feat(client): add stall gate backpressure for slow consumers
Adds StallGate nested class inside NatsClient that blocks producers when the outbound buffer exceeds 75% of maxPending capacity, modelling Go's stc channel and stalledRoute handling in server/client.go.
This commit is contained in:
@@ -966,4 +966,81 @@ public sealed class NatsClient : INatsClient, IDisposable
|
||||
_stream.Dispose();
|
||||
_socket.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Blocks producers when the client's outbound buffer is near capacity.
|
||||
/// Go reference: server/client.go (stc channel, stalledRoute handling).
|
||||
/// When pending bytes exceed 75% of maxPending, producers must wait until
|
||||
/// the write loop drains enough data.
|
||||
/// </summary>
|
||||
public sealed class StallGate
|
||||
{
|
||||
private readonly long _threshold;
|
||||
private volatile SemaphoreSlim? _semaphore;
|
||||
private readonly Lock _gate = new();
|
||||
|
||||
/// <summary>
|
||||
/// Creates a stall gate with the given maxPending capacity.
|
||||
/// The stall threshold is set at 75% of maxPending.
|
||||
/// Go reference: server/client.go stc channel creation.
|
||||
/// </summary>
|
||||
public StallGate(long maxPending)
|
||||
{
|
||||
_threshold = maxPending * 3 / 4;
|
||||
}
|
||||
|
||||
/// <summary>Whether producers are currently being stalled.</summary>
|
||||
public bool IsStalled
|
||||
{
|
||||
get { lock (_gate) return _semaphore is not null; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Updates pending byte count and activates/deactivates the stall gate.
|
||||
/// Go reference: server/client.go stalledRoute check.
|
||||
/// </summary>
|
||||
public void UpdatePending(long pending)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (pending >= _threshold && _semaphore is null)
|
||||
{
|
||||
_semaphore = new SemaphoreSlim(0, 1);
|
||||
}
|
||||
else if (pending < _threshold && _semaphore is not null)
|
||||
{
|
||||
Release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Waits for the stall gate to release. Returns true if released,
|
||||
/// false if timed out (indicating the client should be closed as slow consumer).
|
||||
/// Go reference: server/client.go stc channel receive with timeout.
|
||||
/// </summary>
|
||||
public async Task<bool> WaitAsync(TimeSpan timeout)
|
||||
{
|
||||
SemaphoreSlim? sem;
|
||||
lock (_gate) sem = _semaphore;
|
||||
if (sem is null) return true;
|
||||
return await sem.WaitAsync(timeout);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Releases any blocked producers. Called when the write loop has drained
|
||||
/// enough data to bring pending bytes below the threshold.
|
||||
/// </summary>
|
||||
public void Release()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_semaphore is not null)
|
||||
{
|
||||
_semaphore.Release();
|
||||
_semaphore = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user