diff --git a/conf/bleedingEdge.neon b/conf/bleedingEdge.neon index 3d9d45eb107..318b968315e 100644 --- a/conf/bleedingEdge.neon +++ b/conf/bleedingEdge.neon @@ -1,5 +1,6 @@ parameters: featureToggles: + adaptiveParallelWorkerCount: true bleedingEdge: true checkNonStringableDynamicAccess: true checkParameterCastableToNumberFunctions: true diff --git a/conf/config.neon b/conf/config.neon index 52f5dd061ec..605946bd343 100644 --- a/conf/config.neon +++ b/conf/config.neon @@ -26,6 +26,7 @@ parameters: tooWideImplicitThrowType: false throwTypeCovariance: false featureToggles: + adaptiveParallelWorkerCount: false bleedingEdge: false checkNonStringableDynamicAccess: false checkParameterCastableToNumberFunctions: false diff --git a/conf/parametersSchema.neon b/conf/parametersSchema.neon index fd9dfd4d5d6..b3a244d490f 100644 --- a/conf/parametersSchema.neon +++ b/conf/parametersSchema.neon @@ -30,6 +30,7 @@ parametersSchema: ]) featureToggles: structure([ bleedingEdge: bool(), + adaptiveParallelWorkerCount: bool(), checkNonStringableDynamicAccess: bool(), checkParameterCastableToNumberFunctions: bool(), skipCheckGenericClasses: listOf(string()), diff --git a/src/Parallel/Scheduler.php b/src/Parallel/Scheduler.php index 604792a3406..6b23c0f621e 100644 --- a/src/Parallel/Scheduler.php +++ b/src/Parallel/Scheduler.php @@ -12,7 +12,9 @@ use function floor; use function max; use function min; +use function round; use function sprintf; +use function sqrt; use function usort; #[AutowiredService] @@ -29,6 +31,20 @@ final class Scheduler implements DiagnoseExtension */ private const AUTO_PROCESSES_LIMIT = 20; + /** + * Workers per square root of the file count, when the adaptive strategy is on. + * + * A worker's startup is a fixed cost, so the number of files it needs to earn + * its keep grows with the size of the run. Deriving the worker count from + * sqrt(files) encodes that: about 12 files per worker at 25 files, about 40 at + * 400, saturating at the usable core count from roughly 800 files upward - so + * full runs are scheduled exactly as before and only small ones change. + */ + private const ADAPTIVE_WORKERS_PER_SQRT_FILE = 0.5; + + /** Below this many files a second worker does not pay for its own startup. */ + private const ADAPTIVE_SECOND_WORKER_FILE_THRESHOLD = 9; + /** @var array{int, int, int, int, string}|null */ private ?array $storedData = null; @@ -44,6 +60,8 @@ public function __construct( private int|string $maximumNumberOfProcesses, #[AutowiredParameter(ref: '%parallel.minimumNumberOfJobsPerProcess%')] private int $minimumNumberOfJobsPerProcess, + #[AutowiredParameter(ref: '%featureToggles.adaptiveParallelWorkerCount%')] + private bool $adaptiveParallelWorkerCount = false, ) { } @@ -70,6 +88,19 @@ public function scheduleWork( usort($files, static fn (string $a, string $b): int => $fileSizes[$b] <=> $fileSizes[$a]); $numberOfJobs = (int) ceil(count($files) / $this->jobSize); + + $desiredNumberOfProcesses = null; + if ($this->adaptiveParallelWorkerCount) { + $desiredNumberOfProcesses = $this->resolveDesiredNumberOfProcesses(count($files), $numberOfJobs, $cpuCores); + + // the spawn loop stops when the queue runs dry, so a worker without a + // job of its own never starts - chunk finely enough to feed them all + $numberOfJobs = min( + count($files), + max($numberOfJobs, $desiredNumberOfProcesses * $this->minimumNumberOfJobsPerProcess), + ); + } + $stripedJobs = []; foreach ($files as $i => $file) { $stripedJobs[$i % $numberOfJobs][] = $file; @@ -83,10 +114,12 @@ public function scheduleWork( unset($stripedJob); $jobs = array_values($stripedJobs); - $numberOfProcesses = min( - max((int) floor(count($jobs) / $this->minimumNumberOfJobsPerProcess), 1), - $cpuCores, - ); + $numberOfProcesses = $desiredNumberOfProcesses !== null + ? min($desiredNumberOfProcesses, count($jobs), $cpuCores) + : min( + max((int) floor(count($jobs) / $this->minimumNumberOfJobsPerProcess), 1), + $cpuCores, + ); [$maximumNumberOfProcesses, $decision] = $this->resolveMaximumNumberOfProcesses($cpuCores); $usedNumberOfProcesses = min($numberOfProcesses, $maximumNumberOfProcesses); @@ -95,6 +128,28 @@ public function scheduleWork( return new Schedule($usedNumberOfProcesses, $jobs); } + /** + * How many workers the file count justifies, before the configured maximum applies. + * + * @return positive-int + */ + private function resolveDesiredNumberOfProcesses(int $numberOfFiles, int $numberOfJobs, int $cpuCores): int + { + if ($numberOfFiles < 1) { + return 1; + } + + $floor = $numberOfFiles >= self::ADAPTIVE_SECOND_WORKER_FILE_THRESHOLD ? 2 : 1; + $fromFileCount = (int) round(self::ADAPTIVE_WORKERS_PER_SQRT_FILE * sqrt($numberOfFiles)); + + // never below what the job count alone already justifies: the point is to stop + // small runs being starved, not to take workers away from large ones, and + // sqrt() dips under the existing formula between roughly 400 and 800 files + $fromJobCount = max((int) floor($numberOfJobs / $this->minimumNumberOfJobsPerProcess), 1); + + return max(1, min(max($floor, $fromFileCount, $fromJobCount), $cpuCores, $numberOfFiles)); + } + /** * How many workers may run at once, and a human-readable account of why - which * `diagnose` prints, because a user who thinks the number is wrong needs to see diff --git a/tests/PHPStan/Parallel/SchedulerTest.php b/tests/PHPStan/Parallel/SchedulerTest.php index de2b132980a..9b2c9bdfa2e 100644 --- a/tests/PHPStan/Parallel/SchedulerTest.php +++ b/tests/PHPStan/Parallel/SchedulerTest.php @@ -195,6 +195,78 @@ public function testAutoIsStillCappedByTheJobCount(): void $this->assertSame(1, $schedule->getNumberOfProcesses()); } + public function testAdaptiveWorkerCountLeavesFullRunsAlone(): void + { + // from ~800 files the sqrt rule saturates at the usable cores, so a full + // run is scheduled exactly as it is without the toggle + foreach ([800, 4524] as $numberOfFiles) { + $files = array_fill(0, $numberOfFiles, 'file.php'); + $legacy = (new Scheduler(20, Scheduler::AUTO, 2))->scheduleWork(14, $files, static fn (string $file): int => 0); + $adaptive = (new Scheduler(20, Scheduler::AUTO, 2, true))->scheduleWork(14, $files, static fn (string $file): int => 0); + + $this->assertSame($legacy->getNumberOfProcesses(), $adaptive->getNumberOfProcesses()); + $this->assertSame($legacy->getJobs(), $adaptive->getJobs()); + } + } + + public function testAdaptiveWorkerCountParallelisesSmallRuns(): void + { + // 50 files is one worker today, because jobSize 20 and 2 jobs per process + // ask for 40 files before a second worker is allowed + $files = array_fill(0, 50, 'file.php'); + $callback = static fn (string $file): int => 0; + + $this->assertSame(1, (new Scheduler(20, Scheduler::AUTO, 2))->scheduleWork(14, $files, $callback)->getNumberOfProcesses()); + $this->assertSame(4, (new Scheduler(20, Scheduler::AUTO, 2, true))->scheduleWork(14, $files, $callback)->getNumberOfProcesses()); + } + + public function testAdaptiveWorkerCountKeepsTinyRunsSerial(): void + { + // below the threshold a second worker does not pay for its own startup + $callback = static fn (string $file): int => 0; + $scheduler = new Scheduler(20, Scheduler::AUTO, 2, true); + + $this->assertSame(1, $scheduler->scheduleWork(14, array_fill(0, 1, 'file.php'), $callback)->getNumberOfProcesses()); + $this->assertSame(1, $scheduler->scheduleWork(14, array_fill(0, 8, 'file.php'), $callback)->getNumberOfProcesses()); + $this->assertSame(2, $scheduler->scheduleWork(14, array_fill(0, 9, 'file.php'), $callback)->getNumberOfProcesses()); + } + + public function testAdaptiveWorkerCountIsNeverBelowTheDefault(): void + { + // the rule exists to stop small runs being starved, never to take workers away + // from large ones - sqrt() alone dips under the existing formula between roughly + // 400 and 800 files, which measured 13% slower at 600 + $callback = static fn (string $file): int => 0; + foreach ([1, 5, 9, 25, 50, 100, 200, 300, 400, 600, 800, 1424, 4524] as $numberOfFiles) { + $files = array_fill(0, $numberOfFiles, 'file.php'); + $legacy = (new Scheduler(20, Scheduler::AUTO, 2))->scheduleWork(14, $files, $callback); + $adaptive = (new Scheduler(20, Scheduler::AUTO, 2, true))->scheduleWork(14, $files, $callback); + + $this->assertGreaterThanOrEqual( + $legacy->getNumberOfProcesses(), + $adaptive->getNumberOfProcesses(), + sprintf('%d files', $numberOfFiles), + ); + } + } + + public function testAdaptiveWorkerCountNeverExceedsTheJobCount(): void + { + // a worker with no job never starts, so the schedule must not claim one + $callback = static fn (string $file): int => 0; + $schedule = (new Scheduler(20, Scheduler::AUTO, 2, true))->scheduleWork(14, array_fill(0, 3, 'file.php'), $callback); + + $this->assertLessThanOrEqual(count($schedule->getJobs()), $schedule->getNumberOfProcesses()); + } + + public function testAdaptiveWorkerCountStillRespectsAnExplicitLimit(): void + { + $callback = static fn (string $file): int => 0; + $schedule = (new Scheduler(20, 2, 2, true))->scheduleWork(14, array_fill(0, 200, 'file.php'), $callback); + + $this->assertSame(2, $schedule->getNumberOfProcesses()); + } + public function testAnExplicitLimitStillWins(): void { $scheduler = new Scheduler(1, 20, 1);