[perf] Parallel LPT buckets

phpstan.neon

@@ -25,6 +25,7 @@ parameters:
         - config
         - src
         - rules
+        - scripts
         # tests
         - tests
         - rules-tests
@@ -361,6 +362,7 @@ parameters:
                 - src/NodeTypeResolver/PHPStan/Type/TypeFactory.php
                 - rules/CodeQuality/Rector/Class_/InnerFunctionToPrivateMethodRector.php
                 - rules/CodeQuality/NodeAnalyzer/VariableDimFetchAssignResolver.php
+                - src/Parallel/Experimental/ExperimentalParallelFileProcessor.php
 
         -
             identifier: symplify.noNullableServiceInConstructor

rules/DeadCode/Rector/Property/RemoveUnusedPrivatePropertyRector.php

@@ -68,6 +68,15 @@ public function getNodeTypes(): array
      */
     public function refactor(Node $node): ?Node
     {
+        // skip if has no private property
+        $hasAnyPrivateProperty = array_any($node->getProperties(), function (Property $property): bool {
+            return $property->isPrivate();
+        });
+
+        if (! $hasAnyPrivateProperty) {
+            return null;
+        }
+
         if ($this->shouldSkipClass($node)) {
             return null;
         }

scripts/validate-rule-definitions.php

@@ -3,11 +3,11 @@
 declare(strict_types=1);
 
 use Rector\Console\ExitCode;
+use Rector\RuleDoc\Contract\DocumentedRuleInterface;
 use Rector\Scripts\Finder\RectorClassFinder;
 use Symfony\Component\Console\Input\ArrayInput;
 use Symfony\Component\Console\Output\ConsoleOutput;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use Symplify\RuleDocGenerator\Contract\DocumentedRuleInterface;
 
 require __DIR__ . '/../vendor/autoload.php';
 

src/Application/ApplicationFileProcessor.php

@@ -14,6 +14,8 @@
 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;
@@ -51,6 +53,9 @@ 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,
     ) {
     }
 
@@ -257,18 +262,36 @@ private function runParallel(
         InputInterface $input,
         callable $postFileCallback,
     ): ProcessResult {
-        $schedule = $this->scheduleFactory->create(
-            $this->cpuCoreCountProvider->provide(),
-            SimpleParameterProvider::provideIntParameter(Option::PARALLEL_JOB_SIZE),
-            SimpleParameterProvider::provideIntParameter(Option::PARALLEL_MAX_NUMBER_OF_PROCESSES),
-            $filePaths
-        );
-
         $mainScript = $this->resolveCalledRectorBinary();
         if ($mainScript === null) {
             throw new ParallelShouldNotHappenException('[parallel] Main script was not found');
         }
 
+        $cpuCores = $this->cpuCoreCountProvider->provide();
+        $jobSize = SimpleParameterProvider::provideIntParameter(Option::PARALLEL_JOB_SIZE);
+        $maxNumberOfProcesses = SimpleParameterProvider::provideIntParameter(
+            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);
+
         // mimics see https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-387b8f04e0db7a06678eb52ce0c0d0aff73e0d7d8fc5df834d0a5fbec198e5daR139
         return $this->parallelFileProcessor->process($schedule, $mainScript, $postFileCallback, $input);
     }

src/Configuration/Option.php

@@ -184,6 +184,13 @@ 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
+     * @var string
+     */
+    public const string LPT = 'lpt';
+
     public const string MEMORY_LIMIT = 'memory-limit';
 
     /**

src/Console/ProcessConfigureDecorator.php

@@ -99,6 +99,13 @@ 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(

src/Parallel/Application/ParallelFileProcessor.php

@@ -25,7 +25,6 @@
 use Rector\Parallel\ValueObject\Schedule;
 use Rector\ValueObject\Error\SystemError;
 use Rector\ValueObject\ProcessResult;
-use Rector\ValueObject\Reporting\FileDiff;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Throwable;
@@ -68,16 +67,7 @@ public function process(
         // basic properties setup
         $numberOfProcesses = $schedule->getNumberOfProcesses();
 
-        // initial counters
-
-        /** @var FileDiff[] $fileDiffs */
-        $fileDiffs = [];
-
-        /** @var SystemError[] $systemErrors */
-        $systemErrors = [];
-
-        /** @var array<string, array<string, true>> $usedSkips */
-        $usedSkips = [];
+        $parallelResultCollector = new ParallelResultCollector();
 
         $tcpServer = new TcpServer('127.0.0.1:0', $streamSelectLoop);
         $this->processPool = new ProcessPool($tcpServer);
@@ -118,20 +108,19 @@ public function process(
 
         $systemErrorsCount = 0;
         $reachedSystemErrorsCountLimit = false;
-        $totalChanged = 0;
         $scheduledFilesCount = array_sum(array_map(count(...), $jobs));
         $processedFilesCount = 0;
 
         $handleErrorCallable = function (Throwable $throwable) use (
-            &$systemErrors,
+            $parallelResultCollector,
             &$systemErrorsCount,
             &$reachedSystemErrorsCountLimit
         ): void {
-            $systemErrors[] = new SystemError(
+            $parallelResultCollector->collectSystemError(new SystemError(
                 $throwable->getMessage(),
                 $throwable->getFile(),
                 $throwable->getLine(),
-            );
+            ));
 
             ++$systemErrorsCount;
             $reachedSystemErrorsCountLimit = true;
@@ -147,9 +136,7 @@ public function process(
         $fileChunksBudgetPerProcess = [];
 
         $processSpawner = function () use (
-            &$systemErrors,
-            &$fileDiffs,
-            &$usedSkips,
+            $parallelResultCollector,
             &$jobs,
             $postFileCallback,
             &$systemErrorsCount,
@@ -162,7 +149,6 @@ public function process(
             $handleErrorCallable,
             &$fileChunksBudgetPerProcess,
             &$processSpawner,
-            &$totalChanged,
             &$processedFilesCount
         ): void {
             $processIdentifier = Random::generate();
@@ -182,17 +168,14 @@ public function process(
                 // 1. callable on data
                 function (array $json) use (
                     $parallelProcess,
-                    &$systemErrors,
-                    &$fileDiffs,
-                    &$usedSkips,
+                    $parallelResultCollector,
                     &$jobs,
                     $postFileCallback,
                     &$systemErrorsCount,
                     &$reachedSystemErrorsCountLimit,
                     $processIdentifier,
                     &$fileChunksBudgetPerProcess,
                     &$processSpawner,
-                    &$totalChanged,
                     &$processedFilesCount
                 ): void {
                     /** @var array{
@@ -203,31 +186,9 @@ function (array $json) use (
                      *      system_errors_count: int,
                      *      used_skips: array<string, string[]>
                      * } $json */
-                    $totalChanged += $json[Bridge::TOTAL_CHANGED];
-
-                    foreach ($json[Bridge::USED_SKIPS] as $skip => $paths) {
-                        $usedSkips[$skip] ??= [];
-                        foreach ($paths as $path) {
-                            $usedSkips[$skip][$path] = true;
-                        }
-                    }
-
-                    // decode arrays to objects
-                    foreach ($json[Bridge::SYSTEM_ERRORS] as $jsonError) {
-                        if (is_string($jsonError)) {
-                            $systemErrors[] = new SystemError('System error: ' . $jsonError);
-                            continue;
-                        }
-
-                        $systemErrors[] = SystemError::decode($jsonError);
-                    }
-
-                    foreach ($json[Bridge::FILE_DIFFS] as $jsonFileDiff) {
-                        $fileDiffs[] = FileDiff::decode($jsonFileDiff);
-                    }
-
-                    $postFileCallback($json[Bridge::FILES_COUNT]);
-                    $processedFilesCount += $json[Bridge::FILES_COUNT];
+                    $filesCount = $parallelResultCollector->collectWorkerResult($json);
+                    $postFileCallback($filesCount);
+                    $processedFilesCount += $filesCount;
 
                     $systemErrorsCount += $json[Bridge::SYSTEM_ERRORS_COUNT];
                     if ($systemErrorsCount >= self::SYSTEM_ERROR_LIMIT) {
@@ -261,7 +222,7 @@ function (array $json) use (
                 $handleErrorCallable,
 
                 // 3. callable on exit
-                function ($exitCode, string $stdErr) use (&$systemErrors, $processIdentifier): void {
+                function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIdentifier): void {
                     $this->processPool->tryQuitProcess($processIdentifier);
                     if ($exitCode === Command::SUCCESS) {
                         return;
@@ -271,7 +232,9 @@ function ($exitCode, string $stdErr) use (&$systemErrors, $processIdentifier): v
                         return;
                     }
 
-                    $systemErrors[] = new SystemError('Child process error: ' . $stdErr);
+                    $parallelResultCollector->collectSystemError(new SystemError(
+                        'Child process error: ' . $stdErr
+                    ));
                 }
             );
 
@@ -290,26 +253,21 @@ function ($exitCode, string $stdErr) use (&$systemErrors, $processIdentifier): v
         $streamSelectLoop->run();
 
         if ($reachedSystemErrorsCountLimit) {
-            $systemErrors[] = new SystemError(sprintf(
+            $parallelResultCollector->collectSystemError(new SystemError(sprintf(
                 'Reached system errors count limit of %d, exiting...',
                 self::SYSTEM_ERROR_LIMIT
-            ));
+            )));
         }
 
         // 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) {
-            $systemErrors[] = new SystemError(sprintf(
+            $parallelResultCollector->collectSystemError(new SystemError(sprintf(
                 'Some parallel jobs have not finished, results for %d of %d files are missing',
                 $scheduledFilesCount - $processedFilesCount,
                 $scheduledFilesCount
-            ));
-        }
-
-        $mergedUsedSkips = [];
-        foreach ($usedSkips as $skip => $paths) {
-            $mergedUsedSkips[$skip] = array_keys($paths);
+            )));
         }
 
-        return new ProcessResult($systemErrors, $fileDiffs, $totalChanged, $mergedUsedSkips);
+        return $parallelResultCollector->createProcessResult();
     }
 }