Measure reactive performance across runtimes

.github/workflows/cultmesh-portability.yml

@@ -50,7 +50,7 @@ jobs:
 
       - name: Measure representative reactive workload
         shell: pwsh
-        run: ./scripts/measure-reactive-document-performance.ps1 -Quick
+        run: ./scripts/measure-reactive-document-performance.ps1 -Quick -SkipRuntimeBuild
 
       - name: Churn 100,000 distinct document leases
         run: >-

packages/cultcache-py/tools/reactive_performance_probe.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import argparse
+import json
+import threading
+import time
+import tracemalloc
+from typing import Any
+
+from cultcache_py import define_document_type
+from cultmesh_py import CultMeshReactiveDocumentOptions, create_node
+
+
+def percentile(ordered: list[float], ratio: float) -> float:
+    if not ordered:
+        return 0.0
+    index = max(0, min(len(ordered) - 1, int(len(ordered) * ratio + 0.999999) - 1))
+    return ordered[index]
+
+
+def measure(document_count: int, idle_seconds: float, active_seconds: float) -> dict[str, Any]:
+    payload = "x" * (16 * 1024)
+    changed_document_count = max(1, document_count // 100)
+    document = define_document_type(
+        f"gamecult.mesh.python_performance_probe_{document_count}",
+        encode=lambda value: value,
+        decode=lambda value: value,
+    )
+    node = create_node(runtime_id=f"python-performance-{document_count}")
+    node.register_document(document)
+    reactive_documents = []
+    publish_events = [threading.Event() for _ in range(document_count)]
+    update_started_at = [0.0] * document_count
+    latencies: list[float] = []
+    payload_bytes_published = 0
+    publishes = 0
+    measurement_lock = threading.Lock()
+
+    for index in range(document_count):
+        key = f"performance:{index}"
+        node.put(document, key, {"id": key, "payload": payload, "revision": 1})
+        writer = node.authoritative_writer(document, key)
+        original_write = writer.write
+
+        def measured_write(value: Any, document_index: int = index, write=original_write) -> None:
+            nonlocal payload_bytes_published, publishes
+            encoded_bytes = len(document.encode_payload(value))
+            latency_ms = (time.perf_counter() - update_started_at[document_index]) * 1000
+            with measurement_lock:
+                payload_bytes_published += encoded_bytes
+                publishes += 1
+                latencies.append(latency_ms)
+            write(value)
+            publish_events[document_index].set()
+
+        writer.write = measured_write  # type: ignore[method-assign]
+        reactive_documents.append(
+            writer.reactive(CultMeshReactiveDocumentOptions(flush_delay_seconds=0.0))
+        )
+
+    tracemalloc.start()
+    idle_cpu_start = time.process_time()
+    time.sleep(idle_seconds)
+    idle_current_bytes, idle_peak_bytes = tracemalloc.get_traced_memory()
+    idle_cpu_ms = (time.process_time() - idle_cpu_start) * 1000
+    tracemalloc.reset_peak()
+    active_current_start, _ = tracemalloc.get_traced_memory()
+    active_cpu_start = time.process_time()
+    active_started_at = time.perf_counter()
+    frames = 0
+    while time.perf_counter() - active_started_at < active_seconds:
+        for index in range(changed_document_count):
+            publish_events[index].clear()
+            update_started_at[index] = time.perf_counter()
+            reactive_documents[index].update(
+                lambda draft: draft.__setitem__("revision", draft["revision"] + 1)
+            )
+        for index in range(changed_document_count):
+            if not publish_events[index].wait(timeout=1.0):
+                raise RuntimeError(f"Python reactive publication timed out for document {index}")
+        frames += 1
+        remaining = frames / 60 - (time.perf_counter() - active_started_at)
+        if remaining > 0:
+            time.sleep(remaining)
+
+    active_elapsed_seconds = time.perf_counter() - active_started_at
+    active_cpu_ms = (time.process_time() - active_cpu_start) * 1000
+    active_current_bytes, active_peak_bytes = tracemalloc.get_traced_memory()
+    tracemalloc.stop()
+    for reactive in reactive_documents:
+        reactive.dispose()
+    ordered = sorted(latencies)
+    return {
+        "documentCount": document_count,
+        "changedDocumentCount": changed_document_count,
+        "frames": frames,
+        "expectedPublishes": frames * changed_document_count,
+        "actualPublishes": publishes,
+        "payloadBytesPublished": payload_bytes_published,
+        "idleTracedCurrentBytes": idle_current_bytes,
+        "idleTracedPeakBytes": idle_peak_bytes,
+        "idleCpuMilliseconds": idle_cpu_ms,
+        "activeTracedRetainedGrowthBytes": active_current_bytes - active_current_start,
+        "activeTracedPeakGrowthBytes": active_peak_bytes - active_current_start,
+        "activeCpuMilliseconds": active_cpu_ms,
+        "publishedPayloadBytesPerSecond": payload_bytes_published / active_elapsed_seconds,
+        "p50PublishLatencyMilliseconds": percentile(ordered, 0.50),
+        "p95PublishLatencyMilliseconds": percentile(ordered, 0.95),
+        "p99PublishLatencyMilliseconds": percentile(ordered, 0.99),
+        "schedulerThreads": sum(
+            thread.name == "cultmesh-reactive-scheduler" for thread in threading.enumerate()
+        ),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--quick", action="store_true")
+    args = parser.parse_args()
+    idle_seconds = 0.25 if args.quick else 10.0
+    active_seconds = 0.5 if args.quick else 10.0
+    results = [measure(count, idle_seconds, active_seconds) for count in (1, 100, 1000)]
+    print(json.dumps({
+        "runtime": "python",
+        "workload": {
+            "documentCounts": [1, 100, 1000],
+            "payloadBytes": 16 * 1024,
+            "idleSeconds": idle_seconds,
+            "activeSeconds": active_seconds,
+            "updateRateHz": 60,
+            "changedFraction": 0.01,
+        },
+        "results": results,
+    }, indent=2))
+    return 0 if all(
+        result["actualPublishes"] == result["expectedPublishes"]
+        and result["p99PublishLatencyMilliseconds"] < 250
+        and result["activeTracedPeakGrowthBytes"] < 128 * 1024 * 1024
+        and result["schedulerThreads"] == 1
+        for result in results
+    ) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

packages/cultmesh-ts/src/index.ts

@@ -1312,8 +1312,11 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
       return;
     }
 
-    const timer = setTimeout(flush, 0);
-    this.#cancelScheduledFlush = () => clearTimeout(timer);
+    let cancelled = false;
+    queueMicrotask(() => {
+      if (!cancelled) flush();
+    });
+    this.#cancelScheduledFlush = () => { cancelled = true; };
   }
 
   private flushNow(): void {

packages/cultmesh-ts/test/cultmesh.test.ts

@@ -572,6 +572,42 @@ test("CultMesh TS reactive documents coalesce same-frame explicit updates", asyn
   }
 });
 
+test("CultMesh TS headless reactive documents coalesce on one microtask", async () => {
+  const previousRequestAnimationFrame = globalThis.requestAnimationFrame;
+  delete (globalThis as unknown as {
+    requestAnimationFrame?: (callback: FrameRequestCallback) => number;
+  }).requestAnimationFrame;
+  let current = { noteId: "note:reactive-headless", body: "initial" };
+  const predictions: string[] = [];
+  const document = CultMesh.document(
+    "cultmesh.note:reactive-headless",
+    noteDocument,
+    async () => current,
+    {
+      submitPrediction: async (_context, value) => {
+        predictions.push(value.body);
+        current = value;
+      },
+    },
+  );
+  const reactive = document.predictionWriter().reactive({ watch: false });
+  try {
+    await reactive.ready;
+    reactive.update(value => { value.body = "microtask-1"; });
+    reactive.update(value => { value.body = "microtask-2"; });
+    reactive.update(value => { value.body = "microtask-3"; });
+    assert.equal(predictions.length, 0, "the current turn remains coalesced");
+
+    await waitFor(() => predictions.length === 1, "headless microtask prediction");
+    assert.deepEqual(predictions, ["microtask-3"]);
+  } finally {
+    reactive.dispose();
+    if (previousRequestAnimationFrame) {
+      globalThis.requestAnimationFrame = previousRequestAnimationFrame;
+    }
+  }
+});
+
 test("CultMesh TS reactive scheduling scales with changed documents only", async () => {
   const previousRequestAnimationFrame = globalThis.requestAnimationFrame;
   const previousCancelAnimationFrame = globalThis.cancelAnimationFrame;

scripts/measure-reactive-document-performance.mjs

@@ -0,0 +1,141 @@
+import { monitorEventLoopDelay, performance } from "node:perf_hooks";
+import process from "node:process";
+
+import { CultMesh } from "../packages/cultmesh-ts/dist/index.js";
+
+const quick = process.argv.includes("--quick");
+const idleMs = quick ? 250 : 10_000;
+const activeMs = quick ? 500 : 10_000;
+const results = [];
+
+for (const documentCount of [1, 100, 1000]) {
+  results.push(await measure(documentCount));
+}
+
+console.log(JSON.stringify({
+  runtime: "typescript",
+  workload: {
+    documentCounts: [1, 100, 1000],
+    payloadBytes: 16 * 1024,
+    idleSeconds: idleMs / 1000,
+    activeSeconds: activeMs / 1000,
+    updateRateHz: 60,
+    changedFraction: 0.01,
+  },
+  results,
+}, undefined, 2));
+
+if (results.some(result =>
+  result.actualPublishes !== result.expectedPublishes ||
+  result.p99PublishLatencyMilliseconds >= 250 ||
+  result.activeHeapPeakGrowthBytes >= 128 * 1024 * 1024)) {
+  process.exitCode = 1;
+}
+
+async function measure(documentCount) {
+  const payload = "x".repeat(16 * 1024);
+  const changedDocumentCount = Math.max(1, Math.floor(documentCount / 100));
+  const reactiveDocuments = [];
+  const publishSignals = new Array(documentCount);
+  const updateStartedAt = new Float64Array(documentCount);
+  const latencies = [];
+  let payloadBytesPublished = 0;
+  let publishes = 0;
+
+  for (let index = 0; index < documentCount; index++) {
+    let current = { id: `performance:${index}`, payload, revision: 1 };
+    let watcher;
+    const document = CultMesh.document(
+      `performance:${index}`,
+      { schemaId: "gamecult.mesh.performance_probe.v1" },
+      async () => current,
+      {
+        submitPrediction: async (_context, value) => {
+          payloadBytesPublished += Buffer.byteLength(JSON.stringify(value));
+          latencies.push(performance.now() - updateStartedAt[index]);
+          publishes++;
+          current = value;
+          watcher?.(value);
+          publishSignals[index]?.();
+        },
+        watchDocument: (_context, callback) => {
+          watcher = callback;
+          return () => { watcher = undefined; };
+        },
+      },
+    );
+    const reactive = document.predictionWriter().reactive();
+    reactiveDocuments.push(reactive);
+  }
+
+  await Promise.all(reactiveDocuments.map(document => document.ready));
+  collect();
+  const idleHeapStart = process.memoryUsage().heapUsed;
+  const idleCpuStart = process.cpuUsage();
+  await delay(idleMs);
+  collect();
+  const idleHeapGrowthBytes = process.memoryUsage().heapUsed - idleHeapStart;
+  const idleCpu = process.cpuUsage(idleCpuStart);
+
+  collect();
+  const activeHeapStart = process.memoryUsage().heapUsed;
+  let activeHeapPeak = activeHeapStart;
+  const activeCpuStart = process.cpuUsage();
+  const eventLoop = monitorEventLoopDelay({ resolution: 10 });
+  eventLoop.enable();
+  const activeStartedAt = performance.now();
+  let frames = 0;
+  while (performance.now() - activeStartedAt < activeMs) {
+    const completions = [];
+    for (let index = 0; index < changedDocumentCount; index++) {
+      completions.push(new Promise(resolve => { publishSignals[index] = resolve; }));
+      updateStartedAt[index] = performance.now();
+      reactiveDocuments[index].update(value => { value.revision++; });
+    }
+    await Promise.all(completions);
+    frames++;
+    activeHeapPeak = Math.max(activeHeapPeak, process.memoryUsage().heapUsed);
+    const remaining = frames * (1000 / 60) - (performance.now() - activeStartedAt);
+    if (remaining > 0) await delay(remaining);
+  }
+  eventLoop.disable();
+  const activeElapsedSeconds = (performance.now() - activeStartedAt) / 1000;
+  const activeCpu = process.cpuUsage(activeCpuStart);
+  const ordered = latencies.toSorted((left, right) => left - right);
+
+  for (const document of reactiveDocuments) document.dispose();
+
+  return {
+    documentCount,
+    changedDocumentCount,
+    frames,
+    expectedPublishes: frames * changedDocumentCount,
+    actualPublishes: publishes,
+    payloadBytesPublished,
+    idleHeapGrowthBytes,
+    idleCpuMilliseconds: (idleCpu.user + idleCpu.system) / 1000,
+    activeHeapPeakGrowthBytes: activeHeapPeak - activeHeapStart,
+    activeCpuMilliseconds: (activeCpu.user + activeCpu.system) / 1000,
+    publishedPayloadBytesPerSecond: payloadBytesPublished / activeElapsedSeconds,
+    p50PublishLatencyMilliseconds: percentile(ordered, 0.50),
+    p95PublishLatencyMilliseconds: percentile(ordered, 0.95),
+    p99PublishLatencyMilliseconds: percentile(ordered, 0.99),
+    p99EventLoopDelayMilliseconds: eventLoop.percentile(99) / 1_000_000,
+  };
+}
+
+function percentile(ordered, ratio) {
+  if (ordered.length === 0) return 0;
+  return ordered[Math.max(0, Math.ceil(ordered.length * ratio) - 1)];
+}
+
+function collect() {
+  if (typeof globalThis.gc !== "function") {
+    throw new Error("Run this probe with node --expose-gc.");
+  }
+  globalThis.gc();
+}
+
+function delay(milliseconds) {
+  return new Promise(resolve => setTimeout(resolve, milliseconds));
+}

scripts/measure-reactive-document-performance.ps1

@@ -1,9 +1,43 @@
 param(
-    [switch] $Quick
+    [switch] $Quick,
+    [switch] $SkipRuntimeBuild,
+    [string] $PythonPath = "",
+    [string] $NodePath = ""
 )
 
 $ErrorActionPreference = "Stop"
 $root = Split-Path -Parent $PSScriptRoot
+$nodeExecutable = if ([string]::IsNullOrWhiteSpace($NodePath)) {
+    (Get-Command node -ErrorAction Stop).Source
+}
+else {
+    [IO.Path]::GetFullPath($NodePath)
+}
+$pythonExecutable = if ([string]::IsNullOrWhiteSpace($PythonPath)) {
+    (Get-Command python -ErrorAction Stop).Source
+}
+else {
+    [IO.Path]::GetFullPath($PythonPath)
+}
+foreach ($executable in @($nodeExecutable, $pythonExecutable)) {
+    if (-not (Test-Path -LiteralPath $executable -PathType Leaf)) {
+        throw "Performance probe runtime not found: $executable"
+    }
+}
+
+if (-not $SkipRuntimeBuild) {
+    $typescript = Join-Path $root "node_modules\typescript\bin\tsc"
+    foreach ($dependency in @("cultcache-ts", "cultnet-ts", "cultmesh-ts")) {
+        $dependencyRoot = Join-Path $root "packages\$dependency"
+        if ($dependency -eq "cultnet-ts") {
+            & $nodeExecutable (Join-Path $dependencyRoot "tools\generate-swarm-contracts.mjs")
+            if ($LASTEXITCODE -ne 0) { throw "CultNet TypeScript contract generation failed." }
+        }
+        & $nodeExecutable $typescript -p (Join-Path $dependencyRoot "tsconfig.json") --pretty false
+        if ($LASTEXITCODE -ne 0) { throw "TypeScript performance dependency build failed: $dependency" }
+    }
+}
+
 $project = Join-Path $root "tools\GameCult.Mesh.PerformanceProbe\GameCult.Mesh.PerformanceProbe.csproj"
 $buildArguments = @(
     "build",
@@ -24,5 +58,31 @@ if ($Quick) {
 
 & dotnet @arguments
 if ($LASTEXITCODE -ne 0) {
-    throw "CultMesh reactive document performance probe failed."
+    throw "CultMesh C# reactive document performance probe failed."
+}
+
+$runtimeArguments = @()
+if ($Quick) { $runtimeArguments += "--quick" }
+
+& $nodeExecutable --expose-gc (Join-Path $root "scripts\measure-reactive-document-performance.mjs") @runtimeArguments
+if ($LASTEXITCODE -ne 0) {
+    throw "CultMesh TypeScript reactive document performance probe failed."
+}
+
+$pythonSource = Join-Path $root "packages\cultcache-py\src"
+$previousPythonPath = $env:PYTHONPATH
+$env:PYTHONPATH = if ([string]::IsNullOrWhiteSpace($previousPythonPath)) {
+    $pythonSource
+}
+else {
+    $pythonSource + [IO.Path]::PathSeparator + $previousPythonPath
+}
+try {
+    & $pythonExecutable (Join-Path $root "packages\cultcache-py\tools\reactive_performance_probe.py") @runtimeArguments
+    if ($LASTEXITCODE -ne 0) {
+        throw "CultMesh Python reactive document performance probe failed."
+    }
+}
+finally {
+    $env:PYTHONPATH = $previousPythonPath
 }