ParallelExecutor::generateArguments() - Code Metrics - Inspection of "Fix #1299" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 76afdf...93d214 )

by Anton

created 2017-08-12 16:03 UTC

ParallelExecutor::generateArguments() A

↳ Parent: ParallelExecutor

Complexity

Conditions	2
Paths	2

Size

Total Lines	11
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	5
nc	2
nop	0
dl	0
loc	11
rs	9.4285
c	0
b	0
f	0
ccs	5
cts	5
cp	1
crap	2

<?php
/* (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;

    /**
     * @param InputInterface $input
     * @param OutputInterface $output
     * @param Informer $informer
     * @param Application $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($tasks, $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
     * @param Task $task
     * @return int
     */
    private function runTask(array $hosts, Task $task)
    {
        $processes = [];

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

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

        $this->startProcesses($processes);

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

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

    /**
     * Get process for task on host.
     *
     * @param Host $host
     * @param Task $task
     * @return Process
     */
    protected function getProcess($host, Task $task)
    {
        $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
     * @return bool
     */
    protected function areRunning(array $processes)
    {
        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
     */
    protected function gatherOutput(array $processes, callable $callback)
    {
        foreach ($processes as $host => $process) {
            $methods = [
                Process::OUT => 'getIncrementalOutput',
                Process::ERR => 'getIncrementalErrorOutput',
            ];
            foreach ($methods as $type => $method) {
                $output = $process->{$method}();
                if (!empty($output)) {
                    $callback($type, $host, $output);
                }
            }
        }
    }

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

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

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

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

            if ($value) {
                $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
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		/**
47		* @param InputInterface $input
48		* @param OutputInterface $output
49		* @param Informer $informer
50		* @param Application $console
51		*/
52	3	public function __construct(InputInterface $input, OutputInterface $output, Informer $informer, Application $console)
53		{
54	3	$this->input = $input;
55	3	$this->output = $output;
56	3	$this->informer = $informer;
57	3	$this->console = $console;
58	3	}
59
60		/**
61		* {@inheritdoc}
62		*/
63	3	public function run($tasks, $hosts)
64		{
65	3	$localhost = new Localhost();
66	3	$limit = (int)$this->input->getOption('limit') ?: count($hosts);
67
68		// We need contexts here for usage inside `on` function. Pass input/output to callback of it.
69		// This allows to use code like this in parallel mode:
70		//
71		// host('prod')
72		// ->set('branch', function () {
		1 ignored issue – show Unused Code Comprehensibility introduced 2017-08-12 15:52 UTC by Report Bug Copy Issue Report `62%` of this comment could be valid code. Did you maybe forget this after debugging? Sometimes obsolete code just ends up commented out instead of removed. In this case it is better to remove the code once you have checked you do not need it. The code might also have been commented out for debugging purposes. In this case it is vital that someone uncomments it again or your project may behave in very unexpected ways in production. This check looks for comments that seem to be mostly valid code and reports them. Loading history...
73		// return input()->getOption('branch') ?: 'production';
		1 ignored issue – show Unused Code Comprehensibility introduced 2017-08-12 15:52 UTC by Report Bug Copy Issue Report `65%` of this comment could be valid code. Did you maybe forget this after debugging? Sometimes obsolete code just ends up commented out instead of removed. In this case it is better to remove the code once you have checked you do not need it. The code might also have been commented out for debugging purposes. In this case it is vital that someone uncomments it again or your project may behave in very unexpected ways in production. This check looks for comments that seem to be mostly valid code and reports them. Loading history...
74		// })
75		//
76		// Otherwise `input()` wont be accessible (i.e. null)
77		//
78	3	Context::push(new Context($localhost, $this->input, $this->output));
79		{
80	3	Storage::persist($hosts);
81		}
82	3	Context::pop();
83
84	3	foreach ($tasks as $task) {
85	3	$success = true;
86	3	$this->informer->startTask($task);
87
88	3	if ($task->isLocal()) {
89	2	Storage::load($hosts);
90		{
91	2	$task->run(new Context($localhost, $this->input, $this->output));
92		}
93	2	Storage::flush($hosts);
94		} else {
95	3	foreach (array_chunk($hosts, $limit) as $chunk) {
96	3	$exitCode = $this->runTask($chunk, $task);
97
98		switch ($exitCode) {
99	3	case 1:
100		throw new GracefulShutdownException();
101	3	case 2:
102		$success = false;
103		break;
104	3	case 255:
105	3	throw new Exception();
106		}
107		}
108		}
109
110	3	if ($success) {
111	3	$this->informer->endTask($task);
112		} else {
113	3	$this->informer->taskError();
114		}
115		}
116	3	}
117
118		/**
119		* Run task on hosts.
120		*
121		* @param Host[] $hosts
122		* @param Task $task
123		* @return int
124		*/
125	3	private function runTask(array $hosts, Task $task)
126		{
127	3	$processes = [];
128
129	3	foreach ($hosts as $host) {
130	3	$processes[$host->getHostname()] = $this->getProcess($host, $task);
131		}
132
133	3	$callback = function ($type, $host, $output) {
134	3	$output = rtrim($output);
135	3	if (!empty($output)) {
136	3	$this->output->writeln($output);
137		}
138	3	};
139
140	3	$this->startProcesses($processes);
141
142	3	while ($this->areRunning($processes)) {
143	3	$this->gatherOutput($processes, $callback);
144		}
145	3	$this->gatherOutput($processes, $callback);
146
147	3	return $this->gatherExitCodes($processes);
148		}
149
150		/**
151		* Get process for task on host.
152		*
153		* @param Host $host
154		* @param Task $task
155		* @return Process
156		*/
157	3	protected function getProcess($host, Task $task)
158		{
159	3	$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
160	3	$options = $this->generateOptions();
161	3	$arguments = $this->generateArguments();
162	3	$hostname = $host->getHostname();
163	3	$taskName = $task->getName();
164	3	$configFile = $host->get('host_config_storage');
165	3	$value = $this->input->getOption('file');
166	3	$file = $value ? "--file='$value'" : '';
167
168	3	if ($this->output->isDecorated()) {
169		$options .= ' --ansi';
170		}
171
172	3	$command = "$dep $file worker $arguments $options --hostname $hostname --task $taskName --config-file $configFile";
173	3	$process = new Process($command);
174
175	3	if (!defined('DEPLOYER_PARALLEL_PTY')) {
176		$process->setPty(true);
177		}
178
179	3	return $process;
180		}
181
182		/**
183		* Start all of the processes.
184		*
185		* @param Process[] $processes
186		* @return void
187		*/
188	3	protected function startProcesses(array $processes)
189		{
190	3	foreach ($processes as $process) {
191	3	$process->start();
192		}
193	3	}
194
195		/**
196		* Determine if any of the processes are running.
197		*
198		* @param Process[] $processes
199		* @return bool
200		*/
201	3	protected function areRunning(array $processes)
202		{
203	3	foreach ($processes as $process) {
204	3	if ($process->isRunning()) {
205	3	return true;
206		}
207		}
208	3	return false;
209		}
210
211		/**
212		* Gather the output from all of the processes.
213		*
214		* @param Process[] $processes
215		* @param callable $callback
216		*/
217	3	protected function gatherOutput(array $processes, callable $callback)
218		{
219	3	foreach ($processes as $host => $process) {
220		$methods = [
221	3	Process::OUT => 'getIncrementalOutput',
222		Process::ERR => 'getIncrementalErrorOutput',
223		];
224	3	foreach ($methods as $type => $method) {
225	3	$output = $process->{$method}();
226	3	if (!empty($output)) {
227	3	$callback($type, $host, $output);
228		}
229		}
230		}
231	3	}
232
233		/**
234		* Gather the cumulative exit code for the processes.
235		*
236		* @param Process[] $processes
237		* @return int
238		*/
239	3	protected function gatherExitCodes(array $processes)
240		{
241	3	$code = 0;
242	3	foreach ($processes as $process) {
243	3	if ($process->getExitCode() > 0) {
244	3	$code = $process->getExitCode();
245		}
246		}
247	3	return $code;
248		}
249
250		/**
251		* Generate options and arguments string.
252		* @return string
253		*/
254	3	private function generateOptions()
255		{
256	3	$verbosity = new VerbosityString($this->output);
257	3	$input = $verbosity;
258
259		// Get user arguments
260	3	foreach ($this->console->getUserDefinition()->getArguments() as $argument) {
261		$value = $this->input->getArgument($argument->getName());
262		if ($value) {
263		$input .= " $value";
264		}
265		}
266
267		// Get user options
268	3	foreach ($this->console->getUserDefinition()->getOptions() as $option) {
269	2	$name = $option->getName();
270	2	$value = $this->input->getOption($name);
271
272	2	if ($value) {
273		$input .= " --{$name}";
274
275		if ($option->acceptValue()) {
276	2	$input .= " {$value}";
277		}
278		}
279		}
280
281	3	return $input;
282		}
283
284	3	private function generateArguments(): string
285		{
286	3	$arguments = '';
287
288	3	if ($this->input->hasArgument('stage')) {
289		// Some people rely on stage argument, so pass it to worker too.
290	3	$arguments .= $this->input->getArgument('stage');
291		}
292
293	3	return $arguments;
294		}
295		}
296

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 76afdf...93d214 )

ParallelExecutor::generateArguments() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like