Ackintosh\Snidel::waitSimply()

Complexity

Conditions	5
Paths	5

Size

Total Lines	23
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	Snidel::hasError()	0	4	1
A	Snidel::getError()	0	4	1

<?php
declare(ticks = 1);

namespace Ackintosh;

use Ackintosh\Snidel\ForkContainer;
use Ackintosh\Snidel\Result;
use Ackintosh\Snidel\Token;
use Ackintosh\Snidel\Log;
use Ackintosh\Snidel\Pcntl;
use Ackintosh\Snidel\DataRepository;
use Ackintosh\Snidel\MapContainer;
use Ackintosh\Snidel\Task;
use Ackintosh\Snidel\TaskQueue;
use Ackintosh\Snidel\ResultQueue;
use Ackintosh\Snidel\Exception\SharedMemoryControlException;

class Snidel
{
    /** @var string */
    const VERSION = '0.6.0';

    private $masterProcessId = null;

    /** @var array */
    private $childPids = array();

    /** @var \Ackintosh\Snidel\ForkContainer */
    private $forkContainer;

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

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

    /** @var \Ackintosh\Snidel\Token */
    private $token;

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

    /** @var \Ackintosh\Snidel\DataRepository */
    private $dataRepository;

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

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

    /** @var array */
    private $signals = array(
        SIGTERM,
        SIGINT,
    );

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

    /** @var \Ackintosh\Snidel\Token */
    private $processToken;

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

    public function __construct($concurrency = 5)
    {
        $this->ownerPid         = getmypid();
        $this->childPids        = array();
        $this->concurrency      = $concurrency;
        $this->token            = new Token(getmypid(), $concurrency);
        $this->log              = new Log(getmypid());
        $this->pcntl            = new Pcntl();
        $this->dataRepository   = new DataRepository();
        $this->forkContainer    = new ForkContainer($this->ownerPid);

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

                    $log->info('--> sending a signal to children.');
                    $self->sendSignalToChildren($sig);

                    $log->info('--> deleting token.');
                    unset($token);

                    $log->info('<-- signal handling has been completed successfully.');
                    exit;

The method __construct() contains an exit expression.

                },
                false
            );
        }

        $this->log->info('parent pid: ' . $this->ownerPid);
    }

    /**
     * sets the resource for the log.
     *
     * @param   resource    $resource
     * @return  void
     * @codeCoverageIgnore
     */
    public function setLoggingDestination($resource)
    {
        $this->log->setDestination($resource);
    }

    /**
     * this method uses master / worker model.
     *
     * @param   callable    $callable
     * @param   mixed       $args
     * @param   string      $tag
     * @return  void
     * @throws  \RuntimeException
     */
    public function fork($callable, $args = array(), $tag = null)
    {
        if ($this->masterProcessId === null) {
            $this->forkMaster();
        }

        try {
            $this->forkContainer->enqueue(new Task($callable, $args, $tag));
        } catch (\RuntimeException $e) {
            throw $e;
        }

        $this->log->info('queued task #' . $this->forkContainer->queuedCount());
    }

    /**
     * fork master process
     *
     * @return  void
     */
    private function forkMaster()
    {
        $pid = $this->pcntl->fork();
        $this->masterProcessId = ($pid === 0) ? getmypid() : $pid;
        $this->log->setMasterProcessId($this->masterProcessId);

        if ($pid) {
            // owner
            $this->log->info('pid: ' . getmypid());
        } elseif ($pid === -1) {

This elseif statement is empty, and could be removed.

            // error
        } else {
            // master
            $taskQueue = new TaskQueue($this->ownerPid);
            $this->log->info('pid: ' . $this->masterProcessId);

            foreach ($this->signals as $sig) {
                $this->pcntl->signal($sig, SIG_DFL, true);
            }

            while ($task = $taskQueue->dequeue()) {
                $this->log->info('dequeued task #' . $taskQueue->dequeuedCount());
                if ($this->token->accept()) {
                    $this->forkWorker($task);
                }
            }
            $this->_exit();
        }
    }

    /**
     * fork worker process
     *
     * @param   \Ackintosh\Snidel\Task
     * @return  void
     * @throws  \RuntimeException
     */
    private function forkWorker($task)
    {
        try {
            $fork = $this->forkContainer->fork($task);
        } catch (\RuntimeException $e) {
            $this->log->error($e->getMessage());
            throw $e;
        }

        $fork->setTask($task);

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

            foreach ($this->signals as $sig) {
                $this->pcntl->signal($sig, SIG_DFL, true);
            }

            $resultQueue = new ResultQueue($this->ownerPid);
            register_shutdown_function(function () use ($fork, $resultQueue) {
                if ($fork->hasNoResult() || !$fork->isQueued()) {
                    $result = new Result();
                    $result->setFailure();
                    $fork->setResult($result);
                    $resultQueue->enqueue($fork);
                }
            });

            $this->log->info('----> started the function.');
            $fork->executeTask();
            $this->log->info('<---- completed the function.');

            $resultQueue->enqueue($fork);
            $fork->setQueued();
            $this->log->info('queued the result.');

            $this->token->back();
            $this->log->info('return the token and exit.');
            $this->_exit();
            // @codeCoverageIgnoreEnd
        }
    }

    /**
     * fork process
     * the processes which forked are wait for token.
     *
     * @param   callable                    $callable
     * @param   mixed                       $args
     * @param   string                      $tag
     * @param   \Ackintosh\Snidel\Token     $token
     * @return  void
     * @throws  \RuntimeException
     */
    private function prefork($callable, $args = array(), $tag = null, Token $token = null)
    {
        $this->processToken = $token ? $token : $this->token;
        $task = new Task($callable, $args, $tag);

        try {
            $fork = $this->forkContainer->fork($task);
        } catch (\RuntimeException $e) {
            $this->log->error($e->getMessage());
            throw $e;
        }

        $fork->setTask($task);

        if (getmypid() === $this->ownerPid) {
            // parent
            $this->log->info('created child process. pid: ' . $fork->getPid());
            $this->childPids[] = $fork->getPid();
        } else {
            // @codeCoverageIgnoreStart
            // child
            foreach ($this->signals as $sig) {
                $this->pcntl->signal($sig, SIG_DFL, true);
            }

            /**
             * in php5.3, we can not use $this in anonymous functions
             */
            $dataRepository     = $this->dataRepository;
            $log                = $this->log;
            $processToken       = $this->processToken;
            register_shutdown_function(function () use ($fork, $dataRepository, $log, $processToken) {
                $data = $dataRepository->load(getmypid());
                try {
                    $data->write($fork);
                } catch (SharedMemoryControlException $e) {
                    throw $e;
                }
                $log->info('<-- return token.');
                $processToken->back();
            });

            $log->info('--> waiting for the token come around.');
            if ($processToken->accept()) {
                $log->info('----> started the function.');
                $fork->executeTask();
                $log->info('<---- completed the function.');
            }

            $this->_exit();
            // @codeCoverageIgnoreEnd
        }

        return $fork->getPid();
    }

    /**
     * waits until all tasks that queued by Snidel::fork() are completed
     *
     * @return  void
     */
    public function wait()
    {
        $this->forkContainer->wait();
        $this->joined = true;
    }

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

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

    /**
     * gets results
     *
     * @param   string  $tag
     * @return  \Ackintosh\Snidel\ForkCollection
     * @throws  \InvalidArgumentException
     */
    public function get($tag = null)
    {
        if (!$this->joined) {
            $this->wait();
        }

        if (!$this->forkContainer->hasTag($tag)) {

The expression $this->forkContainer->hasTag($tag) of type boolean|null is loosely compared to false; this is ambiguous if the boolean can be false. You might want to explicitly use !== null instead.

            throw new \InvalidArgumentException('unknown tag: ' . $tag);
        }

        return $this->forkContainer->getCollection($tag);
    }

    /**
     * sends signal to child
     *
     * @param   int     $sig
     * @return  void
     */
    public function sendSignalToChildren($sig)
    {
        foreach ($this->childPids as $pid) {
            $this->log->info('----> sending a signal to child. pid: ' . $pid);
            posix_kill($pid, $sig);
        }
    }

    public function setReceivedSignal($sig)
    {
        $this->receivedSignal = $sig;
    }

    /**
     * delete shared memory
     *
     * @return  void
     * @throws  \Ackintosh\Snidel\Exception\SharedMemoryControlException
     */
    private function deleteAllData()
    {
        foreach ($this->childPids as $pid) {
            $data = $this->dataRepository->load($pid);
            try {
                $data->deleteIfExists();
            } catch (SharedMemoryControlException $e) {
                throw $e;
            }
        }
    }

    /**
     * create map object
     *
     * @param   array       $args
     * @param   callable    $callable
     * @return  \Ackintosh\Snidel\MapContainer
     */
    public function map(Array $args, $callable)
    {
        return new MapContainer($args, $callable, $this->concurrency);
    }

    /**
     * run map object
     *
     * @param   \Ackintosh\Snidel\MapContainer
     * @return  array
     * @throws  \RuntimeException
     */
    public function run(MapContainer $mapContainer)
    {
        try {
            $this->forkTheFirstProcessing($mapContainer);
            $this->waitsAndConnectsProcess($mapContainer);
        } catch (\RuntimeException $e) {
            $this->exceptionHasOccured = true;
            throw $e;
        }

        return $this->getResultsOf($mapContainer);
    }

    /**
     * fork the first processing of the map container
     *
     * @param   \Ackintosh\Snidel\MapContainer
     * @return  void
     * @throws  \RuntimeException
     */
    private function forkTheFirstProcessing(MapContainer $mapContainer)
    {
        foreach ($mapContainer->getFirstArgs() as $args) {
            try {
                $childPid = $this->prefork($mapContainer->getFirstMap()->getCallable(), $args);
            } catch (\RuntimeException $e) {
                throw $e;
            }
            $mapContainer->getFirstMap()->countTheForked();
            $mapContainer->getFirstMap()->addChildPid($childPid);
        }
    }

    /**
     * waits and connects the process of map container
     *
     * @param   \Ackintosh\Snidel\MapContainer
     * @return  void
     * @throws  \RuntimeException
     */
    private function waitsAndConnectsProcess(MapContainer $mapContainer)
    {
        if ($this->joined) {
            return;
        }

        while ($mapContainer->isProcessing()) {
            try {
                $fork = $this->forkContainer->waitSimply();
            } catch (SharedMemoryControlException $e) {
                throw $e;
            }

            $childPid = $fork->getPid();
            if ($fork->hasNotFinishedSuccessfully()) {
                $message = 'an error has occurred in child process. pid: ' . $childPid;
                $this->log->error($message);
                throw new \RuntimeException($message);
            }

            unset($this->childPids[array_search($childPid, $this->childPids)]);
            if ($nextMap = $mapContainer->nextMap($childPid)) {
                try {
                    $nextMapPid = $this->prefork(
                        $nextMap->getCallable(),
                        $fork,
                        null,
                        $nextMap->getToken()
                    );
                } catch (\RuntimeException $e) {
                    throw $e;
                }
                $message = sprintf('processing is connected from [%d] to [%d]', $childPid, $nextMapPid);
                $this->log->info($message);
                $nextMap->countTheForked();
                $nextMap->addChildPid($nextMapPid);
            }
            $mapContainer->countTheCompleted($childPid);
        }

        $this->joined = true;
    }

    /**
     * gets results of map container
     *
     * @param   \Ackintosh\Snidel\MapContainer
     * @return  array
     */
    private function getResultsOf(MapContainer $mapContainer)
    {
        $results = array();
        foreach ($mapContainer->getLastMapPids() as $pid) {
            $results[] = $this->forkContainer->get($pid)->getResult()->getReturn();
        }

        return $results;
    }

    private function _exit($status = 0)
    {
        exit($status);

The method _exit() contains an exit expression.

    }

    public function __destruct()
    {
        if ($this->masterProcessId !== null && $this->ownerPid === getmypid()) {
            $this->log->info('shutdown master process.');
            posix_kill($this->masterProcessId, SIGTERM);

            unset($this->forkContainer);
        }

        if ($this->exceptionHasOccured) {
            $this->log->info('destruct processes are started.(exception has occured)');
            $this->log->info('--> deleting all shared memory.');
            $this->deleteAllData();
        } elseif ($this->ownerPid === getmypid() && !$this->joined && $this->receivedSignal === null) {
            $message = 'snidel will have to wait for the child process is completed. please use Snidel::wait()';
            $this->log->error($message);
            $this->log->info('destruct processes are started.');

            $this->log->info('--> sending a signal to children.');
            $this->sendSignalToChildren(SIGTERM);

            $this->log->info('--> deleting all shared memory.');
            $this->deleteAllData();

            $this->log->info('--> deleting token.');
            unset($this->token);

            $this->log->info('--> destruct processes are finished successfully.');
            throw new \LogicException($message);
        }
    }
}