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

Passed

Push — master ( b3cd52...71bd8a )

by Anton

created 2020-04-29 18:52 UTC

ParallelExecutor::gatherOutput() A

↳ Parent: ParallelExecutor

Complexity

Conditions	4
Paths	5

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes

Metric	Value
cc	4
nc	5
nop	2
dl	0
loc	14
ccs	0
cts	13
cp	0
crap	20
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\Collection\PersistentCollection;
use Deployer\Configuration\Configuration;
use Deployer\Console\Application;
use Deployer\Deployer;
use Deployer\Exception\Exception;
use Deployer\Exception\GracefulShutdownException;
use Deployer\Host\Host;
use Deployer\Host\Localhost;
use Deployer\Host\Storage;
use Deployer\Component\Ssh\Client;
use Deployer\Selector\Selector;
use Deployer\Task\Context;
use Deployer\Task\Task;
use Symfony\Component\Console\Helper\Table;
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 $console;
    private $client;
    private $config;

    public function __construct(
        InputInterface $input,
        OutputInterface $output,
        Messenger $messenger,
        Application $console,
        Client $client,
        Configuration $config
    )
    {
        $this->input = $input;
        $this->output = $output;
        $this->messenger = $messenger;
        $this->console = $console;
        $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
     */
    public function run(array $tasks, array $hosts, $plan = null): int
    {
        $plan || $this->connect($hosts);

        $localhost = new Localhost();
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}
        $globalLimit = (int)$this->input->getOption('limit') ?: count($hosts);

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

            $limit = min($globalLimit, $task->getLimit() ?? $globalLimit);

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

deployphp / deployer

GitHub Access Token became invalid

Push — master ( b3cd52...71bd8a )

ParallelExecutor::gatherOutput() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like