Default transducer.buffer.size to 600; port upstream resource-param docs

tests/test_xarray_buffer_default.py

@@ -0,0 +1,53 @@
+"""``transducer.buffer.size`` is optional and defaults to a sane, large value.
+
+Regression guard for the "lost in translation" vendoring bug: a tiny buffer
+(e.g. 3--4 emit steps) forces the ``XarrayBuffer`` to flush every few simulated
+seconds, degrading latency and compression by ~2 orders of magnitude. The
+library therefore requires no explicit ``buffer.size`` and, when omitted, uses
+:py:data:`~viva_emitters.xarray_emitter.transducer.DEFAULT_BUFFER_SIZE` — sized
+to flush only a handful of times per generation.
+"""
+
+import copy
+
+import pytest
+
+from viva_emitters.xarray_emitter.transducer import (
+    DEFAULT_BUFFER_SIZE,
+    XarrayTransducer,
+)
+
+
+def test_default_is_large_enough_to_avoid_pathological_flushing():
+    # Flushing every few seconds is the bug we are guarding against; the default
+    # should span many minutes of simulated time at a 1 Hz emission rate.
+    assert DEFAULT_BUFFER_SIZE >= 100
+
+
+def test_omitting_buffer_size_uses_default(minimal_xarray_config):
+    config = copy.deepcopy(minimal_xarray_config)
+    del config["transducer"]["buffer"]["size"]
+    transducer = XarrayTransducer(config)
+    assert transducer.buf_size == DEFAULT_BUFFER_SIZE
+
+
+def test_omitting_buffer_key_entirely_uses_default(minimal_xarray_config):
+    config = copy.deepcopy(minimal_xarray_config)
+    del config["transducer"]["buffer"]
+    transducer = XarrayTransducer(config)
+    assert transducer.buf_size == DEFAULT_BUFFER_SIZE
+
+
+def test_explicit_buffer_size_is_honored(minimal_xarray_config):
+    config = copy.deepcopy(minimal_xarray_config)
+    config["transducer"]["buffer"]["size"] = 512
+    transducer = XarrayTransducer(config)
+    assert transducer.buf_size == 512
+
+
+@pytest.mark.parametrize("bad_size", [2, 0, -1, 1.5, "600"])
+def test_invalid_explicit_buffer_size_still_rejected(minimal_xarray_config, bad_size):
+    config = copy.deepcopy(minimal_xarray_config)
+    config["transducer"]["buffer"]["size"] = bad_size
+    with pytest.raises(TypeError):
+        XarrayTransducer(config)

viva_emitters/xarray_emitter/transducer.py

@@ -29,6 +29,20 @@
     from .writer import AsyncBufferWriter
 
 
+#: Default number of *emit steps* held in memory by :py:class:`.XarrayBuffer`
+#: before a flush, used when ``transducer.buffer.size`` is omitted.
+#:
+#: Sized so that, at a 1 Hz emission rate (``subsample.interval == 1``), the
+#: buffer flushes roughly every 10 min of *simulated* time — a handful of
+#: flushes per cell generation. That is the regime the transducer/writer
+#: pipeline is designed for: the transport-layer latency is 2+ orders of
+#: magnitude below the wall-clock time needed to fill one buffer, so flushing
+#: is cheap relative to simulation. Small buffers (e.g. 3--4) instead force a
+#: flush every few simulated seconds, which is pathological for both latency
+#: and compression ratio. See :py:class:`.XarrayTransducer` for how to tune it.
+DEFAULT_BUFFER_SIZE = 600
+
+
 # ==============================================================================
 
 
@@ -546,7 +560,7 @@ class XarrayTransducer:
       {
         "predicate": [...],
         "buffer": {
-          "size": 3
+          "size": 600
         }
       }
 
@@ -555,14 +569,26 @@ class XarrayTransducer:
       - ``predicate`` defines the criterion for which *simulation steps* also
         become *emit steps*, and is parsed by
         :py:class:`.ConjunctiveEmitPredicate`,
-      - while ``size`` is the number of *emit steps* stored in memory by
-        :py:class:`.XarrayBuffer`.
+      - while ``buffer.size`` is the number of *emit steps* stored in memory by
+        :py:class:`.XarrayBuffer` between flushes to the transport layer. It is
+        optional; when omitted it defaults to :py:data:`.DEFAULT_BUFFER_SIZE`.
 
     .. note::
-      The parameter ``size`` is intended to constrain the memory cost of each
-      simulation process, when many parallel simulations are executed in
-      parallel on a node with shared memory. Within that memory budget, larger
-      buffer sizes will result in fewer calls to the transport layer.
+      The parameter ``buffer.size`` is intended to constrain the memory cost of
+      each simulation process, when many parallel simulations are executed in
+      parallel on a node with shared memory. Within that memory budget, **larger
+      buffer sizes** directly translate into **fewer calls to the transport
+      layer**.
+
+      As a rule of thumb, size the buffer so that it flushes only a handful of
+      times per cell generation: the transport-layer latency is 2+ orders of
+      magnitude below the *simulation runtime* needed to fill one buffer, so
+      there is no benefit to flushing more often, and a small buffer (e.g. the
+      ``size == 3`` used by fast CI tests) forces a flush every few *simulated*
+      seconds — pathological for both latency and compression ratio. See
+      :py:data:`.DEFAULT_BUFFER_SIZE` for the omitted-value default, and
+      :py:class:`.AsyncBufferWriter` for how ``writer.buffers_per_chunk`` then
+      maps buffers onto persistent chunk files.
     """
 
     __slots__ = (
@@ -581,8 +607,11 @@ def __init__(self, config: dict[str, Any], /, *, debug: bool=False) -> None:
         self.buffer: XarrayBuffer = XarrayBuffer(view, emit_root)
         """ In-memory cyclic buffer for simulation data. """
 
-        self.buf_size: int = _config["buffer"]["size"]
-        """ Size of time dimension. """
+        self.buf_size: int = (
+            (_config.get("buffer") or {}).get("size") or DEFAULT_BUFFER_SIZE
+        )
+        """ Size of time dimension. Defaults to :py:data:`.DEFAULT_BUFFER_SIZE`
+        when ``transducer.buffer.size`` is omitted. """
         self.buf_tix: int = 0
         """
         Current relative *emit step* inside the cyclic buffer; advanced at the
@@ -602,10 +631,10 @@ def validate_config(cls, config: dict[str, Any], /) -> None:
             case None:
                 raise KeyError(emitter_arg_error(
                     cls, "Missing argument", "\"buffer\": {\"size\": ...}"))
-        match config.get("buffer", {}).get("size"):
+        match (config.get("buffer") or {}).get("size"):
             case None:
-                raise KeyError(emitter_arg_error(
-                    cls, "Missing argument", "\"buffer\": {\"size\": ...}"))
+                # Optional: XarrayTransducer falls back to DEFAULT_BUFFER_SIZE.
+                pass
             case int(buf_size) if buf_size > 2:
                 pass
             case buf_size:

viva_emitters/xarray_emitter/writer.py

@@ -136,25 +136,35 @@ class AsyncBufferWriter[StoreT](ABC):
       - ``store`` is a URI interpreted by the backend,
       - ``threaded`` toggles the use of a separate writer thread,
       - ``buffers_per_chunk`` is the integer-valued size ratio, in terms of
-        *emit step* counts, between one chunk of backend storage and one
-        in-memory buffer,
+        *emit step* counts, between one *backend storage chunk* and one
+        *in-memory buffer*,
       - ``backend`` determines the transport layer subclass instantiated by
         :py:meth:`.dispatch`, e.g., :py:class:`.AsyncZarrBufferWriter`,
       - and ``backend_config`` is interpreted by the backend subclass.
 
     .. note::
-      The parameter ``buffers_per_chunk`` is intended to decouple the number of
-      output files from the choice of ``transducer.buffer.size`` in
-      :py:class:`.XarrayTransducer`. As a rule of thumb, 1 chunk file per
-      variable per generation is desirable in order to minimize the file system
-      pressure, unless a downstream application can benefit from smaller file
-      sizes.
-
-      The latter situation appears to be unlikely under current simulation use
-      cases. However, it may be supported in the future by extending the writer
-      configuration to further distinguish between *chunks* and *shards*, which
-      is `supported`_ by backends like Zarr.
-
+      Once the *in-memory* ``transducer.buffer.size`` in
+      :py:class:`.XarrayTransducer` has been set according to memory constraints
+      and transport latencies, the *number of output files* can be **configured
+      independently** via the parameter ``writer.buffers_per_chunk``.
+
+      As a rule of thumb:
+
+      - For **immutable** object storage systems (e.g., Amazon S3 Standard
+        storage class), ``writer.buffers_per_chunk`` must be set to 1, in order
+        to avoid *copying previous objects* when appending to them.
+      - For **local** or **HPC** file systems, as well as for **appendable**
+        object storage systems (e.g., `Amazon S3 Express One Zone`_ storage
+        class), 1 chunk file *per variable per generation* is desirable in order
+        to minimize the file system pressure --- this amounts to choosing
+        ``transducer.predicate.subsample.interval * transducer.buffer.size *
+        writer.buffers_per_chunk`` to approximately equal the expected number of
+        *simulation steps per generation*.
+      - In case future downstream applications require smaller chunk sizes, the
+        writer configuration may be extended to further distinguish between
+        *chunks* and *shards*, which is `supported`_ by backends like Zarr.
+
+    .. _Amazon S3 Express One Zone: https://docs.aws.amazon.com/AmazonS3/latest/userguide/directory-buckets-objects-append.html
     .. _supported: https://zarr.readthedocs.io/en/latest/user-guide/performance/#sharding
     """