QueueShell::_memoryUsage() - Code Metrics - Inspection of "WIP - Cakephp 3.x upgrade" - Oefenweb/cakephp-queue - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Pull Request — master (#35)

unknown
created 2019-05-22 15:30 UTC
QueueShell::_memoryUsage() A

↳ Parent: QueueShell
Complexity

Conditions	2
Paths	2
Size

Total Lines	10
Code Lines	5
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
cc	2
eloc	5
nc	2
nop	0
dl	0
loc	10
rs	10
c	0
b	0
f	0
<?php
namespace Queue\Shell;

use Cake\Console\Shell;
use Cake\Core\Configure;
use Cake\I18n\Number;
use Cake\Log\Log;
use Cake\ORM\Exception\PersistenceFailedException;
use Cake\Utility\Inflector;
use Cake\Utility\Text;
use Exception;
use Queue\Model\Entity\QueuedTask;
use Queue\Model\QueueException;
use Queue\Queue\Config;
use Queue\Queue\TaskFinder;
use Queue\Shell\Task\AddInterface;
use Queue\Shell\Task\QueueTaskInterface;
use RuntimeException;
use Throwable;
declare(ticks = 1);

/**
 * Main shell to init and run queue workers.
 *
 * @author [email protected]
 * @license http://www.opensource.org/licenses/mit-license.php The MIT License
 * @link http://github.com/MSeven/cakephp_queue
 * @property \Queue\Model\Table\QueuedTasksTable $QueuedTasks
 */
class QueueShell extends Shell
{

    /**
     *
     * @var string
     */
    public $modelClass = 'Queue.QueuedTasks';

    /**
     *
     * @var array|null
     */
    protected $_taskConf;

    /**
     *
     * @var int
     */
    protected $_time = 0;

    /**
     *
     * @var bool
     */
    protected $_exit = false;

    /**
     * Overwrite shell initialize to dynamically load all Queue Related Tasks.
     *
     * @return void
     */
    public function initialize()
    {
        $taskFinder = new TaskFinder();
        $this->tasks = $taskFinder->allAppAndPluginTasks();

        parent::initialize();
    }

    /**
     *
     * @return void
     */
    public function startup()
    {
        if ($this->param('quiet')) {
            $this->interactive = false;
        }

        parent::startup();
    }

    /**
     *
     * @return string
     */
    public function _getDescription()
    {
        $tasks = [];
        foreach ($this->taskNames as $loadedTask) {
            $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask);
        }
        $tasks = implode(PHP_EOL, $tasks);

        $text = <<<TEXT
Simple and minimalistic job queue (or deferred-task) system.

Available Tasks:
$tasks
TEXT;
        return $text;
    }

    /**
     * Look for a Queue Task of hte passed name and try to call add() on it.
     * A QueueTask may provide an add function to enable the user to create new jobs via commandline.
     *
     * @return void
     */
    public function add()
    {
        if (count($this->args) < 1) {
            $this->out('Please call like this:');
            $this->out('       bin/cake queue add <taskname>');
            $this->_displayAvailableTasks();

            return;
        }

        $name = Inflector::camelize($this->args[0]);
        if (in_array('Queue' . $name, $this->taskNames, true)) {
            /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */
            $task = $this->{'Queue' . $name};
            if (! ($task instanceof AddInterface)) {
                $this->abort('This task does not support adding via CLI call');
            }
            $task->add();
        } else {
            $this->out('Error: Task not found: ' . $name);
            $this->_displayAvailableTasks();
        }
    }

    /**
     * Output the task without Queue or Task
     * example: QueueImageTask becomes Image on display
     *
     * @param string $task
     *            Task name
     * @return string Cleaned task name
     */
    protected function _taskName($task)
    {
        if (strpos($task, 'Queue') === 0) {
            return substr($task, 5);
        }
        return $task;
    }

    /**
     * Run a QueueWorker loop.
     * Runs a Queue Worker process which will try to find unassigned jobs in the queue
     * which it may run and try to fetch and execute them.
     *
     * @return int|null
     */
    public function runworker()
    {
        try {
            $pid = $this->_initPid();
        } catch (PersistenceFailedException $exception) {
            $this->err($exception->getMessage());
            $limit = (int) Configure::read('Queue.maxWorkers');
            if ($limit) {
                $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')');
            }
            return static::CODE_ERROR;
        }

        // Enable Garbage Collector (PHP >= 5.3)
        if (function_exists('gc_enable')) {
            gc_enable();
        }
        if (function_exists('pcntl_signal')) {
            pcntl_signal(SIGTERM, [
                &$this,
                '_exit'
            ]);
            pcntl_signal(SIGINT, [
                &$this,
                '_exit'
            ]);
            pcntl_signal(SIGTSTP, [
                &$this,
                '_exit'
            ]);
            pcntl_signal(SIGQUIT, [
                &$this,
                '_exit'
            ]);
        }
        $this->_exit = false;

        $startTime = time();
        $types = $this->_stringToArray($this->param('type'));

        while (! $this->_exit) {
            $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE);

            $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types);

            if ($QueuedTask) {
                $this->runJob($QueuedTask, $pid);
            } elseif (Configure::read('Queue.exitWhenNothingToDo')) {
                $this->out('nothing to do, exiting.');
                $this->_exit = true;
            } else {
                $this->out('nothing to do, sleeping.');
                sleep(Config::sleepTime());
            }

            // check if we are over the maximum runtime and end processing if so.
            if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) {
                $this->_exit = true;
                $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
                $this->out('queue', /** @scrutinizer ignore-type */ 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
            }
            if ($this->_exit || mt_rand(0, 100) > (100 - (int) Config::gcprob())) {
                $this->out(__d('queue', 'Performing old job cleanup.'));
                $this->QueuedTasks->cleanOldJobs();
            }
            $this->hr();
        }

        $this->_deletePid($pid);

        if ($this->param('verbose')) {
            $this->_log('endworker', $pid);
        }
    }

    /**
     *
     * @param \Queue\Model\Entity\QueuedTask $QueuedTask
     * @param string $pid
     * @return void
     */
    protected function runJob(QueuedTask $QueuedTask, $pid)
    {
        $this->out('Running Job of type "' . $QueuedTask->task . '"');
        $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false);
        $taskName = 'Queue' . $QueuedTask->task;

        try {
            $this->_time = time();

            $data = unserialize($QueuedTask->data);
            /** @var \Queue\Shell\Task\QueueTask $task */
            $task = $this->{$taskName};
            if (! $task instanceof QueueTaskInterface) {

                throw new RuntimeException('Task must implement ' . QueueTaskInterface::class);
            }

            $return = $task->run((array) $data, $QueuedTask->id);
class A
{
    function getObject()
    {
        return null;
    }

}

$a = new A();
$object = $a->getObject();
            if ($return !== null) {

                trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED);
            }
            $failureMessage = $taskName . ' failed';
        } catch (Throwable $e) {
            $return = false;

            $failureMessage = get_class($e) . ': ' . $e->getMessage();
            if (! ($e instanceof QueueException)) {
                $failureMessage .= "\n" . $e->getTraceAsString();
            }

            $this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid);
        } catch (Exception $e) {
            $return = false;

            $failureMessage = get_class($e) . ': ' . $e->getMessage();
            $this->_logError($taskName . "\n" . $failureMessage, $pid);
        }

        if ($return === false) {

            $this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage);
            $failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf());
            $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid);
            $this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.');

            return;
        }

        $this->QueuedTasks->markJobDone($QueuedTask);
        $this->out('Job Finished.');
    }

    /**
     * Manually trigger a Finished job cleanup.
     *
     * @return void
     */
    public function clean()
    {
        if (! Configure::read('Queue.cleanupTimeout')) {
            $this->abort('You disabled cleanuptimout in config. Aborting.');
        }

        $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int) Configure::read('Queue.cleanupTimeout')));
        $this->QueuedTasks->cleanOldJobs();
    }

    /**
     * Display current settings
     *
     * @return void
     */
    public function settings()
    {
        $this->out('Current Settings:');
        $conf = (array) Configure::read('Queue');
        foreach ($conf as $key => $val) {
            if ($val === false) {
                $val = 'no';
            }
            if ($val === true) {
                $val = 'yes';
            }
            $this->out('* ' . $key . ': ' . print_r($val, true));
        }

        $this->out();
    }

    /**
     * Display some statistics about Finished Jobs.
     *
     * @return void
     */
    public function stats()
    {
        $this->out('Jobs currently in the queue:');

        $types = $this->QueuedTasks->getTypes()->toArray();
        foreach ($types as $type) {
            $this->out('      ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type));
        }
        $this->hr();
        $this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength());
        $this->hr();
        $this->out('Finished job statistics:');
        $data = $this->QueuedTasks->getStats();
        foreach ($data as $item) {
            $this->out(' ' . $item['task'] . ': ');
            $this->out('   Finished Jobs in Database: ' . $item['num']);
            $this->out('   Average Job existence    : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's');
            $this->out('   Average Execution delay  : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's');
            $this->out('   Average Execution time   : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's');
        }
    }

    /**
     * Get option parser method to parse commandline options
     *
     * @return \Cake\Console\ConsoleOptionParser
     */
    public function getOptionParser()
    {
        $subcommandParser = [
            'options' => [ /*
                             * 'dry-run'=> array(
                             * 'short' => 'd',
                             * 'help' => 'Dry run the update, no jobs will actually be added.',
                             * 'boolean' => true
                             * ),
                             */
            ]
        ];
        $subcommandParserFull = $subcommandParser;
        $subcommandParserFull['options']['type'] = [
            'short' => 't',
            'help' => 'Type (comma separated list possible)',
            'default' => null
        ];

        return parent::getOptionParser()->setDescription($this->_getDescription())
            ->addSubcommand('clean', [
            'help' => 'Remove old jobs (cleanup)',
            'parser' => $subcommandParser
        ])
            ->addSubcommand('add', [
            'help' => 'Add Job',
            'parser' => $subcommandParser
        ])
            ->addSubcommand('stats', [
            'help' => 'Stats',
            'parser' => $subcommandParserFull
        ])
            ->addSubcommand('settings', [
            'help' => 'Settings',
            'parser' => $subcommandParserFull
        ])
            ->addSubcommand('runworker', [
            'help' => 'Run Worker',
            'parser' => $subcommandParserFull
        ]);
    }

    /**
     * Timestamped log.
     *
     * @param string $message
     *            Log type
     * @param string|null $pid
     *            PID of the process
     * @param bool $addDetails
     * @return void
     */
    protected function _log($message, $pid = null, $addDetails = true)
    {
        if (! Configure::read('Queue.log')) {
            return;
        }

        if ($addDetails) {
            $timeNeeded = $this->_timeNeeded();
            $memoryUsage = $this->_memoryUsage();
            $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
        }

        if ($pid) {
            $message .= ' (pid ' . $pid . ')';
        }
        Log::write('info', $message, [
            'scope' => 'queue'
        ]);
    }

    /**
     *
     * @param string $message
     * @param string|null $pid
     *            PID of the process
     * @return void
     */
    protected function _logError($message, $pid = null)
    {
        $timeNeeded = $this->_timeNeeded();
        $memoryUsage = $this->_memoryUsage();
        $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';

        if ($pid) {
            $message .= ' (pid ' . $pid . ')';
        }

        Log::write('error', $message);
    }

    /**
     * Returns a List of available QueueTasks and their individual configurations.
     *
     * @return array
     */
    protected function _getTaskConf()
    {
        if (! is_array($this->_taskConf)) {
            $this->_taskConf = [];
            foreach ($this->tasks as $task) {
                list ($pluginName, $taskName) = pluginSplit($task);

                $this->_taskConf[$taskName]['name'] = substr($taskName, 5);
                $this->_taskConf[$taskName]['plugin'] = $pluginName;
                if (property_exists($this->{$taskName}, 'timeout')) {
                    $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout;
                } else {
                    $this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout();
                }
                if (property_exists($this->{$taskName}, 'retries')) {
                    $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries;
                } else {
                    $this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries();
                }
                if (property_exists($this->{$taskName}, 'cleanupTimeout')) {
                    $this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout;
                } else {
                    $this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout();
                }
            }
        }
        return $this->_taskConf;
    }

    /**
     * Signal handling to queue worker for clean shutdown
     *
     * @param int $signal
     * @return void
     */
    protected function _exit($signal)
    {
        $this->out(__d('queue', 'Caught %d signal, exiting.', $signal));
        $this->_exit = true;
    }

    /**
     *
     * @return void
     */
    protected function _displayAvailableTasks()
    {
        $this->out('Available Tasks:');
        foreach ($this->taskNames as $loadedTask) {
            $this->out("\t" . '* ' . $this->_taskName($loadedTask));
        }
    }

    /**
     *
     * @return string
     */
    protected function _initPid()
    {
        $this->_pid = $this->_retrievePid();

        return $this->_pid;
    }

    /**
     *
     * @return string
     */
    protected function _retrievePid()
    {
        if (function_exists('posix_getpid')) {
            $pid = (string) posix_getpid();
        } else {
            $pid = $this->QueuedTasks->key();
        }
        return $pid;
    }

    /**
     *
     * @return string Memory usage in MB.
     */
    protected function _memoryUsage()
    {
        $limit = ini_get('memory_limit');

        $used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB';
        if ($limit !== '-1') {
            $used .= '/' . $limit;
        }

        return $used;
    }

    /**
     *
     * @param string|null $pid
     *
     * @return void
     */
    protected function _deletePid($pid)
    {
        if (! $pid) {
            $pid = $this->_pid;
        }
        if (! $pid) {
            return;
        }
    }

    /**
     *
     * @return string
     */
    protected function _timeNeeded()
    {
        $diff = $this->_time() - $this->_time($this->_time);
        $seconds = max($diff, 1);

        return $seconds . 's';
    }

    /**
     *
     * @param int|null $providedTime
     *
     * @return int
     */
    protected function _time($providedTime = null)
    {
        if ($providedTime) {
0   == false // true
0   == null  // true
123 == false // false
123 == null  // false

// It is often better to use strict comparison
0 === false // false
0 === null  // false
            return $providedTime;
        }

        return time();
    }

    /**
     *
     * @param string|null $param
     * @return array
     */
    protected function _stringToArray($param)
    {
        if (! $param) {
            return [];
        }

        $array = Text::tokenize($param);

        return array_filter($array);
    }
}

Oefenweb / cakephp-queue

Pull Request — master (#35)

QueueShell::_memoryUsage() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
