Manager::updateTrackerSetQueued() - Code Metrics - Inspection of "fix worker hang after SIGCLD when db conn timeout,..." - comodojo/extender.framework - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — 2.0 ( dbbe01...d43752 )

by Marco

created 2017-09-05 22:37 UTC

Manager::updateTrackerSetQueued() A

↳ Parent: Manager

Complexity

Conditions	1
Paths	1

Size

Total Lines	7
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	7
rs	9.4285
c	0
b	0
f	0
cc	1
eloc	3
nc	1
nop	0

<?php namespace Comodojo\Extender\Task;

use \Comodojo\Foundation\Base\Configuration;
use \Comodojo\Foundation\Events\Manager as EventsManager;
use \Comodojo\Foundation\Logging\LoggerTrait;
use \Comodojo\Foundation\Events\EventsTrait;
use \Comodojo\Daemon\Utils\ProcessTools;
use \Comodojo\Foundation\Base\ConfigurationTrait;
use \Comodojo\Extender\Traits\TasksTableTrait;
use \Comodojo\Extender\Utils\Validator as ExtenderCommonValidations;
use \Comodojo\Extender\Components\Ipc;
use \Comodojo\Extender\Task\Table as TasksTable;
use \Comodojo\Extender\Components\Database;
use \Psr\Log\LoggerInterface;
use \Exception;

/**
* @package     Comodojo Extender
* @author      Marco Giovinazzi <[email protected]>
* @license     MIT
*
* LICENSE:
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
 */

class Manager {

    use ConfigurationTrait;
    use LoggerTrait;
    use EventsTrait;
    use TasksTableTrait;

    /**
     * @var int
     */
    protected $lagger_timeout;

    /**
     * @var bool
     */
    protected $multithread;

    /**
     * @var int
     */
    protected $max_runtime;

    /**
     * @var int
     */
    protected $max_childs;

    /**
     * @var Ipc
     */
    protected $ipc;

    /**
     * @var Locker
     */
    protected $locker;

    /**
     * @var Tracker
     */
    protected $tracker;

    /**
     * Class constructor
     *
     * @param string $manager_name
/**
 * @param array $germany
 * @param array $island
 * @param array $italy
 */
function finale($germany, $island) {
    return "2:1";
}
     * @param Configuration $configuration
     * @param LoggerInterface $logger
     * @param TasksTable $tasks
     * @param EventsManager $events
     * @param EntityManager $em
/**
 * @param array $germany
 * @param array $island
 * @param array $italy
 */
function finale($germany, $island) {
    return "2:1";
}
     */
    public function __construct(
        Locker $locker,
        Configuration $configuration,
        LoggerInterface $logger,
        TasksTable $tasks,
        EventsManager $events
    ) {

        $this->setConfiguration($configuration);
        $this->setLogger($logger);
        $this->setTasksTable($tasks);
        $this->setEvents($events);

        $this->locker = $locker;
        $this->tracker = new Tracker($configuration, $logger);
        $this->ipc = new Ipc($configuration);

        // retrieve parameters
        $this->lagger_timeout = ExtenderCommonValidations::laggerTimeout($this->configuration->get('child-lagger-timeout'));
        $this->multithread = ExtenderCommonValidations::multithread($this->configuration->get('multithread'));
        $this->max_runtime = ExtenderCommonValidations::maxChildRuntime($this->configuration->get('child-max-runtime'));
        $this->max_childs = ExtenderCommonValidations::forkLimit($this->configuration->get('fork-limit'));

        // $logger->debug("Tasks Manager online", array(

        //     'lagger_timeout' => $this->lagger_timeout,

        //     'multithread' => $this->multithread,

        //     'max_runtime' => $this->max_runtime,

        //     'max_childs' => $this->max_childs,

        //     'tasks_count' => count($this->table)

        // ));

        set_error_handler([$this, 'customErrorHandler']);

    }

    public function __destruct() {

        restore_error_handler();

    }

    public function add(Request $request) {

        $this->tracker->setQueued($request);

        return $this;

    }

    public function addBulk(array $requests) {

        foreach ($requests as $id => $request) {

            if ($request instanceof \Comodojo\Extender\Task\Request) {
                $this->add($request);
            } else {
                $this->logger->error("Skipping invalid request with local id $id: class mismatch");
            }

        }

        return $this;

    }

    public function run() {

        $this->updateTrackerSetQueued();

        while ( $this->tracker->countQueued() > 0 ) {

            // Start to cycle queued tasks
            $this->cycle();

        }

        $this->ipc->free();

        return $this->tracker->getCompleted();

    }

    public function customErrorHandler($errno, $errstr, $errfile, $errline) {

        $this->getLogger()->error("Unhandled error ($errno): $errstr [in $errfile line $errline]");

        return true;

    }

    protected function cycle() {

        foreach ($this->tracker->getQueued() as $uid => $request) {

            if ( $this->multithread === false ) {

                $this->runSingleThread($uid, $request);

            } else {

                try {

                    $pid = $this->forker($request);

                } catch (Exception $e) {

                    $result = self::generateSyntheticResult($uid, $e->getMessage(), $request->getJid(), false);

                    $this->updateTrackerSetAborted($uid, $result);

                    if ( $request->isChain() ) $this->evalChain($request, $result);

                    continue;

                }

                $this->updateTrackerSetRunning($uid, $pid);

                if ( $this->max_childs > 0 && $this->tracker->countRunning() >= $this->max_childs ) {

                    while( $this->tracker->countRunning() >= $this->max_childs ) {

                        $this->catcher();

                    }

                }

            }

        }

        // spawn the loop if multithread
        if ( $this->multithread === true ) $this->catcher_loop();

    }

    protected function runSingleThread($uid, Request $request) {

        $pid = ProcessTools::getPid();

        $this->updateTrackerSetRunning($uid, $pid);

        $result = Runner::fastStart(
            $request,
            $this->getConfiguration(),
            $this->getLogger(),
            $this->getTasksTable(),
            $this->getEvents()
        );

        if ( $request->isChain() ) $this->evalChain($request, $result);

        $this->updateTrackerSetCompleted($uid, $result);

        $success = $result->success === false ? "error" : "success";
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
        $this->logger->notice("Task ".$request->getName()."(uid: ".$request->getUid().") ends in $success");

    }

    private function forker(Request $request) {

        $uid = $request->getUid();

        try {

            $this->ipc->init($uid);

        } catch (Exception $e) {

            $this->logger->error("Aborting task ".$request->getName().": ".$e->getMessage());

            $this->ipc->hang($uid);

            throw $e;

        }

        $pid = pcntl_fork();

        if ( $pid == -1 ) {

            // $this->logger->error("Could not fok job, aborting");


            throw new Exception("Unable to fork job, aborting");

        } elseif ( $pid ) {

            //PARENT will take actions on processes later

            $niceness = $request->getNiceness();

            if ( $niceness !== null ) ProcessTools::setNiceness($niceness, $pid);

        } else {

            $this->ipc->close($uid, Ipc::READER);

            $result = Runner::fastStart(
                $request,
                $this->getConfiguration(),
                $this->getLogger(),
                $this->getTasksTable(),
                $this->getEvents()
            );

            $this->ipc->write($uid, serialize($result));

            $this->ipc->close($uid, Ipc::WRITER);

            exit(!$result->success);

        }

        return $pid;

    }

    private function catcher_loop() {

        while ( !empty($this->tracker->getRunning()) ) {

            $this->catcher();

        }

    }

    /**
     * Catch results from completed jobs
     *
     */
    private function catcher() {

        foreach ( $this->tracker->getRunning() as $uid => $request ) {

            if ( ProcessTools::isRunning($request->getPid()) === false ) {

                $this->ipc->close($uid, Ipc::WRITER);

                try {

                    $raw_output = $this->ipc->read($uid);

                    $result = unserialize(rtrim($raw_output));

                    $this->ipc->close($uid, Ipc::READER);

                } catch (Exception $e) {

                    $result = self::generateSyntheticResult($uid, $e->getMessage(), $request->getJid(), false);

                    // $this->logger->error($result);


                }

                if ( $request->isChain() ) $this->evalChain($request, $result);

                $this->updateTrackerSetCompleted($uid, $result);

                $success = $result->success === false ? "error" : "success";
                $this->logger->notice("Task ".$request->getName()."(uid: ".$request->getUid().") ends in $success");

            } else {

                $current_time = microtime(true);

                $request_max_time = $request->getMaxtime();
                $maxtime = $request_max_time === null ? $this->max_runtime : $request_max_time;

                if ( $current_time > $request->getStartTimestamp() + $maxtime ) {

                    $pid = $request->getPid();

                    $this->logger->warning("Killing pid $pid due to maximum exec time reached", [
                        "START_TIME"    => $request->getStartTimestamp(),
                        "CURRENT_TIME"  => $current_time,
                        "MAX_RUNTIME"   => $maxtime
                    ]);

                    $kill = ProcessTools::term($pid, $this->lagger_timeout);

                    if ( $kill ) {
                        $this->logger->warning("Pid $pid killed");
                    } else {
                        $this->logger->warning("Pid $pid could not be killed");
                    }

                    $this->ipc->hang($uid);

                    $result = self::generateSyntheticResult($uid, "Job killed due to max runtime reached", $request->getJid(), false);

                    if ( $request->isChain() ) $this->evalChain($request, $result);

                    $this->updateTrackerSetCompleted($uid, $result);

                    $this->logger->notice("Task ".$request->getName()."(uid: $uid) ends in error");

                }

            }

        }

    }

    private function evalChain(Request $request, Result $result) {

        if ( $result->success && $request->hasOnDone() ) {
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
            $chain_done = $request->getOnDone();
            $chain_done->getParameters()->set('parent', $result);
            $chain_done->setParentUid($result->uid);
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
            $this->add($chain_done);
        }

        if ( $result->success === false && $request->hasOnFail() ) {
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
            $chain_fail = $request->getOnFail();
            $chain_fail->getParameters()->set('parent', $result);
            $chain_fail->setParentUid($result->uid);
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
            $this->add($chain_fail);
        }

        if ( $request->hasPipe() ) {
            $chain_pipe = $request->getPipe();
            $chain_pipe->getParameters()->set('parent', $result);
            $chain_pipe->setParentUid($result->uid);
<?php

/**
 * @property int $x
 * @property int $y
 * @property string $text
 */
class MyLabel
{
    private $properties;

    private $allowedProperties = array('x', 'y', 'text');

    public function __get($name)
    {
        if (isset($properties[$name]) && in_array($name, $this->allowedProperties)) {
            return $properties[$name];
        } else {
            return null;
        }
    }

    public function __set($name, $value)
    {
        if (in_array($name, $this->allowedProperties)) {
            $properties[$name] = $value;
        } else {
            throw new \LogicException("Property $name is not defined.");
        }
    }

}
            $this->add($chain_pipe);
        }

    }

    private function generateSyntheticResult($uid, $message, $jid = null, $success = true) {

        return new Result([
            $uid,
            null,
            $jid,
            null,
            $success,
            null,
            null,
            $message,
            null
        ]);

    }

    private function updateTrackerSetQueued() {

        $this->locker->lock([
            'QUEUED' => $this->tracker->countQueued()
        ]);

    }

    private function updateTrackerSetRunning($uid, $pid) {

        $this->tracker->setRunning($uid, $pid);
        $this->locker->lock([
            'QUEUED' => $this->tracker->countQueued(),
            'RUNNING' => $this->tracker->countRunning()
        ]);

    }

    private function updateTrackerSetCompleted($uid, $result) {

        $this->tracker->setCompleted($uid, $result);

        $lock_data = [
            'RUNNING' => $this->tracker->countRunning(),
            'COMPLETED' => 1
        ];

        if ( $result->success ) {
            $lock_data['SUCCEEDED'] = 1;
        } else {
            $lock_data['FAILED'] = 1;
        }

        $this->locker->lock($lock_data);

    }

    private function updateTrackerSetAborted($uid, $result) {

        $this->tracker->setAborted($uid, $result);
        $lock_data = [
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}
            'RUNNING' => $this->tracker->countRunning(),
            'ABORTED' => 1
        ];

    }

    // private function updateLocker() {

    //
    //     $this->locker->lock([

    //         'QUEUED' => $this->tracker->countQueued(),

    //         'RUNNING' => $this->tracker->countRunning(),

    //         'COMPLETED' => $this->tracker->countCompleted(),

    //         'SUCCEEDED' => $this->tracker->countSucceeded(),

    //         'FAILED' => $this->tracker->countFailed()

    //     ]);
    //
    // }

}


1			<?php namespace Comodojo\Extender\Task;
2
3			use \Comodojo\Foundation\Base\Configuration;
4			use \Comodojo\Foundation\Events\Manager as EventsManager;
5			use \Comodojo\Foundation\Logging\LoggerTrait;
6			use \Comodojo\Foundation\Events\EventsTrait;
7			use \Comodojo\Daemon\Utils\ProcessTools;
8			use \Comodojo\Foundation\Base\ConfigurationTrait;
9			use \Comodojo\Extender\Traits\TasksTableTrait;
10			use \Comodojo\Extender\Utils\Validator as ExtenderCommonValidations;
11			use \Comodojo\Extender\Components\Ipc;
12			use \Comodojo\Extender\Task\Table as TasksTable;
13			use \Comodojo\Extender\Components\Database;
14			use \Psr\Log\LoggerInterface;
15			use \Exception;
16
17			/**
18			* @package Comodojo Extender
19			* @author Marco Giovinazzi <[email protected]>
20			* @license MIT
21			*
22			* LICENSE:
23			*
24			* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
25			* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
26			* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
27			* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
28			* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
29			* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
30			* THE SOFTWARE.
31			*/
32
33			class Manager {
34
35			use ConfigurationTrait;
36			use LoggerTrait;
37			use EventsTrait;
38			use TasksTableTrait;
39
40			/**

comodojo / extender.framework

Push — 2.0 ( dbbe01...d43752 )

Manager::updateTrackerSetQueued() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like