From 95b46981ff997a6a2b2df18ba84ea629f8fa643d Mon Sep 17 00:00:00 2001 From: flink Date: Mon, 20 Jul 2026 05:12:40 -0400 Subject: [PATCH 1/3] Fix NetworkWriter.AsyncFlushPages hanging FlushEvent waiters on send failure When networkSender.SendResponse throws inside AsyncFlushPages (e.g. because the connection was disposed concurrently, such as after a cluster gossip timeout), the catch block logged the exception and disposed the handler but never invoked AsyncFlushPageCallback for that page. That callback is the only place that decrements the page-completion counter and signals FlushEvent, so any caller already blocked on flushEvent.Wait (e.g. GarnetClient.InternalExecuteNoResponse, used by cluster PUBLISH forwarding) would hang until the next gossip round disposed the whole NetworkWriter. This addresses the remaining "main thread blocking" symptom described in #1929 that is not covered by #1931, which only fixes the separate NullReferenceException in GarnetServerNode.TryClusterPublish. This change does not touch that code path and does not claim to fully resolve #1929. Adds a regression test that reproduces the hang by forcing a send failure while a thread is parked on the flush-completion event, and verifies it is released instead of only unblocking via a cancellation timeout. The test needs direct access to NetworkWriter's internals to inject the failure and observe the specific completion-event generation, so Garnet.client grants InternalsVisibleTo to Garnet.test (matching the pattern already used by other libs projects). --- libs/client/Garnet.client.csproj | 4 + libs/client/NetworkWriter.cs | 3 + .../Garnet.test/GarnetClientTests.cs | 102 ++++++++++++++++++ 3 files changed, 109 insertions(+) diff --git a/libs/client/Garnet.client.csproj b/libs/client/Garnet.client.csproj index 0a436edfd6a..b9b756ba48b 100644 --- a/libs/client/Garnet.client.csproj +++ b/libs/client/Garnet.client.csproj @@ -13,4 +13,8 @@ + + + + \ No newline at end of file diff --git a/libs/client/NetworkWriter.cs b/libs/client/NetworkWriter.cs index 9c5ac69e795..6862bc92d33 100644 --- a/libs/client/NetworkWriter.cs +++ b/libs/client/NetworkWriter.cs @@ -333,6 +333,9 @@ public void AsyncFlushPages(long fromAddress, long untilAddress) { logger?.LogError(ex, "Exception calling networkSender.SendResponse in AsyncFlushPages"); networkHandler.Dispose(); + // Still signal completion for this page so FlushEvent waiters (e.g. GarnetClient + // callers blocked in InternalExecuteNoResponse) are not left hanging indefinitely. + AsyncFlushPageCallback(asyncResult); } } if (flushPage == endPage) break; diff --git a/test/standalone/Garnet.test/GarnetClientTests.cs b/test/standalone/Garnet.test/GarnetClientTests.cs index 8f63ddb1ea3..62e13f493f8 100644 --- a/test/standalone/Garnet.test/GarnetClientTests.cs +++ b/test/standalone/Garnet.test/GarnetClientTests.cs @@ -4,10 +4,13 @@ using System.Linq; using System.Net; using System.Net.Sockets; +using System.Reflection; using System.Text; using System.Threading; using System.Threading.Tasks; +using Garnet.client; using Garnet.common; +using Garnet.networking; using NUnit.Framework; using NUnit.Framework.Legacy; @@ -568,5 +571,104 @@ public async Task MultipleSocketPing([Values] bool useTls) result = await db.ExecuteForStringResultAsync("PING").ConfigureAwait(false); ClassicAssert.AreEqual("PONG", result); } + + /// + /// Regression test for: NetworkWriter.AsyncFlushPages swallows send failures without + /// releasing the flush-completion event, hanging callers blocked on FlushEvent (e.g. + /// GarnetClient.InternalExecuteNoResponse, used by cluster PUBLISH forwarding). A + /// thread is parked in flushEvent.Wait (mirroring a caller already blocked when the + /// connection dies) before the failing flush is triggered; if AsyncFlushPageCallback + /// is not invoked when networkSender.SendResponse throws, that thread is never + /// released and only unblocks via our cancellation-token timeout (which the + /// assertions below turn into a test failure instead of a hang). + /// + [Test] + public void AsyncFlushPagesSignalsFlushEventOnSendFailure() + { + using var server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir); + server.Start(); + + using var db = new GarnetClient(TestUtils.EndPoint, sendPageSize: 1 << 12); + db.Connect(); + + var networkWriter = (NetworkWriter)typeof(GarnetClient) + .GetField("networkWriter", BindingFlags.NonPublic | BindingFlags.Instance) + .GetValue(db); + + networkWriter.epoch.Resume(); + try + { + // Reserve some space in the send buffer so AsyncFlushPages below has a real + // (non-empty) range to send, exercising the networkSender.SendResponse call + // rather than the empty-range fast path. + var (_, address) = networkWriter.TryAllocate(64, out var flushEvent); + ClassicAssert.GreaterOrEqual(address, 0); + var untilAddress = networkWriter.GetTailAddress(); + + // Simulate the connection being disposed/broken concurrently with the flush, + // as happens in production when a gossip timeout tears down the connection: + // make the send throw. + typeof(NetworkWriter) + .GetField("networkSender", BindingFlags.NonPublic | BindingFlags.Instance) + .SetValue(networkWriter, new ThrowingNetworkSender()); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + using var waiterStarted = new ManualResetEventSlim(); + Exception waiterException = null; + var waiterThread = new Thread(() => + { + waiterStarted.Set(); + try { flushEvent.Wait(cts.Token); } + catch (Exception ex) { waiterException = ex; } + }) + { IsBackground = true }; + waiterThread.Start(); + waiterStarted.Wait(); + // Give the waiter thread a brief moment to actually enter the semaphore wait + // before triggering the failing flush below, avoiding a race where the flush + // signals completion before the waiter has started waiting on it. + Thread.Sleep(50); + + networkWriter.AsyncFlushPages(0, untilAddress); + + ClassicAssert.IsTrue(waiterThread.Join(TimeSpan.FromSeconds(5)), + "Waiter thread did not complete - FlushEvent was not signaled after a failed flush (regression)"); + ClassicAssert.IsNull(waiterException, $"Waiter threw: {waiterException}"); + } + finally + { + networkWriter.epoch.Suspend(); + } + } + + /// + /// Network sender stub whose SendResponse always throws, simulating the underlying + /// connection being disposed/broken concurrently with a page flush. + /// + sealed unsafe class ThrowingNetworkSender : INetworkSender + { + readonly MaxSizeSettings maxSizeSettings = new(); + + public MaxSizeSettings GetMaxSizeSettings => maxSizeSettings; + public string RemoteEndpointName => ""; + public string LocalEndpointName => ""; + public bool IsLocalConnection() => true; + public void Dispose() { } + public void DisposeNetworkSender(bool waitForSendCompletion) { } + public void Enter() { } + public void EnterAndGetResponseObject(out byte* head, out byte* tail) => throw new NotSupportedException(); + public void Exit() { } + public void ExitAndReturnResponseObject() { } + public void GetResponseObject() { } + public byte* GetResponseObjectHead() => throw new NotSupportedException(); + public byte* GetResponseObjectTail() => throw new NotSupportedException(); + public void ReturnResponseObject() { } + public void SendCallback(object context) { } + public bool SendResponse(int offset, int size) => throw new NotSupportedException(); + public void SendResponse(byte[] buffer, int offset, int count, object context) + => throw new Exception("Simulated send failure (connection disposed)"); + public void Throttle() { } + public bool TryClose() => false; + } } } \ No newline at end of file From 5859996e13d1efc3ec36b39ce9a469ca7ef03a30 Mon Sep 17 00:00:00 2001 From: flink Date: Mon, 20 Jul 2026 05:36:41 -0400 Subject: [PATCH 2/3] Harden regression test synchronization against review-flagged flakiness Replace the fixed Thread.Sleep(50) rendezvous in AsyncFlushPagesSignalsFlushEventOnSendFailure with a deterministic readiness signal: the waiter thread flips a volatile flag immediately before calling flushEvent.Wait, and the main thread spins on that flag (bounded by SpinWait, no fixed delay) before triggering the failing flush. This removes the timing assumption an independent review flagged as a source of a low-frequency flake under load, in a test whose entire purpose is proving the fix doesn't hang. --- test/standalone/Garnet.test/GarnetClientTests.cs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/test/standalone/Garnet.test/GarnetClientTests.cs b/test/standalone/Garnet.test/GarnetClientTests.cs index 62e13f493f8..6a56ef7677c 100644 --- a/test/standalone/Garnet.test/GarnetClientTests.cs +++ b/test/standalone/Garnet.test/GarnetClientTests.cs @@ -613,21 +613,23 @@ public void AsyncFlushPagesSignalsFlushEventOnSendFailure() .SetValue(networkWriter, new ThrowingNetworkSender()); using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); - using var waiterStarted = new ManualResetEventSlim(); + // Flipped immediately before the waiter thread calls flushEvent.Wait, so the + // main thread has a deterministic (bounded-spin, not fixed-sleep) signal that + // the waiter is about to enter its wait rather than merely having started. + var aboutToWait = false; Exception waiterException = null; var waiterThread = new Thread(() => { - waiterStarted.Set(); + Volatile.Write(ref aboutToWait, true); try { flushEvent.Wait(cts.Token); } catch (Exception ex) { waiterException = ex; } }) { IsBackground = true }; waiterThread.Start(); - waiterStarted.Wait(); - // Give the waiter thread a brief moment to actually enter the semaphore wait - // before triggering the failing flush below, avoiding a race where the flush - // signals completion before the waiter has started waiting on it. - Thread.Sleep(50); + + var spinWait = new SpinWait(); + while (!Volatile.Read(ref aboutToWait)) + spinWait.SpinOnce(); networkWriter.AsyncFlushPages(0, untilAddress); From d52bbc48cbd404b43b7ee778981bd7a71de022ba Mon Sep 17 00:00:00 2001 From: hexonal Date: Tue, 21 Jul 2026 23:20:42 -0400 Subject: [PATCH 3/3] Remove residual race in AsyncFlushPages regression test synchronization Per Copilot review feedback: setting the aboutToWait flag before actually calling flushEvent.Wait() still left a narrow window where Set()/Dispose() on the underlying SemaphoreSlim could race the waiter thread's own call to Wait(), occasionally surfacing as an ObjectDisposedException instead of exercising the intended flow. Switch to flushEvent.WaitAsync(), which registers with the semaphore synchronously before returning. This closes the race entirely rather than narrowing it, and removes the need for a separate waiter thread and readiness rendezvous altogether: the test can just trigger the flush and then block on the returned Task with a timeout. --- .../Garnet.test/GarnetClientTests.cs | 30 +++++++------------ 1 file changed, 10 insertions(+), 20 deletions(-) diff --git a/test/standalone/Garnet.test/GarnetClientTests.cs b/test/standalone/Garnet.test/GarnetClientTests.cs index 6a56ef7677c..2a892152eee 100644 --- a/test/standalone/Garnet.test/GarnetClientTests.cs +++ b/test/standalone/Garnet.test/GarnetClientTests.cs @@ -613,29 +613,19 @@ public void AsyncFlushPagesSignalsFlushEventOnSendFailure() .SetValue(networkWriter, new ThrowingNetworkSender()); using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); - // Flipped immediately before the waiter thread calls flushEvent.Wait, so the - // main thread has a deterministic (bounded-spin, not fixed-sleep) signal that - // the waiter is about to enter its wait rather than merely having started. - var aboutToWait = false; - Exception waiterException = null; - var waiterThread = new Thread(() => - { - Volatile.Write(ref aboutToWait, true); - try { flushEvent.Wait(cts.Token); } - catch (Exception ex) { waiterException = ex; } - }) - { IsBackground = true }; - waiterThread.Start(); - - var spinWait = new SpinWait(); - while (!Volatile.Read(ref aboutToWait)) - spinWait.SpinOnce(); + // WaitAsync registers with the underlying SemaphoreSlim synchronously, before + // returning - unlike the previous Wait()-on-a-background-thread approach, there + // is no window between "registered as a waiter" and "flush triggered" for a + // racing Set()/Dispose() to be missed, so no separate thread or readiness + // rendezvous is needed at all. + var waitTask = flushEvent.WaitAsync(cts.Token); networkWriter.AsyncFlushPages(0, untilAddress); - ClassicAssert.IsTrue(waiterThread.Join(TimeSpan.FromSeconds(5)), - "Waiter thread did not complete - FlushEvent was not signaled after a failed flush (regression)"); - ClassicAssert.IsNull(waiterException, $"Waiter threw: {waiterException}"); + var completed = waitTask.Wait(TimeSpan.FromSeconds(5)); + ClassicAssert.IsTrue(completed, + "FlushEvent was not signaled after a failed flush (regression)"); + ClassicAssert.IsFalse(waitTask.IsFaulted, $"Wait faulted: {waitTask.Exception}"); } finally {