diff --git a/doc/epoll.md b/doc/epoll.md index 0931f8a..b1f46dc 100644 --- a/doc/epoll.md +++ b/doc/epoll.md @@ -84,3 +84,31 @@ end. ``` The framework will resolve the `THorse.Listen` call to the native epoll reactor when running under Linux. + +## Pipeline execution modes (Delphi) + +By default, route handlers run in a bounded pool owned by the epoll provider. This keeps blocking handlers (database access, filesystem operations, or remote calls) away from the I/O event-loop threads without changing Delphi's process-wide `TThreadPool.Default` settings. + +The default worker count is eight times the processor count and the default pending-request queue capacity is 2048. Configure either value before calling `Listen`: + +```delphi +uses + Horse, + Horse.Provider.Epoll; + +begin + THorseProviderEpoll.PipelineWorkerThreads := 32; // 0 = automatic + THorseProviderEpoll.PipelineQueueCapacity := 1024; + THorse.Listen(9095); +end. +``` + +Applications whose handlers are known to be short and non-blocking can opt into inline execution: + +```delphi +THorseProviderEpoll.PipelineMode := epmInline; +``` + +Inline mode removes the dispatch hop and can improve throughput and tail latency for very short handlers. It also means that one slow handler blocks its epoll worker and delays every connection assigned to that event loop. Do not use inline mode for handlers that perform blocking I/O. Pipeline settings cannot be changed while the provider is running. + +When the bounded queue is full, the provider closes the newly saturated connection instead of creating more threads or silently losing the request. During shutdown, it stops accepting work and drains requests already accepted by the pipeline pool before releasing their connection contexts. diff --git a/doc/epoll.pt-BR.md b/doc/epoll.pt-BR.md index b3ae716..061491e 100644 --- a/doc/epoll.pt-BR.md +++ b/doc/epoll.pt-BR.md @@ -84,3 +84,31 @@ end. ``` O framework resolverá automaticamente a chamada de `THorse.Listen` para o reactor nativo epoll quando executado em ambientes Linux. + +## Modos de execução da pipeline (Delphi) + +Por padrão, os handlers das rotas são executados em um pool limitado pertencente ao próprio provider epoll. Isso mantém handlers bloqueantes (acesso ao banco de dados, filesystem ou chamadas remotas) fora das threads do event loop sem alterar as configurações globais de `TThreadPool.Default` do processo Delphi. + +A quantidade padrão de workers é oito vezes a quantidade de processadores e a capacidade padrão da fila de requisições pendentes é 2048. Configure esses valores antes de chamar `Listen`: + +```delphi +uses + Horse, + Horse.Provider.Epoll; + +begin + THorseProviderEpoll.PipelineWorkerThreads := 32; // 0 = automático + THorseProviderEpoll.PipelineQueueCapacity := 1024; + THorse.Listen(9095); +end. +``` + +Aplicações cujos handlers sejam comprovadamente curtos e não bloqueantes podem optar pela execução inline: + +```delphi +THorseProviderEpoll.PipelineMode := epmInline; +``` + +O modo inline elimina o despacho para outra thread e pode melhorar throughput e latência de cauda para handlers muito curtos. Em contrapartida, um único handler lento bloqueia seu worker epoll e atrasa todas as conexões atribuídas àquele event loop. Não use o modo inline em handlers que realizem E/S bloqueante. As configurações da pipeline não podem ser alteradas enquanto o provider estiver em execução. + +Quando a fila limitada está cheia, o provider fecha a nova conexão saturada em vez de criar mais threads ou perder silenciosamente a requisição. Durante o shutdown, ele deixa de aceitar trabalho e drena as requisições já aceitas pelo pool antes de liberar os contexts das conexões. diff --git a/src/Horse.Provider.Epoll.pas b/src/Horse.Provider.Epoll.pas index 70c7903..0b65b6f 100644 --- a/src/Horse.Provider.Epoll.pas +++ b/src/Horse.Provider.Epoll.pas @@ -27,7 +27,6 @@ interface System.SyncObjs, System.Generics.Collections, System.Generics.Defaults, - System.Threading, System.NetEncoding, Posix.Base, Posix.SysTypes, @@ -54,6 +53,13 @@ interface Horse.Provider.Socket.WebSocket; type + {$IFNDEF FPC} + TEpollPipelineMode = ( + epmBoundedPool, + epmInline + ); + {$ENDIF} + { Estrutura que representa os segmentos de cabeçalhos indexados durante o parsing preguiçoso para evitar alocações desnecessárias na heap. } TEpollConnectionContext = class; @@ -281,9 +287,19 @@ THorseProviderEpoll = class(THorseProviderAbstract) class var FRunning: Boolean; class var FListenSockets: TList; class var FWorkers: TObjectList; + {$IFNDEF FPC} + class var FPipelineMode: TEpollPipelineMode; + class var FPipelineWorkerThreads: Integer; + class var FPipelineQueueCapacity: Integer; + {$ENDIF} class procedure SetPort(const AValue: Integer); static; class procedure SetHost(const AValue: string); static; + {$IFNDEF FPC} + class procedure SetPipelineMode(const AValue: TEpollPipelineMode); static; + class procedure SetPipelineWorkerThreads(const AValue: Integer); static; + class procedure SetPipelineQueueCapacity(const AValue: Integer); static; + {$ENDIF} class function GetPort: Integer; static; class function GetHost: string; static; class function GetDefaultPort: Integer; static; @@ -295,6 +311,11 @@ THorseProviderEpoll = class(THorseProviderAbstract) public class property Host: string read GetHost write SetHost; class property Port: Integer read GetPort write SetPort; + {$IFNDEF FPC} + class property PipelineMode: TEpollPipelineMode read FPipelineMode write SetPipelineMode; + class property PipelineWorkerThreads: Integer read FPipelineWorkerThreads write SetPipelineWorkerThreads; + class property PipelineQueueCapacity: Integer read FPipelineQueueCapacity write SetPipelineQueueCapacity; + {$ENDIF} class procedure Listen; overload; override; class procedure Listen(const APort: Integer; const AHost: string = '0.0.0.0'; const ACallbackListen: TProc = nil; const ACallbackStopListen: TProc = nil); reintroduce; overload; static; class procedure Listen(const APort: Integer; const ACallbackListen: TProc; const ACallbackStopListen: TProc = nil); reintroduce; overload; static; @@ -471,6 +492,38 @@ TEpollFPCTaskPool = class var GTaskPool: TEpollFPCTaskPool = nil; +{$ELSE} +type + TEpollPipelinePool = class; + + TEpollPipelineThread = class(TThread) + private + FPool: TEpollPipelinePool; + protected + procedure Execute; override; + public + constructor Create(APool: TEpollPipelinePool); + end; + + TEpollPipelinePool = class + private + FActive: Boolean; + FHead: Integer; + FLock: TCriticalSection; + FQueue: TArray; + FQueued: Integer; + FSemaphore: TSemaphore; + FTail: Integer; + FWorkers: TObjectList; + function Dequeue(out ATask: TProc): Boolean; + public + constructor Create(AThreadCount, AQueueCapacity: Integer); + destructor Destroy; override; + function TryQueue(const ATask: TProc): Boolean; + end; + +var + GPipelinePool: TEpollPipelinePool = nil; {$ENDIF} const @@ -949,6 +1002,128 @@ function TEpollFPCTaskPool.DequeueTask: TEpollFPCTask; end; {$ENDIF} +{$IFNDEF FPC} +{ TEpollPipelineThread } + +constructor TEpollPipelineThread.Create(APool: TEpollPipelinePool); +begin + inherited Create(True); + FPool := APool; + FreeOnTerminate := False; + Start; +end; + +procedure TEpollPipelineThread.Execute; +var + LTask: TProc; +begin + while True do + begin + FPool.FSemaphore.WaitFor(1000); + if FPool.Dequeue(LTask) then + begin + try + LTask(); + finally + LTask := nil; + end; + Continue; + end; + + if not FPool.FActive then + Break; + end; +end; + +{ TEpollPipelinePool } + +constructor TEpollPipelinePool.Create(AThreadCount, AQueueCapacity: Integer); +var + I: Integer; +begin + inherited Create; + if AThreadCount < 1 then + raise EArgumentOutOfRangeException.Create('Pipeline worker count must be greater than zero'); + if AQueueCapacity < 1 then + raise EArgumentOutOfRangeException.Create('Pipeline queue capacity must be greater than zero'); + + FActive := True; + FLock := TCriticalSection.Create; + SetLength(FQueue, AQueueCapacity); + FSemaphore := TSemaphore.Create(nil, 0, AQueueCapacity + AThreadCount, ''); + FWorkers := TObjectList.Create(True); + for I := 1 to AThreadCount do + FWorkers.Add(TEpollPipelineThread.Create(Self)); +end; + +destructor TEpollPipelinePool.Destroy; +var + I: Integer; +begin + if FLock <> nil then + begin + FLock.Enter; + try + FActive := False; + finally + FLock.Leave; + end; + end + else + FActive := False; + + if FWorkers <> nil then + begin + if FSemaphore <> nil then + for I := 0 to FWorkers.Count - 1 do + FSemaphore.Release; + for I := 0 to FWorkers.Count - 1 do + FWorkers[I].WaitFor; + FWorkers.Free; + end; + for I := 0 to Length(FQueue) - 1 do + FQueue[I] := nil; + FLock.Free; + FSemaphore.Free; + inherited; +end; + +function TEpollPipelinePool.Dequeue(out ATask: TProc): Boolean; +begin + ATask := nil; + FLock.Enter; + try + Result := FQueued > 0; + if not Result then + Exit; + ATask := FQueue[FHead]; + FQueue[FHead] := nil; + FHead := (FHead + 1) mod Length(FQueue); + Dec(FQueued); + finally + FLock.Leave; + end; +end; + +function TEpollPipelinePool.TryQueue(const ATask: TProc): Boolean; +begin + FLock.Enter; + try + Result := FActive and (FQueued < Length(FQueue)); + if Result then + begin + FQueue[FTail] := ATask; + FTail := (FTail + 1) mod Length(FQueue); + Inc(FQueued); + end; + finally + FLock.Leave; + end; + if Result then + FSemaphore.Release; +end; +{$ENDIF} + { TEpollConnectionContext } @@ -2340,6 +2515,7 @@ procedure THorseEpollWorker.ProcessClientRead(AContext: TEpollConnectionContext) LBodyOffset: Integer; LContentLength: Int64; LWorker: THorseEpollWorker; + LInlinePipeline: TProc; begin LWorker := Self; LRequestComplete := False; @@ -2553,8 +2729,11 @@ procedure THorseEpollWorker.ProcessClientRead(AContext: TEpollConnectionContext) AContext.FProcessing := True; {$IF NOT DEFINED(FPC)} - // Delphi Linux: Executa rotas assincronamente no Task Parallel Library - TTask.Run( + // The same pipeline can run inline for CPU-bound/short handlers or in the + // provider-owned bounded pool when handlers may block. Unlike + // TThreadPool.Default, the latter has deterministic concurrency and queue + // limits and does not alter process-wide thread-pool settings. + LInlinePipeline := procedure var LHorseReq: THorseRequest; @@ -2677,9 +2856,14 @@ procedure THorseEpollWorker.ProcessClientRead(AContext: TEpollConnectionContext) end; end; except - // Captura exceções para segurança na thread + LWorker.CloseConnection(LLocalContext); end; - end); + end; + if THorseProviderEpoll.PipelineMode = epmInline then + LInlinePipeline() + else if (GPipelinePool = nil) or + (not GPipelinePool.TryQueue(LInlinePipeline)) then + LWorker.CloseConnection(AContext); {$ELSE} // Lazarus FPC: Despacha as rotas via GTaskPool de forma assíncrona {$IFNDEF HORSE_EPOLL_SYNCHRONOUS} @@ -3179,6 +3363,11 @@ procedure THorseEpollWorker.Execute; FListenSockets := TList.Create; FWorkers := TObjectList.Create(True); FRunning := False; + {$IFNDEF FPC} + FPipelineMode := epmBoundedPool; + FPipelineWorkerThreads := 0; + FPipelineQueueCapacity := 2048; + {$ENDIF} // Eleva o limite máximo de descritores de arquivos abertos (ulimit -n) do processo para 65535 LRLimit.rlim_cur := 65535; @@ -3187,8 +3376,6 @@ procedure THorseEpollWorker.Execute; fpSetrlimit(RLIMIT_NOFILE, @LRLimit); {$ELSE} setrlimit(RLIMIT_NOFILE, LRLimit); - TThreadPool.Default.MaxWorkerThreads := 2048; - TThreadPool.Default.MinWorkerThreads := TThread.ProcessorCount * 8; {$ENDIF} end; @@ -3224,6 +3411,33 @@ class procedure THorseProviderEpoll.SetHost(const AValue: string); FHost := AValue; end; +{$IFNDEF FPC} +class procedure THorseProviderEpoll.SetPipelineMode(const AValue: TEpollPipelineMode); +begin + if FRunning then + raise EInvalidOperation.Create('Pipeline mode cannot be changed while the epoll provider is running'); + FPipelineMode := AValue; +end; + +class procedure THorseProviderEpoll.SetPipelineWorkerThreads(const AValue: Integer); +begin + if FRunning then + raise EInvalidOperation.Create('Pipeline worker count cannot be changed while the epoll provider is running'); + if AValue < 0 then + raise EArgumentOutOfRangeException.Create('Pipeline worker count cannot be negative'); + FPipelineWorkerThreads := AValue; +end; + +class procedure THorseProviderEpoll.SetPipelineQueueCapacity(const AValue: Integer); +begin + if FRunning then + raise EInvalidOperation.Create('Pipeline queue capacity cannot be changed while the epoll provider is running'); + if AValue < 1 then + raise EArgumentOutOfRangeException.Create('Pipeline queue capacity must be greater than zero'); + FPipelineQueueCapacity := AValue; +end; +{$ENDIF} + class procedure THorseProviderEpoll.SetPort(const AValue: Integer); begin FPort := AValue; @@ -3318,10 +3532,22 @@ class procedure THorseProviderEpoll.InternalListen; begin TriggerBeforeListen; if FRunning then Exit; + FRunning := True; - LThreadCount := TThread.ProcessorCount; - if LThreadCount <= 0 then - LThreadCount := 2; + try + LThreadCount := TThread.ProcessorCount; + if LThreadCount <= 0 then + LThreadCount := 2; + + {$IFNDEF FPC} + if FPipelineMode = epmBoundedPool then + begin + I := FPipelineWorkerThreads; + if I = 0 then + I := LThreadCount * 8; + GPipelinePool := TEpollPipelinePool.Create(I, FPipelineQueueCapacity); + end; + {$ENDIF} {$IFDEF FPC} {$IFNDEF HORSE_EPOLL_SYNCHRONOUS} @@ -3330,7 +3556,6 @@ class procedure THorseProviderEpoll.InternalListen; {$ENDIF} {$ENDIF} - try for I := 1 to LThreadCount do begin LSocket := CreateListenSocket(FPort, FHost); @@ -3340,7 +3565,6 @@ class procedure THorseProviderEpoll.InternalListen; LWorker.Start; end; - FRunning := True; DoOnListen; { [EPOLL-LISTEN-1] (re-derived onto merged 2026-07-17, upstream-PR candidate) @@ -3371,17 +3595,23 @@ class procedure THorseProviderEpoll.InternalStopListen; FRunning := False; + for I := 0 to FWorkers.Count - 1 do + FWorkers[I].TerminateWorker; + {$IFDEF FPC} if GTaskPool <> nil then begin GTaskPool.Free; GTaskPool := nil; end; + {$ELSE} + if GPipelinePool <> nil then + begin + GPipelinePool.Free; + GPipelinePool := nil; + end; {$ENDIF} - for I := 0 to FWorkers.Count - 1 do - FWorkers[I].TerminateWorker; - FWorkers.Clear; for I := 0 to FListenSockets.Count - 1 do