diff --git a/packages/fcl-web/src/websocket/fpwebsocket.pp b/packages/fcl-web/src/websocket/fpwebsocket.pp index cbf9db0ced0fee880dd89cbaa31fc50402dfbed8..3e79f204dfc4485ce8dc026529ca9042cea88574 100644 --- a/packages/fcl-web/src/websocket/fpwebsocket.pp +++ b/packages/fcl-web/src/websocket/fpwebsocket.pp @@ -83,6 +83,7 @@ Const type EWebSocket = Class(Exception); EWSHandShake = class(EWebSocket); + EWSReadInterrupted = class(EWebSocket); TFrameType = (ftContinuation,ftText,ftBinary,ftClose,ftPing,ftPong,ftFutureOpcodes); @@ -194,8 +195,14 @@ type TWSSocketHelper = Class (TObject,IWSTransport) Private FSocket : TSocketStream; + FReadState : LongInt; + Procedure BeginRead; + Procedure EndRead; + function ReadSocket(var aBuffer; aCount : LongInt) : LongInt; + Procedure ReadSocketBuffer(var aBuffer; aCount : LongInt); Public Constructor Create (aSocket : TSocketStream); + Procedure InterruptRead; Function CanRead(aTimeOut: Integer) : Boolean; function PeerIP: string; virtual; function PeerPort: word; virtual; @@ -216,6 +223,7 @@ type Constructor Create(aStream : TSocketStream); Destructor Destroy; override; Procedure CloseSocket; + Procedure InterruptRead; Property Helper : TWSSocketHelper Read FHelper Implements IWSTransport; Property Socket : TSocketStream Read GetSocket; end; @@ -491,6 +499,8 @@ Resourcestring SErrInvalidSizeFlag = 'Invalid size flag: %d'; SErrInvalidFrameType = 'Invalid frame type flag: %d'; SErrWriteReturnedError = 'Write operation returned error: (%d) %s'; + SErrReadInterrupted = 'WebSocket read interrupted'; + SErrConcurrentRead = 'Concurrent reads on one WebSocket transport are not supported'; function DecodeBytesBase64(const s: string; Strict: boolean = false) : TBytes; function EncodeBytesBase64(const aBytes : TBytes) : String; @@ -504,6 +514,54 @@ uses System.StrUtils, System.Hash.Sha1, System.Hash.Base64; uses strutils, sha1, base64; {$ENDIF FPC_DOTTEDUNITS} +Const + WSReadIdle = 0; + WSReadActive = 1; + WSReadInterrupting = 2; + +{$IFDEF MSWINDOWS} +Type + TCancelIoExProc = function(aHandle : PtrUInt; + aOverlapped : Pointer) : LongBool; stdcall; + +function WSGetModuleHandleA(aModuleName : PAnsiChar) : PtrUInt; stdcall; + external 'kernel32.dll' name 'GetModuleHandleA'; +function WSGetProcAddress(aModule : PtrUInt; + aProcName : PAnsiChar) : Pointer; stdcall; + external 'kernel32.dll' name 'GetProcAddress'; +{$ENDIF MSWINDOWS} + +function WakeSocketRead(aSocket : TSocket) : Boolean; +{$IFDEF MSWINDOWS} +Var + KernelModule : PtrUInt; + CancelIO : TCancelIoExProc; +begin + { Winsock shutdown disables later receives but does not release the receive + already blocked on another thread. Do it first so that, after CancelIoEx + releases the current call, neither OpenSSL nor the frame reader can block + by retrying the same socket. This is raw socket state only: the descriptor + stays open and no TLS object is touched or freed here. } + {$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.fpShutdown( + aSocket,{$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.SHUT_RDWR); + + { Resolve CancelIoEx dynamically so merely linking fpwebsocket does not add + a hard dependency on that entry point on older Windows versions. } + KernelModule:=WSGetModuleHandleA('kernel32.dll'); + if KernelModule=0 then + Exit(False); + CancelIO:=TCancelIoExProc(WSGetProcAddress(KernelModule,'CancelIoEx')); + if not Assigned(CancelIO) then + Exit(False); + Result:=CancelIO(PtrUInt(aSocket),Nil); +end; +{$ELSE MSWINDOWS} +begin + Result:={$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.fpShutdown( + aSocket,{$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.SHUT_RDWR)=0; +end; +{$ENDIF MSWINDOWS} + { TFrameTypeHelper } function TFrameTypeHelper.GetAsFlag: Byte; @@ -602,17 +660,82 @@ begin {$IFDEF FPC_DOTTEDUNITS}System.Net.{$ENDIF}sockets.CloseSocket(FStream.Handle); end; +procedure TWSTransport.InterruptRead; +begin + if Assigned(FHelper) then + FHelper.InterruptRead; +end; + { TWSTransport } constructor TWSSocketHelper.Create(aSocket: TSocketStream); begin FSocket:=aSocket; + FReadState:=WSReadIdle; {$if defined(FreeBSD) or defined(Linux)} FSocket.ReadFlags:=MSG_NOSIGNAL; FSocket.WriteFlags:=MSG_NOSIGNAL; {$endif} end; +procedure TWSSocketHelper.BeginRead; +begin + if InterlockedCompareExchange(FReadState,WSReadActive,WSReadIdle) <> + WSReadIdle then + Raise EWebSocket.Create(SErrConcurrentRead); +end; + +procedure TWSSocketHelper.EndRead; +Var + PreviousState : LongInt; +begin + { The state transition made by InterruptRead is itself the interruption + publication. This single atomic exchange closes the former window between + claiming a read and publishing a separate interruption flag. } + PreviousState:=InterlockedExchange(FReadState,WSReadIdle); + if PreviousState=WSReadInterrupting then + Raise EWSReadInterrupted.Create(SErrReadInterrupted); +end; + +function TWSSocketHelper.ReadSocket(var aBuffer; aCount: LongInt): LongInt; +begin + BeginRead; + try + Result:=FSocket.Read(aBuffer,aCount); + finally + EndRead; + end; +end; + +procedure TWSSocketHelper.ReadSocketBuffer(var aBuffer; aCount: LongInt); +begin + BeginRead; + try + FSocket.ReadBuffer(aBuffer,aCount); + finally + EndRead; + end; +end; + +procedure TWSSocketHelper.InterruptRead; +Var + PreviousState : LongInt; +begin + { Claim only a transport whose reader is currently inside a socket read. + A pump may serve several connections, so interrupting every registered + socket would unnecessarily break healthy siblings. } + PreviousState:=InterlockedCompareExchange(FReadState,WSReadInterrupting, + WSReadActive); + if (PreviousState<>WSReadActive) and + (PreviousState<>WSReadInterrupting) then + Exit; + + { Keep WSReadInterrupting set until EndRead observes it and raises. Repeated + termination passes may retry the platform wake, but the connection can no + longer return to the pump as healthy after its socket has been shut down. } + WakeSocketRead(FSocket.Handle); +end; + function TWSSocketHelper.CanRead(aTimeOut: Integer): Boolean; begin Result:=FSocket.CanRead(aTimeout); @@ -658,7 +781,7 @@ begin SetLength(Result,255); aSize:=0; C:=0; - While (FSocket.Read(C,1)=1) and (C<>10) do + While (ReadSocket(C,1)=1) and (C<>10) do begin Inc(aSize); if aSize>Length(Result) then @@ -680,7 +803,7 @@ begin SetLength(aBytes, aCount); repeat SetLength(buf{%H-}, aCount); - Result := FSocket.Read(buf[0], aCount - aPos); + Result := ReadSocket(buf[0], aCount - aPos); if Result <= 0 then break; SetLength(buf, Result); @@ -694,7 +817,7 @@ end; procedure TWSSocketHelper.ReadBuffer(aBytes: TBytes); begin if Length(ABytes)=0 then exit; - FSocket.ReadBuffer(aBytes[0],Length(ABytes)); + ReadSocketBuffer(aBytes[0],Length(ABytes)); end; function TWSSocketHelper.WriteBytes(aBytes: TBytes; aCount: Integer): Integer; diff --git a/packages/fcl-web/src/websocket/fpwebsocketclient.pp b/packages/fcl-web/src/websocket/fpwebsocketclient.pp index e3b36dc21904e1046c57cd09b1f797dfea79942c..5b4bd48df1780cf26351a7662c4f36c1325f767b 100644 --- a/packages/fcl-web/src/websocket/fpwebsocketclient.pp +++ b/packages/fcl-web/src/websocket/fpwebsocketclient.pp @@ -40,12 +40,14 @@ Type 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,19 @@ Type TWSThreadMessagePump = Class(TWSMessagePump) Private FThread : TThread; - Procedure ThreadTerminated(Sender : TObject); + Procedure PollDriverStop(aDriverThread : TThread; + aPollMs : Integer); 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; @@ -135,6 +140,8 @@ Type Protected Procedure CheckInactive; Procedure Loaded; override; + Procedure Notification(aComponent : TComponent; + Operation : TOperation); override; function CreateClientConnection(aTransport : TWSClientTransport): TWebSocketClientConnection; virtual; procedure MessageReceived(Sender: TObject; const aMessage : TWSMessage); Procedure ControlReceived(Sender: TObject; aType : TFrameType; const aData: TBytes);virtual; @@ -460,6 +467,14 @@ begin Connect; end; +procedure TCustomWebsocketClient.Notification(aComponent : TComponent; + Operation : TOperation); +begin + inherited Notification(aComponent,Operation); + if (Operation=opRemove) and (aComponent=FMessagePump) then + FMessagePump:=Nil; +end; + procedure TCustomWebsocketClient.MessageReceived(Sender: TObject; const aMessage : TWSMessage) ; begin if Assigned(OnMessageReceived) and (TWSClientConnection(Sender).HandshakeCompleted) then @@ -577,12 +592,25 @@ end; 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 +682,8 @@ end; constructor TWSMessagePump.Create(aOwner : TComponent); begin + inherited Create(aOwner); + FInterruptList:=TThreadList.Create; FList:=TThreadList.Create; FReads:=[]; FExceptions:=[]; @@ -662,32 +692,90 @@ end; 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; + DisconnectedClient: TWSClientConnection; + IncomingResult: TIncomingResult; I : Integer; begin + DisconnectedClient:=Nil; try aList := List.LockList; try FReads:=[]; - for I := 0 to aList.Count - 1 do + { Notify one removed connection before examining another. A callback may + destroy any remaining client, so retaining several raw connection + pointers across callbacks is unsafe. The next pump pass resumes with + the current registry contents. } + I:=0; + while I0 then + GraceMs:=QWord(Interval)*2+10 + else + GraceMs:=MinStopGraceMs; + if GraceMsMaxStopGraceMs then + GraceMs:=MaxStopGraceMs; + + StartMs:=TThread.GetTickCount64; + while (not DriverThread.Finished) and + ((TThread.GetTickCount64-StartMs)