From 29d99f811aaa2b9f2f1b58e776b85fc85eb7d1a0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Martin=20Poirier=20Th=C3=A9or=C3=AAt?= <mpoiriert@gmail.com>
Date: Tue, 2 Apr 2024 19:24:54 -0400
Subject: [PATCH] [Messenger] Broker maximum process option

---
 .../Command/StartMessengerBrokerCommand.php   | 25 +++++++++-
 .../StartMessengerBrokerCommandTest.php       | 49 +++++++++++++++++--
 2 files changed, 68 insertions(+), 6 deletions(-)

diff --git a/packages/messenger/Broker/Command/StartMessengerBrokerCommand.php b/packages/messenger/Broker/Command/StartMessengerBrokerCommand.php
index 3efe6ad1..e2824942 100644
--- a/packages/messenger/Broker/Command/StartMessengerBrokerCommand.php
+++ b/packages/messenger/Broker/Command/StartMessengerBrokerCommand.php
@@ -68,6 +68,15 @@ protected function configure(): void
                 ),
                 1
             )
+            ->addOption(
+                'maximum-processes',
+                null,
+                InputOption::VALUE_REQUIRED,
+                sprintf(
+                    'Maximum number of processes (used only if "concurrent" is set to "%s")',
+                    self::OPTION_VALUE_CONCURRENT_AUTO
+                )
+            )
             ->addOption(
                 'timeout',
                 null,
@@ -140,6 +149,20 @@ private function calculateAutoConcurrent(InputInterface $input): int
             ));
         }
 
-        return max([$minProcesses, (int) round($processesPerCore * $this->cpuCounter->count())]);
+        $process = max([$minProcesses, (int) round($processesPerCore * $this->cpuCounter->count())]);
+
+        if ($maxProcesses = null === $input->getOption('maximum-processes')) {
+            return $process;
+        }
+
+        $maxProcesses = (int) $input->getOption('maximum-processes');
+        if ($maxProcesses <= 0) {
+            throw new InvalidOptionException(sprintf(
+                'Maximum processes value [%d] is invalid. Must be greater than 0',
+                $maxProcesses
+            ));
+        }
+
+        return min([$maxProcesses, $process]);
     }
 }
diff --git a/packages/messenger/Tests/Broker/Command/StartMessengerBrokerCommandTest.php b/packages/messenger/Tests/Broker/Command/StartMessengerBrokerCommandTest.php
index 993be4ef..4c4d6d00 100644
--- a/packages/messenger/Tests/Broker/Command/StartMessengerBrokerCommandTest.php
+++ b/packages/messenger/Tests/Broker/Command/StartMessengerBrokerCommandTest.php
@@ -80,6 +80,13 @@ public static function provideTestOption(): iterable
             1,
         ];
 
+        yield [
+            'maximum-processes',
+            null,
+            InputOption::VALUE_REQUIRED,
+            null,
+        ];
+
         yield [
             'timeout',
             null,
@@ -134,6 +141,20 @@ public function testExecuteInvalidMinimumProcessesWithAutoConcurrent(): void
         ]);
     }
 
+    public function testExecuteInvalidMaximumProcessesWithAutoConcurrent(): void
+    {
+        $maxProcesses = random_int(\PHP_INT_MIN, 0);
+        $this->expectExceptionObject(new InvalidOptionException(sprintf(
+            'Maximum processes value [%d] is invalid. Must be greater than 0',
+            $maxProcesses
+        )));
+
+        $this->execute([
+            '--concurrent' => 'auto',
+            '--maximum-processes' => $maxProcesses,
+        ]);
+    }
+
     public function testExecute(): void
     {
         $concurrent = random_int(1, 10);
@@ -205,6 +226,7 @@ public function testExecuteWithAutoConcurrent(
         int $numCpus,
         float $processesPerCore,
         int $minProcesses,
+        ?int $maxProcesses,
         int $concurrent
     ): void {
         $this
@@ -223,11 +245,17 @@ function (BrokerStartedEvent $event) use ($concurrent): void {
         );
 
         $this
-            ->execute([
-                '--concurrent' => 'auto',
-                '--processes-per-core' => $processesPerCore,
-                '--minimum-processes' => $minProcesses,
-            ])
+            ->execute(
+                array_filter(
+                    [
+                        '--concurrent' => 'auto',
+                        '--processes-per-core' => $processesPerCore,
+                        '--minimum-processes' => $minProcesses,
+                        '--maximum-processes' => $maxProcesses,
+                    ],
+                    static fn ($value) => null !== $value
+                )
+            )
             ->test(CommandDataTester::create(
                 0,
                 [
@@ -245,6 +273,7 @@ public static function provideDataForTestExecuteWithAutoConcurrent(): iterable
             '$numCpus' => 4,
             '$processesPerCore' => 1.0,
             '$minProcesses' => 2,
+            '$maxProcesses' => null,
             '$concurrent' => 4,
         ];
 
@@ -252,6 +281,7 @@ public static function provideDataForTestExecuteWithAutoConcurrent(): iterable
             '$numCpus' => 4,
             '$processesPerCore' => 0.8,
             '$minProcesses' => 1,
+            '$maxProcesses' => null,
             '$concurrent' => 3,
         ];
 
@@ -259,7 +289,16 @@ public static function provideDataForTestExecuteWithAutoConcurrent(): iterable
             '$numCpus' => 2,
             '$processesPerCore' => 0.8,
             '$minProcesses' => 5,
+            '$maxProcesses' => null,
             '$concurrent' => 5,
         ];
+
+        yield 'maximum processes' => [
+            '$numCpus' => 2,
+            '$processesPerCore' => 5,
+            '$minProcesses' => 1,
+            '$maxProcesses' => 1,
+            '$concurrent' => 1,
+        ];
     }
 }