ParallelExecutor::areRunning() - Code Metrics - Inspection of "make sure `deploy:lock` can be called several time..." - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#1470)

by Markus

created 2017-12-21 16:09 UTC

ParallelExecutor::areRunning() A

↳ Parent: ParallelExecutor

Complexity

Conditions	3
Paths	3

Size

Total Lines	9
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
eloc	5
nc	3
nop	1
dl	0
loc	9
ccs	5
cts	5
cp	1
crap	3
rs	9.6666
c	0
b	0
f	0

<?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	1	public function __construct(InputInterface $input, OutputInterface $output, Informer $informer, Application $console)
53		{
54	1	$this->input = $input;
55	1	$this->output = $output;
56	1	$this->informer = $informer;
57	1	$this->console = $console;
58	1	}
59
60		/**
61		* {@inheritdoc}
62		*/
63	1	public function run($tasks, $hosts)
64		{
65	1	$localhost = new Localhost();
66	1	$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 () {
73		// return input()->getOption('branch') ?: 'production';
74		// })
75		//
76		// Otherwise `input()` wont be accessible (i.e. null)
77		//
78	1	Context::push(new Context($localhost, $this->input, $this->output));
79		{
80	1	Storage::persist($hosts);
81		}
82	1	Context::pop();
83
84	1	foreach ($tasks as $task) {
85	1	$success = true;
86	1	$this->informer->startTask($task);
87
88	1	if ($task->isLocal()) {
89	1	Storage::load($hosts);
90		{
91	1	$task->run(new Context($localhost, $this->input, $this->output));
92		}
93	1	Storage::flush($hosts);
94		} else {
95	1	foreach (array_chunk($hosts, $limit) as $chunk) {
96	1	$exitCode = $this->runTask($chunk, $task);
97
98		switch ($exitCode) {
99	1	case 1:
100		throw new GracefulShutdownException();
101	1	case 2:
102		$success = false;
103		break;
104	1	case 255:
105	1	throw new Exception();
106		}
107		}
108		}
109
110	1	if ($success) {
111	1	$this->informer->endTask($task);
112		} else {
113	1	$this->informer->taskError();
114		}
115		}
116	1	}
117
118		/**
119		* Run task on hosts.
120		*
121		* @param Host[] $hosts
122		* @param Task $task
123		* @return int
124		*/
125	1	private function runTask(array $hosts, Task $task)
126		{
127	1	$processes = [];
128
129	1	foreach ($hosts as $host) {
130	1	$processes[$host->getHostname()] = $this->getProcess($host, $task);
131		}
132
133	1	$callback = function ($type, $host, $output) {
134	1	$output = rtrim($output);
135	1	if (!empty($output)) {
136	1	$this->output->writeln($output);
137		}
138	1	};
139
140	1	$this->startProcesses($processes);
141
142	1	while ($this->areRunning($processes)) {
143	1	$this->gatherOutput($processes, $callback);
144		}
145	1	$this->gatherOutput($processes, $callback);
146
147	1	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	1	protected function getProcess($host, Task $task)
158		{
159	1	$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
160	1	$options = $this->generateOptions();
161	1	$arguments = $this->generateArguments();
162	1	$hostname = $host->getHostname();
163	1	$taskName = $task->getName();
164	1	$configFile = $host->get('host_config_storage');
165	1	$value = $this->input->getOption('file');
166	1	$file = $value ? "--file='$value'" : '';
167
168	1	if ($this->output->isDecorated()) {
169		$options .= ' --ansi';
170		}
171
172	1	$command = "$dep $file worker $arguments $options --hostname $hostname --task $taskName --config-file $configFile";
173	1	$process = new Process($command);
174
175	1	if (!defined('DEPLOYER_PARALLEL_PTY')) {
176		$process->setPty(true);
177		}
178
179	1	return $process;
180		}
181
182		/**
183		* Start all of the processes.
184		*
185		* @param Process[] $processes
186		* @return void
187		*/
188	1	protected function startProcesses(array $processes)
189		{
190	1	foreach ($processes as $process) {
191	1	$process->start();
192		}
193	1	}
194
195		/**
196		* Determine if any of the processes are running.
197		*
198		* @param Process[] $processes
199		* @return bool
200		*/
201	1	protected function areRunning(array $processes)
202		{
203	1	foreach ($processes as $process) {
204	1	if ($process->isRunning()) {
205	1	return true;
206		}
207		}
208	1	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	1	protected function gatherOutput(array $processes, callable $callback)
218		{
219	1	foreach ($processes as $host => $process) {
220		$methods = [
221	1	Process::OUT => 'getIncrementalOutput',
222		Process::ERR => 'getIncrementalErrorOutput',
223		];
224	1	foreach ($methods as $type => $method) {
225	1	$output = $process->{$method}();
226	1	if (!empty($output)) {
227	1	$callback($type, $host, $output);
228		}
229		}
230		}
231	1	}
232
233		/**
234		* Gather the cumulative exit code for the processes.
235		*
236		* @param Process[] $processes
237		* @return int
238		*/
239	1	protected function gatherExitCodes(array $processes)
240		{
241	1	$code = 0;
242	1	foreach ($processes as $process) {
243	1	if ($process->getExitCode() > 0) {
244	1	$code = $process->getExitCode();
245		}
246		}
247	1	return $code;
248		}
249
250		/**
251		* Generate options and arguments string.
252		* @return string
253		*/
254	1	private function generateOptions()
255		{
256	1	$verbosity = new VerbosityString($this->output);
257	1	$input = $verbosity;
258
259		// Get user arguments
260	1	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	1	foreach ($this->console->getUserDefinition()->getOptions() as $option) {
269		$name = $option->getName();
270		$value = $this->input->getOption($name);
271
272		if ($value) {
273		$input .= " --{$name}";
274
275		if ($option->acceptValue()) {
276		$input .= " {$value}";
277		}
278		}
279		}
280
281	1	return $input;
282		}
283
284	1	private function generateArguments(): string
285		{
286	1	$arguments = '';
287
288	1	if ($this->input->hasArgument('stage')) {
289		// Some people rely on stage argument, so pass it to worker too.
290	1	$arguments .= $this->input->getArgument('stage');
291		}
292
293	1	return $arguments;
294		}
295		}
296

deployphp / deployer

GitHub Access Token became invalid

Pull Request — master (#1470)

ParallelExecutor::areRunning() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like