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

Passed

Push — master ( c538d5...8bac07 )

by Anton

created 2020-04-28 17:03 UTC

ParallelExecutor::persistHosts() A

↳ Parent: Project

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
nc	2
nop	1
dl	0
loc	16
ccs	0
cts	13
cp	0
crap	6
rs	9.7333
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\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 $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): int
    {
        $this->connect($hosts);

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

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

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

            if ($limit === 1 || count($hosts) === 1) {
                foreach ($hosts as $host) {
                    try {
                        Exception::setFilepath($task->getFilepath());
                        $host->getConfig()->load();

                        $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);
                    if ($exitCode !== 0) {
                        return $exitCode;
                    }
                }
            }

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

        return 0;
    }

    private function runTask(array $hosts, Task $task): int
    {
        $processes = [];
        foreach ($hosts as $host) {
            if ($task->shouldBePerformed($host)) {
                $processes[] = $this->getProcess($host, $task);
                if ($task->isOnce()) {
                    $task->setHasRun();
                }
            }
        }

        $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->gatherExitCodes($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);
    }

    /**
     * Start all of the processes.
     *
     * @param Process[] $processes
     * @return void
     */
    protected function startProcesses(array $processes)
    {
        foreach ($processes as $process) {
            $process->start();
        }
    }

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

        return false;
    }

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

deployphp / deployer

GitHub Access Token became invalid

Push — master ( c538d5...8bac07 )

ParallelExecutor::persistHosts() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like