Worker::run() - Code Metrics - Inspection of "fix high cpu usage when gearman queue no job" - dominikmank/gearman - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#9)

unknown

created 2018-04-06 02:38 UTC

Worker::run() C

↳ Parent: Worker

Complexity

Conditions	8
Paths	117

Size

Total Lines	60
Code Lines	31

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	39.042

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
dl	0
loc	60
ccs	9
cts	42
cp	0.2143
rs	6.3734
c	1
b	0
f	1
cc	8
eloc	31
nc	117
nop	0
crap	39.042

How to fix Long Method

<?php

namespace dmank\gearman;

use dmank\gearman\event\ConnectToServerEvent;
use dmank\gearman\event\FunctionEvent;
use dmank\gearman\event\FunctionFailureEvent;
use dmank\gearman\event\RegisterFunctionEvent;
use dmank\gearman\event\WorkerEvent;
use dmank\gearman\event\WorkerExceptionEvent;
use dmank\gearman\exception\NoFunctionGiven;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

class Worker
{
    /**
     * @var \GearmanWorker
     */
    private $realWorker;
    /**
     * @var bool
     */
    private $connectedToServer = false;
    /**
     * @var bool
     */
    private $registeredFunctions = false;
    /**
     * @var ServerCollection
     */
    private $serverCollection;

    /**
     * @var JobCollection
     */
    private $jobs = [];

    /**
     * @var EventDispatcherInterface
     */
    private $dispatcher;

    /**
     * @var bool
     */
    private $killRequested = false;

    /**
     * @var int
     */
    private $runInterval = 0;

    public function __construct(
        ServerCollection $servers,
        JobCollection $jobs,
        EventDispatcherInterface $dispatcher,
        $runInterval = 50000
    ) {
        $this->serverCollection = $servers;
        $this->jobs = $jobs;
        $this->dispatcher = $dispatcher;
        $this->runInterval = $runInterval;
    }

    /**
     * @param \GearmanWorker $worker
     */
    public function setImplementation(\GearmanWorker $worker)
    {
        $this->realWorker = $worker;
    }

    public function run()
    {
        $this->getEventDispatcher()->dispatch(
            WorkerEvent::EVENT_BEFORE_RUN,
            new WorkerEvent($this)
        );

        try {

            while ($this->workerIsPending()) {

                $gearmanReturnCode = $this->getWorker()->returnCode();

                if ($gearmanReturnCode == GEARMAN_IO_WAIT) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_IO_WAIT,
                        new WorkerEvent($this)
                    );

                    @$this->getWorker()->wait();
// For example instead of
@mkdir($dir);

// Better use
if (@mkdir($dir) === false) {
    throw new \RuntimeException('The directory '.$dir.' could not be created.');
}
                }

                if ($gearmanReturnCode == GEARMAN_NO_JOBS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_NO_JOBS,
                        new WorkerEvent($this)
                    );

                }

                if ($gearmanReturnCode != GEARMAN_SUCCESS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_WORK,
                        new WorkerEvent($this)
                    );
                }

                if ($gearmanReturnCode == GEARMAN_SUCCESS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_AFTER_RUN,
                        new WorkerEvent($this)
                    );
                }

                usleep($this->runInterval);
            }
        } catch (NoFunctionGiven $e) {
            throw $e;
        } catch (\Exception $e) {
            $this->getEventDispatcher()->dispatch(
                WorkerExceptionEvent::EVENT_ON_FAILURE,
                new WorkerExceptionEvent($this, $e)
            );

        }
    }

    /**
     * @return EventDispatcherInterface
     */
    protected function getEventDispatcher()
    {
        return $this->dispatcher;
    }

    /**
     * @return \GearmanWorker
     */
    private function getWorker()
    {
        if (!$this->realWorker) {
            $this->realWorker = new \GearmanWorker();
            $this->realWorker->setOptions(GEARMAN_WORKER_NON_BLOCKING);
        }

        if (!$this->connectedToServer) {

            $this->getEventDispatcher()->dispatch(
                ConnectToServerEvent::CONNECT_TO_SERVER_EVENT,
                new ConnectToServerEvent($this->serverCollection)
            );

            /* @var \dmank\gearman\Server $server */
            foreach ($this->serverCollection->getServers() as $server) {
                $this->realWorker->addServer($server->getHost(), $server->getPort());
            }

            $this->getEventDispatcher()->dispatch(
                ConnectToServerEvent::CONNECTED_TO_SERVER_EVENT,
                new ConnectToServerEvent($this->serverCollection)
            );

            $this->connectedToServer = true;
        }

        if (!$this->registeredFunctions) {
            $this->registerFunctions();

            $this->registeredFunctions = true;
        }

        return $this->realWorker;
    }

    private function registerFunctions()
    {
        $this->getEventDispatcher()->dispatch(
            RegisterFunctionEvent::EVENT_ON_BEFORE_REGISTER_FUNCTIONS,
            new RegisterFunctionEvent($this->jobs)
        );

        if (count($this->jobs) == 0) {

            throw new NoFunctionGiven(
                sprintf('Didnt have jobs to register. So we need to stop here. My bad!')
            );
        }

        /* @var JobHandlerInterface $jobClass */
        foreach ($this->jobs->getJobs() as $jobName => $jobClass) {
            $this->realWorker->addFunction(
                $jobName,
                function (\GearmanJob $gearmanJob) use ($jobClass, $jobName) {

                    $job = new Job($jobName, $gearmanJob->workload());

                    try {

                        $this->getEventDispatcher()->dispatch(
                            FunctionEvent::FUNCTION_BEFORE_EXECUTE,
                            new FunctionEvent($jobClass, $job)
                        );

                        $result = $jobClass->execute($job);

                        $this->getEventDispatcher()->dispatch(
                            FunctionEvent::FUNCTION_AFTER_EXECUTE,
                            new FunctionEvent($jobClass, $job, $result)
                        );
                    } catch (\Exception $e) {

                        $this->getEventDispatcher()->dispatch(
                            FunctionFailureEvent::FUNCTION_ON_FAILURE,
                            new FunctionFailureEvent($jobClass, $e, $job)
                        );

                        $result = $e;
                    }

                    if (is_object($result) || is_array($result)) {
                        $result = serialize($result);
                    }

                    $gearmanJob->sendComplete($result);
                }
            );
        }

        $this->getEventDispatcher()->dispatch(
            RegisterFunctionEvent::EVENT_ON_AFTER_REGISTER_FUNCTIONS,
            new RegisterFunctionEvent($this->jobs)
        );

    }

    /**
     * @return bool
     */
    private function workerIsPending()
    {
        return (
            $this->getWorker()->work() ||
            $this->getWorker()->returnCode() == GEARMAN_IO_WAIT ||
            $this->getWorker()->returnCode() == GEARMAN_NO_JOBS
        ) && !$this->killRequested;
    }

    public function destroyWorker()
    {
        $this->getEventDispatcher()->dispatch(
            WorkerEvent::EVENT_ON_BEFORE_DESTROY,
            new WorkerEvent($this)
        );

        $this->killRequested = true;
    }
}


1		<?php
2
3		namespace dmank\gearman;
4
5		use dmank\gearman\event\ConnectToServerEvent;
6		use dmank\gearman\event\FunctionEvent;
7		use dmank\gearman\event\FunctionFailureEvent;
8		use dmank\gearman\event\RegisterFunctionEvent;
9		use dmank\gearman\event\WorkerEvent;
10		use dmank\gearman\event\WorkerExceptionEvent;
11		use dmank\gearman\exception\NoFunctionGiven;
12		use Symfony\Component\EventDispatcher\EventDispatcherInterface;
13
14		class Worker
15		{
16		/**
17		* @var \GearmanWorker
18		*/
19		private $realWorker;
20		/**
21		* @var bool
22		*/
23		private $connectedToServer = false;
24		/**
25		* @var bool
26		*/
27		private $registeredFunctions = false;
28		/**
29		* @var ServerCollection
30		*/
31		private $serverCollection;
32
33		/**
34		* @var JobCollection
35		*/
36		private $jobs = [];
37
38		/**
39		* @var EventDispatcherInterface
40		*/
41		private $dispatcher;
42
43		/**
44		* @var bool
45		*/
46		private $killRequested = false;
47
48		/**
49		* @var int
50		*/
51		private $runInterval = 0;
52
53	3	public function __construct(
54		ServerCollection $servers,
55		JobCollection $jobs,
56		EventDispatcherInterface $dispatcher,
57		$runInterval = 50000
58		) {
59	3	$this->serverCollection = $servers;
60	3	$this->jobs = $jobs;
61	3	$this->dispatcher = $dispatcher;
62	3	$this->runInterval = $runInterval;
63	3	}
64
65		/**
66		* @param \GearmanWorker $worker
67		*/
68	3	public function setImplementation(\GearmanWorker $worker)
69		{
70	3	$this->realWorker = $worker;
71	3	}
72
73	3	public function run()
74		{
75	3	$this->getEventDispatcher()->dispatch(
76	3	WorkerEvent::EVENT_BEFORE_RUN,
77	3	new WorkerEvent($this)
78	3	);
79
80		try {
81
82	3	while ($this->workerIsPending()) {
83
84		$gearmanReturnCode = $this->getWorker()->returnCode();
85
86		if ($gearmanReturnCode == GEARMAN_IO_WAIT) {
87
88		$this->getEventDispatcher()->dispatch(
89		WorkerEvent::EVENT_ON_IO_WAIT,
90		new WorkerEvent($this)
91		);
92
93		@$this->getWorker()->wait();
		0 ignored issues – show Security Best Practice introduced 2017-01-16 08:13 UTC by Report Bug Copy Issue Report It seems like you do not handle an error condition here. This can introduce security issues, and is generally not recommended. If you suppress an error, we recommend checking for the error condition explicitly: // For example instead of @mkdir($dir); // Better use if (@mkdir($dir) === false) { throw new \RuntimeException('The directory '.$dir.' could not be created.'); } Loading history...
94		}
95
96		if ($gearmanReturnCode == GEARMAN_NO_JOBS) {
97
98		$this->getEventDispatcher()->dispatch(
99		WorkerEvent::EVENT_ON_NO_JOBS,
100		new WorkerEvent($this)
101		);
102
103		}
104
105		if ($gearmanReturnCode != GEARMAN_SUCCESS) {
106
107		$this->getEventDispatcher()->dispatch(
108		WorkerEvent::EVENT_ON_WORK,
109		new WorkerEvent($this)
110		);
111		}
112
113		if ($gearmanReturnCode == GEARMAN_SUCCESS) {
114
115		$this->getEventDispatcher()->dispatch(
116		WorkerEvent::EVENT_AFTER_RUN,
117		new WorkerEvent($this)
118		);
119		}
120
121		usleep($this->runInterval);
122		}
123	3	} catch (NoFunctionGiven $e) {
124	1	throw $e;
125		} catch (\Exception $e) {
126		$this->getEventDispatcher()->dispatch(
127		WorkerExceptionEvent::EVENT_ON_FAILURE,
128		new WorkerExceptionEvent($this, $e)
129		);
130
131		}
132	2	}
133
134		/**
135		* @return EventDispatcherInterface
136		*/
137	3	protected function getEventDispatcher()
138		{
139	3	return $this->dispatcher;
140		}
141
142		/**
143		* @return \GearmanWorker
144		*/
145	3	private function getWorker()
146		{
147	3	if (!$this->realWorker) {
148		$this->realWorker = new \GearmanWorker();
149		$this->realWorker->setOptions(GEARMAN_WORKER_NON_BLOCKING);
150		}
151
152	3	if (!$this->connectedToServer) {
153
154	3	$this->getEventDispatcher()->dispatch(
155	3	ConnectToServerEvent::CONNECT_TO_SERVER_EVENT,
156	3	new ConnectToServerEvent($this->serverCollection)
157	3	);
158
159		/* @var \dmank\gearman\Server $server */
160	3	foreach ($this->serverCollection->getServers() as $server) {
161	3	$this->realWorker->addServer($server->getHost(), $server->getPort());
162	3	}
163
164	3	$this->getEventDispatcher()->dispatch(
165	3	ConnectToServerEvent::CONNECTED_TO_SERVER_EVENT,
166	3	new ConnectToServerEvent($this->serverCollection)
167	3	);
168
169	3	$this->connectedToServer = true;
170	3	}
171
172	3	if (!$this->registeredFunctions) {
173	3	$this->registerFunctions();
174
175	2	$this->registeredFunctions = true;
176	2	}
177
178	2	return $this->realWorker;
179		}
180
181	3	private function registerFunctions()
182		{
183	3	$this->getEventDispatcher()->dispatch(
184	3	RegisterFunctionEvent::EVENT_ON_BEFORE_REGISTER_FUNCTIONS,
185	3	new RegisterFunctionEvent($this->jobs)
186	3	);
187
188	3	if (count($this->jobs) == 0) {
189
190	1	throw new NoFunctionGiven(
191	1	sprintf('Didnt have jobs to register. So we need to stop here. My bad!')
192	1	);
193		}
194
195		/* @var JobHandlerInterface $jobClass */
196	2	foreach ($this->jobs->getJobs() as $jobName => $jobClass) {
197	2	$this->realWorker->addFunction(
198	2	$jobName,
199		function (\GearmanJob $gearmanJob) use ($jobClass, $jobName) {
200
201		$job = new Job($jobName, $gearmanJob->workload());
202
203		try {
204
205		$this->getEventDispatcher()->dispatch(
206		FunctionEvent::FUNCTION_BEFORE_EXECUTE,
207		new FunctionEvent($jobClass, $job)
208		);
209
210		$result = $jobClass->execute($job);
211
212		$this->getEventDispatcher()->dispatch(
213		FunctionEvent::FUNCTION_AFTER_EXECUTE,
214		new FunctionEvent($jobClass, $job, $result)
215		);
216		} catch (\Exception $e) {
217
218		$this->getEventDispatcher()->dispatch(
219		FunctionFailureEvent::FUNCTION_ON_FAILURE,
220		new FunctionFailureEvent($jobClass, $e, $job)
221		);
222
223		$result = $e;
224		}
225
226		if (is_object($result) \|\| is_array($result)) {
227		$result = serialize($result);
228		}
229
230		$gearmanJob->sendComplete($result);
231		}
232	2	);
233	2	}
234
235	2	$this->getEventDispatcher()->dispatch(
236	2	RegisterFunctionEvent::EVENT_ON_AFTER_REGISTER_FUNCTIONS,
237	2	new RegisterFunctionEvent($this->jobs)
238	2	);
239
240	2	}
241
242		/**
243		* @return bool
244		*/
245	3	private function workerIsPending()
246		{
247		return (
248	3	$this->getWorker()->work() \|\|
249	2	$this->getWorker()->returnCode() == GEARMAN_IO_WAIT \|\|
250	2	$this->getWorker()->returnCode() == GEARMAN_NO_JOBS
251	2	) && !$this->killRequested;
252		}
253
254		public function destroyWorker()
255		{
256		$this->getEventDispatcher()->dispatch(
257		WorkerEvent::EVENT_ON_BEFORE_DESTROY,
258		new WorkerEvent($this)
259		);
260
261		$this->killRequested = true;
262		}
263		}
264

dominikmank / gearman

GitHub Access Token became invalid

Pull Request — master (#9)

Worker::run() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like