ParallelExecutor::gatherExitCodes() - Code Metrics - Inspection of "Migrate to PHP 7.2.25" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 31f734...513d45 )

by Anton

created 2019-12-15 14:31 UTC

ParallelExecutor::gatherExitCodes() A

↳ Parent: ParallelExecutor

Complexity

Conditions	3
Paths	3

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
nc	3
nop	1
dl	0
loc	10
ccs	0
cts	9
cp	0
crap	12
rs	9.9332
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\Input\Argument;
use Deployer\Console\Input\Option;
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
     * @return int
     */
    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()) {
                    $task->setHasRun();
                }
            }
        }

        $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 = Process::fromShellCommandline($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): 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 $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.
     */
    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
    {
        /** @var string[] $inputs */
        $inputs = [
            (string)(new VerbosityString($this->output)),
        ];

        $userDefinition = $this->console->getUserDefinition();
        // Get user arguments
        foreach ($userDefinition->getArguments() as $argument) {
            $inputs[] = Argument::toString($this->input, $argument);
        }

        // Get user options
        foreach ($userDefinition->getOptions() as $option) {
            $inputs[] = Option::toString($this->input, $option);
        }

        return implode(' ', array_filter($inputs, static function (string $item): bool {
            return $item !== '';
        }));
    }

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

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 31f734...513d45 )

ParallelExecutor::gatherExitCodes() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like