From d67554e7884ac4e1ad80e8cf4a54314d3ebdde5d Mon Sep 17 00:00:00 2001 From: "Reymart A. Calicdan" Date: Sun, 21 Jun 2026 00:09:06 +0800 Subject: [PATCH 1/2] Enhance README with explicit lifecycle management details and improve PersistentProcess class for better task handling and resource management --- README.md | 23 ++- src/Internals/PersistentProcess.php | 295 ++++++++++++++++------------ src/Internals/Process.php | 7 +- src/Managers/ProcessPoolManager.php | 2 +- src/ProcessPool.php | 2 +- 5 files changed, 186 insertions(+), 143 deletions(-) diff --git a/README.md b/README.md index 184e0f8..1169ebb 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ Hibla Parallel brings Erlang-style reliability and Node.js-level cluster pool pe **Persistent pools** - [Persistent Worker Pools](#persistent-worker-pools) + - [Explicit Lifecycle Management](#explicit-lifecycle-management) - [Automatic garbage collection between tasks](#automatic-garbage-collection-between-tasks) - [Why `withMaxExecutionsPerWorker`](#why-withmaxexecutionsperworker) - [Lazy vs. Eager Spawning](#lazy-vs-eager-spawning) @@ -781,17 +782,23 @@ for ($i = 0; $i < 4; $i++) { $pool->run($task)->then(fn($pid) => print("Handled by worker: $pid\n")); } -/** - * CRITICAL: Always shut down the pool when done. - * Persistent workers hold open IPC channels that keep the Event Loop alive. - * Without an explicit shutdown the script will never exit. - */ +// Once all tasks are complete and the event loop is empty, the script will +// exit naturally. PHP's Garbage Collector will automatically tear down the +// OS processes via the pool's __destruct() method. No manual cleanup is required! +``` + +### Explicit Lifecycle Management -// Option A: Synchronous — blocks until all queued tasks finish and all workers exit +While Hibla Parallel automatically cleans up idle pools when your script ends, you may still want to shut down pools manually in long-running daemons (like web servers) to free up OS resources immediately when a pool is no longer needed. + +```php +// Option A: Forceful Shutdown — instantly kills all workers in the pool. +// Any currently executing or queued tasks are rejected with a PoolShutdownException. $pool->shutdown(); -// Option B: Graceful — rejects all incoming tasks while still waiting for the current task to finish, then shuts down -// $pool->drain(); +// Option B: Graceful Drain — stops accepting new tasks, but waits for all +// currently executing and queued tasks to finish before tearing down the workers. +$pool->drain(); ``` ### Automatic garbage collection between tasks diff --git a/src/Internals/PersistentProcess.php b/src/Internals/PersistentProcess.php index 8d683db..1b3fe28 100644 --- a/src/Internals/PersistentProcess.php +++ b/src/Internals/PersistentProcess.php @@ -6,12 +6,12 @@ use Hibla\Parallel\Exceptions\ProcessCrashedException; use Hibla\Parallel\Handlers\ExceptionHandler; -use Rcalicdan\ProcessKiller\ProcessKiller; use Hibla\Parallel\ValueObjects\WorkerMessage; use Hibla\Promise\Interfaces\PromiseInterface; use Hibla\Promise\Promise; use Hibla\Stream\Interfaces\PromiseReadableStreamInterface; use Hibla\Stream\Interfaces\PromiseWritableStreamInterface; +use Rcalicdan\ProcessKiller\ProcessKiller; use function Hibla\async; use function Hibla\await; @@ -27,19 +27,21 @@ final class PersistentProcess private array $pendingTasks = []; /** - * @var callable(self): void + * @var (callable(self): void)|null */ - private $onReadyCallback; + private $onReadyCallback = null; /** - * @var callable(self): void + * @var (callable(self): void)|null */ - private $onCrashCallback; + private $onCrashCallback = null; private bool $isAlive = true; private bool $isBusy = true; + private bool $isReading = false; + /** * @var int|null */ @@ -63,13 +65,157 @@ public function startReadLoop(callable $onReadyCallback, callable $onCrashCallba $this->onReadyCallback = $onReadyCallback; $this->onCrashCallback = $onCrashCallback; + // Boot phase: start reading to catch the initial READY frame + $this->ensureReading(); + } + + public function getPid(): int + { + return $this->pid; + } + + public function getWorkerPid(): int + { + return $this->workerPid ?? $this->pid; + } + + /** + * @param callable(WorkerMessage): void|null $onMessage + * + * @return PromiseInterface + */ + public function submitTask(string $taskId, string $payload, string $sourceLocation = 'unknown', ?callable $onMessage = null): PromiseInterface + { + $this->isBusy = true; + + /** @var Promise $promise */ + $promise = new Promise(); + + $this->pendingTasks[$taskId] = [ + 'promise' => $promise, + 'location' => $sourceLocation, + 'onMessage' => $onMessage, + ]; + + $this->ensureReading(); + + async(function () use ($payload) { + try { + if (! $this->isAlive) { + return; + } + + await($this->stdin->writeAsync($payload . PHP_EOL)); + } catch (\Hibla\Stream\Exceptions\StreamException) { + } + }); + + return $promise; + } + + public function isBusy(): bool + { + return $this->isBusy; + } + + public function isAlive(): bool + { + return $this->isAlive; + } + + /** + * Sends a termination signal to the worker process without blocking. + * + * @param bool $executeOsKill If true, issues the OS command to kill the tree. If false, + * only internal state is updated (used during batched pool shutdown). + */ + public function signalTerminate(bool $executeOsKill = true): void + { + if (! $this->isAlive) { + return; + } + + $this->handleCrash(); + + if ($executeOsKill) { + ProcessKiller::killTreesAsync([$this->pid]); + } + } + + /** + * Closes all I/O streams and optionally releases the proc resource. + * + * @param bool $closeProcessResource If false, the proc_open resource is left alive + * so the Windows tree structure isn't broken. + */ + public function cleanupResources(bool $closeProcessResource = true): void + { + $this->stdin->close(); + $this->stdout->close(); + $this->stderr->close(); + + if ($closeProcessResource && \is_resource($this->processResource)) { + if (PHP_OS_FAMILY === 'Windows') { + @proc_terminate($this->processResource); + } else { + @proc_terminate($this->processResource); + @proc_close($this->processResource); + } + } + } + + /** + * Terminates the worker process and releases all associated resources. + */ + public function terminate(): void + { + if (! $this->isAlive) { + return; + } + + $this->signalTerminate(true); + $this->cleanupResources(PHP_OS_FAMILY !== 'Windows'); + } + + private function handleCrash(): void + { + $this->isAlive = false; + $this->isBusy = false; + + foreach ($this->pendingTasks as $taskId => $taskMeta) { + $taskMeta['promise']->reject(new ProcessCrashedException( + "Persistent worker PID {$this->pid} crashed or stream closed while executing task {$taskId}." + )); + } + + $this->pendingTasks = []; + } + + /** + * Demand-driven read loop that only listens when the worker is busy (booting or executing). + * Automatically suspends itself when idle to allow the event loop to exit cleanly. + */ + private function ensureReading(): void + { + if ($this->isReading) { + return; + } + + $this->isReading = true; + async(function () { /** @var array>> $pendingHandlers */ $pendingHandlers = []; $buffer = ''; // JSON Reassembly Buffer try { - while (null !== ($line = await($this->stdout->readLineAsync()))) { + while ($this->isBusy) { + $line = await($this->stdout->readLineAsync()); + + if ($line === null) { + throw new \RuntimeException('Worker stream closed unexpectedly.'); + } + $buffer .= $line; if (trim($buffer) === '') { @@ -109,7 +255,9 @@ public function startReadLoop(callable $onReadyCallback, callable $onCrashCallba if ($status === 'CRASHED' || $status === 'RETIRING') { $this->terminate(); - ($this->onCrashCallback)($this); + if ($this->onCrashCallback !== null) { + ($this->onCrashCallback)($this); + } break; } @@ -119,8 +267,11 @@ public function startReadLoop(callable $onReadyCallback, callable $onCrashCallba $this->workerPid = $data['pid']; } + // Suspend the read loop and notify the pool manager $this->isBusy = false; - ($this->onReadyCallback)($this); + if ($this->onReadyCallback !== null) { + ($this->onReadyCallback)($this); + } continue; } @@ -199,131 +350,15 @@ public function startReadLoop(callable $onReadyCallback, callable $onCrashCallba // and clear all pending handler references before crashing. $pendingHandlers = []; $this->terminate(); - ($this->onCrashCallback)($this); - } finally { - $pendingHandlers = []; - $this->terminate(); - } - }); - } - public function getPid(): int - { - return $this->pid; - } - - public function getWorkerPid(): int - { - return $this->workerPid ?? $this->pid; - } - - /** - * @param callable(WorkerMessage): void|null $onMessage - * - * @return PromiseInterface - */ - public function submitTask(string $taskId, string $payload, string $sourceLocation = 'unknown', ?callable $onMessage = null): PromiseInterface - { - $this->isBusy = true; - - /** @var Promise $promise */ - $promise = new Promise(); - - $this->pendingTasks[$taskId] = [ - 'promise' => $promise, - 'location' => $sourceLocation, - 'onMessage' => $onMessage, - ]; - - async(function () use ($payload) { - try { - if (! $this->isAlive) { - return; + if ($this->onCrashCallback !== null) { + ($this->onCrashCallback)($this); } - - await($this->stdin->writeAsync($payload . PHP_EOL)); - } catch (\Hibla\Stream\Exceptions\StreamException) { + } finally { + $this->isReading = false; + $pendingHandlers = []; + // Unconditional terminate intentionally removed here to allow idle worker to stay alive } }); - - return $promise; - } - - public function isBusy(): bool - { - return $this->isBusy; - } - - public function isAlive(): bool - { - return $this->isAlive; - } - - /** - * Sends a termination signal to the worker process without blocking. - * - * @param bool $executeOsKill If true, issues the OS command to kill the tree. If false, - * only internal state is updated (used during batched pool shutdown). - */ - public function signalTerminate(bool $executeOsKill = true): void - { - if (! $this->isAlive) { - return; - } - - $this->handleCrash(); - - if ($executeOsKill) { - ProcessKiller::killTreesAsync([$this->pid]); - } - } - - /** - * Closes all I/O streams and optionally releases the proc resource. - * - * @param bool $closeProcessResource If false, the proc_open resource is left alive - * so the Windows tree structure isn't broken. - */ - public function cleanupResources(bool $closeProcessResource = true): void - { - $this->stdin->close(); - $this->stdout->close(); - $this->stderr->close(); - - if ($closeProcessResource && \is_resource($this->processResource)) { - if (PHP_OS_FAMILY === 'Windows') { - @proc_terminate($this->processResource); - } else { - @proc_terminate($this->processResource); - @proc_close($this->processResource); - } - } - } - - /** - * Terminates the worker process and releases all associated resources. - */ - public function terminate(): void - { - if (! $this->isAlive) { - return; - } - - $this->signalTerminate(true); - $this->cleanupResources(PHP_OS_FAMILY !== 'Windows'); - } - - private function handleCrash(): void - { - $this->isAlive = false; - $this->isBusy = false; - - foreach ($this->pendingTasks as $taskId => $taskMeta) { - $taskMeta['promise']->reject(new ProcessCrashedException( - "Persistent worker PID {$this->pid} crashed or stream closed while executing task {$taskId}." - )); - } - - $this->pendingTasks = []; } } diff --git a/src/Internals/Process.php b/src/Internals/Process.php index b234f68..b6da03f 100644 --- a/src/Internals/Process.php +++ b/src/Internals/Process.php @@ -7,7 +7,6 @@ use Hibla\Parallel\Exceptions\ProcessCrashedException; use Hibla\Parallel\Exceptions\TimeoutException; use Hibla\Parallel\Handlers\ExceptionHandler; -use Rcalicdan\ProcessKiller\ProcessKiller; use Hibla\Parallel\ValueObjects\WorkerMessage; use Hibla\Promise\Exceptions\TimeoutException as PromiseTimeoutException; use Hibla\Promise\Interfaces\PromiseInterface; @@ -15,6 +14,7 @@ use Hibla\Stream\Exceptions\StreamException; use Hibla\Stream\Interfaces\PromiseReadableStreamInterface; use Hibla\Stream\Interfaces\PromiseWritableStreamInterface; +use Rcalicdan\ProcessKiller\ProcessKiller; use function Hibla\async; use function Hibla\await; @@ -46,7 +46,8 @@ public function __construct( private readonly PromiseReadableStreamInterface $stdout, private readonly PromiseReadableStreamInterface $stderr, private readonly string $sourceLocation = 'unknown' - ) {} + ) { + } /** * Get the result of the background process. @@ -258,7 +259,7 @@ private function readResultFromStream(?callable $onMessage): PromiseInterface data: $data, pid: \is_int($status['pid']) ? $status['pid'] : $this->pid, ); - $pendingHandlers[] = async(fn() => $onMessage($message)); + $pendingHandlers[] = async(fn () => $onMessage($message)); } } elseif ($statusType === 'COMPLETED') { $result = $status['result'] ?? null; diff --git a/src/Managers/ProcessPoolManager.php b/src/Managers/ProcessPoolManager.php index 6a214f7..4e9e146 100644 --- a/src/Managers/ProcessPoolManager.php +++ b/src/Managers/ProcessPoolManager.php @@ -13,11 +13,11 @@ use Hibla\Parallel\Handlers\ProcessSpawnHandler; use Hibla\Parallel\Internals\PersistentProcess; use Hibla\Parallel\Traits\MessageHandlerComposer; -use Rcalicdan\ProcessKiller\ProcessKiller; use Hibla\Parallel\ValueObjects\WorkerMessage; use Hibla\Promise\Exceptions\TimeoutException as PromiseTimeoutException; use Hibla\Promise\Interfaces\PromiseInterface; use Hibla\Promise\Promise; +use Rcalicdan\ProcessKiller\ProcessKiller; use Rcalicdan\Serializer\CallbackSerializationManager; use SplQueue; diff --git a/src/ProcessPool.php b/src/ProcessPool.php index 6f0c803..4e6d123 100644 --- a/src/ProcessPool.php +++ b/src/ProcessPool.php @@ -4,8 +4,8 @@ namespace Hibla\Parallel; -use Hibla\Parallel\Interfaces\ProcessPoolInterface; use Hibla\Parallel\Exceptions\PoolShutdownException; +use Hibla\Parallel\Interfaces\ProcessPoolInterface; use Hibla\Parallel\Managers\ProcessManager; use Hibla\Parallel\Managers\ProcessPoolManager; use Hibla\Parallel\ValueObjects\WorkerMessage; From da5ff0c4e89330769f2cffd477cbafbbba30b673 Mon Sep 17 00:00:00 2001 From: "Reymart A. Calicdan" Date: Sun, 21 Jun 2026 00:28:57 +0800 Subject: [PATCH 2/2] Fix read loop behavior in PersistentProcess to prevent hanging during synchronous task submission --- src/Internals/PersistentProcess.php | 8 ++++++-- tests/Unit/PersistentProcessTest.php | 4 ---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Internals/PersistentProcess.php b/src/Internals/PersistentProcess.php index 1b3fe28..d051564 100644 --- a/src/Internals/PersistentProcess.php +++ b/src/Internals/PersistentProcess.php @@ -267,8 +267,12 @@ private function ensureReading(): void $this->workerPid = $data['pid']; } - // Suspend the read loop and notify the pool manager - $this->isBusy = false; + // Prevent the read loop from exiting if a task was just submitted + // (Fixes test suite hanging due to synchronous task submission) + if (\count($this->pendingTasks) === 0) { + $this->isBusy = false; + } + if ($this->onReadyCallback !== null) { ($this->onReadyCallback)($this); } diff --git a/tests/Unit/PersistentProcessTest.php b/tests/Unit/PersistentProcessTest.php index f0702cf..1160735 100644 --- a/tests/Unit/PersistentProcessTest.php +++ b/tests/Unit/PersistentProcessTest.php @@ -365,7 +365,6 @@ function (WorkerMessage $msg) use (&$log) { $crashFired = false; $lines = [ - json_encode(['status' => 'READY', 'pid' => 1234]), json_encode(['status' => 'CRASHED']), ]; @@ -406,7 +405,6 @@ function (PersistentProcess $p) use (&$crashFired) { it('marks the worker as dead after a CRASHED frame', function () { $lines = [ - json_encode(['status' => 'READY', 'pid' => 1234]), json_encode(['status' => 'CRASHED']), ]; @@ -422,7 +420,6 @@ function (PersistentProcess $p) use (&$crashFired) { $retireFired = false; $lines = [ - json_encode(['status' => 'READY', 'pid' => 1234]), json_encode(['status' => 'RETIRING', 'executions' => 10]), ]; @@ -441,7 +438,6 @@ function (PersistentProcess $p) use (&$retireFired) { it('marks the worker as dead after a RETIRING frame', function () { $lines = [ - json_encode(['status' => 'READY', 'pid' => 1234]), json_encode(['status' => 'RETIRING', 'executions' => 5]), ];