From 6badba866a4c8cad24fd586686cdfc11766afac7 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Thu, 12 Nov 2026 17:15:13 +0700 Subject: [PATCH 1/5] [perf] Fork parallel workers from the main process when PHPStan Turbo is loaded --- .../LazyContainerFactory.php | 11 ++ .../Application/ParallelFileProcessor.php | 103 ++++++++++++-- .../Command/WorkerCommandLineFactory.php | 46 ++++++ src/Parallel/ValueObject/ParallelProcess.php | 131 +++++++++++++++++- .../Command/WorkerCommandLineFactoryTest.php | 28 ++++ .../ValueObject/ParallelProcessTest.php | 36 ++++- 6 files changed, 338 insertions(+), 17 deletions(-) diff --git a/src/DependencyInjection/LazyContainerFactory.php b/src/DependencyInjection/LazyContainerFactory.php index 8b37e0bfd87..0a27a8cb321 100644 --- a/src/DependencyInjection/LazyContainerFactory.php +++ b/src/DependencyInjection/LazyContainerFactory.php @@ -36,6 +36,8 @@ use Rector\NodeTypeResolver\DependencyInjection\PHPStanServicesFactory; use Rector\NodeTypeResolver\NodeTypeResolver; use Rector\NodeTypeResolver\Reflection\BetterReflection\SourceLocatorProvider\DynamicSourceLocatorProvider; +use Rector\Parallel\Application\ParallelFileProcessor; +use Rector\Parallel\Command\WorkerCommandLineFactory; use Rector\PhpAttribute\AnnotationToAttributeMapper; use Rector\PhpAttribute\AnnotationToAttributeMapper\ArrayAnnotationToAttributeMapper; use Rector\PhpAttribute\AnnotationToAttributeMapper\ArrayItemNodeAnnotationToAttributeMapper; @@ -123,6 +125,15 @@ private function registerConsole(RectorConfig $rectorConfig): void $rectorConfig->singleton(ProcessCommand::class); $rectorConfig->singleton(WorkerCommand::class); + + // lazy, as the worker command depends on the processor that forks it + $rectorConfig->singleton( + ParallelFileProcessor::class, + static fn (RectorConfig $rectorConfig): ParallelFileProcessor => new ParallelFileProcessor( + $rectorConfig->make(WorkerCommandLineFactory::class), + static fn (): WorkerCommand => $rectorConfig->make(WorkerCommand::class) + ) + ); $rectorConfig->singleton(ValidateConfigCommand::class); $rectorConfig->singleton(AllRectorRulesProvider::class); $rectorConfig->singleton(ListRulesCommand::class); diff --git a/src/Parallel/Application/ParallelFileProcessor.php b/src/Parallel/Application/ParallelFileProcessor.php index ddb7d1fe232..90fc1c49777 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -6,13 +6,17 @@ use Clue\React\NDJson\Decoder; use Clue\React\NDJson\Encoder; +use Nette\Utils\FileSystem; use Nette\Utils\Random; +use PHPStan\Turbo\TurboExtensionEnabler; +use PHPStan\Turbo\TurboExtensionSelector; use React\EventLoop\StreamSelectLoop; use React\Socket\ConnectionInterface; use React\Socket\TcpServer; use Rector\Configuration\Option; use Rector\Configuration\Parameter\SimpleParameterProvider; use Rector\Console\Command\ProcessCommand; +use Rector\Console\Command\WorkerCommand; use Rector\Parallel\Command\WorkerCommandLineFactory; use Rector\Parallel\Enum\Action; use Rector\Parallel\Enum\Content; @@ -23,16 +27,22 @@ use Rector\Parallel\ValueObject\ParallelProcess; use Rector\Parallel\ValueObject\ProcessPool; use Rector\Parallel\ValueObject\Schedule; +use Rector\Testing\PHPUnit\StaticPHPUnitEnvironment; use Rector\ValueObject\Error\SystemError; use Rector\ValueObject\ProcessResult; use Symfony\Component\Console\Command\Command; +use Symfony\Component\Console\Input\ArrayInput; use Symfony\Component\Console\Input\InputInterface; +use Symfony\Component\Console\Output\ConsoleOutput; use Throwable; /** * Workers process their own balanced bucket first, and are replaced by a fresh one bound to the same bucket * after a few chunks to free memory. Idle workers take pending chunks from the largest remaining queue. * Each chunk advances progress and re-arms the per-job timeout. + * + * Where possible a worker is forked from the main process, so it starts with the container the main process + * already booted; otherwise it is spawned as a new process that boots on its own. */ final class ParallelFileProcessor { @@ -46,9 +56,19 @@ final class ParallelFileProcessor private ProcessPool $processPool; + /** + * @var callable(): WorkerCommand + */ + private $workerCommandResolver; + + /** + * @param callable(): WorkerCommand $workerCommandResolver Lazy, as the worker command itself depends on this class + */ public function __construct( private readonly WorkerCommandLineFactory $workerCommandLineFactory, + callable $workerCommandResolver, ) { + $this->workerCommandResolver = $workerCommandResolver; } /** @@ -148,6 +168,13 @@ public function process( $timeoutInSeconds = SimpleParameterProvider::provideIntParameter(Option::PARALLEL_JOB_TIMEOUT_IN_SECONDS); $fileChunksBudgetPerProcess = []; + $canForkWorkers = $this->canForkWorkers(); + + // forked workers log their PHP errors next to the cache + $cacheDirectory = SimpleParameterProvider::provideStringParameter(Option::CACHE_DIR); + if ($canForkWorkers) { + FileSystem::createDir($cacheDirectory); + } $processSpawner = function (int $bucketKey) use ( $parallelResultCollector, @@ -162,6 +189,8 @@ public function process( $streamSelectLoop, $timeoutInSeconds, $handleErrorCallable, + $canForkWorkers, + $cacheDirectory, &$fileChunksBudgetPerProcess, &$processSpawner, &$processedFilesCount @@ -170,16 +199,26 @@ public function process( $bucketKeyByIdentifier[$processIdentifier] = $bucketKey; $fileChunksBudgetPerProcess[$processIdentifier] = self::MAX_CHUNKS_PER_WORKER; - $workerCommandLine = $this->workerCommandLineFactory->create( - $mainScript, - ProcessCommand::class, - 'worker', - $input, - $processIdentifier, - $serverPort, - ); + if ($canForkWorkers) { + $parallelProcess = new ParallelProcess( + '', + $streamSelectLoop, + $timeoutInSeconds, + $this->createForkedWorker($input, $processIdentifier, $serverPort), + $cacheDirectory + ); + } else { + $workerCommandLine = $this->workerCommandLineFactory->create( + $mainScript, + ProcessCommand::class, + 'worker', + $input, + $processIdentifier, + $serverPort, + ); - $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); + $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); + } $parallelProcess->start( // 1. callable on data @@ -286,6 +325,52 @@ function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIden return $parallelResultCollector->createProcessResult(); } + /** + * Forking skips starting PHP and booting the container again for every worker, including the ones that + * replace a recycled worker. A fork runs on what the main process has loaded, so it is only used when a + * spawned worker would not get more than that - PHPStan Turbo on its command line. + */ + private function canForkWorkers(): bool + { + if (StaticPHPUnitEnvironment::isPHPUnitRun()) { + return false; + } + + if (! extension_loaded('pcntl') || ! extension_loaded('posix')) { + return false; + } + + $disabledFunctions = (string) ini_get('disable_functions'); + if (str_contains($disabledFunctions, 'pcntl_') || str_contains($disabledFunctions, 'posix_kill')) { + return false; + } + + return TurboExtensionEnabler::isLoaded() || TurboExtensionSelector::findExtensionForWorkers() === null; + } + + /** + * @return callable(): int + */ + private function createForkedWorker(InputInterface $input, string $processIdentifier, int $serverPort): callable + { + $workerInput = new ArrayInput($this->workerCommandLineFactory->createInputParameters( + ProcessCommand::class, + $input, + $processIdentifier, + $serverPort + )); + + return function () use ($workerInput): int { + // the main process collects PHP errors on its own while it waits for the workers; + // a worker leaves them to PHP, the same as a spawned one + restore_error_handler(); + + $workerCommand = ($this->workerCommandResolver)(); + + return $workerCommand->run($workerInput, new ConsoleOutput()); + }; + } + /** * @param array>> $jobsPerWorker * @param array $bucketKeyByIdentifier diff --git a/src/Parallel/Command/WorkerCommandLineFactory.php b/src/Parallel/Command/WorkerCommandLineFactory.php index 99161845e71..b622ce68563 100644 --- a/src/Parallel/Command/WorkerCommandLineFactory.php +++ b/src/Parallel/Command/WorkerCommandLineFactory.php @@ -131,6 +131,52 @@ public function create( return implode(' ', $workerCommandArray); } + /** + * The input of the worker command for a worker that is forked from the main process instead of spawned; + * carries the same options as the command line of @see create() + * + * @param class-string $mainCommandClass + * @return array + */ + public function createInputParameters( + string $mainCommandClass, + InputInterface $input, + string $identifier, + int $port + ): array { + $mainCommand = $this->commandFromReflectionFactory->create($mainCommandClass); + + $parameters = []; + foreach ($this->getCommandOptionNames($mainCommand) as $mainCommandOptionName) { + if ($this->shouldSkipOption($input, $mainCommandOptionName)) { + continue; + } + + $optionValue = $input->getOption($mainCommandOptionName); + + // skip clutter + if ($optionValue === null || $optionValue === false) { + continue; + } + + $parameters[self::OPTION_DASHES . $mainCommandOptionName] = $optionValue; + } + + $parameters[self::OPTION_DASHES . Option::PARALLEL_PORT] = (string) $port; + $parameters[self::OPTION_DASHES . Option::PARALLEL_IDENTIFIER] = $identifier; + $parameters[Option::SOURCE] = $input->getArgument(Option::SOURCE); + $parameters[self::OPTION_DASHES . Option::OUTPUT_FORMAT] = JsonOutputFormatter::NAME; + + if ($input->hasOption(Option::CONFIG)) { + $configValue = $input->getOption(Option::CONFIG); + if (is_string($configValue) && $configValue !== '') { + $parameters[self::OPTION_DASHES . Option::CONFIG] = $configValue; + } + } + + return $parameters; + } + private function shouldSkipOption(InputInterface $input, string $optionName): bool { if (! $input->hasOption($optionName)) { diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php index 131fd4f03fc..f15fe73acac 100644 --- a/src/Parallel/ValueObject/ParallelProcess.php +++ b/src/Parallel/ValueObject/ParallelProcess.php @@ -7,6 +7,7 @@ use Clue\React\NDJson\Decoder; use Clue\React\NDJson\Encoder; use Exception; +use Nette\Utils\FileSystem; use React\ChildProcess\Process; use React\EventLoop\LoopInterface; use React\EventLoop\TimerInterface; @@ -44,11 +45,30 @@ final class ParallelProcess private ?TimerInterface $timer = null; + private ?int $forkedProcessId = null; + + private ?TimerInterface $forkedExitTimer = null; + + /** + * @var (callable(): int)|null + */ + private $forkedWorker; + + /** + * @param (callable(): int)|null $forkedWorker Runs the worker in a fork of the current process and returns + * its exit code. The fork starts with the container and the classes the main process already loaded, + * instead of booting them again; the command is not used then. + * @param string|null $forkedErrorLogDirectory Where a forked worker logs its PHP errors, the system + * temp directory by default + */ public function __construct( private readonly string $command, private readonly LoopInterface $loop, - private readonly int $timetoutInSeconds + private readonly int $timetoutInSeconds, + ?callable $forkedWorker = null, + private readonly ?string $forkedErrorLogDirectory = null ) { + $this->forkedWorker = $forkedWorker; } /** @@ -58,6 +78,13 @@ public function __construct( */ public function start(callable $onData, callable $onError, callable $onExit): void { + if ($this->forkedWorker !== null) { + $this->onData = $onData; + $this->onError = $onError; + $this->startForked($onExit); + return; + } + $tmp = tmpfile(); if ($tmp === false) { throw new ParallelShouldNotHappenException('Failed creating temp file.'); @@ -105,7 +132,7 @@ public function request(array $data): void $this->encoder->write($data); $this->timer = $this->loop->addTimer($this->timetoutInSeconds, function (): void { // a worker that does not answer in time cannot be asked to stop either - $this->process->terminate(); + $this->terminate(); $onError = $this->onError; @@ -116,18 +143,20 @@ public function request(array $data): void public function quit(): void { - if (! $this->process->isRunning()) { + if (! $this->isRunning()) { return; } - foreach ($this->process->pipes as $pipe) { - $pipe->close(); + if (isset($this->process)) { + foreach ($this->process->pipes as $pipe) { + $pipe->close(); + } } // the process can be quit before its connection is bound, e.g. on quitAll() after an error; // such a worker cannot be asked to stop if (! isset($this->encoder)) { - $this->process->terminate(); + $this->terminate(); return; } @@ -159,6 +188,96 @@ public function bindConnection(Decoder $decoder, Encoder $encoder): void }); } + /** + * @param callable(?int $onExit, string $output) : void $onExit + */ + private function startForked(callable $onExit): void + { + $errorLogFilePath = tempnam($this->forkedErrorLogDirectory ?? sys_get_temp_dir(), 'rector-worker-'); + if ($errorLogFilePath === false) { + throw new ParallelShouldNotHappenException('Failed creating the error log file of a worker.'); + } + + $processId = pcntl_fork(); + if ($processId === -1) { + throw new ParallelShouldNotHappenException('Failed forking a worker process.'); + } + + if ($processId === 0) { + $this->runForkedWorker($errorLogFilePath); + } + + $this->forkedProcessId = $processId; + + $this->forkedExitTimer = $this->loop->addPeriodicTimer(0.01, function () use ( + $processId, + $errorLogFilePath, + $onExit + ): void { + $status = 0; + if (pcntl_waitpid($processId, $status, WNOHANG) === 0) { + return; + } + + $this->forkedProcessId = null; + $this->cancelTimer(); + + if ($this->forkedExitTimer instanceof TimerInterface) { + $this->loop->cancelTimer($this->forkedExitTimer); + $this->forkedExitTimer = null; + } + + // killed by a signal, e.g. after a timeout: no exit code, same as a spawned worker + $exitCode = pcntl_wifexited($status) ? pcntl_wexitstatus($status) : null; + $onExit($exitCode === false ? null : $exitCode, $this->readErrorLog($errorLogFilePath)); + }); + } + + /** + * Runs in the fork. A spawned worker has its stderr captured; a fork shares the stderr of the main + * process, so PHP logs its errors to a file instead. + */ + private function runForkedWorker(string $errorLogFilePath): never + { + ini_set('display_errors', '0'); + ini_set('log_errors', '1'); + ini_set('error_log', $errorLogFilePath); + + $forkedWorker = $this->forkedWorker; + + // never return to the code of the main process this fork was started from + exit($forkedWorker === null ? 1 : $forkedWorker()); + } + + private function readErrorLog(string $errorLogFilePath): string + { + $errorLog = FileSystem::read($errorLogFilePath); + FileSystem::delete($errorLogFilePath); + + return $errorLog; + } + + private function isRunning(): bool + { + if ($this->forkedWorker !== null) { + return $this->forkedProcessId !== null; + } + + return $this->process->isRunning(); + } + + private function terminate(): void + { + if ($this->forkedProcessId !== null) { + posix_kill($this->forkedProcessId, SIGTERM); + return; + } + + if (isset($this->process)) { + $this->process->terminate(); + } + } + private function cancelTimer(): void { if (! $this->timer instanceof TimerInterface) { diff --git a/tests/Parallel/Command/WorkerCommandLineFactoryTest.php b/tests/Parallel/Command/WorkerCommandLineFactoryTest.php index 5b7d0f7d6eb..3557218b924 100644 --- a/tests/Parallel/Command/WorkerCommandLineFactoryTest.php +++ b/tests/Parallel/Command/WorkerCommandLineFactoryTest.php @@ -198,6 +198,34 @@ public static function provideData(): Iterator ]; } + public function testInputParametersOfForkedWorker(): void + { + $arrayInput = new ArrayInput([ + self::COMMAND => 'process', + Option::SOURCE => ['src'], + '--' . Option::DRY_RUN => true, + '--' . Option::OUTPUT_FORMAT => ConsoleOutputFormatter::NAME, + '--' . Option::ONLY => ['FirstRector', 'SecondRector'], + ], $this->prepareProcessCommandDefinition()); + + $inputParameters = $this->workerCommandLineFactory->createInputParameters( + ProcessCommand::class, + $arrayInput, + 'identifier', + 2000 + ); + + // the same options the command line of a spawned worker carries; unset ones are left out + $this->assertSame([ + '--dry-run' => true, + '--only' => ['FirstRector', 'SecondRector'], + '--port' => '2000', + '--identifier' => 'identifier', + 'source' => ['src'], + '--output-format' => 'json', + ], $inputParameters); + } + /** * The worker gets PHPStan Turbo through two -d options, when PHPStan ships a binary for this runtime */ diff --git a/tests/Parallel/ValueObject/ParallelProcessTest.php b/tests/Parallel/ValueObject/ParallelProcessTest.php index f8c296c4857..e67fc944a4a 100644 --- a/tests/Parallel/ValueObject/ParallelProcessTest.php +++ b/tests/Parallel/ValueObject/ParallelProcessTest.php @@ -68,6 +68,29 @@ public function testTimeoutTerminatesWorker(): void $this->assertSame(['Child process timed out after 1 seconds'], $this->errorMessages); } + #[RequiresFunction('pcntl_fork')] + public function testTimeoutTerminatesForkedWorker(): void + { + $parallelProcess = $this->startWorker(true); + $this->bindConnection($parallelProcess); + + $parallelProcess->request([]); + + $this->assertWorkerIsTerminated(); + $this->assertSame(['Child process timed out after 1 seconds'], $this->errorMessages); + } + + #[RequiresFunction('pcntl_fork')] + public function testQuitTerminatesForkedWorkerThatIsNotConnected(): void + { + $parallelProcess = $this->startWorker(true); + + $parallelProcess->quit(); + + $this->assertWorkerIsTerminated(); + $this->assertSame([], $this->errorMessages); + } + public function testQuitKeepsTimeoutOfBusyWorker(): void { $parallelProcess = $this->startWorker(); @@ -83,8 +106,17 @@ public function testQuitKeepsTimeoutOfBusyWorker(): void /** * Starts a worker that neither connects nor answers, like one that is still booting or stuck in an endless loop */ - private function startWorker(): ParallelProcess + private function startWorker(bool $isForked = false): ParallelProcess { + $pidFile = $this->pidFile; + $forkedWorker = $isForked ? static function () use ($pidFile): int { + file_put_contents($pidFile, getmypid()); + sleep(10); + + return 0; + } + : null; + $command = sprintf( '%s -r %s -- %s', escapeshellarg(PHP_BINARY), @@ -92,7 +124,7 @@ private function startWorker(): ParallelProcess escapeshellarg($this->pidFile) ); - $parallelProcess = new ParallelProcess($command, $this->streamSelectLoop, 1); + $parallelProcess = new ParallelProcess($command, $this->streamSelectLoop, 1, $forkedWorker); $parallelProcess->start( static function (): void { }, From a0e651576379a0fd9630c5df2c478dd3e79d15e7 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Thu, 12 Nov 2026 17:35:29 +0700 Subject: [PATCH 2/5] add comment why ini_set() is needed --- src/Parallel/ValueObject/ParallelProcess.php | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php index f15fe73acac..80ad65686b9 100644 --- a/src/Parallel/ValueObject/ParallelProcess.php +++ b/src/Parallel/ValueObject/ParallelProcess.php @@ -239,7 +239,10 @@ private function startForked(callable $onExit): void */ private function runForkedWorker(string $errorLogFilePath): never { + // bin/rector.php displays errors on stderr, which would print a crash of the fork on the terminal ini_set('display_errors', '0'); + + // off in some php.ini files; the error log stays empty then and the reason of a crash is lost ini_set('log_errors', '1'); ini_set('error_log', $errorLogFilePath); From 43e8b4bb3b5c4a2e7f6f9bb42e521a1e2ac0266b Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Thu, 12 Nov 2026 17:45:04 +0700 Subject: [PATCH 3/5] clean up: make use of cacheDirectory for tempnam() usage --- .../Application/ParallelFileProcessor.php | 11 ++++++--- src/Parallel/ValueObject/ParallelProcess.php | 8 +++---- .../ValueObject/ParallelProcessTest.php | 24 ++++++++++++++++--- 3 files changed, 32 insertions(+), 11 deletions(-) diff --git a/src/Parallel/Application/ParallelFileProcessor.php b/src/Parallel/Application/ParallelFileProcessor.php index 90fc1c49777..0225a66203e 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -204,8 +204,8 @@ public function process( '', $streamSelectLoop, $timeoutInSeconds, - $this->createForkedWorker($input, $processIdentifier, $serverPort), - $cacheDirectory + $cacheDirectory, + $this->createForkedWorker($input, $processIdentifier, $serverPort) ); } else { $workerCommandLine = $this->workerCommandLineFactory->create( @@ -217,7 +217,12 @@ public function process( $serverPort, ); - $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); + $parallelProcess = new ParallelProcess( + $workerCommandLine, + $streamSelectLoop, + $timeoutInSeconds, + $cacheDirectory + ); } $parallelProcess->start( diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php index 80ad65686b9..eeb7636fa4c 100644 --- a/src/Parallel/ValueObject/ParallelProcess.php +++ b/src/Parallel/ValueObject/ParallelProcess.php @@ -58,15 +58,13 @@ final class ParallelProcess * @param (callable(): int)|null $forkedWorker Runs the worker in a fork of the current process and returns * its exit code. The fork starts with the container and the classes the main process already loaded, * instead of booting them again; the command is not used then. - * @param string|null $forkedErrorLogDirectory Where a forked worker logs its PHP errors, the system - * temp directory by default */ public function __construct( private readonly string $command, private readonly LoopInterface $loop, private readonly int $timetoutInSeconds, - ?callable $forkedWorker = null, - private readonly ?string $forkedErrorLogDirectory = null + private readonly string $cacheDirectory, + ?callable $forkedWorker = null ) { $this->forkedWorker = $forkedWorker; } @@ -193,7 +191,7 @@ public function bindConnection(Decoder $decoder, Encoder $encoder): void */ private function startForked(callable $onExit): void { - $errorLogFilePath = tempnam($this->forkedErrorLogDirectory ?? sys_get_temp_dir(), 'rector-worker-'); + $errorLogFilePath = tempnam($this->cacheDirectory, 'rector-worker-'); if ($errorLogFilePath === false) { throw new ParallelShouldNotHappenException('Failed creating the error log file of a worker.'); } diff --git a/tests/Parallel/ValueObject/ParallelProcessTest.php b/tests/Parallel/ValueObject/ParallelProcessTest.php index e67fc944a4a..effbb0efd7f 100644 --- a/tests/Parallel/ValueObject/ParallelProcessTest.php +++ b/tests/Parallel/ValueObject/ParallelProcessTest.php @@ -6,20 +6,25 @@ use Clue\React\NDJson\Decoder; use Clue\React\NDJson\Encoder; +use Nette\Utils\FileSystem; use PHPUnit\Framework\Attributes\RequiresFunction; -use PHPUnit\Framework\TestCase; use React\EventLoop\StreamSelectLoop; use React\Stream\DuplexResourceStream; +use Rector\Configuration\Option; +use Rector\Configuration\Parameter\SimpleParameterProvider; use Rector\Parallel\ValueObject\ParallelProcess; +use Rector\Testing\PHPUnit\AbstractLazyTestCase; use Throwable; #[RequiresFunction('posix_kill')] -final class ParallelProcessTest extends TestCase +final class ParallelProcessTest extends AbstractLazyTestCase { private StreamSelectLoop $streamSelectLoop; private string $pidFile; + private string $cacheDirectory; + /** * @var string[] */ @@ -34,6 +39,13 @@ final class ParallelProcessTest extends TestCase protected function setUp(): void { + parent::setUp(); + + // the container sets the default cache directory + self::getContainer(); + $this->cacheDirectory = SimpleParameterProvider::provideStringParameter(Option::CACHE_DIR); + FileSystem::createDir($this->cacheDirectory); + $this->streamSelectLoop = new StreamSelectLoop(); $this->pidFile = (string) tempnam(sys_get_temp_dir(), 'rector_parallel_process_test'); } @@ -124,7 +136,13 @@ private function startWorker(bool $isForked = false): ParallelProcess escapeshellarg($this->pidFile) ); - $parallelProcess = new ParallelProcess($command, $this->streamSelectLoop, 1, $forkedWorker); + $parallelProcess = new ParallelProcess( + $command, + $this->streamSelectLoop, + 1, + $this->cacheDirectory, + $forkedWorker + ); $parallelProcess->start( static function (): void { }, From 4b241b2f8e5f3cb36613666ae7b1c7c254cd5c35 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Thu, 12 Nov 2026 17:55:24 +0700 Subject: [PATCH 4/5] fix: handle STDERR --- src/Parallel/ValueObject/ParallelProcess.php | 5 +++ .../ValueObject/ParallelProcessTest.php | 35 +++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php index eeb7636fa4c..e9ffbf61a24 100644 --- a/src/Parallel/ValueObject/ParallelProcess.php +++ b/src/Parallel/ValueObject/ParallelProcess.php @@ -16,6 +16,7 @@ use Rector\Parallel\Enum\ReactCommand; use Rector\Parallel\Enum\ReactEvent; use Rector\Parallel\Exception\ParallelShouldNotHappenException; +use Rector\Parallel\Stream\StderrToFileFilter; use Throwable; /** @@ -244,6 +245,10 @@ private function runForkedWorker(string $errorLogFilePath): never ini_set('log_errors', '1'); ini_set('error_log', $errorLogFilePath); + // what the worker itself writes to STDERR, e.g. a console error message + stream_filter_register(StderrToFileFilter::NAME, StderrToFileFilter::class); + stream_filter_append(STDERR, StderrToFileFilter::NAME, STREAM_FILTER_WRITE, $errorLogFilePath); + $forkedWorker = $this->forkedWorker; // never return to the code of the main process this fork was started from diff --git a/tests/Parallel/ValueObject/ParallelProcessTest.php b/tests/Parallel/ValueObject/ParallelProcessTest.php index effbb0efd7f..c80ff514de2 100644 --- a/tests/Parallel/ValueObject/ParallelProcessTest.php +++ b/tests/Parallel/ValueObject/ParallelProcessTest.php @@ -103,6 +103,41 @@ public function testQuitTerminatesForkedWorkerThatIsNotConnected(): void $this->assertSame([], $this->errorMessages); } + #[RequiresFunction('pcntl_fork')] + public function testForkedWorkerStderrIsCaptured(): void + { + $exitCode = null; + $errorOutput = null; + + $parallelProcess = new ParallelProcess( + '', + $this->streamSelectLoop, + 1, + $this->cacheDirectory, + static function (): int { + fwrite(STDERR, 'worker failed'); + + return 3; + } + ); + $parallelProcess->start( + static function (): void { + }, + static function (): void { + }, + function (?int $workerExitCode, string $workerErrorOutput) use (&$exitCode, &$errorOutput): void { + $exitCode = $workerExitCode; + $errorOutput = $workerErrorOutput; + $this->hasExited = true; + } + ); + + $this->runLoopUntil(fn (): bool => $this->hasExited); + + $this->assertSame(3, $exitCode); + $this->assertSame('worker failed', $errorOutput); + } + public function testQuitKeepsTimeoutOfBusyWorker(): void { $parallelProcess = $this->startWorker(); From bd313f93afa5a0edc7508d8404bad628a1eb7b30 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Thu, 12 Nov 2026 17:55:27 +0700 Subject: [PATCH 5/5] fix: handle STDERR --- src/Parallel/Stream/StderrToFileFilter.php | 36 ++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 src/Parallel/Stream/StderrToFileFilter.php diff --git a/src/Parallel/Stream/StderrToFileFilter.php b/src/Parallel/Stream/StderrToFileFilter.php new file mode 100644 index 00000000000..6169a1d42a3 --- /dev/null +++ b/src/Parallel/Stream/StderrToFileFilter.php @@ -0,0 +1,36 @@ +datalen; + file_put_contents((string) $this->params, $bucket->data, FILE_APPEND); + } + + // nothing is passed on to the stream itself + return PSFS_FEED_ME; + } +}