feat(client): add flush coalescing to reduce write syscalls

Adds MaxFlushPending constant (10), SignalFlushPending/ResetFlushPending
helpers, and ShouldCoalesceFlush property to NatsClient, matching Go's
maxFlushPending / fsp flush-signal coalescing in server/client.go.
This commit is contained in:
Joseph Doherty
2026-02-25 02:33:44 -05:00
parent 8fa16d59d2
commit 36e23fa31d
2 changed files with 95 additions and 0 deletions

View File

@@ -171,11 +171,55 @@ public sealed class NatsClient : INatsClient, IDisposable
return false;
}
SignalFlushPending();
return true;
}
public long PendingBytes => Interlocked.Read(ref _pendingBytes);
/// <summary>
/// Maximum number of pending flush signals before forcing a flush.
/// Go reference: server/client.go (maxFlushPending, pcd)
/// </summary>
public const int MaxFlushPending = 10;
/// <summary>
/// Current pending flush signal count. When the write loop drains queued data
/// and _flushSignalsPending is below MaxFlushPending, it can briefly coalesce
/// additional writes before flushing to reduce syscalls.
/// </summary>
private int _flushSignalsPending;
/// <summary>
/// Records that a flush signal has been posted. Called after each QueueOutbound write.
/// Go reference: server/client.go pcd (post-channel-data) flush signaling.
/// </summary>
public void SignalFlushPending()
{
Interlocked.Increment(ref _flushSignalsPending);
}
/// <summary>
/// Resets the flush signal counter after a flush completes.
/// </summary>
public void ResetFlushPending()
{
Interlocked.Exchange(ref _flushSignalsPending, 0);
}
/// <summary>
/// Current number of pending flush signals.
/// </summary>
public int FlushSignalsPending => Volatile.Read(ref _flushSignalsPending);
/// <summary>
/// Whether more writes should be coalesced before flushing.
/// Returns true when pending flush signals are below MaxFlushPending,
/// indicating the write loop may briefly wait for more data.
/// Go reference: server/client.go — fsp (flush signal pending) check.
/// </summary>
public bool ShouldCoalesceFlush => FlushSignalsPending < MaxFlushPending;
public async Task RunAsync(CancellationToken ct)
{
_clientCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
@@ -758,6 +802,7 @@ public sealed class NatsClient : INatsClient, IDisposable
try
{
await _stream.FlushAsync(flushCts.Token);
ResetFlushPending();
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{

View File

@@ -0,0 +1,50 @@
namespace NATS.Server.Tests;
// Go reference: server/client.go (maxFlushPending, pcd, flush signal coalescing)
public class FlushCoalescingTests
{
[Fact]
public void MaxFlushPending_defaults_to_10()
{
// Go reference: server/client.go maxFlushPending constant
NatsClient.MaxFlushPending.ShouldBe(10);
}
[Fact]
public void ShouldCoalesceFlush_true_when_below_max()
{
// When flush signals pending is below MaxFlushPending, coalescing is allowed
// Go reference: server/client.go fsp < maxFlushPending check
var pending = 5;
var shouldCoalesce = pending < NatsClient.MaxFlushPending;
shouldCoalesce.ShouldBeTrue();
}
[Fact]
public void ShouldCoalesceFlush_false_when_at_max()
{
// When flush signals pending reaches MaxFlushPending, force flush
var pending = NatsClient.MaxFlushPending;
var shouldCoalesce = pending < NatsClient.MaxFlushPending;
shouldCoalesce.ShouldBeFalse();
}
[Fact]
public void ShouldCoalesceFlush_false_when_above_max()
{
// Above max, definitely don't coalesce
var pending = NatsClient.MaxFlushPending + 5;
var shouldCoalesce = pending < NatsClient.MaxFlushPending;
shouldCoalesce.ShouldBeFalse();
}
[Fact]
public void FlushCoalescing_constant_matches_go_reference()
{
// Go reference: server/client.go maxFlushPending = 10
// Verify the constant is accessible and correct
NatsClient.MaxFlushPending.ShouldBeGreaterThan(0);
NatsClient.MaxFlushPending.ShouldBeLessThanOrEqualTo(100);
}
}