Issues in Threads.php - Fixed Issues - Inspection of "CamelCase namespaces + bug fixes" - lugnsk/micro - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 579af5...b29473 )

by Oleg

created 2016-02-08 20:56 UTC

micro/cli/Threads.php (2 issues)

Labels

Severity

Major 2

Upgrade to new PHP Analysis Engine

These results are based on our legacy PHP analysis, consider migrating to our new PHP analysis engine instead. Learn more

<?php /** MicroThreads */

namespace Micro\cli;

use Micro\base\Exception;

/**
 * Threads class file.
 *
 * @author Oleg Lunegov <[email protected]>
 * @link https://github.com/lugnsk/micro
 * @copyright Copyright &copy; 2013 Oleg Lunegov
 * @license /LICENSE
 * @package micro
 * @subpackage cli
 * @version 1.0
 * @since 1.0
 */
abstract class Threads
{
    /** @var string $name thread name */
    private $name;
    /** @var integer $pid process ID */
    private $pid;
    /** @var integer $puid process UID */
    private $puid;
    /** @var integer $guid process GUID */
    private $guid;
    /** @var bool $isChild is child */
    private $isChild = false;
    /** @var array $internalIPCArray Internal IPC array */
    private $internalIPCArray = [];
    /** @var integer $internalIPCKey Internal IPC key */
    private $internalIPCKey;
    /** @var integer $internalSemaphoreKey Internal semaphore key */
    private $internalSemaphoreKey;
    /** @var bool $isIPC is IPC */
    private $isIPC = false;
    /** @var bool $running is running */
    private $running = false;
    /** @var string $fileIPC1 file IPC1 */
    private $fileIPC1;
    /** @var string $fileIPC2 file IPC2 */
    private $fileIPC2;


    /**
     * Constructor thread
     *
     * @access public
     *
     * @param     $name
     * @param int $puid
     * @param int $guid
     * @param int $umask
     *
     * @result void
     * @throws Exception
     */
    public function __construct($name, $puid = 0, $guid = 0, $umask = -1)
    {
        if (empty($_SERVER['argc'])) {
            throw new Exception('Threads are permitted only for CLI');
        }
        $this->name = $name;
        $this->guid = $guid;
        $this->puid = $puid;

        if ($umask !== -1) {
            umask($umask);
        }

        $this->isChild = false;
        $this->internalIPCArray = [];
        $this->isIPC = false;

        if ($this->createIPCSegment() && $this->createIPCSemaphore()) {
            $this->isIPC = true;
        }
    }

    /**
     * Create IPC segment
     *
     * @access protected
     * @return bool
     * @throws Exception
     */
    protected function createIPCSegment()
    {
        $this->fileIPC1 = '/tmp/' . mt_rand() . md5($this->getName()) . '.shm';

        touch($this->fileIPC1);

        $shm_key = ftok($this->fileIPC1, 't');
        if ($shm_key === -1) {
            throw new Exception('Fatal exception creating SHM segment (ftok)');
        }

        $this->internalIPCKey = @shmop_open($shm_key, 'c', 0644, 10240);
        if (!$this->internalIPCKey) {
            return false;
        }

        return true;
    }

    /**
     * get thread name
     *
     * @access public
     * @return string
     */
    public function getName()
    {
        return $this->name;
    }

    /**
     * Set thread name
     *
     * @access public
     *
     * @param string $name
     *
     * @return void
     */
    public function setName($name)
    {
        $this->name = $name;
    }

    /**
     * Create IPC semaphore
     *
     * @access protected
     * @return bool
     * @throws Exception
     */
    protected function createIPCSemaphore()
    {
        $this->fileIPC2 = '/tmp/' . mt_rand() . md5($this->getName()) . '.sem';

        touch($this->fileIPC2);

        $sem_key = ftok($this->fileIPC2, 't');
        if ($sem_key === -1) {
            throw new Exception('Fatal exception creating semaphore (ftok)');
        }

        $this->internalSemaphoreKey = shmop_open($sem_key, 'c', 0644, 10);
        if (!$this->internalSemaphoreKey) {
            return false;
        }

        return true;
    }

    /**
     * Is running thread
     *
     * @access public
     * @return bool
     */
    public function isRunning()
    {
        return (bool)$this->running;
    }

    /**
     * Set alive
     *
     * @access public
     * @return void
     * @throws \Micro\base\Exception
     */
    public function setAlive()
    {
        $this->setVariable('_pingTime', time());
    }

    /**
     * Set variable in shared memory
     *
     * @access public
     *
     * @param string $name
     * @param mixed $value
     *
     * @return void
     * @throws \Micro\base\Exception
     */
    public function setVariable($name, $value)
    {
        $this->internalIPCArray[$name] = $value;
        $this->writeToIPCSegment();
    }

    /**
     * Write to IPC segment
     *
     * @access protected
     * @return void
     * @throws Exception
     */
    protected function writeToIPCSegment()
    {
        if (shmop_read($this->internalSemaphoreKey, 1, 1) === 1) {

            return;
        }

        $serialized_IPC_array = serialize($this->internalIPCArray);
        $shm_bytes_written = shmop_write($this->internalIPCKey, $serialized_IPC_array, 0);

        if ($shm_bytes_written !== strlen($serialized_IPC_array)) {
            throw new Exception(
                'Fatal exception writing SHM segment (shmop_write)' . strlen($serialized_IPC_array) .
                '-' . shmop_size($this->internalIPCKey)
            );
        }
    }

    /**
     * Get last alive
     *
     * @access public
     * @return int
     * @throws \Micro\base\Exception
     */
    public function getLastAlive()
    {
        $timestamp = (int)$this->getVariable('_pingTime');
        if ($timestamp === 0) {
            return 0;
        } else {
            return (time() - $timestamp);
        }
    }

    /**
     * Get variable from shared memory
     *
     * @access public
     *
     * @param string $name
     *
     * @return mixed
     * @throws \Micro\base\Exception
     */
    public function getVariable($name)
    {
        $this->readFromIPCSegment();

        return $this->internalIPCArray[$name];
    }

    /**
     * Read from IPC segment
     *
     * @access public
     * @return void
     * @throws Exception
     */
    protected function readFromIPCSegment()
    {
        $serialized_IPC_array = shmop_read($this->internalIPCKey, 0, shmop_size($this->internalIPCKey));

        if (!$serialized_IPC_array) {
            throw new Exception('Fatal exception reading SHM segment (shmop_read)' . "\n");
        }

        unset($this->internalIPCArray);

        $this->internalIPCArray = @unserialize($serialized_IPC_array);
    }

    /**
     * Get thread process ID
     *
     * @access public
     * @return int
     */
    public function getPid()
    {
        return $this->pid;
    }

    /**
     * Register callback func into shared memory
     *
     * @access public
     *
     * @param mixed $argList
     * @param string $methodName
     *
     * @return mixed|void
     * @throws \Micro\base\Exception
     */
    public function register_callback_func($argList, $methodName)
    {
        if (is_array($argList) && count($argList) > 1) {
            if ($argList[1] === -2) {
                $this->internalIPCArray['_call_type'] = -2;
            } else {
                $this->internalIPCArray['_call_type'] = -1;
            }
        } else {
            $this->internalIPCArray['_call_type'] = -1;
        }

        $this->internalIPCArray['_call_method'] = $methodName;
        $this->internalIPCArray['_call_input'] = $argList;

        $this->writeToIPCSegment();

        switch ($this->internalIPCArray['_call_type']) {
            case -1:
                $this->sendSigUsr1();
                break;

            case -2:
                shmop_write($this->internalSemaphoreKey, 1, 0);

                $this->sendSigUsr1();
                $this->waitIPCSemaphore();
                $this->readFromIPCSegment();

                shmop_write($this->internalSemaphoreKey, 0, 1);

                return $this->internalIPCArray['_call_output'];
                break;
        }

        return false;
    }

    /**
     * Send signal USR1
     *
     * @access protected
     * @return void
     */
    protected function sendSigUsr1()
    {
        if ($this->pid > 0) {
            posix_kill($this->pid, SIGUSR1);
        }
    }

    /**
     * Wait IPC semaphore
     *
     * @access protected
     * @return void
     */
    protected function waitIPCSemaphore()
    {
        while (true) {
            $ok = shmop_read($this->internalSemaphoreKey, 0, 1);

            if ($ok === 0) {

                break;
            } else {
                usleep(10);
            }
        }
    }

    /**
     * Start
     *
     * @access public
     * @return void
     * @throws Exception
     */
    public function start()
    {
        if (!$this->isIPC) {
            throw new Exception('Fatal error, unable to create SHM segments for process communications');
        }

        pcntl_signal(SIGCHLD, SIG_IGN);

        $pid = pcntl_fork();
        if ($pid === 0) {
            $this->isChild = true;
            sleep(1);

            pcntl_signal(SIGUSR1, [$this, 'sigHandler']);

            if ($this->guid !== 0) {
                posix_setgid($this->guid);
            }
            if ($this->puid !== 0) {
                posix_setuid($this->puid);
            }
            $this->run();

            exit(0);
        } else {
            $this->isChild = false;
            $this->running = true;
            $this->pid = $pid;
        }
    }

    /**
     * Running thread
     *
     * @access public
     * @return void
     */
    abstract public function run();

    /**
     * Stop thread
     *
     * @access public
     * @return bool
     */
    public function stop()
    {
        $success = false;

        if ($this->pid > 0) {
            posix_kill($this->pid, 9);
            pcntl_waitpid($this->pid, $temp = 0, WNOHANG);

            $success = pcntl_wifexited($temp);

            $this->cleanThreadContext();
        }

        return $success;
    }

    /**
     * Clean thread context
     *
     * @access protected
     * @return void
     */
    protected function cleanThreadContext()
    {
        @shmop_delete($this->internalIPCKey);
        @shmop_delete($this->internalSemaphoreKey);

        @shmop_close($this->internalIPCKey);
        @shmop_close($this->internalSemaphoreKey);

        unlink($this->fileIPC1);
        unlink($this->fileIPC2);

        $this->running = false;
        unset($this->pid);
    }

    /**
     * Signal handler
     *
     * @access protected
     *
     * @param $sigNo
     *
     * @return void
     * @throws \Micro\base\Exception
     */
    protected function sigHandler($sigNo)
    {
        switch ($sigNo) {
            case SIGTERM:
                exit;
                break;

            case SIGHUP:
                break;

            case SIGUSR1:
                $this->readFromIPCSegment();

                $method = $this->internalIPCArray['_call_method'];
                $params = $this->internalIPCArray['_call_input'];

                switch ($this->internalIPCArray['_call_type']) {
                    case -1:
                        $this->$method($params);
                        break;

                    case -2:
                        $this->internalIPCArray['_call_output'] = $this->$method($params);

                        $this->writeToIPCSegment();

                        shmop_write($this->internalSemaphoreKey, 0, 0);
                        shmop_write($this->internalSemaphoreKey, 1, 1);

                        break;
                }
                break;

            default:
                // handle all other signals
        }
    }
}


1		<?php /** MicroThreads */
2
3		namespace Micro\cli;
4
5		use Micro\base\Exception;
6
7		/**
8		* Threads class file.
9		*
10		* @author Oleg Lunegov <[email protected]>
11		* @link https://github.com/lugnsk/micro
12		* @copyright Copyright © 2013 Oleg Lunegov
13		* @license /LICENSE
14		* @package micro
15		* @subpackage cli
16		* @version 1.0
17		* @since 1.0
18		*/
19		abstract class Threads
20		{
21		/** @var string $name thread name */
22		private $name;
23		/** @var integer $pid process ID */
24		private $pid;
25		/** @var integer $puid process UID */
26		private $puid;
27		/** @var integer $guid process GUID */
28		private $guid;
29		/** @var bool $isChild is child */
30		private $isChild = false;
31		/** @var array $internalIPCArray Internal IPC array */
32		private $internalIPCArray = [];
33		/** @var integer $internalIPCKey Internal IPC key */
34		private $internalIPCKey;
35		/** @var integer $internalSemaphoreKey Internal semaphore key */
36		private $internalSemaphoreKey;
37		/** @var bool $isIPC is IPC */
38		private $isIPC = false;
39		/** @var bool $running is running */
40		private $running = false;
41		/** @var string $fileIPC1 file IPC1 */
42		private $fileIPC1;
43		/** @var string $fileIPC2 file IPC2 */
44		private $fileIPC2;
45
46
47		/**
48		* Constructor thread
49		*
50		* @access public
51		*
52		* @param $name
53		* @param int $puid
54		* @param int $guid
55		* @param int $umask
56		*
57		* @result void
58		* @throws Exception
59		*/
60		public function __construct($name, $puid = 0, $guid = 0, $umask = -1)
61		{
62		if (empty($_SERVER['argc'])) {
63		throw new Exception('Threads are permitted only for CLI');
64		}
65		$this->name = $name;
66		$this->guid = $guid;
67		$this->puid = $puid;
68
69		if ($umask !== -1) {
70		umask($umask);
71		}
72
73		$this->isChild = false;
74		$this->internalIPCArray = [];
75		$this->isIPC = false;
76
77		if ($this->createIPCSegment() && $this->createIPCSemaphore()) {
78		$this->isIPC = true;
79		}
80		}
81
82		/**
83		* Create IPC segment
84		*
85		* @access protected
86		* @return bool
87		* @throws Exception
88		*/
89	View Code Duplication	protected function createIPCSegment()
90		{
91		$this->fileIPC1 = '/tmp/' . mt_rand() . md5($this->getName()) . '.shm';
92
93		touch($this->fileIPC1);
94
95		$shm_key = ftok($this->fileIPC1, 't');
96		if ($shm_key === -1) {
97		throw new Exception('Fatal exception creating SHM segment (ftok)');
98		}
99
100		$this->internalIPCKey = @shmop_open($shm_key, 'c', 0644, 10240);
101		if (!$this->internalIPCKey) {
102		return false;
103		}
104
105		return true;
106		}
107
108		/**
109		* get thread name
110		*
111		* @access public
112		* @return string
113		*/
114		public function getName()
115		{
116		return $this->name;
117		}
118
119		/**
120		* Set thread name
121		*
122		* @access public
123		*
124		* @param string $name
125		*
126		* @return void
127		*/
128		public function setName($name)
129		{
130		$this->name = $name;
131		}
132
133		/**
134		* Create IPC semaphore
135		*
136		* @access protected
137		* @return bool
138		* @throws Exception
139		*/
140	View Code Duplication	protected function createIPCSemaphore()
141		{
142		$this->fileIPC2 = '/tmp/' . mt_rand() . md5($this->getName()) . '.sem';
143
144		touch($this->fileIPC2);
145
146		$sem_key = ftok($this->fileIPC2, 't');
147		if ($sem_key === -1) {
148		throw new Exception('Fatal exception creating semaphore (ftok)');
149		}
150
151		$this->internalSemaphoreKey = shmop_open($sem_key, 'c', 0644, 10);
152		if (!$this->internalSemaphoreKey) {
153		return false;
154		}
155
156		return true;
157		}
158
159		/**
160		* Is running thread
161		*
162		* @access public
163		* @return bool
164		*/
165		public function isRunning()
166		{
167		return (bool)$this->running;
168		}
169
170		/**
171		* Set alive
172		*
173		* @access public
174		* @return void
175		* @throws \Micro\base\Exception
176		*/
177		public function setAlive()
178		{
179		$this->setVariable('_pingTime', time());
180		}
181
182		/**
183		* Set variable in shared memory
184		*
185		* @access public
186		*
187		* @param string $name
188		* @param mixed $value
189		*
190		* @return void
191		* @throws \Micro\base\Exception
192		*/
193		public function setVariable($name, $value)
194		{
195		$this->internalIPCArray[$name] = $value;
196		$this->writeToIPCSegment();
197		}
198
199		/**
200		* Write to IPC segment
201		*
202		* @access protected
203		* @return void
204		* @throws Exception
205		*/
206		protected function writeToIPCSegment()
207		{
208		if (shmop_read($this->internalSemaphoreKey, 1, 1) === 1) {
		0 ignored issues – show Unused Code Bug introduced 2015-11-21 13:17 UTC by Report Bug Copy Issue Report Show Similar Issues like this The strict comparison `===` seems to always evaluate to `false` as the types of `shmop_read($this->internalSemaphoreKey, 1, 1)` (`string`) and `1` (`integer`) can never be identical. Maybe you want to use a loose comparison `==` instead? Loading history...
209		return;
210		}
211
212		$serialized_IPC_array = serialize($this->internalIPCArray);
213		$shm_bytes_written = shmop_write($this->internalIPCKey, $serialized_IPC_array, 0);
214
215		if ($shm_bytes_written !== strlen($serialized_IPC_array)) {
216		throw new Exception(
217		'Fatal exception writing SHM segment (shmop_write)' . strlen($serialized_IPC_array) .
218		'-' . shmop_size($this->internalIPCKey)
219		);
220		}
221		}
222
223		/**
224		* Get last alive
225		*
226		* @access public
227		* @return int
228		* @throws \Micro\base\Exception
229		*/
230		public function getLastAlive()
231		{
232		$timestamp = (int)$this->getVariable('_pingTime');
233		if ($timestamp === 0) {
234		return 0;
235		} else {
236		return (time() - $timestamp);
237		}
238		}
239
240		/**
241		* Get variable from shared memory
242		*
243		* @access public
244		*
245		* @param string $name
246		*
247		* @return mixed
248		* @throws \Micro\base\Exception
249		*/
250		public function getVariable($name)
251		{
252		$this->readFromIPCSegment();
253
254		return $this->internalIPCArray[$name];
255		}
256
257		/**
258		* Read from IPC segment
259		*
260		* @access public
261		* @return void
262		* @throws Exception
263		*/
264		protected function readFromIPCSegment()
265		{
266		$serialized_IPC_array = shmop_read($this->internalIPCKey, 0, shmop_size($this->internalIPCKey));
267
268		if (!$serialized_IPC_array) {
269		throw new Exception('Fatal exception reading SHM segment (shmop_read)' . "\n");
270		}
271
272		unset($this->internalIPCArray);
273
274		$this->internalIPCArray = @unserialize($serialized_IPC_array);
275		}
276
277		/**
278		* Get thread process ID
279		*
280		* @access public
281		* @return int
282		*/
283		public function getPid()
284		{
285		return $this->pid;
286		}
287
288		/**
289		* Register callback func into shared memory
290		*
291		* @access public
292		*
293		* @param mixed $argList
294		* @param string $methodName
295		*
296		* @return mixed\|void
297		* @throws \Micro\base\Exception
298		*/
299		public function register_callback_func($argList, $methodName)
300		{
301		if (is_array($argList) && count($argList) > 1) {
302		if ($argList[1] === -2) {
303		$this->internalIPCArray['_call_type'] = -2;
304		} else {
305		$this->internalIPCArray['_call_type'] = -1;
306		}
307		} else {
308		$this->internalIPCArray['_call_type'] = -1;
309		}
310
311		$this->internalIPCArray['_call_method'] = $methodName;
312		$this->internalIPCArray['_call_input'] = $argList;
313
314		$this->writeToIPCSegment();
315
316		switch ($this->internalIPCArray['_call_type']) {
317		case -1:
318		$this->sendSigUsr1();
319		break;
320
321		case -2:
322		shmop_write($this->internalSemaphoreKey, 1, 0);
323
324		$this->sendSigUsr1();
325		$this->waitIPCSemaphore();
326		$this->readFromIPCSegment();
327
328		shmop_write($this->internalSemaphoreKey, 0, 1);
329
330		return $this->internalIPCArray['_call_output'];
331		break;
332		}
333
334		return false;
335		}
336
337		/**
338		* Send signal USR1
339		*
340		* @access protected
341		* @return void
342		*/
343		protected function sendSigUsr1()
344		{
345		if ($this->pid > 0) {
346		posix_kill($this->pid, SIGUSR1);
347		}
348		}
349
350		/**
351		* Wait IPC semaphore
352		*
353		* @access protected
354		* @return void
355		*/
356		protected function waitIPCSemaphore()
357		{
358		while (true) {
359		$ok = shmop_read($this->internalSemaphoreKey, 0, 1);
360
361		if ($ok === 0) {
		0 ignored issues – show Unused Code Bug introduced 2015-11-21 13:17 UTC by Report Bug Copy Issue Report Show Similar Issues like this The strict comparison `===` seems to always evaluate to `false` as the types of `$ok` (`string`) and `0` (`integer`) can never be identical. Maybe you want to use a loose comparison `==` instead? Loading history...
362		break;
363		} else {
364		usleep(10);
365		}
366		}
367		}
368
369		/**
370		* Start
371		*
372		* @access public
373		* @return void
374		* @throws Exception
375		*/
376		public function start()
377		{
378		if (!$this->isIPC) {
379		throw new Exception('Fatal error, unable to create SHM segments for process communications');
380		}
381
382		pcntl_signal(SIGCHLD, SIG_IGN);
383
384		$pid = pcntl_fork();
385		if ($pid === 0) {
386		$this->isChild = true;
387		sleep(1);
388
389		pcntl_signal(SIGUSR1, [$this, 'sigHandler']);
390
391		if ($this->guid !== 0) {
392		posix_setgid($this->guid);
393		}
394		if ($this->puid !== 0) {
395		posix_setuid($this->puid);
396		}
397		$this->run();
398
399		exit(0);
400		} else {
401		$this->isChild = false;
402		$this->running = true;
403		$this->pid = $pid;
404		}
405		}
406
407		/**
408		* Running thread
409		*
410		* @access public
411		* @return void
412		*/
413		abstract public function run();
414
415		/**
416		* Stop thread
417		*
418		* @access public
419		* @return bool
420		*/
421		public function stop()
422		{
423		$success = false;
424
425		if ($this->pid > 0) {
426		posix_kill($this->pid, 9);
427		pcntl_waitpid($this->pid, $temp = 0, WNOHANG);
428
429		$success = pcntl_wifexited($temp);
430
431		$this->cleanThreadContext();
432		}
433
434		return $success;
435		}
436
437		/**
438		* Clean thread context
439		*
440		* @access protected
441		* @return void
442		*/
443		protected function cleanThreadContext()
444		{
445		@shmop_delete($this->internalIPCKey);
446		@shmop_delete($this->internalSemaphoreKey);
447
448		@shmop_close($this->internalIPCKey);
449		@shmop_close($this->internalSemaphoreKey);
450
451		unlink($this->fileIPC1);
452		unlink($this->fileIPC2);
453
454		$this->running = false;
455		unset($this->pid);
456		}
457
458		/**
459		* Signal handler
460		*
461		* @access protected
462		*
463		* @param $sigNo
464		*
465		* @return void
466		* @throws \Micro\base\Exception
467		*/
468		protected function sigHandler($sigNo)
469		{
470		switch ($sigNo) {
471		case SIGTERM:
472		exit;
473		break;
474
475		case SIGHUP:
476		break;
477
478		case SIGUSR1:
479		$this->readFromIPCSegment();
480
481		$method = $this->internalIPCArray['_call_method'];
482		$params = $this->internalIPCArray['_call_input'];
483
484		switch ($this->internalIPCArray['_call_type']) {
485		case -1:
486		$this->$method($params);
487		break;
488
489		case -2:
490		$this->internalIPCArray['_call_output'] = $this->$method($params);
491
492		$this->writeToIPCSegment();
493
494		shmop_write($this->internalSemaphoreKey, 0, 0);
495		shmop_write($this->internalSemaphoreKey, 1, 1);
496
497		break;
498		}
499		break;
500
501		default:
502		// handle all other signals
503		}
504		}
505		}
506

lugnsk / micro

Push — master ( 579af5...b29473 )

micro/cli/Threads.php (2 issues)

Labels

Severity

Introduced By

Upgrade to new PHP Analysis Engine

Duplication Side-by-Side

Filter issues like