ProcessAsync: per-thread scratch cache, scaling gate, async rows, frame fix

.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<string, Handler> _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<T>'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

.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())

.github/workflows/build_pull_request.yml

@@ -34,3 +34,41 @@ jobs:
         run: dotnet test AustinHarris.JsonRpcTestN --configuration Release --no-build
       # Pull requests do not publish. Packages go out from the master workflow through Trusted Publishing, whose
       # nuget.org policy is bound to build_publish_master.yml; nothing else can push.
+
+  # 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

AustinHarris.JsonRpc.AspNetCore/README.md

@@ -156,9 +156,15 @@ 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. On one connection, 256 pipelined requests
+  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 handler waits for the running method to finish and discards its response.
+- **Cost:** every document then goes through `ProcessAsync`. With methods that complete inline, the TCP row measured
+  15.0 M to 15.4 M against 14.3 M to 16.5 M for the synchronous mode on the same day, inside its spread. A method that
+  suspends pays for its own async state, the library's completion state and a continuation per request; the main README's
+  Async table measured 559 B per request for the yielding None row in process at one worker, including the service's own
+  allocations. 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)

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<string> 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<byte>();
+            JsonRpcProcessor.Process(_session, Encoding.UTF8.GetBytes(json).AsSpan(), output, context);
+            return Encoding.UTF8.GetString(output.WrittenSpan);
+        }
         private static TaskCompletionSource<int> Gate() => new TaskCompletionSource<int>(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<int>();
+            Bind("suspend", new Func<Task<int>>(() => suspended.Task));
+            var entered = new ManualResetEventSlim();
+            var release = new ManualResetEventSlim();
+            var left = new ManualResetEventSlim();
+            Bind("hold", new Func<int>(() => { entered.Set(); release.Wait(); return 1; }));
+            var mine = new object();
+            Bind("peek", new Func<int>(() =>
+            {
+                release.Set();
+                left.Wait();
+                return (Handler.RpcRequestId().IsAbsent ? 0 : 1) + (ReferenceEquals(Handler.RpcContext(), mine) ? 2 : 0);
+            }));
+            Bind("warm", new Func<int>(() => 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)
         {