diff --git a/packages/fcl-web/src/websocket/fpwebsocket.pp b/packages/fcl-web/src/websocket/fpwebsocket.pp index cbf9db0ced0fee880dd89cbaa31fc50402dfbed8..9fc534ae494e394384a9d4f1e35a535f4d84fe1a 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,15 @@ type TWSSocketHelper = Class (TObject,IWSTransport) Private FSocket : TSocketStream; + FReadState : LongInt; + FInterruptRequested : 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 +224,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 +500,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 +515,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 +661,93 @@ 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; + FInterruptRequested:=0; {$if defined(FreeBSD) or defined(Linux)} FSocket.ReadFlags:=MSG_NOSIGNAL; FSocket.WriteFlags:=MSG_NOSIGNAL; {$endif} end; +procedure TWSSocketHelper.BeginRead; +begin + if InterlockedCompareExchange(FInterruptRequested,0,0)<>0 then + Raise EWSReadInterrupted.Create(SErrReadInterrupted); + + if InterlockedCompareExchange(FReadState,WSReadActive,WSReadIdle) <> + WSReadIdle then + Raise EWebSocket.Create(SErrConcurrentRead); + + { InterruptRead may have claimed the preceding read just as it completed. + In that race the sticky request prevents this next read from blocking. } + if InterlockedCompareExchange(FInterruptRequested,0,0)<>0 then + begin + InterlockedExchange(FReadState,WSReadIdle); + Raise EWSReadInterrupted.Create(SErrReadInterrupted); + end; +end; + +procedure TWSSocketHelper.EndRead; +begin + InterlockedExchange(FReadState,WSReadIdle); + if InterlockedCompareExchange(FInterruptRequested,0,0)<>0 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; +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. } + if InterlockedCompareExchange(FReadState,WSReadInterrupting, + WSReadActive)<>WSReadActive then + Exit; + + { Keep the request sticky until the reader unwinds. A cancelled fpRecv can + otherwise be treated as a short read and immediately retried by the frame + reader, which would block again. } + InterlockedExchange(FInterruptRequested,1); + + if not WakeSocketRead(FSocket.Handle) then + begin + { The read may not yet have reached the operating system. Let a later + termination pass retry once it is actually blocking. } + InterlockedCompareExchange(FReadState,WSReadActive, + WSReadInterrupting); + end; +end; + function TWSSocketHelper.CanRead(aTimeOut: Integer): Boolean; begin Result:=FSocket.CanRead(aTimeout); @@ -658,7 +793,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 +815,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 +829,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..120d5fa00a9d37e7bac2db2ccbcd6a0c4776c6ac 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,17 @@ Type 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; @@ -577,12 +580,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 +670,8 @@ end; constructor TWSMessagePump.Create(aOwner : TComponent); begin + inherited Create(aOwner); + FInterruptList:=TThreadList.Create; FList:=TThreadList.Create; FReads:=[]; FExceptions:=[]; @@ -662,35 +680,103 @@ 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; + DisconnectedClients: TList; + IncomingResult: TIncomingResult; I : Integer; begin + DisconnectedClients:=TList.Create; try - aList := List.LockList; try - FReads:=[]; - for I := 0 to aList.Count - 1 do + aList := List.LockList; + try + FReads:=[]; + { 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 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)