From 669f63811108f6b524ce3b58b41de92139894bc6 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 02:25:22 +0200 Subject: [PATCH] Share one scheduler across Python reactive documents --- packages/cultcache-py/src/cultmesh_py/node.py | 97 +++++++++++++++++-- packages/cultcache-py/tests/test_cultcache.py | 65 ++++++++++--- scripts/verify-reactive-document-scaling.ps1 | 20 +++- 3 files changed, 158 insertions(+), 24 deletions(-) diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 3afddd24..fb302c8c 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -1,6 +1,9 @@ from __future__ import annotations import threading +import heapq +import time +import traceback from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path @@ -87,6 +90,87 @@ class CultMeshReactiveDocumentOptions: replace_dirty_current_on_canonical_snapshot: bool = False +class _CultMeshScheduledCall: + def __init__(self, callback: Callable[[], None], wake: Callable[[], None]) -> None: + self._callback: Callable[[], None] | None = callback + self._wake = wake + self._lock = threading.Lock() + + def cancel(self) -> None: + with self._lock: + self._callback = None + self._wake() + + def take(self) -> Callable[[], None] | None: + with self._lock: + callback = self._callback + self._callback = None + return callback + + @property + def cancelled(self) -> bool: + with self._lock: + return self._callback is None + + +class _CultMeshReactiveScheduler: + """One lazy daemon worker owns delayed reactive flushes for the process.""" + + def __init__(self) -> None: + self._condition = threading.Condition() + self._queue: list[tuple[float, int, _CultMeshScheduledCall]] = [] + self._sequence = 0 + self._worker: threading.Thread | None = None + + def schedule(self, delay_seconds: float, callback: Callable[[], None]) -> _CultMeshScheduledCall: + call = _CultMeshScheduledCall(callback, self._wake) + with self._condition: + self._sequence += 1 + heapq.heappush( + self._queue, + (time.monotonic() + max(0.0, delay_seconds), self._sequence, call), + ) + if self._worker is None: + self._worker = threading.Thread( + target=self._run, + name="cultmesh-reactive-scheduler", + daemon=True, + ) + self._worker.start() + self._condition.notify() + return call + + def _wake(self) -> None: + with self._condition: + self._condition.notify() + + def _run(self) -> None: + while True: + callback: Callable[[], None] | None = None + with self._condition: + while callback is None: + while self._queue and self._queue[0][2].cancelled: + heapq.heappop(self._queue) + if not self._queue: + self._condition.wait() + continue + deadline, _, call = self._queue[0] + remaining = deadline - time.monotonic() + if remaining > 0: + self._condition.wait(remaining) + continue + heapq.heappop(self._queue) + callback = call.take() + if callback is not None: + try: + callback() + except Exception: + traceback.print_exc() + + +_REACTIVE_SCHEDULER = _CultMeshReactiveScheduler() + + @dataclass class CultMeshNode: cache: CultCache = field(default_factory=CultCache) @@ -1373,7 +1457,7 @@ class CultMeshReactiveDocument: self._dirty = False self._flushing = False self._flush_queued = False - self._flush_timer: threading.Timer | None = None + self._flush_timer: _CultMeshScheduledCall | None = None self._reconciliation_version = 0 self.reconciliation: CultMeshReactiveDocumentReconciliation | None = None self._current = self._clone(self._database.get_required(self._document, self._key)) @@ -1503,15 +1587,12 @@ class CultMeshReactiveDocument: self.reconciliation = None def _schedule_flush_locked(self) -> None: - if self._options.flush_delay_seconds <= 0: - threading.Thread(target=self.flush, daemon=True).start() - return if self._flush_timer is not None: self._flush_timer.cancel() - timer = threading.Timer(self._options.flush_delay_seconds, self.flush) - timer.daemon = True - self._flush_timer = timer - timer.start() + self._flush_timer = _REACTIVE_SCHEDULER.schedule( + self._options.flush_delay_seconds, + self.flush, + ) def _clone(self, value: Any) -> Any: return self._document.decode_payload(self._document.encode_payload(value)) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index f753b12e..82a00845 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -2595,23 +2595,27 @@ class CultCacheTests(unittest.TestCase): scheduled_timers: list[Any] = [] - class FakeTimer: - def __init__(self, interval: float, callback: Callable[[], None]) -> None: - self.interval = interval + class FakeScheduledCall: + def __init__(self, delay_seconds: float, callback: Callable[[], None]) -> None: + self.delay_seconds = delay_seconds self.callback = callback - self.daemon = False self.cancelled = False - self.started = False - scheduled_timers.append(self) - - def start(self) -> None: - self.started = True def cancel(self) -> None: self.cancelled = True + class FakeScheduler: + def schedule( + self, + delay_seconds: float, + callback: Callable[[], None], + ) -> FakeScheduledCall: + call = FakeScheduledCall(delay_seconds, callback) + scheduled_timers.append(call) + return call + reactive_documents: list[Any] = [] - with patch("cultmesh_py.node.threading.Timer", FakeTimer): + with patch("cultmesh_py.node._REACTIVE_SCHEDULER", FakeScheduler()): try: for index in range(document_count): reactive_documents.append( @@ -2634,7 +2638,6 @@ class CultCacheTests(unittest.TestCase): ) self.assertEqual(len(scheduled_timers), changed_document_count) - self.assertTrue(all(timer.started for timer in scheduled_timers)) self.assertEqual( sum(reactive.is_dirty for reactive in reactive_documents), changed_document_count, @@ -2645,6 +2648,46 @@ class CultCacheTests(unittest.TestCase): self.assertTrue(all(timer.cancelled for timer in scheduled_timers)) + def test_cultmesh_reactive_documents_share_one_lazy_scheduler_thread(self) -> None: + document = define_document_type( + "mesh.reactive_shared_scheduler_note", + encode=lambda value: value, + decode=lambda value: value, + ) + node = create_node(runtime_id="python-reactive-shared-scheduler") + node.register_document(document) + reactive_documents: list[Any] = [] + try: + for index in range(100): + node.put(document, f"note:{index}", {"body": "initial", "revision": 1}) + reactive_documents.append( + node.authoritative_writer(document, f"note:{index}").reactive( + CultMeshReactiveDocumentOptions(flush_delay_seconds=60.0) + ) + ) + + before = sum( + thread.name == "cultmesh-reactive-scheduler" + for thread in threading.enumerate() + ) + for index, reactive in enumerate(reactive_documents): + reactive.update( + lambda draft, changed=index: draft.update( + body=f"changed-{changed}", + revision=2, + ) + ) + after = sum( + thread.name == "cultmesh-reactive-scheduler" + for thread in threading.enumerate() + ) + + self.assertLessEqual(before, 1) + self.assertEqual(after, 1) + finally: + for reactive in reactive_documents: + reactive.dispose() + def test_cultmesh_observed_document_cannot_publish_borrowed_mutation(self) -> None: document = define_document_type( "mesh.observed_note", diff --git a/scripts/verify-reactive-document-scaling.ps1 b/scripts/verify-reactive-document-scaling.ps1 index 1729ed4d..e3555ae2 100644 --- a/scripts/verify-reactive-document-scaling.ps1 +++ b/scripts/verify-reactive-document-scaling.ps1 @@ -1,12 +1,22 @@ param( - [string] $PythonPath = "" + [string] $PythonPath = "", + [string] $NodePath = "" ) $ErrorActionPreference = "Stop" $root = Split-Path -Parent $PSScriptRoot $dotnet = Get-Command dotnet -ErrorAction Stop -$node = Get-Command node -ErrorAction Stop +$nodeExecutable = if ([string]::IsNullOrWhiteSpace($NodePath)) { + $node = Get-Command node -ErrorAction SilentlyContinue + if ($node) { $node.Source } else { Join-Path $env:ProgramFiles "nodejs\node.exe" } +} +else { + [IO.Path]::GetFullPath($NodePath) +} +if (-not (Test-Path -LiteralPath $nodeExecutable -PathType Leaf)) { + throw "Node.js executable not found: $nodeExecutable" +} $pythonExecutable = if ([string]::IsNullOrWhiteSpace($PythonPath)) { (Get-Command python -ErrorAction Stop).Source } @@ -31,18 +41,18 @@ if ($LASTEXITCODE -ne 0) { Push-Location (Join-Path $root "packages\cultmesh-ts") try { - & $node.Source $typescript -p tsconfig.json --pretty false + & $nodeExecutable $typescript -p tsconfig.json --pretty false if ($LASTEXITCODE -ne 0) { throw "CultMesh TypeScript source build failed." } - & $node.Source $typescript -p tsconfig.test.json --pretty false + & $nodeExecutable $typescript -p tsconfig.test.json --pretty false if ($LASTEXITCODE -ne 0) { throw "CultMesh TypeScript test build failed." } $testFiles = Get-ChildItem -LiteralPath .\dist-test\test -Filter "*.test.js" | Sort-Object Name | ForEach-Object FullName - & $node.Source --test @testFiles + & $nodeExecutable --test @testFiles if ($LASTEXITCODE -ne 0) { throw "CultMesh TypeScript reactive scaling verification failed." }