From 979a375d4bd9be41ec85a47b6e6c8e1636cecd7c Mon Sep 17 00:00:00 2001 From: Austin Harris Date: Thu, 24 Sep 2026 20:06:34 -0600 Subject: [PATCH 1/6] ProcessAsync: per-thread scratch cache, scaling gate, async rows, frame fix The 30 M+ synchronous row never regressed; the ~10 M figure was the legacy string API's thread-pool mode, and ProcessAsync itself was capped near 4 M RPC/s on every core count by the AsyncScratch pool lock taken once per document. Library - AsyncScratch keeps one idle scratch per thread in a [ThreadStatic] slot in front of the 64-slot locked pool, which becomes the miss and overflow path. Rent clears the slot before exposing the lease; Return caches only a reusable scratch (reader released, buffers at most 64 KiB) and disposes the rest. Retention is one scratch per thread that has run ProcessAsync plus 64 shared. - Handler.AsyncScope.Dispose no longer writes the captured thread frame back. A Flow scope under a pre/post hook that suspends is disposed on the completing thread; that thread was handed the starting thread's frame, and the two then shared it, so RpcContext, RpcRequestId and RpcSetException on either could read or clear the other's during concurrent dispatch. The ambient restore already brings back the current thread's own frame. Harness (TestServer_Console) - --async N W: ProcessAsync rows at W awaited workers with the same loop as --sync (barrier start, exact allocation accounting per row). - --scale [seconds] [workers] [threshold]: the release gate; inline rows at 1, 2 and N workers, three paired runs, medians, exit 1 below the threshold. - --kestrel [seconds] [async]: the host with EnableAsyncMethods = true, with an inline row and a yielding-methods row. - The legacy string API's thread-pool benchmark moves to the 't' menu entry; Enter runs the synchronous byte API. CI and docs - .github/request-path-sync.allowlist and its checker: every lock, Interlocked, Volatile.Write, [ThreadStatic] field and writable static on the request-path files (core and both companion serializers) is listed with a reason. A non-required scaling job runs --scale 3 4 2.0. - README: async section with the 1- and 16-worker rows, per-shape allocation table, the cost of a real suspension, Kestrel async rows; charts and the explorer gain the async and legacy sets; CHANGELOG, Micro and AspNetCore READMEs updated. Tests - AsyncScratchCacheTests: re-entrant, overlapping, cross-thread (16 workers x 400 documents), cancellation after transfer, throwing reader release and oversized-document trimming. - AsyncInvocationTests.FlowScope_CompletedOnAnotherThread_LeavesThatThreadItsOwnFrame: deterministic reproduction of the frame bug. Measured on a busy machine (single runs, to be re-measured idle before a release): ProcessAsync inline rows 21.2 to 25.7 M RPC/s at 16 workers (was 3.8 to 4.2 M), a real suspension 8.3 M (was 3.9 M); --scale 3 16 4.0 passes with 16/1 ratios of 9.1 to 9.6. --- .github/request-path-sync.allowlist | 35 +++ .github/scripts/check_request_path_sync.py | 78 +++++++ .github/workflows/build_pull_request.yml | 38 +++ AustinHarris.JsonRpc.AspNetCore/README.md | 11 +- .../AsyncInvocationTests.cs | 51 ++++ .../AsyncScratchCacheTests.cs | 185 +++++++++++++++ CHANGELOG.md | 3 + Json-Rpc/Handler.Async.cs | 12 +- Json-Rpc/JsonRpcProcessor.Async.cs | 34 ++- README.md | 74 ++++-- TestServer_Console/AsyncBenchmark.cs | 221 ++++++++++++++---- TestServer_Console/KestrelBenchmark.cs | 44 +++- TestServer_Console/Program.cs | 100 +++++--- benchmarks/Micro/README.md | 8 + benchmarks/charts/benchmarks.json | 76 +++++- .../charts/compare-connections-dark.svg | 4 +- benchmarks/charts/compare-connections.svg | 4 +- .../charts/compare-streamjsonrpc-dark.svg | 4 +- benchmarks/charts/compare-streamjsonrpc.svg | 4 +- benchmarks/charts/explorer.html | 14 +- benchmarks/charts/explorer_template.html | 8 +- benchmarks/charts/inprocess-paths-dark.svg | 4 +- benchmarks/charts/inprocess-paths.svg | 4 +- benchmarks/charts/kestrel-transports-dark.svg | 108 +++++---- benchmarks/charts/kestrel-transports.svg | 108 +++++---- benchmarks/charts/sync-threads-dark.svg | 4 +- benchmarks/charts/sync-threads.svg | 4 +- benchmarks/charts/wasm-interop-dark.svg | 4 +- benchmarks/charts/wasm-interop.svg | 4 +- 29 files changed, 1003 insertions(+), 245 deletions(-) create mode 100644 .github/request-path-sync.allowlist create mode 100644 .github/scripts/check_request_path_sync.py create mode 100644 AustinHarris.JsonRpcTestN/AsyncScratchCacheTests.cs diff --git a/.github/request-path-sync.allowlist b/.github/request-path-sync.allowlist new file mode 100644 index 0000000..86f28ad --- /dev/null +++ b/.github/request-path-sync.allowlist @@ -0,0 +1,35 @@ +# Every `lock (`, `Interlocked.`, `Volatile.Write`, `[ThreadStatic]` field and writable static field in the +# request-path files (core and both companion serializers), with the reason it is allowed there. +# Format: file | the code line, trimmed, without its trailing comment | reason, starting with its tag: +# per-thread the field is thread-static, so no line is shared +# miss-path touched only when a per-thread cache misses or overflows, never by a warm inline document +# registration-only written when methods or sessions are bound, never by a request +# read-only-after-init written once at startup, then only read +# shared-write a request can write it and every core reads it: the reason must say why it is tolerated +# Checked by .github/scripts/check_request_path_sync.py in the pull-request workflow. The measurement that backs this +# list is `TestServer_Console --scale` (README, Benchmarks). A `[ThreadStatic]` attribute on its own line is not +# listed; the field line under it is. +Json-Rpc/JsonRpcProcessor.Async.cs | private static int _count; | miss-path: the shared pool's fill count, read and written only under lock (Pool) +Json-Rpc/JsonRpcProcessor.Async.cs | [ThreadStatic] private static AsyncScratch _slot; | per-thread: the one-slot cache of an idle async scratch +Json-Rpc/JsonRpcProcessor.Async.cs | lock (Pool) | miss-path: AsyncScratch.Rent, taken only when the thread's slot is empty (cold thread, nested or overlapping document) +Json-Rpc/JsonRpcProcessor.Async.cs | lock (Pool) | miss-path: AsyncScratch.Return, taken only when the completing thread's slot is already occupied; the array is the bounded overflow +Json-Rpc/JsonRpcProcessor.cs | [ThreadStatic] private static Scratch _current; | per-thread: the synchronous scratch +Json-Rpc/Handler.Async.cs | [ThreadStatic] private static InvocationState __unflowedState; | per-thread: the invocation frame of a RpcContextFlow.None call +Json-Rpc/Handler.cs | private static int _sessionHandlerMasterVersion = 1; | registration-only: written by Interlocked.Increment when a session is created or destroyed; requests read it to validate their per-thread snapshot +Json-Rpc/Handler.cs | Interlocked.Increment(ref _sessionHandlerMasterVersion); | registration-only: GetSessionHandler creating a session and DestroySession; never on the request path +Json-Rpc/Handler.cs | private static Dictionary _sessionHandlersLocal; | per-thread ([ThreadStatic] on the line above): the thread's snapshot of the session registry +Json-Rpc/Handler.cs | private static int _sessionHandlerLocalVersion = 0; | per-thread ([ThreadStatic] on the line above): the snapshot's version +Json-Rpc/Handler.cs | private static string _lastSessionId; | per-thread ([ThreadStatic] on the line above): the last-session cache +Json-Rpc/Handler.cs | private static Handler _lastSessionHandler; | per-thread ([ThreadStatic] on the line above): the last-session cache +Json-Rpc/Handler.cs | private static InvocationState __state; | per-thread ([ThreadStatic] on the line above): the current invocation frame +Json-Rpc/Jsmn/JsmnSerializer.cs | [ThreadStatic] private static JsmnTokenizer _scratch; | per-thread: the tokenizer scratch +Json-Rpc/Jsmn/JsmnSerializer.cs | [ThreadStatic] private static bool _scratchInUse; | per-thread: re-entrancy flag for the tokenizer scratch +Json-Rpc/Serialization/Utf8KeyTable.cs | lock (_writeLock) | registration-only: Set, Remove, Clear and ReplaceAll rebuild a snapshot under the lock; requests read the published snapshot without it +Json-Rpc/Serialization/Utf8KeyTable.cs | System.Threading.Volatile.Write(ref _snapshot, snapshot); | registration-only: publishes a rebuilt snapshot; read-only afterwards +Json-Rpc/RpcBinding.cs | lock (_sync) | registration-only: RpcBinding.Dispose unbinding an interface tree +AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static Utf8JsonWriter _cachedWriter; | per-thread: the cached writer +AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static JsonSerializerOptions _cachedWriterOptions; | per-thread: the options the cached writer was built with +AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static bool _cachedWriterInUse; | per-thread: re-entrancy flag for the cached writer +AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static byte[] _scratch; | per-thread: transcoding scratch +AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | private static Entry _last; | shared-write: TypeInfo's last-options cache, rewritten whenever the options differ from the previous call, so a per-request write when two serializers with different options serve requests concurrently; tolerated because one options object is the common case; the per-thread copy is the first candidate of the 2026-09-25 P6 resolution, pending its A/B +AustinHarris.JsonRpc.Newtonsoft/NewtonsoftJsonRpcSerializer.cs | [ThreadStatic] private static Scratch _current; | per-thread: the Json.NET scratch diff --git a/.github/scripts/check_request_path_sync.py b/.github/scripts/check_request_path_sync.py new file mode 100644 index 0000000..1755f20 --- /dev/null +++ b/.github/scripts/check_request_path_sync.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""Every synchronization point and every writable static on the request path must be listed, with a reason, in +.github/request-path-sync.allowlist: `lock (`, `Interlocked.`, `Volatile.Write`, `[ThreadStatic]` fields and static +fields that are not readonly, in the request-path files of the core and of both companion serializers. + +A process-wide serialization point on the per-document path caps ProcessAsync at a few million requests per second +on every core count (the AsyncScratch pool lock, 2026-09-25), and the single-threaded micro-benchmarks cannot see +it; a static that one request writes and every core reads costs the same kind of cache-line traffic without a lock. +This check is a review aid: it catches a new lock, atomic or writable static on those files and asks for a written +reason (per-thread, miss-path, registration-only, read-only-after-init). It is not the measurement; +`TestServer_Console --scale` is (a probe moved behind an allowed lock would pass here and fail there). Mutable +objects reached through readonly references are outside its reach and belong to review. + +Usage: python3 .github/scripts/check_request_path_sync.py (from the repository root; exit 1 on any unlisted use) +""" +import glob +import os +import re +import sys + +ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +ALLOWLIST = os.path.join(ROOT, ".github", "request-path-sync.allowlist") +PATTERNS = ["Json-Rpc/JsonRpcProcessor*.cs", "Json-Rpc/Handler*.cs", "Json-Rpc/Invocation/*.cs", "Json-Rpc/Jsmn/*.cs", + "Json-Rpc/Serialization/*.cs", "Json-Rpc/JsonRpcContext.cs", "Json-Rpc/JsonRpcRequestId.cs", "Json-Rpc/RpcBinding.cs", + "AustinHarris.JsonRpc.SystemTextJson/*.cs", "AustinHarris.JsonRpc.Newtonsoft/*.cs"] +USE = re.compile(r"\block\s*\(|\bInterlocked\.|\bVolatile\.Write") +# A static field declaration that is not readonly or const: `[ThreadStatic] private static T name;` or `static T name = ...;`. +# Expression-bodied members (`=>`), methods, classes, events and operators are not fields. +FIELD = re.compile(r"^(?:\[ThreadStatic\]\s*)?(?:(?:public|private|internal|protected|new|volatile)\s+)*static\s+" + r"(?!readonly\b|const\b|class\b|void\b|partial\b|event\b|explicit\b|implicit\b|operator\b)" + r"(?!.*=>)(?!.*\()[\w<>\[\],.?\s]+?\s+\w+\s*(?:=[^;]*)?;$") + + +def load_allowlist(): + """{(file, code line stripped): reason}; lines are `file | code | reason`, `#` comments and blanks ignored.""" + allowed = {} + with open(ALLOWLIST, encoding="utf-8") as f: + for n, line in enumerate(f, 1): + line = line.strip() + if not line or line.startswith("#"): + continue + parts = [p.strip() for p in line.split("|", 2)] + if len(parts) != 3 or not all(parts): + sys.exit(f"{ALLOWLIST}:{n}: expected `file | code | reason`") + allowed[(parts[0].replace("\\", "/"), parts[1])] = parts[2] + return allowed + + +def main(): + allowed = load_allowlist() + seen = set() + problems = [] + for pattern in PATTERNS: + for path in sorted(glob.glob(os.path.join(ROOT, pattern))): + rel = os.path.relpath(path, ROOT).replace("\\", "/") + with open(path, encoding="utf-8-sig") as f: + for n, line in enumerate(f, 1): + code = line.split("//", 1)[0].strip() + if not (USE.search(code) or FIELD.match(code)): + continue + key = (rel, code) + if key in allowed: + seen.add(key) + else: + problems.append(f"{rel}:{n}: `{code}` is not in {os.path.relpath(ALLOWLIST, ROOT)}; add it with a reason or take it off the request path") + for key in allowed: + if key not in seen: + problems.append(f"{os.path.relpath(ALLOWLIST, ROOT)}: `{key[1]}` in {key[0]} no longer exists; remove the entry") + for p in problems: + print(p) + if problems: + return 1 + print(f"request-path synchronization: {len(seen)} listed uses, none unlisted") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/build_pull_request.yml b/.github/workflows/build_pull_request.yml index e22a198..22374e5 100644 --- a/.github/workflows/build_pull_request.yml +++ b/.github/workflows/build_pull_request.yml @@ -42,3 +42,41 @@ jobs: for project in Json-Rpc AustinHarris.JsonRpc.Newtonsoft AustinHarris.JsonRpc.SystemTextJson AustinHarris.JsonRpc.AspNetCore; do dotnet nuget push "$project/bin/Release/"*.nupkg --skip-duplicate --source "https://api.nuget.org/v3/index.json" --api-key ${{ secrets.NugetKey }} # API key for the NuGet feed done + + # Every lock and Interlocked on the request-path files must be listed with a reason (a process-wide lock on the + # per-document path once capped ProcessAsync at 4 M RPC/s on every core count). A review aid, not the measurement. + request-path-sync: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - name: Synchronization on the request path is allowlisted + run: python3 .github/scripts/check_request_path_sync.py + + # The measurement: ProcessAsync must scale from 1 to 4 workers. A process-wide serialization point holds the ratio + # near 1.3 on any core count. Diagnostic on the shared runner (its core count and isolation are not promised, so + # this job is not required and continues on error); the release gate is `--scale 3 16 4.0` on the reference machine. + scaling: + runs-on: ubuntu-latest + continue-on-error: true + steps: + - uses: actions/checkout@v7 + - name: Setup .NET + uses: actions/setup-dotnet@v6 + with: + global-json-file: global.json + dotnet-version: 10.0.x + - name: Build the harness + run: dotnet build TestServer_Console --configuration Release + - name: ProcessAsync scales from 1 to 4 workers (4/1 at least 2.0) + run: | + set +e + dotnet run -c Release --no-build --project TestServer_Console -- --scale 3 4 2.0 | tee scale.txt + status=${PIPESTATUS[0]} + { + echo "## ProcessAsync scaling (diagnostic, not required)" + echo + grep -E '^\|' scale.txt + echo + [ "$status" -eq 0 ] && echo "pass: 4/1 at least 2.0" || echo "**flag: 4/1 below 2.0 on this runner; reproduce with --scale 3 16 4.0 on the reference machine before reading it as a regression**" + } >> "$GITHUB_STEP_SUMMARY" + exit $status diff --git a/AustinHarris.JsonRpc.AspNetCore/README.md b/AustinHarris.JsonRpc.AspNetCore/README.md index aed7e4a..0907d6e 100644 --- a/AustinHarris.JsonRpc.AspNetCore/README.md +++ b/AustinHarris.JsonRpc.AspNetCore/README.md @@ -123,9 +123,14 @@ invoked. - **HTTP:** the call is cancelled when the client disconnects (`HttpContext.RequestAborted`). Notifications are awaited and still answer `204`. The body reader stays leased until the invocation finishes. -- **Raw connections:** documents are processed one at a time, in order. Replies already finished are flushed - before the connection waits on a slow method. When the connection closes, the running method is waited for and - its response discarded. +- **Raw connections:** documents are processed one at a time, in order, so 256 pipelined requests on one + connection are 256 sequential invocations, not 256 concurrent suspensions; concurrency comes from connections. + Replies already finished are flushed before the connection waits on a slow method. When the connection closes, + the running method is waited for and its response discarded. +- **Cost:** every document then goes through `ProcessAsync`. With methods that complete inline the host measures + within a few percent of the synchronous mode; a method that really suspends pays its own async state plus the + library's completion state (about 560 B) and a continuation per request. The main README's Kestrel table has + both rows, measured with `TestServer_Console --kestrel 3 async`. A method receives the token by declaring a `[JsonRpcCancellation] CancellationToken` parameter; see [Asynchronous methods and cancellation](https://github.com/Astn/JSON-RPC.NET#asynchronous-methods-and-cancellation) diff --git a/AustinHarris.JsonRpcTestN/AsyncInvocationTests.cs b/AustinHarris.JsonRpcTestN/AsyncInvocationTests.cs index f21c5e7..eb4fd67 100644 --- a/AustinHarris.JsonRpcTestN/AsyncInvocationTests.cs +++ b/AustinHarris.JsonRpcTestN/AsyncInvocationTests.cs @@ -24,6 +24,12 @@ public sealed class AsyncInvocationTests [TearDown] public void TearDown() => Handler.DestroySession(_session); private void Bind(string name, Delegate method, RpcContextFlow flow = RpcContextFlow.Flow) => ServiceBinder.BindMethod(_session, name, method, contextFlow: flow); private Task Run(string json, JsonRpcSerializer serializer = null, object context = null, CancellationToken token = default) => JsonRpcProcessor.ProcessAsync(_session, json, context, serializer, token); + private string Sync(string json, object context = null) + { + var output = new ArrayBufferWriter(); + JsonRpcProcessor.Process(_session, Encoding.UTF8.GetBytes(json).AsSpan(), output, context); + return Encoding.UTF8.GetString(output.WrittenSpan); + } private static TaskCompletionSource Gate() => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); private static string Request(string method, string parameters = null, string id = "1") => "{\"method\":\"" + method + "\"" + (parameters == null ? "" : ",\"params\":" + parameters) + (id == null ? "" : ",\"id\":" + id) + "}"; private static void Error(string json, int code) => Assert.AreEqual(code, (int)JObject.Parse(json)["error"]["code"], json); @@ -309,6 +315,51 @@ public async Task FlowFrame_IsClearedForEscapedExecutionContext() await escaped; } + [Test] + public async Task FlowScope_CompletedOnAnotherThread_LeavesThatThreadItsOwnFrame() + { + // A hooked Flow invocation suspends here and completes on a pool thread, where its scope is + // disposed. That thread must keep its own frame afterwards: when it was handed this thread's + // frame instead, a synchronous dispatch on each thread saved and restored the same frame, and + // the interleaving below left this thread's method reading no id at all. + var handler = Handler.GetSessionHandler(_session); + handler.SetPreProcessHandler((request, context) => null); + int completedOn = 0; + handler.SetPostProcessHandler((request, response, context) => { if (request.Method == "suspend") completedOn = Environment.CurrentManagedThreadId; return null; }); + var suspended = new TaskCompletionSource(); + Bind("suspend", new Func>(() => suspended.Task)); + var entered = new ManualResetEventSlim(); + var release = new ManualResetEventSlim(); + var left = new ManualResetEventSlim(); + Bind("hold", new Func(() => { entered.Set(); release.Wait(); return 1; })); + var mine = new object(); + Bind("peek", new Func(() => + { + release.Set(); + left.Wait(); + return (Handler.RpcRequestId().IsAbsent ? 0 : 1) + (ReferenceEquals(Handler.RpcContext(), mine) ? 2 : 0); + })); + Bind("warm", new Func(() => 1)); + Sync(Request("warm")); // this thread owns a frame before the suspension, as any thread that has dispatched does + var pending = Run(Request("suspend")); + int completingThread = 0; + var other = Task.Run(() => + { + completingThread = Environment.CurrentManagedThreadId; + suspended.SetResult(7); + Sync(Request("hold", id: "2")); + left.Set(); + }); + // Block, do not await, until the other thread is inside "hold": an awaited continuation could be + // run on that thread, ahead of "hold", and wait for itself. + entered.Wait(); + var response = Sync(Request("peek", id: "\"mine\""), mine); + await other; + Assert.AreEqual("{\"jsonrpc\":\"2.0\",\"result\":7,\"id\":1}", await pending); + Assume.That(completedOn, Is.EqualTo(completingThread), "the completion did not run the scope's cleanup on the completing thread"); + Assert.AreEqual("{\"jsonrpc\":\"2.0\",\"result\":3,\"id\":\"mine\"}", response, "this thread's method sees its own id and context while the other thread dispatches"); + } + [TestCaseSource(nameof(Serializers))] public async Task Batch_IsSequential_AwaitsNotifications_AndIsolatesFaults(string name) { diff --git a/AustinHarris.JsonRpcTestN/AsyncScratchCacheTests.cs b/AustinHarris.JsonRpcTestN/AsyncScratchCacheTests.cs new file mode 100644 index 0000000..239b8c6 --- /dev/null +++ b/AustinHarris.JsonRpcTestN/AsyncScratchCacheTests.cs @@ -0,0 +1,185 @@ +using System; +using System.Buffers; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AustinHarris.JsonRpc; +using AustinHarris.JsonRpc.Serialization; +using Newtonsoft.Json.Linq; +using NUnit.Framework; + +namespace AustinHarris.JsonRpcTestN +{ + /// + /// The async scratch lease behind ProcessAsync: a one-slot per-thread cache in front of the shared pool. + /// The lease is exclusive and transferable, so these cases exercise every way a lease can leave the thread that + /// rented it or overlap another lease on that thread, and check that every document still answers correctly. + /// + [TestFixture] + public sealed class AsyncScratchCacheTests + { + private string _session; + [SetUp] public void SetUp() => _session = "scratch-" + Guid.NewGuid().ToString("N"); + [TearDown] public void TearDown() => Handler.DestroySession(_session); + private void Bind(string name, Delegate method, RpcContextFlow flow = RpcContextFlow.None) => ServiceBinder.BindMethod(_session, name, method, contextFlow: flow); + private Task Run(string json, JsonRpcSerializer serializer = null, CancellationToken token = default) => JsonRpcProcessor.ProcessAsync(_session, json, null, serializer, token); + private static string Request(string method, string parameters = null, string id = "1") => "{\"method\":\"" + method + "\"" + (parameters == null ? "" : ",\"params\":" + parameters) + ",\"id\":" + id + "}"; + private static int Result(string json) => (int)JObject.Parse(json)["result"]; + private static TaskCompletionSource Gate() => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + [Test] + public async Task ReentrantDocument_InsideARunningMethod_OnOneThread() + { + // The outer document holds its lease while the inner one rents: the inner must not receive the + // outer's scratch. Inline and suspended outer methods, many times over, so the slot is reused. + Bind("inner", new Func(n => n * 2)); + Bind("outer", new Func>(async n => + { + var inner = await JsonRpcProcessor.ProcessAsync(_session, Request("inner", "[" + n + "]", "2")).ConfigureAwait(false); + return Result(inner) + 1; + })); + Bind("outerYield", new Func>(async n => + { + await Task.Yield(); + var inner = await JsonRpcProcessor.ProcessAsync(_session, Request("inner", "[" + n + "]", "3")).ConfigureAwait(false); + return Result(inner) + 1; + })); + for (int i = 0; i < 300; i++) + { + Assert.AreEqual(2 * i + 1, Result(await Run(Request("outer", "[" + i + "]")))); + Assert.AreEqual(2 * i + 1, Result(await Run(Request("outerYield", "[" + i + "]")))); + Assert.AreEqual(2 * i, Result(await Run(Request("inner", "[" + i + "]")))); + } + } + + [Test] + public async Task OverlappingDocuments_OnOneThread_SuspendedThenInline() + { + // Document A suspends and is completed elsewhere; document B runs inline on the same thread + // meanwhile and returns its lease into the slot. Both answer, and the slot keeps working after. + var gate = Gate(); + Bind("wait", new Func>(async () => { await gate.Task.ConfigureAwait(false); return 7; })); + Bind("now", new Func(n => n)); + var pendingA = Run(Request("wait")); + Assert.IsFalse(pendingA.IsCompleted); + for (int i = 0; i < 50; i++) Assert.AreEqual(i, Result(await Run(Request("now", "[" + i + "]", "2")))); + gate.SetResult(0); + Assert.AreEqual(7, Result(await pendingA)); + for (int i = 0; i < 50; i++) Assert.AreEqual(i, Result(await Run(Request("now", "[" + i + "]", "2")))); + } + + [Test] + public async Task CrossThreadCompletion_ManyWorkers_EveryDocumentAnswers() + { + Bind("yield", new Func>(async n => { await Task.Yield(); return n + 1; })); + Bind("hop", new Func>(n => Task.Run(() => n + 2))); + Bind("inline", new Func(n => n + 3)); + var counts = await Task.WhenAll(Enumerable.Range(0, 16).Select(w => Task.Run(async () => + { + int ok = 0; + for (int i = 0; i < 400; i++) + { + int n = w * 1000 + i; + var (name, expected) = (i % 3) switch { 0 => ("yield", n + 1), 1 => ("hop", n + 2), _ => ("inline", n + 3) }; + if (Result(await Run(Request(name, "[" + n + "]")).ConfigureAwait(false)) == expected) ok++; + } + return ok; + }))); + Assert.AreEqual(16 * 400, counts.Sum()); + } + + [Test] + public async Task CancellationAfterTransfer_CommitsNothing_AndTheThreadKeepsWorking() + { + var gate = Gate(); + Bind("wait", new Func>(async () => { await gate.Task.ConfigureAwait(false); return 7; })); + Bind("now", new Func(() => 1)); + using var cts = new CancellationTokenSource(); + using var output = new PooledByteBufferWriter(); + var pending = JsonRpcProcessor.ProcessAsync(_session, new ReadOnlyMemory(System.Text.Encoding.UTF8.GetBytes(Request("wait"))), output, cancellationToken: cts.Token); + Assert.IsFalse(pending.IsCompleted); + cts.Cancel(); + gate.SetResult(0); + Assert.CatchAsync(async () => await pending); + Assert.AreEqual(0, output.WrittenCount, "cancellation commits no response bytes"); + for (int i = 0; i < 20; i++) Assert.AreEqual(1, Result(await Run(Request("now")))); + } + + [Test] + public async Task ThrowingReaderRelease_SurfacesOnce_AndTheNextDocumentGetsAFreshLease() + { + Bind("now", new Func(() => 1)); + var serializer = new ReleaseThrowingSerializer(SerializerCatalog.Create("jsmn")) { ThrowOnRelease = true }; + // The damaged scratch is disposed, not cached, so the failure is not repeated by the next document. + Assert.ThrowsAsync(async () => await Run(Request("now"), serializer)); + serializer.ThrowOnRelease = false; + Assert.AreEqual(1, Result(await Run(Request("now"), serializer))); + Assert.AreEqual(1, Result(await Run(Request("now"), serializer))); + Assert.AreEqual(2, serializer.ReadersCreated, "the first reader went with its scratch; the second is cached and reused"); + } + + [Test] + public async Task OversizedDocuments_AreTrimmedOnReturn_AndSmallOnesFollow() + { + Bind("echo", new Func(s => s)); + Bind("now", new Func(() => 1)); + var big = new string('x', 70 * 1024); + for (int round = 0; round < 3; round++) + { + var json = await Run(Request("echo", "[\"" + big + "\"]")); + Assert.AreEqual(big, (string)JObject.Parse(json)["result"]); + Assert.AreEqual(1, Result(await Run(Request("now")))); + } + var sequence = new ReadOnlySequence(System.Text.Encoding.UTF8.GetBytes(Request("echo", "[\"" + big + "\"]"))); + using var output = new PooledByteBufferWriter(); + await JsonRpcProcessor.ProcessAsync(_session, sequence, output); + Assert.AreEqual(big, (string)JObject.Parse(output.ToString())["result"]); + Assert.AreEqual(1, Result(await Run(Request("now")))); + } + + private sealed class ReleaseThrowingSerializer : JsonRpcSerializer + { + private readonly JsonRpcSerializer _inner; + internal bool ThrowOnRelease; + internal int ReadersCreated; + internal ReleaseThrowingSerializer(JsonRpcSerializer inner) { _inner = inner; } + public override string Name => "release-throws"; + public override JsonRpcRequestReader CreateReader() { ReadersCreated++; return new Reader(this, _inner.CreateReader()); } + public override T Read(ReadOnlySpan bytes) => _inner.Read(bytes); + public override object Read(ReadOnlySpan bytes, Type type) => _inner.Read(bytes, type); + public override void Write(IBufferWriter output, T value) => _inner.Write(output, value); + public override void Write(IBufferWriter output, object value, Type type) => _inner.Write(output, value, type); + + private sealed class Reader : JsonRpcRequestReader + { + private readonly ReleaseThrowingSerializer _owner; + private readonly JsonRpcRequestReader _inner; + internal Reader(ReleaseThrowingSerializer owner, JsonRpcRequestReader inner) { _owner = owner; _inner = inner; } + public override bool TryParse(ReadOnlyMemory bytes, out string error) => _inner.TryParse(bytes, out error); + public override ReadOnlyMemory Document => _inner.Document; + public override bool IsBatch => _inner.IsBatch; + public override int Count => _inner.Count; + public override bool Select(int index) => _inner.Select(index); + public override bool HasMethod => _inner.HasMethod; + public override string Method => _inner.Method; + public override ReadOnlySpan MethodUtf8 => _inner.MethodUtf8; + public override JsonRpcIdKind IdKind => _inner.IdKind; + public override ReadOnlySpan IdRaw => _inner.IdRaw; + public override object IdValue => _inner.IdValue; + public override JsonRpcParamsKind ParamsKind => _inner.ParamsKind; + public override int ParamCount => _inner.ParamCount; + public override ReadOnlySpan ParamNameUtf8(int i) => _inner.ParamNameUtf8(i); + public override ReadOnlySpan ParamRaw(int i) => _inner.ParamRaw(i); + public override bool ParamIsNull(int i) => _inner.ParamIsNull(i); + public override T ReadParam(int i) => _inner.ReadParam(i); + public override object ReadParam(int i, Type type) => _inner.ReadParam(i, type); + public override object ParamsValue => _inner.ParamsValue; + public override void Release() + { + _inner.Release(); + if (_owner.ThrowOnRelease) throw new InvalidOperationException("Release failed"); + } + } + } + } +} diff --git a/CHANGELOG.md b/CHANGELOG.md index b1a90cd..243b3cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,10 +23,12 @@ behaviour: a breaking change to either means a new major version. - `Config.SetPreProcessHandler(sessionId, …)` and `Config.SetPostProcessHandler(sessionId, …)`, symmetric with the default-session setters. `Config.SetBeforeProcessHandler(sessionId, …)` remains as an obsolete alias. - `protected JsonRpcService(bool autoBind)`: a subclass constructed with `base(false)` binds itself nowhere, for services that a host or an explicit `BindService` call binds. - `SECURITY.md` (private vulnerability reporting) and this changelog. +- `TestServer_Console --scale`: the release gate for the `ProcessAsync` path (the inline rows at 1, 2 and N workers, three paired runs, medians; fails when N/1 is below the threshold), `--kestrel [seconds] async` for the host with `EnableAsyncMethods = true`, and `--async` rows for `ProcessAsync` at 1 and 16 workers in the README. The 1.x string overloads' thread-pool benchmark is the `t` menu entry, no longer the default. ### Changed - The core no longer depends on Json.NET. +- `ProcessAsync` no longer serializes the process on one lock per document. The async scratch (input copy, reader, staged output) is cached one per thread in front of the shared pool, which is now the miss and overflow path only; on 16 threads the inline rows went from about 4 M to over 20 M RPC/s and a real suspension from 3.9 M to 7 M. Retention is one scratch per thread that has run `ProcessAsync` plus 64 shared, buffers at most 64 KiB each. - The request path looks sessions up without creating them. A request for a session id that was never registered answers `-32601` for every call and leaves the registry untouched; sessions are created by binding and by the per-session `Config` setters. Registration adds the session before publishing the registry version, so a thread that misses its snapshot consults the master registry and cannot answer `-32601` for a session that exists. - The AspNetCore host binds every registered service, `JsonRpcService` subclasses included, to its effective session (the registration's session, then `JsonRpcOptions.SessionId`, then the default). It no longer skips a subclass on the default session. - The core package's description says "no JSON library dependency" instead of "no dependencies". The session registry uses the framework's `ConcurrentDictionary`; the `NonBlocking` package reference is gone, so the core has no dependencies on `net8.0` and `net10.0` (measured with `SessionRegistryBenchmarks`: unknown-id lookups and register/destroy cycles got faster, stable lookups and dispatch are unchanged). @@ -47,6 +49,7 @@ behaviour: a breaking change to either means a new major version. - A trailing notification in a batch no longer leaves a dangling comma. - `async void` methods are rejected at registration. +- A method registered with `RpcContextFlow.Flow` that suspends under a pre- or post-processing hook and completes on another thread no longer hands that thread the frame of the thread that started it. Afterwards the two threads shared one frame, so `RpcContext`, `RpcRequestId` and `RpcSetException` on either could read or clear the other's while both dispatched. Found by the new cross-thread `ProcessAsync` test; the frame is now restored through the ambient value alone. ### Security diff --git a/Json-Rpc/Handler.Async.cs b/Json-Rpc/Handler.Async.cs index 983278a..31340bc 100644 --- a/Json-Rpc/Handler.Async.cs +++ b/Json-Rpc/Handler.Async.cs @@ -26,6 +26,10 @@ private static void Changed(AsyncLocalValueChangedArgs change) // Only the async dispatcher uses this scope. The thread frame is restored before returning // an incomplete operation; a flowing frame is owned by that operation until terminal cleanup. + // A None scope is always disposed on the thread that created it. A Flow scope may be disposed on the + // thread that completed the method (HandleBoxedAsync awaits with a live scope), so Dispose never + // writes the captured thread frame back: restoring the ambient value brings back the frame the + // current thread had before the flowing one arrived, whichever thread that is. private readonly struct AsyncScope : IDisposable { internal readonly InvocationState Frame; @@ -59,13 +63,11 @@ internal AsyncScope(object context, JsonRpcRequestReader reader, bool flow) public void Dispose() { - if (FlowFrame == null) - { - Frame.Context = _context; Frame.Exception = _exception; Frame.Reader = _reader; - } if (AsyncAmbient.Current.Value != _parent) AsyncAmbient.Current.Value = _parent; + if (FlowFrame != null) return; + Frame.Context = _context; Frame.Exception = _exception; Frame.Reader = _reader; // A None scope may have created the reusable thread frame; keep it when there was no parent. - if (_thread != null || FlowFrame != null) __state = _thread; + if (_thread != null) __state = _thread; } } diff --git a/Json-Rpc/JsonRpcProcessor.Async.cs b/Json-Rpc/JsonRpcProcessor.Async.cs index ad4b018..564ed31 100644 --- a/Json-Rpc/JsonRpcProcessor.Async.cs +++ b/Json-Rpc/JsonRpcProcessor.Async.cs @@ -188,13 +188,21 @@ private static void CommitAsyncDocument(AsyncScratch scratch, IBufferWriter 0) @@ -264,12 +280,22 @@ internal void Return() Output = new PooledByteBufferWriter(4096); } else Output.Clear(); + // Only a scratch whose reader released cleanly is cached; a throwing Release leaves reusable + // false and the scratch is disposed below instead of being published in a damaged state. bool retained = false; if (reusable) { - lock (Pool) + if (_slot == null) + { + _slot = this; + retained = true; + } + else { - if (_count < Pool.Length) { Pool[_count++] = this; retained = true; } + lock (Pool) + { + if (_count < Pool.Length) { Pool[_count++] = this; retained = true; } + } } } if (!retained) diff --git a/README.md b/README.md index 1dfda2c..32259df 100644 --- a/README.md +++ b/README.md @@ -284,7 +284,9 @@ In the second `-32602` row, "could not convert" means the serializer refused the ## Asynchronous methods and cancellation -A method may return `Task`, `Task`, `ValueTask` or `ValueTask`. Call it through `JsonRpcProcessor.ProcessAsync`. The processor awaits the method and writes its result; `Task` and `ValueTask` answer `null`. A method that completes synchronously runs inline, allocates nothing on the library's side, and the byte overloads then return `Task.CompletedTask`. +A method may return `Task`, `Task`, `ValueTask` or `ValueTask`. Call it through `JsonRpcProcessor.ProcessAsync`. The processor awaits the method and writes its result; `Task` and `ValueTask` answer `null`. A method that completes synchronously runs inline and the byte overloads then return `Task.CompletedTask`. + +**Cost.** With `RpcContextFlow.None`, the built-in numeric fast path adds no dispatcher allocation when the invocation completes inline; the service's own allocations (a `Task.FromResult`, a result string) are separate and included in the harness figures. Flow allocates an `InvocationState` even when the call completes inline. A method that really suspends may allocate its own async state plus completion state in the result writer, the request handler and the document processor: the yielding benchmark measured about 560 B per request under None, and a continuation per request. The figures are in [Async](#async-processasync-awaited-workers). ```csharp [JsonRpcMethod("lookup")] @@ -474,28 +476,34 @@ The `jsonrpc` member policy (`Lenient` by default) is a compatibility setting, n ## Benchmarks -| What | RPC/s | Details | -| --- | ---: | --- | -| Library alone, 16 threads | 30.6 M to 35.8 M | [Sync](#sync-the-library-alone) | -| Kestrel TCP, 256 pipelined | 15.2 M to 15.5 M | [Kestrel](#kestrel-through-the-aspnetcore-package) | -| Kestrel HTTP, batch of 100 per POST | 12.7 M to 13.7 M | | -| Kestrel HTTP, one request per POST | 128 k to 168 k | HTTP/1.1 round trips dominate | +| What | API and mode | RPC/s | Details | +| --- | --- | ---: | --- | +| Library alone, 16 threads | `Process(bytes)`, dedicated threads | 30.6 M to 35.8 M | [Sync](#sync-the-library-alone) | +| Library alone, 16 workers | `ProcessAsync(bytes)`, awaited workers, methods that complete inline | 21.2 M to 25.7 M | [Async](#async-processasync-awaited-workers); the spread is across registrations, not runs | +| Library alone, 16 workers, one real suspension per request | `ProcessAsync(bytes)`, `yieldsOnce` | 8.32 M | | +| Kestrel TCP, 256 pipelined | `EnableAsyncMethods = false` | 15.2 M to 15.5 M | [Kestrel](#kestrel-through-the-aspnetcore-package) | +| Kestrel TCP, 256 pipelined | `EnableAsyncMethods = true`, methods that complete inline | 13.3 M | | +| Kestrel TCP, 256 pipelined | `EnableAsyncMethods = true`, methods that suspend once | 1.13 M | | +| Kestrel HTTP, batch of 100 per POST | `EnableAsyncMethods = false` | 12.7 M to 13.7 M | | +| Kestrel HTTP, one request per POST | `EnableAsyncMethods = false` | 128 k to 168 k | HTTP/1.1 round trips dominate | +| Legacy string API, thread pool | `Task Process(string)`, batches of 36,000 | 12.0 M | [Legacy](#legacy-string-api-scheduled-synchronous-execution); the 1.x overloads, not the byte path | -All numbers below are from an AMD Ryzen 7 7800X3D (8 cores / 16 threads, 4.2 GHz), 64 GB, Windows 11, .NET 10, Release, Server GC, with the built-in serializer, measured 2026-09-23. Where a row gives two figures they are the spread over that day's runs on an otherwise idle machine; the WSL virtual machine, which takes 15 to 25 % of the box when idle, was shut down for the Kestrel and comparison runs. A single benchmark thread on this machine varies with whatever else lands on its core's SMT sibling, so the 1-thread rows are from runs on an idle core. +All numbers below are from an AMD Ryzen 7 7800X3D (8 cores / 16 threads, 4.2 GHz), 64 GB, Windows 11, .NET 10, Release, Server GC, with the built-in serializer, measured 2026-09-23, except the `ProcessAsync` rows and the `EnableAsyncMethods = true` rows, measured 2026-09-25 on the same machine with other sessions running (single runs, so lower bounds; they are re-measured on an idle box before a release). Where a row gives two figures they are the spread over that day's runs on an otherwise idle machine; the WSL virtual machine, which takes 15 to 25 % of the box when idle, was shut down for the Kestrel and comparison runs. A single benchmark thread on this machine varies with whatever else lands on its core's SMT sibling, so the 1-thread rows are from runs on an idle core. `TestServer_Console` is the benchmark harness. It binds one service with five small methods (`add`, `addInt`, `NullableFloatToNullableFloat`, `Test2`, `StringMe`), drives the same five requests through the server, checks every response is a `result` rather than an error, and ends each mode with a bar chart of RPC/s. For one-request timings with an allocation column, the numbers to check before merging a change to the dispatch path, see [benchmarks/Micro](benchmarks/Micro/README.md). ``` dotnet run -c Release --project TestServer_Console -- --sync 3 # library only, 1..N threads (add a thread count, e.g. --sync 3 1, for one row) -dotnet run -c Release --project TestServer_Console -- --async 3 1 # ProcessAsync, Flow and None separately; no published figures yet -dotnet run -c Release --project TestServer_Console -- --kestrel 3 # through the AspNetCore package, HTTP and TCP +dotnet run -c Release --project TestServer_Console -- --async 3 16 # ProcessAsync from 16 awaited workers, one row per registration (--async 3 1 for the 1-worker column) +dotnet run -c Release --project TestServer_Console -- --scale 3 16 4.0 # release gate: ProcessAsync must scale at least 4x from 1 to 16 workers +dotnet run -c Release --project TestServer_Console -- --kestrel 3 # through the AspNetCore package, HTTP and TCP (add `async` for EnableAsyncMethods = true) dotnet run -c Release --project TestServer_Console -- --compare 3 # the same calls through StreamJsonRpc and gRPC for .NET, side by side dotnet run -c Release --project TestServer_Console -- --sweep 2 benchmarks/charts/sweep.json # every library and transport at 1, 2, 4, 8, 16 connections; one file per run -dotnet run -c Release --project TestServer_Console # menu: Enter = Task mode, s = sync, k = Kestrel, x = compare, q = quit +dotnet run -c Release --project TestServer_Console # menu: Enter = Process(bytes), a = ProcessAsync(bytes), t = legacy string API, k = Kestrel, x = compare, q = quit dotnet run --project samples/WasmHost # browser: "Run benchmark" on the page ``` -`--async [seconds] [workers]` uses `ProcessAsync` with separate sessions and the same five wire names as `--sync`. It reports synchronous returns through the new API, completed `Task` and inline `ValueTask`, with Flow and None registrations reported separately, plus a `yieldsOnce` shape that awaits `Task.Yield()`. Responses and ids are validated before timing, and allocation totals include the service methods' own allocations. +The three modes measure three different things and are named for the entry point they call: `Process(bytes), dedicated threads` is the library alone; `ProcessAsync(bytes), awaited workers` is the entry point an asynchronous host calls; `Legacy Process(string), scheduled synchronous work` is the 1.x string overloads through the thread pool. The Kestrel rows say whether `EnableAsyncMethods` was on. ### Sync: the library alone @@ -516,9 +524,37 @@ dotnet run --project samples/WasmHost # browser: " Per-thread cost rises with thread count because the 16 threads share 8 physical cores. -### Task: scheduled synchronous execution +### Async: ProcessAsync, awaited workers + +`--async [seconds] [workers]` calls the byte-level `JsonRpcProcessor.ProcessAsync` from `Task.Run` workers that await each call, with the same five requests as `--sync` and the same loop shape (workers start behind a barrier, stop on a shared flag, walk the inputs with an index; worker startup is reported separately). Each row registers the five methods with one return shape and one `RpcContextFlow`; the `yieldsOnce` rows are one method that awaits `Task.Yield()`, a real suspension per request. Responses are checked before timing. The bytes column is per request including the service method's own allocations, measured at one worker; the inline rows are exact per-thread counts, the `yieldsOnce` rows the process-wide counter. + +| Registration | 1 worker | 16 workers | B per request | +| --- | ---: | ---: | ---: | +| synchronous methods, None | 3.16 M | 21.2 M | 6 | +| `Task`, Flow | 1.84 M | 19.8 M | 251 | +| `Task`, None | 2.42 M | 25.4 M | 67 | +| `ValueTask`, Flow | 1.95 M | 21.9 M | 190 | +| `ValueTask`, None | 2.30 M | 25.7 M | 6 | +| `yieldsOnce`, Flow | 671 k | 6.14 M | 737 | +| `yieldsOnce`, None | 887 k | 8.32 M | 556 | + +The 16-worker rows are an equal-weight mix of the five requests, except `yieldsOnce`, which is one request. Per request shape, bytes per request at one worker, including the method's own allocations (the harness prints these lines before each row): -The default mode submits batches through the `Task`-returning `Process` overload from every core at once, the way an async host would, so it pays for the thread-pool hop, a `Task`, a result string and a continuation per request. Each batch size is repeated for at least half a second after a one-second warm-up. Throughput peaks once a batch is large enough to keep every core busy and falls off again when hundreds of thousands of requests are queued at once: +| Registration | `add` | `addInt` | nullable float | decimal | `StringMe` | +| --- | ---: | ---: | ---: | ---: | ---: | +| synchronous methods, None | 0 | 0 | 0 | 0 | 32 | +| `Task`, Flow | 256 | 184 | 256 | 272 | 288 | +| `Task`, None | 72 | 0 | 72 | 88 | 104 | +| `ValueTask`, Flow | 184 | 184 | 184 | 184 | 216 | +| `ValueTask`, None | 0 | 0 | 0 | 0 | 32 | + +With `RpcContextFlow.None` the dispatcher adds no allocation to a method that completes inline: the `Task` None row is the service's own `Task.FromResult` (`Task` for 8 comes from the runtime's cache), and the 32 bytes of `StringMe` are its result string. Flow allocates the `InvocationState` and the execution-context bridge on every call, inline or not. A real suspension allocates the method's own async state plus completion state in the result writer, the request handler and the document processor: about 560 B per request in the `yieldsOnce` None row. The 7 to 10 M target for a hosted server applies to methods that complete inline; a method that suspends costs a continuation per request as well as those bytes. + +Before 2.0.0, the `ProcessAsync` path was capped near 4 M RPC/s at every worker count by one lock taken per document on the shared scratch pool, which no single-threaded benchmark could see. The scratch is now cached one per thread in front of that pool. `--scale [seconds] [workers] [threshold]` is the gate that catches the next such point: the inline None rows at 1, 2 and 16 workers, three paired runs, medians, and it exits non-zero when any 16/1 ratio is below 4.0 (the lock gave 1.3; the cache gives about 9). Run it on the reference machine before a release and paste its table into the release notes; the pull-request build runs a diagnostic `--scale 3 4 2.0` on the shared runner, and a check that every `lock`, `Interlocked`, `Volatile.Write`, thread-static and writable static field on the request-path files of the core and both companion serializers is listed with a reason (per-thread, miss-path, registration-only, read-only-after-init) in `.github/request-path-sync.allowlist`. + +### Legacy string API: scheduled synchronous execution + +The `t` menu entry submits batches through the 1.x `Task Process(string)` overload from every core at once: the compatibility string API, including scheduling, transcoding, result strings and continuations, so it pays for the thread-pool hop, a `Task`, a result string and a continuation per request. It is not the byte path an asynchronous host uses; the byte entry points are the Sync and Async tables above. Each batch size is repeated for at least half a second after a one-second warm-up. Throughput peaks once a batch is large enough to keep every core busy and falls off again when hundreds of thousands of requests are queued at once: | Batch size | RPC/s | | ---: | ---: | @@ -529,11 +565,11 @@ The default mode submits batches through the `Task`-returning `Process` overload | 252,000 | 7.6 M | | 2,016,000 | 7.2 M | -Task mode is slower than sync mode because it measures the .NET thread pool and a `Task`, a string and a continuation per request as much as the library. The AspNetCore host does not use that string path or allocate a result string: it awaits transport reads and flushes without a thread per connection, as the next table shows. +This mode is slower than the byte modes because it measures the .NET thread pool and a `Task`, a string and a continuation per request as much as the library. The AspNetCore host does not use that string path or allocate a result string: it awaits transport reads and flushes without a thread per connection, as the next table shows. ### Kestrel: through the AspNetCore package -`--kestrel` starts a real Kestrel on loopback with `MapJsonRpc` and `JsonRpcConnectionHandler`, then drives it with 16 clients on the same machine, so every figure is an upper bound for one box talking to itself. The in-process rows are the same requests through the byte entry point with no transport at all, for scale. +`--kestrel [seconds]` starts a real Kestrel on loopback with `MapJsonRpc` and `JsonRpcConnectionHandler`, then drives it with 16 clients on the same machine, so every figure is an upper bound for one box talking to itself. The in-process rows are the same requests through the byte entry point with no transport at all, for scale. The rows below ran with `EnableAsyncMethods = false`, the default, so the host called `Process`; `--kestrel 3 async` runs the same host with `EnableAsyncMethods = true`, every document through `ProcessAsync`, and adds a TCP row whose five methods await `Task.Yield()` before answering. @@ -546,8 +582,10 @@ Task mode is slower than sync mode because it measures the .NET thread pool and | HTTP, 1 request per POST | 128 k to 168 k | 95 to 125 µs per round trip per client depending on the run; HTTP/1.1 request-response is the cost, not the server | | HTTP, batch of 100 per POST | 12.7 M to 13.7 M | | | TCP, 256 pipelined | 15.2 M to 15.5 M | ring-buffer clients, one thread each, streaming framer | +| TCP, 256 pipelined, `EnableAsyncMethods = true`, methods that complete inline | 13.3 M | 2026-09-25, one run on a busy machine; the same run's `false` row was 14.0 M | +| TCP, 256 pipelined, `EnableAsyncMethods = true`, methods that suspend once | 1.13 M | five `async Task` methods awaiting `Task.Yield()` | -The TCP client keeps 256 requests in flight per connection and refills from a precomputed ring of request bytes with one `Send` per refill; the server side is the same `Process` call the HTTP endpoint makes, fed by `JsonFramer`. +The TCP client keeps 256 requests in flight per connection and refills from a precomputed ring of request bytes with one `Send` per refill; the server side is the same `Process` call the HTTP endpoint makes, fed by `JsonFramer`. With `EnableAsyncMethods = true` the connection handler processes the documents of one connection one at a time, in order, so 256 pipelined requests are 256 sequential invocations and a method that suspends is paid for per request; concurrency comes from the 16 connections. ### Versus StreamJsonRpc and gRPC @@ -610,6 +648,8 @@ simdjson was evaluated as a fourth parser and not adopted: through the only main The charts, the explorer page and the figures in this file come from one data file, [benchmarks/charts/benchmarks.json](benchmarks/charts/benchmarks.json); how they are rendered and checked is under [Building](#charts). +On 2026-09-25 the `ProcessAsync` path was found capped near 4 M RPC/s at every worker count: every document took one lock on the shared scratch pool, invisible to the single-threaded micro-benchmarks. A one-slot per-thread cache in front of the pool took the inline rows to over 20 M at 16 workers; the `--scale` gate and the request-path allowlist exist so the next such point is caught before a release. + The 2026-09-23 performance pass (compiled invokers that read the tokens and write the pooled buffer through direct calls instead of virtual, delegate and interface calls; a tokenizer that keeps its scanner state in locals; a last-session cache; envelope keys matched by length; a flat method table) was measured A/B in one session: the same seven runs of `--sync 2 1` went from 3.2 M to 4.1 M (median 3.6 M) before to 4.0 M to 4.8 M (median 4.4 M) after, about 20 to 25 % more on one thread. The transport rows are bound by the loopback round trips rather than by the library and moved less. Under the previous harness (one pass per batch, workstation GC) the two-million batch ran at about 525,000 RPC/s on 1.3 and 1,584,906 RPC/s on 2.0 on the same machine. The 1.x figure published earlier in this README was measured while the benchmark service was not bound, so every request took the "method not found" path; the benchmark now prints the responses so that cannot go unnoticed. diff --git a/TestServer_Console/AsyncBenchmark.cs b/TestServer_Console/AsyncBenchmark.cs index 7046e73..2e7262c 100644 --- a/TestServer_Console/AsyncBenchmark.cs +++ b/TestServer_Console/AsyncBenchmark.cs @@ -1,72 +1,205 @@ using System; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Text; +using System.Threading; using System.Threading.Tasks; using AustinHarris.JsonRpc; using AustinHarris.JsonRpc.Serialization; namespace TestServer_Console; +/// +/// ProcessAsync under load: the same five requests as --sync through the byte entry point, driven by +/// awaited workers on the thread pool. Each row registers the five methods with one return shape (synchronous, +/// completed Task<T>, inline ValueTask<T>) and one ; the +/// yield row is one method that awaits Task.Yield(), a real suspension per request. The timing loop +/// has the shape of the synchronous harness (workers start behind a barrier, stop on a shared flag, walk the inputs +/// with an index) so that a difference between the two modes is a difference between the two entry points. +/// internal static class AsyncBenchmark { + internal static readonly (string shape, RpcContextFlow flow)[] Rows = + { + ("sync", RpcContextFlow.None), + ("Task", RpcContextFlow.Flow), ("Task", RpcContextFlow.None), + ("ValueTask", RpcContextFlow.Flow), ("ValueTask", RpcContextFlow.None), + ("yield", RpcContextFlow.Flow), ("yield", RpcContextFlow.None), + }; + + /// The inline rows a process-wide serialization point shows up in first; the scaling gate runs these. + internal static readonly (string shape, RpcContextFlow flow)[] InlineRows = { ("sync", RpcContextFlow.None), ("Task", RpcContextFlow.None), ("ValueTask", RpcContextFlow.None) }; + + internal readonly record struct Measurement(long Count, double Seconds, double StartupSeconds, long AllocatedBytes) + { + public double RpcPerSec => Count / Seconds; + public double BytesPerRpc => Count == 0 ? 0 : AllocatedBytes / (double)Count; + } + internal static async Task RunAsync(Action print, double seconds, int workers) { if (seconds <= 0 || workers <= 0) throw new ArgumentOutOfRangeException(nameof(seconds)); - var inputs = BenchmarkRunner.taskInputs.Select(t => (ReadOnlyMemory)Encoding.UTF8.GetBytes(t)).ToArray(); - var expected = BenchmarkRunner.taskInputs.Select(t => JsonRpcProcessor.ProcessSync(t)).ToArray(); - foreach (var shape in new[] { "sync", "Task", "ValueTask", "yield" }) - foreach (var flow in shape == "sync" ? new[] { RpcContextFlow.None } : new[] { RpcContextFlow.Flow, RpcContextFlow.None }) + var summary = new List(); + foreach (var (shape, flow) in Rows) { string session = "async-benchmark-" + shape + "-" + flow; try { Register(session, shape, flow); - var rowInputs = shape == "yield" ? new[] { (ReadOnlyMemory)Encoding.UTF8.GetBytes("{\"method\":\"yieldsOnce\",\"id\":6}") } : inputs; - var rowExpected = shape == "yield" ? new[] { "{\"jsonrpc\":\"2.0\",\"result\":7,\"id\":6}" } : expected; - using (var output = new PooledByteBufferWriter()) + var rowInputs = Inputs(shape); + await ValidateAndReportAllocations(print, session, shape, flow, rowInputs); + bool yield = shape == "yield"; + var m = await Measure(session, rowInputs, workers, seconds, processWide: yield); + string label = yield ? "one request (yieldsOnce)" : "equal-weight mix of five requests"; + print($"{shape} / {flow}: {m.Count:N0} RPCs, {workers} workers, {m.Seconds:F3} s, {m.RpcPerSec:N0} RPC/s; worker startup {m.StartupSeconds * 1000:F1} ms; " + + $"{(yield ? "process-wide" : "worker-thread")} allocation total {m.AllocatedBytes:N0} B ({m.BytesPerRpc:F0} B/RPC, including method and harness allocations); {label}."); + summary.Add(new BenchmarkRunner.ChartRow($"{shape} / {flow}", m.RpcPerSec, $"{m.Count,12:N0} RPCs {m.BytesPerRpc,6:F0} B/RPC")); + } + finally { Handler.DestroySession(session); } + } + BenchmarkRunner.PrintBarChart($"ProcessAsync(bytes), awaited workers - {Config.Serializer.Name} - {workers} worker{(workers == 1 ? "" : "s")} - RPC/s by registration", "Registration", summary); + } + + /// + /// The scaling gate: the inline None rows at 1, 2 and workers, + /// paired runs, medians, and the 2/1 and N/1 ratios. Returns false when any N/1 ratio is below + /// . A process-wide serialization point on the async path holds the ratio near 1.3 + /// on any core count; the per-thread cache measured about 6.8 on 16 threads. + /// + internal static async Task ScaleAsync(Action print, double seconds, int workers, double threshold, int runs = 3) + { + if (workers < 2) throw new ArgumentOutOfRangeException(nameof(workers)); + var counts = new[] { 1, 2, workers }.Distinct().ToArray(); + print($"Scaling check: ProcessAsync(bytes), {string.Join(", ", counts)} workers, {runs} paired runs of {seconds:0.#} s per cell, medians; {Environment.ProcessorCount} logical processors; gate {workers}/1 >= {threshold:0.0#}.\n"); + var results = new Dictionary<(string, RpcContextFlow, int), List>(); + foreach (var (shape, flow) in InlineRows) + { + string session = "async-scale-" + shape + "-" + flow; + try + { + Register(session, shape, flow); + var rowInputs = Inputs(shape); + await ValidateAndReportAllocations(print, session, shape, flow, rowInputs); + await Measure(session, rowInputs, workers, Math.Min(seconds, 1)); // warm the pool threads and the JIT + for (int run = 1; run <= runs; run++) + foreach (int w in counts) { - for (int i = 0; i < rowInputs.Length; i++) - { - output.Clear(); - await JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); - if (output.ToString() != rowExpected[i]) throw new InvalidOperationException("Unexpected response: " + output); - for (int warm = 0; warm < 500; warm++) { output.Clear(); await JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); } - if (shape != "yield") - { - long before = GC.GetAllocatedBytesForCurrentThread(); - for (int n = 0; n < 2000; n++) - { - output.Clear(); - var task = JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); - if (!task.IsCompleted) throw new InvalidOperationException("Expected inline completion."); - task.GetAwaiter().GetResult(); - } - long totalBytes = GC.GetAllocatedBytesForCurrentThread() - before; - print($"{shape} / {flow} / shape {i + 1}: {totalBytes} B total over 2000 requests ({totalBytes / 2000.0:F1} B/RPC), including method allocations."); - } - } + var m = await Measure(session, rowInputs, w, seconds); + print($" run {run}: {shape} / {flow}, {w,2} workers: {m.RpcPerSec,14:N0} RPC/s ({m.Count:N0} RPCs in {m.Seconds:F3} s, startup {m.StartupSeconds * 1000:F1} ms)"); + if (!results.TryGetValue((shape, flow, w), out var list)) results[(shape, flow, w)] = list = new List(); + list.Add(m.RpcPerSec); } - var start = Stopwatch.GetTimestamp(); - long beforeProcess = GC.GetTotalAllocatedBytes(true); - var counts = await Task.WhenAll(Enumerable.Range(0, workers).Select(_ => Task.Run(async () => - { - long count = 0; - using var output = new PooledByteBufferWriter(); - while ((Stopwatch.GetTimestamp() - start) / (double)Stopwatch.Frequency < seconds) - { - output.Clear(); - await JsonRpcProcessor.ProcessAsync(session, rowInputs[count % rowInputs.Length], output).ConfigureAwait(false); - count++; - } - return count; - }))); - double elapsed = (Stopwatch.GetTimestamp() - start) / (double)Stopwatch.Frequency; - long allocations = GC.GetTotalAllocatedBytes(true) - beforeProcess; - print($"{shape} / {flow}: {counts.Sum():N0} RPCs, {workers} workers, {elapsed:F3} s, {counts.Sum() / elapsed:N0} RPC/s; process allocation total {allocations:N0} B."); } finally { Handler.DestroySession(session); } } + + bool pass = true; + print(""); + print($"| Registration | 1 worker | 2 workers | {workers} workers | 2/1 | {workers}/1 | gate |"); + print("| --- | ---: | ---: | ---: | ---: | ---: | --- |"); + foreach (var (shape, flow) in InlineRows) + { + double one = Median(results[(shape, flow, 1)]); + double two = Median(results[(shape, flow, 2)]); + double many = Median(results[(shape, flow, workers)]); + bool ok = many / one >= threshold; + pass &= ok; + print($"| {shape} / {flow} | {one:N0} | {two:N0} | {many:N0} | {two / one:F2} | {many / one:F2} | {(ok ? "pass" : "FAIL")} |"); + } + print(""); + print(pass ? $"Scaling check passed: every inline row scales at least {threshold:0.0#}x from 1 to {workers} workers." + : $"Scaling check FAILED: an inline row scales less than {threshold:0.0#}x from 1 to {workers} workers; look for a process-wide serialization point on the ProcessAsync path."); + print("The 2/1 ratio is diagnostic (the lock was already visible at two workers, about 1.3); it is not gated."); + return pass; + } + + private static double Median(List values) + { + var sorted = values.OrderBy(v => v).ToArray(); + int n = sorted.Length; + return n % 2 == 1 ? sorted[n / 2] : (sorted[n / 2 - 1] + sorted[n / 2]) / 2; + } + + private static ReadOnlyMemory[] Inputs(string shape) => shape == "yield" + ? new[] { (ReadOnlyMemory)Encoding.UTF8.GetBytes("{\"method\":\"yieldsOnce\",\"id\":6}") } + : BenchmarkRunner.taskInputs.Select(t => (ReadOnlyMemory)Encoding.UTF8.GetBytes(t)).ToArray(); + + /// Checks every response, warms the row, and prints the per-shape allocation of an inline document. + private static async Task ValidateAndReportAllocations(Action print, string session, string shape, RpcContextFlow flow, ReadOnlyMemory[] rowInputs) + { + var expected = shape == "yield" + ? new[] { "{\"jsonrpc\":\"2.0\",\"result\":7,\"id\":6}" } + : BenchmarkRunner.taskInputs.Select(t => JsonRpcProcessor.ProcessSync(t)).ToArray(); + using var output = new PooledByteBufferWriter(); + for (int i = 0; i < rowInputs.Length; i++) + { + output.Clear(); + await JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); + if (output.ToString() != expected[i]) throw new InvalidOperationException("Unexpected response: " + output); + for (int warm = 0; warm < 500; warm++) { output.Clear(); await JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); } + if (shape == "yield") continue; + long before = GC.GetAllocatedBytesForCurrentThread(); + for (int n = 0; n < 2000; n++) + { + output.Clear(); + var task = JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output); + if (!task.IsCompleted) throw new InvalidOperationException("Expected inline completion."); + task.GetAwaiter().GetResult(); + } + long totalBytes = GC.GetAllocatedBytesForCurrentThread() - before; + print($"{shape} / {flow} / shape {i + 1}: {totalBytes} B total over 2000 requests ({totalBytes / 2000.0:F1} B/RPC), including method allocations."); + } + } + + /// + /// One timed cell. Workers are Task.Run tasks that await each call; they start together behind a barrier, + /// the clock starts once every worker is ready, and they stop on a shared flag. Startup is reported separately. + /// + /// Count allocations process-wide (the yield row: its continuations run on other pool threads, so + /// a worker's thread-local counter is meaningless) instead of summing the worker threads' own counters (exact for inline rows). + internal static async Task Measure(string session, ReadOnlyMemory[] rowInputs, int workers, double seconds, bool processWide = false) + { + int stop = 0; + var counts = new long[workers * 16]; // padded to avoid false sharing + var allocs = new long[workers * 16]; + using var ready = new Barrier(workers + 1); + var startup = Stopwatch.StartNew(); + var tasks = new Task[workers]; + for (int w = 0; w < workers; w++) + { + int slot = w * 16; + tasks[w] = Task.Run(async () => + { + using var output = new PooledByteBufferWriter(); + ready.SignalAndWait(); + // Per worker thread, exact for inline rows (the process-wide counter under-reports at this rate). + long allocated = GC.GetAllocatedBytesForCurrentThread(); + long n = 0; + int i = 0; + while (Volatile.Read(ref stop) == 0) + { + output.Clear(); + await JsonRpcProcessor.ProcessAsync(session, rowInputs[i], output).ConfigureAwait(false); + if (++i == rowInputs.Length) i = 0; + n++; + } + counts[slot] = n; + allocs[slot] = GC.GetAllocatedBytesForCurrentThread() - allocated; + }); + } + ready.SignalAndWait(); + double startupSeconds = startup.Elapsed.TotalSeconds; + long before = GC.GetTotalAllocatedBytes(false); + var sw = Stopwatch.StartNew(); + await Task.Delay(TimeSpan.FromSeconds(seconds)); + Volatile.Write(ref stop, 1); + await Task.WhenAll(tasks); + sw.Stop(); + long processAllocated = GC.GetTotalAllocatedBytes(false) - before; + long total = 0, allocatedTotal = 0; + for (int w = 0; w < workers; w++) { total += counts[w * 16]; allocatedTotal += allocs[w * 16]; } + return new Measurement(total, sw.Elapsed.TotalSeconds, startupSeconds, processWide ? processAllocated : allocatedTotal); } private static void Register(string session, string shape, RpcContextFlow flow) diff --git a/TestServer_Console/KestrelBenchmark.cs b/TestServer_Console/KestrelBenchmark.cs index b833417..888e475 100644 --- a/TestServer_Console/KestrelBenchmark.cs +++ b/TestServer_Console/KestrelBenchmark.cs @@ -27,14 +27,17 @@ namespace TestServer_Console; /// answering the same five requests over HTTP (one request per POST, and a batch per POST) and over a raw /// TCP connection with pipelined requests. Clients and server share the machine, so each figure is an /// upper bound on what one box can do talking to itself; the in-process rows are the same requests through -/// the byte entry point with no transport at all, for scale. +/// the byte entry point with no transport at all, for scale. With asyncMethods the host runs with +/// EnableAsyncMethods = true, so every document goes through ProcessAsync, and a second TCP row +/// answers the same five calls from methods that await Task.Yield() once: a real suspension per request. /// internal static class KestrelBenchmark { private static readonly byte[] ResultPrefix = Encoding.UTF8.GetBytes("{\"jsonrpc\":\"2.0\",\"result\":"); private static readonly byte[] BatchResultPrefix = Encoding.UTF8.GetBytes("[{\"jsonrpc\":\"2.0\",\"result\":"); + private const string YieldPrefix = "yield."; - internal static async Task RunAsync(Action print, double seconds = 3, int clients = 0, int pipeline = 256) + internal static async Task RunAsync(Action print, double seconds = 3, int clients = 0, int pipeline = 256, bool asyncMethods = false) { print ??= Console.WriteLine; if (clients <= 0) clients = Environment.ProcessorCount; @@ -51,17 +54,18 @@ internal static async Task RunAsync(Action print, double seconds = 3, in k.Listen(IPAddress.Loopback, 0); k.Listen(IPAddress.Loopback, tcpPort, l => l.UseConnectionHandler()); }); - builder.Services.AddJsonRpc(); + builder.Services.AddJsonRpc(o => o.EnableAsyncMethods = asyncMethods); var app = builder.Build(); app.MapJsonRpc("/rpc"); await app.StartAsync(); + if (asyncMethods) BindYieldingMethods(); try { var addresses = app.Services.GetRequiredService().Features.Get().Addresses; var httpUrl = addresses.First(a => !a.EndsWith(":" + tcpPort)) + "/rpc"; - print($"Kestrel on {httpUrl} (HTTP) and 127.0.0.1:{tcpPort} (TCP), {clients} clients, pipeline {pipeline}, {seconds:0.#} s per row\n"); + print($"Kestrel on {httpUrl} (HTTP) and 127.0.0.1:{tcpPort} (TCP), {clients} clients, pipeline {pipeline}, {seconds:0.#} s per row, EnableAsyncMethods = {(asyncMethods ? "true" : "false")}\n"); var rows = new List(); var session = Handler.DefaultSessionId(); @@ -90,18 +94,46 @@ internal static async Task RunAsync(Action print, double seconds = 3, in // TCP, pipelined. TcpRun(tcpPort, inputs, clients, 0.5, pipeline, ResultPrefix); (count, secs) = TcpRun(tcpPort, inputs, clients, seconds, pipeline, ResultPrefix); - rows.Add(new BenchmarkRunner.ChartRow($"TCP, {pipeline} pipelined", count / secs, $"{count,12:N0} RPCs")); + rows.Add(new BenchmarkRunner.ChartRow($"TCP, {pipeline} pipelined{(asyncMethods ? ", inline methods" : "")}", count / secs, $"{count,12:N0} RPCs")); print($" TCP done ({count / secs:N0} RPC/s)"); - BenchmarkRunner.PrintBarChart($"Kestrel benchmark - {Config.Serializer.Name} - RPC/s by transport", "Transport", rows); + if (asyncMethods) + { + // The same five calls, each answered by a method that awaits Task.Yield() before returning. + var yieldInputs = inputs.Select(i => Encoding.UTF8.GetBytes(Encoding.UTF8.GetString(i).Replace("\"method\":\"", "\"method\":\"" + YieldPrefix))).ToArray(); + TcpRun(tcpPort, yieldInputs, clients, 0.5, pipeline, ResultPrefix); + (count, secs) = TcpRun(tcpPort, yieldInputs, clients, seconds, pipeline, ResultPrefix); + rows.Add(new BenchmarkRunner.ChartRow($"TCP, {pipeline} pipelined, yielding methods", count / secs, $"{count,12:N0} RPCs")); + print($" TCP yielding done ({count / secs:N0} RPC/s)"); + } + + BenchmarkRunner.PrintBarChart($"Kestrel benchmark - {Config.Serializer.Name} - EnableAsyncMethods = {(asyncMethods ? "true" : "false")} - RPC/s by transport", "Transport", rows); } finally { await app.StopAsync(); await app.DisposeAsync(); + if (asyncMethods) UnbindYieldingMethods(); } } + private static readonly string[] YieldNames = { "add", "addInt", "NullableFloatToNullableFloat", "Test2", "StringMe" }; + + /// The five benchmark methods as async Task<T> methods that yield once, on the default session under a prefix. + private static void BindYieldingMethods() + { + ServiceBinder.BindMethod(YieldPrefix + "add", new Func>(async (l, r) => { await Task.Yield(); return l + r; })); + ServiceBinder.BindMethod(YieldPrefix + "addInt", new Func>(async (l, r) => { await Task.Yield(); return l + r; })); + ServiceBinder.BindMethod(YieldPrefix + "NullableFloatToNullableFloat", new Func>(async a => { await Task.Yield(); return a; })); + ServiceBinder.BindMethod(YieldPrefix + "Test2", new Func>(async x => { await Task.Yield(); return x; })); + ServiceBinder.BindMethod(YieldPrefix + "StringMe", new Func>(async x => { await Task.Yield(); return x; })); + } + + private static void UnbindYieldingMethods() + { + foreach (var name in YieldNames) ServiceBinder.UnbindMethod(YieldPrefix + name); + } + internal static byte[] BuildBatch(byte[][] inputs, int size) { var sb = new StringBuilder("["); diff --git a/TestServer_Console/Program.cs b/TestServer_Console/Program.cs index 1a3adb6..30b653f 100644 --- a/TestServer_Console/Program.cs +++ b/TestServer_Console/Program.cs @@ -1,4 +1,4 @@ -using System; +using System; using AustinHarris.JsonRpc; using Hardware.Info; @@ -12,9 +12,11 @@ class Program static void Main(string[] args) { - // When stdout is a pipe (CI, `dotnet run | tee`) there is no console buffer to clear or reposition. services = new object[] { new CalculatorService() }; - // `dotnet run -- --async [seconds] [workers]` exercises real asynchronous invocation. + // Before any mode runs: the awaited-worker modes (--async, --scale) start their workers on the pool. + System.Threading.ThreadPool.SetMinThreads(Environment.ProcessorCount * 3, Environment.ProcessorCount * 3); + + // `dotnet run -- --async [seconds] [workers]` drives ProcessAsync from awaited workers (the Async table). if (args.Length > 0 && args[0] == "--async") { double seconds = args.Length > 1 && double.TryParse(args[1], out var s) ? s : 3; @@ -23,13 +25,24 @@ static void Main(string[] args) return; } + // `dotnet run -- --scale [seconds] [workers] [threshold]` is the release gate for the ProcessAsync path: + // the inline rows at 1, 2 and N workers, three paired runs, medians; exit code 1 when N/1 is below the threshold. + if (args.Length > 0 && args[0] == "--scale") + { + double seconds = args.Length > 1 && double.TryParse(args[1], out var s) ? s : 3; + int workers = args.Length > 2 && int.TryParse(args[2], out var t) ? t : 16; + double threshold = args.Length > 3 && double.TryParse(args[3], out var r) ? r : 4.0; + bool pass = AsyncBenchmark.ScaleAsync(Console.WriteLine, seconds, workers, threshold).GetAwaiter().GetResult(); + Environment.ExitCode = pass ? 0 : 1; + return; + } + var interactive = !Console.IsOutputRedirected; if (interactive) Console.Clear(); IHardwareInfo hardwareInfo = new HardwareInfo(); hardwareInfo.RefreshAll(); HardwarePrinter.PrintHardware(hardwareInfo); - System.Threading.ThreadPool.SetMinThreads(Environment.ProcessorCount * 3, Environment.ProcessorCount * 3); - Console.WriteLine("Setting task pool size to {0}", Environment.ProcessorCount * 4); + Console.WriteLine("Thread pool minimum set to {0}", Environment.ProcessorCount * 3); // `dotnet run -- --sync [seconds] [threads]` runs the direct synchronous benchmark and exits (CI / scripted runs). if (args.Length > 0 && args[0] == "--sync") @@ -40,11 +53,13 @@ static void Main(string[] args) return; } - // `dotnet run -- --kestrel [seconds]` hosts the AspNetCore package in-process and drives it over HTTP and TCP. + // `dotnet run -- --kestrel [seconds] [async]` hosts the AspNetCore package in-process and drives it over HTTP and TCP; + // `async` turns EnableAsyncMethods on and adds a TCP row with yielding methods. if (args.Length > 0 && args[0] == "--kestrel") { double seconds = args.Length > 1 && double.TryParse(args[1], out var s) ? s : 3; - KestrelBenchmark.RunAsync(Console.WriteLine, seconds).GetAwaiter().GetResult(); + bool asyncMethods = args.Length > 2 && args[2] == "async"; + KestrelBenchmark.RunAsync(Console.WriteLine, seconds, asyncMethods: asyncMethods).GetAwaiter().GetResult(); return; } @@ -67,13 +82,18 @@ static void Main(string[] args) PrintOptions(); for (string line = Console.ReadLine(); line != null && line != "q"; line = Console.ReadLine()) { - if (!interactive && string.IsNullOrWhiteSpace(line)) + if (string.IsNullOrWhiteSpace(line) || line.StartsWith("s", StringComparison.CurrentCultureIgnoreCase)) { - BenchmarkRunner.Benchmark(Console.WriteLine); + BenchmarkRunner.BenchmarkSync(Console.WriteLine); } - else if (line.StartsWith("s", StringComparison.CurrentCultureIgnoreCase)) + else if (line.StartsWith("a", StringComparison.CurrentCultureIgnoreCase)) { - BenchmarkRunner.BenchmarkSync(Console.WriteLine); + AsyncBenchmark.RunAsync(Console.WriteLine, 3, 1).GetAwaiter().GetResult(); + AsyncBenchmark.RunAsync(Console.WriteLine, 3, Environment.ProcessorCount).GetAwaiter().GetResult(); + } + else if (line.StartsWith("t", StringComparison.CurrentCultureIgnoreCase)) + { + LegacyStringBenchmark(interactive, hardwareInfo); } else if (line.StartsWith("k", StringComparison.CurrentCultureIgnoreCase)) { @@ -83,37 +103,45 @@ static void Main(string[] args) { CompareBenchmark.RunAsync(Console.WriteLine).GetAwaiter().GetResult(); } - else if (string.IsNullOrWhiteSpace(line)) - { - Console.CursorVisible = false; - HardwarePrinter.PrintHardware(hardwareInfo); - var pos = Console.CursorTop; - BenchmarkRunner.Benchmark((update) => - { - // Clear the console from pos to current position first - var currentline = Console.CursorTop; - var fullLine = new string(' ', Console.WindowWidth); - Console.SetCursorPosition(0, pos); - for (int i = 0; i < currentline-pos; i++) - { - Console.WriteLine( fullLine); - } - - Console.SetCursorPosition(0, pos); - Console.WriteLine(update); - }); - Console.CursorVisible = true; - } else if (line.StartsWith("c", StringComparison.CurrentCultureIgnoreCase)) ConsoleInput(); PrintOptions(); } } + /// The 1.x string overloads through the thread pool, with live progress when there is a console to reposition. + private static void LegacyStringBenchmark(bool interactive, IHardwareInfo hardwareInfo) + { + if (!interactive) + { + BenchmarkRunner.Benchmark(Console.WriteLine); + return; + } + Console.CursorVisible = false; + HardwarePrinter.PrintHardware(hardwareInfo); + var pos = Console.CursorTop; + BenchmarkRunner.Benchmark((update) => + { + // Clear the console from pos to current position first + var currentline = Console.CursorTop; + var fullLine = new string(' ', Console.WindowWidth); + Console.SetCursorPosition(0, pos); + for (int i = 0; i < currentline - pos; i++) + { + Console.WriteLine(fullLine); + } + + Console.SetCursorPosition(0, pos); + Console.WriteLine(update); + }); + Console.CursorVisible = true; + } + private static void PrintOptions() { - Console.WriteLine("Hit Enter to run the Task-based benchmark"); - Console.WriteLine("'s' to run the direct synchronous benchmark (bytes in, bytes out)"); + Console.WriteLine("Hit Enter (or 's') for Process(bytes), dedicated threads: the library alone, the Sync table (--sync)"); + Console.WriteLine("'a' for ProcessAsync(bytes), awaited workers, at 1 and " + Environment.ProcessorCount + " workers: the Async table (--async)"); + Console.WriteLine("'t' for Legacy Process(string), scheduled synchronous work: the 1.x string overloads through the thread pool, not the byte path"); Console.WriteLine("'k' to run the Kestrel benchmark (AspNetCore package over HTTP and TCP)"); Console.WriteLine("'x' to compare against StreamJsonRpc and gRPC for .NET (same calls, same Kestrel)"); Console.WriteLine("'c' to start reading console input"); @@ -127,9 +155,5 @@ private static void ConsoleInput() JsonRpcProcessor.Process(line).ContinueWith(response => Console.WriteLine( response.Result )); } } - - } - - } diff --git a/benchmarks/Micro/README.md b/benchmarks/Micro/README.md index 5123b60..15aa57a 100644 --- a/benchmarks/Micro/README.md +++ b/benchmarks/Micro/README.md @@ -17,3 +17,11 @@ Benchmark classes: - `AsyncDispatchBenchmarks`: a synchronous method through `Process` and `ProcessAsync`, `Task` and `ValueTask` methods that complete inline, and a method that yields once, each with the default `RpcContextFlow.None` and with `RpcContextFlow.Flow`. The inline default rows should allocate nothing; the `Flow` rows pay for the execution-context bridge; the yielding rows show the cost of a real suspension. Read the `Allocated` column first: a non-zero value on a numeric shape means the request touched the GC, which the fast path must not do. Then compare `Mean`, but only between runs on an idle machine or within one run: background load biases ratios as well as absolute numbers, which is why `BindingComparisonBenchmarks` puts both registrations in one process. + +These rows run on one thread, so they cannot see a process-wide serialization point: the `AsyncScratch` pool lock capped `ProcessAsync` at about 4 M RPC/s on every core count while every row here stayed at 0 B and the same mean. Before a release, and after any change to dispatch, pooling or the async path, also run the scaling gate on the reference machine and paste its table into the release notes: + +```bash +dotnet run -c Release --project TestServer_Console -- --scale 3 16 4.0 +``` + +It fails when an inline row scales less than 4× from 1 to 16 workers (the lock gave 1.3; the per-thread cache gives about 9). The pull-request build runs a diagnostic `--scale 3 4 2.0` on the shared runner and an allowlist check of every `lock`, `Interlocked`, `Volatile.Write`, thread-static and writable static field on the request-path files (`.github/request-path-sync.allowlist`, each with a reason). diff --git a/benchmarks/charts/benchmarks.json b/benchmarks/charts/benchmarks.json index e6acc0f..adb10e7 100644 --- a/benchmarks/charts/benchmarks.json +++ b/benchmarks/charts/benchmarks.json @@ -27,6 +27,77 @@ } ] }, + "async16": { + "title": "ProcessAsync, 16 awaited workers, by registration", + "date": "2026-09-25", + "source": "README.md, Async table, 16-worker column; branch async-scratch-cache", + "harness": "TestServer_Console --async 3 16", + "workload": "the five benchmark requests (yieldsOnce: one request) through the byte-level JsonRpcProcessor.ProcessAsync from Task.Run workers that await each call", + "conditions": "other sessions running on the box; one 3 s run per row, so a lower bound", + "policy": "one run per row; re-measured on an idle box before a release", + "groups": [ + { "heading": "methods that complete inline, RpcContextFlow.None", "rule": false, "series": ["async16-sync-none", "async16-task-none", "async16-valuetask-none"] }, + { "heading": "methods that complete inline, RpcContextFlow.Flow", "rule": false, "series": ["async16-task-flow", "async16-valuetask-flow"] }, + { "heading": "one real suspension per request (yieldsOnce)", "rule": true, "series": ["async16-yield-none", "async16-yield-flow"] } + ], + "series": [ + { "id": "async16-sync-none", "label": "synchronous methods, None", "note": "6 B per request, the StringMe result string", "family": "ours", "low": 21200000, "high": 21200000, "status": "single", "published": "21.2 M" }, + { "id": "async16-task-none", "label": "Task, None", "note": "67 B per request, the service's own Task.FromResult", "family": "ours", "low": 25400000, "high": 25400000, "status": "single", "published": "25.4 M" }, + { "id": "async16-valuetask-none", "label": "ValueTask, None", "note": "6 B per request", "family": "ours", "low": 25700000, "high": 25700000, "status": "single", "published": "25.7 M" }, + { "id": "async16-task-flow", "label": "Task, Flow", "note": "251 B per request", "family": "ours", "low": 19800000, "high": 19800000, "status": "single", "published": "19.8 M" }, + { "id": "async16-valuetask-flow", "label": "ValueTask, Flow", "note": "190 B per request", "family": "ours", "low": 21900000, "high": 21900000, "status": "single", "published": "21.9 M" }, + { "id": "async16-yield-none", "label": "yieldsOnce, None", "note": "556 B per request", "family": "ours", "low": 8320000, "high": 8320000, "status": "single", "published": "8.32 M" }, + { "id": "async16-yield-flow", "label": "yieldsOnce, Flow", "note": "737 B per request", "family": "ours", "low": 6140000, "high": 6140000, "status": "single", "published": "6.14 M" } + ] + }, + "async1": { + "title": "ProcessAsync, one awaited worker, by registration", + "date": "2026-09-25", + "source": "README.md, Async table, 1-worker column; branch async-scratch-cache", + "harness": "TestServer_Console --async 3 1", + "workload": "as async16, one worker", + "conditions": "other sessions running on the box; one 3 s run per row, so a lower bound", + "policy": "one run per row; re-measured on an idle box before a release", + "groups": [ + { "heading": "methods that complete inline, RpcContextFlow.None", "rule": false, "series": ["async1-sync-none", "async1-task-none", "async1-valuetask-none"] }, + { "heading": "methods that complete inline, RpcContextFlow.Flow", "rule": false, "series": ["async1-task-flow", "async1-valuetask-flow"] }, + { "heading": "one real suspension per request (yieldsOnce)", "rule": true, "series": ["async1-yield-none", "async1-yield-flow"] } + ], + "series": [ + { "id": "async1-sync-none", "label": "synchronous methods, None", "family": "ours", "low": 3160000, "high": 3160000, "status": "single", "published": "3.16 M" }, + { "id": "async1-task-none", "label": "Task, None", "family": "ours", "low": 2420000, "high": 2420000, "status": "single", "published": "2.42 M" }, + { "id": "async1-valuetask-none", "label": "ValueTask, None", "family": "ours", "low": 2300000, "high": 2300000, "status": "single", "published": "2.30 M" }, + { "id": "async1-task-flow", "label": "Task, Flow", "family": "ours", "low": 1840000, "high": 1840000, "status": "single", "published": "1.84 M" }, + { "id": "async1-valuetask-flow", "label": "ValueTask, Flow", "family": "ours", "low": 1950000, "high": 1950000, "status": "single", "published": "1.95 M" }, + { "id": "async1-yield-none", "label": "yieldsOnce, None", "family": "ours", "low": 887000, "high": 887000, "status": "single", "published": "887 k" }, + { "id": "async1-yield-flow", "label": "yieldsOnce, Flow", "family": "ours", "low": 671000, "high": 671000, "status": "single", "published": "671 k" } + ] + }, + "legacy": { + "title": "Legacy string API through the thread pool, by batch size", + "date": "2026-09-23", + "source": "README.md, Legacy table; commit 192a97f", + "harness": "TestServer_Console, menu entry t", + "workload": "the five benchmark requests through the 1.x Task Process(string) overload, submitted in batches from every core through Parallel.For", + "conditions": "idle box", + "policy": "one run per row", + "x": { "name": "batch size", "values": [50, 300, 6000, 36000, 252000, 2016000] }, + "series": [ + { + "id": "legacy-string", + "label": "JSON-RPC.Net, 1.x string overloads", + "family": "ours", + "points": [ + { "low": 1500000, "high": 1500000, "status": "single", "published": "1.5 M" }, + { "low": 7800000, "high": 7800000, "status": "single", "published": "7.8 M" }, + { "low": 10900000, "high": 10900000, "status": "single", "published": "10.9 M" }, + { "low": 12000000, "high": 12000000, "status": "single", "published": "12.0 M" }, + { "low": 7600000, "high": 7600000, "status": "single", "published": "7.6 M" }, + { "low": 7200000, "high": 7200000, "status": "single", "published": "7.2 M" } + ] + } + ] + }, "kestrel": { "title": "JSON-RPC.Net by transport", "date": "2026-09-23", @@ -36,13 +107,16 @@ "conditions": "WSL VM shut down; two 3 s runs", "policy": "observed low and high over two runs", "groups": [ - { "heading": "", "rule": false, "series": ["ours-http-1", "ours-http-100", "ours-tcp"] }, + { "heading": "EnableAsyncMethods = false (the default)", "rule": false, "series": ["ours-http-1", "ours-http-100", "ours-tcp"] }, + { "heading": "EnableAsyncMethods = true: every document through ProcessAsync (2026-09-25, one run, busy machine)", "rule": false, "series": ["ours-tcp-async-inline", "ours-tcp-async-yield"] }, { "heading": "Reference: no transport", "rule": true, "series": ["ours-inproc-16"] } ], "series": [ { "id": "ours-http-1", "label": "HTTP, 1 request per POST", "note": "each client awaits one POST at a time; 95 to 125 µs per round trip", "family": "ours", "low": 128000, "high": 168000, "status": "range", "published": "128 k to 168 k" }, { "id": "ours-http-100", "label": "HTTP, batch of 100 per POST", "note": "one POST outstanding per client, 100 RPCs in it", "family": "ours", "low": 12700000, "high": 13700000, "status": "range", "published": "12.7 M to 13.7 M" }, { "id": "ours-tcp", "label": "TCP, 256 pipelined", "note": "JsonRpcConnectionHandler, 256 requests in flight per connection", "family": "ours", "low": 15200000, "high": 15500000, "status": "range", "published": "15.2 M to 15.5 M" }, + { "id": "ours-tcp-async-inline", "label": "TCP, 256 pipelined, methods that complete inline", "note": "--kestrel 3 async; the same run's default-mode TCP row was 14.0 M", "family": "ours", "low": 13300000, "high": 13300000, "status": "single", "published": "13.3 M" }, + { "id": "ours-tcp-async-yield", "label": "TCP, 256 pipelined, methods that suspend once", "note": "five async Task methods awaiting Task.Yield(); documents are sequential per connection", "family": "ours", "low": 1130000, "high": 1130000, "status": "single", "published": "1.13 M" }, { "id": "ours-inproc-16", "label": "in-process, 16 threads", "note": "the same requests through the byte entry point", "family": "scale", "low": 30800000, "high": 31300000, "status": "range", "published": "30.8 M to 31.3 M" } ] }, diff --git a/benchmarks/charts/compare-connections-dark.svg b/benchmarks/charts/compare-connections-dark.svg index 98891d0..1a13499 100644 --- a/benchmarks/charts/compare-connections-dark.svg +++ b/benchmarks/charts/compare-connections-dark.svg @@ -1,5 +1,5 @@ - -Every library and transport, by client connectionsSame five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight per connection; HTTP clients await one POST at a time (1 or 100 RPCs). 5 runs of 2 s per point; whiskers span the runs. Data revision a4f53e6a848f. + +Every library and transport, by client connectionsSame five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight per connection; HTTP clients await one POST at a time (1 or 100 RPCs). 5 runs of 2 s per point; whiskers span the runs. Data revision d3cafb28c8a9. Every library and transport, by client connections Same five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight diff --git a/benchmarks/charts/compare-connections.svg b/benchmarks/charts/compare-connections.svg index 29194c5..afe756b 100644 --- a/benchmarks/charts/compare-connections.svg +++ b/benchmarks/charts/compare-connections.svg @@ -1,5 +1,5 @@ - -Every library and transport, by client connectionsSame five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight per connection; HTTP clients await one POST at a time (1 or 100 RPCs). 5 runs of 2 s per point; whiskers span the runs. Data revision a4f53e6a848f. + +Every library and transport, by client connectionsSame five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight per connection; HTTP clients await one POST at a time (1 or 100 RPCs). 5 runs of 2 s per point; whiskers span the runs. Data revision d3cafb28c8a9. Every library and transport, by client connections Same five calls, same Kestrel, clients on the server's 8 cores, 2026-09-23. TCP and gRPC keep 256 requests in flight diff --git a/benchmarks/charts/compare-streamjsonrpc-dark.svg b/benchmarks/charts/compare-streamjsonrpc-dark.svg index d427a26..8b831c8 100644 --- a/benchmarks/charts/compare-streamjsonrpc-dark.svg +++ b/benchmarks/charts/compare-streamjsonrpc-dark.svg @@ -1,5 +1,5 @@ - -JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connectionsSame five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. Every request carries "jsonrpc":"2.0" (StreamJsonRpc requires it). Intervals span the runs; a dot is a single value. Data revision a4f53e6a848f. + +JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connectionsSame five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. Every request carries "jsonrpc":"2.0" (StreamJsonRpc requires it). Intervals span the runs; a dot is a single value. Data revision d3cafb28c8a9. JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connections Same five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. diff --git a/benchmarks/charts/compare-streamjsonrpc.svg b/benchmarks/charts/compare-streamjsonrpc.svg index f283484..5fc01ea 100644 --- a/benchmarks/charts/compare-streamjsonrpc.svg +++ b/benchmarks/charts/compare-streamjsonrpc.svg @@ -1,5 +1,5 @@ - -JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connectionsSame five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. Every request carries "jsonrpc":"2.0" (StreamJsonRpc requires it). Intervals span the runs; a dot is a single value. Data revision a4f53e6a848f. + +JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connectionsSame five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. Every request carries "jsonrpc":"2.0" (StreamJsonRpc requires it). Intervals span the runs; a dot is a single value. Data revision d3cafb28c8a9. JSON-RPC.Net vs StreamJsonRpc vs gRPC for .NET, 16 connections Same five calls, same Kestrel, 16 connections or channels with 256 requests in flight each, two 3 s runs. diff --git a/benchmarks/charts/explorer.html b/benchmarks/charts/explorer.html index f1323dc..9d1d855 100644 --- a/benchmarks/charts/explorer.html +++ b/benchmarks/charts/explorer.html @@ -5,7 +5,7 @@ JSON-RPC.NET benchmark explorer - +