diff --git a/native/csrc/ring/producer.cu b/native/csrc/ring/producer.cu index 928a96250..ebe68ba72 100644 --- a/native/csrc/ring/producer.cu +++ b/native/csrc/ring/producer.cu @@ -656,13 +656,6 @@ __global__ void record_producer_segmented_pack_kernel( // Host launchers // --------------------------------------------------------------------------- -static inline uint32_t pick_tier_blocks(uint64_t nbytes) { - if (nbytes <= TIER1_THRESHOLD) return TIER0_BLOCKS; - else if (nbytes <= TIER2_THRESHOLD) return TIER1_BLOCKS; - else if (nbytes <= TIER3_THRESHOLD) return TIER2_BLOCKS; - else return TIER3_BLOCKS; -} - void launch_producer_static( const RingState& ring, const uint8_t* d_src, @@ -670,7 +663,7 @@ void launch_producer_static( uint32_t hook_type, cudaStream_t stream) { - const uint32_t n_blocks = pick_tier_blocks(nbytes); + const uint32_t n_blocks = producer_grid_blocks(nbytes); producer_static_kernel<<>>( ring, d_src, nbytes, hook_type); } @@ -684,7 +677,7 @@ void launch_producer_prefix( uint32_t hook_type, cudaStream_t stream) { - const uint32_t n_blocks = pick_tier_blocks(nbytes_upper); + const uint32_t n_blocks = producer_grid_blocks(nbytes_upper); producer_prefix_kernel<<>>( ring, d_src, nbytes_upper, row_count_dev_ptr, row_bytes, hook_type); } @@ -698,11 +691,8 @@ void launch_producer_chunked( uint32_t hook_type, cudaStream_t stream) { - uint32_t n_blocks = pick_tier_blocks(nbytes_upper); - // Round up so n_blocks % K == 0 (clean block-to-chunk partition). - if (n_blocks < K) n_blocks = K; - const uint32_t r = n_blocks % K; - if (r != 0) n_blocks += (K - r); + // A multiple of K (clean block-to-chunk partition) within the cap. + const uint32_t n_blocks = producer_chunked_grid_blocks(nbytes_upper, K); producer_chunked_kernel<<>>( ring, d_src, nbytes_upper, chunk_bytes_dev_ptr, K, hook_type); } @@ -714,7 +704,7 @@ void launch_record_producer_static( const int32_t* emit_gate, int32_t emit_value, cudaStream_t stream) { - const uint32_t blocks = pick_tier_blocks(nbytes); + const uint32_t blocks = producer_grid_blocks(nbytes); record_producer_static_kernel<<>>( ring, d_src, nbytes, emit_gate, emit_value); } @@ -732,7 +722,7 @@ void launch_record_producer_prefix( throw std::invalid_argument( "record prefix producer requires positive row bytes"); } - const uint32_t blocks = pick_tier_blocks(nbytes_upper); + const uint32_t blocks = producer_grid_blocks(nbytes_upper); record_producer_prefix_kernel<<>>( ring, d_src, nbytes_upper, row_count_dev_ptr, row_bytes, emit_gate, emit_value); @@ -752,10 +742,8 @@ void launch_record_producer_chunked( throw std::invalid_argument( "record chunked producer requires a valid evenly divided chunk count"); } - uint32_t blocks = pick_tier_blocks(nbytes_upper); - if (blocks < chunk_count) blocks = chunk_count; - const uint32_t remainder = blocks % chunk_count; - if (remainder != 0) blocks += chunk_count - remainder; + const uint32_t blocks = + producer_chunked_grid_blocks(nbytes_upper, chunk_count); record_producer_chunked_kernel<<>>( ring, d_src, nbytes_upper, chunk_bytes_dev_ptr, chunk_count, emit_gate, emit_value); diff --git a/native/csrc/ring/producer.cuh b/native/csrc/ring/producer.cuh index 05fce355f..1a518a8bb 100644 --- a/native/csrc/ring/producer.cuh +++ b/native/csrc/ring/producer.cuh @@ -28,6 +28,62 @@ namespace ring { +// Grid geometry for the D2D producers: one block per +// PRODUCER_BYTES_PER_BLOCK of the bytes the launch may copy (the +// capture-time upper bound for the prefix and chunked producers), capped at +// PRODUCER_MAX_BLOCKS. Host-only code (and CPU unit tests) can use these +// helpers without a CUDA compiler. +// +// 16 KiB per block is four uint4 stores per thread at 256 threads; 8, 16 and +// 32 KiB per block measured within 0.5 us of each other, and 16 KiB keeps a +// graph captured for many rows cheap when a replay copies only a few. +// +// PRODUCER_MAX_BLOCKS is a bounded heuristic cap, not a residency +// guarantee: 264 is sized for about two resident 256-thread blocks per SM on +// 128/132-SM parts. On smaller GPUs, or when other kernels occupy SMs, a +// capped grid can still take more than one wave. Correctness never depends +// on the grid fitting in one wave: the kernels stride over the bytes they +// actually copy, so a grid sized by the upper bound stays correct when a +// replay copies fewer bytes, and surplus blocks only join the arrival count. +static constexpr uint64_t PRODUCER_BYTES_PER_BLOCK = 16 * 1024; +static constexpr uint32_t PRODUCER_MAX_BLOCKS = 264; + +#if defined(__CUDACC__) +#define RING_PRODUCER_HD __host__ __device__ +#else +#define RING_PRODUCER_HD +#endif + +// Overflow-free ceiling division: nbytes + PRODUCER_BYTES_PER_BLOCK - 1 +// would wrap for byte counts near UINT64_MAX. +RING_PRODUCER_HD inline uint32_t producer_grid_blocks(uint64_t nbytes) { + if (nbytes <= PRODUCER_BYTES_PER_BLOCK) return 1; + const uint64_t blocks = (nbytes - 1) / PRODUCER_BYTES_PER_BLOCK + 1; + return blocks > PRODUCER_MAX_BLOCKS + ? PRODUCER_MAX_BLOCKS + : static_cast(blocks); +} + +// Grid for the chunked producers, which partition blocks evenly across +// `nchunks` chunks and therefore need a multiple of `nchunks` blocks. The +// proportional grid is rounded up to that multiple while it stays within +// PRODUCER_MAX_BLOCKS; once rounding up would pass the cap, the largest +// multiple of `nchunks` not above the cap is used instead. Only when +// `nchunks` itself exceeds the cap does the grid exceed it (one block per +// chunk is the minimum the partition needs). `nchunks == 0` is rejected by +// the launchers; it maps to the unchunked grid here. +RING_PRODUCER_HD inline uint32_t producer_chunked_grid_blocks( + uint64_t nbytes_upper, uint32_t nchunks) { + const uint32_t raw = producer_grid_blocks(nbytes_upper); + if (nchunks <= 1) return raw; + if (nchunks >= PRODUCER_MAX_BLOCKS || raw <= nchunks) return nchunks; + const uint32_t rounded_up = ((raw - 1) / nchunks + 1) * nchunks; + if (rounded_up <= PRODUCER_MAX_BLOCKS) return rounded_up; + return (PRODUCER_MAX_BLOCKS / nchunks) * nchunks; +} + +#undef RING_PRODUCER_HD + // Upper bound on K (chunk count) for shared-memory sizing in the // chunked kernel and host-side argument validation. static constexpr uint32_t PRODUCER_MAX_K = 64; @@ -43,17 +99,6 @@ namespace ring { static constexpr uint32_t PRODUCER_BLOCK_DIM = 256; -// Size-tier thresholds for grid selection (bytes). -static constexpr uint64_t TIER1_THRESHOLD = 64 * 1024; // 64 KB -static constexpr uint64_t TIER2_THRESHOLD = 4 * 1024 * 1024; // 4 MB -static constexpr uint64_t TIER3_THRESHOLD = 32 * 1024 * 1024; // 32 MB - -// Block counts per tier. -static constexpr uint32_t TIER0_BLOCKS = 1; // <= 64 KB -static constexpr uint32_t TIER1_BLOCKS = 4; // <= 4 MB -static constexpr uint32_t TIER2_BLOCKS = 16; // <= 32 MB -static constexpr uint32_t TIER3_BLOCKS = 64; // > 32 MB - // Record-only row packers use one block per logical output row and cap the // grid so a conservative upper bound cannot create an unbounded launch. static constexpr uint32_t RECORD_PACK_MAX_BLOCKS = 1024; diff --git a/tests/native/ring/Makefile b/tests/native/ring/Makefile index 43d78562e..5c7018b8b 100644 --- a/tests/native/ring/Makefile +++ b/tests/native/ring/Makefile @@ -69,7 +69,8 @@ TARGETS := $(BUILD)/test_rings $(BUILD)/test_producer \ $(BUILD)/test_clickhouse_record_sink .PHONY: all test_rings test_producer test_ring_engine test_null_mode \ - test_record_consumer test_clickhouse_record_sink run clean FORCE + test_record_consumer test_clickhouse_record_sink run clean FORCE \ + bench_producer_grid all: $(TARGETS) @@ -125,6 +126,13 @@ test_record_consumer: $(BUILD)/test_record_consumer test_clickhouse_record_sink: $(BUILD)/test_clickhouse_record_sink $(BUILD)/test_clickhouse_record_sink +# Manual microbenchmark; not part of `all` or `run`. +$(BUILD)/bench_producer_grid: bench_producer_grid.cu $(PRODUCER_CU) $(CUDA_CONFIG_STAMP) | $(BUILD) + $(CUDA_NVCC) $(NVCC_FLAGS) bench_producer_grid.cu -o $@ + +bench_producer_grid: $(BUILD)/bench_producer_grid + $(BUILD)/bench_producer_grid + run: all $(BUILD)/test_rings $(BUILD)/test_producer diff --git a/tests/native/ring/bench_producer_grid.cu b/tests/native/ring/bench_producer_grid.cu new file mode 100644 index 000000000..21ef1d5de --- /dev/null +++ b/tests/native/ring/bench_producer_grid.cu @@ -0,0 +1,128 @@ +// Microbenchmark: D2D producer copy time under the previous size-tier grid +// (1/4/16/64 blocks at 64 KB / 4 MB / 32 MB) and the proportional grid +// (ring::producer_grid_blocks). Not part of `make run`; build and run with +// make -C tests/native/ring bench_producer_grid +// +// Each configuration replays a CUDA graph of 64 back-to-back producer +// launches so launch overhead is amortized the same way it is in a captured +// decode step; the reported time is per launch. + +#include "../../../native/csrc/ring/producer.cu" +#include "ring/ring_alloc.h" + +#include + +#include +#include +#include + +#define CUDA_CHECK(expr) \ + do { \ + cudaError_t error_ = (expr); \ + if (error_ != cudaSuccess) { \ + std::fprintf(stderr, "CUDA error at %s:%d: %s\n", \ + __FILE__, __LINE__, cudaGetErrorString(error_)); \ + std::exit(1); \ + } \ + } while (0) + +static uint32_t previous_tier_blocks(uint64_t nbytes) { + if (nbytes <= 64ull * 1024) return 1; + if (nbytes <= 4ull * 1024 * 1024) return 4; + if (nbytes <= 32ull * 1024 * 1024) return 16; + return 64; +} + +static constexpr int kLaunchesPerGraph = 64; +static constexpr int kReplays = 50; + +// Returns microseconds per producer launch. +static double time_static(ring::RingState& state, const uint8_t* src, + uint64_t nbytes, uint32_t blocks, + cudaStream_t stream) { + cudaGraph_t graph; + cudaGraphExec_t exec; + CUDA_CHECK(cudaStreamBeginCapture(stream, cudaStreamCaptureModeGlobal)); + for (int i = 0; i < kLaunchesPerGraph; ++i) { + ring::producer_static_kernel<<>>(state, src, nbytes, 0); + } + CUDA_CHECK(cudaStreamEndCapture(stream, &graph)); + CUDA_CHECK(cudaGraphInstantiate(&exec, graph, 0)); + + cudaEvent_t start, stop; + CUDA_CHECK(cudaEventCreate(&start)); + CUDA_CHECK(cudaEventCreate(&stop)); + // Warm up, then reset the ring heads between replays from the host + // (outside the timed region) so every replay writes the same slots. + for (int warm = 0; warm < 3; ++warm) { + *state.task_head = 0; + *state.payload_head = 0; + CUDA_CHECK(cudaGraphLaunch(exec, stream)); + CUDA_CHECK(cudaStreamSynchronize(stream)); + } + float total_ms = 0.0f; + for (int replay = 0; replay < kReplays; ++replay) { + *state.task_head = 0; + *state.payload_head = 0; + CUDA_CHECK(cudaEventRecord(start, stream)); + CUDA_CHECK(cudaGraphLaunch(exec, stream)); + CUDA_CHECK(cudaEventRecord(stop, stream)); + CUDA_CHECK(cudaEventSynchronize(stop)); + float ms = 0.0f; + CUDA_CHECK(cudaEventElapsedTime(&ms, start, stop)); + total_ms += ms; + } + CUDA_CHECK(cudaEventDestroy(start)); + CUDA_CHECK(cudaEventDestroy(stop)); + CUDA_CHECK(cudaGraphExecDestroy(exec)); + CUDA_CHECK(cudaGraphDestroy(graph)); + return 1000.0 * total_ms / (kReplays * kLaunchesPerGraph); +} + +int main() { + setbuf(stdout, nullptr); + ring::set_ring_null_mode(false); + cudaDeviceProp prop; + CUDA_CHECK(cudaGetDeviceProperties(&prop, 0)); + std::printf("device: %s (%d SMs)\n", prop.name, prop.multiProcessorCount); + + const std::vector sizes = { + 8ull * 1024, // one decode row of a 4096-wide bf16 hidden state + 64ull * 1024, // 8 rows + 256ull * 1024, // 32 rows + 1200ull * 1024, // ~150 rows + 4ull * 1024 * 1024, // 512 rows + 16ull * 1024 * 1024, // a 2048-token prefill chunk + }; + uint64_t largest = sizes.back(); + ring::RingConfig cfg; + cfg.task_ring_entries = 128; + cfg.payload_ring_bytes = largest * (kLaunchesPerGraph + 1); + cfg.pinned_staging_bytes = 4096; + ring::AllocatedRing allocated(cfg); + allocated.init(); + ring::RingState& state = allocated.state(); + + uint8_t* src = nullptr; + CUDA_CHECK(cudaMalloc(&src, largest)); + CUDA_CHECK(cudaMemset(src, 0x5A, largest)); + cudaStream_t stream; + CUDA_CHECK(cudaStreamCreateWithFlags(&stream, cudaStreamNonBlocking)); + + std::printf("%12s %10s %12s %10s %12s %8s\n", "bytes", "old_blocks", + "old_us", "new_blocks", "new_us", "speedup"); + for (uint64_t nbytes : sizes) { + const uint32_t old_blocks = previous_tier_blocks(nbytes); + const uint32_t new_blocks = ring::producer_grid_blocks(nbytes); + const double old_us = time_static(state, src, nbytes, old_blocks, stream); + const double new_us = time_static(state, src, nbytes, new_blocks, stream); + std::printf("%12llu %10u %12.2f %10u %12.2f %7.2fx (%.0f -> %.0f GB/s)\n", + static_cast(nbytes), old_blocks, old_us, + new_blocks, new_us, old_us / new_us, + nbytes / (old_us * 1e3), nbytes / (new_us * 1e3)); + } + CUDA_CHECK(cudaStreamDestroy(stream)); + CUDA_CHECK(cudaFree(src)); + return 0; +} diff --git a/tests/native/ring/test_producer.cu b/tests/native/ring/test_producer.cu index cd1ed9927..eb946a34e 100644 --- a/tests/native/ring/test_producer.cu +++ b/tests/native/ring/test_producer.cu @@ -644,6 +644,208 @@ static void test_record_device_gate_rejects_without_publication() { CUDA_CHECK(cudaFree(device)); } +static void test_grid_geometry() { + banner("producer grid is proportional to bytes and capped"); + constexpr uint64_t per_block = ring::PRODUCER_BYTES_PER_BLOCK; + EXPECT(ring::producer_grid_blocks(0) == 1); + EXPECT(ring::producer_grid_blocks(1) == 1); + EXPECT(ring::producer_grid_blocks(per_block) == 1); + EXPECT(ring::producer_grid_blocks(per_block + 1) == 2); + // A 1.125 MiB hook (e.g. 144 rows of 8 KiB): 72 blocks, previously 4. + EXPECT(ring::producer_grid_blocks(72 * per_block) == 72); + EXPECT(ring::producer_grid_blocks( + ring::PRODUCER_MAX_BLOCKS * per_block) == + ring::PRODUCER_MAX_BLOCKS); + EXPECT(ring::producer_grid_blocks( + ring::PRODUCER_MAX_BLOCKS * per_block + 1) == + ring::PRODUCER_MAX_BLOCKS); + EXPECT(ring::producer_grid_blocks(uint64_t(1) << 40) == + ring::PRODUCER_MAX_BLOCKS); + // No wrap near the top of the range (a naive ceiling division wraps to 0). + EXPECT(ring::producer_grid_blocks(UINT64_MAX) == ring::PRODUCER_MAX_BLOCKS); + EXPECT(ring::producer_grid_blocks(UINT64_MAX - per_block + 2) == + ring::PRODUCER_MAX_BLOCKS); + // Monotone in the byte count. + uint32_t previous = 0; + bool monotone = true; + for (uint64_t bytes = 0; bytes < 300 * per_block; bytes += 4093) { + const uint32_t blocks = ring::producer_grid_blocks(bytes); + monotone = monotone && blocks >= previous; + previous = blocks; + } + EXPECT(monotone); + + // Chunked grids are a multiple of the chunk count and stay within the + // cap whenever the chunk count does (64 * 5 = 320 would pass 264). + EXPECT(ring::producer_chunked_grid_blocks(0, 3) == 3); + EXPECT(ring::producer_chunked_grid_blocks(4 * per_block, 64) == 64); + EXPECT(ring::producer_chunked_grid_blocks(70 * per_block, 64) == 128); + EXPECT(ring::producer_chunked_grid_blocks( + ring::PRODUCER_MAX_BLOCKS * per_block, 64) == 256); + EXPECT(ring::producer_chunked_grid_blocks(uint64_t(1) << 40, 64) == 256); + EXPECT(ring::producer_chunked_grid_blocks(uint64_t(1) << 40, 3) == 264); + EXPECT(ring::producer_chunked_grid_blocks(uint64_t(1) << 40, 1) == + ring::PRODUCER_MAX_BLOCKS); + for (uint32_t k = 1; k <= ring::PRODUCER_MAX_K; ++k) { + for (uint64_t bytes : {uint64_t(0), 5 * per_block + 1, + uint64_t(1) << 40}) { + const uint32_t blocks = ring::producer_chunked_grid_blocks(bytes, k); + EXPECT(blocks % k == 0); + EXPECT(blocks >= k); + EXPECT(blocks <= ring::PRODUCER_MAX_BLOCKS); + } + } +} + +// Sizes chosen to land on 1 block, a partial last block, a mid-size grid, +// exactly the cap, and past the cap (grid-stride beyond one pass). +static const uint64_t kWideGridSizes[] = { + 333, + 3 * ring::PRODUCER_BYTES_PER_BLOCK + 5, + 72 * ring::PRODUCER_BYTES_PER_BLOCK, + uint64_t(ring::PRODUCER_MAX_BLOCKS) * ring::PRODUCER_BYTES_PER_BLOCK, + uint64_t(ring::PRODUCER_MAX_BLOCKS) * ring::PRODUCER_BYTES_PER_BLOCK * 3 + 4097, +}; + +static void test_wide_grid_static_and_wrap() { + banner("static producer copies exactly on wide grids, with and without wrap"); + constexpr uint64_t payload_bytes = 16ull * 1024 * 1024; + for (uint64_t size : kWideGridSizes) { + for (bool wrap : {false, true}) { + ring::AllocatedRing allocated(make_config(payload_bytes)); + allocated.init(); + ring::RingState& state = allocated.state(); + // Start 1 KiB + 16 before the end so the copy crosses the wrap. + const uint64_t start = wrap ? payload_bytes - 1040 : 0; + *state.payload_head = start; + const std::vector source = + pattern(size, static_cast(size)); + uint8_t* device = upload(source); + ring::launch_producer_static(state, device, size, 0); + CUDA_CHECK(cudaDeviceSynchronize()); + expect_publication(state, 0, size); + EXPECT(*state.task_head == 1); + EXPECT(*state.payload_head == + start + ring::align_up(size, ring::PAYLOAD_ALIGN)); + EXPECT(*state.actual_bytes_counter == size); + EXPECT(read_payload(state, start, size) == source); + CUDA_CHECK(cudaFree(device)); + } + } +} + +static void test_wide_grid_prefix_selects_fewer_rows() { + banner("prefix grid sized by the upper bound copies only the selected rows"); + constexpr uint64_t row_bytes = 8192; + constexpr uint64_t upper_rows = 512; // 4 MiB upper bound -> capped grid + const std::vector source = pattern(upper_rows * row_bytes, 41); + uint8_t* device = upload(source); + for (int64_t rows : {int64_t(0), int64_t(1), int64_t(3), int64_t(37), + int64_t(upper_rows)}) { + ring::AllocatedRing allocated(make_config(8ull * 1024 * 1024)); + allocated.init(); + ring::RingState& state = allocated.state(); + int64_t* row_count = upload_counts({rows}); + ring::launch_producer_prefix(state, device, source.size(), row_count, + row_bytes, 1); + CUDA_CHECK(cudaDeviceSynchronize()); + const uint64_t expected_bytes = static_cast(rows) * row_bytes; + expect_publication(state, 0, expected_bytes); + EXPECT(*state.payload_head == + ring::align_up(expected_bytes, ring::PAYLOAD_ALIGN)); + EXPECT(*state.actual_bytes_counter == expected_bytes); + const std::vector expected(source.begin(), + source.begin() + expected_bytes); + EXPECT(read_payload(state, 0, expected_bytes) == expected); + CUDA_CHECK(cudaFree(row_count)); + } + CUDA_CHECK(cudaFree(device)); +} + +static void test_wide_grid_chunked_partition() { + banner("chunked producer keeps a block-per-chunk partition on wide grids"); + // 3 chunks of 2 MiB: 384 proportional blocks, capped at 264 (already a + // multiple of 3). + constexpr uint32_t chunks = 3; + constexpr uint64_t chunk_input_bytes = 2ull * 1024 * 1024; + const std::vector counts = { + static_cast(chunk_input_bytes), + 5, + static_cast(chunk_input_bytes / 2 + 3)}; + const std::vector source = pattern(chunks * chunk_input_bytes, 77); + std::vector expected; + for (uint32_t chunk = 0; chunk < chunks; ++chunk) { + const auto begin = source.begin() + chunk * chunk_input_bytes; + expected.insert(expected.end(), begin, begin + counts[chunk]); + } + for (bool record : {false, true}) { + ring::AllocatedRing allocated(make_config(8ull * 1024 * 1024)); + allocated.init(); + ring::RingState& state = allocated.state(); + uint8_t* device = upload(source); + int64_t* device_counts = upload_counts(counts); + if (record) { + ring::launch_record_producer_chunked( + state, device, source.size(), device_counts, chunks, + nullptr, 0); + } else { + ring::launch_producer_chunked(state, device, source.size(), + device_counts, chunks, 2); + } + CUDA_CHECK(cudaDeviceSynchronize()); + expect_publication(state, 0, expected.size()); + EXPECT(*state.actual_bytes_counter == expected.size()); + EXPECT(read_payload(state, 0, expected.size()) == expected); + CUDA_CHECK(cudaFree(device_counts)); + CUDA_CHECK(cudaFree(device)); + } +} + +static void test_mixed_grid_launches_share_arrival_counter() { + banner("back-to-back launches of different grid sizes publish in order"); + constexpr uint64_t payload_bytes = 16ull * 1024 * 1024; + ring::AllocatedRing allocated(make_config(payload_bytes)); + allocated.init(); + ring::RingState& state = allocated.state(); + // 1 block, capped grid, 7 blocks, 1 block, 72 blocks: the shared + // last-block-arrives counter must reset between grids of any size. + const std::vector sizes = { + 100, + uint64_t(ring::PRODUCER_MAX_BLOCKS) * ring::PRODUCER_BYTES_PER_BLOCK + 33, + 7 * ring::PRODUCER_BYTES_PER_BLOCK, + 16, + 72 * ring::PRODUCER_BYTES_PER_BLOCK - 1, + }; + std::vector> sources; + std::vector devices; + for (uint64_t i = 0; i < sizes.size(); ++i) { + sources.push_back(pattern(sizes[i], static_cast(90 + i))); + devices.push_back(upload(sources.back())); + if (i % 2 == 0) { + ring::launch_producer_static(state, devices.back(), sizes[i], 0); + } else { + ring::launch_record_producer_static(state, devices.back(), + sizes[i], nullptr, 0); + } + } + CUDA_CHECK(cudaDeviceSynchronize()); + uint64_t expected_head = 0; + uint64_t expected_actual = 0; + for (uint64_t sequence = 0; sequence < sizes.size(); ++sequence) { + expect_publication(state, sequence, sizes[sequence]); + EXPECT(read_payload(state, expected_head, sizes[sequence]) == + sources[sequence]); + expected_head += ring::align_up(sizes[sequence], ring::PAYLOAD_ALIGN); + expected_actual += sizes[sequence]; + } + EXPECT(*state.task_head == sizes.size()); + EXPECT(*state.payload_head == expected_head); + EXPECT(*state.actual_bytes_counter == expected_actual); + for (uint8_t* device : devices) { + CUDA_CHECK(cudaFree(device)); + } +} + int main() { setbuf(stdout, nullptr); ring::set_ring_null_mode(false); @@ -662,6 +864,11 @@ int main() { test_record_sequence_and_segmented_pack(); test_record_device_gate_rejects_without_publication(); test_all_producers_handle_unaligned_addresses(); + test_grid_geometry(); + test_wide_grid_static_and_wrap(); + test_wide_grid_prefix_selects_fewer_rows(); + test_wide_grid_chunked_partition(); + test_mixed_grid_launches_share_arrival_counter(); std::printf("Results: %d passed, %d failed\n", g_pass, g_fail); return g_fail == 0 ? 0 : 1; diff --git a/tests/native/ring/test_producer_grid_geometry.cpp b/tests/native/ring/test_producer_grid_geometry.cpp new file mode 100644 index 000000000..7ea270597 --- /dev/null +++ b/tests/native/ring/test_producer_grid_geometry.cpp @@ -0,0 +1,83 @@ +// Host-only checks of the producer grid geometry helpers in producer.cuh. +// Built with a plain C++ compiler by tests/test_producer_grid_geometry.py; +// no CUDA toolkit or device is involved. +#include +#include +#include + +#include "ring/producer.cuh" + +static int failures = 0; + +#define EXPECT(cond) \ + do { \ + if (!(cond)) { \ + std::fprintf(stderr, "FAIL %s:%d: %s\n", __FILE__, __LINE__, \ + #cond); \ + ++failures; \ + } \ + } while (0) + +int main() { + constexpr uint64_t per_block = ring::PRODUCER_BYTES_PER_BLOCK; + constexpr uint32_t cap = ring::PRODUCER_MAX_BLOCKS; + + EXPECT(ring::producer_grid_blocks(0) == 1); + EXPECT(ring::producer_grid_blocks(per_block) == 1); + EXPECT(ring::producer_grid_blocks(per_block + 1) == 2); + EXPECT(ring::producer_grid_blocks(cap * per_block) == cap); + EXPECT(ring::producer_grid_blocks(cap * per_block + 1) == cap); + EXPECT(ring::producer_grid_blocks(UINT64_MAX) == cap); + + // The audit's example: a capped 264-block grid for 64 chunks used to be + // rounded up to 320; it now rounds down to 256, within the cap. + EXPECT(ring::producer_chunked_grid_blocks(cap * per_block, 64) == 256); + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, 64) == 256); + // Rounding up is kept while it stays within the cap. + EXPECT(ring::producer_chunked_grid_blocks(70 * per_block, 64) == 128); + EXPECT(ring::producer_chunked_grid_blocks(200 * per_block, 64) == 256); + EXPECT(ring::producer_chunked_grid_blocks(10 * per_block, 3) == 12); + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, 3) == 264); + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, 5) == 260); + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, 7) == 259); + // Small transfers still get one block per chunk. + EXPECT(ring::producer_chunked_grid_blocks(0, 64) == 64); + EXPECT(ring::producer_chunked_grid_blocks(per_block, 3) == 3); + // A single chunk is the unchunked grid. + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, 1) == cap); + EXPECT(ring::producer_chunked_grid_blocks(5 * per_block, 1) == 5); + // Only a chunk count above the cap can exceed it (one block per chunk). + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, cap) == cap); + EXPECT(ring::producer_chunked_grid_blocks(UINT64_MAX, cap + 1) == cap + 1); + EXPECT(ring::producer_chunked_grid_blocks(0, 300) == 300); + + for (uint32_t k = 1; k <= cap; ++k) { + for (uint64_t bytes : {uint64_t(0), uint64_t(1), per_block, + 5 * per_block + 1, 63 * per_block, + uint64_t(cap) * per_block, + uint64_t(cap) * per_block + 1, + uint64_t(1) << 40, UINT64_MAX}) { + const uint32_t raw = ring::producer_grid_blocks(bytes); + const uint32_t blocks = ring::producer_chunked_grid_blocks(bytes, k); + EXPECT(blocks % k == 0); + EXPECT(blocks >= k); + EXPECT(blocks <= cap); + // Never more than one partition step away from the proportional + // grid: the nearest multiple of k at or above it, or the largest + // multiple at or below the cap. + if (raw <= k) { + EXPECT(blocks == k); + } else { + const uint32_t up = ((raw - 1) / k + 1) * k; + EXPECT(blocks == (up <= cap ? up : (cap / k) * k)); + } + } + } + + if (failures != 0) { + std::fprintf(stderr, "%d failure(s)\n", failures); + return 1; + } + std::printf("producer grid geometry: ok\n"); + return 0; +} diff --git a/tests/test_producer_grid_geometry.py b/tests/test_producer_grid_geometry.py new file mode 100644 index 000000000..6fd186bb4 --- /dev/null +++ b/tests/test_producer_grid_geometry.py @@ -0,0 +1,34 @@ +"""Host-side checks of the D2D producer grid geometry (producer.cuh). + +Compiles tests/native/ring/test_producer_grid_geometry.cpp with the host C++ +compiler. The helpers are plain integer arithmetic shared with the CUDA +launchers, so no CUDA toolkit or device is needed. +""" + +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path + +import pytest + + +@pytest.mark.cpu +def test_producer_grid_geometry(tmp_path): + compiler = shutil.which("g++") or shutil.which("c++") + if compiler is None: + pytest.skip("a C++17 compiler is required") + root = Path(__file__).resolve().parents[1] + source = root / "tests" / "native" / "ring" / "test_producer_grid_geometry.cpp" + executable = tmp_path / "test_producer_grid_geometry" + build = subprocess.run( + [compiler, "-std=c++17", "-O0", "-Wall", "-Wextra", "-Werror", + f"-I{root / 'native' / 'csrc'}", str(source), "-o", str(executable)], + capture_output=True, text=True, check=False, + ) + assert build.returncode == 0, build.stdout + build.stderr + run = subprocess.run([str(executable)], capture_output=True, text=True, + check=False, timeout=60) + assert run.returncode == 0, run.stdout + run.stderr + assert "producer grid geometry: ok" in run.stdout