diff --git a/.github/workflows/cultmesh-portability.yml b/.github/workflows/cultmesh-portability.yml index 9c59c2ce..42d3237a 100644 --- a/.github/workflows/cultmesh-portability.yml +++ b/.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: >- diff --git a/packages/cultcache-py/tools/reactive_performance_probe.py b/packages/cultcache-py/tools/reactive_performance_probe.py new file mode 100644 index 00000000..86ab9441 --- /dev/null +++ b/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()) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 7300ad01..76177269 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -1312,8 +1312,11 @@ class CultMeshReactiveDocumentHandle return; } - const timer = setTimeout(flush, 0); - this.#cancelScheduledFlush = () => clearTimeout(timer); + let cancelled = false; + queueMicrotask(() => { + if (!cancelled) flush(); + }); + this.#cancelScheduledFlush = () => { cancelled = true; }; } private flushNow(): void { diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 18e40a32..653a1e95 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/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; diff --git a/scripts/measure-reactive-document-performance.mjs b/scripts/measure-reactive-document-performance.mjs new file mode 100644 index 00000000..786cb3ed --- /dev/null +++ b/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)); +} diff --git a/scripts/measure-reactive-document-performance.ps1 b/scripts/measure-reactive-document-performance.ps1 index 45598ed0..7b410e41 100644 --- a/scripts/measure-reactive-document-performance.ps1 +++ b/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 }