ParallelExecutor::gatherOutput() - Code Metrics - Inspection of "Next version" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 6c9176...e80da1 )

by Anton

created 2020-05-01 13:04 UTC

ParallelExecutor::gatherOutput() A

↳ Parent: ParallelExecutor

Complexity

Conditions	4
Paths	5

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	4.0218

Importance

Changes

Metric	Value
cc	4
nc	5
nop	2
dl	0
loc	14
ccs	8
cts	9
cp	0.8889
crap	4.0218
rs	9.7998
c	0
b	0
f	0

<?php declare(strict_types=1);
/* (c) Anton Medvedev <[email protected]>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Deployer\Executor;

use Deployer\Component\Ssh\Client;
use Deployer\Configuration\Configuration;
use Deployer\Exception\Exception;
use Deployer\Exception\GracefulShutdownException;
use Deployer\Host\Host;
use Deployer\Host\Localhost;
use Deployer\Selector\Selector;
use Deployer\Task\Context;
use Deployer\Task\Task;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\Process;

const FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];

function spinner($message = '')
{
    $frame = FRAMES[(int)(microtime(true) * 10) % count(FRAMES)];
    return "  $frame $message\r";
}

class ParallelExecutor
{
    private $input;
    private $output;
    private $messenger;
    private $client;
    private $config;

    public function __construct(
        InputInterface $input,
        OutputInterface $output,
        Messenger $messenger,
        Client $client,
        Configuration $config
    )
    {
        $this->input = $input;
        $this->output = $output;
        $this->messenger = $messenger;
        $this->client = $client;
        $this->config = $config;
    }

    /**
     * @param Host[] $hosts
     */
    private function connect(array $hosts)
    {
        $callback = function (string $output) {
            $output = preg_replace('/\n$/', '', $output);
            if (strlen($output) !== 0) {
                $this->output->writeln($output);
            }
        };

        // Connect to each host sequentially, to prevent getting locked.
        foreach ($hosts as $host) {
            if ($host instanceof Localhost) {
                continue;
            }
            $process = $this->getProcess($host, new Task('connect'));
            $process->start();

            while ($process->isRunning()) {
                $this->gatherOutput([$process], $callback);
                $this->output->write(spinner(str_pad("connect {$host->tag()}", intval(getenv('COLUMNS')) - 1)));
                usleep(1000);
            }
        }

        // Clear spinner.
        $this->output->write(str_repeat(' ', intval(getenv('COLUMNS')) - 1) . "\r");
    }

    /**
     * @param Task[] $tasks
     * @param Host[] $hosts
     * @param Planner|null $plan
     * @return int
     */
    public function run(array $tasks, array $hosts, $plan = null): int
    {
        $plan || $this->connect($hosts);

        $globalLimit = (int)$this->input->getOption('limit') ?: count($hosts);

        foreach ($tasks as $task) {
            $plan || $this->messenger->startTask($task);

            $limit = min($globalLimit, $task->getLimit() ?? $globalLimit);
            $plannedHosts = $hosts;
            if ($task->isOnce()) {
                $plannedHosts = [];
                foreach ($hosts as $host) {
                    if (Selector::apply($task->getSelector(), $host)) {
                        $plannedHosts[] = $host;
                        break;
                    }
                }
            }

            if ($limit === 1 || count($plannedHosts) === 1) {
                foreach ($plannedHosts as $host) {
                    if (!Selector::apply($task->getSelector(), $host)) {
                        if ($plan) {
                            $plan->commit([], $task);
                        }
                        continue;
                    }

                    if ($plan) {
                        $plan->commit([$host], $task);
                        continue;
                    }

                    try {
                        $host->getConfig()->load();
                        Exception::setTaskSourceLocation($task->getSourceLocation());

                        $task->run(new Context($host, $this->input, $this->output));

                        $host->getConfig()->save();
                    } catch (GracefulShutdownException $exception) {
                        $this->messenger->renderException($exception, $host);
                        return GracefulShutdownException::EXIT_CODE;
                    } catch (\Throwable $exception) {
                        $this->messenger->renderException($exception, $host);
                        return 1;
                    }
                }
            } else {
                foreach (array_chunk($hosts, $limit) as $chunk) {
                    $exitCode = $this->runTask($chunk, $task, $plan);
                    if ($exitCode !== 0) {
                        return $exitCode;
                    }
                }
            }

            $this->messenger->endTask($task);
        }

        return 0;
    }

    private function runTask(array $hosts, Task $task, Planner $plan = null): int
    {
        $processes = [];
        $selectedHosts = [];
        foreach ($hosts as $host) {
            $selector = $task->getSelector();
            if ($selector === null || Selector::apply($selector, $host)) {
                $selectedHosts[] = $host;
                $plan || $processes[] = $this->getProcess($host, $task);
            }
        }

        if ($plan) {
            $plan->commit($selectedHosts, $task);
            return 0;
        }

        $callback = function (string $output) {
            $output = preg_replace('/\n$/', '', $output);
            if (strlen($output) !== 0) {
                $this->output->writeln($output);
            }
        };

        $this->startProcesses($processes);

        while ($this->areRunning($processes)) {
            $this->gatherOutput($processes, $callback);
            $this->output->write(spinner());
            usleep(1000);
        }

        // Clear spinner.
        $this->output->write("    \r");

        $this->gatherOutput($processes, $callback);

        return $this->cumulativeExitCode($processes);
    }

    protected function getProcess(Host $host, Task $task): Process
    {
        $dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
        $configDirectory = $host->get('config_directory');
        $decorated = $this->output->isDecorated() ? '--decorated' : '';
        $command = "$dep worker $task {$host->alias()} $configDirectory {$this->input} $decorated";

        if ($this->output->isDebug()) {
            $this->output->writeln("[{$host->tag()}] $command");
        }

        return Process::fromShellCommandline($command);
    }

    /**
     * @param Process[] $processes
     */
    protected function startProcesses(array $processes)
    {
        foreach ($processes as $process) {
            $process->start();
        }
    }

    /**
     * @param Process[] $processes
     */
    protected function areRunning(array $processes): bool
    {
        foreach ($processes as $process) {
            if ($process->isRunning()) {
                return true;
            }
        }

        return false;
    }

    /**
     * @param Process[] $processes
     */
    protected function gatherOutput(array $processes, callable $callback)
    {
        foreach ($processes as $process) {
            $output = $process->getIncrementalOutput();
            if (strlen($output) !== 0) {
                $callback($output);
            }

            $errorOutput = $process->getIncrementalErrorOutput();
            if (strlen($errorOutput) !== 0) {
                $callback($errorOutput);
            }
        }
    }

    /**
     * Gather the cumulative exit code for the processes.
     */
    protected function cumulativeExitCode(array $processes): int
    {
        foreach ($processes as $process) {
            if ($process->getExitCode() > 0) {
                return $process->getExitCode();
            }
        }

        return 0;
    }
}


1			<?php declare(strict_types=1);
2			/* (c) Anton Medvedev <[email protected]>
3			*
4			* For the full copyright and license information, please view the LICENSE
5			* file that was distributed with this source code.
6			*/
7
8			namespace Deployer\Executor;
9
10			use Deployer\Component\Ssh\Client;
11			use Deployer\Configuration\Configuration;
12			use Deployer\Exception\Exception;
13			use Deployer\Exception\GracefulShutdownException;
14			use Deployer\Host\Host;
15			use Deployer\Host\Localhost;
16			use Deployer\Selector\Selector;
17			use Deployer\Task\Context;
18			use Deployer\Task\Task;
19			use Symfony\Component\Console\Input\InputInterface;
20			use Symfony\Component\Console\Output\OutputInterface;
21			use Symfony\Component\Process\Process;
22
23	1		const FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
24
25			function spinner($message = '')
26			{
27	1		$frame = FRAMES[(int)(microtime(true) * 10) % count(FRAMES)];
28	1		return " $frame $message\r";
29			}
30
31			class ParallelExecutor
32			{
33			private $input;
34			private $output;
35			private $messenger;
36			private $client;
37			private $config;
38
39	8		public function __construct(
40			InputInterface $input,
41			OutputInterface $output,
42			Messenger $messenger,
43			Client $client,
44			Configuration $config
45			)
46			{
47	8		$this->input = $input;
48	8		$this->output = $output;
49	8		$this->messenger = $messenger;
50	8		$this->client = $client;
51	8		$this->config = $config;
52	8		}
53
54			/**
55			* @param Host[] $hosts
56			*/
57	8		private function connect(array $hosts)
58			{
59		View Code Duplication	$callback = function (string $output) {
60			$output = preg_replace('/\n$/', '', $output);
61			if (strlen($output) !== 0) {
62			$this->output->writeln($output);
63			}
64	8		};
65
66			// Connect to each host sequentially, to prevent getting locked.
67	8		foreach ($hosts as $host) {
68	8		if ($host instanceof Localhost) {
69	8		continue;
70			}
71			$process = $this->getProcess($host, new Task('connect'));
72			$process->start();
73
74			while ($process->isRunning()) {
75			$this->gatherOutput([$process], $callback);
76			$this->output->write(spinner(str_pad("connect {$host->tag()}", intval(getenv('COLUMNS')) - 1)));
77			usleep(1000);
78			}
79			}
80
81			// Clear spinner.
82	8		$this->output->write(str_repeat(' ', intval(getenv('COLUMNS')) - 1) . "\r");
83	8		}
84
85			/**
86			* @param Task[] $tasks
87			* @param Host[] $hosts
88			* @param Planner\|null $plan
89			* @return int
90			*/
91	8		public function run(array $tasks, array $hosts, $plan = null): int
92			{
93	8		$plan \|\| $this->connect($hosts);
94
95	8		$globalLimit = (int)$this->input->getOption('limit') ?: count($hosts);
96
97	8		foreach ($tasks as $task) {
98	8		$plan \|\| $this->messenger->startTask($task);
99
100	8		$limit = min($globalLimit, $task->getLimit() ?? $globalLimit);
101	8		$plannedHosts = $hosts;
102	8		if ($task->isOnce()) {
103			$plannedHosts = [];
104			foreach ($hosts as $host) {
105			if (Selector::apply($task->getSelector(), $host)) {
106			$plannedHosts[] = $host;
107			break;
108			}
109			}
110			}
111
112	8		if ($limit === 1 \|\| count($plannedHosts) === 1) {
113	7		foreach ($plannedHosts as $host) {
114	7		if (!Selector::apply($task->getSelector(), $host)) {
115			if ($plan) {
116			$plan->commit([], $task);
117			}
118			continue;
119			}
120
121	7		if ($plan) {
122			$plan->commit([$host], $task);
123			continue;
124			}
125
126			try {
127	7		$host->getConfig()->load();
128	7		Exception::setTaskSourceLocation($task->getSourceLocation());
129
130	7		$task->run(new Context($host, $this->input, $this->output));
131
132	7		$host->getConfig()->save();
133	1		} catch (GracefulShutdownException $exception) {
134			$this->messenger->renderException($exception, $host);
135			return GracefulShutdownException::EXIT_CODE;
136	1		} catch (\Throwable $exception) {
137	1		$this->messenger->renderException($exception, $host);
138	1		return 1;
139			}
140			}
141			} else {
142	1		foreach (array_chunk($hosts, $limit) as $chunk) {
143	1		$exitCode = $this->runTask($chunk, $task, $plan);
144	1		if ($exitCode !== 0) {
145			return $exitCode;
146			}
147			}
148			}
149
150	8		$this->messenger->endTask($task);
151			}
152
153	8		return 0;
154			}
155
156	1		private function runTask(array $hosts, Task $task, Planner $plan = null): int
157			{
158	1		$processes = [];
159	1		$selectedHosts = [];
160	1		foreach ($hosts as $host) {
161	1		$selector = $task->getSelector();
162	1		if ($selector === null \|\| Selector::apply($selector, $host)) {
163	1		$selectedHosts[] = $host;
164	1		$plan \|\| $processes[] = $this->getProcess($host, $task);
165			}
166			}
167
168	1		if ($plan) {
169			$plan->commit($selectedHosts, $task);
170			return 0;
171			}
172
173		View Code Duplication	$callback = function (string $output) {
174	1		$output = preg_replace('/\n$/', '', $output);
175	1		if (strlen($output) !== 0) {
176	1		$this->output->writeln($output);
177			}
178	1		};
179
180	1		$this->startProcesses($processes);
181
182	1		while ($this->areRunning($processes)) {
183	1		$this->gatherOutput($processes, $callback);
184	1		$this->output->write(spinner());
185	1		usleep(1000);
186			}
187
188			// Clear spinner.
189	1		$this->output->write(" \r");
190
191	1		$this->gatherOutput($processes, $callback);
192
193	1		return $this->cumulativeExitCode($processes);
194			}
195
196	1		protected function getProcess(Host $host, Task $task): Process
197			{
198	1		$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
199	1		$configDirectory = $host->get('config_directory');
200	1		$decorated = $this->output->isDecorated() ? '--decorated' : '';
201	1		$command = "$dep worker $task {$host->alias()} $configDirectory {$this->input} $decorated";
202
203	1		if ($this->output->isDebug()) {
204			$this->output->writeln("[{$host->tag()}] $command");
205			}
206
207	1		return Process::fromShellCommandline($command);
208			}
209
210			/**
211			* @param Process[] $processes
212			*/
213	1		protected function startProcesses(array $processes)
214			{
215	1		foreach ($processes as $process) {
216	1		$process->start();
217			}
218	1		}
219
220			/**
221			* @param Process[] $processes
222			*/
223	1		protected function areRunning(array $processes): bool
224			{
225	1		foreach ($processes as $process) {
226	1		if ($process->isRunning()) {
227	1		return true;
228			}
229			}
230
231	1		return false;
232			}
233
234			/**
235			* @param Process[] $processes
236			*/
237	1		protected function gatherOutput(array $processes, callable $callback)
238			{
239	1		foreach ($processes as $process) {
240	1		$output = $process->getIncrementalOutput();
241	1		if (strlen($output) !== 0) {
242	1		$callback($output);
243			}
244
245	1		$errorOutput = $process->getIncrementalErrorOutput();
246	1		if (strlen($errorOutput) !== 0) {
247			$callback($errorOutput);
248			}
249			}
250	1		}
251
252			/**
253			* Gather the cumulative exit code for the processes.
254			*/
255	1		protected function cumulativeExitCode(array $processes): int
256			{
257	1		foreach ($processes as $process) {
258	1		if ($process->getExitCode() > 0) {
259			return $process->getExitCode();
260			}
261			}
262
263	1		return 0;
264			}
265			}
266

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 6c9176...e80da1 )

ParallelExecutor::gatherOutput() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like