Map packed input into edge chunks with scatter lookup tables

README.md

@@ -29,6 +29,12 @@ LOD pyramids built on the fly: after the base level (L0) is chunked, the pipelin
 scatters, reduces, and chunks each coarser level before compressing and delivering
 it alongside L0.
 
+Input remains tightly packed. At partial chunk edges, precomputed row and
+column offsets map logical coordinates into zeroed output chunks through the
+same lookup scatter used for LOD. Padding requires no work from the caller.
+CPU scatter caches contiguous run boundaries so rows and groups of rows can
+be copied in bulk.
+
 **Supported element types:** u8, u16, u32, u64, i8, i16, i32, i64, f16, f32, f64
 (see `enum dtype` in `src/dtype.h`).
 
@@ -224,7 +230,8 @@ and multiscale, generated/image inputs, and discard/filesystem/S3/throttled sink
 measured input, and `--duration S` requests a minimum measured append time;
 neither changes that layout. The underlying frame dimension is unbounded. Source
 preparation and stream creation precede warmup. Generated inputs use a fixed 64 MiB source ring. Images use a preloaded,
-chunk-padded ring; loading and padding are excluded from timing.
+tightly packed ring; loading is excluded from timing. Chunk padding is handled
+by the stream's scatter and LOD mapping.
 
 Warmup continues to a full batch boundary with no partial append-downsample
 accumulator, drains earlier work and sink metadata, and resets stage timings,
@@ -240,8 +247,10 @@ TTY and top-level JSON rates/stage metrics describe this same window. The
 `measurement` object records the policy, requested and actual durations, warmup
 and measured work, effective per-dimension geometry, epoch/batch/staging sizes,
 and boundary samples. `throughput_in_gibs` counts submitted bytes;
-`throughput_logical_gibs` excludes image padding. `compression_fold` counts
-full decoded chunks, while `logical_compression_fold` counts logical input.
+`throughput_logical_gibs` counts the same packed image bytes. `compression_fold`
+counts full decoded chunks, while `logical_compression_fold` counts logical input.
+Image results with `source_input_bytes` use packed input; older
+`source_padded_bytes` results include caller padding in submitted-byte rates.
 Version 12 adds images to the common version-11 window. Neither is directly
 comparable to older whole-run or no-drain sustained rates. The specialized two-stream driver
 retains its explicit fixed-frame, whole-run policy and rejects timing options.
@@ -394,7 +403,7 @@ error and unaccepted remainder before recycling the entire camera buffer.
 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:
+Size input capacity from the tightly packed bytes actually submitted:
 
 `capacity >= submitted_bytes_per_second * pause_seconds + active_drain_bytes`
 

bench/bench_input.c

@@ -9,31 +9,20 @@ bench_input_load(struct bench_input* input,
                  const char* path,
                  enum dtype dtype,
                  size_t width,
-                 size_t height,
-                 size_t chunk_width,
-                 size_t chunk_height)
+                 size_t height)
 {
   const uint16_t endian = 1;
   const size_t bpe = dtype_bpe(dtype);
   if (!input || !path ||
       (dtype != dtype_u8 && dtype != dtype_u16 && dtype != dtype_f32) ||
-      !width || !height || !chunk_width || !chunk_height ||
-      width > SIZE_MAX - (chunk_width - 1) ||
-      height > SIZE_MAX - (chunk_height - 1) ||
-      width > SIZE_MAX / bpe / height ||
+      !width || !height || width > SIZE_MAX / bpe / height ||
       (bpe > 1 && *(const unsigned char*)&endian != 1)) {
     fprintf(stderr,
             "Image replay requires little-endian u8, u16, or f32 frames\n");
     return 1;
   }
   *input = (struct bench_input){ 0 };
-  size_t padded_width = (width + chunk_width - 1) / chunk_width * chunk_width;
-  size_t padded_height =
-    (height + chunk_height - 1) / chunk_height * chunk_height;
-  if (padded_width > SIZE_MAX / bpe / padded_height)
-    return 1;
   size_t frame_elements = width * height;
-  size_t padded_frame = padded_width * padded_height;
   struct platform_clock clock = { 0 };
   platform_toc(&clock);
   FILE* file = fopen(path, "rb");
@@ -49,28 +38,11 @@ bench_input_load(struct bench_input* input,
       fseek(file, 0, SEEK_SET))
     goto Fail;
   size_t bytes = (size_t)length;
-  size_t frames = bytes / (frame_elements * bpe);
-  if (frames > SIZE_MAX / bpe / padded_frame)
-    goto Fail;
-  input->elements = frames * padded_frame;
-  input->frame_elements = padded_frame;
-  input->logical_frame_elements = frame_elements;
+  input->frame_elements = frame_elements;
   input->source_bytes = bytes;
-  input->data = calloc(input->elements, bpe);
-  if (!input->data)
+  input->data = malloc(bytes);
+  if (!input->data || fread(input->data, 1, bytes, file) != bytes)
     goto Fail;
-  if (padded_frame == frame_elements) {
-    if (fread(input->data, 1, bytes, file) != bytes)
-      goto Fail;
-  } else {
-    for (size_t frame = 0; frame < frames; ++frame)
-      for (size_t row = 0; row < height; ++row) {
-        unsigned char* data =
-          input->data + (frame * padded_frame + row * padded_width) * bpe;
-        if (fread(data, bpe, width, file) != width)
-          goto Fail;
-      }
-  }
   if (fgetc(file) != EOF || ferror(file))
     goto Fail;
   if (fclose(file)) {

bench/bench_input.h

@@ -9,21 +9,18 @@
 struct bench_input
 {
   unsigned char* data;
-  size_t elements;
   size_t frame_elements;
-  size_t logical_frame_elements;
   size_t source_bytes;
   float load_s;
 };
 
-// Prepared cyclic input. All offsets and sizes are writer bytes; logical
-// frame bytes exclude image edge padding. Preparation precedes measurement.
+// Prepared cyclic input. Frames are tightly packed; chunk padding is supplied
+// by the stream mapping. All offsets and sizes are writer bytes.
 struct bench_source
 {
   const unsigned char* data;
   size_t bytes;
   size_t frame_bytes;
-  size_t logical_frame_bytes;
 };
 
 struct slice
@@ -36,9 +33,7 @@ bench_input_load(struct bench_input* input,
                  const char* path,
                  enum dtype dtype,
                  size_t width,
-                 size_t height,
-                 size_t chunk_width,
-                 size_t chunk_height);
+                 size_t height);
 
 void
 bench_input_free(struct bench_input* input);

bench/bench_report.c

@@ -1031,8 +1031,8 @@ print_bench_json_pass(const struct stream_metrics* m,
     jw_uint(&jw, measurement->append_bytes / dtype_bpe(dtype));
     jw_key(&jw, "source_bytes");
     jw_uint(&jw, images->input->source_bytes);
-    jw_key(&jw, "source_padded_bytes");
-    jw_uint(&jw, images->input->elements * dtype_bpe(dtype));
+    jw_key(&jw, "source_input_bytes");
+    jw_uint(&jw, images->input->source_bytes);
     jw_key(&jw, "order");
     jw_string(&jw, "cyclic");
     jw_key(&jw, "load_s");

bench/bench_util.c

@@ -599,9 +599,8 @@ run_bench(const struct bench_config* cfg)
     if (cfg->input) {
       source = (struct bench_source){
         .data = (const unsigned char*)cfg->input->data,
-        .bytes = cfg->input->elements * bpe,
+        .bytes = cfg->input->source_bytes,
         .frame_bytes = cfg->input->frame_elements * bpe,
-        .logical_frame_bytes = cfg->input->logical_frame_elements * bpe,
       };
       measurement.source_bytes = source.bytes;
       if (!cfg->append_elements)
@@ -617,7 +616,6 @@ run_bench(const struct bench_config* cfg)
       source.frame_bytes = bpe;
       for (uint8_t d = 1; d < rank; ++d)
         source.frame_bytes *= dims[d].size;
-      source.logical_frame_bytes = source.frame_bytes;
     }
     measurement.prep_s = (platform_monotonic_ns() - prep_start) * 1e-9 +
                          (cfg->input ? cfg->input->load_s : 0);
@@ -850,10 +848,7 @@ run_bench(const struct bench_config* cfg)
   const float wall_s = (float)measurement.elapsed_s;
   measurement.output_bytes =
     sink_bytes(&meter, &tss, &dss) - measurement.warmup_output_bytes;
-  measurement.logical_input_bytes = cfg->input ? measurement.input_bytes /
-                                                   source.frame_bytes *
-                                                   source.logical_frame_bytes
-                                               : measurement.input_bytes;
+  measurement.logical_input_bytes = measurement.input_bytes;
 
   if (platform_peak_resident_memory(&mem_used.host_peak_bytes) != 0) {
     log_warn("  host peak memory reading unavailable");
@@ -1344,16 +1339,7 @@ bench_stream_main(int ac, char* av[], struct bench_spec spec)
       return bench_failed(a.json_output);
     }
 
-    const size_t chunk_height = (size_t)dims[1].chunk_size;
-    const size_t chunk_width = (size_t)dims[2].chunk_size;
-    size_t padded_width =
-      (size_t)((a.width + chunk_width - 1) / chunk_width * chunk_width);
-    size_t padded_height =
-      (size_t)((a.height + chunk_height - 1) / chunk_height * chunk_height);
-    if (padded_width > SIZE_MAX / bpe / padded_height)
-      return bench_failed(a.json_output);
-    size_t padded_frame = padded_width * padded_height;
-    if (a.frames > SIZE_MAX / bpe / padded_frame ||
+    if (a.frames > SIZE_MAX / frame_bytes ||
         a.append_elements > SIZE_MAX / bpe) {
       fprintf(stderr, "Image stream size overflows addressable memory\n");
       return bench_failed(a.json_output);
@@ -1364,14 +1350,9 @@ bench_stream_main(int ac, char* av[], struct bench_spec spec)
     if (!a.max_threads)
       a.max_threads = 4;
     if (!a.append_elements)
-      a.append_elements = padded_frame;
-    if (bench_input_load(&input,
-                         a.input_path,
-                         a.dtype,
-                         (size_t)a.width,
-                         (size_t)a.height,
-                         chunk_width,
-                         chunk_height))
+      a.append_elements = frame_elements;
+    if (bench_input_load(
+          &input, a.input_path, a.dtype, (size_t)a.width, (size_t)a.height))
       return bench_failed(a.json_output);
   } else if (a.input_path || a.width || a.height || a.chunk_depth) {
     fprintf(stderr, "Image options require bench_stream_microscopy\n");

bench/test_bench_images.py

@@ -1,4 +1,4 @@
-"""Image replay shares coverage and reports logical/padded work independently."""
+"""Image replay submits packed pixels and pads only the output chunks."""
 
 from array import array
 import json
@@ -46,20 +46,18 @@ def check_type(dtype, typecode, replay_dtype, zarr_dtype):
             geometry = window["geometry"]
             assert replay["dtype"] == replay_dtype
             assert math.prod(replay["chunk_shape"]) * source.itemsize == 32 * 1024
-            cy, cx = replay["chunk_shape"][1:]
-            padded_frame = math.ceil(height / cy) * cy * math.ceil(width / cx) * cx
             measured = replay["shape"][0]
             assert measured >= 7
-            assert result["submitted_bytes"] == window["input_bytes"] == measured * padded_frame * source.itemsize
+            assert result["submitted_bytes"] == window["input_bytes"] == measured * frame * source.itemsize
             assert result["logical_input_bytes"] == measured * frame * source.itemsize
-            assert result["logical_input_bytes"] < result["submitted_bytes"]
+            assert result["logical_input_bytes"] == result["submitted_bytes"]
             assert math.isclose(result["throughput_in_gibs"],
                                 result["submitted_bytes"] / 2**30 / window["elapsed_s"], rel_tol=1e-5)
             assert math.isclose(result["throughput_logical_gibs"],
                                 result["logical_input_bytes"] / 2**30 / window["elapsed_s"], rel_tol=1e-5)
             assert replay["source_bytes"] == len(source) * source.itemsize
-            assert window["source_bytes"] == planes * padded_frame * source.itemsize
-            assert replay["source_padded_bytes"] == window["source_bytes"]
+            assert window["source_bytes"] == planes * frame * source.itemsize
+            assert replay["source_input_bytes"] == window["source_bytes"]
             assert replay["reference_shape"] == [64, height, width]
 
         # Compression keeps this filesystem metadata/index readback small.
@@ -71,7 +69,7 @@ def check_type(dtype, typecode, replay_dtype, zarr_dtype):
         result = json.loads(process.stdout)
         window = result["measurement"]
         metadata = json.loads((output / "images" / "zarr.json").read_text())
-        total_frames = (window["warmup_input_bytes"] + window["input_bytes"]) // (padded_frame * source.itemsize)
+        total_frames = (window["warmup_input_bytes"] + window["input_bytes"]) // (frame * source.itemsize)
         assert metadata["shape"] == [total_frames, height, width]
         assert metadata["data_type"] == zarr_dtype
         assert window["boundary_timing"] is False

docs/buffered-writer.md

@@ -11,15 +11,14 @@ sustained throughput.
 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. 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
+(excluding output chunk padding) times the pause allowance, plus input 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
+For example, 4 GiB/s of tightly packed frames with a 250 ms pause allowance and an
+active 64 MiB drain requires 1,088 MiB when rounded up to
 64 MiB blocks. The camera pool is additional. Tune the sizes below to your workload:
 
 ```c

scripts/datasets/README.md

@@ -186,9 +186,10 @@ Image chunks preserve the default T:Y:X bit ratio `1:4:4`; every requested
 decoded size is checked against the actual layout. All tiers retain the 0.5 GiB
 uncompressed full-shard floor, 1 GiB ceiling, 64 MiB batch target, four workers,
 one scale, and discard sink. Images retain native dimensions. The replay buffer
-is zero-padded to whole chunks before timing; source and padded bytes are
-reported separately. Primary throughput counts submitted bytes including this
-padding; `throughput_logical_gibs` counts only native image bytes. A memory constraint or unattainable chunk target
+contains tightly packed pixels. Scatter and LOD mapping pad output chunks
+internally, so both input throughput fields count native image bytes.
+`source_input_bytes` records the replay buffer size; archived runs used
+`source_padded_bytes` for a caller-padded buffer. A memory constraint or unattainable chunk target
 causes failure instead of silently changing geometry.
 
 The unified runner writes the normal

scripts/datasets/run.py

@@ -280,8 +280,12 @@ def check_result(
     if result["padded_input_bytes"] != padded_frames * padded_frame_bytes:
         raise ValueError("Padded input byte count is wrong")
     source_frames = pack["bytes"] // (pack["height"] * pack["width"] * 2)
-    if replay["source_padded_bytes"] != source_frames * padded_frame_bytes:
-        raise ValueError("Preloaded padded source byte count is wrong")
+    if "source_input_bytes" in replay:
+        source_valid = replay["source_input_bytes"] == pack["bytes"]
+    else:
+        source_valid = replay["source_padded_bytes"] == source_frames * padded_frame_bytes
+    if not source_valid:
+        raise ValueError("Preloaded source byte count is wrong")
     full_shard_bytes = (
         math.prod(replay["chunk_shape"])
         * math.prod(replay["chunks_per_shard"])

scripts/datasets/verify_output.py

@@ -57,25 +57,22 @@ def check_replay(report, images, minimum_frames, backend, profile):
         raise ValueError("Invalid image chunk shape")
     if math.prod(chunk) * images.dtype.itemsize != 32 * 1024:
         raise ValueError("Image chunk target changed")
-    padded_frame = math.prod(
-        (size + step - 1) // step * step
-        for size, step in zip((height, width), chunk[1:])
-    ) * images.dtype.itemsize
-    measured, partial = divmod(window["input_bytes"], padded_frame)
-    warmup, warmup_partial = divmod(window["warmup_input_bytes"], padded_frame)
+    frame_bytes = height * width * images.dtype.itemsize
+    measured, partial = divmod(window["input_bytes"], frame_bytes)
+    warmup, warmup_partial = divmod(window["warmup_input_bytes"], frame_bytes)
     if partial or warmup_partial or measured < minimum_frames:
         raise ValueError("Image replay has partial or insufficient frames")
     expected = {
         "backend": backend, "codec": profile, "dtype": image_dtype(images),
         "shape": [measured, height, width], "source_bytes": images.nbytes,
-        "source_padded_bytes": len(images) * padded_frame, "order": "cyclic",
+        "source_input_bytes": len(images) * frame_bytes, "order": "cyclic",
         "codec_level": 0 if profile == "none" else 3,
         "shuffle": "bit" if profile.startswith("blosc-") else "none",
     }
     for key, value in expected.items():
         if replay.get(key) != value:
             raise ValueError(f"Image replay changed {key}: expected {value}")
-    if (report["submitted_bytes"] != measured * padded_frame
+    if (report["submitted_bytes"] != measured * frame_bytes
             or report["logical_input_bytes"] != measured * height * width * images.dtype.itemsize):
         raise ValueError("Image logical/submitted byte accounting disagrees")
     return warmup + measured

scripts/sweep/microscopy_data.py

@@ -79,11 +79,15 @@ def check_observation(record, task, config, definition, resources=None):
         raise ValueError("Invalid logical byte accounting")
     frames, height, width = replay["shape"]
     frame_bytes = math.ceil(height / chunk[1]) * chunk[1] * math.ceil(width / chunk[2]) * chunk[2] * element_size
+    # Archived observations used caller-padded frames; new runs submit pixels.
+    if "source_input_bytes" in replay:
+        frame_bytes = height * width * element_size
+    source_bytes = replay.get("source_input_bytes", replay.get("source_padded_bytes"))
     if (window["input_bytes"] != frames * frame_bytes
             or result["submitted_bytes"] != window["input_bytes"]
-            or replay["source_padded_bytes"] != window["source_bytes"]
+            or source_bytes != window["source_bytes"]
             or replay["append_elements"] * element_size != window["append_bytes"]):
-        raise ValueError("Invalid padded byte accounting or replay buffer")
+        raise ValueError("Invalid input byte accounting or replay buffer")
     if definition["geometry_frames"] is not None and window["reference_frames"] != definition["geometry_frames"]:
         raise ValueError("Observation changed reference geometry")
     if not math.isfinite(result["process_wall_s"]) or result["process_wall_s"] <= 0: