From 8ff22f2b1b688c79154010e9097caeaff2315cd0 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Mon, 12 Oct 2026 19:35:35 +0700 Subject: [PATCH 1/3] Make LPT bucket scheduling on parallel by default --- .github/workflows/build_scoped_rector.yaml | 2 +- .github/workflows/rector.yaml | 2 +- PARALLEL_LPT_POC_NOTES.md | 349 ------------------ .../expected-output.diff | 20 +- e2e/parallel-unfinished-jobs/rector.php | 2 +- e2e/parallel-unfinished-jobs/src/File2.php | 2 + .../expected-output.diff | 72 ++-- e2e/parallel-worker-respawn/rector.php | 2 +- phpstan.neon | 8 +- src/Application/ApplicationFileProcessor.php | 22 -- src/Configuration/Option.php | 6 - src/Console/ProcessConfigureDecorator.php | 7 - .../Application/ParallelFileProcessor.php | 146 +++++--- .../ExperimentalParallelFileProcessor.php | 311 ---------------- .../Experimental/LptScheduleFactory.php | 87 ----- .../ValueObject/BucketSchedule.php | 31 -- src/Parallel/ScheduleFactory.php | 69 +++- src/Parallel/ValueObject/Schedule.php | 19 +- 18 files changed, 221 insertions(+), 936 deletions(-) delete mode 100644 PARALLEL_LPT_POC_NOTES.md delete mode 100644 src/Parallel/Experimental/ExperimentalParallelFileProcessor.php delete mode 100644 src/Parallel/Experimental/LptScheduleFactory.php delete mode 100644 src/Parallel/Experimental/ValueObject/BucketSchedule.php 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/PARALLEL_LPT_POC_NOTES.md b/PARALLEL_LPT_POC_NOTES.md deleted file mode 100644 index e4e480922f7..00000000000 --- a/PARALLEL_LPT_POC_NOTES.md +++ /dev/null @@ -1,349 +0,0 @@ -# PoC: LPT bucket scheduling for parallel run - -Branch: `poc/parallel-lpt-buckets` (local only, **not committed** — no tests yet) - -Opt-in flag: `--lpt`. Without it nothing changes. - -Source: https://github.com/rectorphp/rector-src/issues/8494#issuecomment-5664864017 (@samsonasik) - -## What samsonasik proposed - -1. LPT (Longest Processing Time first): sort files largest-first by byte size, assign each file - to the worker with the smallest total byte load so far. -2. One worker per bucket, started once. -3. Poll workers until no pending data. No timeout tweak, no job size tweak, no repeated - close-open of workers. - -Reference: `boundwize/structarmed` `src/Analyser/Parallel/ParallelAnalysisNodeExtractor.php`. - -## What was actually implemented here - -Faithful to 1 + 3, with one deliberate deviation on 2. - -| | StructArmed | this PoC | -| --- | --- | --- | -| bucketing | LPT by filesize | LPT by filesize (same greedy) | -| assignment | static, 1 bucket per worker | static, **plus stealing once a worker runs dry** | -| dispatch | whole bucket in one shot | bucket sub-chunked into `jobSize` jobs | -| worker lifetime | one process per bucket, never respawned | one process per bucket, never respawned | -| transport | `proc_open` + stdout polling | unchanged (ReactPHP + NDJSON over TCP) | - -### Why the bucket is still sub-chunked - -Rector's worker reports back once per job, and that single message does two things: - -- advances the progress bar (`Bridge::FILES_COUNT`), -- cancels/re-arms the per-job timeout (`ParallelProcess::request()`). - -A true one-shot bucket would mean: progress bar frozen at 0 % until a worker finishes its whole -bucket, and `parallelJobTimeout` (default 120 s) applied to the *entire* bucket — i.e. guaranteed -`Child process timed out after 120 seconds` on any non-trivial project. That is precisely the -failure mode reported in #8494. - -Sub-chunking inside a statically-assigned bucket keeps the heartbeat while preserving the actual -win: **the worker process is never recycled**, so its warm in-memory caches (PHPStan reflection, -parsed AST, node scope resolver) survive across the whole bucket. - -The round-trip cost of sub-chunking turned out to be far cheaper than expected — small chunks win -outright, see the `jobSize` section below. - -### The other half of the change: no forced respawn - -`ParallelFileProcessor::MAX_CHUNKS_PER_WORKER = 8` kills a worker after 8 chunks and spawns a fresh -one. At `jobSize: 16` that is a full container boot + cold caches every 128 files. On 13 111 files / -16 workers that is ~100 respawns. Strong suspicion this — not the chunk round-trips — is what makes -`jobSize: 16` (218 s) three times slower than `jobSize: 300` (75 s) in the issue's table. - -In bucket mode the respawn is disabled entirely. **This is the main thing to benchmark and the main -risk** (see below). - -## Files touched - -New, all isolated under `src/Parallel/Experimental/`: - -- `LptScheduleFactory.php` — LPT bucketing, heartbeat sub-chunking, worker-count math -- `ExperimentalParallelFileProcessor.php` — static bucket→worker binding, no respawn -- `ValueObject/BucketSchedule.php` - -Existing files, minimal edits only: - -- `src/Configuration/Option.php` — `LPT` constant -- `src/Console/ProcessConfigureDecorator.php` — registers `--lpt` -- `src/Application/ApplicationFileProcessor.php` — two constructor deps + one branch in `runParallel()` - -Nothing on the default path changed, so a run without the flag is byte-identical to `main`. - -## Measurements - -Machine: **Apple M3 Max, 10 performance + 4 efficiency cores** (`hw.ncpu` = 14, which is what Rector -uses as the worker count). Corpus `vendor/`, 5 746 files. -`--dry-run --clear-cache --no-progress-bar --output-format=json`, runs strictly sequential. - -| scheduler | workers | wall | user | sys | peak RSS | busy cores | -| --- | --: | --: | --: | --: | --: | --: | -| legacy | 14 | 152.2 s | 1043.8 s | 52.7 s | 751 MB | 7.2 | -| legacy | 10 | 164.3 s | 1026.9 s | 37.1 s | 691 MB | 6.5 | -| LPT, static buckets | 14 | 127.8 s | 1014.3 s | 39.6 s | 938 MB | 8.2 | -| LPT, static buckets | 10 | 153.8 s | 961.6 s | 31.9 s | 1256 MB | 6.5 | -| LPT + work stealing | 14 | 122.1 s | 1089.2 s | 52.3 s | 902 MB | 9.3 | -| LPT + work stealing | 12 | 113.6 s | 999.1 s | 33.3 s | 1075 MB | 9.1 | -| LPT + work stealing | 10 | **113.2 s** | 930.6 s | 31.5 s | 1156 MB | 8.5 | - -All seven runs produce **3 338 identical diffs** (same sha over the file-sorted diff list) and the -same 2 errors. The scheduler does not change the outcome. - -"busy cores" = `(user + sys) / real`, i.e. how many of the 14 were actually kept busy. - -Every row above is at `jobSize: 16`, which was the hardcoded chunk at the time. `jobSize` turned out -to matter more than anything in this table — see its own section below. - -Headline: best legacy **152.2 s** → best `--lpt` **90.7 s** (14 workers, `jobSize: 2`), i.e. **-40 %**. -At the same worker count and the same `jobSize: 16`, it is 152.2 s → 113.4 s, **-25 %**. - -### Work stealing is the load-bearing part, not LPT - -At 10 workers — all of them on performance cores, so the hardware is homogeneous — static buckets -give 153.8 s and stealing gives 113.2 s. That is **-26 % from stealing alone**, on hardware where -static assignment had no excuse. - -So the E-core theory is at best a secondary effect. The primary one is that **byte size is a bad -predictor of processing cost**: a bucket balanced perfectly by bytes still finishes at wildly -different times. LPT bucketing on its own (samsonasik's proposal as written) lands at 153.8 s vs -legacy's 164.3 s at the same worker count — a real but modest gain. Nearly all of the win comes from -keeping the workers alive and letting idle ones take over work. - -This also sharpens item 1 in "what is left": a timing-based cost model would make the initial -buckets right, but stealing is what makes the schedule robust when the model is wrong. Both, not -either. - -### Worker count: a red herring once the chunk size is right - -Measured at `jobSize: 16`, dropping from 14 to 10 workers looked like a win (122.1 → 113.2), and -legacy got worse at 10 (152.2 → 164.3). That suggested tuning the worker count, and that -`CpuCoreCountProvider` counting logical cores was a problem on big.LITTLE hardware. - -**That was an artefact of the coarse chunk.** At `jobSize: 4` the ordering flips: 14 workers give -99.2 s, 12 workers give 107.6 s. With fine-grained stealing an efficiency core simply takes fewer -chunks and contributes instead of straggling, so there is nothing left to compensate for by starving -the pool. - -Conclusion: fix the granularity and the worker-count question mostly disappears. `hw.ncpu` is a fine -default. Do not tune the worker count before tuning `jobSize` — you will tune against a symptom. - -### jobSize: was inert, now it is the most important knob - -Originally this PoC hardcoded the chunk at 16 files and let `jobSize` feed only the worker-count -formula `min(ceil(files / jobSize), cpuCores, maxProcesses)`. That makes it inert on any real -project: once `ceil(files / jobSize) >= cpuCores` — from ~224 files at 14 cores and `jobSize: 16` — -it stops changing anything. On a 13 k file project it did nothing below `jobSize` ~936. - -`jobSize` is now the chunk a worker is handed per request (it still implies at least one whole job -per worker). It therefore controls three things at once: **the granularity at which an idle worker -can steal work**, the heartbeat that advances the progress bar, and the window the per-job timeout -measures. - -14 workers, LPT + stealing, `vendor/`. Run-to-run spread across batches is ~5 %, so the two figures -for `jobSize: 4` are the same measurement taken twice. - -| jobSize | wall | diffs | | -| --: | --: | --: | --- | -| 1 | 98.8 s | 3 338 | round-trip overhead takes over | -| 2 | **90.7 / 91.4 s** | 3 338 | optimum | -| 4 | 96.7 / 99.2 s | 3 338 | | -| 8 | 104.6 s | 3 338 | | -| 16 | 113.4 s | 3 338 | today's default | -| 50 | 118.1 s | 3 338 | | -| 150 | 127.5 s | **3 237** | **timed out, results truncated** | - -The curve is U-shaped with a clear floor at **`jobSize: 2`** — reproduced twice at 90.7 s and 91.4 s, -so the ~6 % gap to `jobSize: 4` is outside run-to-run noise. Below that, at `jobSize: 1`, the round -trips finally cost more than the balance they buy (and `user` time peaks at 1 173 s, the highest of -any run). - -This **inverts the recommendation in #8494**. There, raising `jobSize` was the only lever that -reduced respawn churn, which is why 150 and 300 looked good. With respawns gone, a large `jobSize` -buys nothing and only coarsens stealing — and stealing is what carries the win. - -Caveat before generalising: the optimum is a balance between round-trip cost and stealing -granularity, and both scale with the corpus. A heavier rule set makes each file cost more, which -shifts the floor upward. Treat "2" as the answer for this corpus on this machine, and re-measure -the 2-8 range elsewhere rather than hardcoding it. - -### A large jobSize both fails the build and truncates the result - -`jobSize: 150` hit `Child process timed out after 120 seconds`. That calls `handleErrorCallable` → -`quitAll()`, and the run finished **reporting 3 237 diffs instead of 3 338** — 101 missing. - -The build does go red: `ProcessCommand` returns `ExitCode::FAILURE` whenever there are system errors -(`src/Console/Command/ProcessCommand.php:232`), which matches the `exit 1` reported in #8494. The -extra hazard is that the *output* is silently short by 101 diffs on top of that, so anyone who -tolerates or filters system errors in CI loses findings without a signal. - -Worth reporting upstream separately: the message printed in that case is -`Reached system errors count limit of 50, exiting...` even though only 4 errors occurred, because -`ParallelFileProcessor::$handleErrorCallable` sets `$reachedSystemErrorsCountLimit = true` -unconditionally on *any* error (`src/Parallel/Application/ParallelFileProcessor.php:134`). This is a -pre-existing bug on `main`, faithfully copied into the experimental processor. It hides the real -cause behind a wrong one — exactly the kind of thing that makes a timeout hard to diagnose. - -### Memory is the consistent cost - -Peak RSS is higher in every experimental run: 691-751 MB legacy vs 902-1256 MB experimental, worst -case **+82 %** (1256 MB vs 691 MB at 10 workers). It gets worse as workers get fewer, because each -worker then holds more files and nothing ever releases. This is the direct price of removing the -respawn and the main reason a memory-driven recycle is worth building (see below). - -Control run on `rules/` (802 files, ~57 per worker): 6.17 s legacy vs 6.67 s experimental — pure -noise. At that size a worker never reaches the `MAX_CHUNKS_PER_WORKER × jobSize = 128` respawn -threshold, so both schedulers do exactly the same thing. **The gain appears precisely where the -respawns do**, which supports reading #8494's table as respawn churn rather than round-trip overhead. - -Any project past ~2 000 files sits beyond that threshold on a 14-core machine. - -### Why work stealing matters here - -Static LPT assumes every worker is equally fast. On a big.LITTLE CPU it is not: 4 of the 14 workers -land on efficiency cores that are several times slower, so their bucket finishes long after the rest -and the run waits for them. Legacy's shared job pool corrected for that automatically; static -bucketing throws that correction away. Stealing puts it back while keeping the no-respawn win — -hence it beats both. - -Consequence for benchmarking: **the split between static and stealing depends on the hardware.** On a -homogeneous CI runner static buckets should lose much less. Worth measuring on the target CI too, not -only on a laptop. `RECTOR_LPT_NO_STEAL=1` turns stealing off for that A/B (temporary knob). - -### What is left on the table - -~9.3 of 14 cores busy means roughly a third of the machine is still idle. Ranked by expected value: - -1. **Memory-driven worker recycle.** Not a speed item — the one that decides whether this is shippable - at all. Report `memory_get_peak_usage()` in the result payload and recycle a worker above a - threshold, handing its remaining chunks to the replacement. Cheap with sub-chunked buckets, - impossible with true one-shot buckets. -2. **Cost model.** Byte size mispredicts badly (see above). Rector already writes a per-file cache - entry; persisting the measured processing time there and using it as the LPT weight on the next - run would make the first assignment right, with byte size as the cold-start fallback. Stealing - already absorbs most of the error, so measure before assuming this is worth much on top. -3. **Steal policy.** Currently: take the largest remaining chunk from whoever has the most left. - Untuned. Stealing the *smallest* chunk near the end of a run avoids creating a fresh straggler; - a smaller chunk size for the tail would do the same. Given stealing is worth 26 %, its policy is - worth an experiment. -4. **Cache write contention.** `sys` is 32-53 s, up to ~0.4 cores of kernel time; Rector writes one - cache file per processed file, from every worker into one directory. Batching those writes per - chunk is outside the scheduler and may be worth more than further scheduling work. -5. **Round-trip idle.** A worker sits idle between finishing a chunk and being handed the next one by - the single-threaded main loop. Cheap fix: a one-chunk prefetch buffer. Bounded upside — `--no-diffs` - and full-JSON runs differ little, so the main loop is not saturated. -6. **Worker boot.** 14 container boots, concurrent, ~1 s out of 113. Not worth chasing. - -## How to A/B benchmark on a real project - -The flag is opt-in, so baseline and experiment differ by one argument — no branch switching: - -```bash -vendor/bin/rector process --dry-run --clear-cache --no-progress-bar -``` - -```bash -vendor/bin/rector process --dry-run --clear-cache --no-progress-bar --lpt -``` - -Run them sequentially, never in parallel, and keep `--clear-cache` on both (or off on both). - -Suggested matrix: - -| run | flag | `jobSize` | why | -| --- | --- | --: | --- | -| baseline | — | 16 | today's default | -| baseline tuned | — | 150 | what #8494 recommends raising it to | -| lpt | `--lpt` | 16 | same knob as the default, scheduler is the only variable | -| lpt, fine chunks | `--lpt` | 4 | fastest here; expected best | -| lpt, coarse chunks | `--lpt` | 150 | expected worst, and watch for the timeout | - -**`jobSize` means something different under `--lpt`.** On the default path it is the chunk any worker -may pull from a shared pool. Under `--lpt` it is the chunk handed to a worker from its own bucket, -which makes it the granularity at which an idle worker can steal from a busy one — the setting that -turned out to matter most. It still implies at least one whole job per worker, so it also caps the -worker count on small file sets, exactly as before. - -Do not carry a tuned `jobSize` across the two paths: the optimum moves in opposite directions. - -Watch, besides wall time: - -- peak RSS (`/usr/bin/time -l` on macOS reports the largest single child) — removing the respawn is a - memory trade -- any `Child process timed out after N seconds` -- CPU % — a higher number at equal user time means better balance -- diffs identical across runs (`--output-format=json`, sort by file, compare) — the correctness gate - -## Open risks / what is worth testing later - -Nothing below is covered by a test yet — the user asked for a PoC first. - -### Correctness - -- [ ] `ScheduleFactory`: every input file lands in exactly one bucket, exactly once (no loss, no dupes). - Easy to get wrong with the greedy loop. -- [ ] `ScheduleFactory`: duplicate file paths in the input list must not collapse (that is why the - implementation sorts a list of tuples rather than using paths as array keys). -- [ ] `ScheduleFactory`: numeric-looking paths (e.g. `/tmp/123`) must not be cast to int array keys. -- [ ] `ScheduleFactory`: `$files === []` and `count($files) < $numberOfProcesses` → no empty buckets - handed to `ParallelFileProcessor` (an empty job would trip `Assert::notEmpty()` in `WorkerCommand`). -- [ ] `ScheduleFactory`: non-existent / unreadable file → size 0, no warning, no crash. -- [ ] worker count math is unchanged vs. legacy: `min(ceil(count/jobSize), cpuCores, maxProcesses)`. -- [ ] `ParallelFileProcessor`: a worker only ever receives files from its own bucket. -- [ ] `ParallelFileProcessor`: all buckets drain; the run does not hang when one bucket is empty. -- [ ] end-to-end: file diffs produced in bucket mode == legacy mode == non-parallel mode. - -### Failure handling - -- [ ] a worker that dies mid-bucket: its remaining sub-chunks are currently **lost** (legacy mode had - the same hole, but respawn masked it). Decide: redistribute, or fail loudly. -- [ ] `SYSTEM_ERROR_LIMIT` / `quitAll()` path still terminates cleanly with static buckets. -- [ ] timeout still fires for a genuinely hung worker (heartbeat must not accidentally mask it). - -### Memory (the big one) - -- [ ] a worker now processes its whole bucket without recycling. On a 13 k file project that is - ~800 files in one process vs. 128 today. Measure peak RSS; consider a memory-driven respawn - (`memory_get_peak_usage()` reported in the result payload → recycle above a threshold and hand - the remaining sub-chunks to the fresh worker). This is *easy* with sub-chunked buckets and - *impossible* with true one-shot buckets — a point in favour of the deviation above. - -### Scheduling quality - -- [ ] byte size is a proxy for work, not work itself. A 2 kB file hitting an expensive rule can cost - more than a 50 kB one. Worth measuring per-worker finish times: if the spread is wide, LPT on - bytes is not enough and work-stealing (idle worker pulls from the fattest remaining bucket) - should be added. Ordering largest-first already helps, since the tail is fine-grained. -- [ ] `filesize()` is called once per file in the main process (13 k stats). Measure; the file list - may already be able to carry sizes. -- [ ] does LPT interact badly with `--max-changes` (early break inside `processFiles`)? - -### Cleanup before this could become a PR - -- [ ] **Give the bucket path its own chunk-size default, separate from `PARALLEL_JOB_SIZE`.** - Right now `--lpt` reuses the shared `jobSize`, whose default is 16 - (`src/Config/RectorConfig.php:137` and `src/Configuration/RectorConfigBuilder.php:106`), - untouched by this PoC. Two consequences: - - `--lpt` alone gives 113.4 s, not the 90.7 s headline. Reaching the best number needs an - explicit `withParallel(jobSize: 2)`. - - The optimum moves in **opposite directions** on the two paths — down to ~2 under `--lpt`, - up towards 100-300 on the default path per #8494 — so one shared default cannot serve both, - and lowering the shared one would hurt every run without the flag. - - A dedicated default for the bucket path lets each side sit at its own optimum. Pick it by - measuring on more than one corpus first: "2" comes from a single repo on a single machine, and - a heavier rule set raises the per-file cost and shifts the floor upward. -- [ ] drop the `--lpt` flag and `RECTOR_LPT_NO_STEAL` env var, or promote them to real options -- [ ] `MAX_CHUNKS_PER_WORKER` is dead in bucket mode — decide its fate (see the memory-driven - recycle item above, which would give it a real job again) -- [x] the shared result accumulation is extracted into `ParallelResultCollector` — this is what the - `duplicated_code` CI gate caught (169 tokens against a 150 threshold). Default-path behaviour - verified unchanged: `vendor/` before and after gives 3 338 diffs, identical sha, same 2 errors. - The system-error counter stayed in the processors on purpose — three sites append an error but - only two increment it, and unifying that would move when the 50-error limit trips. -- [ ] fold `ExperimentalParallelFileProcessor` back into `ParallelFileProcessor` — the TCP wiring and - the spawn loop are still parallel structures, kept apart only so the default path stays - provably untouched -- [ ] `composer check-cs`, `composer phpstan`, `vendor/bin/phpunit` diff --git a/e2e/parallel-unfinished-jobs/expected-output.diff b/e2e/parallel-unfinished-jobs/expected-output.diff index 59be2f290fe..9cc3a503e90 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,23 @@ Applied rules: * LongArrayToShortArrayRector +2) src/File4.php:1 + + ---------- begin diff ---------- +@@ Line 1 @@ + parallel(maxNumberOfProcess: 1, jobSize: 1); diff --git a/e2e/parallel-unfinished-jobs/src/File2.php b/e2e/parallel-unfinished-jobs/src/File2.php index e91165552de..b4e086d65e4 100644 --- a/e2e/parallel-unfinished-jobs/src/File2.php +++ b/e2e/parallel-unfinished-jobs/src/File2.php @@ -1,3 +1,5 @@ parallel(maxNumberOfProcess: 1, jobSize: 1); $rectorConfig->rules([ diff --git a/phpstan.neon b/phpstan.neon index 9d5e1d70a33..a19977c6e96 100644 --- a/phpstan.neon +++ b/phpstan.neon @@ -192,12 +192,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 @@ -361,7 +355,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..b14914ff269 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,23 +268,6 @@ 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); // mimics see https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-387b8f04e0db7a06678eb52ce0c0d0aff73e0d7d8fc5df834d0a5fbec198e5daR139 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..437dd0a3c48 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -30,22 +30,15 @@ use Throwable; /** - * Inspired from @see - * https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-39c7a3b0cbb217bbfff96fbb454e6e5e60c74cf92fbb0f9d246b8bebbaad2bb0 - * - * https://github.com/phpstan/phpstan-src/commit/b84acd2e3eadf66189a64fdbc6dd18ff76323f67#diff-7f625777f1ce5384046df08abffd6c911cfbb1cfc8fcb2bdeaf78f337689e3e2R150 + * Workers keep their caches for the whole run and process their own balanced bucket first. + * Idle workers take pending chunks from the largest remaining queue. + * Each chunk advances progress and re-arms the per-job 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, @@ -61,22 +54,35 @@ public function process( 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(...), + $schedule->getJobsPerWorker() + ); - // basic properties setup - $numberOfProcesses = $schedule->getNumberOfProcesses(); + /** @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 (&$jobs): void { + $tcpServer->on(ReactEvent::CONNECTION, function (ConnectionInterface $connection) use ( + &$jobsPerWorker, + &$bucketKeyByIdentifier + ): void { $inDecoder = new Decoder($connection, true, StreamFormat::DEPTH, 0, StreamFormat::MAX_LENGTH); $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 +92,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, @@ -106,10 +111,13 @@ public function process( /** @var int $serverPort */ $serverPort = parse_url($serverAddress, PHP_URL_PORT); + $scheduledFilesCount = 0; + foreach ($jobsPerWorker as $jobs) { + $scheduledFilesCount += array_sum(array_map(count(...), $jobs)); + } + $processedFilesCount = 0; $systemErrorsCount = 0; $reachedSystemErrorsCountLimit = false; - $scheduledFilesCount = array_sum(array_map(count(...), $jobs)); - $processedFilesCount = 0; $handleErrorCallable = function (Throwable $throwable) use ( $parallelResultCollector, @@ -126,18 +134,17 @@ 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, @@ -147,11 +154,11 @@ public function process( $streamSelectLoop, $timeoutInSeconds, $handleErrorCallable, - &$fileChunksBudgetPerProcess, - &$processSpawner, &$processedFilesCount ): void { $processIdentifier = Random::generate(); + $bucketKeyByIdentifier[$processIdentifier] = $bucketKey; + $workerCommandLine = $this->workerCommandLineFactory->create( $mainScript, ProcessCommand::class, @@ -160,7 +167,6 @@ public function process( $processIdentifier, $serverPort, ); - $fileChunksBudgetPerProcess[$processIdentifier] = self::MAX_CHUNKS_PER_WORKER; $parallelProcess = new ParallelProcess($workerCommandLine, $streamSelectLoop, $timeoutInSeconds); @@ -169,13 +175,12 @@ public function process( function (array $json) use ( $parallelProcess, $parallelResultCollector, - &$jobs, + &$jobsPerWorker, + &$bucketKeyByIdentifier, $postFileCallback, &$systemErrorsCount, &$reachedSystemErrorsCountLimit, $processIdentifier, - &$fileChunksBudgetPerProcess, - &$processSpawner, &$processedFilesCount ): void { /** @var array{ @@ -197,25 +202,17 @@ function (array $json) use ( return; } - if ($jobs === []) { - $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)(); + $jobsChunk = $this->popJobChunk($processIdentifier, $jobsPerWorker, $bucketKeyByIdentifier); + if ($jobsChunk === null) { + // bucket done, this worker has nothing left to do $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 +238,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,7 +251,7 @@ 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 + // Workers may exit without reporting their assigned files. if ($processedFilesCount < $scheduledFilesCount) { $parallelResultCollector->collectSystemError(new SystemError(sprintf( 'Some parallel jobs have not finished, results for %d of %d files are missing', @@ -270,4 +262,54 @@ function ($exitCode, string $stdErr) use ($parallelResultCollector, $processIden 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 $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/Experimental/ValueObject/BucketSchedule.php b/src/Parallel/Experimental/ValueObject/BucketSchedule.php deleted file mode 100644 index dff8464d651..00000000000 --- a/src/Parallel/Experimental/ValueObject/BucketSchedule.php +++ /dev/null @@ -1,31 +0,0 @@ ->> $jobsPerWorker - */ - public function __construct( - private array $jobsPerWorker - ) { - } - - /** - * @return array>> - */ - public function getJobsPerWorker(): array - { - return $this->jobsPerWorker; - } -} diff --git a/src/Parallel/ScheduleFactory.php b/src/Parallel/ScheduleFactory.php index d8c83d8cacb..e702089caa1 100644 --- a/src/Parallel/ScheduleFactory.php +++ b/src/Parallel/ScheduleFactory.php @@ -8,23 +8,76 @@ use Webmozart\Assert\Assert; /** - * Used from - * https://github.com/phpstan/phpstan-src/blob/9124c66dcc55a222e21b1717ba5f60771f7dda92/src/Parallel/Scheduler.php + * Balances files by byte size into one bucket per worker using LPT (Longest Processing Time first). + * + * 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): Schedule { Assert::positiveInteger($jobSize); + Assert::notEmpty($filePaths); - $jobs = array_chunk($files, $jobSize); - $numberOfProcesses = min(count($jobs), $cpuCores); + // 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 Schedule($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/ValueObject/Schedule.php b/src/Parallel/ValueObject/Schedule.php index c66797e9e59..3693a1b1d8c 100644 --- a/src/Parallel/ValueObject/Schedule.php +++ b/src/Parallel/ValueObject/Schedule.php @@ -5,30 +5,23 @@ namespace Rector\Parallel\ValueObject; /** - * From - * https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-bc84213b079ef3456caece03c00ba34c07886dcae12180cd1192fbb223d65b15 + * Each worker owns a queue of jobs, balanced up front by @see \Rector\Parallel\ScheduleFactory. */ final readonly class Schedule { /** - * @param array> $jobs + * @param array>> $jobsPerWorker */ public function __construct( - private int $numberOfProcesses, - private array $jobs + private array $jobsPerWorker ) { } - public function getNumberOfProcesses(): int - { - return $this->numberOfProcesses; - } - /** - * @return array> + * @return array>> */ - public function getJobs(): array + public function getJobsPerWorker(): array { - return $this->jobs; + return $this->jobsPerWorker; } } From ced4435b2e9fa4372dd342f17ad998e7baaf8207 Mon Sep 17 00:00:00 2001 From: GitHub Action Date: Mon, 12 Oct 2026 12:40:03 +0000 Subject: [PATCH 2/3] [ci-review] Rector Rectify --- src/Parallel/Application/ParallelFileProcessor.php | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Parallel/Application/ParallelFileProcessor.php b/src/Parallel/Application/ParallelFileProcessor.php index 437dd0a3c48..ccce30cc841 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -115,6 +115,7 @@ public function process( foreach ($jobsPerWorker as $jobs) { $scheduledFilesCount += array_sum(array_map(count(...), $jobs)); } + $processedFilesCount = 0; $systemErrorsCount = 0; $reachedSystemErrorsCountLimit = false; From 973ca80ea1826d7de743ba74e50b970a775e15e2 Mon Sep 17 00:00:00 2001 From: Abdul Malik Ikhsan Date: Mon, 12 Oct 2026 19:43:53 +0700 Subject: [PATCH 3/3] more proper comment --- e2e/parallel-unfinished-jobs/rector.php | 2 +- e2e/parallel-unfinished-jobs/src/File2.php | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/e2e/parallel-unfinished-jobs/rector.php b/e2e/parallel-unfinished-jobs/rector.php index 998b678ff72..9bc9582230d 100644 --- a/e2e/parallel-unfinished-jobs/rector.php +++ b/e2e/parallel-unfinished-jobs/rector.php @@ -14,7 +14,7 @@ ]); // size-based scheduling processes File3.php last, where the only worker exits without reporting it; - // the syntax error in File2.php must not hide that + // expect both File2.php's syntax error and a separate error for File3.php's missing result $rectorConfig->parallel(maxNumberOfProcess: 1, jobSize: 1); $rectorConfig->rules([ diff --git a/e2e/parallel-unfinished-jobs/src/File2.php b/e2e/parallel-unfinished-jobs/src/File2.php index b4e086d65e4..62e39425526 100644 --- a/e2e/parallel-unfinished-jobs/src/File2.php +++ b/e2e/parallel-unfinished-jobs/src/File2.php @@ -2,4 +2,4 @@ $items = ; -// Syntax errors must not hide unfinished parallel jobs. +// Keep this fixture larger than File3.php so LPT reports this syntax error before that worker exits.