Implement worker auto-scaler

Author: AJenbo

conf/config.neon

@@ -106,7 +106,7 @@ parameters:
 	parallel:
 		jobSize: 20
 		processTimeout: 600.0
-		maximumNumberOfProcesses: 8
+		maximumNumberOfProcesses: auto
 		minimumNumberOfJobsPerProcess: 2
 		buffer: 134217728 # 128 MB
 		loadLimit: 1.0

conf/parametersSchema.neon

@@ -108,7 +108,7 @@ parametersSchema:
 	parallel: structure([
 		jobSize: int(),
 		processTimeout: float(),
-		maximumNumberOfProcesses: int(),
+		maximumNumberOfProcesses: anyOf(int(), 'auto'),
 		minimumNumberOfJobsPerProcess: int(),
 		buffer: int(),
 		loadLimit: schema(float(), nullable())

src/Diagnose/SystemResourcesDiagnoseExtension.php

@@ -0,0 +1,47 @@
+<?php declare(strict_types = 1);
+
+namespace PHPStan\Diagnose;
+
+use PHPStan\Command\Output;
+use PHPStan\DependencyInjection\AutowiredService;
+use PHPStan\Process\CpuCoreCounter;
+use PHPStan\Process\SystemResources;
+use function sprintf;
+
+/**
+ * Reports what PHPStan believes about the machine, so a user who disagrees with the
+ * number of workers it chose can see which input was wrong.
+ */
+#[AutowiredService]
+final class SystemResourcesDiagnoseExtension implements DiagnoseExtension
+{
+
+	public function __construct(
+		private CpuCoreCounter $cpuCoreCounter,
+		private SystemResources $systemResources,
+	)
+	{
+	}
+
+	public function print(Output $output): void
+	{
+		$output->writeLineFormatted('<info>System resources:</info>');
+		$output->writeLineFormatted(sprintf('Detected CPU cores:        %d', $this->cpuCoreCounter->getDetectedNumberOfCpuCores()));
+
+		$quota = $this->systemResources->getCpuQuota();
+		$output->writeLineFormatted(sprintf(
+			'cgroup CPU quota:          %s',
+			$quota === null ? 'none' : sprintf('%d cores', $quota),
+		));
+
+		$output->writeLineFormatted(sprintf('Usable CPU cores:          %d', $this->cpuCoreCounter->getNumberOfCpuCores()));
+
+		$memory = $this->systemResources->getAvailableMemoryBytes();
+		$output->writeLineFormatted(sprintf(
+			'Available memory:          %s',
+			$memory === null ? 'unknown' : sprintf('%.1f GB', $memory / 1024 / 1024 / 1024),
+		));
+		$output->writeLineFormatted('');
+	}
+
+}

src/Parallel/Scheduler.php

@@ -6,6 +6,7 @@
 use PHPStan\DependencyInjection\AutowiredParameter;
 use PHPStan\DependencyInjection\AutowiredService;
 use PHPStan\Diagnose\DiagnoseExtension;
+use PHPStan\Process\SystemResources;
 use function array_values;
 use function ceil;
 use function count;
@@ -19,21 +20,32 @@
 final class Scheduler implements DiagnoseExtension
 {
 
-	/** @var array{int, int, int, int}|null */
+	public const AUTO = 'auto';
+
+	/**
+	 * Used when the platform cannot say how much memory is available. This is the
+	 * historical fixed default: conservative on a big machine, but the only safe
+	 * answer when flying blind.
+	 */
+	private const UNKNOWN_MEMORY_PROCESSES_LIMIT = 8;
+
+	/** @var array{int, int, int, int, string}|null */
 	private ?array $storedData = null;
 
 	/**
 	 * @param positive-int $jobSize
-	 * @param positive-int $maximumNumberOfProcesses
+	 * @param positive-int|self::AUTO $maximumNumberOfProcesses
 	 * @param positive-int $minimumNumberOfJobsPerProcess
 	 */
 	public function __construct(
 		#[AutowiredParameter(ref: '%parallel.jobSize%')]
 		private int $jobSize,
 		#[AutowiredParameter(ref: '%parallel.maximumNumberOfProcesses%')]
-		private int $maximumNumberOfProcesses,
+		private int|string $maximumNumberOfProcesses,
 		#[AutowiredParameter(ref: '%parallel.minimumNumberOfJobsPerProcess%')]
 		private int $minimumNumberOfJobsPerProcess,
+		private SystemResources $systemResources,
+		private WorkerMemoryBudget $workerMemoryBudget,
 	)
 	{
 	}
@@ -78,25 +90,68 @@ public function scheduleWork(
 			$cpuCores,
 		);
 
-		$usedNumberOfProcesses = min($numberOfProcesses, $this->maximumNumberOfProcesses);
-		$this->storedData = [$cpuCores, count($files), count($jobs), $usedNumberOfProcesses];
+		[$maximumNumberOfProcesses, $decision] = $this->resolveMaximumNumberOfProcesses($cpuCores);
+		$usedNumberOfProcesses = min($numberOfProcesses, $maximumNumberOfProcesses);
+		$this->storedData = [$cpuCores, count($files), count($jobs), $usedNumberOfProcesses, $decision];
 
 		return new Schedule($usedNumberOfProcesses, $jobs);
 	}
 
+	/**
+	 * 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
+	 * which input produced it.
+	 *
+	 * @return array{positive-int, string}
+	 */
+	private function resolveMaximumNumberOfProcesses(int $cpuCores): array
+	{
+		if ($this->maximumNumberOfProcesses !== self::AUTO) {
+			return [$this->maximumNumberOfProcesses, 'configured'];
+		}
+
+		$availableMemory = $this->systemResources->getAvailableMemoryBytes();
+		if ($availableMemory === null) {
+			return [
+				self::UNKNOWN_MEMORY_PROCESSES_LIMIT,
+				sprintf('auto, available memory unknown so capped at %d', self::UNKNOWN_MEMORY_PROCESSES_LIMIT),
+			];
+		}
+
+		$affordableProcesses = $this->workerMemoryBudget->getAffordableWorkerCount($availableMemory);
+
+		if ($affordableProcesses < $cpuCores) {
+			return [
+				$affordableProcesses,
+				sprintf(
+					'auto, %d MB available memory fits %d workers of %d MB',
+					(int) ($availableMemory / 1024 / 1024),
+					$affordableProcesses,
+					WorkerMemoryBudget::EXPECTED_WORKER_MEMORY_LIMIT / 1024 / 1024,
+				),
+			];
+		}
+
+		return [
+			max(1, $cpuCores),
+			sprintf('auto, limited by %d usable CPU cores', $cpuCores),
+		];
+	}
+
 	public function print(Output $output): void
 	{
 		if ($this->storedData === null) {
 			return;
 		}
 
-		[$cpuCores, $filesCount, $jobsCount, $usedNumberOfProcesses] = $this->storedData;
+		[$cpuCores, $filesCount, $jobsCount, $usedNumberOfProcesses, $decision] = $this->storedData;
 
 		$output->writeLineFormatted('<info>Parallel processing scheduler:</info>');
 		$output->writeLineFormatted(sprintf('# of detected CPU cores:   %d', $cpuCores));
 		$output->writeLineFormatted(sprintf('# of analysed files:       %d', $filesCount));
 		$output->writeLineFormatted(sprintf('# of jobs:                 %d', $jobsCount));
 		$output->writeLineFormatted(sprintf('# of spawned processes:    %d', $usedNumberOfProcesses));
+		$output->writeLineFormatted(sprintf('Process limit:             %s', $decision));
 		$output->writeLineFormatted('');
 	}
 

src/Parallel/WorkerMemoryBudget.php

@@ -0,0 +1,58 @@
+<?php declare(strict_types = 1);
+
+namespace PHPStan\Parallel;
+
+use PHPStan\DependencyInjection\AutowiredService;
+use function floor;
+use function max;
+
+/**
+ * How many workers the available memory can pay for.
+ *
+ * Decided once, before any worker starts, and not revisited: the pool cannot shrink, so
+ * a width that turns out to be too wide is unrecoverable, and the readings that would
+ * justify widening it later cannot be trusted (see the note on
+ * EXPECTED_WORKER_MEMORY_LIMIT).
+ */
+#[AutowiredService]
+final class WorkerMemoryBudget
+{
+
+	/**
+	 * What one worker is assumed to cost.
+	 *
+	 * Sampled peak PSS per worker across 25 real projects (26 to 5.3k files, four
+	 * workers each, cold): most sit between 100 MB and 300 MB, but analysis-hostile
+	 * codebases go much further - phpBB 487 MB, WordPress 635 MB, and large Laravel
+	 * applications 750-785 MB. File count predicts this badly; what a project does to
+	 * the type system matters more than how big it is, which is why this is a flat
+	 * figure rather than a per-file one.
+	 *
+	 * It is a divisor rather than a ceiling, so a project costing more than this is not
+	 * automatically a problem: at MEMORY_USAGE_FRACTION_LIMIT the pool still fits inside
+	 * available memory for real costs up to this figure divided by that fraction, i.e.
+	 * about 1 GB per worker. Past that, set the count explicitly.
+	 *
+	 * A flat estimate is used rather than the workers' own memory because no reliable
+	 * reading of a *running* worker exists. A worker's footprint climbs in steps -
+	 * level for seconds while it analyses ordinary files, then jumping - so neither a
+	 * plateau nor a completed-job count distinguishes "finished growing" from "between
+	 * growth spurts".
+	 */
+	public const EXPECTED_WORKER_MEMORY_LIMIT = 768 * 1024 * 1024;
+
+	/**
+	 * Share of the memory PHPStan is willing to plan for, leaving the rest to the main
+	 * process, the OS page cache, and whatever else the machine is doing.
+	 */
+	private const MEMORY_USAGE_FRACTION_LIMIT = 0.75;
+
+	/** @return positive-int */
+	public function getAffordableWorkerCount(int $availableBytes): int
+	{
+		$plannable = max(0, $availableBytes) * self::MEMORY_USAGE_FRACTION_LIMIT;
+
+		return max(1, (int) floor($plannable / self::EXPECTED_WORKER_MEMORY_LIMIT));
+	}
+
+}

src/Process/CpuCoreCounter.php

@@ -6,33 +6,61 @@
 use Fidry\CpuCoreCounter\NumberOfCpuCoreNotFound;
 use PHPStan\DependencyInjection\AutowiredParameter;
 use PHPStan\DependencyInjection\AutowiredService;
+use function min;
 
 #[AutowiredService]
 final class CpuCoreCounter
 {
 
 	private ?int $count = null;
 
+	private ?int $detectedCount = null;
+
 	public function __construct(
 		#[AutowiredParameter(ref: '%parallel.loadLimit%')]
 		private ?float $loadLimit,
+		private SystemResources $systemResources,
 	)
 	{
 	}
 
+	/**
+	 * Cores PHPStan may actually use: what the machine reports, capped by the CPU
+	 * quota of the cgroup it runs in.
+	 */
 	public function getNumberOfCpuCores(): int
 	{
 		if ($this->count !== null) {
 			return $this->count;
 		}
 
+		$count = $this->getDetectedNumberOfCpuCores();
+
+		// fidry/cpu-core-counter has no cgroup finder, and its nproc-based default
+		// honours a cpuset affinity mask but not a CFS bandwidth quota, so inside a
+		// `docker run --cpus=2` container it reports the host's core count
+		$quota = $this->systemResources->getCpuQuota();
+		if ($quota !== null) {
+			$count = min($count, $quota);
+		}
+
+		return $this->count = $count;
+	}
+
+	/** What the machine reports before any cgroup quota is applied. */
+	public function getDetectedNumberOfCpuCores(): int
+	{
+		if ($this->detectedCount !== null) {
+			return $this->detectedCount;
+		}
+
 		try {
-			$this->count = (new FidryCpuCoreCounter())->getAvailableForParallelisation(0, null, $this->loadLimit)->availableCpus;
+			$this->detectedCount = (new FidryCpuCoreCounter())->getAvailableForParallelisation(0, null, $this->loadLimit)->availableCpus;
 		} catch (NumberOfCpuCoreNotFound) {
-			$this->count = 1;
+			$this->detectedCount = 1;
 		}
 
-		return $this->count;
+		return $this->detectedCount;
 	}
 
 }