From 3c2a4a775994d540dd416ae4bfff0b7f206c791c Mon Sep 17 00:00:00 2001 From: Matt Date: Mon, 5 Oct 2026 11:49:46 +0100 Subject: [PATCH 1/2] Fix non atomicity in metric writes in run --- simvue/run.py | 32 +++++++--- tests/functional/test_run_class.py | 97 ++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 7 deletions(-) diff --git a/simvue/run.py b/simvue/run.py index f53ce048..36c1b927 100644 --- a/simvue/run.py +++ b/simvue/run.py @@ -189,6 +189,13 @@ def __init__( self._dispatcher: DispatcherBaseClass | None = None self._meta_cache: dict[str, typing.Any] = {} + # Protects the pending metric units in ``_meta_cache`` which is + # accessed concurrently by ``log_metrics`` (fetch + delete) and + # ``set_metric_units`` (create + insert). Without the lock these + # non-atomic check-then-act sequences can interleave and raise + # ``KeyError: 'metrics'`` when metrics are logged from multiple + # threads (e.g. a file-tail thread and a snapshot poller thread). + self._meta_cache_lock: threading.Lock = threading.Lock() self._folder: Folder | None = None self._term_color: bool = True @@ -1779,10 +1786,17 @@ def log_metrics( ``` """ - # If there are any metric units to be uploaded do so now - if _units := self._meta_cache.get("metrics"): + # If there are any metric units to be uploaded do so now. + # Pop the pending units under the lock so that a concurrent + # ``set_metric_units`` call cannot interleave its + # ``setdefault``/item-assignment between our fetch and delete + # (which would raise ``KeyError: 'metrics'``). The (potentially + # slow) metadata upload is done outside the lock so that + # ``set_metric_units`` calls are not blocked on the I/O. + with self._meta_cache_lock: + _units = self._meta_cache.pop("metrics", None) + if _units: self.update_metadata({"simvue": {"metrics": _units}}) - del self._meta_cache["metrics"] # TODO: When metrics and grids are combined into a single entity # this can be removed. For now need to separate tensor based metrics @@ -2753,18 +2767,22 @@ def set_metric_units( ``` """ - self._meta_cache.setdefault("metrics", {}) - try: _unit_obj: unyt_quantity = unyt_quantity.from_string(units) - self._meta_cache["metrics"][metric_name] = { + _payload = { "units": units, "mks_conversion": mks_conversion or float(_unit_obj.in_mks().value), "mks_units": mks_unit or f"{_unit_obj.in_mks().units}", } except (UnitParseError, ValueError): - self._meta_cache["metrics"][metric_name] = { + _payload = { "units": units, "mks_conversion": mks_conversion, "mks_units": mks_unit, } + + # Register the units under the same lock used by ``log_metrics`` + # so that this check-then-act (setdefault + item-assignment) is + # atomic with respect to the fetch-and-delete performed there. + with self._meta_cache_lock: + self._meta_cache.setdefault("metrics", {})[metric_name] = _payload diff --git a/tests/functional/test_run_class.py b/tests/functional/test_run_class.py index b6ba71f3..2639c7c7 100644 --- a/tests/functional/test_run_class.py +++ b/tests/functional/test_run_class.py @@ -12,6 +12,7 @@ import inspect import tempfile import threading +import traceback import uuid import psutil import pathlib @@ -1779,3 +1780,99 @@ def test_set_metric_units(create_plain_run: tuple[sv_run.Run, dict]) -> None: assert _metric_data["x"] == {"units": "ft", "mks_units": "m", "mks_conversion": 0.3048} assert _metric_data["y"] == {"units": "m", "mks_units": "m", "mks_conversion": 1} assert _metric_data["z"] == {"units": "Foobars", "mks_units": "m", "mks_conversion": 3.542} + + +@pytest.mark.run +@pytest.mark.offline +def test_set_metric_units_thread_safety( + create_plain_run_offline: tuple[sv_run.Run, dict], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Concurrent set_metric_units / log_metrics must not corrupt the units cache. + + Regression test for a thread-safety race on ``Run._meta_cache["metrics"]``: + ``set_metric_units`` (setdefault + item assignment) and ``log_metrics`` + (fetch + delete) used to be non-atomic, so two threads logging metrics + concurrently (e.g. a log-tail thread and a snapshot poller thread) could + interleave and raise ``KeyError: 'metrics'``, killing the logging thread. + + The test forces the fatal interleaving deterministically by freezing + ``log_metrics`` between the cache fetch and the delete (inside + ``update_metadata``) and ``set_metric_units`` between the setdefault and + the item assignment (inside ``unyt`` parsing), then releasing them in + the order fetch / setdefault / delete / item-assignment. + """ + from unyt import unyt_quantity + + run, _ = create_plain_run_offline + + # Seed a pending units entry so log_metrics takes the fetch path + run.set_metric_units("seed", units="m") + assert "metrics" in run._meta_cache + + log_frozen = threading.Event() + log_released = threading.Event() + units_frozen = threading.Event() + units_released = threading.Event() + + _orig_update_metadata = run.update_metadata + + def _blocking_update_metadata(metadata: dict[str, typing.Any]) -> bool: + """Freeze log_metrics after the units fetch, before the cache delete.""" + log_frozen.set() + log_released.wait(timeout=10) + return _orig_update_metadata(metadata) + + _orig_from_string = unyt_quantity.from_string + + def _blocking_from_string( + cls: type, s: str, *args: typing.Any, **kwargs: typing.Any + ): + """Freeze set_metric_units after the setdefault, before the item assignment.""" + units_frozen.set() + units_released.wait(timeout=10) + return _orig_from_string(s, *args, **kwargs) + + monkeypatch.setattr(run, "update_metadata", _blocking_update_metadata) + monkeypatch.setattr( + unyt_quantity, "from_string", classmethod(_blocking_from_string) + ) + + errors: list[str] = [] + + def log_metrics_task() -> None: + try: + run.log_metrics({"a": 1.0}, step=0) + except Exception: + errors.append(f"log_metrics:\n{traceback.format_exc()}") + + def set_metric_units_task() -> None: + try: + run.set_metric_units("b", units="ft") + except Exception: + errors.append(f"set_metric_units:\n{traceback.format_exc()}") + + log_thread = threading.Thread(target=log_metrics_task, name="log-metrics") + log_thread.start() + assert log_frozen.wait(timeout=10), "log_metrics did not reach update_metadata" + + units_thread = threading.Thread( + target=set_metric_units_task, name="set-metric-units" + ) + units_thread.start() + assert units_frozen.wait(timeout=10), "set_metric_units did not reach unit parsing" + + # Release in the fatal order: fetch / setdefault / delete / item-assignment + log_released.set() + log_thread.join(timeout=10) + units_released.set() + units_thread.join(timeout=10) + + assert not log_thread.is_alive() and not units_thread.is_alive() + assert not errors, "concurrent metric logging raised:\n" + "\n".join(errors) + + # log_metrics must have uploaded the seeded units, and the units set by + # set_metric_units must still be pending for the next log_metrics call + assert run._meta_cache.get("metrics") == { + "b": {"units": "ft", "mks_units": "m", "mks_conversion": 0.3048} + } From 66068ae55836bdc717becc2a6a42c7bd17adeb28 Mon Sep 17 00:00:00 2001 From: Matt Date: Mon, 5 Oct 2026 12:06:45 +0100 Subject: [PATCH 2/2] Remove crazy long comments --- simvue/run.py | 19 ++----------------- tests/functional/test_run_class.py | 15 +-------------- 2 files changed, 3 insertions(+), 31 deletions(-) diff --git a/simvue/run.py b/simvue/run.py index 36c1b927..9df2054c 100644 --- a/simvue/run.py +++ b/simvue/run.py @@ -189,12 +189,6 @@ def __init__( self._dispatcher: DispatcherBaseClass | None = None self._meta_cache: dict[str, typing.Any] = {} - # Protects the pending metric units in ``_meta_cache`` which is - # accessed concurrently by ``log_metrics`` (fetch + delete) and - # ``set_metric_units`` (create + insert). Without the lock these - # non-atomic check-then-act sequences can interleave and raise - # ``KeyError: 'metrics'`` when metrics are logged from multiple - # threads (e.g. a file-tail thread and a snapshot poller thread). self._meta_cache_lock: threading.Lock = threading.Lock() self._folder: Folder | None = None @@ -1786,13 +1780,7 @@ def log_metrics( ``` """ - # If there are any metric units to be uploaded do so now. - # Pop the pending units under the lock so that a concurrent - # ``set_metric_units`` call cannot interleave its - # ``setdefault``/item-assignment between our fetch and delete - # (which would raise ``KeyError: 'metrics'``). The (potentially - # slow) metadata upload is done outside the lock so that - # ``set_metric_units`` calls are not blocked on the I/O. + # Ensure this cannot get into a race with set_metric_units with self._meta_cache_lock: _units = self._meta_cache.pop("metrics", None) if _units: @@ -2780,9 +2768,6 @@ def set_metric_units( "mks_conversion": mks_conversion, "mks_units": mks_unit, } - - # Register the units under the same lock used by ``log_metrics`` - # so that this check-then-act (setdefault + item-assignment) is - # atomic with respect to the fetch-and-delete performed there. + # Ensure this cannot get into a race with log_metrics with self._meta_cache_lock: self._meta_cache.setdefault("metrics", {})[metric_name] = _payload diff --git a/tests/functional/test_run_class.py b/tests/functional/test_run_class.py index 2639c7c7..c593ead2 100644 --- a/tests/functional/test_run_class.py +++ b/tests/functional/test_run_class.py @@ -19,6 +19,7 @@ import concurrent.futures import random import datetime +from unyt import unyt_quantity import simvue from simvue.api.objects import Alert, EventsAlert, Metrics, MetricsRangeAlert, MetricsThresholdAlert, UserAlert from simvue.api.objects.grids import GridMetrics @@ -1789,21 +1790,7 @@ def test_set_metric_units_thread_safety( monkeypatch: pytest.MonkeyPatch, ) -> None: """Concurrent set_metric_units / log_metrics must not corrupt the units cache. - - Regression test for a thread-safety race on ``Run._meta_cache["metrics"]``: - ``set_metric_units`` (setdefault + item assignment) and ``log_metrics`` - (fetch + delete) used to be non-atomic, so two threads logging metrics - concurrently (e.g. a log-tail thread and a snapshot poller thread) could - interleave and raise ``KeyError: 'metrics'``, killing the logging thread. - - The test forces the fatal interleaving deterministically by freezing - ``log_metrics`` between the cache fetch and the delete (inside - ``update_metadata``) and ``set_metric_units`` between the setdefault and - the item assignment (inside ``unyt`` parsing), then releasing them in - the order fetch / setdefault / delete / item-assignment. """ - from unyt import unyt_quantity - run, _ = create_plain_run_offline # Seed a pending units entry so log_metrics takes the fetch path