PERF: Deploy to all servers in parallel

src/Task/PendingTask.php

@@ -0,0 +1,82 @@
+<?php
+
+/*
+ * This file is part of the EasyDeploy project.
+ *
+ * (c) Javier Eguiluz <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace EasyCorp\Bundle\EasyDeployBundle\Task;
+
+use EasyCorp\Bundle\EasyDeployBundle\Helper\Str;
+use EasyCorp\Bundle\EasyDeployBundle\Logger;
+use EasyCorp\Bundle\EasyDeployBundle\Server\Server;
+use Symfony\Component\Process\Process;
+use Symfony\Component\Process\Exception\ProcessFailedException;
+
+class PendingTask {
+    private $server;
+    private $dryRun;
+    private $process;
+    private $logger;
+
+    public function __construct (Server $server, string $shellCommand, Logger $logger, bool $dryRun = false)
+    {
+        $this->server = $server;
+        $this->dryRun = $dryRun;
+        $this->logger = $logger;
+
+        if ($this->dryRun) {
+            return;
+        }
+
+        if ($server->isLocalHost()) {
+            $this->process = $this->createProcess($shellCommand);
+        } else {
+            $this->process = $this->createProcess(sprintf('%s %s', $server->getSshConnectionString(), escapeshellarg($shellCommand)));
+        }
+
+        $this->process->setTimeout(null);
+    }
+
+    private function createProcess(string $shellCommand): Process
+    {
+        if (method_exists(Process::class, 'fromShellCommandline')) {
+            return Process::fromShellCommandline($shellCommand);
+        }
+
+        return new Process($shellCommand);
+    }
+
+    public function start ()
+    {
+        if ($this->dryRun) {
+            return;
+        }
+        $this->process->start(function ($type, $buffer) {
+            if (Process::ERR === $type) {
+                $this->logger->log(Str::prefix(rtrim($buffer, PHP_EOL), sprintf('| [<server>%s</>] <stream>err ::</> ', $this->server)));
+            } else {
+                $this->logger->log(Str::prefix(rtrim($buffer, PHP_EOL), sprintf('| [<server>%s</>] <stream>out ::</> ', $this->server)));
+            }
+        });
+    }
+
+    public function getCompletionResult ()
+    {
+        if ($this->dryRun) {
+            return new TaskCompleted($this->server, '', 0);
+        }
+
+        // Make sure we ran without errors
+        // As in https://github.com/symfony/symfony/blob/4.4/src/Symfony/Component/Process/Process.php#L266
+        if (0 !== $this->process->wait()) {
+            throw new ProcessFailedException($this->process);
+        }
+
+        return new TaskCompleted($this->server, $this->process->getOutput(), $this->process->getExitCode());
+    }
+}

src/Task/TaskRunner.php

@@ -33,24 +33,22 @@ public function __construct(bool $isDryRun, Logger $logger)
      */
     public function run(Task $task): array
     {
-        $results = [];
+        // Start all processes asynchronously
+        $processes = [];
         foreach ($task->getServers() as $server) {
-            $results[] = $this->doRun($server, $server->resolveProperties($task->getShellCommand()), $task->getEnvVars());
+            $processes[] = $this->startProcess($server, $server->resolveProperties($task->getShellCommand()), $task->getEnvVars());
         }
 
-        return $results;
-    }
-
-    private function createProcess(string $shellCommand): Process
-    {
-        if (method_exists(Process::class, 'fromShellCommandline')) {
-            return Process::fromShellCommandline($shellCommand);
+        // Collect all their results
+        $results = [];
+        foreach ($processes as $process) {
+            $results[] = $process->getCompletionResult();
         }
 
-        return new Process($shellCommand);
+        return $results;
     }
 
-    private function doRun(Server $server, string $shellCommand, array $envVars): TaskCompleted
+    private function startProcess(Server $server, string $shellCommand, array $envVars): PendingTask
     {
         if ($server->has(Property::project_dir)) {
             $shellCommand = sprintf('cd %s && %s', $server->get(Property::project_dir), $shellCommand);
@@ -67,26 +65,9 @@ private function doRun(Server $server, string $shellCommand, array $envVars): Ta
 
         $this->logger->log(sprintf('[<server>%s</>] Executing command: <command>%s</>', $server, $shellCommand));
 
-        if ($this->isDryRun) {
-            return new TaskCompleted($server, '', 0);
-        }
-
-        if ($server->isLocalHost()) {
-            $process = $this->createProcess($shellCommand);
-        } else {
-            $process = $this->createProcess(sprintf('%s %s', $server->getSshConnectionString(), escapeshellarg($shellCommand)));
-        }
-
-        $process->setTimeout(null);
-
-        $process = $process->mustRun(function ($type, $buffer) {
-            if (Process::ERR === $type) {
-                $this->logger->log(Str::prefix(rtrim($buffer, PHP_EOL), '| <stream>err ::</> '));
-            } else {
-                $this->logger->log(Str::prefix(rtrim($buffer, PHP_EOL), '| <stream>out ::</> '));
-            }
-        });
+        $pendingTask = new PendingTask($server, $shellCommand, $this->logger, $this->isDryRun);
+        $pendingTask->start();
 
-        return new TaskCompleted($server, $process->getOutput(), $process->getExitCode());
+        return $pendingTask;
     }
 }