Pool::onRunCompleted() - Code Metrics - Inspection of "coverage->100%, refactor priorities, add more test..." - graze/parallel-process - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#19)

by Harry

created 2018-09-24 15:37 UTC

Pool::onRunCompleted() A

↳ Parent: Pool

Complexity

Conditions	5
Paths	6

Size

Total Lines	16
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	5

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	16
rs	9.5555
c	0
b	0
f	0
ccs	12
cts	12
cp	1
cc	5
nc	6
nop	1
crap	5

<?php
/**
 * This file is part of graze/parallel-process.
 *
 * Copyright © 2018 Nature Delivered Ltd. <https://www.graze.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 *
 * @license https://github.com/graze/parallel-process/blob/master/LICENSE.md
 * @link    https://github.com/graze/parallel-process
 */

namespace Graze\ParallelProcess;

use Exception;
use Graze\DataStructure\Collection\Collection;
use Graze\ParallelProcess\Event\EventDispatcherTrait;
use Graze\ParallelProcess\Event\PoolRunEvent;
use Graze\ParallelProcess\Event\PriorityChangedEvent;
use Graze\ParallelProcess\Event\RunEvent;
use InvalidArgumentException;
use Symfony\Component\Process\Process;
use Throwable;

/**
 * Class Pool
 *
 * A Pool is a arbitrary collection of runs that can be used to group runs together when displaying with a
 * Table
 *
 * A Pool can transition from `not_running` back to `running` again. But it cannot transition back to `not_started`.
 * This means that multiple `COMPLETED` and `STARTED` events can be sent out for a single pool.
 *
 * ```
 * not_started -> running <-> not_running
 * ```
 *
 * @package Graze\ParallelProcess
 */
class Pool extends Collection implements RunInterface, PoolInterface, PrioritisedInterface
{
    use EventDispatcherTrait;
    use RunningStateTrait;
    use PrioritisedTrait;

    /** @var RunInterface[] */
    protected $items = [];
    /** @var RunInterface[] */
    protected $waiting = [];
    /** @var RunInterface[] */
    protected $running = [];
    /** @var RunInterface[] */
    protected $complete = [];
    /** @var Exception[]|Throwable[] */
    private $exceptions = [];
    /** @var array */
    private $tags;

    /**
     * RunCollection constructor.
     *
     * @param RunInterface[] $runs
     * @param array          $tags
     * @param float          $priority
     */
    public function __construct(array $runs = [], array $tags = [], $priority = 1.0)
    {
        parent::__construct([]);

        $this->tags = $tags;

        array_map([$this, 'add'], $runs);
        $this->priority = $priority;
    }

    /**
     * @param RunInterface|Process $item
     * @param array                $tags
     *
     * @return $this
     */
    public function add($item, array $tags = [])
    {
        if ($item instanceof Process) {
            return $this->add(new ProcessRun($item, $tags));
        }
        if (!$item instanceof RunInterface) {

            throw new InvalidArgumentException('item must implement `RunInterface`');
        }

        parent::add($item);
        $status = 'waiting';
        if ($item->isRunning()) {
            $status = 'running';
            $this->running[] = $item;
        } elseif ($item->hasStarted()) {
            $status = 'finished';
            $this->complete[] = $item;
        } else {
            $this->waiting[] = $item;
        }

        $item->addListener(RunEvent::STARTED, [$this, 'onRunStarted']);
        $item->addListener(RunEvent::COMPLETED, [$this, 'onRunCompleted']);
        $item->addListener(RunEvent::FAILED, [$this, 'onRunFailed']);

        $this->dispatch(PoolRunEvent::POOL_RUN_ADDED, new PoolRunEvent($this, $item));

        if ($status != 'waiting' && $this->state != static::STATE_RUNNING) {
            $this->setStarted();
            $this->dispatch(RunEvent::STARTED, new RunEvent($this));
        }
        if ($status == 'finished' && $this->state != static::STATE_NOT_RUNNING) {
            $this->setFinished();
            $this->dispatch(RunEvent::COMPLETED, new RunEvent($this));
        }

        return $this;
    }

    /**
     * When a run starts, check our current state and start ourselves in required
     *
     * @param RunEvent $event
     */
    public function onRunStarted(RunEvent $event)
    {
        $index = array_search($event->getRun(), $this->waiting, true);
        if ($index !== false) {
            unset($this->waiting[$index]);
        }
        $this->running[] = $event->getRun();
        if ($this->state == static::STATE_NOT_STARTED) {
            $this->setStarted();
            $this->dispatch(RunEvent::STARTED, new RunEvent($this));
        }
        $this->dispatch(RunEvent::UPDATED, new RunEvent($this));
    }

    /**
     * When a run is completed, check if everything has finished
     *
     * @param RunEvent $event
     */
    public function onRunCompleted(RunEvent $event)
    {
        $index = array_search($event->getRun(), $this->running, true);
        if ($index !== false) {
            unset($this->running[$index]);
        }
        $this->complete[] = $event->getRun();
        $this->dispatch(RunEvent::UPDATED, new RunEvent($this));
        if (count($this->waiting) === 0 && count($this->running) === 0) {
            $this->setFinished();
            if ($this->isSuccessful()) {
                $this->dispatch(RunEvent::SUCCESSFUL, new RunEvent($this));
            } else {
                $this->dispatch(RunEvent::FAILED, new RunEvent($this));
            }
            $this->dispatch(RunEvent::COMPLETED, new RunEvent($this));
        }
    }

    /**
     * Handle any errors returned from the child run
     *
     * @param RunEvent $event
     */
    public function onRunFailed(RunEvent $event)
    {
        $this->exceptions = array_merge($this->exceptions, $event->getRun()->getExceptions());
    }

    /**
     * Has this run been started before
     *
     * @return bool
     */
    public function hasStarted()
    {
        return $this->getState() !== static::STATE_NOT_STARTED;
    }

    /**
     * Start all non running children
     *
     * @return $this
     *
     * @throws \Graze\ParallelProcess\Exceptions\NotRunningException
     */
    public function start()
    {
        foreach ($this->items as $run) {
            if (!$run->hasStarted()) {
                $run->start();
            }
        }
        return $this;
    }

    /**
     * Was this run successful
     *
     * @return bool
     */
    public function isSuccessful()
    {
        if ($this->getState() === static::STATE_NOT_RUNNING) {
            foreach ($this->items as $run) {
                if (!$run->isSuccessful()) {
                    return false;
                }
            }
            return true;
        }
        return false;
    }

    /**
     * If the run was unsuccessful, get the error if applicable
     *
     * @return Exception[]|Throwable[]
     */
    public function getExceptions()
    {
        return $this->exceptions;
    }

    /**
     * We think this is running
     *
     * @return bool
     */
    public function isRunning()
    {
        return $this->getState() === static::STATE_RUNNING;
    }

    /**
     * Pools to see if this process is running
     *
     * @return bool
     */
    public function poll()
    {
        foreach ($this->running as $run) {
            $run->poll();
        }
        return $this->isRunning();
    }

    /**
     * Get a set of tags associated with this run
     *
     * @return array
     */
    public function getTags()
    {
        return $this->tags;
    }

    /**
     * @return float[]|null an array of values of the current position, max, and percentage. null if not applicable
     */
    public function getProgress()
    {
        return [count($this->complete), count($this->items), count($this->complete) / count($this->items)];
    }

    /**
     * @return string[]
     */
    protected function getEventNames()
    {
        return [
            RunEvent::STARTED,
            RunEvent::COMPLETED,
            RunEvent::SUCCESSFUL,
            RunEvent::FAILED,
            RunEvent::UPDATED,
            PoolRunEvent::POOL_RUN_ADDED,
            PriorityChangedEvent::CHANGED,
        ];
    }

    /**
     * Run this pool of runs and block until they are complete.
     *
     * Note this will run the parent pool
     *
     * @param float $interval
     *
     * @return bool `true` if all the runs were successful
     */
    public function run($interval = self::CHECK_INTERVAL)
    {
        $this->start();

        $sleep = (int) ($interval * 1000000);
        while ($this->poll()) {
            usleep($sleep);
        }

        return $this->isSuccessful();
    }

    /**
     * @return RunInterface[]
     */
    public function getWaiting()
    {
        return array_values($this->waiting);
    }

    /**
     * @return RunInterface[]
     */
    public function getRunning()
    {
        return array_values($this->running);
    }

    /**
     * @return RunInterface[]
     */
    public function getFinished()
    {
        return array_values($this->complete);
    }
}


1		<?php
2		/**
3		* This file is part of graze/parallel-process.
4		*
5		* Copyright © 2018 Nature Delivered Ltd. <https://www.graze.com>
6		*
7		* For the full copyright and license information, please view the LICENSE
8		* file that was distributed with this source code.
9		*
10		* @license https://github.com/graze/parallel-process/blob/master/LICENSE.md
11		* @link https://github.com/graze/parallel-process
12		*/
13
14		namespace Graze\ParallelProcess;
15
16		use Exception;
17		use Graze\DataStructure\Collection\Collection;
18		use Graze\ParallelProcess\Event\EventDispatcherTrait;
19		use Graze\ParallelProcess\Event\PoolRunEvent;
20		use Graze\ParallelProcess\Event\PriorityChangedEvent;
21		use Graze\ParallelProcess\Event\RunEvent;
22		use InvalidArgumentException;
23		use Symfony\Component\Process\Process;
24		use Throwable;
25
26		/**
27		* Class Pool
28		*
29		* A Pool is a arbitrary collection of runs that can be used to group runs together when displaying with a
30		* Table
31		*
32		* A Pool can transition from `not_running` back to `running` again. But it cannot transition back to `not_started`.
33		* This means that multiple `COMPLETED` and `STARTED` events can be sent out for a single pool.
34		*
35		* ```
36		* not_started -> running <-> not_running
37		* ```
38		*
39		* @package Graze\ParallelProcess
40		*/
41		class Pool extends Collection implements RunInterface, PoolInterface, PrioritisedInterface
42		{
43		use EventDispatcherTrait;
44		use RunningStateTrait;
45		use PrioritisedTrait;
46
47		/** @var RunInterface[] */
48		protected $items = [];
49		/** @var RunInterface[] */
50		protected $waiting = [];
51		/** @var RunInterface[] */
52		protected $running = [];
53		/** @var RunInterface[] */
54		protected $complete = [];
55		/** @var Exception[]\|Throwable[] */
56		private $exceptions = [];
57		/** @var array */
58		private $tags;
59
60		/**
61		* RunCollection constructor.
62		*
63		* @param RunInterface[] $runs
64		* @param array $tags
65		* @param float $priority
66		*/
67	73	public function __construct(array $runs = [], array $tags = [], $priority = 1.0)
68		{
69	73	parent::__construct([]);
70
71	73	$this->tags = $tags;
72
73	73	array_map([$this, 'add'], $runs);
74	73	$this->priority = $priority;
75	73	}
76
77		/**
78		* @param RunInterface\|Process $item
79		* @param array $tags
80		*
81		* @return $this
82		*/
83	57	public function add($item, array $tags = [])
84		{
85	57	if ($item instanceof Process) {
86	32	return $this->add(new ProcessRun($item, $tags));
87		}
88	57	if (!$item instanceof RunInterface) {
		0 ignored issues – show introduced 2018-05-22 13:13 UTC by Report Bug Copy Issue Report `$item` is always a sub-type of `Graze\ParallelProcess\RunInterface`. Loading history...
89	2	throw new InvalidArgumentException('item must implement `RunInterface`');
90		}
91
92	55	parent::add($item);
93	55	$status = 'waiting';
94	55	if ($item->isRunning()) {
95	1	$status = 'running';
96	1	$this->running[] = $item;
97	55	} elseif ($item->hasStarted()) {
98	1	$status = 'finished';
99	1	$this->complete[] = $item;
100		} else {
101	54	$this->waiting[] = $item;
102		}
103
104	55	$item->addListener(RunEvent::STARTED, [$this, 'onRunStarted']);
105	55	$item->addListener(RunEvent::COMPLETED, [$this, 'onRunCompleted']);
106	55	$item->addListener(RunEvent::FAILED, [$this, 'onRunFailed']);
107
108	55	$this->dispatch(PoolRunEvent::POOL_RUN_ADDED, new PoolRunEvent($this, $item));
109
110	55	if ($status != 'waiting' && $this->state != static::STATE_RUNNING) {
111	1	$this->setStarted();
112	1	$this->dispatch(RunEvent::STARTED, new RunEvent($this));
113		}
114	55	if ($status == 'finished' && $this->state != static::STATE_NOT_RUNNING) {
115	1	$this->setFinished();
116	1	$this->dispatch(RunEvent::COMPLETED, new RunEvent($this));
117		}
118
119	55	return $this;
120		}
121
122		/**
123		* When a run starts, check our current state and start ourselves in required
124		*
125		* @param RunEvent $event
126		*/
127	37	public function onRunStarted(RunEvent $event)
128		{
129	37	$index = array_search($event->getRun(), $this->waiting, true);
130	37	if ($index !== false) {
131	37	unset($this->waiting[$index]);
132		}
133	37	$this->running[] = $event->getRun();
134	37	if ($this->state == static::STATE_NOT_STARTED) {
135	37	$this->setStarted();
136	37	$this->dispatch(RunEvent::STARTED, new RunEvent($this));
137		}
138	37	$this->dispatch(RunEvent::UPDATED, new RunEvent($this));
139	37	}
140
141		/**
142		* When a run is completed, check if everything has finished
143		*
144		* @param RunEvent $event
145		*/
146	36	public function onRunCompleted(RunEvent $event)
147		{
148	36	$index = array_search($event->getRun(), $this->running, true);
149	36	if ($index !== false) {
150	35	unset($this->running[$index]);
151		}
152	36	$this->complete[] = $event->getRun();
153	36	$this->dispatch(RunEvent::UPDATED, new RunEvent($this));
154	36	if (count($this->waiting) === 0 && count($this->running) === 0) {
155	34	$this->setFinished();
156	34	if ($this->isSuccessful()) {
157	28	$this->dispatch(RunEvent::SUCCESSFUL, new RunEvent($this));
158		} else {
159	6	$this->dispatch(RunEvent::FAILED, new RunEvent($this));
160		}
161	34	$this->dispatch(RunEvent::COMPLETED, new RunEvent($this));
162		}
163	36	}
164
165		/**
166		* Handle any errors returned from the child run
167		*
168		* @param RunEvent $event
169		*/
170	7	public function onRunFailed(RunEvent $event)
171		{
172	7	$this->exceptions = array_merge($this->exceptions, $event->getRun()->getExceptions());
173	7	}
174
175		/**
176		* Has this run been started before
177		*
178		* @return bool
179		*/
180	14	public function hasStarted()
181		{
182	14	return $this->getState() !== static::STATE_NOT_STARTED;
183		}
184
185		/**
186		* Start all non running children
187		*
188		* @return $this
189		*
190		* @throws \Graze\ParallelProcess\Exceptions\NotRunningException
191		*/
192	8	public function start()
193		{
194	8	foreach ($this->items as $run) {
195	8	if (!$run->hasStarted()) {
196	8	$run->start();
197		}
198		}
199	8	return $this;
200		}
201
202		/**
203		* Was this run successful
204		*
205		* @return bool
206		*/
207	40	public function isSuccessful()
208		{
209	40	if ($this->getState() === static::STATE_NOT_RUNNING) {
210	34	foreach ($this->items as $run) {
211	34	if (!$run->isSuccessful()) {
212	34	return false;
213		}
214		}
215	28	return true;
216		}
217	10	return false;
218		}
219
220		/**
221		* If the run was unsuccessful, get the error if applicable
222		*
223		* @return Exception[]\|Throwable[]
224		*/
225	1	public function getExceptions()
226		{
227	1	return $this->exceptions;
228		}
229
230		/**
231		* We think this is running
232		*
233		* @return bool
234		*/
235	62	public function isRunning()
236		{
237	62	return $this->getState() === static::STATE_RUNNING;
238		}
239
240		/**
241		* Pools to see if this process is running
242		*
243		* @return bool
244		*/
245	35	public function poll()
246		{
247	35	foreach ($this->running as $run) {
248	25	$run->poll();
249		}
250	35	return $this->isRunning();
251		}
252
253		/**
254		* Get a set of tags associated with this run
255		*
256		* @return array
257		*/
258	3	public function getTags()
259		{
260	3	return $this->tags;
261		}
262
263		/**
264		* @return float[]\|null an array of values of the current position, max, and percentage. null if not applicable
265		*/
266	3	public function getProgress()
267		{
268	3	return [count($this->complete), count($this->items), count($this->complete) / count($this->items)];
269		}
270
271		/**
272		* @return string[]
273		*/
274	62	protected function getEventNames()
275		{
276		return [
277	62	RunEvent::STARTED,
278		RunEvent::COMPLETED,
279		RunEvent::SUCCESSFUL,
280		RunEvent::FAILED,
281		RunEvent::UPDATED,
282	62	PoolRunEvent::POOL_RUN_ADDED,
283		PriorityChangedEvent::CHANGED,
284		];
285		}
286
287		/**
288		* Run this pool of runs and block until they are complete.
289		*
290		* Note this will run the parent pool
291		*
292		* @param float $interval
293		*
294		* @return bool `true` if all the runs were successful
295		*/
296	7	public function run($interval = self::CHECK_INTERVAL)
297		{
298	7	$this->start();
299
300	7	$sleep = (int) ($interval * 1000000);
301	7	while ($this->poll()) {
302	1	usleep($sleep);
303		}
304
305	7	return $this->isSuccessful();
306		}
307
308		/**
309		* @return RunInterface[]
310		*/
311	7	public function getWaiting()
312		{
313	7	return array_values($this->waiting);
314		}
315
316		/**
317		* @return RunInterface[]
318		*/
319	6	public function getRunning()
320		{
321	6	return array_values($this->running);
322		}
323
324		/**
325		* @return RunInterface[]
326		*/
327	6	public function getFinished()
328		{
329	6	return array_values($this->complete);
330		}
331		}
332

graze / parallel-process

Pull Request — master (#19)

Pool::onRunCompleted() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like