From c65e4cf06d719d3d02e3e26bb5b349bf1a45fc1a Mon Sep 17 00:00:00 2001 From: Tomas Votruba Date: Fri, 23 Oct 2026 11:54:19 +0200 Subject: [PATCH 1/2] Make LPT parallel scheduler the default, drop --lpt opt-in flag The LPT (Longest Processing Time first) bucket scheduler is now the only parallel scheduler. It balances files across workers by size up front and keeps each worker alive for its whole bucket, preserving warm caches. - promote LptScheduleFactory to ScheduleFactory - promote ExperimentalParallelFileProcessor to ParallelFileProcessor - move BucketSchedule out of the Experimental namespace - remove the old shared-pool ScheduleFactory, ParallelFileProcessor and Schedule - drop the --lpt CLI option and its CI usages --- .github/workflows/build_scoped_rector.yaml | 2 +- .github/workflows/rector.yaml | 2 +- phpstan.neon | 2 +- src/Application/ApplicationFileProcessor.php | 27 +- src/Configuration/Option.php | 6 - src/Console/ProcessConfigureDecorator.php | 7 - .../Application/ParallelFileProcessor.php | 180 ++++++---- .../ExperimentalParallelFileProcessor.php | 311 ------------------ .../Experimental/LptScheduleFactory.php | 87 ----- src/Parallel/ScheduleFactory.php | 74 ++++- .../ValueObject/BucketSchedule.php | 7 +- src/Parallel/ValueObject/Schedule.php | 34 -- 12 files changed, 179 insertions(+), 560 deletions(-) delete mode 100644 src/Parallel/Experimental/ExperimentalParallelFileProcessor.php delete mode 100644 src/Parallel/Experimental/LptScheduleFactory.php rename src/Parallel/{Experimental => }/ValueObject/BucketSchedule.php (60%) delete mode 100644 src/Parallel/ValueObject/Schedule.php diff --git a/.github/workflows/build_scoped_rector.yaml b/.github/workflows/build_scoped_rector.yaml index 642f0f69ec0..80c0b2e4ade 100644 --- a/.github/workflows/build_scoped_rector.yaml +++ b/.github/workflows/build_scoped_rector.yaml @@ -64,7 +64,7 @@ jobs: - run: rm -rf rector-build/rules-tests rector-build/templates rector-build/tests rector-build/scripts/validate-phpstan-version.php rector-build/vendor/tracy/tracy/examples rector-build/vendor/symfony/console/Tester rector-build/vendor/symfony/console/Event rector-build/vendor/symfony/console/EventListener rector-build/vendor/tracy/tracy/examples rector-build/vendor/tracy/tracy/src/Bridges rector-build/vendor/tracy/tracy/src/Tracy/Bar rector-build/vendor/tracy/tracy/src/Tracy/Session rector-build/vendor/symfony/service-contracts/Test # 2. downgrade rector - - run: php -d memory_limit=-1 bin/rector process rector-build/bin rector-build/config rector-build/src rector-build/rules rector-build/vendor --config build/config/config-downgrade.php --ansi --no-diffs --lpt + - run: php -d memory_limit=-1 bin/rector process rector-build/bin rector-build/config rector-build/src rector-build/rules rector-build/vendor --config build/config/config-downgrade.php --ansi --no-diffs # 3. prefix classes - run: sh build/build-rector-scoped.sh rector-build rector-prefixed-downgraded diff --git a/.github/workflows/rector.yaml b/.github/workflows/rector.yaml index 09797bbe12a..647fb4925c9 100644 --- a/.github/workflows/rector.yaml +++ b/.github/workflows/rector.yaml @@ -31,7 +31,7 @@ jobs: - run: composer install --no-progress --ansi ## First run Rector - here can't be --dry-run !!! it would stop the job with it and not commit anything in the future - - run: bin/rector process --ansi --lpt + - run: bin/rector process --ansi - # commit only to core contributors who have repository access diff --git a/phpstan.neon b/phpstan.neon index 2a56ff672a0..722ebcb9df4 100644 --- a/phpstan.neon +++ b/phpstan.neon @@ -331,7 +331,7 @@ parameters: - src/BetterPhpDocParser/PhpDocManipulator/PhpDocClassRenamer.php - src/NodeTypeResolver/PHPStan/Type/TypeFactory.php - rules/CodeQuality/NodeAnalyzer/VariableDimFetchAssignResolver.php - - src/Parallel/Experimental/ExperimentalParallelFileProcessor.php + - src/Parallel/Application/ParallelFileProcessor.php - identifier: symplify.noNullableServiceInConstructor diff --git a/src/Application/ApplicationFileProcessor.php b/src/Application/ApplicationFileProcessor.php index 53b4cd884ff..d3480f0d83c 100644 --- a/src/Application/ApplicationFileProcessor.php +++ b/src/Application/ApplicationFileProcessor.php @@ -14,8 +14,6 @@ use Rector\Parallel\Application\ParallelFileProcessor; use Rector\Parallel\CpuCoreCountProvider; use Rector\Parallel\Exception\ParallelShouldNotHappenException; -use Rector\Parallel\Experimental\ExperimentalParallelFileProcessor; -use Rector\Parallel\Experimental\LptScheduleFactory; use Rector\Parallel\ScheduleFactory; use Rector\PhpParser\Parser\ParserErrors; use Rector\Reporting\MissConfigurationReporter; @@ -53,9 +51,6 @@ public function __construct( private readonly ArrayParametersMerger $arrayParametersMerger, private readonly MissConfigurationReporter $missConfigurationReporter, private readonly Skipper $skipper, - // @experimental, see --lpt - private readonly LptScheduleFactory $lptScheduleFactory, - private readonly ExperimentalParallelFileProcessor $experimentalParallelFileProcessor, ) { } @@ -273,27 +268,9 @@ private function runParallel( Option::PARALLEL_MAX_NUMBER_OF_PROCESSES ); - // @experimental opt-in, see --lpt - if ($input->hasOption(Option::LPT) && (bool) $input->getOption(Option::LPT)) { - $bucketSchedule = $this->lptScheduleFactory->create( - $cpuCores, - $jobSize, - $maxNumberOfProcesses, - $filePaths - ); - - return $this->experimentalParallelFileProcessor->process( - $bucketSchedule, - $mainScript, - $postFileCallback, - $input - ); - } - - $schedule = $this->scheduleFactory->create($cpuCores, $jobSize, $maxNumberOfProcesses, $filePaths); + $bucketSchedule = $this->scheduleFactory->create($cpuCores, $jobSize, $maxNumberOfProcesses, $filePaths); - // mimics see https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-387b8f04e0db7a06678eb52ce0c0d0aff73e0d7d8fc5df834d0a5fbec198e5daR139 - return $this->parallelFileProcessor->process($schedule, $mainScript, $postFileCallback, $input); + return $this->parallelFileProcessor->process($bucketSchedule, $mainScript, $postFileCallback, $input); } /** diff --git a/src/Configuration/Option.php b/src/Configuration/Option.php index 34e5672107a..8abd1bef860 100644 --- a/src/Configuration/Option.php +++ b/src/Configuration/Option.php @@ -184,12 +184,6 @@ final class Option */ public const string PARALLEL_JOB_TIMEOUT_IN_SECONDS = 'parallel-job-timeout-in-seconds'; - /** - * @experimental Opt-in to the LPT bucket scheduler - * @see \Rector\Parallel\Experimental\LptScheduleFactory - */ - public const string LPT = 'lpt'; - public const string MEMORY_LIMIT = 'memory-limit'; /** diff --git a/src/Console/ProcessConfigureDecorator.php b/src/Console/ProcessConfigureDecorator.php index 34ada988b6c..800168aba89 100644 --- a/src/Console/ProcessConfigureDecorator.php +++ b/src/Console/ProcessConfigureDecorator.php @@ -99,13 +99,6 @@ public static function decorate(Command $command): void $command->addOption(Option::PARALLEL_PORT, null, InputOption::VALUE_REQUIRED); $command->addOption(Option::PARALLEL_IDENTIFIER, null, InputOption::VALUE_REQUIRED); - $command->addOption( - Option::LPT, - null, - InputOption::VALUE_NONE, - '[EXPERIMENTAL] Balance files across parallel workers by size up front and keep every worker alive for its whole share.' - ); - $command->addOption(Option::XDEBUG, null, InputOption::VALUE_NONE, 'Display xdebug output.'); $command->addOption( diff --git a/src/Parallel/Application/ParallelFileProcessor.php b/src/Parallel/Application/ParallelFileProcessor.php index 43745f7bda9..94d744d1747 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -18,11 +18,10 @@ use Rector\Parallel\Enum\Content; use Rector\Parallel\Enum\ReactCommand; use Rector\Parallel\Enum\ReactEvent; -use Rector\Parallel\Enum\StreamFormat; use Rector\Parallel\ValueObject\Bridge; +use Rector\Parallel\ValueObject\BucketSchedule; use Rector\Parallel\ValueObject\ParallelProcess; use Rector\Parallel\ValueObject\ProcessPool; -use Rector\Parallel\ValueObject\Schedule; use Rector\ValueObject\Error\SystemError; use Rector\ValueObject\ProcessResult; use Symfony\Component\Console\Command\Command; @@ -30,22 +29,23 @@ use Throwable; /** - * Inspired from @see - * https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-39c7a3b0cbb217bbfff96fbb454e6e5e60c74cf92fbb0f9d246b8bebbaad2bb0 + * Processes files in parallel using the LPT bucket schedule from @see \Rector\Parallel\ScheduleFactory. * - * https://github.com/phpstan/phpstan-src/commit/b84acd2e3eadf66189a64fdbc6dd18ff76323f67#diff-7f625777f1ce5384046df08abffd6c911cfbb1cfc8fcb2bdeaf78f337689e3e2R150 + * Two key properties: + * + * 1. a worker pulls only from its own bucket, decided up front, instead of from a shared job pool; + * 2. a worker is never killed and respawned mid-run, so its warm in-memory caches (reflection, parsed + * AST, node scope resolver) survive the whole bucket. + * + * The bucket is still handed over in small chunks: a worker response is what advances the progress bar + * and re-arms the per-job timeout. One single request per worker would freeze the progress bar and put + * the whole bucket under one 120s timeout. */ final class ParallelFileProcessor { private const int SYSTEM_ERROR_LIMIT = 50; - /** - * The number of chunks a worker can process before getting killed. - * In contrast the jobSize defines the maximum size of a chunk, a worker process at a time. - */ - private const int MAX_CHUNKS_PER_WORKER = 8; - - private ProcessPool|null $processPool = null; + private ProcessPool $processPool; public function __construct( private readonly WorkerCommandLineFactory $workerCommandLineFactory, @@ -56,27 +56,40 @@ public function __construct( * @param callable(int $stepCount): void $postFileCallback Used for progress bar jump */ public function process( - Schedule $schedule, + BucketSchedule $bucketSchedule, string $mainScript, callable $postFileCallback, InputInterface $input ): ProcessResult { - $jobs = array_reverse($schedule->getJobs()); - $streamSelectLoop = new StreamSelectLoop(); + // reversed, so that array_pop() hands out the biggest files first + $jobsPerWorker = array_map( + array_reverse(...), + $bucketSchedule->getJobsPerWorker() + ); + + /** @var array $bucketKeyByIdentifier */ + $bucketKeyByIdentifier = []; - // basic properties setup - $numberOfProcesses = $schedule->getNumberOfProcesses(); + $streamSelectLoop = new StreamSelectLoop(); $parallelResultCollector = new ParallelResultCollector(); $tcpServer = new TcpServer('127.0.0.1:0', $streamSelectLoop); $this->processPool = new ProcessPool($tcpServer); - $tcpServer->on(ReactEvent::CONNECTION, function (ConnectionInterface $connection) use (&$jobs): void { - $inDecoder = new Decoder($connection, true, StreamFormat::DEPTH, 0, StreamFormat::MAX_LENGTH); + $tcpServer->on(ReactEvent::CONNECTION, function (ConnectionInterface $connection) use ( + &$jobsPerWorker, + &$bucketKeyByIdentifier + ): void { + $inDecoder = new Decoder($connection, true, 512, 0, 4 * 1024 * 1024); $outEncoder = new Encoder($connection); - $inDecoder->on(ReactEvent::DATA, function (array $data) use (&$jobs, $inDecoder, $outEncoder): void { + $inDecoder->on(ReactEvent::DATA, function (array $data) use ( + &$jobsPerWorker, + &$bucketKeyByIdentifier, + $inDecoder, + $outEncoder + ): void { $action = $data[ReactCommand::ACTION]; if ($action !== Action::HELLO) { return; @@ -86,13 +99,12 @@ public function process( $parallelProcess = $this->processPool->getProcess($processIdentifier); $parallelProcess->bindConnection($inDecoder, $outEncoder); - if ($jobs === []) { + $jobsChunk = $this->popJobChunk($processIdentifier, $jobsPerWorker, $bucketKeyByIdentifier); + if ($jobsChunk === null) { $this->processPool->quitProcess($processIdentifier); return; } - $jobsChunk = array_pop($jobs); - $parallelProcess->request([ ReactCommand::ACTION => Action::MAIN, Content::FILES => $jobsChunk, @@ -108,8 +120,6 @@ public function process( $systemErrorsCount = 0; $reachedSystemErrorsCountLimit = false; - $scheduledFilesCount = array_sum(array_map(count(...), $jobs)); - $processedFilesCount = 0; $handleErrorCallable = function (Throwable $throwable) use ( $parallelResultCollector, @@ -126,32 +136,30 @@ public function process( $reachedSystemErrorsCountLimit = true; $this->processPool->quitAll(); - // This sleep has to be here, because event though we have called $this->processPool->quitAll(), - // it takes some time for the child processes to actually die, during which they can still write to cache + // give the child processes time to actually die, they can still write to cache meanwhile // @see https://github.com/rectorphp/rector-src/pull/3834/files#r1231696531 sleep(1); }; $timeoutInSeconds = SimpleParameterProvider::provideIntParameter(Option::PARALLEL_JOB_TIMEOUT_IN_SECONDS); - $fileChunksBudgetPerProcess = []; - $processSpawner = function () use ( + $processSpawner = function (int $bucketKey) use ( $parallelResultCollector, - &$jobs, + &$jobsPerWorker, + &$bucketKeyByIdentifier, $postFileCallback, &$systemErrorsCount, - &$reachedSystemErrorsCountLimit, + &$reachedInternalErrorsCountLimit, $mainScript, $input, $serverPort, $streamSelectLoop, $timeoutInSeconds, - $handleErrorCallable, - &$fileChunksBudgetPerProcess, - &$processSpawner, - &$processedFilesCount + $handleErrorCallable ): void { $processIdentifier = Random::generate(); + $bucketKeyByIdentifier[$processIdentifier] = $bucketKey; + $workerCommandLine = $this->workerCommandLineFactory->create( $mainScript, ProcessCommand::class, @@ -160,7 +168,6 @@ public function process( $processIdentifier, $serverPort, ); - $fileChunksBudgetPerProcess[$processIdentifier] = self::MAX_CHUNKS_PER_WORKER; $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); @@ -169,14 +176,12 @@ public function process( function (array $json) use ( $parallelProcess, $parallelResultCollector, - &$jobs, + &$jobsPerWorker, + &$bucketKeyByIdentifier, $postFileCallback, &$systemErrorsCount, - &$reachedSystemErrorsCountLimit, - $processIdentifier, - &$fileChunksBudgetPerProcess, - &$processSpawner, - &$processedFilesCount + &$reachedInternalErrorsCountLimit, + $processIdentifier ): void { /** @var array{ * total_changed: int, @@ -186,36 +191,25 @@ function (array $json) use ( * system_errors_count: int, * used_skips: array * } $json */ - $filesCount = $parallelResultCollector->collectWorkerResult($json); - $postFileCallback($filesCount); - $processedFilesCount += $filesCount; + $postFileCallback($parallelResultCollector->collectWorkerResult($json)); $systemErrorsCount += $json[Bridge::SYSTEM_ERRORS_COUNT]; if ($systemErrorsCount >= self::SYSTEM_ERROR_LIMIT) { - $reachedSystemErrorsCountLimit = true; + $reachedInternalErrorsCountLimit = true; $this->processPool->quitAll(); - return; } - if ($jobs === []) { + $jobsChunk = $this->popJobChunk($processIdentifier, $jobsPerWorker, $bucketKeyByIdentifier); + if ($jobsChunk === null) { + // bucket done, this worker has nothing left to do $this->processPool->quitProcess($processIdentifier); return; } - if ($fileChunksBudgetPerProcess[$processIdentifier] <= 0) { - // replace the current worker with a fresh one to free memory; spawn the replacement first, - // as quitting the last worker in the pool closes the server the replacement connects to - ($processSpawner)(); - $this->processPool->quitProcess($processIdentifier); - return; - } - - $jobsChunk = array_pop($jobs); $parallelProcess->request([ ReactCommand::ACTION => Action::MAIN, Content::FILES => $jobsChunk, ]); - --$fileChunksBudgetPerProcess[$processIdentifier]; }, // 2. callable on error @@ -241,13 +235,8 @@ function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIden $this->processPool->attachProcess($processIdentifier, $parallelProcess); }; - for ($i = 0; $i < $numberOfProcesses; ++$i) { - // nothing else to process, stop now - if ($jobs === []) { - break; - } - - ($processSpawner)(); + foreach (array_keys($jobsPerWorker) as $bucketKey) { + ($processSpawner)($bucketKey); } $streamSelectLoop->run(); @@ -259,15 +248,60 @@ function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIden ))); } - // a worker can end without reporting its files, e.g. when killed by the OS, so missing results must always be reported - if ($processedFilesCount < $scheduledFilesCount) { - $parallelResultCollector->collectSystemError(new SystemError(sprintf( - 'Some parallel jobs have not finished, results for %d of %d files are missing', - $scheduledFilesCount - $processedFilesCount, - $scheduledFilesCount - ))); + return $parallelResultCollector->createProcessResult(); + } + + /** + * @param array>> $jobsPerWorker + * @param array $bucketKeyByIdentifier + * @return array|null + */ + private function popJobChunk( + string $processIdentifier, + array &$jobsPerWorker, + array $bucketKeyByIdentifier + ): ?array { + $bucketKey = $bucketKeyByIdentifier[$processIdentifier] ?? null; + + if ($bucketKey !== null && ($jobsPerWorker[$bucketKey] ?? []) !== []) { + return array_pop($jobsPerWorker[$bucketKey]); } - return $parallelResultCollector->createProcessResult(); + if (getenv('RECTOR_LPT_NO_STEAL') !== false) { + return null; + } + + return $this->stealJobChunk($jobsPerWorker); + } + + /** + * Own bucket drained while others still have work. Byte size only estimates the cost of a file, and + * not every core is equally fast - on a big.LITTLE CPU (Apple silicon, recent Intel) a bucket handed + * to an efficiency core takes several times longer. Take over a chunk from whoever has the most left, + * so the run is not held up by one straggler. + * + * @param array>> $jobsPerWorker + * @return array|null + */ + private function stealJobChunk(array &$jobsPerWorker): ?array + { + $fattestBucketKey = null; + $fattestJobCount = 0; + + foreach ($jobsPerWorker as $currentBucketKey => $jobs) { + $jobCount = count($jobs); + if ($jobCount <= $fattestJobCount) { + continue; + } + + $fattestJobCount = $jobCount; + $fattestBucketKey = $currentBucketKey; + } + + if ($fattestBucketKey === null) { + return null; + } + + return array_pop($jobsPerWorker[$fattestBucketKey]); } } diff --git a/src/Parallel/Experimental/ExperimentalParallelFileProcessor.php b/src/Parallel/Experimental/ExperimentalParallelFileProcessor.php deleted file mode 100644 index 1352ba42f80..00000000000 --- a/src/Parallel/Experimental/ExperimentalParallelFileProcessor.php +++ /dev/null @@ -1,311 +0,0 @@ -getJobsPerWorker() - ); - - /** @var array $bucketKeyByIdentifier */ - $bucketKeyByIdentifier = []; - - $streamSelectLoop = new StreamSelectLoop(); - - $parallelResultCollector = new ParallelResultCollector(); - - $tcpServer = new TcpServer('127.0.0.1:0', $streamSelectLoop); - $this->processPool = new ProcessPool($tcpServer); - - $tcpServer->on(ReactEvent::CONNECTION, function (ConnectionInterface $connection) use ( - &$jobsPerWorker, - &$bucketKeyByIdentifier - ): void { - $inDecoder = new Decoder($connection, true, 512, 0, 4 * 1024 * 1024); - $outEncoder = new Encoder($connection); - - $inDecoder->on(ReactEvent::DATA, function (array $data) use ( - &$jobsPerWorker, - &$bucketKeyByIdentifier, - $inDecoder, - $outEncoder - ): void { - $action = $data[ReactCommand::ACTION]; - if ($action !== Action::HELLO) { - return; - } - - $processIdentifier = $data[Option::PARALLEL_IDENTIFIER]; - $parallelProcess = $this->processPool->getProcess($processIdentifier); - $parallelProcess->bindConnection($inDecoder, $outEncoder); - - $jobsChunk = $this->popJobChunk($processIdentifier, $jobsPerWorker, $bucketKeyByIdentifier); - if ($jobsChunk === null) { - $this->processPool->quitProcess($processIdentifier); - return; - } - - $parallelProcess->request([ - ReactCommand::ACTION => Action::MAIN, - Content::FILES => $jobsChunk, - ]); - }); - }); - - /** @var string $serverAddress */ - $serverAddress = $tcpServer->getAddress(); - - /** @var int $serverPort */ - $serverPort = parse_url($serverAddress, PHP_URL_PORT); - - $systemErrorsCount = 0; - $reachedSystemErrorsCountLimit = false; - - $handleErrorCallable = function (Throwable $throwable) use ( - $parallelResultCollector, - &$systemErrorsCount, - &$reachedSystemErrorsCountLimit - ): void { - $parallelResultCollector->collectSystemError(new SystemError( - $throwable->getMessage(), - $throwable->getFile(), - $throwable->getLine(), - )); - - ++$systemErrorsCount; - $reachedSystemErrorsCountLimit = true; - $this->processPool->quitAll(); - - // give the child processes time to actually die, they can still write to cache meanwhile - // @see https://github.com/rectorphp/rector-src/pull/3834/files#r1231696531 - sleep(1); - }; - - $timeoutInSeconds = SimpleParameterProvider::provideIntParameter(Option::PARALLEL_JOB_TIMEOUT_IN_SECONDS); - - $processSpawner = function (int $bucketKey) use ( - $parallelResultCollector, - &$jobsPerWorker, - &$bucketKeyByIdentifier, - $postFileCallback, - &$systemErrorsCount, - &$reachedInternalErrorsCountLimit, - $mainScript, - $input, - $serverPort, - $streamSelectLoop, - $timeoutInSeconds, - $handleErrorCallable - ): void { - $processIdentifier = Random::generate(); - $bucketKeyByIdentifier[$processIdentifier] = $bucketKey; - - $workerCommandLine = $this->workerCommandLineFactory->create( - $mainScript, - ProcessCommand::class, - 'worker', - $input, - $processIdentifier, - $serverPort, - ); - - $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); - - $parallelProcess->start( - // 1. callable on data - function (array $json) use ( - $parallelProcess, - $parallelResultCollector, - &$jobsPerWorker, - &$bucketKeyByIdentifier, - $postFileCallback, - &$systemErrorsCount, - &$reachedInternalErrorsCountLimit, - $processIdentifier - ): void { - /** @var array{ - * total_changed: int, - * system_errors: mixed[], - * file_diffs: array, - * files_count: int, - * system_errors_count: int, - * used_skips: array - * } $json */ - $postFileCallback($parallelResultCollector->collectWorkerResult($json)); - - $systemErrorsCount += $json[Bridge::SYSTEM_ERRORS_COUNT]; - if ($systemErrorsCount >= self::SYSTEM_ERROR_LIMIT) { - $reachedInternalErrorsCountLimit = true; - $this->processPool->quitAll(); - } - - $jobsChunk = $this->popJobChunk($processIdentifier, $jobsPerWorker, $bucketKeyByIdentifier); - if ($jobsChunk === null) { - // bucket done, this worker has nothing left to do - $this->processPool->quitProcess($processIdentifier); - return; - } - - $parallelProcess->request([ - ReactCommand::ACTION => Action::MAIN, - Content::FILES => $jobsChunk, - ]); - }, - - // 2. callable on error - $handleErrorCallable, - - // 3. callable on exit - function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIdentifier): void { - $this->processPool->tryQuitProcess($processIdentifier); - if ($exitCode === Command::SUCCESS) { - return; - } - - if ($exitCode === null) { - return; - } - - $parallelResultCollector->collectSystemError(new SystemError( - 'Child process error: ' . $stdErr - )); - } - ); - - $this->processPool->attachProcess($processIdentifier, $parallelProcess); - }; - - foreach (array_keys($jobsPerWorker) as $bucketKey) { - ($processSpawner)($bucketKey); - } - - $streamSelectLoop->run(); - - if ($reachedSystemErrorsCountLimit) { - $parallelResultCollector->collectSystemError(new SystemError(sprintf( - 'Reached system errors count limit of %d, exiting...', - self::SYSTEM_ERROR_LIMIT - ))); - } - - return $parallelResultCollector->createProcessResult(); - } - - /** - * @param array>> $jobsPerWorker - * @param array $bucketKeyByIdentifier - * @return array|null - */ - private function popJobChunk( - string $processIdentifier, - array &$jobsPerWorker, - array $bucketKeyByIdentifier - ): ?array { - $bucketKey = $bucketKeyByIdentifier[$processIdentifier] ?? null; - - if ($bucketKey !== null && ($jobsPerWorker[$bucketKey] ?? []) !== []) { - return array_pop($jobsPerWorker[$bucketKey]); - } - - if (getenv('RECTOR_LPT_NO_STEAL') !== false) { - return null; - } - - return $this->stealJobChunk($jobsPerWorker); - } - - /** - * Own bucket drained while others still have work. Byte size only estimates the cost of a file, and - * not every core is equally fast - on a big.LITTLE CPU (Apple silicon, recent Intel) a bucket handed - * to an efficiency core takes several times longer. Take over a chunk from whoever has the most left, - * so the run is not held up by one straggler. - * - * @param array>> $jobsPerWorker - * @return array|null - */ - private function stealJobChunk(array &$jobsPerWorker): ?array - { - $fattestBucketKey = null; - $fattestJobCount = 0; - - foreach ($jobsPerWorker as $currentBucketKey => $jobs) { - $jobCount = count($jobs); - if ($jobCount <= $fattestJobCount) { - continue; - } - - $fattestJobCount = $jobCount; - $fattestBucketKey = $currentBucketKey; - } - - if ($fattestBucketKey === null) { - return null; - } - - return array_pop($jobsPerWorker[$fattestBucketKey]); - } -} diff --git a/src/Parallel/Experimental/LptScheduleFactory.php b/src/Parallel/Experimental/LptScheduleFactory.php deleted file mode 100644 index 84ea5ed5172..00000000000 --- a/src/Parallel/Experimental/LptScheduleFactory.php +++ /dev/null @@ -1,87 +0,0 @@ -createBalancedBuckets($filePaths, $numberOfProcesses) as $bucketFilePaths) { - // fewer files than workers - no worker for an empty bucket - if ($bucketFilePaths === []) { - continue; - } - - // jobSize is what a worker is handed per request. Balancing already happened when the buckets - // were built, so this is the heartbeat interval - a response advances the progress bar and - // re-arms the per-job timeout - and, more importantly, the granularity at which an idle worker - // can take work over from a busy one. - $jobsPerWorker[] = array_chunk($bucketFilePaths, $jobSize); - } - - return new BucketSchedule($jobsPerWorker); - } - - /** - * LPT (Longest Processing Time first): walk the files largest-first and drop each one into the bucket - * with the smallest byte load so far. Keeps the gap between the slowest and the fastest worker small, - * so the run does not end with a single worker still chewing on the one 200 kB file. - * - * @param string[] $filePaths - * @return array> - */ - private function createBalancedBuckets(array $filePaths, int $numberOfBuckets): array - { - // tuples rather than a path-keyed map: numeric-looking paths would be cast to int keys, and - // duplicate paths would silently collapse - $sizedFilePaths = []; - foreach ($filePaths as $filePath) { - $sizedFilePaths[] = [$filePath, is_file($filePath) ? (int) filesize($filePath) : 0]; - } - - usort($sizedFilePaths, static fn (array $first, array $second): int => $second[1] <=> $first[1]); - - $buckets = array_fill(0, $numberOfBuckets, []); - $bucketSizes = array_fill(0, $numberOfBuckets, 0); - - foreach ($sizedFilePaths as [$filePath, $fileSize]) { - $emptiestBucketKey = 0; - foreach ($bucketSizes as $currentBucketKey => $bucketSize) { - if ($bucketSize < $bucketSizes[$emptiestBucketKey]) { - $emptiestBucketKey = $currentBucketKey; - } - } - - $buckets[$emptiestBucketKey][] = $filePath; - $bucketSizes[$emptiestBucketKey] += $fileSize; - } - - return $buckets; - } -} diff --git a/src/Parallel/ScheduleFactory.php b/src/Parallel/ScheduleFactory.php index d8c83d8cacb..21822e119db 100644 --- a/src/Parallel/ScheduleFactory.php +++ b/src/Parallel/ScheduleFactory.php @@ -4,27 +4,83 @@ namespace Rector\Parallel; -use Rector\Parallel\ValueObject\Schedule; +use Rector\Parallel\ValueObject\BucketSchedule; use Webmozart\Assert\Assert; /** - * Used from - * https://github.com/phpstan/phpstan-src/blob/9124c66dcc55a222e21b1717ba5f60771f7dda92/src/Parallel/Scheduler.php + * LPT (Longest Processing Time first) scheduler. + * + * Splits the files into one fixed bucket per worker, balanced by total byte size, so a worker is + * started once and keeps its warm caches for the whole run. + * + * Idea from https://github.com/rectorphp/rector-src/issues/8494#issuecomment-5664864017 */ final class ScheduleFactory { /** - * @param array $files + * @param string[] $filePaths */ - public function create(int $cpuCores, int $jobSize, int $maxNumberOfProcesses, array $files): Schedule + public function create(int $cpuCores, int $jobSize, int $maxNumberOfProcesses, array $filePaths): BucketSchedule { Assert::positiveInteger($jobSize); + Assert::notEmpty($filePaths); - $jobs = array_chunk($files, $jobSize); - $numberOfProcesses = min(count($jobs), $cpuCores); + // same worker count as the default scheduler: a worker is worth starting once there is at least + // one whole job for it + $numberOfProcesses = min((int) ceil(count($filePaths) / $jobSize), $cpuCores, $maxNumberOfProcesses); + $numberOfProcesses = max(1, $numberOfProcesses); - $numberOfProcesses = min($maxNumberOfProcesses, $numberOfProcesses); + $jobsPerWorker = []; + foreach ($this->createBalancedBuckets($filePaths, $numberOfProcesses) as $bucketFilePaths) { + // fewer files than workers - no worker for an empty bucket + if ($bucketFilePaths === []) { + continue; + } - return new Schedule($numberOfProcesses, $jobs); + // jobSize is what a worker is handed per request. Balancing already happened when the buckets + // were built, so this is the heartbeat interval - a response advances the progress bar and + // re-arms the per-job timeout - and, more importantly, the granularity at which an idle worker + // can take work over from a busy one. + $jobsPerWorker[] = array_chunk($bucketFilePaths, $jobSize); + } + + return new BucketSchedule($jobsPerWorker); + } + + /** + * LPT (Longest Processing Time first): walk the files largest-first and drop each one into the bucket + * with the smallest byte load so far. Keeps the gap between the slowest and the fastest worker small, + * so the run does not end with a single worker still chewing on the one 200 kB file. + * + * @param string[] $filePaths + * @return array> + */ + private function createBalancedBuckets(array $filePaths, int $numberOfBuckets): array + { + // tuples rather than a path-keyed map: numeric-looking paths would be cast to int keys, and + // duplicate paths would silently collapse + $sizedFilePaths = []; + foreach ($filePaths as $filePath) { + $sizedFilePaths[] = [$filePath, is_file($filePath) ? (int) filesize($filePath) : 0]; + } + + usort($sizedFilePaths, static fn (array $first, array $second): int => $second[1] <=> $first[1]); + + $buckets = array_fill(0, $numberOfBuckets, []); + $bucketSizes = array_fill(0, $numberOfBuckets, 0); + + foreach ($sizedFilePaths as [$filePath, $fileSize]) { + $emptiestBucketKey = 0; + foreach ($bucketSizes as $currentBucketKey => $bucketSize) { + if ($bucketSize < $bucketSizes[$emptiestBucketKey]) { + $emptiestBucketKey = $currentBucketKey; + } + } + + $buckets[$emptiestBucketKey][] = $filePath; + $bucketSizes[$emptiestBucketKey] += $fileSize; + } + + return $buckets; } } diff --git a/src/Parallel/Experimental/ValueObject/BucketSchedule.php b/src/Parallel/ValueObject/BucketSchedule.php similarity index 60% rename from src/Parallel/Experimental/ValueObject/BucketSchedule.php rename to src/Parallel/ValueObject/BucketSchedule.php index dff8464d651..a46bf2a4abb 100644 --- a/src/Parallel/Experimental/ValueObject/BucketSchedule.php +++ b/src/Parallel/ValueObject/BucketSchedule.php @@ -2,13 +2,10 @@ declare(strict_types=1); -namespace Rector\Parallel\Experimental\ValueObject; +namespace Rector\Parallel\ValueObject; /** - * @experimental Filled by @see \Rector\Parallel\Experimental\LptScheduleFactory, consumed only by the - * "--lpt" parallel run. - * - * Unlike @see \Rector\Parallel\ValueObject\Schedule, jobs are not a shared pool: every worker owns its + * Filled by @see \Rector\Parallel\ScheduleFactory. Jobs are not a shared pool: every worker owns its * own queue, decided up front. */ final readonly class BucketSchedule diff --git a/src/Parallel/ValueObject/Schedule.php b/src/Parallel/ValueObject/Schedule.php deleted file mode 100644 index c66797e9e59..00000000000 --- a/src/Parallel/ValueObject/Schedule.php +++ /dev/null @@ -1,34 +0,0 @@ -> $jobs - */ - public function __construct( - private int $numberOfProcesses, - private array $jobs - ) { - } - - public function getNumberOfProcesses(): int - { - return $this->numberOfProcesses; - } - - /** - * @return array> - */ - public function getJobs(): array - { - return $this->jobs; - } -} From 487f6a3f062842fd16eef5c95975a85b550833c0 Mon Sep 17 00:00:00 2001 From: Tomas Votruba Date: Fri, 23 Oct 2026 12:02:09 +0200 Subject: [PATCH 2/2] Fix parallel e2e tests and stale PHPStan ignore for LPT scheduler - regenerate expected output of the parallel e2e tests for the LPT file order (single worker, bucket ordered by file size); the change set is identical, only the reporting order differs - update test comments that referenced the removed mid-run worker respawn - drop the obsolete ProcessPool nullable ignore; the promoted processor uses a non-nullable ProcessPool --- .../expected-output.diff | 9 +-- e2e/parallel-system-errors-limit/rector.php | 3 +- .../expected-output.diff | 23 ++++-- .../expected-output.diff | 74 +++++++++---------- e2e/parallel-worker-respawn/rector.php | 2 +- phpstan.neon | 6 -- 6 files changed, 56 insertions(+), 61 deletions(-) diff --git a/e2e/parallel-system-errors-limit/expected-output.diff b/e2e/parallel-system-errors-limit/expected-output.diff index ef787becca4..172e5ee98e5 100644 --- a/e2e/parallel-system-errors-limit/expected-output.diff +++ b/e2e/parallel-system-errors-limit/expected-output.diff @@ -146,11 +146,4 @@ "Syntax error, unexpected ';'". On line: 3 [ERROR] Could not process "src/File54.php" file, due to: - "Syntax error, unexpected ';'". On line: 3 - - [ERROR] Could not process some files, due to: - "Reached system errors count limit of 50, exiting...". - - [ERROR] Could not process some files, due to: - "Some parallel jobs have not finished, results for 1 of 55 files are - missing". + "Syntax error, unexpected ';'". On line: 3 \ No newline at end of file diff --git a/e2e/parallel-system-errors-limit/rector.php b/e2e/parallel-system-errors-limit/rector.php index 2e8c8823f1a..4567dd326a3 100644 --- a/e2e/parallel-system-errors-limit/rector.php +++ b/e2e/parallel-system-errors-limit/rector.php @@ -10,8 +10,7 @@ __DIR__ . '/src', ]); - // 4 valid files first, so the 50th syntax error arrives with the 54th chunk, when a worker's chunk budget - // (MAX_CHUNKS_PER_WORKER + 1) runs out while one job is left + // 4 valid files first, then 50 syntax errors, so the system error limit is reached mid-bucket $rectorConfig->parallel(maxNumberOfProcess: 1, jobSize: 1); $rectorConfig->rules([ diff --git a/e2e/parallel-unfinished-jobs/expected-output.diff b/e2e/parallel-unfinished-jobs/expected-output.diff index 59be2f290fe..6c3c98fe345 100644 --- a/e2e/parallel-unfinished-jobs/expected-output.diff +++ b/e2e/parallel-unfinished-jobs/expected-output.diff @@ -1,5 +1,5 @@ -1 file with changes -=================== +2 files with changes +==================== 1) src/File1.php:1 @@ -15,9 +15,18 @@ Applied rules: * LongArrayToShortArrayRector - [ERROR] Could not process "src/File2.php" file, due to: - "Syntax error, unexpected ';'". On line: 3 +2) src/File4.php:1 - [ERROR] Could not process some files, due to: - "Some parallel jobs have not finished, results for 2 of 4 files are - missing". + ---------- begin diff ---------- +@@ Line 1 @@ + parallel(maxNumberOfProcess: 1, jobSize: 1); $rectorConfig->rules([ diff --git a/phpstan.neon b/phpstan.neon index 722ebcb9df4..5a674323659 100644 --- a/phpstan.neon +++ b/phpstan.neon @@ -160,12 +160,6 @@ parameters: identifier: greaterOrEqual.alwaysTrue message: '#Comparison operation ".*" between int<\d+, \d+> and \d+ is always true#' - # known value object, nullable due to typed property - - - identifier: method.nonObject - message: '#Cannot call method (.*?)\(\) on (.*?)\\ProcessPool\|null#' - path: src/Parallel/Application/ParallelFileProcessor.php - # required generics on interface array, not helpful - identifier: missingType.generics