dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: batch stat increments per read cycle in ProcessCommandsAsync

Browse Source 
Accumulate InMsgs/InBytes locally per ReadAsync cycle and flush once,
reducing from 4 Interlocked operations per published message to 2 per
read cycle. This matches the Go server's approach of batching stats.
This commit is contained in:
Joseph Doherty
2026-02-22 23:52:09 -05:00
parent c522ce99f5
commit bce793fd42
1 changed files with 32 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/NATS.Server/NatsClient.cs
View File

@@ -221,10 +221,37 @@ public sealed class NatsClient : IDisposable
var result = await reader.ReadAsync(ct);
var buffer = result.Buffer;
long localInMsgs = 0;
long localInBytes = 0;
while (_parser.TryParse(ref buffer, out var cmd))
{
Interlocked.Exchange(ref _lastIn, Environment.TickCount64);
await DispatchCommandAsync(cmd, ct);
// Handle Pub/HPub inline to allow ref parameter passing for stat batching.
// DispatchCommandAsync is async and cannot accept ref parameters.
if (cmd.Type is CommandType.Pub or CommandType.HPub
&& (!_authService.IsAuthRequired || ConnectReceived))
{
Interlocked.Exchange(ref _lastActivityTicks, DateTime.UtcNow.Ticks);
ProcessPub(cmd, ref localInMsgs, ref localInBytes);
}
else
{
await DispatchCommandAsync(cmd, ct);
}
}
if (localInMsgs > 0)
{
Interlocked.Add(ref InMsgs, localInMsgs);
Interlocked.Add(ref _serverStats.InMsgs, localInMsgs);
}
if (localInBytes > 0)
{
Interlocked.Add(ref InBytes, localInBytes);
Interlocked.Add(ref _serverStats.InBytes, localInBytes);
}
reader.AdvanceTo(buffer.Start, buffer.End);
@@ -285,7 +312,7 @@ public sealed class NatsClient : IDisposable
case CommandType.Pub:
case CommandType.HPub:
ProcessPub(cmd);
// Pub/HPub is handled inline in ProcessCommandsAsync for stat batching
break;
}
}
@@ -386,12 +413,10 @@ public sealed class NatsClient : IDisposable
Account?.SubList.Remove(sub);
}
private void ProcessPub(ParsedCommand cmd)
private void ProcessPub(ParsedCommand cmd, ref long localInMsgs, ref long localInBytes)
{
Interlocked.Increment(ref InMsgs);
Interlocked.Add(ref InBytes, cmd.Payload.Length);
Interlocked.Increment(ref _serverStats.InMsgs);
Interlocked.Add(ref _serverStats.InBytes, cmd.Payload.Length);
localInMsgs++;
localInBytes += cmd.Payload.Length;
// Max payload validation (always, hard close)
if (cmd.Payload.Length > _options.MaxPayload)