diff --git a/src/Horse.Provider.Socket.WebSocket.pas b/src/Horse.Provider.Socket.WebSocket.pas index eb3b31c..309f126 100644 --- a/src/Horse.Provider.Socket.WebSocket.pas +++ b/src/Horse.Provider.Socket.WebSocket.pas @@ -9,12 +9,23 @@ interface uses SysUtils, Classes, {$IF DEFINED(FPC)} + SyncObjs, Sockets, + {$IFDEF MSWINDOWS} + { FIX-WS-NONBLOCK WSAGetLastError / select / TFDSet } + WinSock2, + {$ELSE} + { FIX-WS-NONBLOCK fpgeterrno, ESysEAGAIN, fpSelect, fpFD_* , TTimeVal } + BaseUnix, + {$ENDIF} {$ELSE} + System.SyncObjs, {$IFDEF MSWINDOWS} Winapi.WinSock2, {$ELSE} Posix.SysSocket, Posix.Unistd, + { FIX-WS-NONBLOCK errno / EAGAIN / select / fd_set / timeval } + Posix.Errno, Posix.SysSelect, Posix.SysTime, {$ENDIF} {$ENDIF} Horse.Core.WebSocket; @@ -37,10 +48,24 @@ THorseWebSocketSocketTransport = class(TInterfacedObject, IHorseWebSocketTrans private FSocket: TSocket; FIsClosed: Boolean; + FSocketClosed: Boolean; + FActiveOperations: Integer; + FStateLock: TCriticalSection; + FOperationsFinished: TEvent; FClientIP: string; FServerPort: Integer; + { FIX-WS-NONBLOCK } + function BeginOperation(out ASocket: TSocket): Boolean; + procedure EndOperation; + function GetLastSocketError: Integer; + function IsInterrupted(const AError: Integer): Boolean; + function WouldBlock(const AError: Integer): Boolean; + function WaitReadable(const ASocket: TSocket; const ATimeoutMS: Integer): Boolean; + function Closing: Boolean; + procedure RequestClose(const AWaitForOperations: Boolean); public constructor Create(ASocket: TSocket; const AClientIP: string = ''; const AServerPort: Integer = 0); + destructor Destroy; override; function Read(var ABuffer: TBytes; const ACount: Integer): Integer; function Write(const ABuffer: TBytes; const ACount: Integer): Integer; procedure Close; @@ -61,6 +86,11 @@ THorseWebSocketSocketUpgrader = class(THorseWebSocketUpgrader) function Upgrade(const APath: string; const AHeartbeatInterval: Integer = 0): IHorseWebSocketConnection; override; end; +const + { FIX-WS-NONBLOCK one select() tick. A timeout is not a disconnect, so Read + simply retries; this only bounds how often the loop re-checks FIsClosed. } + WS_SOCKET_READ_TICK_MS = 250; + implementation { THorseWebSocketSocketTransport } @@ -70,85 +100,366 @@ constructor THorseWebSocketSocketTransport.Create(ASocket: TSocket; const AClien inherited Create; FSocket := ASocket; FIsClosed := False; + FSocketClosed := False; + FActiveOperations := 0; + FStateLock := TCriticalSection.Create; + FOperationsFinished := TEvent.Create(nil, True, True, ''); FClientIP := AClientIP; FServerPort := AServerPort; end; -function THorseWebSocketSocketTransport.Read(var ABuffer: TBytes; const ACount: Integer): Integer; +destructor THorseWebSocketSocketTransport.Destroy; begin - Result := 0; - if FIsClosed then - Exit; + Close; + FOperationsFinished.Free; + FStateLock.Free; + inherited; +end; + +{ FIX-WS-NONBLOCK distinguish "no data yet" from "peer closed". + + IOCP and epoll both put accepted client sockets into non-blocking mode -- + epoll requires it (Horse.Provider.Epoll.pas sets O_NONBLOCK on every accepted + fd). On a non-blocking socket, recv returns -1 with EAGAIN/EWOULDBLOCK to mean + "nothing available right now", which is the normal state of an idle WebSocket + peer, not a disconnect. + + Treating every non-positive result as closed made the upgrader's read loop + break on its very first iteration: the connection was marked dead about a + millisecond after the 101, before any frame could arrive. The handler then + returned and the HTTP pipeline resumed, writing a stray "HTTP/1.1 200 OK" + onto a socket that had already been handed to WebSocket. + + Only two results actually mean the connection is over: + recv = 0 orderly shutdown by the peer + recv < 0 with an errno that is NOT EAGAIN/EWOULDBLOCK/EINTR + + Anything else means wait. WaitReadable blocks in select() so an idle peer + costs no CPU, and the loop is bounded only by the socket's own lifetime -- + which is correct: a WebSocket peer may legitimately stay silent for minutes. } + +function THorseWebSocketSocketTransport.BeginOperation(out ASocket: TSocket): Boolean; +begin + FStateLock.Enter; + try + Result := not FIsClosed and not FSocketClosed; + if not Result then + Exit; + ASocket := FSocket; + if FActiveOperations = 0 then + FOperationsFinished.ResetEvent; + Inc(FActiveOperations); + finally + FStateLock.Leave; + end; +end; + +procedure THorseWebSocketSocketTransport.EndOperation; +var + LSocket: TSocket; + LCloseSocket: Boolean; + LSignalFinished: Boolean; +begin + LSocket := 0; + LCloseSocket := False; + FStateLock.Enter; try + Dec(FActiveOperations); + LSignalFinished := FActiveOperations = 0; + if LSignalFinished and FIsClosed and not FSocketClosed then + begin + LSocket := FSocket; + FSocketClosed := True; + LCloseSocket := True; + end; + finally + FStateLock.Leave; + end; + + if LCloseSocket then + begin {$IF DEFINED(FPC)} - Result := fprecv(FSocket, @ABuffer[0], ACount, 0); + CloseSocket(LSocket); {$ELSE} {$IFDEF MSWINDOWS} - Result := recv(FSocket, ABuffer[0], ACount, 0); + closesocket(LSocket); {$ELSE} - Result := recv(FSocket, ABuffer[0], ACount, 0); + Posix.Unistd.__close(LSocket); {$ENDIF} {$ENDIF} - if Result <= 0 then - begin - Result := 0; - FIsClosed := True; - end; - except - Result := 0; - FIsClosed := True; end; + + if LSignalFinished then + FOperationsFinished.SetEvent; end; -function THorseWebSocketSocketTransport.Write(const ABuffer: TBytes; const ACount: Integer): Integer; +function THorseWebSocketSocketTransport.GetLastSocketError: Integer; +{$IF DEFINED(FPC)} + {$IFDEF MSWINDOWS} + begin + Result := WSAGetLastError; + end; + {$ELSE} + begin + Result := fpgeterrno; + end; + {$ENDIF} +{$ELSE} + {$IFDEF MSWINDOWS} + begin + Result := WSAGetLastError; + end; + {$ELSE} + begin + Result := errno; + end; + {$ENDIF} +{$ENDIF} + +function THorseWebSocketSocketTransport.IsInterrupted(const AError: Integer): Boolean; +begin + {$IF DEFINED(FPC)} + {$IFDEF MSWINDOWS} + Result := AError = WSAEINTR; + {$ELSE} + Result := AError = ESysEINTR; + {$ENDIF} + {$ELSE} + {$IFDEF MSWINDOWS} + Result := AError = WSAEINTR; + {$ELSE} + Result := AError = EINTR; + {$ENDIF} + {$ENDIF} +end; + +function THorseWebSocketSocketTransport.WouldBlock(const AError: Integer): Boolean; +begin + {$IF DEFINED(FPC)} + {$IFDEF MSWINDOWS} + Result := AError = WSAEWOULDBLOCK; + {$ELSE} + Result := (AError = ESysEAGAIN) or (AError = ESysEWOULDBLOCK); + {$ENDIF} + {$ELSE} + {$IFDEF MSWINDOWS} + Result := AError = WSAEWOULDBLOCK; + {$ELSE} + Result := (AError = EAGAIN) or (AError = EWOULDBLOCK); + {$ENDIF} + {$ENDIF} +end; + +function THorseWebSocketSocketTransport.WaitReadable(const ASocket: TSocket; + const ATimeoutMS: Integer): Boolean; +{$IF DEFINED(FPC)} + {$IFDEF MSWINDOWS} + var + LSet: TFDSet; + LTimeout: TTimeVal; + begin + LSet.fd_count := 1; + LSet.fd_array[0] := ASocket; + LTimeout.tv_sec := ATimeoutMS div 1000; + LTimeout.tv_usec := (ATimeoutMS mod 1000) * 1000; + Result := select(0, @LSet, nil, nil, @LTimeout) > 0; + end; + {$ELSE} + var + LSet: TFDSet; + LTimeout: TTimeVal; + begin + fpFD_ZERO(LSet); + fpFD_SET(ASocket, LSet); + LTimeout.tv_sec := ATimeoutMS div 1000; + LTimeout.tv_usec := (ATimeoutMS mod 1000) * 1000; + Result := fpSelect(ASocket + 1, @LSet, nil, nil, @LTimeout) > 0; + end; + {$ENDIF} +{$ELSE} + {$IFDEF MSWINDOWS} + var + LSet: TFDSet; + LTimeout: TTimeVal; + begin + { Winapi.WinSock2 exposes FD_SET as a TYPE, not the usual macro-style + procedure, so the members are filled in by hand. } + LSet.fd_count := 1; + LSet.fd_array[0] := ASocket; + LTimeout.tv_sec := ATimeoutMS div 1000; + LTimeout.tv_usec := (ATimeoutMS mod 1000) * 1000; + Result := select(0, @LSet, nil, nil, @LTimeout) > 0; + end; + {$ELSE} + var + LSet: fd_set; + LTimeout: timeval; + begin + FD_ZERO(LSet); + FD_SET(ASocket, LSet); + LTimeout.tv_sec := ATimeoutMS div 1000; + LTimeout.tv_usec := (ATimeoutMS mod 1000) * 1000; + Result := select(ASocket + 1, @LSet, nil, nil, @LTimeout) > 0; + end; + {$ENDIF} +{$ENDIF} + +function THorseWebSocketSocketTransport.Read(var ABuffer: TBytes; const ACount: Integer): Integer; +var + LError: Integer; + LSocket: TSocket; begin Result := 0; - if FIsClosed then + if (ACount <= 0) or (Length(ABuffer) < ACount) or not BeginOperation(LSocket) then Exit; try - {$IF DEFINED(FPC)} - Result := fpsend(FSocket, @ABuffer[0], ACount, 0); - {$ELSE} - {$IFDEF MSWINDOWS} - Result := send(FSocket, ABuffer[0], ACount, 0); - {$ELSE} - Result := send(FSocket, ABuffer[0], ACount, 0); - {$ENDIF} - {$ENDIF} - if Result < 0 then - begin + try + while not Closing do + begin + {$IF DEFINED(FPC)} + Result := fprecv(LSocket, @ABuffer[0], ACount, 0); + {$ELSE} + {$IFDEF MSWINDOWS} + Result := recv(LSocket, ABuffer[0], ACount, 0); + {$ELSE} + Result := recv(LSocket, ABuffer[0], ACount, 0); + {$ENDIF} + {$ENDIF} + + if Result > 0 then + Exit; + + if Result = 0 then + begin + RequestClose(False); + Exit; + end; + + LError := GetLastSocketError; + if IsInterrupted(LError) then + Continue; + if not WouldBlock(LError) then + begin + Result := 0; + RequestClose(False); + Exit; + end; + + WaitReadable(LSocket, WS_SOCKET_READ_TICK_MS); + end; Result := 0; - FIsClosed := True; + except + Result := 0; + RequestClose(False); end; - except - Result := 0; - FIsClosed := True; + finally + EndOperation; end; end; -procedure THorseWebSocketSocketTransport.Close; +function THorseWebSocketSocketTransport.Write(const ABuffer: TBytes; const ACount: Integer): Integer; +var + LSocket: TSocket; begin - if not FIsClosed then - begin - FIsClosed := True; + Result := 0; + if (ACount <= 0) or (Length(ABuffer) < ACount) or not BeginOperation(LSocket) then + Exit; + try try {$IF DEFINED(FPC)} - CloseSocket(FSocket); + Result := fpsend(LSocket, @ABuffer[0], ACount, 0); {$ELSE} {$IFDEF MSWINDOWS} - closesocket(FSocket); + Result := send(LSocket, ABuffer[0], ACount, 0); {$ELSE} - Posix.Unistd.__close(FSocket); + Result := send(LSocket, ABuffer[0], ACount, 0); {$ENDIF} {$ENDIF} + if Result < 0 then + begin + Result := 0; + RequestClose(False); + end; except + Result := 0; + RequestClose(False); end; + finally + EndOperation; end; end; +procedure THorseWebSocketSocketTransport.Close; +begin + RequestClose(True); +end; + function THorseWebSocketSocketTransport.IsConnected: Boolean; begin - Result := not FIsClosed; + Result := not Closing; +end; + +function THorseWebSocketSocketTransport.Closing: Boolean; +begin + FStateLock.Enter; + try + Result := FIsClosed; + finally + FStateLock.Leave; + end; +end; + +procedure THorseWebSocketSocketTransport.RequestClose( + const AWaitForOperations: Boolean); +var + LSocket: TSocket; + LCloseSocket: Boolean; + LHasOperations: Boolean; +begin + LCloseSocket := False; + FStateLock.Enter; + try + FIsClosed := True; + if FSocketClosed then + Exit; + LSocket := FSocket; + LHasOperations := FActiveOperations > 0; + if not LHasOperations then + begin + FSocketClosed := True; + LCloseSocket := True; + end; + finally + FStateLock.Leave; + end; + + if LHasOperations then + begin + {$IF DEFINED(FPC)} + fpShutdown(LSocket, 2); + {$ELSE} + {$IFDEF MSWINDOWS} + shutdown(LSocket, SD_BOTH); + {$ELSE} + shutdown(LSocket, SHUT_RDWR); + {$ENDIF} + {$ENDIF} + if AWaitForOperations then + FOperationsFinished.WaitFor(INFINITE); + end + else if LCloseSocket then + begin + {$IF DEFINED(FPC)} + CloseSocket(LSocket); + {$ELSE} + {$IFDEF MSWINDOWS} + closesocket(LSocket); + {$ELSE} + Posix.Unistd.__close(LSocket); + {$ENDIF} + {$ENDIF} + FOperationsFinished.SetEvent; + end; end; function THorseWebSocketSocketTransport.GetClientIP: string; diff --git a/tests/Dockerfile.tests-lazarus b/tests/Dockerfile.tests-lazarus index 0796be7..d2a3df9 100644 --- a/tests/Dockerfile.tests-lazarus +++ b/tests/Dockerfile.tests-lazarus @@ -15,6 +15,11 @@ COPY . . RUN mkdir -p tests/lib && cd tests/src && \ fpc -Mdelphi -Sh -FE.. -FU../lib -Fu"../../src:modules/jhonson/src:modules/restrequest4delphi/src:modules/cors/src:modules/basic-auth/src" "-dHORSE_CONSOLE" "-dHORSE_PROVIDER_EPOLL" IntegrationServer.dpr +# Regressao do transporte WebSocket non-blocking usado pelo provider epoll. +RUN mkdir -p tests/lib-websocket && cd tests/src && \ + fpc -Mdelphi -Sh -FE.. -FU../lib-websocket -Fu"../../src" "-dHORSE_CONSOLE" "-dHORSE_PROVIDER_EPOLL" WebSocketEpollRegression.dpr && \ + ../WebSocketEpollRegression + WORKDIR /usr/src/app/tests # Executa o servidor de integração que valida as rotas e finaliza autonomamente diff --git a/tests/src/WebSocketEpollRegression.dpr b/tests/src/WebSocketEpollRegression.dpr new file mode 100644 index 0000000..a5ac9f4 --- /dev/null +++ b/tests/src/WebSocketEpollRegression.dpr @@ -0,0 +1,170 @@ +program WebSocketEpollRegression; + +{$MODE DELPHI}{$H+} + +uses + cthreads, + SysUtils, + Classes, + SyncObjs, + Sockets, + BaseUnix, + Horse.Core.WebSocket, + Horse.Provider.Socket.WebSocket; + +const + TEST_TIMEOUT_MS = 3000; + TEST_IDLE_MS = 500; + +type + TTransportReadThread = class(TThread) + private + FTransport: IHorseWebSocketTransport; + FFinished: TEvent; + FReadCount: Integer; + FData: TBytes; + protected + procedure Execute; override; + public + constructor Create(const ATransport: IHorseWebSocketTransport); + destructor Destroy; override; + property FinishedEvent: TEvent read FFinished; + property ReadCount: Integer read FReadCount; + property Data: TBytes read FData; + end; + +constructor TTransportReadThread.Create( + const ATransport: IHorseWebSocketTransport); +begin + inherited Create(True); + FreeOnTerminate := False; + FTransport := ATransport; + FFinished := TEvent.Create(nil, True, False, ''); + SetLength(FData, 16); +end; + +destructor TTransportReadThread.Destroy; +begin + FFinished.Free; + inherited; +end; + +procedure TTransportReadThread.Execute; +begin + try + FReadCount := FTransport.Read(FData, Length(FData)); + finally + FFinished.SetEvent; + end; +end; + +procedure Check(const ACondition: Boolean; const AMessage: string); +begin + if not ACondition then + raise Exception.Create(AMessage); +end; + +procedure SetNonBlocking(const ASocket: cint); +var + LFlags: cint; +begin + LFlags := fpFcntl(ASocket, F_GETFL, 0); + Check(LFlags >= 0, 'Nao foi possivel ler as flags do socket'); + Check(fpFcntl(ASocket, F_SETFL, LFlags or O_NONBLOCK) = 0, + 'Nao foi possivel configurar o socket como non-blocking'); +end; + +procedure TestInitialEagainAndLaterData; +const + PAYLOAD: array[0..3] of Byte = (Ord('p'), Ord('i'), Ord('n'), Ord('g')); +var + LSockets: array[0..1] of cint; + LTransport: IHorseWebSocketTransport; + LReader: TTransportReadThread; +begin + Check(fpSocketPair(AF_UNIX, SOCK_STREAM, 0, @LSockets[0]) = 0, + 'Nao foi possivel criar o socket pair'); + try + SetNonBlocking(LSockets[0]); + LTransport := THorseWebSocketSocketTransport.Create(LSockets[0]); + LReader := TTransportReadThread.Create(LTransport); + try + LReader.Start; + + { Longer than the transport select tick: an idle non-blocking socket must + remain connected instead of treating EAGAIN as a closed peer. } + Sleep(TEST_IDLE_MS); + Check(LReader.FinishedEvent.WaitFor(0) <> wrSignaled, + 'EAGAIN encerrou a leitura enquanto o peer permanecia conectado'); + Check(LTransport.IsConnected, + 'O transporte marcou um socket ocioso como desconectado'); + + Check(fpSend(LSockets[1], @PAYLOAD[0], Length(PAYLOAD), 0) = Length(PAYLOAD), + 'Nao foi possivel enviar o payload de teste'); + Check(LReader.FinishedEvent.WaitFor(TEST_TIMEOUT_MS) = wrSignaled, + 'A leitura nao recebeu os dados enviados depois do EAGAIN'); + Check(LReader.ReadCount = Length(PAYLOAD), + 'A leitura retornou uma quantidade inesperada de bytes'); + Check(CompareByte(LReader.Data[0], PAYLOAD[0], Length(PAYLOAD)) = 0, + 'A leitura retornou um payload diferente do enviado'); + finally + LTransport.Close; + LReader.WaitFor; + LReader.Free; + LTransport := nil; + end; + finally + fpClose(LSockets[1]); + end; +end; + +procedure TestCloseUnblocksPendingRead; +var + LSockets: array[0..1] of cint; + LTransport: IHorseWebSocketTransport; + LReader: TTransportReadThread; +begin + Check(fpSocketPair(AF_UNIX, SOCK_STREAM, 0, @LSockets[0]) = 0, + 'Nao foi possivel criar o socket pair'); + try + SetNonBlocking(LSockets[0]); + LTransport := THorseWebSocketSocketTransport.Create(LSockets[0]); + LReader := TTransportReadThread.Create(LTransport); + try + LReader.Start; + Sleep(TEST_IDLE_MS); + Check(LReader.FinishedEvent.WaitFor(0) <> wrSignaled, + 'A leitura terminou antes do fechamento solicitado'); + + LTransport.Close; + Check(LReader.FinishedEvent.WaitFor(TEST_TIMEOUT_MS) = wrSignaled, + 'Close nao desbloqueou a leitura pendente'); + Check(LReader.ReadCount = 0, + 'A leitura desbloqueada por Close deveria retornar zero'); + Check(not LTransport.IsConnected, + 'O transporte permaneceu conectado depois de Close'); + finally + LTransport.Close; + LReader.WaitFor; + LReader.Free; + LTransport := nil; + end; + finally + fpClose(LSockets[1]); + end; +end; + +begin + try + TestInitialEagainAndLaterData; + TestCloseUnblocksPendingRead; + Writeln('WEBSOCKET EPOLL REGRESSION TEST: SUCCESS'); + ExitCode := 0; + except + on E: Exception do + begin + Writeln('WEBSOCKET EPOLL REGRESSION TEST: FAILED: ', E.Message); + ExitCode := 1; + end; + end; +end.