Skip to content

Commit 979a375

Browse files
AstnAstn
committed
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.
1 parent 160897a commit 979a375

29 files changed

Lines changed: 1003 additions & 245 deletions

‎.github/request-path-sync.allowlist‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
# Every `lock (`, `Interlocked.`, `Volatile.Write`, `[ThreadStatic]` field and writable static field in the
2+
# request-path files (core and both companion serializers), with the reason it is allowed there.
3+
# Format: file | the code line, trimmed, without its trailing comment | reason, starting with its tag:
4+
# per-thread the field is thread-static, so no line is shared
5+
# miss-path touched only when a per-thread cache misses or overflows, never by a warm inline document
6+
# registration-only written when methods or sessions are bound, never by a request
7+
# read-only-after-init written once at startup, then only read
8+
# shared-write a request can write it and every core reads it: the reason must say why it is tolerated
9+
# Checked by .github/scripts/check_request_path_sync.py in the pull-request workflow. The measurement that backs this
10+
# list is `TestServer_Console --scale` (README, Benchmarks). A `[ThreadStatic]` attribute on its own line is not
11+
# listed; the field line under it is.
12+
Json-Rpc/JsonRpcProcessor.Async.cs | private static int _count; | miss-path: the shared pool's fill count, read and written only under lock (Pool)
13+
Json-Rpc/JsonRpcProcessor.Async.cs | [ThreadStatic] private static AsyncScratch _slot; | per-thread: the one-slot cache of an idle async scratch
14+
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)
15+
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
16+
Json-Rpc/JsonRpcProcessor.cs | [ThreadStatic] private static Scratch _current; | per-thread: the synchronous scratch
17+
Json-Rpc/Handler.Async.cs | [ThreadStatic] private static InvocationState __unflowedState; | per-thread: the invocation frame of a RpcContextFlow.None call
18+
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
19+
Json-Rpc/Handler.cs | Interlocked.Increment(ref _sessionHandlerMasterVersion); | registration-only: GetSessionHandler creating a session and DestroySession; never on the request path
20+
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
21+
Json-Rpc/Handler.cs | private static int _sessionHandlerLocalVersion = 0; | per-thread ([ThreadStatic] on the line above): the snapshot's version
22+
Json-Rpc/Handler.cs | private static string _lastSessionId; | per-thread ([ThreadStatic] on the line above): the last-session cache
23+
Json-Rpc/Handler.cs | private static Handler _lastSessionHandler; | per-thread ([ThreadStatic] on the line above): the last-session cache
24+
Json-Rpc/Handler.cs | private static InvocationState __state; | per-thread ([ThreadStatic] on the line above): the current invocation frame
25+
Json-Rpc/Jsmn/JsmnSerializer.cs | [ThreadStatic] private static JsmnTokenizer _scratch; | per-thread: the tokenizer scratch
26+
Json-Rpc/Jsmn/JsmnSerializer.cs | [ThreadStatic] private static bool _scratchInUse; | per-thread: re-entrancy flag for the tokenizer scratch
27+
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
28+
Json-Rpc/Serialization/Utf8KeyTable.cs | System.Threading.Volatile.Write(ref _snapshot, snapshot); | registration-only: publishes a rebuilt snapshot; read-only afterwards
29+
Json-Rpc/RpcBinding.cs | lock (_sync) | registration-only: RpcBinding.Dispose unbinding an interface tree
30+
AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static Utf8JsonWriter _cachedWriter; | per-thread: the cached writer
31+
AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static JsonSerializerOptions _cachedWriterOptions; | per-thread: the options the cached writer was built with
32+
AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static bool _cachedWriterInUse; | per-thread: re-entrancy flag for the cached writer
33+
AustinHarris.JsonRpc.SystemTextJson/SystemTextJsonRpcSerializer.cs | [ThreadStatic] private static byte[] _scratch; | per-thread: transcoding scratch
34+
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
35+
AustinHarris.JsonRpc.Newtonsoft/NewtonsoftJsonRpcSerializer.cs | [ThreadStatic] private static Scratch _current; | per-thread: the Json.NET scratch

‎.github/scripts/check_request_path_sync.py‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
#!/usr/bin/env python3
2+
"""Every synchronization point and every writable static on the request path must be listed, with a reason, in
3+
.github/request-path-sync.allowlist: `lock (`, `Interlocked.`, `Volatile.Write`, `[ThreadStatic]` fields and static
4+
fields that are not readonly, in the request-path files of the core and of both companion serializers.
5+
6+
A process-wide serialization point on the per-document path caps ProcessAsync at a few million requests per second
7+
on every core count (the AsyncScratch pool lock, 2026-09-25), and the single-threaded micro-benchmarks cannot see
8+
it; a static that one request writes and every core reads costs the same kind of cache-line traffic without a lock.
9+
This check is a review aid: it catches a new lock, atomic or writable static on those files and asks for a written
10+
reason (per-thread, miss-path, registration-only, read-only-after-init). It is not the measurement;
11+
`TestServer_Console --scale` is (a probe moved behind an allowed lock would pass here and fail there). Mutable
12+
objects reached through readonly references are outside its reach and belong to review.
13+
14+
Usage: python3 .github/scripts/check_request_path_sync.py (from the repository root; exit 1 on any unlisted use)
15+
"""
16+
import glob
17+
import os
18+
import re
19+
import sys
20+
21+
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
22+
ALLOWLIST = os.path.join(ROOT, ".github", "request-path-sync.allowlist")
23+
PATTERNS = ["Json-Rpc/JsonRpcProcessor*.cs", "Json-Rpc/Handler*.cs", "Json-Rpc/Invocation/*.cs", "Json-Rpc/Jsmn/*.cs",
24+
"Json-Rpc/Serialization/*.cs", "Json-Rpc/JsonRpcContext.cs", "Json-Rpc/JsonRpcRequestId.cs", "Json-Rpc/RpcBinding.cs",
25+
"AustinHarris.JsonRpc.SystemTextJson/*.cs", "AustinHarris.JsonRpc.Newtonsoft/*.cs"]
26+
USE = re.compile(r"\block\s*\(|\bInterlocked\.|\bVolatile\.Write")
27+
# A static field declaration that is not readonly or const: `[ThreadStatic] private static T name;` or `static T name = ...;`.
28+
# Expression-bodied members (`=>`), methods, classes, events and operators are not fields.
29+
FIELD = re.compile(r"^(?:\[ThreadStatic\]\s*)?(?:(?:public|private|internal|protected|new|volatile)\s+)*static\s+"
30+
r"(?!readonly\b|const\b|class\b|void\b|partial\b|event\b|explicit\b|implicit\b|operator\b)"
31+
r"(?!.*=>)(?!.*\()[\w<>\[\],.?\s]+?\s+\w+\s*(?:=[^;]*)?;$")
32+
33+
34+
def load_allowlist():
35+
"""{(file, code line stripped): reason}; lines are `file | code | reason`, `#` comments and blanks ignored."""
36+
allowed = {}
37+
with open(ALLOWLIST, encoding="utf-8") as f:
38+
for n, line in enumerate(f, 1):
39+
line = line.strip()
40+
if not line or line.startswith("#"):
41+
continue
42+
parts = [p.strip() for p in line.split("|", 2)]
43+
if len(parts) != 3 or not all(parts):
44+
sys.exit(f"{ALLOWLIST}:{n}: expected `file | code | reason`")
45+
allowed[(parts[0].replace("\\", "/"), parts[1])] = parts[2]
46+
return allowed
47+
48+
49+
def main():
50+
allowed = load_allowlist()
51+
seen = set()
52+
problems = []
53+
for pattern in PATTERNS:
54+
for path in sorted(glob.glob(os.path.join(ROOT, pattern))):
55+
rel = os.path.relpath(path, ROOT).replace("\\", "/")
56+
with open(path, encoding="utf-8-sig") as f:
57+
for n, line in enumerate(f, 1):
58+
code = line.split("//", 1)[0].strip()
59+
if not (USE.search(code) or FIELD.match(code)):
60+
continue
61+
key = (rel, code)
62+
if key in allowed:
63+
seen.add(key)
64+
else:
65+
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")
66+
for key in allowed:
67+
if key not in seen:
68+
problems.append(f"{os.path.relpath(ALLOWLIST, ROOT)}: `{key[1]}` in {key[0]} no longer exists; remove the entry")
69+
for p in problems:
70+
print(p)
71+
if problems:
72+
return 1
73+
print(f"request-path synchronization: {len(seen)} listed uses, none unlisted")
74+
return 0
75+
76+
77+
if __name__ == "__main__":
78+
sys.exit(main())

‎.github/workflows/build_pull_request.yml‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,3 +42,41 @@ jobs:
4242
for project in Json-Rpc AustinHarris.JsonRpc.Newtonsoft AustinHarris.JsonRpc.SystemTextJson AustinHarris.JsonRpc.AspNetCore; do
4343
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
4444
done
45+
46+
# Every lock and Interlocked on the request-path files must be listed with a reason (a process-wide lock on the
47+
# per-document path once capped ProcessAsync at 4 M RPC/s on every core count). A review aid, not the measurement.
48+
request-path-sync:
49+
runs-on: ubuntu-latest
50+
steps:
51+
- uses: actions/checkout@v7
52+
- name: Synchronization on the request path is allowlisted
53+
run: python3 .github/scripts/check_request_path_sync.py
54+
55+
# The measurement: ProcessAsync must scale from 1 to 4 workers. A process-wide serialization point holds the ratio
56+
# near 1.3 on any core count. Diagnostic on the shared runner (its core count and isolation are not promised, so
57+
# this job is not required and continues on error); the release gate is `--scale 3 16 4.0` on the reference machine.
58+
scaling:
59+
runs-on: ubuntu-latest
60+
continue-on-error: true
61+
steps:
62+
- uses: actions/checkout@v7
63+
- name: Setup .NET
64+
uses: actions/setup-dotnet@v6
65+
with:
66+
global-json-file: global.json
67+
dotnet-version: 10.0.x
68+
- name: Build the harness
69+
run: dotnet build TestServer_Console --configuration Release
70+
- name: ProcessAsync scales from 1 to 4 workers (4/1 at least 2.0)
71+
run: |
72+
set +e
73+
dotnet run -c Release --no-build --project TestServer_Console -- --scale 3 4 2.0 | tee scale.txt
74+
status=${PIPESTATUS[0]}
75+
{
76+
echo "## ProcessAsync scaling (diagnostic, not required)"
77+
echo
78+
grep -E '^\|' scale.txt
79+
echo
80+
[ "$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**"
81+
} >> "$GITHUB_STEP_SUMMARY"
82+
exit $status

‎AustinHarris.JsonRpc.AspNetCore/README.md‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -123,9 +123,14 @@ invoked.
123123

124124
- **HTTP:** the call is cancelled when the client disconnects (`HttpContext.RequestAborted`). Notifications are
125125
awaited and still answer `204`. The body reader stays leased until the invocation finishes.
126-
- **Raw connections:** documents are processed one at a time, in order. Replies already finished are flushed
127-
before the connection waits on a slow method. When the connection closes, the running method is waited for and
128-
its response discarded.
126+
- **Raw connections:** documents are processed one at a time, in order, so 256 pipelined requests on one
127+
connection are 256 sequential invocations, not 256 concurrent suspensions; concurrency comes from connections.
128+
Replies already finished are flushed before the connection waits on a slow method. When the connection closes,
129+
the running method is waited for and its response discarded.
130+
- **Cost:** every document then goes through `ProcessAsync`. With methods that complete inline the host measures
131+
within a few percent of the synchronous mode; a method that really suspends pays its own async state plus the
132+
library's completion state (about 560 B) and a continuation per request. The main README's Kestrel table has
133+
both rows, measured with `TestServer_Console --kestrel 3 async`.
129134

130135
A method receives the token by declaring a `[JsonRpcCancellation] CancellationToken` parameter; see
131136
[Asynchronous methods and cancellation](https://github.com/Astn/JSON-RPC.NET#asynchronous-methods-and-cancellation)

‎AustinHarris.JsonRpcTestN/AsyncInvocationTests.cs‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,12 @@ public sealed class AsyncInvocationTests
2424
[TearDown] public void TearDown() => Handler.DestroySession(_session);
2525
private void Bind(string name, Delegate method, RpcContextFlow flow = RpcContextFlow.Flow) => ServiceBinder.BindMethod(_session, name, method, contextFlow: flow);
2626
private Task<string> Run(string json, JsonRpcSerializer serializer = null, object context = null, CancellationToken token = default) => JsonRpcProcessor.ProcessAsync(_session, json, context, serializer, token);
27+
private string Sync(string json, object context = null)
28+
{
29+
var output = new ArrayBufferWriter<byte>();
30+
JsonRpcProcessor.Process(_session, Encoding.UTF8.GetBytes(json).AsSpan(), output, context);
31+
return Encoding.UTF8.GetString(output.WrittenSpan);
32+
}
2733
private static TaskCompletionSource<int> Gate() => new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
2834
private static string Request(string method, string parameters = null, string id = "1") => "{\"method\":\"" + method + "\"" + (parameters == null ? "" : ",\"params\":" + parameters) + (id == null ? "" : ",\"id\":" + id) + "}";
2935
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()
309315
await escaped;
310316
}
311317

318+
[Test]
319+
public async Task FlowScope_CompletedOnAnotherThread_LeavesThatThreadItsOwnFrame()
320+
{
321+
// A hooked Flow invocation suspends here and completes on a pool thread, where its scope is
322+
// disposed. That thread must keep its own frame afterwards: when it was handed this thread's
323+
// frame instead, a synchronous dispatch on each thread saved and restored the same frame, and
324+
// the interleaving below left this thread's method reading no id at all.
325+
var handler = Handler.GetSessionHandler(_session);
326+
handler.SetPreProcessHandler((request, context) => null);
327+
int completedOn = 0;
328+
handler.SetPostProcessHandler((request, response, context) => { if (request.Method == "suspend") completedOn = Environment.CurrentManagedThreadId; return null; });
329+
var suspended = new TaskCompletionSource<int>();
330+
Bind("suspend", new Func<Task<int>>(() => suspended.Task));
331+
var entered = new ManualResetEventSlim();
332+
var release = new ManualResetEventSlim();
333+
var left = new ManualResetEventSlim();
334+
Bind("hold", new Func<int>(() => { entered.Set(); release.Wait(); return 1; }));
335+
var mine = new object();
336+
Bind("peek", new Func<int>(() =>
337+
{
338+
release.Set();
339+
left.Wait();
340+
return (Handler.RpcRequestId().IsAbsent ? 0 : 1) + (ReferenceEquals(Handler.RpcContext(), mine) ? 2 : 0);
341+
}));
342+
Bind("warm", new Func<int>(() => 1));
343+
Sync(Request("warm")); // this thread owns a frame before the suspension, as any thread that has dispatched does
344+
var pending = Run(Request("suspend"));
345+
int completingThread = 0;
346+
var other = Task.Run(() =>
347+
{
348+
completingThread = Environment.CurrentManagedThreadId;
349+
suspended.SetResult(7);
350+
Sync(Request("hold", id: "2"));
351+
left.Set();
352+
});
353+
// Block, do not await, until the other thread is inside "hold": an awaited continuation could be
354+
// run on that thread, ahead of "hold", and wait for itself.
355+
entered.Wait();
356+
var response = Sync(Request("peek", id: "\"mine\""), mine);
357+
await other;
358+
Assert.AreEqual("{\"jsonrpc\":\"2.0\",\"result\":7,\"id\":1}", await pending);
359+
Assume.That(completedOn, Is.EqualTo(completingThread), "the completion did not run the scope's cleanup on the completing thread");
360+
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");
361+
}
362+
312363
[TestCaseSource(nameof(Serializers))]
313364
public async Task Batch_IsSequential_AwaitsNotifications_AndIsolatesFaults(string name)
314365
{

0 commit comments

Comments
 (0)