POC: Parallel LPT buckets

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 UsedSkipCollector $usedSkipCollector,
+        // @experimental, see --lpt
+        private readonly LptScheduleFactory $lptScheduleFactory,
+        private readonly ExperimentalParallelFileProcessor $experimentalParallelFileProcessor,
     ) {
     }
 
@@ -255,18 +260,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

@@ -24,7 +24,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;
@@ -67,16 +66,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);
@@ -117,18 +107,17 @@ public function process(
 
         $systemErrorsCount = 0;
         $reachedSystemErrorsCountLimit = false;
-        $totalChanged = 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;
@@ -144,9 +133,7 @@ public function process(
         $fileChunksBudgetPerProcess = [];
 
         $processSpawner = function () use (
-            &$systemErrors,
-            &$fileDiffs,
-            &$usedSkips,
+            $parallelResultCollector,
             &$jobs,
             $postFileCallback,
             &$systemErrorsCount,
@@ -158,8 +145,7 @@ public function process(
             $timeoutInSeconds,
             $handleErrorCallable,
             &$fileChunksBudgetPerProcess,
-            &$processSpawner,
-            &$totalChanged
+            &$processSpawner
         ): void {
             $processIdentifier = Random::generate();
             $workerCommandLine = $this->workerCommandLineFactory->create(
@@ -178,17 +164,14 @@ public function process(
                 // 1. callable on data
                 function (array $json) use (
                     $parallelProcess,
-                    &$systemErrors,
-                    &$fileDiffs,
-                    &$usedSkips,
+                    $parallelResultCollector,
                     &$jobs,
                     $postFileCallback,
                     &$systemErrorsCount,
                     &$reachedInternalErrorsCountLimit,
                     $processIdentifier,
                     &$fileChunksBudgetPerProcess,
-                    &$processSpawner,
-                    &$totalChanged
+                    &$processSpawner
                 ): void {
                     /** @var array{
                      *      total_changed: int,
@@ -198,30 +181,7 @@ 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]);
+                    $postFileCallback($parallelResultCollector->collectWorkerResult($json));
 
                     $systemErrorsCount += $json[Bridge::SYSTEM_ERRORS_COUNT];
                     if ($systemErrorsCount >= self::SYSTEM_ERROR_LIMIT) {
@@ -254,7 +214,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;
@@ -264,7 +224,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
+                    ));
                 }
             );
 
@@ -283,17 +245,12 @@ 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
-            ));
-        }
-
-        $mergedUsedSkips = [];
-        foreach ($usedSkips as $skip => $paths) {
-            $mergedUsedSkips[$skip] = array_keys($paths);
+            )));
         }
 
-        return new ProcessResult($systemErrors, $fileDiffs, $totalChanged, $mergedUsedSkips);
+        return $parallelResultCollector->createProcessResult();
     }
 }

src/Parallel/Application/ParallelResultCollector.php

@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Rector\Parallel\Application;
+
+use Rector\Parallel\ValueObject\Bridge;
+use Rector\ValueObject\Error\SystemError;
+use Rector\ValueObject\ProcessResult;
+use Rector\ValueObject\Reporting\FileDiff;
+
+/**
+ * Gathers what the workers report back over the run, so a parallel file processor only has to decide
+ * who gets which files.
+ *
+ * Holds mutable per-run state - create one per process() call, never share it.
+ */
+final class ParallelResultCollector
+{
+    /**
+     * @var SystemError[]
+     */
+    private array $systemErrors = [];
+
+    /**
+     * @var FileDiff[]
+     */
+    private array $fileDiffs = [];
+
+    /**
+     * Paths are kept as keys while collecting, to deduplicate them per skip
+     *
+     * @var array<string, array<string, true>>
+     */
+    private array $usedSkips = [];
+
+    private int $totalChanged = 0;
+
+    /**
+     * @param array{
+     *      total_changed: int,
+     *      system_errors: mixed[],
+     *      file_diffs: array<string, mixed>,
+     *      files_count: int,
+     *      system_errors_count: int,
+     *      used_skips: array<string, string[]>
+     * } $json
+     * @return int files covered by this result, to advance the progress bar
+     */
+    public function collectWorkerResult(array $json): int
+    {
+        $this->totalChanged += $json[Bridge::TOTAL_CHANGED];
+
+        foreach ($json[Bridge::USED_SKIPS] as $skip => $paths) {
+            $this->usedSkips[$skip] ??= [];
+            foreach ($paths as $path) {
+                $this->usedSkips[$skip][$path] = true;
+            }
+        }
+
+        // decode arrays to objects
+        foreach ($json[Bridge::SYSTEM_ERRORS] as $jsonError) {
+            if (is_string($jsonError)) {
+                $this->systemErrors[] = new SystemError('System error: ' . $jsonError);
+                continue;
+            }
+
+            $this->systemErrors[] = SystemError::decode($jsonError);
+        }
+
+        foreach ($json[Bridge::FILE_DIFFS] as $jsonFileDiff) {
+            $this->fileDiffs[] = FileDiff::decode($jsonFileDiff);
+        }
+
+        return $json[Bridge::FILES_COUNT];
+    }
+
+    public function collectSystemError(SystemError $systemError): void
+    {
+        $this->systemErrors[] = $systemError;
+    }
+
+    public function createProcessResult(): ProcessResult
+    {
+        $mergedUsedSkips = [];
+        foreach ($this->usedSkips as $skip => $paths) {
+            $mergedUsedSkips[$skip] = array_keys($paths);
+        }
+
+        return new ProcessResult($this->systemErrors, $this->fileDiffs, $this->totalChanged, $mergedUsedSkips);
+    }
+}