From 46e1a24cc5272072d0db858b7e868f61a42d88a4 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:46:34 +0000 Subject: [PATCH 01/10] Implement worker console message output alongside ProgressBar - Add ParallelWorker::write() and ParallelWorker::writeln() methods. - Route messages through ProgressBarWorker when a progress bar is active (clear/write/display). - Add a ConsoleWorker fallback for workers without a progress bar. - Use a shared stderr OutputInterface so ProgressBar and messages share the same stream. - Add WriteOutputMessage command and ConsoleWorker/HasChannels infrastructure. - Update README and CHANGELOG. Co-Authored-By: Hermann D. Schimpf --- CHANGELOG.md | 10 +++ README.md | 27 +++++++ .../Commands/Output/WriteOutputMessage.php | 23 ++++++ src/Internals/ConsoleWorker.php | 32 ++++++++ src/Internals/ConsoleWorker/HasChannels.php | 41 ++++++++++ src/Internals/ProgressBarWorker.php | 7 ++ .../ProgressBarWorker/HasProgressBar.php | 10 ++- src/Internals/Runner.php | 3 + src/Internals/Runner/HasSharedConsole.php | 80 +++++++++++++++++++ src/Internals/Runner/ManagesTasks.php | 12 ++- .../CommunicatesWithProgressBarWorker.php | 59 ++++++++++++++ tests/ParallelTest.php | 66 +++++++++++++++ 12 files changed, 368 insertions(+), 2 deletions(-) create mode 100644 src/Internals/Commands/Output/WriteOutputMessage.php create mode 100644 src/Internals/ConsoleWorker.php create mode 100644 src/Internals/ConsoleWorker/HasChannels.php create mode 100644 src/Internals/Runner/HasSharedConsole.php diff --git a/CHANGELOG.md b/CHANGELOG.md index d871b69..998b513 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ All notable changes to **parallel-sdk** are documented in this file. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and the project adheres to [Semantic Versioning](https://semver.org/). +## Unreleased + +### Added +- `ParallelWorker::write()` and `ParallelWorker::writeln()` methods to emit console messages from workers without them being overwritten by the ProgressBar. +- `ConsoleWorker` thread that owns a fallback `ConsoleOutput` for workers that do not use a ProgressBar. +- `WriteOutputMessage` command to route `write()`/`writeln()` calls through the existing channel infrastructure. + +### Changed +- `ProgressBarWorker` now uses a shared `stderr` `OutputInterface` for both the ProgressBar and messages, so `clear()`/`write()`/`display()` work correctly together. + ## `3.0.0` – 2025-07-04 ### Added diff --git a/README.md b/README.md index f9ad3a2..57436bd 100644 --- a/README.md +++ b/README.md @@ -321,6 +321,8 @@ Available methods are: - `setProgress(int $step)` - `display()` - `clear()` +- `write(string $message, bool $newline = false)` +- `writeln(string $message)` ```php use HDSSolutions\Console\Parallel\ParallelWorker; @@ -332,6 +334,7 @@ final class ExampleWorker extends ParallelWorker { $microseconds = random_int(100, 500); $this->setMessage(sprintf("ExampleWorker >> Hello from task #%u, I'll wait %sms", $number, $microseconds)); usleep($microseconds * 1000); + $this->writeln(sprintf("ExampleWorker >> Finished task #%u", $number)); $this->advance(); // end example process @@ -349,6 +352,30 @@ final class ExampleWorker extends ParallelWorker { memory: 562 KiB, threads: 12x ~474 KiB, Σ 5,6 MiB ↑ 5,6 MiB ``` +#### Console messages + +Use `write()` or `writeln()` to emit ad-hoc console messages from a worker. When a ProgressBar is active, the bar is temporarily hidden, the message is printed, and the bar is redrawn below it so the message is not overwritten. + +These methods also work for workers that did not enable `withProgress()`; messages are then routed to a dedicated console output handler. + +```php +use HDSSolutions\Console\Parallel\ParallelWorker; + +final class ExampleWorker extends ParallelWorker { + + protected function process(int $number = 0): int { + $this->writeln(sprintf("Processing task #%u", $number)); + + // ... do work ... + + $this->writeln(sprintf("Finished task #%u", $number)); + + return $number; + } + +} +``` + ### References 1. [parallel\bootstrap()](https://www.php.net/manual/en/parallel.bootstrap.php) 2. [parallel\Runtime](https://www.php.net/manual/en/class.parallel-runtime.php) diff --git a/src/Internals/Commands/Output/WriteOutputMessage.php b/src/Internals/Commands/Output/WriteOutputMessage.php new file mode 100644 index 0000000..bb6b822 --- /dev/null +++ b/src/Internals/Commands/Output/WriteOutputMessage.php @@ -0,0 +1,23 @@ +openChannels(); + $this->output = (new ConsoleOutput)->getErrorOutput(); + } + + public function afterListening(): void { + $this->closeChannels(); + } + + private function writeOutput(string $message, bool $newline = true): void { + $this->output->write($message, $newline); + } + +} diff --git a/src/Internals/ConsoleWorker/HasChannels.php b/src/Internals/ConsoleWorker/HasChannels.php new file mode 100644 index 0000000..55ed2b2 --- /dev/null +++ b/src/Internals/ConsoleWorker/HasChannels.php @@ -0,0 +1,41 @@ +console_channel = TwoWayChannel::make(self::class.'@'.$this->uuid); + } + + protected function recv(): mixed { + return $this->console_channel->receive(); + } + + protected function send(mixed $value): mixed { + return $this->console_channel->send($value); + } + + protected function release(): bool { + return $this->send(true); + } + + private function closeChannels(): void { + // gracefully join + $this->console_channel->send(false); + // close channel + $this->console_channel->close(); + } + +} diff --git a/src/Internals/ProgressBarWorker.php b/src/Internals/ProgressBarWorker.php index 8b7839f..eefc1a2 100644 --- a/src/Internals/ProgressBarWorker.php +++ b/src/Internals/ProgressBarWorker.php @@ -64,6 +64,13 @@ private function progressBarAction(string $action, array $args): void { } } + private function writeOutput(string $message, bool $newline = true): void { + // clear the bar, write the message, then redraw the bar below it + $this->progressBar->clear(); + $this->output->write($message, $newline); + $this->progressBar->display(); + } + private function statsReport(string $worker_id, int $memory_usage): void { // save memory usage of thread $this->threads_memory['current'][$worker_id] = $memory_usage; diff --git a/src/Internals/ProgressBarWorker/HasProgressBar.php b/src/Internals/ProgressBarWorker/HasProgressBar.php index d35f822..6110852 100644 --- a/src/Internals/ProgressBarWorker/HasProgressBar.php +++ b/src/Internals/ProgressBarWorker/HasProgressBar.php @@ -4,6 +4,7 @@ use Symfony\Component\Console\Helper\ProgressBar; use Symfony\Component\Console\Output\ConsoleOutput; +use Symfony\Component\Console\Output\OutputInterface; trait HasProgressBar { @@ -17,8 +18,15 @@ trait HasProgressBar { */ private bool $progressBarStarted = false; + /** + * @var OutputInterface Output stream used for both the ProgressBar and messages + */ + private OutputInterface $output; + private function createProgressBar(): void { - $this->progressBar = new ProgressBar(new ConsoleOutput); + // use the stderr output stream; ProgressBar uses the same stream internally + $this->output = (new ConsoleOutput)->getErrorOutput(); + $this->progressBar = new ProgressBar($this->output); // configure ProgressBar settings $this->progressBar->setBarWidth(80); diff --git a/src/Internals/Runner.php b/src/Internals/Runner.php index 0d48d07..bd5d5c3 100644 --- a/src/Internals/Runner.php +++ b/src/Internals/Runner.php @@ -14,6 +14,7 @@ final class Runner { use Runner\HasChannels; use Runner\HasEater; use Runner\HasSharedProgressBar; + use Runner\HasSharedConsole; use Runner\ManagesWorkers; use Runner\ManagesTasks; @@ -33,6 +34,7 @@ public function __construct( protected function afterListening(): void { $this->stopEater(); $this->stopRunningTasks(); + $this->stopConsole(); $this->closeChannels(); } @@ -244,6 +246,7 @@ private function await(?int $wait_until = null): bool { } public function __destruct() { + $this->stopConsole(); $this->stopProgressBar(); } diff --git a/src/Internals/Runner/HasSharedConsole.php b/src/Internals/Runner/HasSharedConsole.php new file mode 100644 index 0000000..2e21de6 --- /dev/null +++ b/src/Internals/Runner/HasSharedConsole.php @@ -0,0 +1,80 @@ +consoleOutput ??= new ConsoleOutput(); + + return; + } + + // already started + if ($this->console_started) return; + + // create a ConsoleWorker instance inside a thread + $this->console ??= parallel\run(static function(string $uuid): void { + // create ConsoleWorker instance + $console = new ConsoleWorker($uuid); + // listen for events + $console->listen(); + }, [ $this->uuid ]); + + // open communication channel with the Console worker + while ($this->console_channel === null) { + try { $this->console_channel = TwoWayChannel::open(ConsoleWorker::class.'@'.$this->uuid); + // wait 1ms if channel does not exist yet and retry + } catch (Channel\Error\Existence) { usleep(1_000); } + } + + // wait until Console worker starts + $this->console_channel->receive(); + $this->console_started = true; + } + + private function stopConsole(): void { + if (! PARALLEL_EXT_LOADED || ! $this->console_started) return; + + // stop Console worker instance + $this->console_channel->send(Event\Type::Close); + // wait until Console worker instance shutdowns + $this->console_channel->receive(); + } + + private function writeOutput(string $message, bool $newline = true): void { + $this->consoleOutput ??= new ConsoleOutput(); + $this->consoleOutput->getErrorOutput()->write($message, $newline); + } + +} diff --git a/src/Internals/Runner/ManagesTasks.php b/src/Internals/Runner/ManagesTasks.php index 86c0cfe..0511405 100644 --- a/src/Internals/Runner/ManagesTasks.php +++ b/src/Internals/Runner/ManagesTasks.php @@ -99,6 +99,10 @@ private function startNextPendingTask(): void { $worker->connectProgressBar($uuid, $GLOBALS['worker_thread_id'] ??= sprintf('%s@%s', $uuid, substr(md5(uniqid($worker_class, true)), 0, 16))); } + // initialize console output and connect worker to it + $this->initConsole(); + $worker->connectConsole($uuid); + // notify that thread started Channel::open(sprintf('starter@%s', $uuid))->send(true); @@ -141,6 +145,9 @@ private function startNextPendingTask(): void { // process task using user Worker : [ ...$task->getInput() ]; + // initialize console output (also initializes the local ConsoleOutput on non-threaded environments) + $this->initConsole(); + // check if worker has ProgressBar enabled if ($registered_worker->hasProgressEnabled()) { // init progressbar @@ -151,9 +158,12 @@ private function startNextPendingTask(): void { steps: $registered_worker->getSteps(), )); // connect worker to ProgressBar - $worker->connectProgressBar(fn(Commands\ProgressBar\ProgressBarActionMessage $message) => $this->progressBar->processMessage($message)); + $worker->connectProgressBar(fn(Commands\ParallelCommandMessage $message) => $this->progressBar->processMessage($message)); } + // connect worker to console output fallback + $worker->connectConsole(fn(Commands\Output\WriteOutputMessage $message) => $this->writeOutput(...$message->args)); + $task->setState(Task::STATE_Processing); // process task using worker diff --git a/src/Internals/Worker/CommunicatesWithProgressBarWorker.php b/src/Internals/Worker/CommunicatesWithProgressBarWorker.php index e16919c..e5db495 100644 --- a/src/Internals/Worker/CommunicatesWithProgressBarWorker.php +++ b/src/Internals/Worker/CommunicatesWithProgressBarWorker.php @@ -5,6 +5,7 @@ use Closure; use HDSSolutions\Console\Parallel\Internals\Communication\TwoWayChannel; use HDSSolutions\Console\Parallel\Internals\Commands; +use HDSSolutions\Console\Parallel\Internals\ConsoleWorker; use HDSSolutions\Console\Parallel\Internals\ProgressBarWorker; use parallel\Channel; @@ -15,6 +16,11 @@ trait CommunicatesWithProgressBarWorker { */ private TwoWayChannel | Closure | null $progressbar_channel = null; + /** + * @var TwoWayChannel|Closure|null Channel of communication between Task and Console output + */ + private TwoWayChannel | Closure | null $console_channel = null; + final public function connectProgressBar(string | Closure $uuid, string $identifier = null): bool { if (! PARALLEL_EXT_LOADED) { $this->progressbar_channel = $uuid; @@ -36,6 +42,24 @@ final public function connectProgressBar(string | Closure $uuid, string $identif return true; } + final public function connectConsole(string | Closure $uuid, string $identifier = null): bool { + if (! PARALLEL_EXT_LOADED) { + $this->console_channel = $uuid; + + return true; + } + + // open channel if not already opened + while ($this->console_channel === null) { + // open channel to communicate with the Console output worker instance + try { $this->console_channel = TwoWayChannel::open(ConsoleWorker::class.'@'.$uuid); + // wait 1ms if channel does not exist yet and retry + } catch (Channel\Error\Existence) { usleep(1_000); } + } + + return true; + } + final public function setMessage(string $message, string $name = 'message'): void { $this->newProgressBarAction(__FUNCTION__, $message, $name); } @@ -56,6 +80,41 @@ final public function clear(): void { $this->newProgressBarAction(__FUNCTION__); } + final public function write(string $message, bool $newline = false): void { + $this->sendOutputMessage(new Commands\Output\WriteOutputMessage($message, $newline)); + } + + final public function writeln(string $message): void { + $this->write($message, true); + } + + private function sendOutputMessage(Commands\Output\WriteOutputMessage $message): void { + // if a progress bar is active, route the message through the ProgressBar worker + if ($this->progressbar_channel !== null) { + if (PARALLEL_EXT_LOADED) { + $this->progressbar_channel->send($message); + } else { + ($this->progressbar_channel)($message); + } + + return; + } + + // if a console output channel is connected, route the message there + if ($this->console_channel !== null) { + if (PARALLEL_EXT_LOADED) { + $this->console_channel->send($message); + } else { + ($this->console_channel)($message); + } + + return; + } + + // fallback when no coordinator is available + fwrite(STDERR, $message->args[0].($message->args[1] ? PHP_EOL : '')); + } + private function newProgressBarAction(string $action, ...$args): void { // check if progressbar is active if ($this->progressbar_channel === null) return; diff --git a/tests/ParallelTest.php b/tests/ParallelTest.php index 0da9674..e7e09f5 100644 --- a/tests/ParallelTest.php +++ b/tests/ParallelTest.php @@ -226,4 +226,70 @@ public function testThatCpuUsageCanBeControlled(): void { $this->assertGreaterThanOrEqual(1, time() - $start); } + public function testThatWorkerCanWriteMessagesWithoutProgressBar(): void { + $output = $this->runWorkerScript(<<<'PHP' +Scheduler::using(Writer::class); +foreach (range(1, 3) as $i) { + Scheduler::runTask($i); +} +Scheduler::awaitTasksCompletion(); +PHP); + + $this->assertStringContainsString('Starting #1', $output); + $this->assertStringContainsString('Starting #2', $output); + $this->assertStringContainsString('Starting #3', $output); + } + + public function testThatWorkerCanWriteMessagesWithProgressBar(): void { + $output = $this->runWorkerScript(<<<'PHP' +Scheduler::using(Writer::class)->withProgress(steps: 3); +foreach (range(1, 3) as $i) { + Scheduler::runTask($i); +} +Scheduler::awaitTasksCompletion(); +PHP); + + $this->assertStringContainsString('Starting #1', $output); + $this->assertStringContainsString('Done #3', $output); + $this->assertStringContainsString('3 of 3: Task #3', $output); + } + + private function runWorkerScript(string $body): string { + $autoload = __DIR__.'/../vendor/autoload.php'; + + $script = <<<'PHP' +setMessage(sprintf('Task #%d', $n)); + $this->writeln(sprintf('Starting #%d', $n)); + $this->writeln(sprintf('Done #%d', $n)); + $this->advance(); + + return $n; + } +} + +__BODY__ +PHP; + + $file = tempnam(sys_get_temp_dir(), 'parallel_sdk_test_').'.php'; + file_put_contents($file, str_replace(['__AUTOLOAD__', '__BODY__'], [var_export($autoload, true), $body], $script)); + + $output = []; + $exit = 0; + exec(sprintf('%s %s 2>&1', escapeshellarg(PHP_BINARY), escapeshellarg($file)), $output, $exit); + + unlink($file); + + $this->assertSame(0, $exit, 'Worker script exited with an error'); + + return implode("\n", $output); + } + } From def3d9a155ac7d31c170d4c2dd950b67aa1399f4 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:10:38 +0000 Subject: [PATCH 02/10] docs: add console message example output Co-Authored-By: Hermann D. Schimpf --- README.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/README.md b/README.md index 57436bd..2b6f236 100644 --- a/README.md +++ b/README.md @@ -376,6 +376,20 @@ final class ExampleWorker extends ParallelWorker { } ``` +#### Example output +When used together with `withProgress()`, messages are printed between progress-bar redraws instead of being overwritten: + +```bash + 0 of 10: Starting... + [>------------------------------------------------------------------------] 0% + elapsed: < 1 sec, remaining: < 1 sec, ?? items/s,memory: ?? + Processing task #5 + Finished task #5 + 1 of 10: Task #5 + [=====>-------------------------------------------------------------------] 10% + elapsed: < 1 sec, remaining: < 1 sec, ~1.00 items/s,memory: ?? +``` + ### References 1. [parallel\bootstrap()](https://www.php.net/manual/en/parallel.bootstrap.php) 2. [parallel\Runtime](https://www.php.net/manual/en/class.parallel-runtime.php) From faf3bcc7756e8a21a4c03ac2839c14a4c69f9912 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:14:06 +0000 Subject: [PATCH 03/10] tests: bootstrap parallel extension in spawned test scripts Co-Authored-By: Hermann D. Schimpf --- tests/ParallelTest.php | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/ParallelTest.php b/tests/ParallelTest.php index e7e09f5..58e9984 100644 --- a/tests/ParallelTest.php +++ b/tests/ParallelTest.php @@ -261,6 +261,10 @@ private function runWorkerScript(string $body): string { Date: Thu, 23 Jul 2026 21:19:01 +0000 Subject: [PATCH 04/10] fix: init console worker from Runner thread, not from task thread Co-Authored-By: Hermann D. Schimpf --- src/Internals/Runner/ManagesTasks.php | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/Internals/Runner/ManagesTasks.php b/src/Internals/Runner/ManagesTasks.php index 0511405..5a45e6b 100644 --- a/src/Internals/Runner/ManagesTasks.php +++ b/src/Internals/Runner/ManagesTasks.php @@ -74,6 +74,9 @@ private function startNextPendingTask(): void { $task = $this->tasks[$task_id = array_shift($this->pending_tasks)]; $task->setState(Task::STATE_Starting); + // ensure console output worker is available for this task + $this->initConsole(); + // process task inside a thread (if parallel extension is available) if (PARALLEL_EXT_LOADED) { // create starter channel to wait threads start event @@ -99,8 +102,7 @@ private function startNextPendingTask(): void { $worker->connectProgressBar($uuid, $GLOBALS['worker_thread_id'] ??= sprintf('%s@%s', $uuid, substr(md5(uniqid($worker_class, true)), 0, 16))); } - // initialize console output and connect worker to it - $this->initConsole(); + // connect worker to console output $worker->connectConsole($uuid); // notify that thread started From 7b2cf17f479e9a114ff9ba983c60f0b01d487be0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:35:28 +0000 Subject: [PATCH 05/10] fix: use dedicated Runtime for ConsoleWorker and reset state on stop Co-Authored-By: Hermann D. Schimpf --- src/Internals/Runner/HasSharedConsole.php | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/Internals/Runner/HasSharedConsole.php b/src/Internals/Runner/HasSharedConsole.php index 2e21de6..1d392b1 100644 --- a/src/Internals/Runner/HasSharedConsole.php +++ b/src/Internals/Runner/HasSharedConsole.php @@ -7,8 +7,8 @@ use parallel\Channel; use parallel\Events\Event; use parallel\Future; +use parallel\Runtime; use Symfony\Component\Console\Output\ConsoleOutput; -use parallel; trait HasSharedConsole { @@ -43,8 +43,8 @@ private function initConsole(): void { // already started if ($this->console_started) return; - // create a ConsoleWorker instance inside a thread - $this->console ??= parallel\run(static function(string $uuid): void { + // create a ConsoleWorker instance inside a dedicated thread + $this->console ??= (new Runtime(PARALLEL_AUTOLOADER))->run(static function(string $uuid): void { // create ConsoleWorker instance $console = new ConsoleWorker($uuid); // listen for events @@ -70,6 +70,9 @@ private function stopConsole(): void { $this->console_channel->send(Event\Type::Close); // wait until Console worker instance shutdowns $this->console_channel->receive(); + + $this->console_started = false; + $this->console_channel = null; } private function writeOutput(string $message, bool $newline = true): void { From 51d95a8105ba68b4f324a6733fdb52de260c9d49 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:41:58 +0000 Subject: [PATCH 06/10] fix: drop ConsoleWorker thread, write non-progress messages directly to STDERR Co-Authored-By: Hermann D. Schimpf --- CHANGELOG.md | 1 - README.md | 2 +- .../Commands/Output/WriteOutputMessage.php | 4 +- src/Internals/ConsoleWorker.php | 32 ---------- src/Internals/ConsoleWorker/HasChannels.php | 41 ------------- src/Internals/Runner.php | 2 - src/Internals/Runner/HasSharedConsole.php | 61 +------------------ src/Internals/Runner/ManagesTasks.php | 6 -- .../CommunicatesWithProgressBarWorker.php | 14 +---- 9 files changed, 7 insertions(+), 156 deletions(-) delete mode 100644 src/Internals/ConsoleWorker.php delete mode 100644 src/Internals/ConsoleWorker/HasChannels.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 998b513..6ee1f64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,6 @@ All notable changes to **parallel-sdk** are documented in this file. The format ### Added - `ParallelWorker::write()` and `ParallelWorker::writeln()` methods to emit console messages from workers without them being overwritten by the ProgressBar. -- `ConsoleWorker` thread that owns a fallback `ConsoleOutput` for workers that do not use a ProgressBar. - `WriteOutputMessage` command to route `write()`/`writeln()` calls through the existing channel infrastructure. ### Changed diff --git a/README.md b/README.md index 2b6f236..8ed634a 100644 --- a/README.md +++ b/README.md @@ -356,7 +356,7 @@ final class ExampleWorker extends ParallelWorker { Use `write()` or `writeln()` to emit ad-hoc console messages from a worker. When a ProgressBar is active, the bar is temporarily hidden, the message is printed, and the bar is redrawn below it so the message is not overwritten. -These methods also work for workers that did not enable `withProgress()`; messages are then routed to a dedicated console output handler. +These methods also work for workers that did not enable `withProgress()`; messages are then written directly to the console. ```php use HDSSolutions\Console\Parallel\ParallelWorker; diff --git a/src/Internals/Commands/Output/WriteOutputMessage.php b/src/Internals/Commands/Output/WriteOutputMessage.php index bb6b822..99a7202 100644 --- a/src/Internals/Commands/Output/WriteOutputMessage.php +++ b/src/Internals/Commands/Output/WriteOutputMessage.php @@ -6,9 +6,7 @@ /** * Message sent to {@see \HDSSolutions\Console\Parallel\Internals\ProgressBarWorker} - * or {@see \HDSSolutions\Console\Parallel\Internals\ConsoleWorker} to execute - * {@see \HDSSolutions\Console\Parallel\Internals\ProgressBarWorker::writeOutput()} - * or {@see \HDSSolutions\Console\Parallel\Internals\ConsoleWorker::writeOutput()}. + * to execute {@see \HDSSolutions\Console\Parallel\Internals\ProgressBarWorker::writeOutput()}. */ final readonly class WriteOutputMessage extends ParallelCommandMessage { diff --git a/src/Internals/ConsoleWorker.php b/src/Internals/ConsoleWorker.php deleted file mode 100644 index 8fb6978..0000000 --- a/src/Internals/ConsoleWorker.php +++ /dev/null @@ -1,32 +0,0 @@ -openChannels(); - $this->output = (new ConsoleOutput)->getErrorOutput(); - } - - public function afterListening(): void { - $this->closeChannels(); - } - - private function writeOutput(string $message, bool $newline = true): void { - $this->output->write($message, $newline); - } - -} diff --git a/src/Internals/ConsoleWorker/HasChannels.php b/src/Internals/ConsoleWorker/HasChannels.php deleted file mode 100644 index 55ed2b2..0000000 --- a/src/Internals/ConsoleWorker/HasChannels.php +++ /dev/null @@ -1,41 +0,0 @@ -console_channel = TwoWayChannel::make(self::class.'@'.$this->uuid); - } - - protected function recv(): mixed { - return $this->console_channel->receive(); - } - - protected function send(mixed $value): mixed { - return $this->console_channel->send($value); - } - - protected function release(): bool { - return $this->send(true); - } - - private function closeChannels(): void { - // gracefully join - $this->console_channel->send(false); - // close channel - $this->console_channel->close(); - } - -} diff --git a/src/Internals/Runner.php b/src/Internals/Runner.php index bd5d5c3..7435c67 100644 --- a/src/Internals/Runner.php +++ b/src/Internals/Runner.php @@ -34,7 +34,6 @@ public function __construct( protected function afterListening(): void { $this->stopEater(); $this->stopRunningTasks(); - $this->stopConsole(); $this->closeChannels(); } @@ -246,7 +245,6 @@ private function await(?int $wait_until = null): bool { } public function __destruct() { - $this->stopConsole(); $this->stopProgressBar(); } diff --git a/src/Internals/Runner/HasSharedConsole.php b/src/Internals/Runner/HasSharedConsole.php index 1d392b1..a7c7ab0 100644 --- a/src/Internals/Runner/HasSharedConsole.php +++ b/src/Internals/Runner/HasSharedConsole.php @@ -2,77 +2,20 @@ namespace HDSSolutions\Console\Parallel\Internals\Runner; -use HDSSolutions\Console\Parallel\Internals\Communication\TwoWayChannel; -use HDSSolutions\Console\Parallel\Internals\ConsoleWorker; -use parallel\Channel; -use parallel\Events\Event; -use parallel\Future; -use parallel\Runtime; use Symfony\Component\Console\Output\ConsoleOutput; trait HasSharedConsole { /** - * @var Future|ConsoleWorker|null Instance of the Console output worker + * @var ConsoleOutput|null Local ConsoleOutput used on non-threaded environments */ - private Future | ConsoleWorker | null $console = null; - - /** - * @var bool Flag to identify if Console output worker is already started - */ - private bool $console_started = false; - - /** - * @var TwoWayChannel|null Channel of communication with the Console output worker - */ - private ?TwoWayChannel $console_channel = null; - - /** - * @var ConsoleOutput Local ConsoleOutput used as a fallback on non-threaded environments - */ - private ConsoleOutput $consoleOutput; + private ?ConsoleOutput $consoleOutput = null; private function initConsole(): void { // on non-threaded environments, just initialize the local ConsoleOutput if (! PARALLEL_EXT_LOADED) { $this->consoleOutput ??= new ConsoleOutput(); - - return; } - - // already started - if ($this->console_started) return; - - // create a ConsoleWorker instance inside a dedicated thread - $this->console ??= (new Runtime(PARALLEL_AUTOLOADER))->run(static function(string $uuid): void { - // create ConsoleWorker instance - $console = new ConsoleWorker($uuid); - // listen for events - $console->listen(); - }, [ $this->uuid ]); - - // open communication channel with the Console worker - while ($this->console_channel === null) { - try { $this->console_channel = TwoWayChannel::open(ConsoleWorker::class.'@'.$this->uuid); - // wait 1ms if channel does not exist yet and retry - } catch (Channel\Error\Existence) { usleep(1_000); } - } - - // wait until Console worker starts - $this->console_channel->receive(); - $this->console_started = true; - } - - private function stopConsole(): void { - if (! PARALLEL_EXT_LOADED || ! $this->console_started) return; - - // stop Console worker instance - $this->console_channel->send(Event\Type::Close); - // wait until Console worker instance shutdowns - $this->console_channel->receive(); - - $this->console_started = false; - $this->console_channel = null; } private function writeOutput(string $message, bool $newline = true): void { diff --git a/src/Internals/Runner/ManagesTasks.php b/src/Internals/Runner/ManagesTasks.php index 5a45e6b..10c2efc 100644 --- a/src/Internals/Runner/ManagesTasks.php +++ b/src/Internals/Runner/ManagesTasks.php @@ -74,9 +74,6 @@ private function startNextPendingTask(): void { $task = $this->tasks[$task_id = array_shift($this->pending_tasks)]; $task->setState(Task::STATE_Starting); - // ensure console output worker is available for this task - $this->initConsole(); - // process task inside a thread (if parallel extension is available) if (PARALLEL_EXT_LOADED) { // create starter channel to wait threads start event @@ -102,9 +99,6 @@ private function startNextPendingTask(): void { $worker->connectProgressBar($uuid, $GLOBALS['worker_thread_id'] ??= sprintf('%s@%s', $uuid, substr(md5(uniqid($worker_class, true)), 0, 16))); } - // connect worker to console output - $worker->connectConsole($uuid); - // notify that thread started Channel::open(sprintf('starter@%s', $uuid))->send(true); diff --git a/src/Internals/Worker/CommunicatesWithProgressBarWorker.php b/src/Internals/Worker/CommunicatesWithProgressBarWorker.php index e5db495..bf7bd6f 100644 --- a/src/Internals/Worker/CommunicatesWithProgressBarWorker.php +++ b/src/Internals/Worker/CommunicatesWithProgressBarWorker.php @@ -5,7 +5,6 @@ use Closure; use HDSSolutions\Console\Parallel\Internals\Communication\TwoWayChannel; use HDSSolutions\Console\Parallel\Internals\Commands; -use HDSSolutions\Console\Parallel\Internals\ConsoleWorker; use HDSSolutions\Console\Parallel\Internals\ProgressBarWorker; use parallel\Channel; @@ -43,20 +42,13 @@ final public function connectProgressBar(string | Closure $uuid, string $identif } final public function connectConsole(string | Closure $uuid, string $identifier = null): bool { + // on non-threaded environments the Runner provides a closure that writes to ConsoleOutput if (! PARALLEL_EXT_LOADED) { $this->console_channel = $uuid; - - return true; - } - - // open channel if not already opened - while ($this->console_channel === null) { - // open channel to communicate with the Console output worker instance - try { $this->console_channel = TwoWayChannel::open(ConsoleWorker::class.'@'.$uuid); - // wait 1ms if channel does not exist yet and retry - } catch (Channel\Error\Existence) { usleep(1_000); } } + // on threaded environments there is no console coordinator to connect to: + // messages are written directly to STDERR when no progress bar is active return true; } From f3f8dcd07b7ce1923470a5306aa46c962c079964 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:50:39 +0000 Subject: [PATCH 07/10] test: disable opcache JIT before parallel tests to avoid thread races Co-Authored-By: Hermann D. Schimpf --- tests/ParallelTest.php | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/ParallelTest.php b/tests/ParallelTest.php index 58e9984..3bae9b8 100644 --- a/tests/ParallelTest.php +++ b/tests/ParallelTest.php @@ -12,6 +12,13 @@ final class ParallelTest extends TestCase { + public static function setUpBeforeClass(): void { + // The parallel extension is not compatible with OPcache JIT in threaded environments. + if (extension_loaded('parallel')) { + ini_set('opcache.jit', 'disable'); + } + } + public function testThatParallelExtensionIsAvailable(): void { // check that ext-parallel is available $this->assertTrue(extension_loaded('parallel'), 'Parallel extension isn\'t available'); From 19caddc96b68026db3f6e325d31b86263324bf7c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:55:52 +0000 Subject: [PATCH 08/10] test: disable opcache JIT via phpunit.xml to avoid parallel thread races Co-Authored-By: Hermann D. Schimpf --- phpunit.xml | 3 ++- tests/ParallelTest.php | 7 ------- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/phpunit.xml b/phpunit.xml index 31aace4..d22743e 100644 --- a/phpunit.xml +++ b/phpunit.xml @@ -20,6 +20,7 @@ - + + diff --git a/tests/ParallelTest.php b/tests/ParallelTest.php index 3bae9b8..58e9984 100644 --- a/tests/ParallelTest.php +++ b/tests/ParallelTest.php @@ -12,13 +12,6 @@ final class ParallelTest extends TestCase { - public static function setUpBeforeClass(): void { - // The parallel extension is not compatible with OPcache JIT in threaded environments. - if (extension_loaded('parallel')) { - ini_set('opcache.jit', 'disable'); - } - } - public function testThatParallelExtensionIsAvailable(): void { // check that ext-parallel is available $this->assertTrue(extension_loaded('parallel'), 'Parallel extension isn\'t available'); From 2a307fab8b8fea580e253d43a8b1128b58d1dae5 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 22:00:32 +0000 Subject: [PATCH 09/10] Revert phpunit.xml JIT override (cannot change opcache.jit at runtime) Co-Authored-By: Hermann D. Schimpf --- phpunit.xml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/phpunit.xml b/phpunit.xml index d22743e..31aace4 100644 --- a/phpunit.xml +++ b/phpunit.xml @@ -20,7 +20,6 @@ - - + From 53982e1c512b354d5a82013d621aa3d0a7bdf6a6 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 22:09:00 +0000 Subject: [PATCH 10/10] test: revert HasProgressBar and writeOutput to baseline to isolate hang Co-Authored-By: Hermann D. Schimpf --- src/Internals/ProgressBarWorker.php | 3 ++- src/Internals/ProgressBarWorker/HasProgressBar.php | 10 +--------- 2 files changed, 3 insertions(+), 10 deletions(-) diff --git a/src/Internals/ProgressBarWorker.php b/src/Internals/ProgressBarWorker.php index eefc1a2..515c18b 100644 --- a/src/Internals/ProgressBarWorker.php +++ b/src/Internals/ProgressBarWorker.php @@ -3,6 +3,7 @@ namespace HDSSolutions\Console\Parallel\Internals; use Symfony\Component\Console\Helper\Helper; +use Symfony\Component\Console\Output\ConsoleOutput; final class ProgressBarWorker { use ProgressBarWorker\HasChannels; @@ -67,7 +68,7 @@ private function progressBarAction(string $action, array $args): void { private function writeOutput(string $message, bool $newline = true): void { // clear the bar, write the message, then redraw the bar below it $this->progressBar->clear(); - $this->output->write($message, $newline); + (new ConsoleOutput)->getErrorOutput()->write($message, $newline); $this->progressBar->display(); } diff --git a/src/Internals/ProgressBarWorker/HasProgressBar.php b/src/Internals/ProgressBarWorker/HasProgressBar.php index 6110852..d35f822 100644 --- a/src/Internals/ProgressBarWorker/HasProgressBar.php +++ b/src/Internals/ProgressBarWorker/HasProgressBar.php @@ -4,7 +4,6 @@ use Symfony\Component\Console\Helper\ProgressBar; use Symfony\Component\Console\Output\ConsoleOutput; -use Symfony\Component\Console\Output\OutputInterface; trait HasProgressBar { @@ -18,15 +17,8 @@ trait HasProgressBar { */ private bool $progressBarStarted = false; - /** - * @var OutputInterface Output stream used for both the ProgressBar and messages - */ - private OutputInterface $output; - private function createProgressBar(): void { - // use the stderr output stream; ProgressBar uses the same stream internally - $this->output = (new ConsoleOutput)->getErrorOutput(); - $this->progressBar = new ProgressBar($this->output); + $this->progressBar = new ProgressBar(new ConsoleOutput); // configure ProgressBar settings $this->progressBar->setBarWidth(80);