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/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 2a56ff672a0..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 @@ -331,7 +325,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; - } -}