From 605fe3f05d76ddc36323c0a272dd8a905b8a0268 Mon Sep 17 00:00:00 2001 From: Kevin Kristensen Date: Wed, 12 Aug 2026 21:00:33 +0000 Subject: [PATCH 1/3] fix(cudf): order GPU reads after writers --- .../cudf/gpu_data_representation.hpp | 24 +++--- src/cudf/gpu_data_representation.cpp | 46 +++++++++--- .../representation_converter_builtins.cpp | 73 +++++++------------ 3 files changed, 77 insertions(+), 66 deletions(-) diff --git a/include/cucascade/cudf/gpu_data_representation.hpp b/include/cucascade/cudf/gpu_data_representation.hpp index b6d463a..33b83fd 100644 --- a/include/cucascade/cudf/gpu_data_representation.hpp +++ b/include/cucascade/cudf/gpu_data_representation.hpp @@ -118,13 +118,15 @@ class gpu_table_representation : public idata_representation { std::size_t get_uncompressed_data_size_in_bytes() const override; /** - * @brief Create a deep copy of this GPU table representation. + * @brief Create an independently owned copy of this GPU table * - * The cloned representation will have its own copy of the underlying cuDF table, - * residing in the same memory space as the original. + * Orders the copy after the recorded writer event, or synchronizes the source device if no event + * is available. The copy uses this memory space's default allocator on @p stream, and the method + * synchronizes @p stream before returning. A stream with a non-null handle is recorded as the + * result's writer stream. * - * @param stream CUDA stream for memory operations - * @return std::unique_ptr A new gpu_table_representation with copied data + * @param stream Stream on this representation's device used for the copy + * @return Independently owned copy in the same memory space */ std::unique_ptr clone(rmm::cuda_stream_view stream) override; @@ -136,12 +138,16 @@ class gpu_table_representation : public idata_representation { cudf::table_view get_table_view() const; /** - * @brief Release ownership of the underlying cuDF table + * @brief Move out an owned cuDF table or materialize a table view * - * After calling this method, this representation no longer owns the table. + * An owned table is moved out without synchronization; the caller must order subsequent access + * after any outstanding writer work. A view-backed table is copied on @p stream using this memory + * space's default allocator after the recorded writer event, or after synchronizing the source + * device when no event exists. The method synchronizes @p stream before releasing the external + * owner. In either case, this representation is left without a table. * - * @param stream CUDA stream (used to materialize the table from a view path before release) - * @return std::unique_ptr The cuDF table + * @param stream Stream on this representation's device used only for view materialization + * @return Moved table or independently owned materialization of the table view */ std::unique_ptr release_table(rmm::cuda_stream_view stream); diff --git a/src/cudf/gpu_data_representation.cpp b/src/cudf/gpu_data_representation.cpp index d56985a..9ae41dd 100644 --- a/src/cudf/gpu_data_representation.cpp +++ b/src/cudf/gpu_data_representation.cpp @@ -23,6 +23,8 @@ #include #include +#include + namespace cucascade { gpu_table_representation::gpu_table_representation(std::unique_ptr table, @@ -73,11 +75,25 @@ cudf::table_view gpu_table_representation::get_table_view() const } } -std::unique_ptr gpu_table_representation::release_table( - [[maybe_unused]] rmm::cuda_stream_view stream) +std::unique_ptr gpu_table_representation::release_table(rmm::cuda_stream_view stream) { if (std::holds_alternative(_table)) { - _table = std::make_unique(std::get(_table).view, stream); + rmm::cuda_set_device_raii device_guard{rmm::cuda_device_id{get_device_id()}}; + // Wait for the latest writer before materializing the view. Eventless representations require + // a source-device synchronization. + if (_writer_event != nullptr) { + cucascade::cuda::cuda_event_view{_writer_event}.wait(stream); + } else { + CUCASCADE_CUDA_TRY(cudaDeviceSynchronize()); + } + + auto materialized = std::make_unique( + std::get(_table).view, stream, get_memory_space().get_default_allocator()); + // cuDF enqueues the deep copy asynchronously. Replacing the variant destroys the external + // owner, so the materialization stream must finish reading the view before that owner can + // release its source buffers. + stream.synchronize(); + _table = std::move(materialized); } return std::move(std::get>(_table)); } @@ -102,14 +118,22 @@ void gpu_table_representation::rebind_stream(rmm::cuda_stream_view stream) std::unique_ptr gpu_table_representation::clone(rmm::cuda_stream_view stream) { - // Create a deep copy of the cuDF table using the provided stream. - // STREAM-LINEAGE: the clone has been written by `stream`; record an event on - // it so any cross-stream/cross-device reader of the clone honors the - // producer-consumer ordering established by record_writer_event(). - cudf::table_view view = get_table_view(); - auto cloned = std::make_unique( - std::make_unique(view, stream), get_memory_space(), stream); - return cloned; + rmm::cuda_set_device_raii device_guard{rmm::cuda_device_id{get_device_id()}}; + // Wait for the latest writer before copying the source. Eventless representations require a + // source-device synchronization. + if (_writer_event != nullptr) { + cucascade::cuda::cuda_event_view{_writer_event}.wait(stream); + } else { + CUCASCADE_CUDA_TRY(cudaDeviceSynchronize()); + } + + auto cloned_table = std::make_unique( + get_table_view(), stream, get_memory_space().get_default_allocator()); + // The source may be destroyed as soon as clone() returns, so finish all asynchronous reads from + // it before publishing the independently owned result. + stream.synchronize(); + return std::make_unique( + std::move(cloned_table), get_memory_space(), stream); } void gpu_table_representation::record_writer_event(rmm::cuda_stream_view writer_stream) diff --git a/src/cudf/representation_converter_builtins.cpp b/src/cudf/representation_converter_builtins.cpp index aab773b..e7ce39c 100644 --- a/src/cudf/representation_converter_builtins.cpp +++ b/src/cudf/representation_converter_builtins.cpp @@ -44,6 +44,7 @@ #include #include +#include #include #include #include @@ -81,6 +82,22 @@ inline cudf::type_id as_cudf_type_id(int32_t type_id) return static_cast(type_id); } +// Orders `source_read_stream` after the source's latest recorded writer. An event-backed wait is +// asynchronous and does not extend source lifetime; callers must retain the source until their +// reads complete. If no event is recorded, this function synchronizes the source device before +// returning. The device associated with `source_read_stream` must be current on entry. +void wait_for_gpu_source(gpu_table_representation const& source, + rmm::cuda_stream_view source_read_stream) +{ + if (auto const writer_event = source.get_writer_event(); writer_event != nullptr) { + cuda::cuda_event_view{writer_event}.wait(source_read_stream); + return; + } + + rmm::cuda_set_device_raii source_device_guard{rmm::cuda_device_id{source.get_device_id()}}; + CUCASCADE_CUDA_TRY(cudaDeviceSynchronize()); +} + // Forward declaration. convert_gpu_to_gpu is defined below convert_gpu_to_host_fast // so it can reuse BatchCopyAccumulator and the column-tree reconstruction helpers, // peer-copying each column buffer directly and avoiding cudf::pack (whose internal @@ -101,11 +118,9 @@ std::unique_ptr convert_gpu_to_host( rmm::cuda_stream_view stream, memory::reservation* reservation) { - // Synchronize the stream to ensure any prior operations (like table creation) - // are complete before we read from the source table - stream.synchronize(); - auto& gpu_source = source.cast(); + rmm::cuda_set_device_raii source_device_guard{rmm::cuda_device_id{source.get_device_id()}}; + wait_for_gpu_source(gpu_source, stream); auto packed_data = cudf::pack(gpu_source.get_table_view(), stream); auto mr = target_memory_space->get_memory_resource_as(); @@ -497,7 +512,9 @@ std::unique_ptr convert_gpu_to_host_fast( rmm::cuda_stream_view stream, memory::reservation* reservation) { - auto& gpu_source = source.cast(); + auto& gpu_source = source.cast(); + rmm::cuda_set_device_raii source_device_guard{rmm::cuda_device_id{source.get_device_id()}}; + wait_for_gpu_source(gpu_source, stream); const cudf::table_view view = gpu_source.get_table_view(); // --- Pass 1: plan the allocation layout --- @@ -866,11 +883,6 @@ std::unique_ptr convert_gpu_to_gpu( rmm::cuda_stream_view stream, [[maybe_unused]] memory::reservation* reservation) { - // Sync the caller's stream so the source table's buffers are stable on the source - // device before we issue peer copies. The caller's stream is the one that produced - // (or last touched) the source representation. - stream.synchronize(); - auto& gpu_source = source.cast(); // Same-device case: clone via source's own clone() method. @@ -881,25 +893,6 @@ std::unique_ptr convert_gpu_to_gpu( auto const src_device_id = gpu_source.get_device_id(); auto const dst_device_id = target_memory_space->get_device_id(); - // STREAM-LINEAGE INVARIANT: cross-device peer copies of cudaMallocAsync - // allocations require explicit event-ordered synchronization with the - // writer stream. A source-device-wide cudaDeviceSynchronize() does NOT - // establish the cross-mempool visibility the driver needs — under - // compute-sanitizer this site emits hundreds of stream-ordered-race errors - // even with a brute-force device sync. Producer-consumer pairing: - // producer = the stream that wrote gpu_source (recorded via - // gpu_table_representation::record_writer_event) - // consumer = target_stream (acquired from target memory space below) - // We resolve this in two passes: - // 1) Wait on the writer event (if recorded) on the *target* stream so the - // reader sees the writer's allocation/copy ordering. This is the precise - // primitive the sanitizer recognizes as closing the race. - // 2) Keep the source-device cudaDeviceSynchronize() as defense-in-depth for - // callers that have not yet been migrated to record writer events - // (get_writer_event() == nullptr). When the writer event is set the - // cudaDeviceSynchronize is technically redundant but harmless. - cudaEvent_t const writer_event = gpu_source.get_writer_event(); - rmm::cuda_set_device_raii target_guard{rmm::cuda_device_id{dst_device_id}}; // Target-bound stream from the target memory_space's stream pool. All peer copies @@ -907,21 +900,7 @@ std::unique_ptr convert_gpu_to_gpu( // completion without explicit cross-stream events. auto target_stream = target_memory_space->acquire_stream(); auto mr = target_memory_space->get_default_allocator(); - - if (writer_event != nullptr) { - // STREAM-LINEAGE pass 1: tie the reader stream's timeline to the writer's - // recorded event. After this point the target_stream observes all - // writer-side cudaMallocAsync allocations and writes in proper order. - cucascade::cuda::cuda_event_view{writer_event}.wait(target_stream); - } else { - // STREAM-LINEAGE pass 2 (fallback): no writer event recorded — fall back to - // a coarser source-device sync. This path is documented as insufficient for - // cross-mempool cudaMallocAsync allocations but is preserved for - // representations produced by code paths that have not yet been migrated to - // record_writer_event(). - rmm::cuda_set_device_raii src_sync_guard{rmm::cuda_device_id{src_device_id}}; - CUCASCADE_CUDA_TRY(cudaDeviceSynchronize()); - } + wait_for_gpu_source(gpu_source, target_stream); cudf::table_view const src_view = gpu_source.get_table_view(); @@ -1614,8 +1593,10 @@ static std::unique_ptr convert_gpu_to_disk( rmm::cuda_stream_view stream, [[maybe_unused]] memory::reservation* reservation) { - auto& backend = target_memory_space->get_io_backend(); - auto& gpu_source = source.cast(); + auto& backend = target_memory_space->get_io_backend(); + auto& gpu_source = source.cast(); + rmm::cuda_set_device_raii source_device_guard{rmm::cuda_device_id{source.get_device_id()}}; + wait_for_gpu_source(gpu_source, stream); cudf::table_view tv = gpu_source.get_table_view(); // Generate unique file path under the disk memory space's mount directory From c5bac4c3c7b4c976c4a6a855f0d40452187601da Mon Sep 17 00:00:00 2001 From: Kevin Kristensen Date: Thu, 13 Aug 2026 14:43:17 +0000 Subject: [PATCH 2/3] test(cudf): cover cross-stream clone ordering --- test/data/test_data_representation.cpp | 156 ++++++++++++++++++ .../test_reservation_manager_configurator.cpp | 2 +- 2 files changed, 157 insertions(+), 1 deletion(-) diff --git a/test/data/test_data_representation.cpp b/test/data/test_data_representation.cpp index d263953..ac5122e 100644 --- a/test/data/test_data_representation.cpp +++ b/test/data/test_data_representation.cpp @@ -46,8 +46,13 @@ #include +#include #include +#include +#include +#include #include +#include #include #include #include @@ -639,6 +644,70 @@ TEST_CASE("Representations polymorphism", // Clone Tests // ============================================================================= +namespace { + +/** + * @brief Deterministically hold a CUDA stream inside a host callback until released. + * + * The callback only uses C++ atomics; CUDA APIs are forbidden from CUDA host callbacks. + */ +class cuda_stream_gate { + public: + cuda_stream_gate() = default; + cuda_stream_gate(cuda_stream_gate const&) = delete; + cuda_stream_gate& operator=(cuda_stream_gate const&) = delete; + + ~cuda_stream_gate() + { + release(); + if (_enqueued) { _exited.wait(false, std::memory_order_acquire); } + } + + void enqueue(rmm::cuda_stream_view stream) + { + CUCASCADE_CUDA_TRY(cudaLaunchHostFunc(stream.value(), &cuda_stream_gate::wait, this)); + _enqueued = true; + } + + void wait_until_entered() const { _entered.wait(false, std::memory_order_acquire); } + + void release() noexcept + { + _released.store(true, std::memory_order_release); + _released.notify_all(); + } + + private: + static void CUDART_CB wait(void* data) + { + auto& gate = *static_cast(data); + gate._entered.store(true, std::memory_order_release); + gate._entered.notify_all(); + gate._released.wait(false, std::memory_order_acquire); + gate._exited.store(true, std::memory_order_release); + gate._exited.notify_all(); + } + + bool _enqueued{false}; + mutable std::atomic _entered{false}; + std::atomic _released{false}; + std::atomic _exited{false}; +}; + +class scoped_stream_gate_release { + public: + explicit scoped_stream_gate_release(cuda_stream_gate& gate) : _gate(gate) {} + ~scoped_stream_gate_release() { _gate.release(); } + + scoped_stream_gate_release(scoped_stream_gate_release const&) = delete; + scoped_stream_gate_release& operator=(scoped_stream_gate_release const&) = delete; + + private: + cuda_stream_gate& _gate; +}; + +} // namespace + TEST_CASE("gpu_table_representation clone creates independent copy", "[gpu_data_representation]") { auto gpu_space = make_mock_memory_space(memory::Tier::GPU, 0); @@ -674,6 +743,93 @@ TEST_CASE("gpu_table_representation clone creates independent copy", "[gpu_data_ } } +TEST_CASE("gpu_table_representation clone waits for a distinct writer stream", + "[gpu_data_representation][stream_ordering]") +{ + using namespace std::chrono_literals; + + memory::gpu_memory_space_config config; + config.device_id = 0; + config.memory_capacity = 64ULL << 20; + config.mr_factory_fn = test::make_shared_current_device_resource; + auto gpu_space = std::make_shared(config); + CUCASCADE_CUDA_TRY(cudaSetDevice(config.device_id)); + rmm::cuda_stream producer_stream; + rmm::cuda_stream consumer_stream; + + constexpr cudf::size_type num_rows = 1024; + constexpr std::size_t data_size = + static_cast(num_rows) * sizeof(std::int32_t); + constexpr unsigned char expected_byte = 0x5a; + + auto column = cudf::make_numeric_column(cudf::data_type{cudf::type_id::INT32}, + num_rows, + cudf::mask_state::UNALLOCATED, + consumer_stream.view(), + gpu_space->get_default_allocator()); + + // Establish a known stale value before deliberately blocking the real producer write. + CUCASCADE_CUDA_TRY(cudaMemsetAsync( + column->mutable_view().head(), 0, data_size, consumer_stream.value())); + consumer_stream.synchronize(); + + cuda_stream_gate producer_gate; + producer_gate.enqueue(producer_stream.view()); + CUCASCADE_CUDA_TRY(cudaMemsetAsync(column->mutable_view().head(), + expected_byte, + data_size, + producer_stream.value())); + + std::vector> columns; + columns.push_back(std::move(column)); + gpu_table_representation source(std::make_unique(std::move(columns)), + *gpu_space, + producer_stream.view()); + + // Make sure the producer cannot reach either the write or source's recorded writer event. + producer_gate.wait_until_entered(); + auto const writer_status_while_blocked = cudaEventQuery(source.get_writer_event()); + + std::atomic clone_started{false}; + std::future> clone_future; + scoped_stream_gate_release release_on_exit{producer_gate}; + clone_future = std::async(std::launch::async, [&] { + CUCASCADE_CUDA_TRY(cudaSetDevice(source.get_device_id())); + clone_started.store(true, std::memory_order_release); + clone_started.notify_all(); + return source.clone(consumer_stream.view()); + }); + clone_started.wait(false, std::memory_order_acquire); + + // The fixed implementation waits for source's writer event and cannot return while the + // producer is gated. Main queues the copy without that wait and returns immediately. + auto const status_while_writer_blocked = clone_future.wait_for(1s); + if (status_while_writer_blocked == std::future_status::ready) { + // On the buggy implementation, finish the premature copy while the source still contains the + // stale pattern. This turns the ordering failure into deterministic data corruption too. + consumer_stream.synchronize(); + } + + producer_gate.release(); + auto cloned_base = clone_future.get(); + producer_stream.synchronize(); + consumer_stream.synchronize(); + + REQUIRE(writer_status_while_blocked == cudaErrorNotReady); + CHECK(status_while_writer_blocked == std::future_status::timeout); + + auto* clone = dynamic_cast(cloned_base.get()); + REQUIRE(clone != nullptr); + + std::vector bytes(data_size); + CUCASCADE_CUDA_TRY(cudaMemcpy(bytes.data(), + clone->get_table_view().column(0).head(), + data_size, + cudaMemcpyDeviceToHost)); + REQUIRE(std::all_of( + bytes.cbegin(), bytes.cend(), [](uint8_t value) { return value == expected_byte; })); +} + TEST_CASE("gpu_table_representation clone empty table", "[gpu_data_representation]") { auto gpu_space = make_mock_memory_space(memory::Tier::GPU, 0); diff --git a/test/memory/test_reservation_manager_configurator.cpp b/test/memory/test_reservation_manager_configurator.cpp index fcb5001..f9bd1e2 100644 --- a/test/memory/test_reservation_manager_configurator.cpp +++ b/test/memory/test_reservation_manager_configurator.cpp @@ -26,7 +26,7 @@ #include #include -#include +#include #include #include From 160ca5f0d2021804619053d3be7d41fb66edf1e3 Mon Sep 17 00:00:00 2001 From: Kevin Kristensen Date: Thu, 13 Aug 2026 15:02:29 +0000 Subject: [PATCH 3/3] docs(cudf): streamline ordering comments --- src/cudf/gpu_data_representation.cpp | 6 ++-- test/data/test_data_representation.cpp | 39 +++++++++----------------- 2 files changed, 15 insertions(+), 30 deletions(-) diff --git a/src/cudf/gpu_data_representation.cpp b/src/cudf/gpu_data_representation.cpp index 9ae41dd..5c5f472 100644 --- a/src/cudf/gpu_data_representation.cpp +++ b/src/cudf/gpu_data_representation.cpp @@ -79,8 +79,7 @@ std::unique_ptr gpu_table_representation::release_table(rmm::cuda_s { if (std::holds_alternative(_table)) { rmm::cuda_set_device_raii device_guard{rmm::cuda_device_id{get_device_id()}}; - // Wait for the latest writer before materializing the view. Eventless representations require - // a source-device synchronization. + // Without an event, the producing stream is unknown and requires a device-wide fallback. if (_writer_event != nullptr) { cucascade::cuda::cuda_event_view{_writer_event}.wait(stream); } else { @@ -119,8 +118,7 @@ void gpu_table_representation::rebind_stream(rmm::cuda_stream_view stream) std::unique_ptr gpu_table_representation::clone(rmm::cuda_stream_view stream) { rmm::cuda_set_device_raii device_guard{rmm::cuda_device_id{get_device_id()}}; - // Wait for the latest writer before copying the source. Eventless representations require a - // source-device synchronization. + // Without an event, the producing stream is unknown and requires a device-wide fallback. if (_writer_event != nullptr) { cucascade::cuda::cuda_event_view{_writer_event}.wait(stream); } else { diff --git a/test/data/test_data_representation.cpp b/test/data/test_data_representation.cpp index ac5122e..1b2d72d 100644 --- a/test/data/test_data_representation.cpp +++ b/test/data/test_data_representation.cpp @@ -646,11 +646,7 @@ TEST_CASE("Representations polymorphism", namespace { -/** - * @brief Deterministically hold a CUDA stream inside a host callback until released. - * - * The callback only uses C++ atomics; CUDA APIs are forbidden from CUDA host callbacks. - */ +// CUDA host callbacks cannot call CUDA APIs, so use C++ atomics to gate the stream. class cuda_stream_gate { public: cuda_stream_gate() = default; @@ -757,9 +753,8 @@ TEST_CASE("gpu_table_representation clone waits for a distinct writer stream", rmm::cuda_stream producer_stream; rmm::cuda_stream consumer_stream; - constexpr cudf::size_type num_rows = 1024; - constexpr std::size_t data_size = - static_cast(num_rows) * sizeof(std::int32_t); + constexpr cudf::size_type num_rows = 1024; + constexpr std::size_t data_size = static_cast(num_rows) * sizeof(std::int32_t); constexpr unsigned char expected_byte = 0x5a; auto column = cudf::make_numeric_column(cudf::data_type{cudf::type_id::INT32}, @@ -769,24 +764,20 @@ TEST_CASE("gpu_table_representation clone waits for a distinct writer stream", gpu_space->get_default_allocator()); // Establish a known stale value before deliberately blocking the real producer write. - CUCASCADE_CUDA_TRY(cudaMemsetAsync( - column->mutable_view().head(), 0, data_size, consumer_stream.value())); + CUCASCADE_CUDA_TRY( + cudaMemsetAsync(column->mutable_view().head(), 0, data_size, consumer_stream.value())); consumer_stream.synchronize(); cuda_stream_gate producer_gate; producer_gate.enqueue(producer_stream.view()); - CUCASCADE_CUDA_TRY(cudaMemsetAsync(column->mutable_view().head(), - expected_byte, - data_size, - producer_stream.value())); + CUCASCADE_CUDA_TRY(cudaMemsetAsync( + column->mutable_view().head(), expected_byte, data_size, producer_stream.value())); std::vector> columns; columns.push_back(std::move(column)); - gpu_table_representation source(std::make_unique(std::move(columns)), - *gpu_space, - producer_stream.view()); + gpu_table_representation source( + std::make_unique(std::move(columns)), *gpu_space, producer_stream.view()); - // Make sure the producer cannot reach either the write or source's recorded writer event. producer_gate.wait_until_entered(); auto const writer_status_while_blocked = cudaEventQuery(source.get_writer_event()); @@ -801,12 +792,10 @@ TEST_CASE("gpu_table_representation clone waits for a distinct writer stream", }); clone_started.wait(false, std::memory_order_acquire); - // The fixed implementation waits for source's writer event and cannot return while the - // producer is gated. Main queues the copy without that wait and returns immediately. + // clone() must not return until the source writer event can complete. auto const status_while_writer_blocked = clone_future.wait_for(1s); if (status_while_writer_blocked == std::future_status::ready) { - // On the buggy implementation, finish the premature copy while the source still contains the - // stale pattern. This turns the ordering failure into deterministic data corruption too. + // Complete a premature copy before releasing the producer to make stale data deterministic. consumer_stream.synchronize(); } @@ -822,10 +811,8 @@ TEST_CASE("gpu_table_representation clone waits for a distinct writer stream", REQUIRE(clone != nullptr); std::vector bytes(data_size); - CUCASCADE_CUDA_TRY(cudaMemcpy(bytes.data(), - clone->get_table_view().column(0).head(), - data_size, - cudaMemcpyDeviceToHost)); + CUCASCADE_CUDA_TRY(cudaMemcpy( + bytes.data(), clone->get_table_view().column(0).head(), data_size, cudaMemcpyDeviceToHost)); REQUIRE(std::all_of( bytes.cbegin(), bytes.cend(), [](uint8_t value) { return value == expected_byte; })); }