diff --git a/README.md b/README.md index 7637f750..6ecf715e 100644 --- a/README.md +++ b/README.md @@ -319,7 +319,9 @@ For Blosc, see the [binary format specification][blosc-format], the The main entry points live in [`src/stream.gpu.h`][src-stream-gpu-h] (GPU) and [`src/stream.cpu.h`][src-stream-cpu-h] (CPU). Both backends expose the same -`struct writer` interface for feeding data. +`struct writer` interface for feeding data. For frames arriving on a regular +camera clock, use the [buffered input adapter](#camera-frames-at-a-regular-frequency) +described below. ```c // 1. Estimate GPU memory requirements @@ -351,6 +353,88 @@ dimensions, shard layout, LOD reduction method, etc.). See [`src/stream.gpu.h`][src-stream-gpu-h] or [`src/stream.cpu.h`][src-stream-cpu-h] for the full API. +### Camera frames at a regular frequency + +Use the [buffered input writer](src/writer.buffered.h) when frames arrive on a +camera's clock. Direct stream appends can spend time processing a batch or waiting +for output capacity. The buffered adapter copies accepted input into a bounded +queue and forwards it to the stream on one worker thread. The camera can reuse +accepted bytes as soon as append returns, while the stream continues processing. + +These buffers have different roles: + +| Buffer | Control | Purpose | +| --- | --- | --- | +| Camera pool | Acquisition application | Holds arriving frames until the writer accepts them | +| Buffered input queue | `buffered_writer_config.capacity_bytes` | Owns copies of accepted input while the forwarding thread feeds the stream | +| Stream processing buffers | `tile_stream_configuration.buffer_capacity_bytes` and `target_batch_bytes` | Size the stream's internal input capacity and processing batches | + +Setting a stream buffer or batch to 64 MiB does **not** enable the buffered input +adapter. Create it explicitly after creating the stream and before starting the +camera. The example below allows 250 ms at 4 useful GiB/s for BBBC022's padded +frames; size the queue for your own submitted byte rate and pause allowance: + +```c +#include "writer.buffered.h" + +struct buffered_writer_config input_config = { + .capacity_bytes = 1152u * 1024 * 1024, + .max_drain_bytes = 64u * 1024 * 1024, +}; +struct buffered_writer* input = buffered_writer_create( + tile_stream_cpu_writer(stream), &input_config); +if (!input) + return 1; +struct writer* frames = buffered_writer_as_writer(input); +``` + +Submit each frame with `writer_append_wait(frames, frame_slice)` and check its +error and unaccepted remainder before recycling the entire camera buffer. +`max_drain_bytes` caps each call to the stream; the worker can forward a smaller +available batch. Both byte limits must respect the stream's input granularity, +and the drain limit cannot exceed capacity. Zero drain limit uses capacity. + +Size input capacity from the bytes actually submitted, including padding: + +`capacity >= submitted_bytes_per_second * pause_seconds + active_drain_bytes` + +Round up for whole frames during the pause and for the downstream input +granularity. The extra drain allowance covers input still owned by the forwarding +call. This assumes the queue has cleared earlier backlog before the pause. +Keep `max_drain_bytes` separate from capacity: increasing the latency allowance +need not increase the compression batch or drain size. + +At 4 useful GiB/s, BBBC022's 520×696 uint16 frames padded to 544×704 for 16 KiB +chunks produce 4.23 GiB/s of submitted input. A 250 ms pause requires about +1.06 GiB, plus one 64 MiB drain; rounding up to a drain-sized block gives +1,152 MiB. The camera pool is additional. A 64 MiB queue alone holds only about +14.8 ms of input at that rate. + +Creation touches every page of the adapter queue and waits for its forwarding +thread. CPU stream creation also touches its main payload buffers and waits for +its processing workers. Initialization takes time and commits memory before the +first append; include it in startup and memory budgets. The input queue uses +ordinary host memory. GPU staging and output buffers are page-locked separately. +On NUMA hosts, set CPU affinity and memory binding before creation; for one GPU, +use its local node. Touching pages does not prevent later NUMA-balancing faults. + +Append blocks when the input queue is full. Keep enough camera buffers to absorb +that wait when acquisition cannot pause, and keep capture independent of the +append thread. The queue adds a copy, its configured +memory capacity, and one forwarding thread; measure total memory and buffer-return +latency for the intended frame rate. `buffered_writer_get_stats()` reports pending +bytes, peak occupancy, and time spent waiting for space. + +Use the adapter exclusively while it owns the stream's writer interface. At the +end, call `buffered_writer_destroy(input)` and check its return value: it drains, +flushes, and closes the downstream writer before freeing the adapter. Destroy the +stream and its sink afterwards. Returning a camera buffer only means its input +was copied; output completion requires the final drain and the sink's completion +guarantees. Flush finalizes the stream, so do not flush after individual frames. + +Test startup and accumulated backlog at the intended camera rate, and count +both the camera pool and adapter queue in the memory budget. + ### GPU host placement On Linux, GPU streams use the NUMA node of the current CUDA context's device diff --git a/docs/buffered-writer.md b/docs/buffered-writer.md index 3f3f5adb..4d8b6814 100644 --- a/docs/buffered-writer.md +++ b/docs/buffered-writer.md @@ -10,7 +10,17 @@ sustained throughput. `capacity_bytes` bounds queued and active input. `max_drain_bytes` limits each downstream batch; zero uses the full capacity. Capacity must be nonzero and the cap must not exceed it. Both values and input sizes must respect downstream -granularity, such as whole elements. Tune these example sizes to your workload: +granularity, such as whole elements. Capacity should cover the submitted byte rate +(including padding) times the pause allowance, plus input already occupied by an +active drain. Round up for whole arriving frames and downstream granularity. Keep +the drain limit independent of capacity so a larger queue does not create larger +handoffs. The queue must have room when the pause begins; sustained overload or +repeated pauses without recovery can still fill it. + +For example, 4 useful GiB/s of BBBC022 frames becomes 4.23 GiB/s after padding +520×696 frames to 544×704 for 16 KiB chunks. Allowing 250 ms plus an active +64 MiB drain requires 1,152 MiB when rounded up to +64 MiB blocks. The camera pool is additional. Tune the sizes below to your workload: ```c #include "writer.buffered.h" @@ -18,8 +28,8 @@ granularity, such as whole elements. Tune these example sizes to your workload: int write_buffered(struct writer* downstream, struct slice input) { const struct buffered_writer_config config = { - .capacity_bytes = 128u * 1024 * 1024, - .max_drain_bytes = 16u * 1024 * 1024, + .capacity_bytes = 1152u * 1024 * 1024, + .max_drain_bytes = 64u * 1024 * 1024, }; struct buffered_writer* buffered = buffered_writer_create(downstream, &config); if (!buffered) @@ -31,6 +41,20 @@ int write_buffered(struct writer* downstream, struct slice input) } ``` +Creation writes to every queue page and waits for the forwarding thread before +returning. Create the adapter before starting acquisition and include that time +and committed memory in the startup budget. This moves initial page allocation +out of append; it does not lock pages in memory or prepare a custom downstream +writer's internal buffers. + +On a NUMA host, set CPU affinity and memory policy before creating the writer. +For a single GPU, use cores and memory on the GPU's local node. Page locking and +NUMA placement are separate: GPU staging and output buffers are page-locked, +while the buffered input queue uses ordinary host memory. Linux automatic NUMA +balancing can introduce page faults even after pages have been touched; explicit +memory binding avoids that source of variability for a process confined to one +node. Keep placement under application control when using multiple GPUs or nodes. + ## Input and ownership Append copies available input and returns the unaccepted suffix in `rest`. diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 42cdf87c..f915878c 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -32,7 +32,7 @@ add_src_lib(lod_plan SOURCES lod/lod_plan.c lod/lod_plan.h ) add_src_lib(crc32c SOURCES zarr/crc32c.c zarr/crc32c.h LINKS platform) -add_src_lib(platform SOURCES platform/platform.h +add_src_lib(platform SOURCES platform/platform.h platform/platform_memory.c platform/placement.c platform/placement.h platform/topology.c platform/topology.h LINKS Threads::Threads chucky_log $<$:psapi> diff --git a/src/cpu/stream.c b/src/cpu/stream.c index 4230370c..35525377 100644 --- a/src/cpu/stream.c +++ b/src/cpu/stream.c @@ -73,12 +73,15 @@ tile_stream_cpu_create(const struct tile_stream_configuration* config, // Chunk pool: K epochs' worth across all levels. s->chunk_pool = calloc((uint64_t)K * total_chunks, chunk_stride_bytes); CHECK(Fail, s->chunk_pool); + platform_touch_pages(s->chunk_pool, + (uint64_t)K * total_chunks * chunk_stride_bytes); s->pool_fully_covered = (s->layout.chunk_stride == s->layout.chunk_elements); // Compressed output buffer (K epochs). s->compressed = malloc((uint64_t)K * total_chunks * max_out); CHECK(Fail, s->compressed); + platform_touch_pages(s->compressed, (uint64_t)K * total_chunks * max_out); s->comp_sizes = (size_t*)calloc((uint64_t)K * total_chunks, sizeof(size_t)); CHECK(Fail, s->comp_sizes); @@ -166,11 +169,14 @@ tile_stream_cpu_create(const struct tile_stream_configuration* config, // Linear epoch buffer: input is accumulated here before LOD scatter. s->linear = calloc(s->layout.epoch_elements, bytes_per_element); CHECK(Fail, s->linear); + platform_touch_pages(s->linear, + s->layout.epoch_elements * bytes_per_element); uint64_t total_lod_elements = s->cl.plan.level_spans.ends[s->cl.plan.levels.nlod - 1]; s->lod_values = calloc(total_lod_elements, bytes_per_element); CHECK(Fail, s->lod_values); + platform_touch_pages(s->lod_values, total_lod_elements * bytes_per_element); // Append accumulator: total elements in levels 1+ (packed). // Uses the source dtype (not a wider accumulator) for integer mean — @@ -184,6 +190,7 @@ tile_stream_cpu_create(const struct tile_stream_configuration* config, if (append_total > 0) { s->append_accum = calloc(append_total, bytes_per_element); CHECK(Fail, s->append_accum); + platform_touch_pages(s->append_accum, append_total * bytes_per_element); } memset(s->append_counts, 0, sizeof(s->append_counts)); } diff --git a/src/platform/platform.h b/src/platform/platform.h index a4690a59..7610e855 100644 --- a/src/platform/platform.h +++ b/src/platform/platform.h @@ -25,6 +25,9 @@ platform_aligned_alloc(size_t alignment, size_t size); void platform_aligned_free(void* ptr); +void +platform_touch_pages(void* ptr, size_t bytes); + // Return the available physical memory in bytes, or 0 on failure. size_t platform_available_memory(void); diff --git a/src/platform/platform_memory.c b/src/platform/platform_memory.c new file mode 100644 index 00000000..03ad02c2 --- /dev/null +++ b/src/platform/platform_memory.c @@ -0,0 +1,14 @@ +#include "platform/platform.h" + +void +platform_touch_pages(void* ptr, size_t bytes) +{ + if (!ptr || !bytes) + return; + volatile unsigned char* data = ptr; + const size_t page = platform_page_alignment(); + const size_t count = 1 + (bytes - 1) / page; + for (size_t i = 0; i < count; ++i) + data[i * page] = 0; + data[bytes - 1] = 0; +} diff --git a/src/stream/host_output_pool.c b/src/stream/host_output_pool.c index a9e192cc..2f950149 100644 --- a/src/stream/host_output_pool.c +++ b/src/stream/host_output_pool.c @@ -36,7 +36,9 @@ static void* default_allocate(void* ctx, size_t alignment, size_t bytes) { (void)ctx; - return platform_aligned_alloc(alignment, bytes); + void* data = platform_aligned_alloc(alignment, bytes); + platform_touch_pages(data, bytes); + return data; } static void diff --git a/src/threadpool/threadpool.c b/src/threadpool/threadpool.c index 2221cbd9..a57c0f09 100644 --- a/src/threadpool/threadpool.c +++ b/src/threadpool/threadpool.c @@ -50,6 +50,7 @@ struct threadpool struct platform_cond* cv; struct worker* workers; int nworkers; + int ready; // Current task (set by dispatcher before bumping epoch). pool_task_fn task_fn; @@ -65,6 +66,11 @@ worker_main(void* arg) struct threadpool* p = w->pool; uint64_t last_epoch = 0; + platform_mutex_lock(p->mu); + ++p->ready; + platform_cond_broadcast(p->cv); + platform_mutex_unlock(p->mu); + for (;;) { // Phase A: brief spin watching for new work. uint64_t epoch = atomic_load_explicit(&p->epoch, memory_order_acquire); @@ -179,6 +185,11 @@ threadpool_new(int nthreads) } } + platform_mutex_lock(p->mu); + while (p->ready < p->nworkers) + platform_cond_wait(p->cv, p->mu); + platform_mutex_unlock(p->mu); + return p; fail: diff --git a/src/writer.buffered.c b/src/writer.buffered.c index fe65fea4..00db19a5 100644 --- a/src/writer.buffered.c +++ b/src/writer.buffered.c @@ -18,6 +18,7 @@ struct buffered_writer size_t read_offset; size_t write_offset; struct buffered_writer_stats stats; + int ready; int flush_requested; int flushed; int close_requested; @@ -52,6 +53,8 @@ worker_main(void* arg) { struct buffered_writer* b = arg; platform_mutex_lock(b->mutex); + b->ready = 1; + platform_cond_broadcast(b->changed); for (;;) { if (b->stats.pending_bytes) { // Snapshot a contiguous byte prefix, independent of producer appends. @@ -230,9 +233,14 @@ buffered_writer_create(struct writer* downstream, b->changed = platform_cond_new(); if (!b->data || !b->mutex || !b->changed) goto fail; + platform_touch_pages(b->data, b->capacity_bytes); b->worker = platform_thread_start(worker_main, b); if (!b->worker) goto fail; + platform_mutex_lock(b->mutex); + while (!b->ready) + platform_cond_wait(b->changed, b->mutex); + platform_mutex_unlock(b->mutex); return b; fail: release(b); diff --git a/src/zarr/shard_delivery.c b/src/zarr/shard_delivery.c index 0bcbf8b2..29e76dae 100644 --- a/src/zarr/shard_delivery.c +++ b/src/zarr/shard_delivery.c @@ -72,6 +72,8 @@ init_shard_state(struct shard_state* ss, const struct level_layout_info* li) (uint8_t*)platform_aligned_alloc(page, ss->footer_buf_pool_bytes); if (!ss->footer_buf_pool) return 1; + platform_touch_pages(ss->tail_buf_pool, ss->tail_buf_pool_bytes); + platform_touch_pages(ss->footer_buf_pool, ss->footer_buf_pool_bytes); ss->shard_file_capacity = shard_file_capacity_for(li->chunks_per_shard_total, li->agg_layout.max_comp_chunk_bytes, diff --git a/tests/test_buffered_writer.c b/tests/test_buffered_writer.c index 00649fea..c92c6ba5 100644 --- a/tests/test_buffered_writer.c +++ b/tests/test_buffered_writer.c @@ -595,10 +595,40 @@ test_destroy_and_config(void) return 1; } +static int +test_memory_ready_before_append(void) +{ + struct fake_writer f; + if (fake_init(&f)) + return 1; + struct buffered_writer* b = NULL; + const size_t capacity = 64u * 1024 * 1024; + uint64_t before = 0, after = 0; + CHECK(Fail, !platform_resident_memory(&before)); + b = buffered_writer_create( + &f.writer, + &(struct buffered_writer_config){ .capacity_bytes = capacity, + .max_drain_bytes = 1024 }); + CHECK(Fail, b); + CHECK(Fail, !platform_resident_memory(&after)); + CHECK(Fail, after >= before + capacity / 2); + CHECK(Fail, buffered_writer_get_stats(b).accepted_bytes == 0); + int failed = buffered_writer_destroy(b); + b = NULL; + CHECK(Fail, !failed); + fake_free(&f); + return 0; +Fail: + buffered_writer_destroy(b); + fake_free(&f); + return 1; +} + int main(void) { - int failed = test_destroy_and_config(); + int failed = test_memory_ready_before_append(); + failed += test_destroy_and_config(); failed += test_queue(0, 0); failed += test_queue(writer_error_fail, 3); failed += test_queue(writer_error_finished, 3);