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..5c5f472 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,24 @@ 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()}}; + // 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 { + 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 +117,21 @@ 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()}}; + // 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 { + 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 diff --git a/test/data/test_data_representation.cpp b/test/data/test_data_representation.cpp index d263953..1b2d72d 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,66 @@ TEST_CASE("Representations polymorphism", // Clone Tests // ============================================================================= +namespace { + +// CUDA host callbacks cannot call CUDA APIs, so use C++ atomics to gate the stream. +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 +739,84 @@ 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()); + + 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); + + // 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) { + // Complete a premature copy before releasing the producer to make stale data deterministic. + 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