diff --git a/include/cucascade/cudf/datasource.hpp b/include/cucascade/cudf/datasource.hpp index 9ec79a8..2a8fb90 100644 --- a/include/cucascade/cudf/datasource.hpp +++ b/include/cucascade/cudf/datasource.hpp @@ -20,11 +20,13 @@ #include #include +#include #include #include #include +#include namespace cucascade::io { @@ -138,6 +140,22 @@ class datasource : public cudf::io::datasource { /// @c disposable call at consume time. void fadvise(std::span ranges, std::optional dev_id); + /** + * @brief Submit all byte ranges as a single vectorized host read. + * + * Uses the ioctx's scatter-read backend to fetch all segments in as few + * HTTP requests as possible, writing each range directly into the + * caller-supplied destination pointer in its @p segment. + * + * @param segments File offsets, sizes, and destination buffers. + * @return A future that resolves when every segment has been written. + */ + [[nodiscard]] std::future host_read_ranges_async( + std::span segments); + + [[nodiscard]] std::future host_read_ranges_async( + std::vector& segments); + void prefetch(cache::prefetching_stage site); private: diff --git a/include/cucascade/cudf/rest_datasource_engine.hpp b/include/cucascade/cudf/rest_datasource_engine.hpp new file mode 100644 index 0000000..d4fd7a6 --- /dev/null +++ b/include/cucascade/cudf/rest_datasource_engine.hpp @@ -0,0 +1,98 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include + +namespace cucascade::io { + +class ioctx; + +/** + * @brief Self-contained REST datasource engine for S3/HTTP object-store reads. + * + * Owns a NUMA-local pinned host staging pool and a pool of libcurl reactor + * threads backed by SigV4 presigned-URL signing. Callers open individual + * @c datasource instances via @c open(); each datasource shares the engine's + * @c ioctx and memory pool but carries its own per-scan @c prefetching_handle. + * + * The engine must outlive every datasource it produces. + * + * @code{.cpp} + * auto engine = cucascade::io::rest_datasource_engine::make_s3(...); + * auto ds = engine->open("s3://my-bucket/data/lineitem.parquet"); + * ds->fadvise(byte_ranges, device_id); + * @endcode + */ +class rest_datasource_engine { + public: + static constexpr std::size_t default_block_size = 1UL << 20; + static constexpr std::size_t default_pool_capacity = 20UL * 128UL * (1UL << 20); + + /** + * @brief Construct an S3-backed REST engine with SigV4 presigned-URL signing. + * + * Credentials are static for the lifetime of the engine. For short-lived STS + * tokens, reconstruct the engine before the token expires. + * + * @param access_key_id AWS access key ID. + * @param secret_access_key AWS secret access key. + * @param session_token STS session token; empty for long-lived credentials. + * @param region AWS region (e.g. @c "us-east-1"). + * @param endpoint S3-compatible endpoint host (e.g. @c "s3.amazonaws.com" + * or a MinIO host:port). Leave empty to derive from region. + * @param n_reactors Number of libcurl reactor threads. + * @param tls_verify Whether to verify TLS peer certificates. + * @param pool_capacity Total capacity of the pinned host staging pool in bytes. + * @param block_size Fixed block size in bytes for the staging pool. + * @return A ready-to-use engine. The engine's @c ioctx is started before returning. + */ + explicit rest_datasource_engine(std::string access_key_id, + std::string secret_access_key, + std::string session_token, + std::string region, + std::string endpoint, + std::size_t n_reactors = 4, + bool tls_verify = true, + std::size_t pool_capacity = default_pool_capacity, + std::size_t block_size = default_block_size, + std::size_t max_connections = 16, + std::size_t chunk_size = 8UL << 20, + std::size_t max_n_chunks = 16, + bool enable_cache = false); + + ~rest_datasource_engine(); + + rest_datasource_engine(rest_datasource_engine const&) = delete; + rest_datasource_engine& operator=(rest_datasource_engine const&) = delete; + + /** + * @brief Open a datasource for the given S3 URI. + * + * Issues an HTTP HEAD request to resolve the object size. + * + * @param path S3 URI of the form @c "s3://bucket/key". + * @return A @c datasource bound to this engine's @c ioctx. The returned + * datasource must not outlive this engine. + * @throw std::runtime_error if the HEAD request fails or the URI is malformed. + */ + [[nodiscard]] std::unique_ptr open(std::string path) const; + + private: + cucascade::memory::numa_region_pinned_host_memory_resource _upstream; + cucascade::memory::fixed_size_host_memory_resource _host_mr; + std::shared_ptr _io_ctx; + std::unique_ptr _reservation_manager; +}; + +} // namespace cucascade::io diff --git a/include/cucascade/cudf/uring_datasource_engine.hpp b/include/cucascade/cudf/uring_datasource_engine.hpp new file mode 100644 index 0000000..765b6d7 --- /dev/null +++ b/include/cucascade/cudf/uring_datasource_engine.hpp @@ -0,0 +1,82 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace cucascade::io { + +/** + * @brief Self-contained io_uring datasource engine for local NVMe reads. + * + * Owns the full stack needed for O_DIRECT io_uring reads: a NUMA-local pinned + * host memory pool, a pool of io_uring reactor threads, and an @c ioctx. + * Callers open individual @c datasource instances via @c open(); each datasource + * shares the engine's @c ioctx and memory pool but carries its own per-scan + * @c prefetching_handle. + * + * The engine must outlive every datasource it produces. + * + * @code{.cpp} + * cucascade::io::uring_datasource_engine engine; + * auto ds = engine.open("/mnt/nvme/data/lineitem.parquet"); + * ds->fadvise(byte_ranges, device_id); + * // ... read through ds as a cudf::io::datasource ... + * @endcode + */ +class uring_datasource_engine { + public: + static constexpr std::size_t default_block_size = 1UL << 20; ///< 1 MiB + static constexpr std::size_t default_pool_capacity = 20UL * 128UL * (1UL << 20); ///< ~2.5 GiB + + /** + * @brief Construct a uring datasource engine. + * + * @param n_reactors Number of io_uring reactor threads. + * @param pool_capacity Total capacity of the pinned host staging pool in bytes. + * @param block_size Size of each fixed-size block in the staging pool in bytes. + * Must be a power of two and at least the alignment required + * by O_DIRECT on the target filesystem. + * @param use_odirect Whether to open files with @c O_DIRECT (bypasses page cache). + * @param numa_node NUMA node from which to allocate the pinned staging pool. + */ + explicit uring_datasource_engine(std::size_t n_reactors = 2, + std::size_t pool_capacity = default_pool_capacity, + std::size_t block_size = default_block_size, + bool use_odirect = true, + int numa_node = 0); + + ~uring_datasource_engine(); + + uring_datasource_engine(uring_datasource_engine const&) = delete; + uring_datasource_engine& operator=(uring_datasource_engine const&) = delete; + + /** + * @brief Open a datasource for the given local file path. + * + * @param path Absolute or relative path to a local file. + * @return A @c datasource bound to this engine's @c ioctx. The returned + * datasource must not outlive this engine. + * @throw std::runtime_error if the file cannot be opened. + */ + [[nodiscard]] std::unique_ptr open(std::string path) const; + + private: + cucascade::memory::numa_region_pinned_host_memory_resource _upstream; + cucascade::memory::fixed_size_host_memory_resource _host_mr; + std::shared_ptr _reactor_ctx; + std::shared_ptr _io_ctx; +}; + +} // namespace cucascade::io diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt new file mode 100644 index 0000000..b100808 --- /dev/null +++ b/python/CMakeLists.txt @@ -0,0 +1,60 @@ +# ============================================================================= +# cmake-format: off +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# cmake-format: on +# ============================================================================= + +cmake_minimum_required(VERSION 4.0 FATAL_ERROR) + +project( + cucascade_python + VERSION 0.1.0 + LANGUAGES CXX +) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) + +find_package(PkgConfig REQUIRED) +pkg_check_modules(LIBURING REQUIRED IMPORTED_TARGET liburing) +pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl) +find_package(OpenSSL REQUIRED) + +find_package(cudf REQUIRED) +find_package(cuCascade REQUIRED) + +set(rapids-cmake-dir + "" + CACHE PATH "Optional path to an already-fetched rapids-cmake source tree (skips download)" +) +if(rapids-cmake-dir) + list(APPEND CMAKE_MODULE_PATH "${rapids-cmake-dir}/rapids-cmake") +else() + set(rapids-cmake-version "24.12") + include(FetchContent) + FetchContent_Declare( + rapids-cmake + GIT_REPOSITORY https://github.com/rapidsai/rapids-cmake.git + GIT_TAG "branch-${rapids-cmake-version}" + GIT_SHALLOW TRUE + ) + FetchContent_MakeAvailable(rapids-cmake) + list(APPEND CMAKE_MODULE_PATH "${rapids-cmake_SOURCE_DIR}/rapids-cmake") +endif() + +execute_process( + COMMAND "${Python_EXECUTABLE}" -c + "import pylibcudf, pathlib; print(pathlib.Path(pylibcudf.__file__).parent.parent)" + OUTPUT_VARIABLE PYLIBCUDF_SOURCE_DIR + OUTPUT_STRIP_TRAILING_WHITESPACE + COMMAND_ERROR_IS_FATAL ANY +) + +include(rapids-cython-core) +rapids_cython_init() + +set(CYTHON_FLAGS "${CYTHON_FLAGS} -I${PYLIBCUDF_SOURCE_DIR}" CACHE STRING "" FORCE) +message(STATUS "cucascade: pylibcudf source dir: ${PYLIBCUDF_SOURCE_DIR}") + +add_subdirectory(cucascade) diff --git a/python/README.md b/python/README.md new file mode 100644 index 0000000..42525af --- /dev/null +++ b/python/README.md @@ -0,0 +1,78 @@ +# cucascade Python bindings + +Python bindings for cuCascade's cudf datasource layer, exposing `UringEngine` +(local NVMe via io_uring) and `RestEngine` (S3/HTTP) with advisory prefetch +support via `CuCascadeDatasource.fadvise()`. + +## Prerequisites + +- CUDA toolkit +- cudf built or installed (cmake config must be findable) +- cuCascade C++ libraries built and installed +- pylibcudf installed +- Python >= 3.11 + +## 1. Build and install cuCascade C++ + +```bash +CUCASCADE_SRC=/path/to/cuCascade +CUCASCADE_BUILD=${CUCASCADE_SRC}/build +CUDF_CMAKE_DIR=/path/to/cudf/build # directory containing cudf-config.cmake + +cmake -S "${CUCASCADE_SRC}" -B "${CUCASCADE_BUILD}" \ + -DCMAKE_BUILD_TYPE=Release \ + -DCMAKE_CUDA_ARCHITECTURES=native \ + -DCMAKE_INSTALL_PREFIX="${CONDA_PREFIX}" \ + -DCUCASCADE_BUILD_CUDF=ON \ + -DCUCASCADE_BUILD_IO=ON \ + -DCUCASCADE_BUILD_TESTS=OFF \ + -DCUCASCADE_BUILD_BENCHMARKS=OFF \ + -DCUCASCADE_BUILD_SHARED_LIBS=ON \ + -DCUCASCADE_BUILD_STATIC_LIBS=OFF \ + -Dcudf_DIR="${CUDF_CMAKE_DIR}" \ + -DCMAKE_PREFIX_PATH="${CONDA_PREFIX}" + +cmake --build "${CUCASCADE_BUILD}" -j$(nproc) +cmake --install "${CUCASCADE_BUILD}" +``` + +Install to `${CONDA_PREFIX}` so the shared libraries are on the default +dynamic linker search path at runtime. + +## 2. Build and install the Python package + +```bash +CUCASCADE_SRC=/path/to/cuCascade +CUCASCADE_BUILD=${CUCASCADE_SRC}/build +CUDF_CMAKE_DIR=/path/to/cudf/build + +pip install --no-build-isolation --no-deps \ + --config-settings "cmake.args=-DCMAKE_PREFIX_PATH=${CONDA_PREFIX};${CUDF_CMAKE_DIR}" \ + --config-settings "cmake.args=-Dconcurrentqueue_dir=${CUCASCADE_BUILD}/_deps/concurrentqueue-src" \ + "${CUCASCADE_SRC}/python" +``` + +**`CMAKE_PREFIX_PATH`** must include: +- `${CONDA_PREFIX}` — finds cuCascade, liburing, libcurl, OpenSSL +- the cudf cmake build directory — lets `find_package(cudf)` resolve + +**`concurrentqueue_dir`** — the moodycamel concurrentqueue headers are +fetched by the cuCascade C++ build and live at +`${CUCASCADE_BUILD}/_deps/concurrentqueue-src`. + +**`rapids-cmake-dir`** (optional) — if rapids-cmake is already available +(e.g. inside a pylibcudf build directory), pass it to avoid a network fetch: + +```bash + --config-settings "cmake.args=-Drapids-cmake-dir=/path/to/rapids-cmake-src" \ +``` + +Otherwise the build fetches rapids-cmake automatically from GitHub. + +## Verify + +```python +import cucascade +engine = cucascade.UringEngine() +ds = engine.open("/path/to/file.parquet") +``` diff --git a/python/cucascade/CMakeLists.txt b/python/cucascade/CMakeLists.txt new file mode 100644 index 0000000..dbb903b --- /dev/null +++ b/python/cucascade/CMakeLists.txt @@ -0,0 +1,31 @@ +# ============================================================================= +# cmake-format: off +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# cmake-format: on +# ============================================================================= + +set(cython_sources datasource.pyx) +set(linked_libraries cuCascade::cucascade_cudf cudf::cudf) + +rapids_cython_create_modules( + CXX + SOURCE_FILES "${cython_sources}" + LINKED_LIBRARIES "${linked_libraries}" + MODULE_PREFIX cucascade_ + ASSOCIATED_TARGETS cucascade_cudf_shared +) + +get_target_property(_cudf_loc cudf::cudf LOCATION) +get_filename_component(_cudf_lib_dir "${_cudf_loc}" DIRECTORY) +set_target_properties(cucascade_datasource + PROPERTIES INSTALL_RPATH "${_cudf_lib_dir}" +) + +set(concurrentqueue_dir + "" + CACHE PATH "Path to the moodycamel concurrentqueue source directory" +) +if(concurrentqueue_dir) + target_include_directories(cucascade_datasource PRIVATE "${concurrentqueue_dir}") +endif() diff --git a/python/cucascade/__init__.py b/python/cucascade/__init__.py new file mode 100644 index 0000000..dba196c --- /dev/null +++ b/python/cucascade/__init__.py @@ -0,0 +1,6 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from cucascade.datasource import CuCascadeDatasource, ReadFuture, RestEngine, UringEngine + +__all__ = ["CuCascadeDatasource", "ReadFuture", "RestEngine", "UringEngine"] diff --git a/python/cucascade/datasource.pxd b/python/cucascade/datasource.pxd new file mode 100644 index 0000000..e2623b9 --- /dev/null +++ b/python/cucascade/datasource.pxd @@ -0,0 +1,60 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from libc.stdint cimport int64_t, uint8_t +from libcpp cimport bool as cpp_bool +from libcpp.future cimport future +from libcpp.memory cimport unique_ptr +from libcpp.string cimport string +from libcpp.vector cimport vector + + +from pylibcudf.libcudf.io.datasource cimport datasource as cudf_datasource + + +cdef extern from "cudf/io/text/byte_range_info.hpp" namespace "cudf::io::text" nogil: + cdef cppclass byte_range_info: + byte_range_info(int64_t offset, int64_t size) except + + int64_t offset() const + int64_t size() const + + +cdef extern from "cucascade/io/types.hpp" namespace "cucascade::io" nogil: + cdef cppclass io_object_segment: + io_object_segment(size_t offset, size_t size, uint8_t* buffer) except + + + +cdef extern from "cucascade/cudf/datasource.hpp" namespace "cucascade::io" nogil: + cdef cppclass cc_datasource "cucascade::io::datasource"(cudf_datasource): + unique_ptr[cc_datasource] duplicate() except + + void fadvise(const vector[byte_range_info]& ranges, int dev_id) except + + future[size_t] host_read_async(size_t offset, size_t size, uint8_t* dst) except + + future[size_t] host_read_ranges_async(vector[io_object_segment]& segments) except + + + +cdef extern from "cucascade/cudf/uring_datasource_engine.hpp" namespace "cucascade::io" nogil: + cdef cppclass uring_datasource_engine: + uring_datasource_engine(size_t n_reactors, + size_t pool_capacity, + size_t block_size, + cpp_bool use_odirect, + int numa_node) except + + unique_ptr[cc_datasource] open(string path) except + + + +cdef extern from "cucascade/cudf/rest_datasource_engine.hpp" namespace "cucascade::io" nogil: + cdef cppclass rest_datasource_engine: + rest_datasource_engine(string access_key_id, + string secret_access_key, + string session_token, + string region, + string endpoint, + size_t n_reactors, + cpp_bool tls_verify, + size_t pool_capacity, + size_t block_size, + size_t max_connections, + size_t chunk_size, + size_t max_n_chunks, + cpp_bool enable_cache) except + + unique_ptr[cc_datasource] open(string path) except + diff --git a/python/cucascade/datasource.pyi b/python/cucascade/datasource.pyi new file mode 100644 index 0000000..d5a293b --- /dev/null +++ b/python/cucascade/datasource.pyi @@ -0,0 +1,53 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from pylibcudf.io.datasource import Datasource + + +class ReadFuture: + def get(self) -> int: ... + + +class CuCascadeDatasource(Datasource): + def fadvise(self, ranges: list[tuple[int, int]], dev_id: int = -1) -> None: ... + def duplicate(self) -> CuCascadeDatasource: ... + def read_ranges_async( + self, ranges: list[tuple[int, int]], buffer: memoryview + ) -> list[ReadFuture]: ... + def read_all_ranges_async( + self, ranges: list[tuple[int, int]], buffer: memoryview + ) -> ReadFuture: ... + + +class UringEngine: + def __init__( + self, + n_reactors: int = 2, + pool_capacity: int = 2684354560, + block_size: int = 1048576, + use_odirect: bool = True, + numa_node: int = 0, + ) -> None: ... + def open(self, path: str) -> CuCascadeDatasource: ... + + +class RestEngine: + def __init__( + self, + access_key_id: str = "", + secret_access_key: str = "", + session_token: str = "", + region: str = "us-east-1", + endpoint: str = "", + n_reactors: int = 4, + tls_verify: bool = True, + pool_capacity: int = 2684354560, + block_size: int = 1048576, + max_connections: int = 16, + chunk_size: int = 8388608, + max_n_chunks: int = 16, + enable_cache: bool = False, + ) -> None: ... + def open(self, path: str) -> CuCascadeDatasource: ... diff --git a/python/cucascade/datasource.pyx b/python/cucascade/datasource.pyx new file mode 100644 index 0000000..fa6323b --- /dev/null +++ b/python/cucascade/datasource.pyx @@ -0,0 +1,367 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""cuCascade datasource bindings for use with pylibcudf readers. + +Provides :class:`UringEngine` (local NVMe via io_uring) and +:class:`RestEngine` (S3/HTTP via libcurl), both producing +:class:`CuCascadeDatasource` instances that implement +``cudf::io::datasource`` and support advisory prefetch via +:meth:`CuCascadeDatasource.fadvise`. +""" + +from cython.operator cimport dereference as deref +from libc.stdint cimport int64_t, uint8_t +from libcpp.future cimport future +from libcpp.memory cimport unique_ptr +from libcpp.string cimport string +from libcpp.utility cimport move +from libcpp.vector cimport vector + +from pylibcudf.io.datasource cimport Datasource +from pylibcudf.libcudf.io.datasource cimport datasource as cudf_datasource + +from cucascade.datasource cimport ( + byte_range_info, + cc_datasource, + io_object_segment, + rest_datasource_engine, + uring_datasource_engine, +) + +__all__ = ["CuCascadeDatasource", "ReadFuture", "RestEngine", "UringEngine"] + + +cdef class ReadFuture: + """A pending async host read issued by :meth:`CuCascadeDatasource.read_ranges_async`. + + Call :meth:`get` to block until the read completes. + """ + + cdef future[size_t] _fut + + def get(self): + """Block until the read completes. + + Returns + ------- + int + Number of bytes transferred. + """ + cdef size_t result + with nogil: + result = self._fut.get() + return result + + +cdef class CuCascadeDatasource(Datasource): + """A pylibcudf :class:`~pylibcudf.io.datasource.Datasource` backed by + a ``cucascade::io::datasource``. + + Instances are produced by :meth:`UringEngine.open` or + :meth:`RestEngine.open` and are not constructed directly. + + The owning engine must outlive every datasource it produces. + When a single file is read by multiple concurrent splits, call + :meth:`duplicate` to obtain an independent datasource per split so + that their :meth:`fadvise` calls do not interfere with each other. + """ + + cdef unique_ptr[cc_datasource] _ds + + cdef cudf_datasource* get_datasource(self) except * nogil: + return self._ds.get() + + def fadvise(self, list ranges, int dev_id=-1): + """Hint the IO layer about byte ranges this scan will read soon. + + Queues the ranges into the engine's prefetch cache so that the + data is staged into pinned host memory before the caller blocks + on :meth:`~pylibcudf.io.types.SourceInfo`-based reads. + + Parameters + ---------- + ranges : list[tuple[int, int]] + Byte ranges as ``(offset, size)`` pairs, for example from + :meth:`~pylibcudf.io.experimental.HybridScanReader.filter_column_chunks_byte_ranges` + or + :meth:`~pylibcudf.io.experimental.HybridScanReader.payload_column_chunks_byte_ranges`. + dev_id : int, optional + Preferred CUDA device id for pinned-host staging placement. + Pass ``-1`` (the default) to express no preference. + """ + cdef vector[byte_range_info] c_ranges + cdef int64_t off, sz + for off, sz in ranges: + c_ranges.emplace_back(off, sz) + with nogil: + deref(self._ds).fadvise(c_ranges, dev_id) + + def duplicate(self): + """Return a datasource sharing the same file handle with an independent + prefetch handle. + + Use one duplicate per split when a single file is read by several + concurrent :class:`~pylibcudf.io.experimental.HybridScanReader` + instances so that per-split :meth:`fadvise` calls do not overwrite + each other's prefetch state. + + Returns + ------- + CuCascadeDatasource + A new datasource over the same underlying file. + """ + cdef CuCascadeDatasource out = CuCascadeDatasource.__new__(CuCascadeDatasource) + with nogil: + out._ds = move(deref(self._ds).duplicate()) + return out + + def read_ranges_async(self, list ranges, object buffer): + """Submit async host reads for each range into a contiguous buffer. + + The reads are submitted immediately and run on the engine's reactor + threads. Call :meth:`ReadFuture.get` on each returned future to wait + for the corresponding range to complete. + + Parameters + ---------- + ranges : list[tuple[int, int]] + Byte ranges as ``(offset, size)`` pairs in the same order that data + should appear in ``buffer``. + buffer : memoryview + Contiguous writable host buffer sized to hold the sum of all range + sizes. Typically a slice of a :class:`PinnedBuffer` array. + + Returns + ------- + list[ReadFuture] + One future per range, in the same order as ``ranges``. + """ + cdef uint8_t[::1] c_buf = buffer + cdef uint8_t* base = &c_buf[0] + cdef list futures = [] + cdef ReadFuture rf + cdef int64_t off, sz + cdef size_t dst_offset = 0 + for off, sz in ranges: + rf = ReadFuture.__new__(ReadFuture) + rf._fut = deref(self._ds).host_read_async(off, sz, base + dst_offset) + futures.append(rf) + dst_offset += sz + return futures + + def read_all_ranges_async(self, list ranges, object buffer): + """Submit all byte ranges as a single vectorized host read. + + Uses the engine's scatter-read backend to fetch all ranges in as few + HTTP requests as possible, writing each range into ``buffer`` at the + corresponding offset. Returns a single future that resolves when every + range has been written. + + Parameters + ---------- + ranges : list[tuple[int, int]] + Byte ranges as ``(offset, size)`` pairs in file order. + buffer : memoryview + Contiguous writable host buffer sized to hold the sum of all range + sizes. + + Returns + ------- + ReadFuture + A single future that resolves when all ranges have been written. + """ + cdef uint8_t[::1] c_buf = buffer + cdef uint8_t* base = &c_buf[0] + cdef vector[io_object_segment] segments + cdef int64_t off, sz + cdef size_t dst_offset = 0 + for off, sz in ranges: + segments.emplace_back(off, sz, base + dst_offset) + dst_offset += sz + cdef ReadFuture rf = ReadFuture.__new__(ReadFuture) + rf._fut = deref(self._ds).host_read_ranges_async(segments) + return rf + + +cdef class UringEngine: + """io_uring-backed datasource engine for local NVMe reads. + + Owns a NUMA-local pinned host staging pool and a pool of io_uring + reactor threads. All datasources produced by :meth:`open` share the + engine's resources; the engine must therefore outlive every datasource + it produces. + + Parameters + ---------- + n_reactors : int, optional + Number of io_uring reactor threads. Default is 2. + pool_capacity : int, optional + Total capacity of the pinned host staging pool in bytes. + Default is approximately 2.5 GiB (20 × 128 MiB). + block_size : int, optional + Fixed block size in bytes for the staging pool. Must be a power + of two and at least the O_DIRECT alignment requirement of the + target filesystem. Default is 1 MiB. + use_odirect : bool, optional + Whether to open files with ``O_DIRECT`` to bypass the page cache. + Default is ``True``. + numa_node : int, optional + NUMA node from which to allocate the pinned staging pool. + Default is 0. + """ + + cdef unique_ptr[uring_datasource_engine] _engine + + def __cinit__( + self, + size_t n_reactors=2, + size_t pool_capacity=2684354560, + size_t block_size=1048576, + bint use_odirect=True, + int numa_node=0, + ): + with nogil: + self._engine.reset( + new uring_datasource_engine( + n_reactors, pool_capacity, block_size, use_odirect, numa_node + ) + ) + + def open(self, str path): + """Open a datasource for a local file. + + Parameters + ---------- + path : str + Path to the local file. + + Returns + ------- + CuCascadeDatasource + A datasource bound to this engine. Must not outlive the engine. + + Raises + ------ + RuntimeError + If the file cannot be opened. + """ + cdef CuCascadeDatasource ds = CuCascadeDatasource.__new__(CuCascadeDatasource) + cdef string c_path = path.encode() + with nogil: + ds._ds = move(deref(self._engine).open(c_path)) + return ds + + +cdef class RestEngine: + """libcurl-backed datasource engine for S3/HTTP object-store reads. + + Owns a NUMA-local pinned host staging pool and a pool of libcurl + reactor threads with SigV4 presigned-URL signing. All datasources + produced by :meth:`open` share the engine's resources; the engine + must therefore outlive every datasource it produces. + + Parameters + ---------- + access_key_id : str, optional + AWS access key ID. Default is ``""`` (reads from environment). + secret_access_key : str, optional + AWS secret access key. Default is ``""`` (reads from environment). + session_token : str, optional + STS session token; leave empty for long-lived credentials. + region : str, optional + AWS region. Default is ``"us-east-1"``. + endpoint : str, optional + S3-compatible endpoint host (e.g. ``"s3.amazonaws.com"`` or a + MinIO ``host:port``). Leave empty to derive from region. + n_reactors : int, optional + Number of libcurl reactor threads. Default is 4. + tls_verify : bool, optional + Whether to verify TLS peer certificates. Default is ``True``. + pool_capacity : int, optional + Total capacity of the pinned host staging pool in bytes. + Default is approximately 2.5 GiB. + block_size : int, optional + Fixed block size in bytes for the staging pool. Default is 1 MiB. + max_connections : int, optional + Maximum concurrent in-flight HTTP connections per reactor. + Default is 16. + chunk_size : int, optional + Maximum bytes per ranged GET request. Adjacent segments are fused + up to this size; oversized segments are split. Default is 8 MiB. + max_n_chunks : int, optional + Maximum number of destination buffers fused into a single scatter + GET. Default is 16. + enable_cache : bool, optional + Whether to enable cuCascade's internal prefetch cache. When ``True``, + :meth:`CuCascadeDatasource.fadvise` queues S3 downloads into cuCascade's + bounce buffer pool so that subsequent reads may be served from cache + rather than S3. Default is ``False``. + """ + + cdef unique_ptr[rest_datasource_engine] _engine + + def __cinit__( + self, + str access_key_id="", + str secret_access_key="", + str session_token="", + str region="us-east-1", + str endpoint="", + size_t n_reactors=4, + bint tls_verify=True, + size_t pool_capacity=2684354560, + size_t block_size=1048576, + size_t max_connections=16, + size_t chunk_size=8388608, + size_t max_n_chunks=16, + bint enable_cache=False, + ): + cdef string c_access_key_id = access_key_id.encode() + cdef string c_secret_access_key = secret_access_key.encode() + cdef string c_session_token = session_token.encode() + cdef string c_region = region.encode() + cdef string c_endpoint = endpoint.encode() + with nogil: + self._engine.reset( + new rest_datasource_engine( + c_access_key_id, + c_secret_access_key, + c_session_token, + c_region, + c_endpoint, + n_reactors, + tls_verify, + pool_capacity, + block_size, + max_connections, + chunk_size, + max_n_chunks, + enable_cache, + ) + ) + + def open(self, str path): + """Open a datasource for an S3 URI. + + Issues an HTTP HEAD request to resolve the object size. + + Parameters + ---------- + path : str + S3 URI of the form ``s3://bucket/key``. + + Returns + ------- + CuCascadeDatasource + A datasource bound to this engine. Must not outlive the engine. + + Raises + ------ + RuntimeError + If the HEAD request fails or the URI is malformed. + """ + cdef CuCascadeDatasource ds = CuCascadeDatasource.__new__(CuCascadeDatasource) + cdef string c_path = path.encode() + with nogil: + ds._ds = move(deref(self._engine).open(c_path)) + return ds diff --git a/python/pyproject.toml b/python/pyproject.toml new file mode 100644 index 0000000..72d8f45 --- /dev/null +++ b/python/pyproject.toml @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +build-backend = "scikit_build_core.build" +requires = [ + "scikit-build-core[pyproject]>=0.11.0", + "cmake>=4.0", + "cython>=3.0.0", + "ninja", +] + +[project] +name = "cucascade" +version = "0.1.0" +description = "Python bindings for cuCascade" +requires-python = ">=3.11" +dependencies = [ + "pylibcudf", +] + +[tool.scikit-build] +build-dir = "build/{wheel_tag}" +cmake.build-type = "Release" +ninja.make-fallback = false +wheel.packages = ["cucascade"] + diff --git a/src/cudf/CMakeLists.txt b/src/cudf/CMakeLists.txt index 3302dd9..9918dac 100644 --- a/src/cudf/CMakeLists.txt +++ b/src/cudf/CMakeLists.txt @@ -25,6 +25,9 @@ target_sources( # The cudf::io::datasource bridge over the cudf-free io core — only built when # the io library is enabled (the io core itself carries no cudf dependency). if(CUCASCADE_BUILD_IO) - target_sources(cucascade_cudf_objects - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/datasource.cpp) + target_sources( + cucascade_cudf_objects + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/datasource.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/uring_datasource_engine.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/rest_datasource_engine.cpp) endif() diff --git a/src/cudf/datasource.cpp b/src/cudf/datasource.cpp index 76143c4..b6bd597 100644 --- a/src/cudf/datasource.cpp +++ b/src/cudf/datasource.cpp @@ -106,6 +106,16 @@ std::future datasource::host_read_async(size_t offset, size_t size, uint _io_ctx->host_read_async(*_io_object, offset, size, dst, &_prefetch_handle)); } +std::future datasource::host_read_ranges_async(std::span segments) +{ + return bridge_semi_to_std(_io_ctx->host_read_ranges_async_io(*_io_object, segments)); +} + +std::future datasource::host_read_ranges_async(std::vector& segments) +{ + return host_read_ranges_async(std::span{segments}); +} + std::future> datasource::host_read_async( size_t offset, size_t size) { diff --git a/src/cudf/rest_datasource_engine.cpp b/src/cudf/rest_datasource_engine.cpp new file mode 100644 index 0000000..d30d2bb --- /dev/null +++ b/src/cudf/rest_datasource_engine.cpp @@ -0,0 +1,91 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cucascade::io { + +rest_datasource_engine::rest_datasource_engine(std::string access_key_id, + std::string secret_access_key, + std::string session_token, + std::string region, + std::string endpoint, + std::size_t n_reactors, + bool tls_verify, + std::size_t pool_capacity, + std::size_t block_size, + std::size_t max_connections, + std::size_t chunk_size, + std::size_t max_n_chunks, + bool enable_cache) + : _upstream(0, true), + _host_mr(0, _upstream, pool_capacity, pool_capacity, block_size, 128, 1) +{ + rest::s3::static_credentials creds{.access_key_id = std::move(access_key_id), + .secret_access_key = std::move(secret_access_key), + .session_token = std::move(session_token), + .expires_at = std::nullopt}; + + auto authorizer = std::make_shared( + std::move(creds), std::move(region), std::move(endpoint)); + + rest::config rest_cfg{}; + rest_cfg.bounce_block_size = _host_mr.get_block_size(); + rest_cfg.tls_verify = tls_verify; + rest_cfg.max_connections = max_connections; + rest_cfg.chunk_size = chunk_size; + rest_cfg.max_n_chunks = max_n_chunks; + + auto rest_ctx = std::make_shared( + std::move(rest_cfg), std::move(authorizer), &_host_mr); + + auto io_ctx = std::make_shared(n_reactors, std::move(rest_ctx)); + io_ctx->start(); + _io_ctx = io_ctx; + + if (enable_cache) { + memory::topology_discovery discovery; + static_cast(discovery.discover()); + auto const& topology = discovery.get_topology(); + + auto configs = memory::reservation_manager_configurator{} + .set_number_of_gpus(1) + .use_host_per_numa() + .set_total_host_capacity(pool_capacity) + .build(topology); + + _reservation_manager = + std::make_unique(std::move(configs)); + + io::cache::config cache_cfg{}; + cache_cfg.dispose_after_use = true; + + auto topo_index = + std::make_shared(topology, std::vector{0}); + + _io_ctx->initialize_cache(*_reservation_manager, cache_cfg, std::move(topo_index)); + } +} + +rest_datasource_engine::~rest_datasource_engine() +{ + _io_ctx->pre_destroy(); + _io_ctx->shutdown(); +} + +std::unique_ptr rest_datasource_engine::open(std::string path) const +{ + return open_datasource(_io_ctx, std::move(path)); +} + +} // namespace cucascade::io diff --git a/src/cudf/uring_datasource_engine.cpp b/src/cudf/uring_datasource_engine.cpp new file mode 100644 index 0000000..ff63324 --- /dev/null +++ b/src/cudf/uring_datasource_engine.cpp @@ -0,0 +1,35 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include + +namespace cucascade::io { + +uring_datasource_engine::uring_datasource_engine(std::size_t n_reactors, + std::size_t pool_capacity, + std::size_t block_size, + bool use_odirect, + int numa_node) + : _upstream(numa_node, true), + _host_mr(0, _upstream, pool_capacity, pool_capacity, block_size, 128, 1), + _reactor_ctx(std::make_shared( + uring::uring_reactor::reactor_config_type{.bounce_size = _host_mr.get_block_size(), + .use_odirect = use_odirect}, + &_host_mr)), + _io_ctx(std::make_shared(n_reactors, _reactor_ctx)) +{ + _io_ctx->start(); +} + +uring_datasource_engine::~uring_datasource_engine() { _io_ctx->shutdown(); } + +std::unique_ptr uring_datasource_engine::open(std::string path) const +{ + return open_datasource(_io_ctx, std::move(path)); +} + +} // namespace cucascade::io