text: // Copyright (c) Microsoft Corporation. All rights reserved.
  // Licensed under the MIT license.
  
  #include <cstring>
  #include <sys/ioctl.h>
  #include <linux/fs.h>
  #include <errno.h>
  #include <fcntl.h>
  #include <libgen.h>
  #include <stdio.h>
  #include <time.h>
  #include "file_linux.h"
  
  namespace FASTER {
  namespace environment {
  
  #ifdef _DEBUG
  #define DCHECK_ALIGNMENT(o, l, b) \
  do { \
    assert(reinterpret_cast<uintptr_t>(b) % device_alignment() == 0); \
    assert((o) % device_alignment() == 0); \
    assert((l) % device_alignment() == 0); \
  } while (0)
  #else
  #define DCHECK_ALIGNMENT(o, l, b) do {} while(0)
  #endif
  
  Status File::Open(int flags, FileCreateDisposition create_disposition, bool* exists) {
    if(exists) {
      *exists = false;
    }
  
    int create_flags = GetCreateDisposition(create_disposition);
  
    /// Always unbuffered (O_DIRECT).
    fd_ = ::open(filename_.c_str(), flags | O_RDWR | create_flags, S_IRUSR | S_IWUSR);
  
    if(exists) {
      // Let the caller know whether the file we tried to open or create (already) exists.
      if(create_disposition == FileCreateDisposition::CreateOrTruncate ||
          create_disposition == FileCreateDisposition::OpenOrCreate) {
        *exists = (errno == EEXIST);
      } else if(create_disposition == FileCreateDisposition::OpenExisting) {
        *exists = (errno != ENOENT);
        if(!*exists) {
          // The file doesn't exist. Don't return an error, since the caller is expecting this case.
          return Status::Ok;
        }
      }
    }
    if(fd_ == -1) {
      int error = errno;
      return Status::IOError;
    }
  
    Status result = GetDeviceAlignment();
    if(result != Status::Ok) {
      Close();
    }
    owner_ = true;
    return result;
  }
  
  Status File::Close() {
    if(fd_ != -1) {
      int result = ::close(fd_);
      fd_ = -1;
      if(result == -1) {
        int error = errno;
        return Status::IOError;
      }
    }
    owner_ = false;
    return Status::Ok;
  }
  
  Status File::Delete() {
    int result = ::remove(filename_.c_str());
    if(result == -1) {
      int error = errno;
      return Status::IOError;
    }
    return Status::Ok;
  }
  
  Status File::GetDeviceAlignment() {
    // For now, just hardcode 512-byte alignment.
    device_alignment_ = 512;
    return Status::Ok;
  }
  
  int File::GetCreateDisposition(FileCreateDisposition create_disposition) {
    switch(create_disposition) {
    case FileCreateDisposition::CreateOrTruncate:
      return O_CREAT | O_TRUNC;
    case FileCreateDisposition::OpenOrCreate:
      return O_CREAT;
    case FileCreateDisposition::OpenExisting:
      return 0;
    default:
      assert(false);
      return 0; // not reached
    }
  }
  
  /// Added later for IO Uring
  void UringQueueIoHandler::UringIoCompletionCallback(io_context_t ctx, struct iocb* iocb, long res,
      long res2) {
    auto callback_context = make_context_unique_ptr<UringIoCallbackContext>(
                              reinterpret_cast<UringIoCallbackContext*>(iocb));
    size_t bytes_transferred;
    Status return_status;
    if(res < 0) {
      return_status = Status::IOError;
      bytes_transferred = 0;
    } else {
      return_status = Status::Ok;
      bytes_transferred = res;
    }
    callback_context->callback(callback_context->caller_context, return_status, bytes_transferred);
  }
  /// Added later for IO Uring
  
  
  /// Added later for IO Uring
  bool UringQueueIoHandler::TryComplete() {
    struct timespec timeout;
    std::memset(&timeout, 0, sizeof(timeout));
    struct io_event events[1];
    int result = ::io_getevents(io_object_, 1, 1, events, &timeout);
    if(result == 1) {
      io_callback_t callback = reinterpret_cast<io_callback_t>(events[0].data);
      callback(io_object_, events[0].obj, events[0].res, events[0].res2);
      return true;
    } else {
      return false;
    }
  }
  /// Added later for IO Uring
  
  /// Added later for IO Uring
  Status UringQueueFile::Open(FileCreateDisposition create_disposition, const FileOptions& options,
                         UringQueueIoHandler* handler, bool* exists) {
    int flags = 0;
    if(options.unbuffered) {
      flags |= O_DIRECT;
    }
    RETURN_NOT_OK(File::Open(flags, create_disposition, exists));
    if(exists && !*exists) {
      return Status::Ok;
    }
  
    io_object_ = handler->io_object();
    return Status::Ok;
  }
  
  Status UringQueueFile::Read(size_t offset, uint32_t length, uint8_t* buffer,
                         IAsyncContext& context, AsyncIOCallback callback) const {
    DCHECK_ALIGNMENT(offset, length, buffer);
  #ifdef IO_STATISTICS
    ++read_count_;
    bytes_read_ += length;
  #endif
    return const_cast<UringQueueFile*>(this)->ScheduleOperation(FileOperationType::Read, buffer,
           offset, length, context, callback);
  }
  
  Status UringQueueFile::Write(size_t offset, uint32_t length, const uint8_t* buffer,
                          IAsyncContext& context, AsyncIOCallback callback) {
    DCHECK_ALIGNMENT(offset, length, buffer);
  #ifdef IO_STATISTICS
    bytes_written_ += length;
  #endif
    return ScheduleOperation(FileOperationType::Write, const_cast<uint8_t*>(buffer), offset, length,
                             context, callback);
  }
  
  Status UringQueueFile::ScheduleOperation(FileOperationType operationType, uint8_t* buffer,
                                      size_t offset, uint32_t length, IAsyncContext& context,
                                      AsyncIOCallback callback) {
    auto io_context = alloc_context<UringQueueIoHandler::UringIoCallbackContext>(sizeof(
                        UringQueueIoHandler::UringIoCallbackContext));
    if(!io_context.get()) return Status::OutOfMemory;
  
    IAsyncContext* caller_context_copy;
    RETURN_NOT_OK(context.DeepCopy(caller_context_copy));
  
    new(io_context.get()) UringQueueIoHandler::UringIoCallbackContext(operationType, fd_, offset, length,
        buffer, caller_context_copy, callback);
  
    struct iocb* iocbs[1];
    iocbs[0] = reinterpret_cast<struct iocb*>(io_context.get());
  
    int result = ::io_submit(io_object_, 1, iocbs);
    if(result != 1) {
      return Status::IOError;
    }
  
    io_context.release();
    return Status::Ok;
  }
  /// Added later for IO Uring
  
  #undef DCHECK_ALIGNMENT
  
  }
  } // namespace FASTER::environment
  