From 66f2a582954e03b949701af0cf634e524b91eac5 Mon Sep 17 00:00:00 2001 From: Tomas Votruba Date: Sat, 29 Aug 2026 10:44:46 +0200 Subject: [PATCH 1/2] Inline symplify/easy-parallel into src/Parallel The symplify/easy-parallel package is deprecated. Move its classes into Rector\Parallel namespace under src/Parallel and drop the dependency. The transitive deps it provided (fidry/cpu-core-counter, react/child-process) are now required directly. Claude-Session: https://claude.ai/code/session_01PMRRBcHZHHSXHYNMFbNAnF --- composer.json | 3 +- src/Application/ApplicationFileProcessor.php | 6 +- .../ValueObject/RectorWithLineChange.php | 2 +- src/Console/Command/WorkerCommand.php | 8 +- .../Application/ParallelFileProcessor.php | 14 +- .../Command/WorkerCommandLineFactory.php | 4 +- .../Contract/SerializableInterface.php | 15 ++ src/Parallel/CpuCoreCountProvider.php | 22 +++ src/Parallel/Enum/Action.php | 14 ++ src/Parallel/Enum/Content.php | 12 ++ src/Parallel/Enum/ReactCommand.php | 12 ++ src/Parallel/Enum/ReactEvent.php | 16 ++ .../ParallelShouldNotHappenException.php | 11 ++ .../CommandFromReflectionFactory.php | 37 +++++ src/Parallel/ScheduleFactory.php | 30 ++++ src/Parallel/ValueObject/ParallelProcess.php | 155 ++++++++++++++++++ src/Parallel/ValueObject/ProcessPool.php | 67 ++++++++ src/Parallel/ValueObject/Schedule.php | 34 ++++ src/ValueObject/Error/SystemError.php | 2 +- src/ValueObject/Reporting/FileDiff.php | 2 +- 20 files changed, 446 insertions(+), 20 deletions(-) create mode 100644 src/Parallel/Contract/SerializableInterface.php create mode 100644 src/Parallel/CpuCoreCountProvider.php create mode 100644 src/Parallel/Enum/Action.php create mode 100644 src/Parallel/Enum/Content.php create mode 100644 src/Parallel/Enum/ReactCommand.php create mode 100644 src/Parallel/Enum/ReactEvent.php create mode 100644 src/Parallel/Exception/ParallelShouldNotHappenException.php create mode 100644 src/Parallel/Reflection/CommandFromReflectionFactory.php create mode 100644 src/Parallel/ScheduleFactory.php create mode 100644 src/Parallel/ValueObject/ParallelProcess.php create mode 100644 src/Parallel/ValueObject/ProcessPool.php create mode 100644 src/Parallel/ValueObject/Schedule.php diff --git a/composer.json b/composer.json index 2b75955037e..aa60d6f9044 100644 --- a/composer.json +++ b/composer.json @@ -19,11 +19,13 @@ "composer/xdebug-handler": "^3.0.5", "doctrine/inflector": "^2.1", "entropy/entropy": "^0.4.12", + "fidry/cpu-core-counter": "^1.1", "nette/utils": "^4.1.4", "nikic/php-parser": "^5.8", "ondram/ci-detector": "^4.2", "phpstan/phpdoc-parser": "^2.3.3", "phpstan/phpstan": "^2.2.6", + "react/child-process": "^0.6.5", "react/event-loop": "^1.6", "react/promise": "^3.3", "react/socket": "^1.17", @@ -37,7 +39,6 @@ "symfony/filesystem": "^8.1", "symfony/finder": "^8.1", "symfony/process": "^8.1", - "symplify/easy-parallel": "^11.2.2", "symplify/rule-doc-generator-contracts": "^11.2", "webmozart/assert": "^2.4" }, diff --git a/src/Application/ApplicationFileProcessor.php b/src/Application/ApplicationFileProcessor.php index 09a9b1e7b73..eeb05d5a297 100644 --- a/src/Application/ApplicationFileProcessor.php +++ b/src/Application/ApplicationFileProcessor.php @@ -12,6 +12,9 @@ use Rector\Configuration\Parameter\SimpleParameterProvider; use Rector\FileSystem\FilesFinder; use Rector\Parallel\Application\ParallelFileProcessor; +use Rector\Parallel\CpuCoreCountProvider; +use Rector\Parallel\Exception\ParallelShouldNotHappenException; +use Rector\Parallel\ScheduleFactory; use Rector\PhpParser\Parser\ParserErrors; use Rector\Reporting\MissConfigurationReporter; use Rector\Skipper\Skipper\UsedSkipCollector; @@ -25,9 +28,6 @@ use Rector\ValueObject\Reporting\FileDiff; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Style\SymfonyStyle; -use Symplify\EasyParallel\CpuCoreCountProvider; -use Symplify\EasyParallel\Exception\ParallelShouldNotHappenException; -use Symplify\EasyParallel\ScheduleFactory; use Throwable; final class ApplicationFileProcessor diff --git a/src/ChangesReporting/ValueObject/RectorWithLineChange.php b/src/ChangesReporting/ValueObject/RectorWithLineChange.php index 6db365bd4ef..a69910a16c6 100644 --- a/src/ChangesReporting/ValueObject/RectorWithLineChange.php +++ b/src/ChangesReporting/ValueObject/RectorWithLineChange.php @@ -5,8 +5,8 @@ namespace Rector\ChangesReporting\ValueObject; use Rector\Contract\Rector\RectorInterface; +use Rector\Parallel\Contract\SerializableInterface; use Rector\PostRector\Contract\Rector\PostRectorInterface; -use Symplify\EasyParallel\Contract\SerializableInterface; use Webmozart\Assert\Assert; final readonly class RectorWithLineChange implements SerializableInterface diff --git a/src/Console/Command/WorkerCommand.php b/src/Console/Command/WorkerCommand.php index ef2a8b251be..2fe40cc6cc4 100644 --- a/src/Console/Command/WorkerCommand.php +++ b/src/Console/Command/WorkerCommand.php @@ -15,6 +15,9 @@ use Rector\Configuration\ConfigurationRuleFilter; use Rector\Configuration\Option; use Rector\Console\ProcessConfigureDecorator; +use Rector\Parallel\Enum\Action; +use Rector\Parallel\Enum\ReactCommand; +use Rector\Parallel\Enum\ReactEvent; use Rector\Parallel\ValueObject\Bridge; use Rector\StaticReflection\DynamicSourceLocatorDecorator; use Rector\Util\MemoryLimiter; @@ -23,9 +26,6 @@ use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; -use Symplify\EasyParallel\Enum\Action; -use Symplify\EasyParallel\Enum\ReactCommand; -use Symplify\EasyParallel\Enum\ReactEvent; use Throwable; use Webmozart\Assert\Assert; @@ -154,7 +154,7 @@ private function runWorker( ); /** - * this invokes all listeners listening $decoder->on(...) @see \Symplify\EasyParallel\Enum\ReactEvent::DATA + * this invokes all listeners listening $decoder->on(...) @see \Rector\Parallel\Enum\ReactEvent::DATA */ $encoder->write([ ReactCommand::ACTION => Action::RESULT, diff --git a/src/Parallel/Application/ParallelFileProcessor.php b/src/Parallel/Application/ParallelFileProcessor.php index 16d3e2e1ba0..e0331571938 100644 --- a/src/Parallel/Application/ParallelFileProcessor.php +++ b/src/Parallel/Application/ParallelFileProcessor.php @@ -14,19 +14,19 @@ use Rector\Configuration\Parameter\SimpleParameterProvider; use Rector\Console\Command\ProcessCommand; use Rector\Parallel\Command\WorkerCommandLineFactory; +use Rector\Parallel\Enum\Action; +use Rector\Parallel\Enum\Content; +use Rector\Parallel\Enum\ReactCommand; +use Rector\Parallel\Enum\ReactEvent; use Rector\Parallel\ValueObject\Bridge; +use Rector\Parallel\ValueObject\ParallelProcess; +use Rector\Parallel\ValueObject\ProcessPool; +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 Symplify\EasyParallel\Enum\Action; -use Symplify\EasyParallel\Enum\Content; -use Symplify\EasyParallel\Enum\ReactCommand; -use Symplify\EasyParallel\Enum\ReactEvent; -use Symplify\EasyParallel\ValueObject\ParallelProcess; -use Symplify\EasyParallel\ValueObject\ProcessPool; -use Symplify\EasyParallel\ValueObject\Schedule; use Throwable; /** diff --git a/src/Parallel/Command/WorkerCommandLineFactory.php b/src/Parallel/Command/WorkerCommandLineFactory.php index b8bbba0ad16..36f44109f2a 100644 --- a/src/Parallel/Command/WorkerCommandLineFactory.php +++ b/src/Parallel/Command/WorkerCommandLineFactory.php @@ -7,10 +7,10 @@ use Rector\ChangesReporting\Output\JsonOutputFormatter; use Rector\Configuration\Option; use Rector\FileSystem\FilePathHelper; +use Rector\Parallel\Exception\ParallelShouldNotHappenException; +use Rector\Parallel\Reflection\CommandFromReflectionFactory; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputInterface; -use Symplify\EasyParallel\Exception\ParallelShouldNotHappenException; -use Symplify\EasyParallel\Reflection\CommandFromReflectionFactory; /** * @see \Rector\Tests\Parallel\Command\WorkerCommandLineFactoryTest diff --git a/src/Parallel/Contract/SerializableInterface.php b/src/Parallel/Contract/SerializableInterface.php new file mode 100644 index 00000000000..250986e4c66 --- /dev/null +++ b/src/Parallel/Contract/SerializableInterface.php @@ -0,0 +1,15 @@ + $json + */ + public static function decode(array $json): self; +} diff --git a/src/Parallel/CpuCoreCountProvider.php b/src/Parallel/CpuCoreCountProvider.php new file mode 100644 index 00000000000..59c5356340d --- /dev/null +++ b/src/Parallel/CpuCoreCountProvider.php @@ -0,0 +1,22 @@ +getCount(); + } catch (NumberOfCpuCoreNotFound) { + return self::DEFAULT_CORE_COUNT; + } + } +} diff --git a/src/Parallel/Enum/Action.php b/src/Parallel/Enum/Action.php new file mode 100644 index 00000000000..7658e1e3484 --- /dev/null +++ b/src/Parallel/Enum/Action.php @@ -0,0 +1,14 @@ + $className + */ + public function create(string $className): Command + { + $commandReflectionClass = new ReflectionClass($className); + + $command = $commandReflectionClass->newInstanceWithoutConstructor(); + $parentClassReflection = $commandReflectionClass->getParentClass(); + + if (! $parentClassReflection instanceof ReflectionClass) { + throw new ParallelShouldNotHappenException(); + } + + $parentConstructorReflectionMethod = $parentClassReflection->getConstructor(); + if (! $parentConstructorReflectionMethod instanceof ReflectionMethod) { + throw new ParallelShouldNotHappenException(); + } + + $parentConstructorReflectionMethod->invoke($command); + + return $command; + } +} diff --git a/src/Parallel/ScheduleFactory.php b/src/Parallel/ScheduleFactory.php new file mode 100644 index 00000000000..d8c83d8cacb --- /dev/null +++ b/src/Parallel/ScheduleFactory.php @@ -0,0 +1,30 @@ + $files + */ + public function create(int $cpuCores, int $jobSize, int $maxNumberOfProcesses, array $files): Schedule + { + Assert::positiveInteger($jobSize); + + $jobs = array_chunk($files, $jobSize); + $numberOfProcesses = min(count($jobs), $cpuCores); + + $numberOfProcesses = min($maxNumberOfProcesses, $numberOfProcesses); + + return new Schedule($numberOfProcesses, $jobs); + } +} diff --git a/src/Parallel/ValueObject/ParallelProcess.php b/src/Parallel/ValueObject/ParallelProcess.php new file mode 100644 index 00000000000..914f03eeacc --- /dev/null +++ b/src/Parallel/ValueObject/ParallelProcess.php @@ -0,0 +1,155 @@ +stdErr = $tmp; + $this->process = new Process($this->command, null, null, [ + 2 => $this->stdErr, + // todo is it fine to not have 0 and 1 FD? + ]); + $this->process->start($this->loop); + + $this->onData = $onData; + $this->onError = $onError; + + $this->process->on(ReactEvent::EXIT, function ($exitCode) use ($onExit): void { + $stdErr = $this->stdErr; + if ($stdErr === null) { + throw new ParallelShouldNotHappenException(); + } + + $this->cancelTimer(); + + rewind($stdErr); + + /** @var string $streamContents */ + $streamContents = stream_get_contents($stdErr); + $onExit($exitCode, $streamContents); + + fclose($stdErr); + }); + } + + /** + * @param mixed[] $data + */ + public function request(array $data): void + { + $this->cancelTimer(); + $this->encoder->write($data); + $this->timer = $this->loop->addTimer($this->timetoutInSeconds, function (): void { + $onError = $this->onError; + + $errorMessage = sprintf('Child process timed out after %d seconds', $this->timetoutInSeconds); + $onError(new Exception($errorMessage)); + }); + } + + public function quit(): void + { + $this->cancelTimer(); + if (! $this->process->isRunning()) { + return; + } + + foreach ($this->process->pipes as $pipe) { + $pipe->close(); + } + + $this->encoder->end(); + } + + public function bindConnection(Decoder $decoder, Encoder $encoder): void + { + $decoder->on(ReactEvent::DATA, function (array $json): void { + $this->cancelTimer(); + if ($json[ReactCommand::ACTION] !== Action::RESULT) { + return; + } + + $onData = $this->onData; + $onData($json[Content::RESULT]); + }); + $this->encoder = $encoder; + + $decoder->on(ReactEvent::ERROR, function (Throwable $throwable): void { + $onError = $this->onError; + $onError($throwable); + }); + + $encoder->on(ReactEvent::ERROR, function (Throwable $throwable): void { + $onError = $this->onError; + $onError($throwable); + }); + } + + private function cancelTimer(): void + { + if (! $this->timer instanceof TimerInterface) { + return; + } + + $this->loop->cancelTimer($this->timer); + $this->timer = null; + } +} diff --git a/src/Parallel/ValueObject/ProcessPool.php b/src/Parallel/ValueObject/ProcessPool.php new file mode 100644 index 00000000000..6341cb83964 --- /dev/null +++ b/src/Parallel/ValueObject/ProcessPool.php @@ -0,0 +1,67 @@ + + */ + private array $processes = []; + + public function __construct( + private readonly TcpServer $tcpServer + ) { + } + + public function getProcess(string $identifier): ParallelProcess + { + if (! \array_key_exists($identifier, $this->processes)) { + throw new ParallelShouldNotHappenException(\sprintf('Process "%s" not found.', $identifier)); + } + + return $this->processes[$identifier]; + } + + public function attachProcess(string $identifier, ParallelProcess $parallelProcess): void + { + $this->processes[$identifier] = $parallelProcess; + } + + public function tryQuitProcess(string $identifier): void + { + if (! \array_key_exists($identifier, $this->processes)) { + return; + } + + $this->quitProcess($identifier); + } + + public function quitProcess(string $identifier): void + { + $parallelProcess = $this->getProcess($identifier); + $parallelProcess->quit(); + + unset($this->processes[$identifier]); + if ($this->processes !== []) { + return; + } + + $this->tcpServer->close(); + } + + public function quitAll(): void + { + foreach (\array_keys($this->processes) as $identifier) { + $this->quitProcess($identifier); + } + } +} diff --git a/src/Parallel/ValueObject/Schedule.php b/src/Parallel/ValueObject/Schedule.php new file mode 100644 index 00000000000..fe2ab1d8500 --- /dev/null +++ b/src/Parallel/ValueObject/Schedule.php @@ -0,0 +1,34 @@ +> $jobs + */ + public function __construct( + private readonly int $numberOfProcesses, + private readonly array $jobs + ) { + } + + public function getNumberOfProcesses(): int + { + return $this->numberOfProcesses; + } + + /** + * @return array> + */ + public function getJobs(): array + { + return $this->jobs; + } +} diff --git a/src/ValueObject/Error/SystemError.php b/src/ValueObject/Error/SystemError.php index 860d7328135..e39b4a29a11 100644 --- a/src/ValueObject/Error/SystemError.php +++ b/src/ValueObject/Error/SystemError.php @@ -5,8 +5,8 @@ namespace Rector\ValueObject\Error; use Nette\Utils\Strings; +use Rector\Parallel\Contract\SerializableInterface; use Rector\Parallel\ValueObject\BridgeItem; -use Symplify\EasyParallel\Contract\SerializableInterface; /** * @see \Rector\Tests\ValueObject\Error\SystemErrorTest diff --git a/src/ValueObject/Reporting/FileDiff.php b/src/ValueObject/Reporting/FileDiff.php index cac1ec6978b..f5c1cb5c283 100644 --- a/src/ValueObject/Reporting/FileDiff.php +++ b/src/ValueObject/Reporting/FileDiff.php @@ -7,9 +7,9 @@ use Nette\Utils\Strings; use Rector\ChangesReporting\ValueObject\RectorWithLineChange; use Rector\Contract\Rector\RectorInterface; +use Rector\Parallel\Contract\SerializableInterface; use Rector\Parallel\ValueObject\BridgeItem; use Rector\Util\RectorClassesSorter; -use Symplify\EasyParallel\Contract\SerializableInterface; use Webmozart\Assert\Assert; /** From f34492f3cd37b4e316e95e1fd6ad1f0c594d9696 Mon Sep 17 00:00:00 2001 From: GitHub Action Date: Sat, 29 Aug 2026 16:46:30 +0000 Subject: [PATCH 2/2] [ci-review] Rector Rectify --- src/Parallel/CpuCoreCountProvider.php | 2 +- src/Parallel/ValueObject/Schedule.php | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Parallel/CpuCoreCountProvider.php b/src/Parallel/CpuCoreCountProvider.php index 59c5356340d..49923cfb46f 100644 --- a/src/Parallel/CpuCoreCountProvider.php +++ b/src/Parallel/CpuCoreCountProvider.php @@ -14,7 +14,7 @@ final class CpuCoreCountProvider public function provide(): int { try { - return (new CpuCoreCounter())->getCount(); + return new CpuCoreCounter()->getCount(); } catch (NumberOfCpuCoreNotFound) { return self::DEFAULT_CORE_COUNT; } diff --git a/src/Parallel/ValueObject/Schedule.php b/src/Parallel/ValueObject/Schedule.php index fe2ab1d8500..c66797e9e59 100644 --- a/src/Parallel/ValueObject/Schedule.php +++ b/src/Parallel/ValueObject/Schedule.php @@ -8,14 +8,14 @@ * From * https://github.com/phpstan/phpstan-src/commit/9124c66dcc55a222e21b1717ba5f60771f7dda92#diff-bc84213b079ef3456caece03c00ba34c07886dcae12180cd1192fbb223d65b15 */ -final class Schedule +final readonly class Schedule { /** * @param array> $jobs */ public function __construct( - private readonly int $numberOfProcesses, - private readonly array $jobs + private int $numberOfProcesses, + private array $jobs ) { }