Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions include/cucascade/io/datasource_factory.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#include <cucascade/io/config.hpp>
#include <cucascade/io/io_context.hpp>

#include <atomic>
#include <functional>
#include <memory>
#include <shared_mutex>
Expand Down Expand Up @@ -80,6 +81,39 @@ class io_context_registry {
*/
void register_ioctx(io_context_type type, scheme_checker_type checker, factory_type factory);

/**
* @brief Replaces one backend registration before path routing begins.
*
* A URI scheme may have more than one backend implementation, but only one
* of them should claim that scheme explicitly. For example, the registry
* initially assigns `s3://` paths to @c io_context_type::restful; when
* S3-over-RDMA is selected, the caller replaces that registration with
* @c io_context_type::s3rdma.
*
* The checker and factory for @p new_type are installed and the @p old_type
* entry is removed while holding the registry's exclusive lock, so no lookup
* observes an intermediate routing state.
*
* This operation changes registration metadata only. It does not construct,
* shut down, or migrate any ioctx instance.
*
* @param old_type Registered backend to remove.
* @param new_type Backend type to register.
* @param checker Predicate used to claim paths for @p new_type.
* @param factory Factory used by @c make_ioctx for @p new_type.
*
* @throws std::invalid_argument if @p old_type is absent, @p new_type is
* already registered, or @p checker or @p factory is empty.
* @throws std::logic_error if @c lookup_path has already been called.
*
* Provides the strong exception guarantee: the registry is unchanged if the
* replacement fails. Intended for single-threaded bootstrap.
*/
void replace_ioctx(io_context_type old_type,
io_context_type new_type,
scheme_checker_type checker,
factory_type factory);

/// Resolve the backend for a full @p path (not a bare scheme — the checkers
/// parse the URI / stat the filesystem themselves). Explicit backends
/// (uring / restful) take precedence over the kvikio catch-all, so `s3://`
Expand All @@ -105,6 +139,9 @@ class io_context_registry {
cucascade::memory::memory_reservation_manager& _reservation_manager;
mutable std::shared_mutex _mtx;
std::unordered_map<io_context_type, entry> _entries;
/// Set by the first @c lookup_path; @c replace_ioctx refuses afterwards
/// (bootstrap-only — see its contract).
mutable std::atomic<bool> _lookup_latched{false};
};

// ---------------------------------------------------------------------------
Expand Down
2 changes: 1 addition & 1 deletion include/cucascade/io/io_context.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@

namespace cucascade::io {

enum class io_context_type { uring, restful, kvikio };
enum class io_context_type { uring, restful, kvikio, s3rdma };

/// Hint passed to @c open_io_object so a backend can tailor how it resolves an
/// object's metadata. @c generic resolves the size however is cheapest for the
Expand Down
57 changes: 57 additions & 0 deletions include/cucascade/io/object_store_listing.hpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

#pragma once

// The page/entry types live under rest/s3 but are S3-PROTOCOL shapes
// (ListObjectsV2 responses), not REST-transport shapes — any backend that
// lists an S3-compatible store speaks them, whatever its data plane.
#include <cucascade/io/rest/s3/list_parser.hpp>

#include <cstddef>
#include <functional>
#include <optional>
#include <string_view>

namespace cucascade::io {

/// Listing capability of an object-store backend. A glob / LIST layer
/// depends on this interface, not a concrete ioctx type. Listing is prefix
/// resolution on the control plane; it is independent of how the data plane
/// reads the resolved keys.
class object_store_listing {
public:
virtual ~object_store_listing() = default;

/// Stream ListObjectsV2 pages under @p prefix to @p sink, one call per
/// page. @p sink returns false to stop early. @p page_size is clamped
/// to [1,1000] (0 and >1000 mean 1000). Throws (never truncates) on a
/// truncated page without a continuation token, and once more than
/// @p max_scanned entries have been scanned across pages.
virtual void list_objects_paged(
std::string_view bucket,
std::string_view prefix,
std::size_t page_size,
std::function<bool(rest::s3::list_objects_v2_page const&)> const& sink,
std::optional<std::size_t> max_scanned = std::nullopt) = 0;

/// The backend's configured matched cap for glob resolution.
[[nodiscard]] virtual std::size_t list_max_matches() const = 0;
};

} // namespace cucascade::io
7 changes: 4 additions & 3 deletions include/cucascade/io/rest/rest_ioctx.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

#pragma once

#include <cucascade/io/object_store_listing.hpp>
#include <cucascade/io/rest/rest_reactor.hpp>
#include <cucascade/io/rest/s3/list_parser.hpp>
#include <cucascade/io/templated_ioctx.hpp>
Expand Down Expand Up @@ -46,7 +47,7 @@ namespace cucascade::io::rest {
* via a blocking HEAD before constructing the @c rest_io_object — the static
* reactor factory cannot do this since it needs the authorizer + a round-trip.
*/
class rest_ioctx : public templated_ioctx<rest_reactor> {
class rest_ioctx : public templated_ioctx<rest_reactor>, public object_store_listing {
public:
/// Build a pool of @p n_reactors reactors, all sharing @p ctx (one context per
/// pool: it carries the per-reactor @c config, the presigning authorizer, and
Expand Down Expand Up @@ -74,7 +75,7 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
std::string_view prefix,
std::size_t page_size,
std::function<bool(s3::list_objects_v2_page const&)> const& sink,
std::optional<std::size_t> max_scanned = std::nullopt);
std::optional<std::size_t> max_scanned = std::nullopt) override;

/// Whole-listing convenience over @c list_objects_paged: every object under
/// @p prefix, in document order, with sizes. Throws (never truncates) when
Expand All @@ -90,7 +91,7 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
/// glob layer one level up can bound its match set without a reactor handle.
/// Falls back to the built-in default when the pool is empty (never in
/// practice).
[[nodiscard]] std::size_t list_max_matches() const;
[[nodiscard]] std::size_t list_max_matches() const override;

protected:
/// Backend hook invoked by @c ioctx::open_io_object: parse @p path
Expand Down
111 changes: 111 additions & 0 deletions include/cucascade/io/s3rdma/s3rdma_ioctx.hpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

#pragma once

#include <cucascade/exec/semi_future.hpp>
#include <cucascade/io/io_context.hpp>
#include <cucascade/io/types.hpp>

#include <rmm/cuda_stream_view.hpp>

#include <cstddef>
#include <cstdint>
#include <memory>
#include <optional>
#include <span>
#include <string>
#include <string_view>
#include <type_traits>
#include <vector>

namespace cucascade::io::s3rdma {

// ---------------------------------------------------------------------------
// s3rdma_ioctx (placeholder)
// ---------------------------------------------------------------------------

/**
* @brief Non-constructible placeholder for the S3-over-RDMA I/O context.
*
* The planned backend uses RDMA for S3 range-read payloads. Data lands in
* registered GPU memory and is copied on device to the caller's destination.
* LIST and HEAD use the configured HTTP(S) control plane; glob resolution
* uses LIST results.
*
* Integration uses @c io_context_type::s3rdma,
* @c io_context_registry::replace_ioctx, @c object_store_listing, and
* @c templated_ioctx::on_device_dispatch_failure.
*
* This declaration has no implementation or factory registration. Its deleted
* constructor prevents accidental use.
*
* @see https://github.com/sirius-db/sirius/blob/dev/experimental/s3-rdma-transport-design.md
*/
class s3rdma_ioctx : public ioctx {
public:
s3rdma_ioctx() = delete;

[[nodiscard]] io_context_type type() const noexcept override;

void shutdown() noexcept override;

[[nodiscard]] bool supports(std::string_view path) const noexcept override;

[[nodiscard]] bool supports_device_read() const noexcept override;
[[nodiscard]] bool supports_host_to_device_read() const noexcept override;
[[nodiscard]] bool supports_vector_host_read() const noexcept override;
[[nodiscard]] cache::prefetching_stage preferred_prefetching_stage() const noexcept override;

[[nodiscard]] std::vector<byte_range> align_and_coalesce(
std::span<const byte_range> ranges,
std::optional<size_t> alignment = std::nullopt) const noexcept override;

size_t host_read_io(const io_object& obj, size_t offset, size_t size, uint8_t* dst) override;

exec::semi_future<size_t> host_read_async_io(const io_object& obj,
size_t offset,
size_t size,
uint8_t* dst) noexcept override;

exec::semi_future<size_t> device_read_async_io(const io_object& obj,
size_t offset,
size_t size,
uint8_t* dst,
rmm::cuda_stream_view stream) noexcept override;

exec::semi_future<size_t> host_to_device_read_async_io(
const io_object& obj,
std::span<io_object_segment> slices,
size_t offset,
size_t size,
uint8_t* dst,
rmm::cuda_stream_view stream) noexcept override;

exec::semi_future<size_t> host_read_ranges_async_io(
const io_object& obj, std::span<io_object_segment> segments) noexcept override;

protected:
std::shared_ptr<io_object> create_io_object(std::string path) override;
};

static_assert(!std::is_default_constructible_v<s3rdma_ioctx>,
"s3rdma_ioctx is a placeholder and must stay non-constructible "
"until the backend implementation lands");

} // namespace cucascade::io::s3rdma
22 changes: 22 additions & 0 deletions include/cucascade/io/templated_ioctx.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -380,6 +380,7 @@ class templated_ioctx : public ioctx {
});
return semi;
} catch (...) {
on_device_dispatch_failure();
return exec::make_semi_future<size_t>(std::current_exception());
}
} else {
Expand All @@ -388,6 +389,26 @@ class templated_ioctx : public ioctx {
}
}

protected:
/**
* @brief Applies backend policy after synchronous device dispatch fails.
*
* Called from the exception handlers in device_read_async_io() and
* host_to_device_read_async_io(), before the exception is returned through
* an errored future.
*
* An S3-over-RDMA backend overrides this hook to check for a sticky CUDA
* context error. Returning such an error as an ordinary request failure
* could allow registered GPU memory to be reused or released before RDMA
* writes and CUDA work are known to be quiescent. In that case the backend
* must invoke its fatal policy instead of returning.
*
* The default implementation does nothing. An override must not throw or
* re-enter this ioctx.
*/
virtual void on_device_dispatch_failure() noexcept {}

public:
exec::semi_future<size_t> host_to_device_read_async_io(
const io_object& obj,
std::span<io_object_segment> slices,
Expand Down Expand Up @@ -417,6 +438,7 @@ class templated_ioctx : public ioctx {
});
return semi;
} catch (...) {
on_device_dispatch_failure();
return exec::make_semi_future<size_t>(std::current_exception());
}
} else {
Expand Down
1 change: 1 addition & 0 deletions src/io/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ target_sources(
${CMAKE_CURRENT_SOURCE_DIR}/rest/curl_handle.cpp
${CMAKE_CURRENT_SOURCE_DIR}/rest/rest_ioctx.cpp
${CMAKE_CURRENT_SOURCE_DIR}/rest/rest_reactor.cpp
${CMAKE_CURRENT_SOURCE_DIR}/s3rdma/s3rdma_ioctx.cpp
${CMAKE_CURRENT_SOURCE_DIR}/uring/uring_ioctx.cpp
${CMAKE_CURRENT_SOURCE_DIR}/uring/uring_reactor.cpp
${CMAKE_CURRENT_SOURCE_DIR}/kvikio/kvikio_context.cpp
Expand Down
27 changes: 27 additions & 0 deletions src/io/datasource_factory.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -185,10 +185,37 @@ void io_context_registry::register_ioctx(io_context_type type,
_entries[type] = {type, std::move(checker), std::move(factory)};
}

void io_context_registry::replace_ioctx(io_context_type old_type,
io_context_type new_type,
scheme_checker_type checker,
factory_type factory)
{
if (!checker) {
throw std::invalid_argument("datasource_registry: replace_ioctx: null scheme checker");
}
if (!factory) { throw std::invalid_argument("datasource_registry: replace_ioctx: null factory"); }
std::lock_guard lk{_mtx};
if (_lookup_latched.load(std::memory_order_acquire)) {
throw std::logic_error(
"datasource_registry: replace_ioctx after the first lookup_path (bootstrap-only)");
}
if (!_entries.contains(old_type)) {
throw std::invalid_argument("datasource_registry: replace_ioctx: old type not registered");
}
if (_entries.contains(new_type)) {
throw std::invalid_argument("datasource_registry: replace_ioctx: new type already registered");
}
// Strong guarantee: the emplace is the only throwing step and precedes the
// erase; erase by KEY, not by a pre-emplace iterator (emplace may rehash).
_entries.emplace(new_type, entry{new_type, std::move(checker), std::move(factory)});
_entries.erase(old_type);
}

std::optional<io_context_type> io_context_registry::lookup_path(
std::string_view path) const noexcept
{
std::shared_lock lk{_mtx};
_lookup_latched.store(true, std::memory_order_release);
// kvikio's checker matches everything; _entries iterates in unspecified order,
// so defer the catch-all and let an explicit backend (uring/restful) win.
std::optional<io_context_type> fallback;
Expand Down
25 changes: 25 additions & 0 deletions src/io/s3rdma/s3rdma_ioctx.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

// Placeholder translation unit: compiles the s3rdma_ioctx declaration so CI
// verifies the header, and is replaced wholesale together with it when the
// S3-over-RDMA backend is contributed upstream (see the header's class doc).
// The class is deliberately non-constructible until then — no definitions
// here.

#include <cucascade/io/s3rdma/s3rdma_ioctx.hpp>
3 changes: 3 additions & 0 deletions test/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,12 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY AND CUCASCADE_BUILD_IO)
# IO test executable - links the cudf-free cucascade-io datasource layer.
add_executable(
cucascade_io_tests
io/test_datasource_registry.cpp
io/test_dispatch_failure_hook.cpp
io/test_uri_parser.cpp
io/cache/test_metadata_store.cpp
io/kvikio/test_kvikio_config.cpp
io/rest/test_object_store_listing.cpp
io/rest/test_rest_perf_snapshot.cpp
io/rest/test_rest_validation_tag.cpp
io/rest/test_shared_byte_span.cpp
Expand Down
Loading