Okay, here we go:I have debugged and made a patch based on todays trunk (attached) for a fpwebsocketclient recovery/shutdown problem on Linux with OpenSSL. After a network interruption, the client could remain active or “reconnecting” while receiving no further data until the application is restarted. During reconnect/shutdown I also saw a SIGSEGV with the reader thread inside SSL_read.
Cause: fpwebsocketclient did not always stop and join its reader thread
before the connection and OpenSSL objects are destroyed. It also failed to
act on irClose. The result could be either a stale connection that never
received data again after a network interruption, or a SIGSEGV when
transport cleanup raced with SSL_read.
The best solution was to fix this inside FPC: make the message pump own the
reader thread, wake a genuinely stuck socket when necessary, join the
reader, and only then destroy the connection and TLS objects. I have
modified my local files and done successful runtimes tests. The rest of the
fpc websocket code should be unaffected.
Proposed changes
• The pump should own a named, non-self-freeing thread using
FreeOnTerminate := False.
• Execute, Terminate, and destruction should be idempotent.
• Shutdown should follow this order:
1. Request thread termination.
2. Allow the bounded polling loop a short opportunity to exit
normally.
3. If it remains blocked, wake the socket with fpShutdown(...,
SHUT_RDWR).
4. Join and free the reader thread.
5. Only afterward close and free the connection, transport, and TLS
objects.
The wake operation must be raw socket shutdown only. Calling the TLS
handler’s Shutdown, SSL_shutdown, or freeing OpenSSL state while another
thread is inside SSL_read recreates the race.
ReadConnections should also process irClose: unregister the connection,
mark the client inactive, and notify its owner, preferably after releasing
the pump-list lock.
Because the reader currently holds the main connection-list lock while
reading, termination needs a separate cold-path transport registry or
another safe snapshot. This requires no additional per-message or hot-path
locking.
For compatibility, socket interruption should only be the fallback for a
reader that did not stop normally. Healthy connections should not be
unconditionally shut down merely because the pump was stopped.
More in depth, the problems appear to be:
TWSThreadMessagePump.Terminate can return without proving that the worker
has stopped. The timeout path clears the thread reference, allowing the
connection, transport, or TLS objects to be freed while the worker may
still be inside CheckIncoming/SSL_read.
ReadConnections ignores the irClose result, so a peer-closed connection can
remain registered and appear active.
WaitFor alone is not sufficient: select is bounded, but after the first
bytes of a frame arrive, reading the remainder of a partial WebSocket frame
or TLS record can block indefinitely.
Suggested changes i detail:
Make the pump own its thread: create it suspended, set FreeOnTerminate :=
False, publish the reference, then start it.
Make Execute, Terminate, and destruction idempotent.
During termination, first request termination and allow a short bounded
interval for the normal polling loop to exit.
If the reader is still running, wake it with raw fpShutdown(Socket.Handle,
SHUT_RDWR), then call WaitFor and free the thread.
Do not call TSocketHandler.Shutdown, SSL_shutdown, close the descriptor, or
free transport/TLS objects before the reader has joined. The
OpenSSL socket handler’s shutdown path destroys SSL state and is unsafe
concurrently with SSL_read.
Keep a separate cold-path registry or another safe snapshot of registered
transports, because ReadConnections currently holds the main list lock
while reading. The termination path must not wait for that same lock before
it can wake the blocked reader.
Handle irClose by safely unregistering the connection, marking the client
inactive, and notifying its owner—preferably after releasing the pump-list
lock.
Preserve normal client-processing order and existing protected signatures
where practical. Raw socket interruption should be the stuck-reader
fallback rather than unconditional behavior, so applications that
temporarily stop and restart a healthy pump retain compatibility.
Useful regression tests would cover partial-frame/TLS-record stalls,
repeated Execute/Terminate, destruction during a blocked read, peer close
and OnDisconnect, and multiple clients sharing one pump.
/Roger
<<attachment: fpc-main-2efddc1b-websocket-updated-full-files.zip>>
diff --git a/packages/fcl-web/src/websocket/fpwebsocket.pp b/packages/fcl-web/src/websocket/fpwebsocket.pp
index cbf9db0ced0fee880dd89cbaa31fc50402dfbed8..e656095a00d637941f1ffa6356313ff071b17d55 100644
--- a/packages/fcl-web/src/websocket/fpwebsocket.pp
+++ b/packages/fcl-web/src/websocket/fpwebsocket.pp
@@ -216,6 +216,7 @@ TWSTransport = class(TObject, IWSTransport)
Constructor Create(aStream : TSocketStream);
Destructor Destroy; override;
Procedure CloseSocket;
+ Procedure InterruptRead;
Property Helper : TWSSocketHelper Read FHelper Implements IWSTransport;
Property Socket : TSocketStream Read GetSocket;
end;
@@ -602,6 +603,21 @@ procedure TWSTransport.CloseSocket;
{$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.CloseSocket(FStream.Handle);
end;
+procedure TWSTransport.InterruptRead;
+begin
+ if not Assigned(FStream) then
+ Exit;
+ if FStream.Closed then
+ Exit;
+
+ { Wake a thread blocked below SSL_read/recv without closing the descriptor
+ or touching the TLS handler. Connection and TLS object destruction must
+ remain on the owning thread after the reader has been joined. }
+ {$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.fpShutdown(
+ FStream.Handle,
+ {$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.SHUT_RDWR);
+end;
+
{ TWSTransport }
constructor TWSSocketHelper.Create(aSocket: TSocketStream);
diff --git a/packages/fcl-web/src/websocket/fpwebsocketclient.pp b/packages/fcl-web/src/websocket/fpwebsocketclient.pp
index e3b36dc21904e1046c57cd09b1f797dfea79942c..5e9a8031bf9c20179ac17908802b911ce6836dee 100644
--- a/packages/fcl-web/src/websocket/fpwebsocketclient.pp
+++ b/packages/fcl-web/src/websocket/fpwebsocketclient.pp
@@ -40,12 +40,14 @@ interface
TWSMessagePump = Class (TComponent)
private
FInterval:Integer;
+ FInterruptList: TThreadList;
FList: TThreadList;
FReads: TSocketStreamArray;
FExceptions : TSocketStreamArray;
FOnError: TWSErrorEvent;
procedure SetInterval(AValue: Integer);
Protected
+ Procedure InterruptConnections;
function WaitForData: Boolean;
Function CheckConnections : Boolean; virtual;
Procedure ReadConnections;
@@ -66,16 +68,17 @@ interface
TWSThreadMessagePump = Class(TWSMessagePump)
Private
FThread : TThread;
- Procedure ThreadTerminated(Sender : TObject);
Protected
Type
TMessageDriverThread = Class(TThread)
Public
FPump : TWSThreadMessagePump;
- Constructor Create(aPump : TWSThreadMessagePump; aTerminate : TNotifyEvent);
+ Constructor Create(aPump : TWSThreadMessagePump;
+ aTerminate : TNotifyEvent);
Procedure Execute;override;
End;
Public
+ Destructor Destroy; override;
Procedure Execute; override;
Procedure Terminate; override;
End;
@@ -132,6 +135,8 @@ TWebSocketClientConnection = class(TWSClientConnection)
procedure SetAutoCheckMessages(const Value: Boolean);
procedure SendHeaders(aHeaders: TStrings);
procedure ConnectionDisconnected(Sender: TObject);
+ procedure MessagePumpDisconnected(
+ AConnection: TWebSocketClientConnection);
Protected
Procedure CheckInactive;
Procedure Loaded; override;
@@ -286,6 +291,20 @@ procedure TCustomWebsocketClient.ConnectionDisconnected(Sender : TObject);
// We cannot free the connection here, because it still needs to call it's own OnDisconnect.
end;
+procedure TCustomWebsocketClient.MessagePumpDisconnected(
+ AConnection: TWebSocketClientConnection);
+begin
+ if FConnection<>AConnection then
+ Exit;
+
+ { ReadConnections already removed this connection from the pump's locked
+ list. Keep the connection object alive until the owner reconnects or is
+ destroyed, matching ConnectionDisconnected's lifetime rule. }
+ FActive:=False;
+ if Assigned(OnDisconnect) then
+ OnDisconnect(AConnection);
+end;
+
procedure TCustomWebsocketClient.Connect;
var
SSLHandler: TSSLSocketHandler;
@@ -577,12 +596,25 @@ procedure TCustomWebsocketClient.SetUseSSL(const Value: Boolean);
procedure TWSMessagePump.AddClient(aConnection: TWSClientConnection);
begin
- List.Add(aConnection);
+ { Keep interruption registration independent from FList. ReadConnections
+ holds FList while it reads a complete frame, so termination must not need
+ that same lock to wake a blocked transport operation. }
+ FInterruptList.Add(aConnection);
+ try
+ List.Add(aConnection);
+ except
+ FInterruptList.Remove(aConnection);
+ raise;
+ end;
end;
procedure TWSMessagePump.RemoveClient(aConnection: TWSClientConnection);
begin
+ { Remove from the reader list first. When this returns the reader can no
+ longer start using the connection. Removal from FInterruptList then waits
+ for any in-progress termination wake before the caller may free it. }
FList.Remove(aConnection);
+ FInterruptList.Remove(aConnection);
end;
procedure TWSMessagePump.SetInterval(AValue: Integer);
@@ -654,6 +686,8 @@ function TWSMessagePump.CheckConnections: Boolean;
constructor TWSMessagePump.Create(aOwner : TComponent);
begin
+ inherited Create(aOwner);
+ FInterruptList:=TThreadList.Create;
FList:=TThreadList.Create;
FReads:=[];
FExceptions:=[];
@@ -662,15 +696,39 @@ constructor TWSMessagePump.Create(aOwner : TComponent);
destructor TWSMessagePump.Destroy;
begin
+ FreeAndNil(FInterruptList);
FreeAndNil(FList);
inherited;
end;
+procedure TWSMessagePump.InterruptConnections;
+Var
+ aList : TList;
+ aClient: TWSClientConnection;
+ I : Integer;
+
+begin
+ aList:=FInterruptList.LockList;
+ try
+ for I:=0 to aList.Count-1 do
+ begin
+ aClient:=TWSClientConnection(aList.Items[I]);
+ if Assigned(aClient) then
+ if Assigned(aClient.ClientTransport) then
+ aClient.ClientTransport.InterruptRead;
+ end;
+ finally
+ FInterruptList.UnlockList;
+ end;
+end;
+
procedure TWSMessagePump.ReadConnections;
Var
aList : TList;
aClient: TWSClientConnection;
+ aWebSocketClient: TWebSocketClientConnection;
+ IncomingResult: TIncomingResult;
I : Integer;
begin
@@ -678,11 +736,36 @@ procedure TWSMessagePump.ReadConnections;
aList := List.LockList;
try
FReads:=[];
- for I := 0 to aList.Count - 1 do
+ { Keep the existing client-processing order. Deleting a closed client at
+ the current index makes the next client take that same index. }
+ I:=0;
+ while I<aList.Count do
begin
aClient:= TWSClientConnection(aList.Items[I]);
- if assigned(aClient.Transport) then
- aClient.CheckIncoming(1);
+ if assigned(aClient) then
+ if assigned(aClient.Transport) then
+ begin
+ IncomingResult:=aClient.CheckIncoming(1);
+ if IncomingResult=irClose then
+ begin
+ { The connection-level CheckIncoming contract returns irClose to
+ request owner disconnection. Remove it while the list is locked,
+ then publish the disconnect without freeing the connection on
+ its own read stack. }
+ aList.Delete(I);
+ FInterruptList.Remove(aClient);
+ if aClient is TWebSocketClientConnection then
+ begin
+ aWebSocketClient:=TWebSocketClientConnection(aClient);
+ aWebSocketClient.WebsocketClient.MessagePumpDisconnected(
+ aWebSocketClient);
+ end
+ else
+ aClient.Disconnect;
+ Continue;
+ end;
+ end;
+ Inc(I);
end;
finally
List.UnlockList;
@@ -699,51 +782,78 @@ procedure TWSMessagePump.ReadConnections;
procedure TWSThreadMessagePump.Execute;
begin
- FThread:=TMessageDriverThread.Create(Self,@ThreadTerminated);
+ if Assigned(FThread) then
+ Exit;
+
+ FThread:=TMessageDriverThread.Create(Self,Nil);
+ try
+ FThread.Start;
+ except
+ FreeAndNil(FThread);
+ raise;
+ end;
end;
-procedure TWSThreadMessagePump.ThreadTerminated(Sender: TObject);
+destructor TWSThreadMessagePump.Destroy;
begin
- FThread:=Nil;
+ Terminate;
+ inherited Destroy;
end;
procedure TWSThreadMessagePump.Terminate;
-var
- lThread: TThread;
- lCounter: Integer;
+Const
+ MinStopGraceMs = 100;
+ MaxStopGraceMs = 1000;
+ StopPollMs = 1;
+Var
+ GraceMs : QWord;
+ StartMs : QWord;
begin
- lThread := FThread;
- if Assigned(lThread) then
- begin
- lThread.Terminate;
+ if not Assigned(FThread) then
+ Exit;
- // Wait till it stops
- lCounter := 0;
- while Assigned(FThread) and (lCounter < 200) do // 5 second timeout
- begin
- Sleep(10);
- Inc(lCounter);
- end;
+ FThread.Terminate;
- // If thread still hasn't finished, there's a serious problem
- if Assigned(FThread) then
+ { Preserve the previous ability to stop and restart a healthy pump without
+ disconnecting its clients. Normally the bounded polling loop exits within
+ two intervals. Interrupt sockets only when that graceful stop fails. }
+ if Interval>0 then
+ GraceMs:=QWord(Interval)*2+10
+ else
+ GraceMs:=MinStopGraceMs;
+ if GraceMs<MinStopGraceMs then
+ GraceMs:=MinStopGraceMs
+ else if GraceMs>MaxStopGraceMs then
+ GraceMs:=MaxStopGraceMs;
+
+ StartMs:=TThread.GetTickCount64;
+ while (not FThread.Finished) and
+ ((TThread.GetTickCount64-StartMs)<GraceMs) do
+ TThread.Sleep(StopPollMs);
+
+ if not FThread.Finished then
begin
- FThread.OnTerminate:=Nil;
- // Force cleanup as last resort
- FThread := nil;
+ { Select is bounded, but reading the remainder of a partial frame can
+ block below SSL_read/recv. Shut down only the socket directions to wake
+ that operation; do not close the descriptor or destroy TLS state until
+ the reader has left and WaitFor has completed. }
+ InterruptConnections;
end;
- end;
+
+ FThread.WaitFor;
+ FreeAndNil(FThread);
end;
{ TWSThreadMessagePump.TMessageDriverThread }
-constructor TWSThreadMessagePump.TMessageDriverThread.Create(aPump: TWSThreadMessagePump; aTerminate : TNotifyEvent);
+constructor TWSThreadMessagePump.TMessageDriverThread.Create(
+ aPump: TWSThreadMessagePump; aTerminate: TNotifyEvent);
begin
FPump:=aPump;
OnTerminate:=aTerminate;
- FreeOnTerminate:=True;
- Inherited Create(False);
+ Inherited Create(True);
+ FreeOnTerminate:=False;
end;
procedure TWSThreadMessagePump.TMessageDriverThread.Execute;
@@ -756,13 +866,6 @@ procedure TWSThreadMessagePump.TMessageDriverThread.Execute;
begin
TThread.Sleep(FPump.Interval);
end;
- // OnTerminate is called in a synchronize. However, if no-one calls CheckSynchronize, it is never called.
- // So we call it ourselves.
- if assigned(OnTerminate) then
- begin
- OnTerminate(Self);
- OnTerminate:=Nil;
- end;
end;
end.
_______________________________________________ fpc-pascal maillist - [email protected] https://lists.freepascal.org/cgi-bin/mailman/listinfo/fpc-pascal
