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..0225a66203e 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,31 @@ public function process( $bucketKeyByIdentifier[$processIdentifier] = $bucketKey; $fileChunksBudgetPerProcess[$processIdentifier] = self::MAX_CHUNKS_PER_WORKER; - $workerCommandLine = $this->workerCommandLineFactory->create( - $mainScript, - ProcessCommand::class, - 'worker', - $input, - $processIdentifier, - $serverPort, - ); - - $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); + if ($canForkWorkers) { + $parallelProcess = new ParallelProcess( + '', + $streamSelectLoop, + $timeoutInSeconds, + $cacheDirectory, + $this->createForkedWorker($input, $processIdentifier, $serverPort) + ); + } else { + $workerCommandLine = $this->workerCommandLineFactory->create( + $mainScript, + ProcessCommand::class, + 'worker', + $input, + $processIdentifier, + $serverPort, + ); + + $parallelProcess = new ParallelProcess( + $workerCommandLine, + $streamSelectLoop, + $timeoutInSeconds, + $cacheDirectory + ); + } $parallelProcess->start( // 1. callable on data @@ -286,6 +330,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/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; + } +} diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php index 131fd4f03fc..e9ffbf61a24 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; @@ -15,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; /** @@ -44,11 +46,28 @@ 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. + */ public function __construct( private readonly string $command, private readonly LoopInterface $loop, - private readonly int $timetoutInSeconds + private readonly int $timetoutInSeconds, + private readonly string $cacheDirectory, + ?callable $forkedWorker = null ) { + $this->forkedWorker = $forkedWorker; } /** @@ -58,6 +77,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 +131,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 +142,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 +187,103 @@ 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->cacheDirectory, '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 + { + // 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); + + // 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 + 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..c80ff514de2 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'); } @@ -68,6 +80,64 @@ 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); + } + + #[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(); @@ -83,8 +153,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 +171,13 @@ private function startWorker(): ParallelProcess escapeshellarg($this->pidFile) ); - $parallelProcess = new ParallelProcess($command, $this->streamSelectLoop, 1); + $parallelProcess = new ParallelProcess( + $command, + $this->streamSelectLoop, + 1, + $this->cacheDirectory, + $forkedWorker + ); $parallelProcess->start( static function (): void { },