Add a KvikIO-backed DiskResource for blocking disk spill I/O

cmake/thirdparty/get_kvikio.cmake

@@ -0,0 +1,12 @@
+# =============================================================================
+# cmake-format: off
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+# cmake-format: on
+# =============================================================================
+
+rapids_find_package(
+  kvikio REQUIRED
+  BUILD_EXPORT_SET rapidsmpf-exports
+  INSTALL_EXPORT_SET rapidsmpf-exports
+)

conda/environments/all_cuda-129_arch-aarch64.yaml

@@ -26,6 +26,7 @@ dependencies:
 - gcc_linux-aarch64=14.*
 - gdb
 - ipython
+- libkvikio==26.10.*,>=0.0.0a0
 - libnuma
 - libpmix-devel >=5.0,<6.0
 - librmm==26.10.*,>=0.0.0a0

conda/environments/all_cuda-129_arch-x86_64.yaml

@@ -26,6 +26,7 @@ dependencies:
 - gcc_linux-64=14.*
 - gdb
 - ipython
+- libkvikio==26.10.*,>=0.0.0a0
 - libnuma
 - libpmix-devel >=5.0,<6.0
 - librmm==26.10.*,>=0.0.0a0

conda/environments/all_cuda-133_arch-aarch64.yaml

@@ -26,6 +26,7 @@ dependencies:
 - gcc_linux-aarch64=14.*
 - gdb
 - ipython
+- libkvikio==26.10.*,>=0.0.0a0
 - libnuma
 - libpmix-devel >=5.0,<6.0
 - librmm==26.10.*,>=0.0.0a0

conda/environments/all_cuda-133_arch-x86_64.yaml

@@ -26,6 +26,7 @@ dependencies:
 - gcc_linux-64=14.*
 - gdb
 - ipython
+- libkvikio==26.10.*,>=0.0.0a0
 - libnuma
 - libpmix-devel >=5.0,<6.0
 - librmm==26.10.*,>=0.0.0a0

conda/recipes/librapidsmpf/recipe.yaml

@@ -77,6 +77,7 @@ cache:
       - cuda-version =${{ cuda_version }}
       - cuda-cupti-dev
       - cuda-nvml-dev
+      - libkvikio =${{ minor_version }}
       - libnuma
       - libpmix-devel >=5.0,<6.0
       - librmm =${{ minor_version }}
@@ -111,13 +112,15 @@ outputs:
         - cuda-version =${{ cuda_version }}
         - cuda-cupti-dev
         - cuda-nvml-dev
+        - libkvikio =${{ minor_version }}
         - libpmix-devel >=5.0,<6.0
         - libucxx ${{ ucxx_version }}
         - openmpi >=5.0
         - libnuma
       run:
         - ${{ pin_compatible("cuda-version", upper_bound="x", lower_bound="x") }}
         - cuda-cupti
+        - libkvikio =${{ minor_version }}
         - librmm =${{ minor_version }}
         - libpmix >=5.0,<6.0
         - libucxx ${{ ucxx_version }}
@@ -130,6 +133,7 @@ outputs:
           - ${{ stdlib("c") }}
         by_name:
           - cuda-cupti
+          - libkvikio
           - librmm
           - openmpi
     tests:

cpp/CMakeLists.txt

@@ -132,6 +132,7 @@ include(../cmake/thirdparty/get_nvtx.cmake)
 include(../cmake/thirdparty/get_rmm.cmake)
 include(../cmake/thirdparty/get_cuco.cmake)
 include(../cmake/thirdparty/get_cucascade.cmake)
+include(../cmake/thirdparty/get_kvikio.cmake)
 
 if(RAPIDSMPF_HAVE_UCXX)
   rapids_find_package(
@@ -202,6 +203,7 @@ add_library(
   src/memory/scoped_memory_record.cpp
   src/memory/spill.cpp
   src/memory/spill_manager.cpp
+  src/disk/disk_resource.cpp
   src/pausable_thread_loop.cpp
   src/progress_thread.cpp
   src/rmm_resource_adaptor.cpp
@@ -306,8 +308,8 @@ endif()
 
 target_link_libraries(
   rapidsmpf
-  PUBLIC rmm::rmm CCCL::CCCL $<TARGET_NAME_IF_EXISTS:ucxx::ucxx> $<TARGET_NAME_IF_EXISTS:libcoro>
-         CUDA::cudart_static
+  PUBLIC rmm::rmm CCCL::CCCL kvikio::kvikio $<TARGET_NAME_IF_EXISTS:ucxx::ucxx>
+         $<TARGET_NAME_IF_EXISTS:libcoro> CUDA::cudart_static
   PRIVATE cuco::cuco
           cuCascade::cucascade_topology_discovery
           $<$<BOOL:${RAPIDSMPF_HAVE_NUMA}>:numa>

cpp/include/rapidsmpf/config_defaults.hpp

@@ -37,6 +37,7 @@ inline const std::unordered_map<std::string, std::string> DEFAULTS{
     {"allow_overbooking_by_default", "true"},
     {"log", "WARN"},
     {"ucxx_progress_mode", "thread-blocking"},
+    {"disk_spill_dir", ""},
 };
 
 }  // namespace rapidsmpf::config

cpp/include/rapidsmpf/disk/disk_resource.hpp

@@ -0,0 +1,135 @@
+/**
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#pragma once
+
+#include <cstddef>
+#include <filesystem>
+#include <memory>
+
+#include <rapidsmpf/config.hpp>
+#include <rapidsmpf/memory/memory_type.hpp>
+
+namespace rapidsmpf::disk {
+
+/**
+ * @brief Future for a DiskResource read or write.
+ *
+ * I/O is submitted before the corresponding DiskResource method returns. The
+ * pointer passed to write()/read() must remain valid until get() returns.
+ *
+ * Concrete implementations live in the translation unit and are returned as
+ * `std::unique_ptr<DiskFuture>`.
+ */
+class DiskFuture {
+  public:
+    DiskFuture() = default;
+    virtual ~DiskFuture() = default;
+    DiskFuture(DiskFuture&&) = default;  ///< Movable.
+    /**
+     * @brief Move assignment.
+     * @returns Moved this.
+     */
+    DiskFuture& operator=(DiskFuture&&) = default;
+    DiskFuture(DiskFuture const&) = delete;  ///< Not copyable.
+    DiskFuture& operator=(DiskFuture const&) = delete;  ///< Not copy-assignable.
+
+    /**
+     * @brief Whether this future refers to outstanding I/O.
+     *
+     * @return True until get() consumes the result.
+     */
+    [[nodiscard]] virtual bool valid() const noexcept = 0;
+
+    /**
+     * @brief Wait for the transfer, release backend resources, and return the
+     * byte count.
+     *
+     * @return Number of bytes transferred. The caller must check this against
+     *         the requested size.
+     *
+     * @note Like `std::future::get()`, this is not thread-safe. Calling get()
+     *       concurrently from multiple threads is undefined behavior.
+     */
+    [[nodiscard]] virtual std::size_t get() = 0;
+};
+
+/**
+ * @brief Non-stream-ordered disk I/O for host or device byte buffers.
+ *
+ * Uses KvikIO with CompatMode::AUTO (GDS when available, POSIX/compat otherwise).
+ *
+ * Callers must synchronize any CUDA stream that produced or consumes a device
+ * pointer before calling write() or read(). KvikIO is not asked to synchronize
+ * the default stream (`sync_default_stream=false`).
+ *
+ * Disk I/O is intentionally outside the MemoryType / BufferResource taxonomy.
+ */
+class DiskResource {
+  public:
+    DiskResource() = default;
+    ~DiskResource() = default;
+
+    DiskResource(DiskResource const&) = delete;
+    DiskResource& operator=(DiskResource const&) = delete;
+    DiskResource(DiskResource&&) = delete;
+    DiskResource& operator=(DiskResource&&) = delete;
+
+    /**
+     * @brief Write bytes to a file.
+     *
+     * @param path File path.
+     * @param data Host or device pointer to the source bytes.
+     * @param size Number of bytes to write.
+     * @param mem_type Memory type of @p data.
+     * @param file_offset Byte offset within the file.
+     * @return Future that owns backend resources until it is waited.
+     */
+    [[nodiscard]] std::unique_ptr<DiskFuture> write(
+        std::filesystem::path const& path,
+        void const* data,
+        std::size_t size,
+        MemoryType mem_type,
+        std::size_t file_offset = 0
+    );
+
+    /**
+     * @brief Read bytes from a file.
+     *
+     * @param path File path.
+     * @param data Host or device pointer to the destination buffer.
+     * @param size Number of bytes to read.
+     * @param mem_type Memory type of @p data.
+     * @param file_offset Byte offset within the file.
+     * @return Future that owns backend resources until it is waited.
+     */
+    [[nodiscard]] std::unique_ptr<DiskFuture> read(
+        std::filesystem::path const& path,
+        void* data,
+        std::size_t size,
+        MemoryType mem_type,
+        std::size_t file_offset = 0
+    );
+
+    /**
+     * @brief Durably synchronize file data to storage (fdatasync).
+     *
+     * Not used on the default spill path; exposed for benchmark durability cases.
+     *
+     * @param path File path.
+     */
+    void flush(std::filesystem::path const& path);
+};
+
+/**
+ * @brief Spill directory from `disk_spill_dir` (`RAPIDSMPF_DISK_SPILL_DIR`).
+ *
+ * An empty option uses the system temporary directory.
+ *
+ * @param options Configuration options.
+ * @return Directory used for spill files.
+ */
+[[nodiscard]] std::filesystem::path default_spill_directory(config::Options options);
+
+}  // namespace rapidsmpf::disk

cpp/src/disk/disk_resource.cpp

@@ -0,0 +1,133 @@
+/**
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <cerrno>
+#include <cstring>
+#include <future>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <utility>
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <kvikio/compat_mode.hpp>
+#include <kvikio/file_handle.hpp>
+
+#include <rapidsmpf/config.hpp>
+#include <rapidsmpf/disk/disk_resource.hpp>
+#include <rapidsmpf/error.hpp>
+
+namespace rapidsmpf::disk {
+
+namespace {
+
+class KvikioDiskFuture final : public DiskFuture {
+  public:
+    KvikioDiskFuture(
+        std::unique_ptr<kvikio::FileHandle> file, std::future<std::size_t> io
+    )
+        : file_{std::move(file)}, io_{std::move(io)} {}
+
+    ~KvikioDiskFuture() override {
+        if (io_.valid()) {
+            try {
+                std::ignore = get();
+            } catch (...) {
+            }
+        }
+    }
+
+    [[nodiscard]] bool valid() const noexcept override {
+        return io_.valid();
+    }
+
+    [[nodiscard]] std::size_t get() override {
+        auto const n = io_.get();
+        file_->close();
+        return n;
+    }
+
+  private:
+    std::unique_ptr<kvikio::FileHandle> file_;
+    std::future<std::size_t> io_;
+};
+
+}  // namespace
+
+std::unique_ptr<DiskFuture> DiskResource::write(
+    std::filesystem::path const& path,
+    void const* data,
+    std::size_t size,
+    [[maybe_unused]] MemoryType mem_type,
+    std::size_t file_offset
+) {
+    auto file = std::make_unique<kvikio::FileHandle>(
+        path.string(), "w+", kvikio::FileHandle::m644, kvikio::CompatMode::AUTO
+    );
+    auto io = file->pwrite(
+        data,
+        size,
+        file_offset,
+        kvikio::defaults::task_size(),
+        kvikio::defaults::gds_threshold(),
+        false  // sync_default_stream
+    );
+    return std::make_unique<KvikioDiskFuture>(std::move(file), std::move(io));
+}
+
+std::unique_ptr<DiskFuture> DiskResource::read(
+    std::filesystem::path const& path,
+    void* data,
+    std::size_t size,
+    [[maybe_unused]] MemoryType mem_type,
+    std::size_t file_offset
+) {
+    auto file = std::make_unique<kvikio::FileHandle>(
+        path.string(), "r", kvikio::FileHandle::m644, kvikio::CompatMode::AUTO
+    );
+    auto io = file->pread(
+        data,
+        size,
+        file_offset,
+        kvikio::defaults::task_size(),
+        kvikio::defaults::gds_threshold(),
+        false  // sync_default_stream
+    );
+    return std::make_unique<KvikioDiskFuture>(std::move(file), std::move(io));
+}
+
+void DiskResource::flush(std::filesystem::path const& path) {
+    auto const fd = ::open(path.c_str(), O_RDONLY);
+    RAPIDSMPF_EXPECTS(
+        fd >= 0,
+        "open for fdatasync failed: " + std::string{std::strerror(errno)},
+        std::runtime_error
+    );
+    if (::fdatasync(fd) != 0) {
+        auto const error = std::string{std::strerror(errno)};
+        ::close(fd);
+        RAPIDSMPF_FAIL("fdatasync failed: " + error, std::runtime_error);
+    }
+    RAPIDSMPF_EXPECTS(
+        ::close(fd) == 0,
+        "close after fdatasync failed: " + std::string{std::strerror(errno)},
+        std::runtime_error
+    );
+}
+
+std::filesystem::path default_spill_directory(config::Options options) {
+    return options.get<std::filesystem::path>(
+        "disk_spill_dir", [](std::string const& value) {
+            if (value.empty()) {
+                return std::filesystem::temp_directory_path();
+            }
+            return std::filesystem::path{value};
+        }
+    );
+}
+
+}  // namespace rapidsmpf::disk

cpp/tests/CMakeLists.txt

@@ -84,6 +84,7 @@ target_sources(
           test_config.cpp
           test_cuda_stream.cpp
           test_cupti_monitor.cpp
+          test_disk_resource.cpp
           test_error_macros.cpp
           test_host_buffer.cpp
           test_memory_resources.cpp