Pool::withBinary() - Code Metrics - Inspection of "Configurable executable" - spatie/async - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#111)

by Alexander

created 2020-04-14 13:00 UTC

Pool::withBinary() A

↳ Parent: Pool

Complexity

Conditions	1
Paths	1

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
nc	1
nop	1
dl	0
loc	6
rs	10
c	0
b	0
f	0

<?php

namespace Spatie\Async;

use ArrayAccess;
use InvalidArgumentException;
use Spatie\Async\Process\ParallelProcess;
use Spatie\Async\Process\Runnable;
use Spatie\Async\Process\SynchronousProcess;
use Spatie\Async\Runtime\ParentRuntime;

class Pool implements ArrayAccess
{
    public const DEFAULT_PHP_BINARY = 'php';

    public static $forceSynchronous = false;

    protected $concurrency = 20;
    protected $tasksPerProcess = 1;
    protected $timeout = 300;
    protected $sleepTime = 50000;

    /** @var \Spatie\Async\Process\Runnable[] */
    protected $queue = [];

    /** @var \Spatie\Async\Process\Runnable[] */
    protected $inProgress = [];

    /** @var \Spatie\Async\Process\Runnable[] */
    protected $finished = [];

    /** @var \Spatie\Async\Process\Runnable[] */
    protected $failed = [];

    /** @var \Spatie\Async\Process\Runnable[] */
    protected $timeouts = [];

    protected $results = [];

    protected $status;

    protected $stopped = false;

    protected $binary = self::DEFAULT_PHP_BINARY;

    public function __construct()
    {
        if (static::isSupported()) {
            $this->registerListener();
        }

        $this->status = new PoolStatus($this);
    }

    /**
     * @return static
     */
    public static function create()
    {
        return new static();
    }

    public static function isSupported(): bool
    {
        return
            function_exists('pcntl_async_signals')
            && function_exists('posix_kill')
            && ! self::$forceSynchronous;
    }

    public function concurrency(int $concurrency): self
    {
        $this->concurrency = $concurrency;

        return $this;
    }

    public function timeout(float $timeout): self
    {
        $this->timeout = $timeout;
$answer = 42;

$correct = false;

$correct = (bool) $answer;

        return $this;
    }

    public function withBinary(string $binary = self::DEFAULT_PHP_BINARY): self
    {
        $this->binary = $binary;

        return $this;
    }

    public function autoload(string $autoloader): self
    {
        ParentRuntime::init($autoloader);

        return $this;
    }

    public function sleepTime(int $sleepTime): self
    {
        $this->sleepTime = $sleepTime;

        return $this;
    }

    public function notify()
    {
        if (count($this->inProgress) >= $this->concurrency) {
            return;
        }

        $process = array_shift($this->queue);

        if (! $process) {
            return;
        }

        $this->putInProgress($process);
    }

    /**
     * @param \Spatie\Async\Process\Runnable|callable $process
     * @param int|null $outputLength
     *
     * @return \Spatie\Async\Process\Runnable
     */
    public function add($process, ?int $outputLength = null): Runnable
    {
        if (! is_callable($process) && ! $process instanceof Runnable) {
            throw new InvalidArgumentException('The process passed to Pool::add should be callable.');
        }

        if (! $process instanceof Runnable) {
            $process = ParentRuntime::createProcess($process, $outputLength, $this->binary);
        }

        $process->withBinary($this->binary);
        $this->putInQueue($process);

        return $process;
    }

    public function wait(?callable $intermediateCallback = null): array
    {
        while ($this->inProgress) {

            foreach ($this->inProgress as $process) {
                if ($process->getCurrentExecutionTime() > $this->timeout) {
                    $this->markAsTimedOut($process);
                }

                if ($process instanceof SynchronousProcess) {
                    $this->markAsFinished($process);
                }
            }

            if (! $this->inProgress) {

                break;
            }

            if ($intermediateCallback) {
                call_user_func_array($intermediateCallback, [$this]);
            }

            usleep($this->sleepTime);
        }

        return $this->results;
    }

    public function putInQueue(Runnable $process)
    {
        $this->queue[$process->getId()] = $process;

        $this->notify();
    }

    public function putInProgress(Runnable $process)
    {
        if ($this->stopped) {
            return;
        }

        if ($process instanceof ParallelProcess) {
            $process->getProcess()->setTimeout($this->timeout);
        }

        $process->start();

        unset($this->queue[$process->getId()]);

        $this->inProgress[$process->getPid()] = $process;
    }

    public function markAsFinished(Runnable $process)
    {
        unset($this->inProgress[$process->getPid()]);

        $this->notify();

        $this->results[] = $process->triggerSuccess();

        $this->finished[$process->getPid()] = $process;
    }

    public function markAsTimedOut(Runnable $process)
    {
        unset($this->inProgress[$process->getPid()]);

        $this->notify();

        $process->triggerTimeout();

        $this->timeouts[$process->getPid()] = $process;
    }

    public function markAsFailed(Runnable $process)
    {
        unset($this->inProgress[$process->getPid()]);

        $this->notify();

        $process->triggerError();

        $this->failed[$process->getPid()] = $process;
    }

    public function offsetExists($offset)
    {
        // TODO

        return false;
    }

    public function offsetGet($offset)
    {
        // TODO
    }

    public function offsetSet($offset, $value)
    {
        $this->add($value);
    }

    public function offsetUnset($offset)
    {
        // TODO
    }

    /**
     * @return \Spatie\Async\Process\Runnable[]
     */
    public function getQueue(): array
    {
        return $this->queue;
    }

    /**
     * @return \Spatie\Async\Process\Runnable[]
     */
    public function getInProgress(): array
    {
        return $this->inProgress;
    }

    /**
     * @return \Spatie\Async\Process\Runnable[]
     */
    public function getFinished(): array
    {
        return $this->finished;
    }

    /**
     * @return \Spatie\Async\Process\Runnable[]
     */
    public function getFailed(): array
    {
        return $this->failed;
    }

    /**
     * @return \Spatie\Async\Process\Runnable[]
     */
    public function getTimeouts(): array
    {
        return $this->timeouts;
    }

    public function status(): PoolStatus
    {
        return $this->status;
    }

    protected function registerListener()
    {
        pcntl_async_signals(true);

        pcntl_signal(SIGCHLD, function ($signo, $status) {
            while (true) {
                $pid = pcntl_waitpid(-1, $processState, WNOHANG | WUNTRACED);

                if ($pid <= 0) {
                    break;
                }

                $process = $this->inProgress[$pid] ?? null;

                if (! $process) {
                    continue;
                }

                if ($status['status'] === 0) {
                    $this->markAsFinished($process);

                    continue;
                }

                $this->markAsFailed($process);
            }
        });
    }

    public function stop()
    {
        $this->stopped = true;
    }
}


1			<?php
2
3			namespace Spatie\Async;
4
5			use ArrayAccess;
6			use InvalidArgumentException;
7			use Spatie\Async\Process\ParallelProcess;
8			use Spatie\Async\Process\Runnable;
9			use Spatie\Async\Process\SynchronousProcess;
10			use Spatie\Async\Runtime\ParentRuntime;
11
12			class Pool implements ArrayAccess
13			{
14			public const DEFAULT_PHP_BINARY = 'php';
15
16			public static $forceSynchronous = false;
17
18			protected $concurrency = 20;
19			protected $tasksPerProcess = 1;
20			protected $timeout = 300;
21			protected $sleepTime = 50000;
22
23			/** @var \Spatie\Async\Process\Runnable[] */
24			protected $queue = [];
25
26			/** @var \Spatie\Async\Process\Runnable[] */
27			protected $inProgress = [];
28
29			/** @var \Spatie\Async\Process\Runnable[] */
30			protected $finished = [];
31
32			/** @var \Spatie\Async\Process\Runnable[] */
33			protected $failed = [];
34
35			/** @var \Spatie\Async\Process\Runnable[] */
36			protected $timeouts = [];
37
38			protected $results = [];
39
40			protected $status;
41
42			protected $stopped = false;
43
44			protected $binary = self::DEFAULT_PHP_BINARY;
45
46			public function __construct()
47			{
48			if (static::isSupported()) {
49			$this->registerListener();
50			}
51
52			$this->status = new PoolStatus($this);
53			}
54
55			/**
56			* @return static
57			*/
58			public static function create()
59			{
60			return new static();
61			}
62
63			public static function isSupported(): bool
64			{
65			return
66			function_exists('pcntl_async_signals')
67			&& function_exists('posix_kill')
68			&& ! self::$forceSynchronous;
69			}
70
71			public function concurrency(int $concurrency): self
72			{
73			$this->concurrency = $concurrency;
74
75			return $this;
76			}
77
78			public function timeout(float $timeout): self
79			{
80			$this->timeout = $timeout;
			0 ignored issues – show Documentation Bug introduced 2020-03-13 16:11 UTC by Report Bug Copy Issue Report The property `$timeout` was declared of type `integer`, but `$timeout` is of type `double`. Maybe add a type cast? This check looks for assignments to scalar types that may be of the wrong type. To ensure the code behaves as expected, it may be a good idea to add an explicit type cast. $answer = 42; $correct = false; $correct = (bool) $answer; Loading history...
81
82			return $this;
83			}
84
85			public function withBinary(string $binary = self::DEFAULT_PHP_BINARY): self
86			{
87			$this->binary = $binary;
88
89			return $this;
90			}
91
92			public function autoload(string $autoloader): self
93			{
94			ParentRuntime::init($autoloader);
95
96			return $this;
97			}
98
99			public function sleepTime(int $sleepTime): self
100			{
101			$this->sleepTime = $sleepTime;
102
103			return $this;
104			}
105
106			public function notify()
107			{
108			if (count($this->inProgress) >= $this->concurrency) {
109			return;
110			}
111
112			$process = array_shift($this->queue);
113
114			if (! $process) {
115			return;
116			}
117
118			$this->putInProgress($process);
119			}
120
121			/**
122			* @param \Spatie\Async\Process\Runnable\|callable $process
123			* @param int\|null $outputLength
124			*
125			* @return \Spatie\Async\Process\Runnable
126			*/
127			public function add($process, ?int $outputLength = null): Runnable
128			{
129			if (! is_callable($process) && ! $process instanceof Runnable) {
130			throw new InvalidArgumentException('The process passed to Pool::add should be callable.');
131			}
132
133			if (! $process instanceof Runnable) {
134			$process = ParentRuntime::createProcess($process, $outputLength, $this->binary);
135			}
136
137			$process->withBinary($this->binary);
138			$this->putInQueue($process);
139
140			return $process;
141			}
142
143			public function wait(?callable $intermediateCallback = null): array
144			{
145			while ($this->inProgress) {
			0 ignored issues – show Bug Best Practice introduced 2018-01-30 14:27 UTC by Report Bug Copy Issue Report The expression `$this->inProgress` of type `Spatie\Async\Process\Runnable[]` is implicitly converted to a boolean; are you sure this is intended? If so, consider using `! empty($expr)` instead to make it clear that you intend to check for an array without elements. This check marks implicit conversions of arrays to boolean values in a comparison. While in PHP an empty array is considered to be equal (but not identical) to false, this is not always apparent. Consider making the comparison explicit by using `empty(..)` or `! empty(...)` instead. Loading history...
146			foreach ($this->inProgress as $process) {
147			if ($process->getCurrentExecutionTime() > $this->timeout) {
148			$this->markAsTimedOut($process);
149			}
150
151			if ($process instanceof SynchronousProcess) {
152			$this->markAsFinished($process);
153			}
154			}
155
156			if (! $this->inProgress) {
			0 ignored issues – show Bug Best Practice introduced 2020-02-06 11:59 UTC by Report Bug Copy Issue Report The expression `$this->inProgress` of type `Spatie\Async\Process\Runnable[]` is implicitly converted to a boolean; are you sure this is intended? If so, consider using `empty($expr)` instead to make it clear that you intend to check for an array without elements. This check marks implicit conversions of arrays to boolean values in a comparison. While in PHP an empty array is considered to be equal (but not identical) to false, this is not always apparent. Consider making the comparison explicit by using `empty(..)` or `! empty(...)` instead. Loading history...
157			break;
158			}
159
160			if ($intermediateCallback) {
161			call_user_func_array($intermediateCallback, [$this]);
162			}
163
164			usleep($this->sleepTime);
165			}
166
167			return $this->results;
168			}
169
170			public function putInQueue(Runnable $process)
171			{
172			$this->queue[$process->getId()] = $process;
173
174			$this->notify();
175			}
176
177			public function putInProgress(Runnable $process)
178			{
179			if ($this->stopped) {
180			return;
181			}
182
183			if ($process instanceof ParallelProcess) {
184			$process->getProcess()->setTimeout($this->timeout);
185			}
186
187			$process->start();
188
189			unset($this->queue[$process->getId()]);
190
191			$this->inProgress[$process->getPid()] = $process;
192			}
193
194			public function markAsFinished(Runnable $process)
195			{
196			unset($this->inProgress[$process->getPid()]);
197
198			$this->notify();
199
200			$this->results[] = $process->triggerSuccess();
201
202			$this->finished[$process->getPid()] = $process;
203			}
204
205			public function markAsTimedOut(Runnable $process)
206			{
207			unset($this->inProgress[$process->getPid()]);
208
209			$this->notify();
210
211			$process->triggerTimeout();
212
213			$this->timeouts[$process->getPid()] = $process;
214			}
215
216			public function markAsFailed(Runnable $process)
217			{
218			unset($this->inProgress[$process->getPid()]);
219
220			$this->notify();
221
222			$process->triggerError();
223
224			$this->failed[$process->getPid()] = $process;
225			}
226
227			public function offsetExists($offset)
228			{
229			// TODO
230
231			return false;
232			}
233
234			public function offsetGet($offset)
235			{
236			// TODO
237			}
238
239			public function offsetSet($offset, $value)
240			{
241			$this->add($value);
242			}
243
244			public function offsetUnset($offset)
245			{
246			// TODO
247			}
248
249			/**
250			* @return \Spatie\Async\Process\Runnable[]
251			*/
252			public function getQueue(): array
253			{
254			return $this->queue;
255			}
256
257			/**
258			* @return \Spatie\Async\Process\Runnable[]
259			*/
260			public function getInProgress(): array
261			{
262			return $this->inProgress;
263			}
264
265			/**
266			* @return \Spatie\Async\Process\Runnable[]
267			*/
268			public function getFinished(): array
269			{
270			return $this->finished;
271			}
272
273			/**
274			* @return \Spatie\Async\Process\Runnable[]
275			*/
276			public function getFailed(): array
277			{
278			return $this->failed;
279			}
280
281			/**
282			* @return \Spatie\Async\Process\Runnable[]
283			*/
284			public function getTimeouts(): array
285			{
286			return $this->timeouts;
287			}
288
289			public function status(): PoolStatus
290			{
291			return $this->status;
292			}
293
294			protected function registerListener()
295			{
296			pcntl_async_signals(true);
297
298			pcntl_signal(SIGCHLD, function ($signo, $status) {
299			while (true) {
300			$pid = pcntl_waitpid(-1, $processState, WNOHANG \| WUNTRACED);
301
302			if ($pid <= 0) {
303			break;
304			}
305
306			$process = $this->inProgress[$pid] ?? null;
307
308			if (! $process) {
309			continue;
310			}
311
312			if ($status['status'] === 0) {
313			$this->markAsFinished($process);
314
315			continue;
316			}
317
318			$this->markAsFailed($process);
319			}
320			});
321			}
322
323			public function stop()
324			{
325			$this->stopped = true;
326			}
327			}
328

spatie / async

GitHub Access Token became invalid

Pull Request — master (#111)

Pool::withBinary() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like