Prepare stream buffers before capture

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

docs/buffered-writer.md

@@ -10,16 +10,26 @@ 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"
 
 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`.

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 $<$<PLATFORM_ID:Windows>:psapi>

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));
     }

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);

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;
+}

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

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:

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);

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,

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);