Pool::getInProgress() - Code Metrics - Inspection of "Add output too large error" - spatie/async - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8b76df...221683 )

by Brent

created 2018-01-30 14:22 UTC

Pool::getInProgress() A

↳ Parent: Pool

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nc	1
nop	0
dl	0
loc	4
rs	10
c	0
b	0
f	0

<?php

namespace Spatie\Async;

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

class Pool implements ArrayAccess
{
    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;

    public function __construct()
    {
        $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(int $timeout): self
    {
        $this->timeout = $timeout;

        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
     *
     * @return \Spatie\Async\Process\Runnable
     */
    public function add($process): 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);
        }

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


1			<?php
2
3			namespace Spatie\Async;
4
5			use ArrayAccess;
6			use InvalidArgumentException;
7			use Spatie\Async\Process\Runnable;
8			use Spatie\Async\Runtime\ParentRuntime;
9			use Spatie\Async\Process\ParallelProcess;
10			use Spatie\Async\Process\SynchronousProcess;
11
12			class Pool implements ArrayAccess
13			{
14			public static $forceSynchronous = false;
15
16			protected $concurrency = 20;
17			protected $tasksPerProcess = 1;
18			protected $timeout = 300;
19			protected $sleepTime = 50000;
20
21			/** @var \Spatie\Async\Process\Runnable[] */
22			protected $queue = [];
23
24			/** @var \Spatie\Async\Process\Runnable[] */
25			protected $inProgress = [];
26
27			/** @var \Spatie\Async\Process\Runnable[] */
28			protected $finished = [];
29
30			/** @var \Spatie\Async\Process\Runnable[] */
31			protected $failed = [];
32
33			/** @var \Spatie\Async\Process\Runnable[] */
34			protected $timeouts = [];
35
36			protected $results = [];
37
38			protected $status;
39
40			public function __construct()
41			{
42			$this->registerListener();
43
44			$this->status = new PoolStatus($this);
45			}
46
47			/**
48			* @return static
49			*/
50			public static function create()
51			{
52			return new static();
53			}
54
55			public static function isSupported(): bool
56			{
57			return
58			function_exists('pcntl_async_signals')
59			&& function_exists('posix_kill')
60			&& ! self::$forceSynchronous;
61			}
62
63			public function concurrency(int $concurrency): self
64			{
65			$this->concurrency = $concurrency;
66
67			return $this;
68			}
69
70			public function timeout(int $timeout): self
71			{
72			$this->timeout = $timeout;
73
74			return $this;
75			}
76
77			public function autoload(string $autoloader): self
78			{
79			ParentRuntime::init($autoloader);
80
81			return $this;
82			}
83
84			public function sleepTime(int $sleepTime): self
85			{
86			$this->sleepTime = $sleepTime;
87
88			return $this;
89			}
90
91			public function notify()
92			{
93			if (count($this->inProgress) >= $this->concurrency) {
94			return;
95			}
96
97			$process = array_shift($this->queue);
98
99			if (! $process) {
100			return;
101			}
102
103			$this->putInProgress($process);
104			}
105
106			/**
107			* @param \Spatie\Async\Process\Runnable\|callable $process
108			*
109			* @return \Spatie\Async\Process\Runnable
110			*/
111			public function add($process): Runnable
112			{
113			if (! is_callable($process) && ! $process instanceof Runnable) {
114			throw new InvalidArgumentException('The process passed to Pool::add should be callable.');
115			}
116
117			if (! $process instanceof Runnable) {
118			$process = ParentRuntime::createProcess($process);
119			}
120
121			$this->putInQueue($process);
122
123			return $process;
124			}
125
126			public function wait(?callable $intermediateCallback = null): array
127			{
128			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...
129			foreach ($this->inProgress as $process) {
130			if ($process->getCurrentExecutionTime() > $this->timeout) {
131			$this->markAsTimedOut($process);
132			}
133
134			if ($process instanceof SynchronousProcess) {
135			$this->markAsFinished($process);
136			}
137			}
138
139			if (! $this->inProgress) {
			0 ignored issues – show Bug Best Practice introduced 2018-01-16 07:43 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...
140			break;
141			}
142
143			if ($intermediateCallback) {
144			call_user_func_array($intermediateCallback, [$this]);
145			}
146
147			usleep($this->sleepTime);
148			}
149
150			return $this->results;
151			}
152
153			public function putInQueue(Runnable $process)
154			{
155			$this->queue[$process->getId()] = $process;
156
157			$this->notify();
158			}
159
160			public function putInProgress(Runnable $process)
161			{
162			if ($process instanceof ParallelProcess) {
163			$process->getProcess()->setTimeout($this->timeout);
164			}
165
166			$process->start();
167
168			unset($this->queue[$process->getId()]);
169
170			$this->inProgress[$process->getPid()] = $process;
171			}
172
173			public function markAsFinished(Runnable $process)
174			{
175			unset($this->inProgress[$process->getPid()]);
176
177			$this->notify();
178
179			$this->results[] = $process->triggerSuccess();
180
181			$this->finished[$process->getPid()] = $process;
182			}
183
184			public function markAsTimedOut(Runnable $process)
185			{
186			unset($this->inProgress[$process->getPid()]);
187
188			$this->notify();
189
190			$process->triggerTimeout();
191
192			$this->timeouts[$process->getPid()] = $process;
193			}
194
195			public function markAsFailed(Runnable $process)
196			{
197			unset($this->inProgress[$process->getPid()]);
198
199			$this->notify();
200
201			$process->triggerError();
202
203			$this->failed[$process->getPid()] = $process;
204			}
205
206			public function offsetExists($offset)
207			{
208			// TODO
209
210			return false;
211			}
212
213			public function offsetGet($offset)
214			{
215			// TODO
216			}
217
218			public function offsetSet($offset, $value)
219			{
220			$this->add($value);
221			}
222
223			public function offsetUnset($offset)
224			{
225			// TODO
226			}
227
228			/**
229			* @return \Spatie\Async\Process\Runnable[]
230			*/
231			public function getQueue(): array
232			{
233			return $this->queue;
234			}
235
236			/**
237			* @return \Spatie\Async\Process\Runnable[]
238			*/
239			public function getInProgress(): array
240			{
241			return $this->inProgress;
242			}
243
244			/**
245			* @return \Spatie\Async\Process\Runnable[]
246			*/
247			public function getFinished(): array
248			{
249			return $this->finished;
250			}
251
252			/**
253			* @return \Spatie\Async\Process\Runnable[]
254			*/
255			public function getFailed(): array
256			{
257			return $this->failed;
258			}
259
260			/**
261			* @return \Spatie\Async\Process\Runnable[]
262			*/
263			public function getTimeouts(): array
264			{
265			return $this->timeouts;
266			}
267
268			public function status(): PoolStatus
269			{
270			return $this->status;
271			}
272
273			protected function registerListener()
274			{
275			pcntl_async_signals(true);
276
277			pcntl_signal(SIGCHLD, function ($signo, $status) {
278			while (true) {
279			$pid = pcntl_waitpid(-1, $processState, WNOHANG \| WUNTRACED);
280
281			if ($pid <= 0) {
282			break;
283			}
284
285			$process = $this->inProgress[$pid] ?? null;
286
287			if (! $process) {
288			continue;
289			}
290
291			if ($status['status'] === 0) {
292			$this->markAsFinished($process);
293
294			continue;
295			}
296
297			$this->markAsFailed($process);
298			}
299			});
300			}
301			}
302

spatie / async

GitHub Access Token became invalid

Push — master ( 8b76df...221683 )

Pool::getInProgress() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like