ParallelExecutor::generateOptions() - Code Metrics - Inspection of "[Collection], [Configuration], [Exception], [Execu..." - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 075b5b...370756 )

by Anton

created 2018-10-08 05:08 UTC

ParallelExecutor::generateOptions() B

↳ Parent: ParallelExecutor

Complexity

Conditions	7
Paths	12

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	9.2312

Importance

Changes

Metric	Value
cc	7
nc	12
nop	0
dl	0
loc	28
ccs	9
cts	14
cp	0.6429
crap	9.2312
rs	8.5386
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\Console\Application;
use Deployer\Console\Output\Informer;
use Deployer\Console\Output\VerbosityString;
use Deployer\Exception\Exception;
use Deployer\Exception\GracefulShutdownException;
use Deployer\Host\Host;
use Deployer\Host\Localhost;
use Deployer\Host\Storage;
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;

class ParallelExecutor implements ExecutorInterface
{
    /**
     * @var InputInterface
     */
    private $input;

    /**
     * @var OutputInterface
     */
    private $output;

    /**
     * @var Informer
     */
    private $informer;

    /**
     * @var Application
     */
    private $console;

    public function __construct(
        InputInterface $input,
        OutputInterface $output,
        Informer $informer,
        Application $console
    ) {
        $this->input = $input;
        $this->output = $output;
        $this->informer = $informer;
        $this->console = $console;
    }

    /**
     * {@inheritdoc}
     */
    public function run(array $tasks, array $hosts)
    {
        $localhost = new Localhost();
        $limit = (int)$this->input->getOption('limit') ?: count($hosts);

        // We need contexts here for usage inside `on` function. Pass input/output to callback of it.
        // This allows to use code like this in parallel mode:
        //
        //     host('prod')
        //         ->set('branch', function () {
        //             return input()->getOption('branch') ?: 'production';
        //     })
        //
        // Otherwise `input()` wont be accessible (i.e. null)
        //
        Context::push(new Context($localhost, $this->input, $this->output));
        {
            Storage::persist($hosts);
        }
        Context::pop();

        foreach ($tasks as $task) {
            $success = true;
            $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);

                    switch ($exitCode) {
                        case 1:
                            throw new GracefulShutdownException();
                        case 2:
                            $success = false;
                            break;
                        case 255:
                            throw new Exception();
                    }
                }
            }

            if ($success) {
                $this->informer->endTask($task);
            } else {
                $this->informer->taskError();
            }
        }
    }

    /**
     * Run task on hosts.
     *
     * @param Host[] $hosts
     */
    private function runTask(array $hosts, Task $task): int
    {
        $processes = [];

        foreach ($hosts as $host) {
            if ($task->shouldBePerformed($host)) {
                $processes[$host->getHostname()] = $this->getProcess($host, $task);
                if ($task->isOnce()) {
                    break;
                }
            }
        }

        $callback = function (string $type, string $host, string $output) {
            $output = rtrim($output);
            if (strlen($output) !== 0) {
                $this->output->writeln($output);
            }
        };

        $this->startProcesses($processes);

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

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

    /**
     * Get process for task on host.
     */
    protected function getProcess(Host $host, Task $task): Process
    {
        $dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
        $options = $this->generateOptions();
        $arguments = $this->generateArguments();
        $hostname = $host->getHostname();
        $taskName = $task->getName();
        $configFile = $host->get('host_config_storage');
        $value = $this->input->getOption('file');
        $file = $value ? "--file='$value'" : '';

        if ($this->output->isDecorated()) {
            $options .= ' --ansi';
        }

        $command = "$dep $file worker $arguments $options --hostname $hostname --task $taskName --config-file $configFile";
        $process = new Process($command);

        if (!defined('DEPLOYER_PARALLEL_PTY')) {
            $process->setPty(true);
        }

        return $process;
    }

    /**
     * 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
     */
    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
     * @return void
     */
    protected function gatherOutput(array $processes, callable $callback)
    {
        foreach ($processes as $host => $process) {
            $output = $process->getIncrementalOutput();
            if (strlen($output) !== 0) {
                $callback(Process::OUT, $host, $output);
            }

            $errorOutput = $process->getIncrementalErrorOutput();
            if (strlen($errorOutput) !== 0) {
                $callback(Process::ERR, $host, $errorOutput);
            }
        }
    }

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

        return 0;
    }

    /**
     * Generate options and arguments string.
     */
    private function generateOptions(): string
    {
        $input = (string)(new VerbosityString($this->output));

        // Get user arguments
        foreach ($this->console->getUserDefinition()->getArguments() as $argument) {
            $value = $this->input->getArgument($argument->getName());
            if (strlen($value) !== 0) {
                $input .= " $value";
            }
        }

        // Get user options
        foreach ($this->console->getUserDefinition()->getOptions() as $option) {
            $name = $option->getName();
            $value = $this->input->getOption($name);

            if (null !== $value && strlen($value) !== 0) {
                $input .= " --{$name}";

                if ($option->acceptValue()) {
                    $input .= " {$value}";
                }
            }
        }

        return $input;
    }

    private function generateArguments(): string
    {
        $arguments = '';

        if ($this->input->hasArgument('stage')) {
            // Some people rely on stage argument, so pass it to worker too.
            $arguments .= $this->input->getArgument('stage');
        }

        return $arguments;
    }
}


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\Console\Application;
11		use Deployer\Console\Output\Informer;
12		use Deployer\Console\Output\VerbosityString;
13		use Deployer\Exception\Exception;
14		use Deployer\Exception\GracefulShutdownException;
15		use Deployer\Host\Host;
16		use Deployer\Host\Localhost;
17		use Deployer\Host\Storage;
18		use Deployer\Task\Context;
19		use Deployer\Task\Task;
20		use Symfony\Component\Console\Input\InputInterface;
21		use Symfony\Component\Console\Output\OutputInterface;
22		use Symfony\Component\Process\Process;
23
24		class ParallelExecutor implements ExecutorInterface
25		{
26		/**
27		* @var InputInterface
28		*/
29		private $input;
30
31		/**
32		* @var OutputInterface
33		*/
34		private $output;
35
36		/**
37		* @var Informer
38		*/
39		private $informer;
40
41		/**
42		* @var Application
43		*/
44		private $console;
45
46	4	public function __construct(
47		InputInterface $input,
48		OutputInterface $output,
49		Informer $informer,
50		Application $console
51		) {
52	4	$this->input = $input;
53	4	$this->output = $output;
54	4	$this->informer = $informer;
55	4	$this->console = $console;
56	4	}
57
58		/**
59		* {@inheritdoc}
60		*/
61	4	public function run(array $tasks, array $hosts)
62		{
63	4	$localhost = new Localhost();
64	4	$limit = (int)$this->input->getOption('limit') ?: count($hosts);
65
66		// We need contexts here for usage inside `on` function. Pass input/output to callback of it.
67		// This allows to use code like this in parallel mode:
68		//
69		// host('prod')
70		// ->set('branch', function () {
71		// return input()->getOption('branch') ?: 'production';
72		// })
73		//
74		// Otherwise `input()` wont be accessible (i.e. null)
75		//
76	4	Context::push(new Context($localhost, $this->input, $this->output));
77		{
78	4	Storage::persist($hosts);
79		}
80	4	Context::pop();
81
82	4	foreach ($tasks as $task) {
83	4	$success = true;
84	4	$this->informer->startTask($task);
85
86	4	if ($task->isLocal()) {
87	2	Storage::load($hosts);
88		{
89	2	$task->run(new Context($localhost, $this->input, $this->output));
90		}
91	2	Storage::flush($hosts);
92		} else {
93	4	foreach (array_chunk($hosts, $limit) as $chunk) {
94	4	$exitCode = $this->runTask($chunk, $task);
95
96		switch ($exitCode) {
97	4	case 1:
98		throw new GracefulShutdownException();
99	4	case 2:
100		$success = false;
101		break;
102	4	case 255:
103	4	throw new Exception();
104		}
105		}
106		}
107
108	4	if ($success) {
109	4	$this->informer->endTask($task);
110		} else {
111	4	$this->informer->taskError();
112		}
113		}
114	4	}
115
116		/**
117		* Run task on hosts.
118		*
119		* @param Host[] $hosts
120		*/
121	4	private function runTask(array $hosts, Task $task): int
122		{
123	4	$processes = [];
124
125	4	foreach ($hosts as $host) {
126	4	if ($task->shouldBePerformed($host)) {
127	4	$processes[$host->getHostname()] = $this->getProcess($host, $task);
128	4	if ($task->isOnce()) {
129	4	break;
130		}
131		}
132		}
133
134	4	$callback = function (string $type, string $host, string $output) {
135	4	$output = rtrim($output);
136	4	if (strlen($output) !== 0) {
137	4	$this->output->writeln($output);
138		}
139	4	};
140
141	4	$this->startProcesses($processes);
142
143	4	while ($this->areRunning($processes)) {
144	4	$this->gatherOutput($processes, $callback);
145	4	usleep(1000);
146		}
147	4	$this->gatherOutput($processes, $callback);
148
149	4	return $this->gatherExitCodes($processes);
150		}
151
152		/**
153		* Get process for task on host.
154		*/
155	4	protected function getProcess(Host $host, Task $task): Process
156		{
157	4	$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
158	4	$options = $this->generateOptions();
159	4	$arguments = $this->generateArguments();
160	4	$hostname = $host->getHostname();
161	4	$taskName = $task->getName();
162	4	$configFile = $host->get('host_config_storage');
163	4	$value = $this->input->getOption('file');
164	4	$file = $value ? "--file='$value'" : '';
165
166	4	if ($this->output->isDecorated()) {
167		$options .= ' --ansi';
168		}
169
170	4	$command = "$dep $file worker $arguments $options --hostname $hostname --task $taskName --config-file $configFile";
171	4	$process = new Process($command);
172
173	4	if (!defined('DEPLOYER_PARALLEL_PTY')) {
174		$process->setPty(true);
175		}
176
177	4	return $process;
178		}
179
180		/**
181		* Start all of the processes.
182		*
183		* @param Process[] $processes
184		* @return void
185		*/
186	4	protected function startProcesses(array $processes)
187		{
188	4	foreach ($processes as $process) {
189	4	$process->start();
190		}
191	4	}
192
193		/**
194		* Determine if any of the processes are running.
195		*
196		* @param Process[] $processes
197		*/
198	4	protected function areRunning(array $processes): bool
199		{
200	4	foreach ($processes as $process) {
201	4	if ($process->isRunning()) {
202	4	return true;
203		}
204		}
205	4	return false;
206		}
207
208		/**
209		* Gather the output from all of the processes.
210		*
211		* @param Process[] $processes
212		* @return void
213		*/
214	4	protected function gatherOutput(array $processes, callable $callback)
215		{
216	4	foreach ($processes as $host => $process) {
217	4	$output = $process->getIncrementalOutput();
218	4	if (strlen($output) !== 0) {
219	4	$callback(Process::OUT, $host, $output);
220		}
221
222	4	$errorOutput = $process->getIncrementalErrorOutput();
223	4	if (strlen($errorOutput) !== 0) {
224	4	$callback(Process::ERR, $host, $errorOutput);
225		}
226		}
227	4	}
228
229		/**
230		* Gather the cumulative exit code for the processes.
231		*
232		* @param Process[] $processes
233		*/
234	4	protected function gatherExitCodes(array $processes): int
235		{
236	4	foreach ($processes as $process) {
237	4	if ($process->getExitCode() > 0) {
238	4	return $process->getExitCode();
239		}
240		}
241
242	4	return 0;
243		}
244
245		/**
246		* Generate options and arguments string.
247		*/
248	4	private function generateOptions(): string
249		{
250	4	$input = (string)(new VerbosityString($this->output));
251
252		// Get user arguments
253	4	foreach ($this->console->getUserDefinition()->getArguments() as $argument) {
254		$value = $this->input->getArgument($argument->getName());
255		if (strlen($value) !== 0) {
256		$input .= " $value";
257		}
258		}
259
260		// Get user options
261	4	foreach ($this->console->getUserDefinition()->getOptions() as $option) {
262	3	$name = $option->getName();
263	3	$value = $this->input->getOption($name);
264
265	3	if (null !== $value && strlen($value) !== 0) {
266		$input .= " --{$name}";
267
268		if ($option->acceptValue()) {
269	3	$input .= " {$value}";
270		}
271		}
272		}
273
274	4	return $input;
275		}
276
277	4	private function generateArguments(): string
278		{
279	4	$arguments = '';
280
281	4	if ($this->input->hasArgument('stage')) {
282		// Some people rely on stage argument, so pass it to worker too.
283	4	$arguments .= $this->input->getArgument('stage');
284		}
285
286	4	return $arguments;
287		}
288		}
289

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 075b5b...370756 )

ParallelExecutor::generateOptions() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like