file_handle.hpp

/*
 * SPDX-FileCopyrightText: Copyright (c) 2021-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 */
#pragma once
 
#include <sys/stat.h>
#include <sys/types.h>
 
#include <atomic>
#include <cstddef>
#include <cstdlib>
 
#include <kvikio/buffer.hpp>
#include <kvikio/compat_mode.hpp>
#include <kvikio/compat_mode_manager.hpp>
#include <kvikio/cufile/config.hpp>
#include <kvikio/defaults.hpp>
#include <kvikio/error.hpp>
#include <kvikio/file_utils.hpp>
#include <kvikio/shim/cufile.hpp>
#include <kvikio/shim/cufile_h_wrapper.hpp>
#include <kvikio/stream.hpp>
#include <kvikio/threadpool_wrapper.hpp>
#include <kvikio/utils.hpp>
 
namespace kvikio {
 
class FileHandle {
 private:
  // We use two file descriptors, one opened with the O_DIRECT flag and one without.
  FileWrapper _file_direct_on{};
  FileWrapper _file_direct_off{};
  bool _initialized{false};
  // The size of the underlying file, zero meaning unknown.
  mutable std::atomic<std::size_t> _nbytes{0};
  CUFileHandleWrapper _cufile_handle{};
  CompatModeManager _compat_mode_manager;
  std::string _file_path;  // Reported to the monitors, see `Observation::source`.
  friend class CompatModeManager;
 
  std::size_t read_impl(void* devPtr_base,
                        std::size_t size,
                        std::size_t file_offset,
                        std::size_t devPtr_offset,
                        bool sync_default_stream);
 
  std::size_t write_impl(void const* devPtr_base,
                         std::size_t size,
                         std::size_t file_offset,
                         std::size_t devPtr_offset,
                         bool sync_default_stream);
  ThreadPool* _thread_pool{};
 
 public:
  // 644 is a common setting of Unix file permissions: read and write for owner, read-only for group
  // and others.
  static constexpr mode_t m644 = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
  FileHandle() noexcept        = default;
 
  FileHandle(std::string const& file_path,
             std::string const& flags = "r",
             mode_t mode              = m644,
             CompatMode compat_mode   = defaults::compat_mode());
 
  FileHandle(FileHandle const&)            = delete;
  FileHandle& operator=(FileHandle const&) = delete;
  FileHandle(FileHandle&& o) noexcept;
  FileHandle& operator=(FileHandle&& o) noexcept;
  ~FileHandle() noexcept;
 
  [[nodiscard]] bool closed() const noexcept;
 
  void close() noexcept;
 
  [[nodiscard]] CUfileHandle_t handle();
 
  [[nodiscard]] int fd(bool o_direct = false) const noexcept;
 
  [[nodiscard]] int fd_open_flags(bool o_direct = false) const;
 
  [[nodiscard]] std::size_t nbytes() const;
 
  std::size_t read(void* devPtr_base,
                   std::size_t size,
                   std::size_t file_offset,
                   std::size_t devPtr_offset,
                   bool sync_default_stream = true);
 
  std::size_t write(void const* devPtr_base,
                    std::size_t size,
                    std::size_t file_offset,
                    std::size_t devPtr_offset,
                    bool sync_default_stream = true);
 
  std::future<std::size_t> pread(void* buf,
                                 std::size_t size,
                                 std::size_t file_offset   = 0,
                                 std::size_t task_size     = defaults::task_size(),
                                 std::size_t gds_threshold = defaults::gds_threshold(),
                                 bool sync_default_stream  = true,
                                 ThreadPool* thread_pool   = &defaults::thread_pool());
 
  std::future<std::size_t> pwrite(void const* buf,
                                  std::size_t size,
                                  std::size_t file_offset   = 0,
                                  std::size_t task_size     = defaults::task_size(),
                                  std::size_t gds_threshold = defaults::gds_threshold(),
                                  bool sync_default_stream  = true,
                                  ThreadPool* thread_pool   = &defaults::thread_pool());
 
  void read_async(void* devPtr_base,
                  std::size_t* size_p,
                  off_t* file_offset_p,
                  off_t* devPtr_offset_p,
                  ssize_t* bytes_read_p,
                  CUstream stream);
 
  [[nodiscard]] StreamFuture read_async(void* devPtr_base,
                                        std::size_t size,
                                        off_t file_offset   = 0,
                                        off_t devPtr_offset = 0,
                                        CUstream stream     = nullptr);
 
  void write_async(void* devPtr_base,
                   std::size_t* size_p,
                   off_t* file_offset_p,
                   off_t* devPtr_offset_p,
                   ssize_t* bytes_written_p,
                   CUstream stream);
 
  [[nodiscard]] StreamFuture write_async(void* devPtr_base,
                                         std::size_t size,
                                         off_t file_offset   = 0,
                                         off_t devPtr_offset = 0,
                                         CUstream stream     = nullptr);
 
  const CompatModeManager& get_compat_mode_manager() const noexcept;
 
  bool is_direct_io_supported() const noexcept;
};
 
}  // namespace kvikio