[perf] Fork parallel workers from the main process when PHPStan Turbo is loaded

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);

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<int, array<int, array<string>>> $jobsPerWorker
      * @param array<string, int> $bucketKeyByIdentifier

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<Command> $mainCommandClass
+     * @return array<string, mixed>
+     */
+    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)) {

src/Parallel/Stream/StderrToFileFilter.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Rector\Parallel\Stream;
+
+use php_user_filter;
+
+/**
+ * Moves what is written to a stream into the file given as the filter parameter, instead of the stream.
+ *
+ * A forked worker shares the stderr of the main process and PHP cannot redirect it, so this is attached
+ * to STDERR to collect the output the same way a spawned worker has its stderr captured.
+ *
+ * @see \Rector\Parallel\ValueObject\ParallelProcess::runForkedWorker()
+ */
+final class StderrToFileFilter extends php_user_filter
+{
+    public const string NAME = 'rector.stderr_to_file';
+
+    /**
+     * @param resource $in
+     * @param resource $out
+     * @param int|null $consumed
+     */
+    public function filter($in, $out, &$consumed, bool $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            file_put_contents((string) $this->params, $bucket->data, FILE_APPEND);
+        }
+
+        // nothing is passed on to the stream itself
+        return PSFS_FEED_ME;
+    }
+}