diff --git a/AUTHORS.md b/AUTHORS.md index ce4fe13e..433b7cf6 100644 --- a/AUTHORS.md +++ b/AUTHORS.md @@ -12,6 +12,7 @@ Contributors * Ilya Loginov — [@iloginov](https://github.com/iloginov) * James Hurst — [@JamesWhurst](https://github.com/JamesWhurst) * Jay Shelton — [@jayshelton](https://github.com/jayshelton) +* Jonas Follesø — [@follesoe](https://github.com/follesoe) * Miguel Angel Jimenez — [@majimenezp](https://github.com/majimenezp) * Nicolas Dextraze — [@ndextraze-pbp](https://github.com/ndextraze-pbp) * Peter H. Merkel — [@mph911](https://github.com/mph911) diff --git a/src/NetMQ.Tests/TcpListenerTests.cs b/src/NetMQ.Tests/TcpListenerTests.cs new file mode 100644 index 00000000..a040223a --- /dev/null +++ b/src/NetMQ.Tests/TcpListenerTests.cs @@ -0,0 +1,61 @@ +using System; +using System.Net; +using System.Net.Sockets; +using NetMQ.Sockets; +using Xunit; + +namespace NetMQ.Tests +{ + public class TcpListenerTests : IClassFixture + { + public TcpListenerTests() => NetMQConfig.Cleanup(); + + [Fact] + public void AcceptedSocketFailingSetupDoesNotCrashProcessOrStopListener() + { + using (var pub = new PublisherSocket()) + { + int port = pub.BindRandomPort("tcp://127.0.0.1"); + + // Hammer the listener with connections that are reset immediately: + // SO_LINGER=0 + Close sends RST, which can land between the proactor + // completing an accept and TcpListener applying socket options to the + // accepted socket. On macOS the option calls then throw + // SocketException (EINVAL). Unhandled, that exception killed the + // proactor thread and with it the process; swallowed carelessly, it + // skips the re-arming Accept() and the listener goes deaf instead. + for (int i = 0; i < 400; i++) + { + using (var raw = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)) + { + raw.LingerState = new LingerOption(true, 0); + try + { + raw.Connect(IPAddress.Loopback, port); + } + catch (SocketException) + { + } + } + } + + // The listener must still be alive and accepting: a subscriber that + // connects after the storm must complete the handshake and receive. + using (var sub = new SubscriberSocket()) + { + sub.Connect("tcp://127.0.0.1:" + port); + sub.SubscribeToAnyTopic(); + + var received = false; + for (int i = 0; i < 100 && !received; i++) + { + pub.SendFrame("hello"); + received = sub.TryReceiveFrameString(TimeSpan.FromMilliseconds(100), out string? _); + } + + Assert.True(received, "Listener stopped accepting connections after a flood of immediately-reset connects"); + } + } + } + } +} diff --git a/src/NetMQ/Core/Transports/Tcp/TcpListener.cs b/src/NetMQ/Core/Transports/Tcp/TcpListener.cs index 99a37288..79e8cd9c 100644 --- a/src/NetMQ/Core/Transports/Tcp/TcpListener.cs +++ b/src/NetMQ/Core/Transports/Tcp/TcpListener.cs @@ -199,11 +199,14 @@ public void InCompleted(SocketError socketError, int bytesTransferred) // TODO: check TcpFilters var acceptedSocket = m_handle.GetAcceptedSocket(); + StreamEngine engine; + try + { acceptedSocket.NoDelay = true; - if (m_options.TcpKeepalive != -1) - { - acceptedSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, m_options.TcpKeepalive); + if (m_options.TcpKeepalive != -1) + { + acceptedSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, m_options.TcpKeepalive); #if NET if (m_options.TcpKeepaliveIdle != -1) acceptedSocket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, m_options.TcpKeepaliveIdle / 1000); @@ -213,24 +216,54 @@ public void InCompleted(SocketError socketError, int bytesTransferred) acceptedSocket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, m_options.TcpKeepaliveCnt); #else - if (m_options.TcpKeepaliveIdle != -1 && m_options.TcpKeepaliveIntvl != -1) - { - var bytes = new ByteArraySegment(new byte[12]); + if (m_options.TcpKeepaliveIdle != -1 && m_options.TcpKeepaliveIntvl != -1) + { + var bytes = new ByteArraySegment(new byte[12]); - Endianness endian = BitConverter.IsLittleEndian ? Endianness.Little : Endianness.Big; + Endianness endian = BitConverter.IsLittleEndian ? Endianness.Little : Endianness.Big; - bytes.PutInteger(endian, m_options.TcpKeepalive, 0); - bytes.PutInteger(endian, m_options.TcpKeepaliveIdle, 4); - bytes.PutInteger(endian, m_options.TcpKeepaliveIntvl, 8); + bytes.PutInteger(endian, m_options.TcpKeepalive, 0); + bytes.PutInteger(endian, m_options.TcpKeepaliveIdle, 4); + bytes.PutInteger(endian, m_options.TcpKeepaliveIntvl, 8); - acceptedSocket.IOControl(IOControlCode.KeepAliveValues, (byte[])bytes, null); - } + acceptedSocket.IOControl(IOControlCode.KeepAliveValues, (byte[])bytes, null); + } #endif + } + + // Create the engine object for this connection. The engine + // constructor also touches the socket (send/receive buffer + // sizes), so it shares the failure window below. + engine = new StreamEngine(acceptedSocket, m_options, m_endpoint); } + catch (SocketException ex) + { + // The peer can reset the connection between the accept + // completing and the socket options above being applied; the + // accepted socket is then already dead and the option calls + // throw (EINVAL/InvalidArgument on macOS and Linux). Without + // this catch the exception escapes on the proactor thread and + // terminates the process. Treat it as a failed accept, as + // libzmq does: drop the socket and keep listening. The raw + // SocketError is deliberately not passed through ToErrorCode, + // which Debug.Asserts on unmapped values; ConnectionReset is + // the effective outcome whatever the exact reported code. + Debug.WriteLine($"TcpListener dropping accepted socket that failed setup: {ex.SocketErrorCode}"); + + try + { + acceptedSocket.Dispose(); + } + catch (SocketException) + { + } + + m_socket.EventAcceptFailed(m_endpoint, ErrorCode.ConnectionReset); - // Create the engine object for this connection. - var engine = new StreamEngine(acceptedSocket, m_options, m_endpoint); + Accept(); + break; + } // Choose I/O thread to run connector in. Given that we are already // running in an I/O thread, there must be at least one available.