Skip to content

Improve handling of connections closed by the peer #248

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
josesimoes merged 2 commits into from
Sep 24, 2026
Merged

Improve handling of connections closed by the peer #248

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
74057ad
Improve handling of connections closed by the peer
josesimoes Sep 24, 2026
bd2d9f3
Several fixes from code review
josesimoes Sep 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 124 additions & 89 deletions WebSockets/ReceiveAndControllThread.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,98 +26,124 @@

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()

Check failure on line 60 in WebSockets/ReceiveAndControllThread.cs

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 28 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=nanoframework_nanoframework.System.Net.WebSockets&issues=AaDTAvy6IHrwXwdORIiG&open=AaDTAvy6IHrwXwdORIiG&pullRequest=248
{
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();
}


Expand All @@ -129,33 +155,42 @@
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)
Expand Down
47 changes: 36 additions & 11 deletions WebSockets/WebSocket.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/// <summary>
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
Expand All @@ -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");
}
}
}

Expand Down
7 changes: 7 additions & 0 deletions WebSockets/WebSocketReceiver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Comment thread
coderabbitai[bot] marked this conversation as resolved.
offset += bytes;
size -= bytes;
}
Expand Down