From cd172bb7aaaf0548cfe5d4fba05572bea0fa903e Mon Sep 17 00:00:00 2001 From: Vincent Vanbesien Date: Sat, 2 May 2026 01:20:49 +0200 Subject: [PATCH] Fix infinite blocking in TrySendMultipartMessage at low bandwidth Apply timeout to all frames in TrySendMultipartMessage, not just the first one. Previously, only the first frame used TrySendFrame with timeout, while subsequent frames used blocking SendMoreFrame/SendFrame methods. This caused infinite hangs when network bandwidth was extremely low (< 10 kbit/s). The fix ensures all frames respect the timeout parameter, preventing infinite blocking and allowing graceful failure when send operations cannot complete. Added three unit tests to validate the fix: - TrySendMultipartMessage_AppliesTimeoutToAllFrames - TrySendMultipartMessage_FailsGracefullyOnMiddleFrameTimeout - TrySendMultipartMessage_FailsGracefullyOnLastFrameTimeout Fixes #1127 --- .../OutgoingSocketExtensionsTests.cs | 105 ++++++++++++++++++ src/NetMQ/OutgoingSocketExtensions.cs | 20 +--- 2 files changed, 111 insertions(+), 14 deletions(-) diff --git a/src/NetMQ.Tests/OutgoingSocketExtensionsTests.cs b/src/NetMQ.Tests/OutgoingSocketExtensionsTests.cs index 96cc7b1c..cb1614a2 100644 --- a/src/NetMQ.Tests/OutgoingSocketExtensionsTests.cs +++ b/src/NetMQ.Tests/OutgoingSocketExtensionsTests.cs @@ -345,5 +345,110 @@ public void TrySendFrameBiggerBufferThanLength() Assert.True(socket.TrySendFrame(TimeSpan.Zero, buffer, data.Length)); } + + /// + /// Test that TrySendMultipartMessage applies timeout to ALL frames, not just the first. + /// This validates the fix for issue #1127 where low bandwidth caused infinite blocking + /// because only the first frame respected the timeout parameter. + /// + [Fact] + public void TrySendMultipartMessage_AppliesTimeoutToAllFrames() + { + var timeout = TimeSpan.FromMilliseconds(500); + var frameCount = 0; + + var socket = new MockOutgoingSocket((ref Msg msg, TimeSpan actualTimeout, bool more) => + { + // CRITICAL: ALL frames must respect the timeout parameter + Assert.Equal(timeout, actualTimeout); + + frameCount++; + + // Verify 'more' flag is correct for each frame + if (frameCount < 3) + Assert.True(more, $"Frame {frameCount} should have 'more' flag set"); + else + Assert.False(more, "Last frame should not have 'more' flag"); + + return true; + }); + + var message = new NetMQMessage(); + message.Append(new byte[] { 1 }); // Frame 1: Identity + message.Append(new byte[] { }); // Frame 2: Empty frame + message.Append(new byte[] { 2 }); // Frame 3: Payload + + bool result = socket.TrySendMultipartMessage(timeout, message); + + Assert.True(result, "Send should succeed when all frames are sent"); + Assert.Equal(3, frameCount); + } + + /// + /// Test that TrySendMultipartMessage fails gracefully when a middle frame times out. + /// This validates that the timeout is respected on ALL frames, preventing infinite blocking. + /// + [Fact] + public void TrySendMultipartMessage_FailsGracefullyOnMiddleFrameTimeout() + { + var timeout = TimeSpan.FromMilliseconds(500); + var frameCount = 0; + + var socket = new MockOutgoingSocket((ref Msg msg, TimeSpan actualTimeout, bool more) => + { + // Verify timeout is applied to all frames + Assert.Equal(timeout, actualTimeout); + + frameCount++; + + // Simulate timeout on the second frame + if (frameCount == 2) + return false; // Timeout! + + return true; + }); + + var message = new NetMQMessage(); + message.Append(new byte[] { 1 }); + message.Append(new byte[] { }); + message.Append(new byte[] { 2 }); + + bool result = socket.TrySendMultipartMessage(timeout, message); + + Assert.False(result, "Send should fail when a middle frame times out"); + Assert.Equal(2, frameCount); // Should stop after the failed frame + } + + /// + /// Test that TrySendMultipartMessage fails gracefully when the last frame times out. + /// + [Fact] + public void TrySendMultipartMessage_FailsGracefullyOnLastFrameTimeout() + { + var timeout = TimeSpan.FromMilliseconds(500); + var frameCount = 0; + + var socket = new MockOutgoingSocket((ref Msg msg, TimeSpan actualTimeout, bool more) => + { + Assert.Equal(timeout, actualTimeout); + frameCount++; + + // Simulate timeout on the last frame + if (frameCount == 3) + return false; + + return true; + }); + + var message = new NetMQMessage(); + message.Append(new byte[] { 1 }); + message.Append(new byte[] { }); + message.Append(new byte[] { 2 }); + + bool result = socket.TrySendMultipartMessage(timeout, message); + + Assert.False(result, "Send should fail when the last frame times out"); + Assert.Equal(3, frameCount); + } } } diff --git a/src/NetMQ/OutgoingSocketExtensions.cs b/src/NetMQ/OutgoingSocketExtensions.cs index d2583d42..78f668e6 100644 --- a/src/NetMQ/OutgoingSocketExtensions.cs +++ b/src/NetMQ/OutgoingSocketExtensions.cs @@ -456,25 +456,17 @@ public static bool TrySendMultipartMessage(this IOutgoingSocket socket, TimeSpan { if (message.FrameCount == 0) throw new ArgumentException("message is empty", nameof(message)); - else if (message.FrameCount == 1) - { - return TrySendFrame(socket, timeout, message[0].Buffer, message[0].MessageSize); - } - else + + // Send all frames except the last with timeout and 'more' flag + for (int i = 0; i < message.FrameCount - 1; i++) { - bool sentSuccessfully = TrySendFrame(socket, timeout, message[0].Buffer, message[0].MessageSize, true); + bool sentSuccessfully = TrySendFrame(socket, timeout, message[i].Buffer, message[i].MessageSize, true); if (!sentSuccessfully) return false; } - for (int i = 1; i < message.FrameCount - 1; i++) - { - socket.SendMoreFrame(message[i].Buffer, message[i].MessageSize); - } - - socket.SendFrame(message.Last.Buffer, message.Last.MessageSize); - - return true; + // Send the last frame with timeout (no 'more' flag) + return TrySendFrame(socket, timeout, message.Last.Buffer, message.Last.MessageSize, false); } #endregion