Start async device writes and default host reads eagerly instead of deferring I/O to future wait#23231
Open
Emillock wants to merge 2 commits into
Open
Start async device writes and default host reads eagerly instead of deferring I/O to future wait#23231Emillock wants to merge 2 commits into
Emillock wants to merge 2 commits into
Conversation
The kvikio pwrite() call was placed inside a std::launch::deferred lambda, so the write was not initiated until the returned future was waited on. The Parquet and ORC writers fan out one future per column chunk / stripe stream and then wait on the futures in order, which serialized all device writes: each write began only after the previous one had fully completed. Introduced in rapidsai#10593 as an opt-in backend, this became the default path in rapidsai#12574 and the only path in rapidsai#17764. Initiate the write at call time via a lambda capture-initializer and defer only the wait, so the wrapper keeps converting kvikio's future<size_t> to future<void> without delaying the I/O. The eager stream.synchronize() above already guarantees the source buffer is ready, offsets are reserved before submission, and the buffers outlive the futures, so eager initiation is safe. Benchmark (isolated sink phase, replaying the 228 column-chunk sizes of a 12M-row NYC-taxi Parquet write, 243 MB total, RTX 5070, ext4): - 4 kvikio threads: 0.098 s -> 0.081 s per write phase (1.21x) - 8 kvikio threads: 0.102 s -> 0.080 s per write phase (1.28x) A pattern-level test with 4 x 100 ms simulated writes shows 0/4 writes in flight before the wait loop without the fix and 4/4 with it (460 ms -> 110 ms). Gains scale with per-write latency and chunk count; page-cache-backed local writes are the lower bound.
The base-class host_read_async implementations wrapped the synchronous host_read() in std::async(std::launch::deferred), so the read only executed when the returned future was waited on, in the waiting thread. The ORC reader fans out one host_read_async per stripe data range and then waits on the futures in order (reader_impl_chunking.cu), so for any datasource that does not override host_read_async -- notably user-provided datasources such as Python/fsspec-backed remote sources -- the reads ran strictly serially, one full read latency after another. Submit the read to cudf::detail::host_worker_pool() instead, so fanned-out reads execute in parallel. Concurrent host_read() calls are already required of every datasource by the multi-threaded Parquet reader, which submits host_read() to the same pool from multiple threads (parquet_io_utils.cpp), and the hierarchical pool design rules out nested-wait deadlocks. Built-in kvikio-backed sources override host_read_async and keep their native async paths. Benchmark (ORC-shaped fan-out of 20 x 2 MiB reads from a real Parquet file through a datasource without a host_read_async override, with cudf's hierarchical_thread_pool logic compiled into the harness, 8 workers, 12-core machine; per-read latency injected to model the source type): - local page cache (no latency): 8 ms -> 7 ms (1.05-1.14x) - 100 us/read (Python overhead): 11 ms -> 7 ms (1.63x) - 500 us/read (NFS-like): 20 ms -> 7 ms (2.88x) - 2 ms/read (remote-like): 52 ms -> 12 ms (4.42x) Read results validated by checksum against the serial path.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (2)
📝 WalkthroughSummary by CodeRabbit
WalkthroughChangesThe PR updates asynchronous I/O scheduling: file sink writes begin before deferred waiting, while datasource host reads run through the shared host worker pool. Asynchronous I/O behavior
Estimated code review effort: 2 (Simple) | ~10 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Fixes two instances of the same defect class in
cpp/src/io/utilities: an "async" API wraps the initiation of I/O insidestd::async(std::launch::deferred, ...), so the I/O does not start until the returned future is waited on. Since the writers/readers fan out these futures and then wait on them in order, the I/O executes strictly serially — one operation's full latency after another — instead of in parallel as the calling code intends.1.
file_sink::device_write_async(data_sink.cpp)The kvikio
pwrite()call was placed inside the deferred lambda, so the write began only at.get(). The Parquet writer fans out one future per column chunk per rowgroup (write_parquet_data_to_sink) and the ORC writer one per stripe stream (write_data_stream), then both wait in order — serializing every device write in every file written through kvikio. Introduced in #10593 as an opt-in backend (the then-default cufile path initiated eagerly via its thread pool), this became the default path in #12574 and the only path when cufile was removed in #17764.The fix initiates the write in a lambda capture-initializer (evaluated at call time) and defers only the wait, preserving the original purpose of the wrapper — converting kvikio's
future<size_t>tofuture<void>. Safety: the existing eagerstream.synchronize()guarantees source data is ready beforepwritereads it, offsets are reserved at submission time, writes target disjoint ranges (POSIXpwritesemantics, which kvikio already relies on internally), and the device buffers outlive the futures, which are all waited in the same scope.2. Default
datasource::host_read_async(datasource.cpp)The base-class default wrapped the synchronous
host_read()inlaunch::deferred. The ORC reader fans out onehost_read_asyncper stripe data range (reader_impl_chunking.cu) — so for any datasource that does not override the async API (user-provided datasources; built-in kvikio sources override it), the reads ran serially. The fix submits the read tocudf::detail::host_worker_pool(). Concurrenthost_read()calls are already required of every datasource by the multi-threaded Parquet reader, which submitshost_read()to this same pool from multiple threads (parquet_io_utils.cpp), and the hierarchical pool design precludes nested-wait deadlocks. The documented contract ("the read operation may be deferred") permits eager execution.Benchmarks
Measured with isolated harnesses running the verbatim before/after implementations against real files (a 12M-row NYC-taxi table; write side replays the 228 real column-chunk sizes of its Parquet output; read side the 24 x 12 MB stripe layout of its ORC output), with kvikio-equivalent staging (4 MiB task splitting, pinned-bounce D2H) and cudf's
hierarchical_thread_poollogic compiled in. Read results checksum-validated against the serial path. 12-core machine, RTX 5070, ext4.Write sink phase (243 MB, 228 chunks):
Read fan-out phase (20 x 2 MiB, 8 pool workers, per-read latency modeling source type):
Both gains scale with per-write/read latency and operation count; page-cache-backed local I/O is the lower bound. A pattern-level test of the write fix (4 x 100 ms simulated writes) shows 0/4 writes in flight before the wait loop on the old code vs 4/4 on the new (460 ms -> 110 ms). I was unable to run the nvbench suites locally (no libcudf build environment on this machine) — a
PARQUET_WRITER_NVBENCH/ ORC reader run on RAPIDS hardware to confirm end-to-end numbers would be welcome.Checklist
cpp/tests/iosink/source and Parquet/ORC writer-reader tests exercise both paths).host_read_asyncdoc contract already states reads "may be deferred"; eager execution remains within it).