diff --git a/WebSockets/ReceiveAndControllThread.cs b/WebSockets/ReceiveAndControllThread.cs index 2c19107..e658057 100644 --- a/WebSockets/ReceiveAndControllThread.cs +++ b/WebSockets/ReceiveAndControllThread.cs @@ -26,98 +26,124 @@ public ReceiveAndControllThread(WebSocket webSocket) while (!_webSocket.Stopped) { - var messageFrame = _webSocket.WebSocketReceiver.StartReceivingMessage(); - - if (messageFrame == null) + try { - //Here we could let the thread sleep to safe resources + ProcessIncomingMessage(); } - else + catch (Exception ex) { - //handle error - if (messageFrame.Error) + // failure reading from the stream (connection closed, reset, disposed, etc.) + if (!_webSocket.Stopped) { _webSocket.HasError = true; - Debug.WriteLine($"{_webSocket.RemoteEndPoint} closed with error: {messageFrame.ErrorMessage}"); + Debug.WriteLine($"{_webSocket.RemoteEndPoint} closed with error: {ex.Message}"); - _webSocket.RawClose(messageFrame.CloseStatus, Encoding.UTF8.GetBytes(messageFrame.ErrorMessage), true); + // don't leak internal exception details to the peer + _webSocket.RawClose(WebSocketCloseStatus.EndpointUnavailable, Encoding.UTF8.GetBytes("Connection error"), true); + + // RawClose returns without closing if a close was already sent (or the socket is closing), + // and the timeout checker is disposed below, so make sure the connection is closed + _webSocket.HardClose(); } - else if (messageFrame.IsControllFrame) - { - byte[] buffer = _webSocket.WebSocketReceiver.ReadBuffer(messageFrame.MessageLength, messageFrame.Masks); - _webSocket.LastContactTimeStamp = DateTime.UtcNow; + // stream can't be used anymore + break; + } + } - switch (messageFrame.OpCode) - { - case OpCode.PingFrame: - // need to send a Pong - var pong = new SendMessageFrame() { Buffer = buffer, OpCode = OpCode.PongFrame}; - messageFrame.OpCode = OpCode.PongFrame; - messageFrame.IsMasked = false; - _webSocket.QueueMessageToSend(pong); - break; - - case OpCode.PongFrame: - // received a Pong - // checking if content Pong matches Ping is not implemented due to thread safety and memory consumption considerations - _webSocket.Pinging = false; - break; - - case OpCode.ConnectionCloseFrame: - _webSocket.CloseStatus = WebSocketCloseStatus.Empty; - - if (buffer.Length > 1) - { - byte[] closeByteCode = new byte[] { buffer[1], buffer[0] }; - UInt16 statusCode = BitConverter.ToUInt16(closeByteCode, 0); - if (statusCode > 999 && statusCode < 1012) - { - _webSocket.CloseStatus = (WebSocketCloseStatus)statusCode; - } - } + _webSocket.ReceiveStream.Close(); + timeoutCheckerTimer.Change(Timeout.Infinite, Timeout.Infinite); + timeoutCheckerTimer.Dispose(); + } - //connection asked to be closed return answer - if (_webSocket.State != WebSocketFrame.WebSocketState.CloseSent) - { - _webSocket.State = WebSocketFrame.WebSocketState.CloseReceived; + private void ProcessIncomingMessage() + { + var messageFrame = _webSocket.WebSocketReceiver.StartReceivingMessage(); - _webSocket.RawClose(WebSocketCloseStatus.NormalClosure, buffer, true); - } - //response to connection close we can shut down the socket. - else + if (messageFrame == null) + { + //Here we could let the thread sleep to safe resources + } + else + { + //handle error + if (messageFrame.Error) + { + _webSocket.HasError = true; + + Debug.WriteLine($"{_webSocket.RemoteEndPoint} closed with error: {messageFrame.ErrorMessage}"); + + _webSocket.RawClose(messageFrame.CloseStatus, Encoding.UTF8.GetBytes(messageFrame.ErrorMessage), true); + } + else if (messageFrame.IsControllFrame) + { + byte[] buffer = _webSocket.WebSocketReceiver.ReadBuffer(messageFrame.MessageLength, messageFrame.Masks); + + _webSocket.LastContactTimeStamp = DateTime.UtcNow; + + switch (messageFrame.OpCode) + { + case OpCode.PingFrame: + // need to send a Pong + var pong = new SendMessageFrame() { Buffer = buffer, OpCode = OpCode.PongFrame}; + messageFrame.OpCode = OpCode.PongFrame; + messageFrame.IsMasked = false; + _webSocket.QueueMessageToSend(pong); + break; + + case OpCode.PongFrame: + // received a Pong + // checking if content Pong matches Ping is not implemented due to thread safety and memory consumption considerations + _webSocket.Pinging = false; + break; + + case OpCode.ConnectionCloseFrame: + _webSocket.CloseStatus = WebSocketCloseStatus.Empty; + + if (buffer.Length > 1) + { + byte[] closeByteCode = new byte[] { buffer[1], buffer[0] }; + UInt16 statusCode = BitConverter.ToUInt16(closeByteCode, 0); + if (statusCode > 999 && statusCode < 1012) { - _webSocket.HardClose(); + _webSocket.CloseStatus = (WebSocketCloseStatus)statusCode; } - break; - } + } + + //connection asked to be closed return answer + if (_webSocket.State != WebSocketFrame.WebSocketState.CloseSent) + { + _webSocket.State = WebSocketFrame.WebSocketState.CloseReceived; + + _webSocket.RawClose(WebSocketCloseStatus.NormalClosure, buffer, true); + } + //response to connection close we can shut down the socket. + else + { + _webSocket.HardClose(); + } + break; } - else + } + else + { + if (messageFrame.Error) { - if (messageFrame.Error) - { - Debug.WriteLine($"Error message from '{_webSocket.RemoteEndPoint}' error - {messageFrame.ErrorMessage}"); + Debug.WriteLine($"Error message from '{_webSocket.RemoteEndPoint}' error - {messageFrame.ErrorMessage}"); - _webSocket.RawClose(messageFrame.CloseStatus, Encoding.UTF8.GetBytes(messageFrame.ErrorMessage), true); - } - else - { - messageFrame.Buffer = _webSocket.WebSocketReceiver.ReadBuffer(messageFrame.MessageLength, messageFrame.Masks); + _webSocket.RawClose(messageFrame.CloseStatus, Encoding.UTF8.GetBytes(messageFrame.ErrorMessage), true); + } + else + { + messageFrame.Buffer = _webSocket.WebSocketReceiver.ReadBuffer(messageFrame.MessageLength, messageFrame.Masks); - _webSocket.LastContactTimeStamp = DateTime.UtcNow; + _webSocket.LastContactTimeStamp = DateTime.UtcNow; - OnNewMessage(messageFrame); - } + OnNewMessage(messageFrame); } } - - } - - _webSocket.ReceiveStream.Close(); - timeoutCheckerTimer.Change(Timeout.Infinite, Timeout.Infinite); - timeoutCheckerTimer.Dispose(); } @@ -129,33 +155,42 @@ private void CheckTimeouts(object thread) receiveThread.Suspend(); #pragma warning restore S3889 // Neither "Thread.Resume" nor "Thread.Suspend" should be used - //Controlling ping and ControllerMessagesTimeout - if (_webSocket.Pinging - && _webSocket.PingTime.Add(_webSocket.ServerTimeout) < DateTime.UtcNow) + try { - _webSocket.RawClose(WebSocketCloseStatus.PolicyViolation, Encoding.UTF8.GetBytes("Ping timeout"), true); + //Controlling ping and ControllerMessagesTimeout + if (_webSocket.Pinging + && _webSocket.PingTime.Add(_webSocket.ServerTimeout) < DateTime.UtcNow) + { + _webSocket.RawClose(WebSocketCloseStatus.PolicyViolation, Encoding.UTF8.GetBytes("Ping timeout"), true); - Debug.WriteLine($"{_webSocket.RemoteEndPoint} ping timed out"); - } + Debug.WriteLine($"{_webSocket.RemoteEndPoint} ping timed out"); + } - if (_webSocket.State == WebSocketFrame.WebSocketState.CloseSent - && _webSocket.ClosingTime.Add(_webSocket.ServerTimeout) < DateTime.UtcNow) - { - _webSocket.HardClose(); - } + if (_webSocket.State == WebSocketFrame.WebSocketState.CloseSent + && _webSocket.ClosingTime.Add(_webSocket.ServerTimeout) < DateTime.UtcNow) + { + _webSocket.HardClose(); + } - if (_webSocket.KeepAliveInterval != Timeout.InfiniteTimeSpan - && _webSocket.State != WebSocketFrame.WebSocketState.CloseSent - && !_webSocket.Pinging - && _webSocket.LastContactTimeStamp.Add(_webSocket.KeepAliveInterval) < DateTime.UtcNow) + if (_webSocket.KeepAliveInterval != Timeout.InfiniteTimeSpan + && _webSocket.State != WebSocketFrame.WebSocketState.CloseSent + && !_webSocket.Pinging + && _webSocket.LastContactTimeStamp.Add(_webSocket.KeepAliveInterval) < DateTime.UtcNow) + { + _webSocket.SendPing(); + } + } + catch (Exception ex) { - _webSocket.SendPing(); + Debug.WriteLine($"{_webSocket.RemoteEndPoint} error checking timeouts: {ex.Message}"); } - + finally + { + // always resume the receive thread, otherwise it will be left suspended forever #pragma warning disable S3889 // OK to use in .NET nanoFramework context - receiveThread.Resume(); + receiveThread.Resume(); #pragma warning restore S3889 // Neither "Thread.Resume" nor "Thread.Suspend" should be used - + } } private void OnNewMessage(ReceiveMessageFrame message) diff --git a/WebSockets/WebSocket.cs b/WebSockets/WebSocket.cs index f6f1de0..ad5fc82 100644 --- a/WebSockets/WebSocket.cs +++ b/WebSockets/WebSocket.cs @@ -27,6 +27,7 @@ public abstract class WebSocket : IDisposable internal WebSocketSender _webSocketSender; private readonly object _syncLock = new object(); + private int _hardClosed = 0; internal MessageReceivedEventHandler CallbacksMessageReceivedEventHandler; /// @@ -133,6 +134,7 @@ protected void ConnectToStream(NetworkStream stream, bool isServer, Socket socke _socket = socket; RemoteEndPoint = (IPEndPoint)socket.RemoteEndPoint; LastContactTimeStamp = DateTime.UtcNow; + _hardClosed = 0; //start server sending and receiving async WebSocketReceiver = new WebSocketReceiver(stream, RemoteEndPoint, this, IsServer, MaxReceiveFrameSize, OnReadError); @@ -275,15 +277,26 @@ internal void RawClose(WebSocketCloseStatus closeStatus = WebSocketCloseStatus.E if (CloseImmediately) { + State = WebSocketState.CloseSent; + ClosingTime = DateTime.UtcNow; + + // Give it a moment for sending a close message. This will block the thread. + // Wait is bounded because the send thread can be stuck on a dead connection. + int maxWaitMs = (int)ServerTimeout.TotalMilliseconds; + + if (maxWaitMs <= 0) + { + // infinite (or invalid) server timeout, use a sensible default + maxWaitMs = 5000; + } + int msWaited = 0; - //Give it a moment for sending a close message. This will block the thread. - while (!_webSocketSender.CloseMessageSent ) + while (!_webSocketSender.CloseMessageSent + && msWaited < maxWaitMs) { msWaited += 50; Thread.Sleep(50); - State = WebSocketState.CloseSent; - ClosingTime = DateTime.UtcNow; } HardClose(); @@ -292,24 +305,36 @@ internal void RawClose(WebSocketCloseStatus closeStatus = WebSocketCloseStatus.E internal void HardClose() { + // can be called from several threads (receive, send error, timeout checker) + // lock-free guard: the timeout checker suspends the receive thread, so a lock here could deadlock + if (Interlocked.CompareExchange(ref _hardClosed, 1, 0) != 0) + { + return; + } + State = WebSocketState.Closed; StopReceiving(); _webSocketSender.StopSender(); Debug.WriteLine($"Connection - {RemoteEndPoint.ToString()} - Closed"); - - ConnectionClosed?.Invoke(this, new EventArgs()); - //Let the tcp socket linger for a second so it can try and send all data out before final close. try { - _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Linger, 1); - _socket.Close(); + ConnectionClosed?.Invoke(this, new EventArgs()); } - catch (ObjectDisposedException e) + finally { - Debug.WriteLine("socket could not be closed properly because it was already disposed"); + //Let the tcp socket linger for a second so it can try and send all data out before final close. + try + { + _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Linger, 1); + _socket.Close(); + } + catch (ObjectDisposedException e) + { + Debug.WriteLine("socket could not be closed properly because it was already disposed"); + } } } diff --git a/WebSockets/WebSocketReceiver.cs b/WebSockets/WebSocketReceiver.cs index 246274a..24bf610 100644 --- a/WebSockets/WebSocketReceiver.cs +++ b/WebSockets/WebSocketReceiver.cs @@ -199,6 +199,13 @@ byte[] ReadFixedSizeBuffer(int size, byte[] masks = null) while (size > 0) { int bytes = _inputStream.Read(buffer, offset, size); + + if (bytes <= 0) + { + // zero bytes read means the remote end has closed the connection + throw new SocketException(SocketError.ConnectionReset); + } + offset += bytes; size -= bytes; }