ParallelExecutor::run() - Code Metrics - Inspection of "Fix tricky bug with passing IO inside on" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 94bf29...92eaf3 )

by Anton

created 2017-08-12 15:43 UTC

ParallelExecutor::run() B

↳ Parent: ParallelExecutor

Complexity

Conditions	9
Paths	14

Size

Total Lines	54
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	23
CRAP Score	9.1244

Importance

Changes

Metric	Value
cc	9
eloc	28
nc	14
nop	2
dl	0
loc	54
ccs	23
cts	26
cp	0.8846
crap	9.1244
rs	7.255
c	0
b	0
f	0

How to fix Long Method

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


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	$hostname = $host->getHostname();
162	3	$taskName = $task->getName();
163	3	$configFile = $host->get('host_config_storage');
164	3	$value = $this->input->getOption('file');
165	3	$file = $value ? "--file='$value'" : '';
166
167	3	if ($this->output->isDecorated()) {
168		$options .= ' --ansi';
169		}
170
171	3	$command = "$dep $file worker $options --hostname $hostname --task $taskName --config-file $configFile";
172	3	$process = new Process($command);
173
174	3	if (!defined('DEPLOYER_PARALLEL_PTY')) {
175		$process->setPty(true);
176		}
177
178	3	return $process;
179		}
180
181		/**
182		* Start all of the processes.
183		*
184		* @param Process[] $processes
185		* @return void
186		*/
187	3	protected function startProcesses(array $processes)
188		{
189	3	foreach ($processes as $process) {
190	3	$process->start();
191		}
192	3	}
193
194		/**
195		* Determine if any of the processes are running.
196		*
197		* @param Process[] $processes
198		* @return bool
199		*/
200	3	protected function areRunning(array $processes)
201		{
202	3	foreach ($processes as $process) {
203	3	if ($process->isRunning()) {
204	3	return true;
205		}
206		}
207	3	return false;
208		}
209
210		/**
211		* Gather the output from all of the processes.
212		*
213		* @param Process[] $processes
214		* @param callable $callback
215		*/
216	3	protected function gatherOutput(array $processes, callable $callback)
217		{
218	3	foreach ($processes as $host => $process) {
219		$methods = [
220	3	Process::OUT => 'getIncrementalOutput',
221		Process::ERR => 'getIncrementalErrorOutput',
222		];
223	3	foreach ($methods as $type => $method) {
224	3	$output = $process->{$method}();
225	3	if (!empty($output)) {
226	3	$callback($type, $host, $output);
227		}
228		}
229		}
230	3	}
231
232		/**
233		* Gather the cumulative exit code for the processes.
234		*
235		* @param Process[] $processes
236		* @return int
237		*/
238	3	protected function gatherExitCodes(array $processes)
239		{
240	3	$code = 0;
241	3	foreach ($processes as $process) {
242	3	if ($process->getExitCode() > 0) {
243	3	$code = $process->getExitCode();
244		}
245		}
246	3	return $code;
247		}
248
249		/**
250		* Generate options and arguments string.
251		* @return string
252		*/
253	3	private function generateOptions()
254		{
255	3	$verbosity = new VerbosityString($this->output);
256	3	$input = $verbosity;
257
258		// Get user arguments
259	3	foreach ($this->console->getUserDefinition()->getArguments() as $argument) {
260		$value = $this->input->getArgument($argument->getName());
261		if ($value) {
262		$input .= " $value";
263		}
264		}
265
266		// Get user options
267	3	foreach ($this->console->getUserDefinition()->getOptions() as $option) {
268	2	$name = $option->getName();
269	2	$value = $this->input->getOption($name);
270
271	2	if ($value) {
272		$input .= " --{$name}";
273
274		if ($option->acceptValue()) {
275	2	$input .= " {$value}";
276		}
277		}
278		}
279
280	3	return $input;
281		}
282		}
283

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 94bf29...92eaf3 )

ParallelExecutor::run() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like