From e6f550da5a6105ba5b37bb76263fd24342aa3acb Mon Sep 17 00:00:00 2001 From: Eran Date: Fri, 21 Aug 2026 01:26:30 -0400 Subject: [PATCH] feat(xarray): default transducer.buffer.size to 600 (was required); port upstream resource-param docs The XArrayEmitter's in-memory buffer flushes to the zarr store every `transducer.buffer.size` emit steps. Sizing it small (3-4) forces a flush every few *simulated* seconds, degrading write latency and compression by ~2 orders of magnitude, because the transport-layer latency is meant to be 2+ orders of magnitude below the time to fill one buffer. - Make `buffer.size` optional; when omitted, default to DEFAULT_BUFFER_SIZE = 600 (a handful of flushes per generation at 1 Hz emission, ~10 min of simulated time per flush). Explicit values are still validated (int > 2). - Port the clarified resource-parameter docs from vEcoli PR #414 (commit febe381): the transducer buffer.size note + example (3 -> 600) and the writer.buffers_per_chunk rule of thumb (interval * size * buffers_per_chunk ~= steps per generation). - Add tests/test_xarray_buffer_default.py covering the default, explicit override, and rejection of invalid explicit sizes. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_xarray_buffer_default.py | 53 ++++++++++++++++++++++ viva_emitters/xarray_emitter/transducer.py | 53 +++++++++++++++++----- viva_emitters/xarray_emitter/writer.py | 38 ++++++++++------ 3 files changed, 118 insertions(+), 26 deletions(-) create mode 100644 tests/test_xarray_buffer_default.py diff --git a/tests/test_xarray_buffer_default.py b/tests/test_xarray_buffer_default.py new file mode 100644 index 0000000..3f9f876 --- /dev/null +++ b/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) diff --git a/viva_emitters/xarray_emitter/transducer.py b/viva_emitters/xarray_emitter/transducer.py index 7ed3513..3415b55 100644 --- a/viva_emitters/xarray_emitter/transducer.py +++ b/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: diff --git a/viva_emitters/xarray_emitter/writer.py b/viva_emitters/xarray_emitter/writer.py index 5d4973f..57f27a2 100644 --- a/viva_emitters/xarray_emitter/writer.py +++ b/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 """