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

Passed

Pull Request — master (#1904)

by Anton

created 2020-04-25 19:57 UTC

ParallelExecutor::generateOptions() A

↳ Parent: Project

Complexity

Conditions	3
Paths	4

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
nc	4
nop	0
dl	0
loc	22
ccs	0
cts	15
cp	0
crap	12
rs	9.568
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 $informer;
    private $console;
    private $client;
    private $config;

    public function __construct(
        InputInterface $input,
        OutputInterface $output,
        Status $informer,
        Application $console,
        Client $client,
        Configuration $config
    )
    {
        $this->input = $input;
        $this->output = $output;
        $this->informer = $informer;
        $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->persistHosts($hosts);
        $this->connect($hosts);

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

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

            if ($task->isLocal()) {
                Storage::load(...$hosts);
                {
                    $task->run(new Context($localhost, $this->input, $this->output));
                }
                Storage::flush(...$hosts);
            } else {
                foreach (array_chunk($hosts, $limit) as $chunk) {
                    $exitCode = $this->runTask($chunk, $task);
                    if ($exitCode !== 0) {
                        return $exitCode;
                    }
                }
            }

            $this->informer->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);
        }
        $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;
        $configFile = $host->get('worker-config');
        $decorated = $this->output->isDecorated() ? '--decorated' : '';
        $command = "$dep worker $task {$host->alias()} $configFile {$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;
    }

    /**
     * @param Host[] $hosts
     */
    private function persistHosts(array $hosts)
    {
        foreach ($hosts as $host) {
            Context::push(new Context($host, $this->input, $this->output));

            $values = $host->getConfig()->persist();
            $workerConfig = sys_get_temp_dir() . '/' . uniqid('deployer-') . '-' . $host->alias() . '.dep';
            $values['worker-config'] = $workerConfig;

            $persistentCollection = new PersistentCollection($workerConfig, $values);
            $persistentCollection->flush();
            $host->getConfig()->setCollection($persistentCollection);

            Context::pop();
        }
    }
}


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 $informer;
39		private $console;
40		private $client;
41		private $config;
42
43		public function __construct(
44		InputInterface $input,
45		OutputInterface $output,
46		Status $informer,
47		Application $console,
48		Client $client,
49		Configuration $config
50		)
51		{
52		$this->input = $input;
53		$this->output = $output;
54		$this->informer = $informer;
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->persistHosts($hosts);
98		$this->connect($hosts);
99
100		$localhost = new Localhost();
101		$limit = (int)$this->input->getOption('limit') ?: count($hosts);
102
103		foreach ($tasks as $task) {
104		$this->informer->startTask($task);
105
106		if ($task->isLocal()) {
107		Storage::load(...$hosts);
108		{
109		$task->run(new Context($localhost, $this->input, $this->output));
110		}
111		Storage::flush(...$hosts);
112		} else {
113		foreach (array_chunk($hosts, $limit) as $chunk) {
114		$exitCode = $this->runTask($chunk, $task);
115		if ($exitCode !== 0) {
116		return $exitCode;
117		}
118		}
119		}
120
121		$this->informer->endTask($task);
122		}
123
124		return 0;
125		}
126
127		private function runTask(array $hosts, Task $task): int
128		{
129		$processes = [];
130		foreach ($hosts as $host) {
131		if ($task->shouldBePerformed($host)) {
132		$processes[] = $this->getProcess($host, $task);
133		if ($task->isOnce()) {
134		$task->setHasRun();
135		}
136		}
137		}
138
139	View Code Duplication	$callback = function (string $output) {
140		$output = preg_replace('/\n$/', '', $output);
141		if (strlen($output) !== 0) {
142		$this->output->writeln($output);
143		}
144		};
145
146		$this->startProcesses($processes);
147
148		while ($this->areRunning($processes)) {
149		$this->gatherOutput($processes, $callback);
150		$this->output->write(spinner());
151		usleep(1000);
152		}
153		$this->output->write("\r");
154		$this->gatherOutput($processes, $callback);
155
156		return $this->gatherExitCodes($processes);
157		}
158
159		protected function getProcess(Host $host, Task $task): Process
160		{
161		$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
162		$configFile = $host->get('worker-config');
163		$decorated = $this->output->isDecorated() ? '--decorated' : '';
164		$command = "$dep worker $task {$host->alias()} $configFile {$this->input} $decorated";
165
166		if ($this->output->isDebug()) {
167		$this->output->writeln("[{$host->tag()}] $command");
168		}
169
170		return Process::fromShellCommandline($command);
171		}
172
173		/**
174		* Start all of the processes.
175		*
176		* @param Process[] $processes
177		* @return void
178		*/
179		protected function startProcesses(array $processes)
180		{
181		foreach ($processes as $process) {
182		$process->start();
183		}
184		}
185
186		/**
187		* Determine if any of the processes are running.
188		*
189		* @param Process[] $processes
190		* @return bool
191		*/
192		protected function areRunning(array $processes): bool
193		{
194		foreach ($processes as $process) {
195		if ($process->isRunning()) {
196		return true;
197		}
198		}
199
200		return false;
201		}
202
203		/**
204		* Gather the output from all of the processes.
205		*
206		* @param Process[] $processes
207		* @param callable $callback
208		* @return void
209		*/
210		protected function gatherOutput(array $processes, callable $callback)
211		{
212		foreach ($processes as $process) {
213		$output = $process->getIncrementalOutput();
214		if (strlen($output) !== 0) {
215		$callback($output);
216		}
217
218		$errorOutput = $process->getIncrementalErrorOutput();
219		if (strlen($errorOutput) !== 0) {
220		$callback($errorOutput);
221		}
222		}
223		}
224
225		/**
226		* Gather the cumulative exit code for the processes.
227		*/
228		protected function gatherExitCodes(array $processes): int
229		{
230		foreach ($processes as $process) {
231		if ($process->getExitCode() > 0) {
232		return $process->getExitCode();
233		}
234		}
235
236		return 0;
237		}
238
239		/**
240		* @param Host[] $hosts
241		*/
242		private function persistHosts(array $hosts)
243		{
244		foreach ($hosts as $host) {
245		Context::push(new Context($host, $this->input, $this->output));
246
247		$values = $host->getConfig()->persist();
248		$workerConfig = sys_get_temp_dir() . '/' . uniqid('deployer-') . '-' . $host->alias() . '.dep';
249		$values['worker-config'] = $workerConfig;
250
251		$persistentCollection = new PersistentCollection($workerConfig, $values);
252		$persistentCollection->flush();
253		$host->getConfig()->setCollection($persistentCollection);
254
255		Context::pop();
256		}
257		}
258		}
259

deployphp / deployer

GitHub Access Token became invalid

Pull Request — master (#1904)

ParallelExecutor::generateOptions() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like