Container::results() - Code Metrics - Inspection of "[WIP] Bernard" - ackintosh/snidel - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#19)

by Akihito

created 2017-07-29 07:23 UTC

Container::results() A

↳ Parent: Container

Complexity

Conditions	3
Paths	3

Size

Total Lines	16
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	16
rs	9.4285
cc	3
eloc	10
nc	3
nop	0

<?php
declare(ticks = 1);
namespace Ackintosh\Snidel\Fork;

use Ackintosh\Snidel\ActiveWorkerSet;
use Ackintosh\Snidel\Config;
use Ackintosh\Snidel\Error;
use Ackintosh\Snidel\Pcntl;
use Ackintosh\Snidel\Result\Formatter as ResultFormatter;
use Ackintosh\Snidel\Task\Formatter as TaskFormatter;
use Ackintosh\Snidel\Worker;
use Bernard\Consumer;
use Bernard\Driver\FlatFileDriver;
use Bernard\Message\PlainMessage;
use Bernard\Producer;
use Bernard\QueueFactory\PersistentFactory;
use Bernard\Router\SimpleRouter;
use Bernard\Serializer;
use Symfony\Component\EventDispatcher\EventDispatcher;

class Container
{
    /** @var Process */
    private $master;

    /** @var \Ackintosh\Snidel\Pcntl */
    private $pcntl;

    /** @var \Ackintosh\Snidel\Error */
    private $error;

    /** @var \Ackintosh\Snidel\Task\QueueInterface */
    private $taskQueue;



    /** @var \Ackintosh\Snidel\Log */
    private $log;

    /** @var array */
    private $signals = [
        SIGTERM,
        SIGINT,
    ];

    /** @var \Ackintosh\Snidel\Config */
    private $config;

    /** @var  int */
    private $receivedSignal;

    private $queuedCount = 0;
    private $dequeuedCount = 0;

    private $factory;
    private $producer;
    private $consumer;
    private $resultQueue;

    /**
     * @param   int     $ownerPid
/**
 * @param array $germany
 * @param array $island
 * @param array $italy
 */
function finale($germany, $island) {
    return "2:1";
}
     */
    public function __construct(Config $config, $log)
    {
        $this->log              = $log;
        $this->config           = $config;
        $this->pcntl            = new Pcntl();
        $this->error            = new Error();

        $driver = new FlatFileDriver('/tmp/hoge');
        $this->factory = new PersistentFactory($driver, new Serializer());
        $this->producer = new Producer($this->factory, new EventDispatcher());

        $router = new SimpleRouter();
        $router->add('Result', $this);
        $this->consumer = new Consumer($router, new EventDispatcher());
    }

    /**
     * @param   \Ackintosh\Snidel\Task
     * @return  void
     * @throws  \RuntimeException
     */
    public function enqueue($task)
    {
        try {
            $this->producer->produce(
                new PlainMessage(
                    'Task',
                    [
                        'task' => TaskFormatter::serialize($task),
                    ]
                )
            );
            $this->queuedCount++;

        } catch (\RuntimeException $e) {
            throw $e;
        }
    }

    /**
     * @return  int
     */
    public function queuedCount()
    {
        return $this->queuedCount;
    }

    /**
     * @return  int
     */
    public function dequeuedCount()
    {
        return $this->dequeuedCount;
    }

    /**
     * fork master process
     *
     * @return Process $master
     * @throws \RuntimeException
     */
    public function forkMaster()
    {
        try {
            $this->master = $this->pcntl->fork();
        } catch (\RuntimeException $e) {
            $message = 'failed to fork master: ' . $e->getMessage();
            $this->log->error($message);
            throw new \RuntimeException($message);
        }

        $this->log->setMasterPid($this->master->getPid());

        if (getmypid() === $this->config->get('ownerPid')) {
            // owner
            $this->log->info('pid: ' . getmypid());
            $this->resultQueue  = $this->factory->create('result');

            return $this->master;
        } else {
            // master
            $activeWorkerSet = new ActiveWorkerSet();
            $this->log->info('pid: ' . $this->master->getPid());

            foreach ($this->signals as $sig) {
                $this->pcntl->signal($sig, function ($sig) use ($activeWorkerSet) {
                    $this->receivedSignal = $sig;
                    $this->log->info('received signal: ' . $sig);

                    if ($activeWorkerSet->count() === 0) {
                        $this->log->info('no worker is active.');
                    } else {
                        $this->log->info('------> sending signal to workers. signal: ' . $sig);
                        $activeWorkerSet->terminate($sig);
                        $this->log->info('<------ sent signal');
                    }
                    exit;
                });
            }

            $concurrency = (int)$this->config->get('concurrency');
            for ($i = 0; $i < $concurrency; $i++) {
                $activeWorkerSet->add($this->forkWorker());
            }
            $status = null;
            while (($workerPid = $this->pcntl->waitpid(-1, $status, WNOHANG)) !== -1) {
                if ($workerPid === true || $workerPid === 0) {
                    usleep(100000);
                    continue;
                }
                $activeWorkerSet->delete($workerPid);
                $activeWorkerSet->add($this->forkWorker());
                $status = null;
            }
            exit;
        }
    }

    /**
     * fork worker process
     *
     * @return  \Ackintosh\Snidel\Worker
     * @throws  \RuntimeException
     */
    private function forkWorker()
    {
        try {
            $process = $this->pcntl->fork();
        } catch (\RuntimeException $e) {
            $message = 'failed to fork worker: ' . $e->getMessage();
            $this->log->error($message);
            throw new \RuntimeException($message);
        }

        $worker = new Worker($process);

        if (getmypid() === $this->master->getPid()) {
            // master
            $this->log->info('forked worker. pid: ' . $worker->getPid());
            return $worker;
        } else {
            // worker
            // @codeCoverageIgnoreStart
            $this->log->info('has forked. pid: ' . getmypid());

            foreach ($this->signals as $sig) {
                $this->pcntl->signal($sig, function ($sig) {
                    $this->receivedSignal = $sig;
                    exit;
                }, false);
            }

            register_shutdown_function(function () use ($worker) {
                if (!$worker->hasTask()) {
                    return;
                }

                if (!$worker->done() && $this->receivedSignal === null) {
                    $worker->error();
                }
            });

            $this->log->info('----> started the function.');
            try {
                $worker->run();
            } catch (\RuntimeException $e) {
                $this->log->error($e->getMessage());
                exit;
            }
            $this->log->info('<---- completed the function.');

            $this->log->info('queued the result and exit.');
            exit;
            // @codeCoverageIgnoreEnd
        }
    }

    /**
     * @return  bool
     */
    public function existsMaster()
    {
        return $this->master !== null;
    }

    /**
     * send signal to master process
     *
     * @return  void
     */
    public function sendSignalToMaster($sig = SIGTERM)
    {
        $this->log->info('----> sending signal to master. signal: ' . $sig);
        posix_kill($this->master->getPid(), $sig);
        $this->log->info('<---- sent signal.');

        $this->log->info('----> waiting for master shutdown.');
        $status = null;
        $this->pcntl->waitpid($this->master->getPid(), $status);
        $this->log->info('<---- master shutdown. status: ' . $status);
        $this->master = null;
    }

    /**
     * @return void
     */
    public function wait()
    {
        foreach ($this->results() as $_) {}

    }

    /**
     * @return \Generator
     */
    public function results()
    {
        for (; $this->queuedCount() > $this->dequeuedCount();) {
            $result = ResultFormatter::unserialize(
                $this->resultQueue->dequeue()->getMessage()['result']
            );
            $this->dequeuedCount++;

            if ($result->isFailure()) {
                $pid = $result->getProcess()->getPid();
                $this->error[$pid] = $result;
            } else {
                yield $result;
            }
        }
    }

    /**
     * @return  bool
     */
    public function hasError()
    {
        return $this->error->exists();
    }

    /**
     * @return  \Ackintosh\Snidel\Error
     */
    public function getError()
    {
        return $this->error;
    }

    public function __destruct()
    {
        unset($this->resultQueue);
    }
}


1			<?php
2			declare(ticks = 1);
3			namespace Ackintosh\Snidel\Fork;
4
5			use Ackintosh\Snidel\ActiveWorkerSet;
6			use Ackintosh\Snidel\Config;
7			use Ackintosh\Snidel\Error;
8			use Ackintosh\Snidel\Pcntl;
9			use Ackintosh\Snidel\Result\Formatter as ResultFormatter;
10			use Ackintosh\Snidel\Task\Formatter as TaskFormatter;
11			use Ackintosh\Snidel\Worker;
12			use Bernard\Consumer;
13			use Bernard\Driver\FlatFileDriver;
14			use Bernard\Message\PlainMessage;
15			use Bernard\Producer;
16			use Bernard\QueueFactory\PersistentFactory;
17			use Bernard\Router\SimpleRouter;
18			use Bernard\Serializer;
19			use Symfony\Component\EventDispatcher\EventDispatcher;
20
21			class Container
22			{
23			/** @var Process */
24			private $master;
25
26			/** @var \Ackintosh\Snidel\Pcntl */
27			private $pcntl;
28
29			/** @var \Ackintosh\Snidel\Error */
30			private $error;
31
32			/** @var \Ackintosh\Snidel\Task\QueueInterface */
33			private $taskQueue;
			0 ignored issues – show Unused Code introduced 2017-07-29 07:27 UTC by Report Bug Copy Issue Report The property `$taskQueue` is not used and could be removed. This check marks private properties in classes that are never used. Those properties can be removed. Loading history...
34
35
36			/** @var \Ackintosh\Snidel\Log */
37			private $log;
38
39			/** @var array */
40			private $signals = [
41			SIGTERM,
42			SIGINT,
43			];
44
45			/** @var \Ackintosh\Snidel\Config */
46			private $config;
47
48			/** @var int */
49			private $receivedSignal;
50
51			private $queuedCount = 0;
52			private $dequeuedCount = 0;
53
54			private $factory;
55			private $producer;
56			private $consumer;
57			private $resultQueue;
58
59			/**
60			* @param int $ownerPid
			0 ignored issues – show Bug introduced 2017-05-19 11:44 UTC by Report Bug Copy Issue Report There is no parameter named `$ownerPid`. Was it maybe removed? This check looks for PHPDoc comments describing methods or function parameters that do not exist on the corresponding method or function. Consider the following example. The parameter `$italy` is not defined by the method `finale(...)`. /** * @param array $germany * @param array $island * @param array $italy */ function finale($germany, $island) { return "2:1"; } The most likely cause is that the parameter was removed, but the annotation was not. Loading history...
61			*/
62			public function __construct(Config $config, $log)
63			{
64			$this->log = $log;
65			$this->config = $config;
66			$this->pcntl = new Pcntl();
67			$this->error = new Error();
68
69			$driver = new FlatFileDriver('/tmp/hoge');
70			$this->factory = new PersistentFactory($driver, new Serializer());
71			$this->producer = new Producer($this->factory, new EventDispatcher());
72
73			$router = new SimpleRouter();
74			$router->add('Result', $this);
75			$this->consumer = new Consumer($router, new EventDispatcher());
76			}
77
78			/**
79			* @param \Ackintosh\Snidel\Task
80			* @return void
81			* @throws \RuntimeException
82			*/
83			public function enqueue($task)
84			{
85			try {
86			$this->producer->produce(
87			new PlainMessage(
88			'Task',
89			[
90			'task' => TaskFormatter::serialize($task),
91			]
92			)
93			);
94			$this->queuedCount++;
95
96			} catch (\RuntimeException $e) {
97			throw $e;
98			}
99			}
100
101			/**
102			* @return int
103			*/
104			public function queuedCount()
105			{
106			return $this->queuedCount;
107			}
108
109			/**
110			* @return int
111			*/
112			public function dequeuedCount()
113			{
114			return $this->dequeuedCount;
115			}
116
117			/**
118			* fork master process
119			*
120			* @return Process $master
121			* @throws \RuntimeException
122			*/
123			public function forkMaster()
124			{
125			try {
126			$this->master = $this->pcntl->fork();
127			} catch (\RuntimeException $e) {
128			$message = 'failed to fork master: ' . $e->getMessage();
129			$this->log->error($message);
130			throw new \RuntimeException($message);
131			}
132
133			$this->log->setMasterPid($this->master->getPid());
134
135			if (getmypid() === $this->config->get('ownerPid')) {
136			// owner
137			$this->log->info('pid: ' . getmypid());
138			$this->resultQueue = $this->factory->create('result');
139
140			return $this->master;
141			} else {
142			// master
143			$activeWorkerSet = new ActiveWorkerSet();
144			$this->log->info('pid: ' . $this->master->getPid());
145
146			foreach ($this->signals as $sig) {
147			$this->pcntl->signal($sig, function ($sig) use ($activeWorkerSet) {
148			$this->receivedSignal = $sig;
149			$this->log->info('received signal: ' . $sig);
150
151			if ($activeWorkerSet->count() === 0) {
152			$this->log->info('no worker is active.');
153			} else {
154			$this->log->info('------> sending signal to workers. signal: ' . $sig);
155			$activeWorkerSet->terminate($sig);
156			$this->log->info('<------ sent signal');
157			}
158			exit;
159			});
160			}
161
162			$concurrency = (int)$this->config->get('concurrency');
163			for ($i = 0; $i < $concurrency; $i++) {
164			$activeWorkerSet->add($this->forkWorker());
165			}
166			$status = null;
167			while (($workerPid = $this->pcntl->waitpid(-1, $status, WNOHANG)) !== -1) {
168			if ($workerPid === true \|\| $workerPid === 0) {
169			usleep(100000);
170			continue;
171			}
172			$activeWorkerSet->delete($workerPid);
173			$activeWorkerSet->add($this->forkWorker());
174			$status = null;
175			}
176			exit;
177			}
178			}
179
180			/**
181			* fork worker process
182			*
183			* @return \Ackintosh\Snidel\Worker
184			* @throws \RuntimeException
185			*/
186			private function forkWorker()
187			{
188			try {
189			$process = $this->pcntl->fork();
190			} catch (\RuntimeException $e) {
191			$message = 'failed to fork worker: ' . $e->getMessage();
192			$this->log->error($message);
193			throw new \RuntimeException($message);
194			}
195
196			$worker = new Worker($process);
197
198			if (getmypid() === $this->master->getPid()) {
199			// master
200			$this->log->info('forked worker. pid: ' . $worker->getPid());
201			return $worker;
202			} else {
203			// worker
204			// @codeCoverageIgnoreStart
205			$this->log->info('has forked. pid: ' . getmypid());
206
207			foreach ($this->signals as $sig) {
208			$this->pcntl->signal($sig, function ($sig) {
209			$this->receivedSignal = $sig;
210			exit;
211			}, false);
212			}
213
214			register_shutdown_function(function () use ($worker) {
215			if (!$worker->hasTask()) {
216			return;
217			}
218
219			if (!$worker->done() && $this->receivedSignal === null) {
220			$worker->error();
221			}
222			});
223
224			$this->log->info('----> started the function.');
225			try {
226			$worker->run();
227			} catch (\RuntimeException $e) {
228			$this->log->error($e->getMessage());
229			exit;
230			}
231			$this->log->info('<---- completed the function.');
232
233			$this->log->info('queued the result and exit.');
234			exit;
235			// @codeCoverageIgnoreEnd
236			}
237			}
238
239			/**
240			* @return bool
241			*/
242			public function existsMaster()
243			{
244			return $this->master !== null;
245			}
246
247			/**
248			* send signal to master process
249			*
250			* @return void
251			*/
252			public function sendSignalToMaster($sig = SIGTERM)
253			{
254			$this->log->info('----> sending signal to master. signal: ' . $sig);
255			posix_kill($this->master->getPid(), $sig);
256			$this->log->info('<---- sent signal.');
257
258			$this->log->info('----> waiting for master shutdown.');
259			$status = null;
260			$this->pcntl->waitpid($this->master->getPid(), $status);
261			$this->log->info('<---- master shutdown. status: ' . $status);
262			$this->master = null;
263			}
264
265			/**
266			* @return void
267			*/
268			public function wait()
269			{
270			foreach ($this->results() as $_) {}
			0 ignored issues – show Unused Code introduced 2017-07-29 07:27 UTC by Report Bug Copy Issue Report This `foreach` statement is empty and can be removed. This check looks for `foreach` loops that have no statements or where all statements have been commented out. This may be the result of changes for debugging or the code may simply be obsolete. Consider removing the loop. Loading history...
271			}
272
273			/**
274			* @return \Generator
275			*/
276			public function results()
277			{
278			for (; $this->queuedCount() > $this->dequeuedCount();) {
279			$result = ResultFormatter::unserialize(
280			$this->resultQueue->dequeue()->getMessage()['result']
281			);
282			$this->dequeuedCount++;
283
284			if ($result->isFailure()) {
285			$pid = $result->getProcess()->getPid();
286			$this->error[$pid] = $result;
287			} else {
288			yield $result;
289			}
290			}
291			}
292
293			/**
294			* @return bool
295			*/
296			public function hasError()
297			{
298			return $this->error->exists();
299			}
300
301			/**
302			* @return \Ackintosh\Snidel\Error
303			*/
304			public function getError()
305			{
306			return $this->error;
307			}
308
309			public function __destruct()
310			{
311			unset($this->resultQueue);
312			}
313			}
314

ackintosh / snidel

Pull Request — master (#19)

Container::results() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like