diff --git a/CHANGELOG.md b/CHANGELOG.md index d871b69..6ee1f64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ 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. +- `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..8ed634a 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,44 @@ 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 written directly to the console. + +```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; + } + +} +``` + +#### 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) diff --git a/src/Internals/Commands/Output/WriteOutputMessage.php b/src/Internals/Commands/Output/WriteOutputMessage.php new file mode 100644 index 0000000..99a7202 --- /dev/null +++ b/src/Internals/Commands/Output/WriteOutputMessage.php @@ -0,0 +1,21 @@ +progressBar->clear(); + (new ConsoleOutput)->getErrorOutput()->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/Runner.php b/src/Internals/Runner.php index 0d48d07..7435c67 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; diff --git a/src/Internals/Runner/HasSharedConsole.php b/src/Internals/Runner/HasSharedConsole.php new file mode 100644 index 0000000..a7c7ab0 --- /dev/null +++ b/src/Internals/Runner/HasSharedConsole.php @@ -0,0 +1,26 @@ +consoleOutput ??= new ConsoleOutput(); + } + } + + 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..10c2efc 100644 --- a/src/Internals/Runner/ManagesTasks.php +++ b/src/Internals/Runner/ManagesTasks.php @@ -141,6 +141,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 +154,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..bf7bd6f 100644 --- a/src/Internals/Worker/CommunicatesWithProgressBarWorker.php +++ b/src/Internals/Worker/CommunicatesWithProgressBarWorker.php @@ -15,6 +15,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 +41,17 @@ final public function connectProgressBar(string | Closure $uuid, string $identif return true; } + 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; + } + + // 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; + } + final public function setMessage(string $message, string $name = 'message'): void { $this->newProgressBarAction(__FUNCTION__, $message, $name); } @@ -56,6 +72,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..58e9984 100644 --- a/tests/ParallelTest.php +++ b/tests/ParallelTest.php @@ -226,4 +226,74 @@ 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); + } + }