Display usefull stats about the workers.

Note: The workers status is conveniently stored by ResqueStatus.

@return void
@see ResqueStatus\ResqueStatus::isSchedulerWorker()
@see ResqueStatus\ResqueStatus::getPausedWorker()
11839
public function stats()
    {
        $ResqueStatus = $this->ResqueStatus;

        $workers = CakeResque::getWorkers();

        // List of all queues
        $queues = array_unique(CakeResque::getQueues());

        // List of queues monitored by a worker
        $activeQueues = [];
        foreach ($workers as $worker) {
            $workerParams = explode(':', $worker);
            $activeQueues = array_merge($activeQueues, explode(',', array_pop($workerParams)));
        }

        $this->out("\n");
        $this->out('<info>' . __d('cake_resque', 'Resque Statistics') . '</info>');
        $this->hr();
        $this->out("\n");

        $this->out('<info>' . __d('cake_resque', 'Jobs Stats') . '</info>');
        $this->out('   ' . __d('cake_resque', 'Processed Jobs : {0}', number_format(Resque_Stat::get('processed'))));
        $this->out('   <warning>' . __d('cake_resque', 'Failed Jobs    : {0}', number_format(Resque_Stat::get('failed'))) . '</warning>');

        if (Configure::read('CakeResque.Scheduler.enabled') === true) {
            $this->out('   ' . __d('cake_resque', 'Scheduled Jobs : {0}', number_format(Stat::get())));
        }

        $this->out("\n");

        $count = [];
        $this->out('<info>' . __d('cake_resque', 'Queues Stats') . '</info>');
        for ($i = count($queues) - 1; $i >= 0; --$i) {
            $count[ $queues[ $i ] ] = CakeResque::getQueueSize($queues[ $i ]);
            if (!in_array($queues[ $i ], $activeQueues) && $count[ $queues[ $i ] ] == 0) {
                unset($queues[ $i ]);
            }
        }

        $this->out('   ' . __d('cake_resque', 'Queues count : {0}', count($queues)));
        foreach ($queues as $queue) {
            $this->out(sprintf("\t- %-15s : %12s %s", $queue, number_format($count[ $queue ]), __dn('cake_resque', 'pending job', 'pending jobs', $count[ $queue ]) . (!in_array($queue, $activeQueues) ? " <error>(unmonitored queue)</error>" : '')));
        }

        $this->out("\n");
        $this->out('<info>' . __d('cake_resque', 'Workers Stats') . '</info>');
        $this->out('   ' . __d('cake_resque', 'Workers count : {0}', count($workers)));

        $pausedWorkers = $ResqueStatus->getPausedWorker();
        $schedulerWorkers = [];

        if (!empty($workers)) {
            $this->out("\t<info>" . strtoupper(__d('cake_resque', 'regular workers')) . "</info>");
            foreach ($workers as $worker) {
                if (Configure::read('CakeResque.Scheduler.enabled') === true && $ResqueStatus->isSchedulerWorker($worker)) {
                    $schedulerWorkers[] = $worker;
                    continue;
                }
                $this->out("\t* <bold>" . (string)$worker . '</bold>' . (in_array((string)$worker, $pausedWorkers) ? ' <warning>(' . __d('cake_resque', 'paused') . ')</warning>' : ''));
                $this->out("\t   - " . __d('cake_resque', 'Started on') . "     : " . CakeResque::getWorkerStartDate($worker));
                $this->out("\t   - " . __d('cake_resque', 'Processed Jobs') . " : " . $worker->getStat('processed'));
                $worker->getStat('failed') == 0
                    ? $this->out("\t   - " . __d('cake_resque', 'Failed Jobs') . "    : " . $worker->getStat('failed'))
                    : $this->out("\t   - <warning>" . __d('cake_resque', 'Failed Jobs') . "    : " . $worker->getStat('failed') . "</warning>");
            }
        }

        $this->out("\n");

        if (!empty($schedulerWorkers)) {
            $this->out("\t<info>" . strtoupper(__d('cake_resque', 'scheduler worker')) . "</info>" . (in_array((string)$schedulerWorkers[0], $pausedWorkers) ? ' <warning>(' . __d('cake_resque', 'paused') . ')</warning>' : ''));
            foreach ($schedulerWorkers as $worker) {
                $schedulerWorker = new ResqueScheduler();
                $delayedJobCount = $schedulerWorker->getDelayedQueueScheduleSize();
                $this->out("\t   - " . __d('cake_resque', 'Started on') . "     : " . CakeResque::getWorkerStartDate($worker));
                $this->out("\t   - " . __d('cake_resque', 'Delayed Jobs') . "   : " . $delayedJobCount);

                if ($delayedJobCount > 0) {
                    $this->out("\t   - " . __d('cake_resque', 'Next Job on') . "    : " . strftime('%a %b %d %H:%M:%S %Z %Y', $schedulerWorker->nextDelayedTimestamp()));
                }
            }
            $this->out("\n");
        } elseif (Configure::read('CakeResque.Scheduler.enabled') === true) {
            $jobsCount = ResqueScheduler::getDelayedQueueScheduleSize();
            if ($jobsCount > 0) {
                $this->out("\t<error>************ " . __d('cake_resque', 'Alert') . " ************</error>");
                $this->out("\t<bold>" . __d('cake_resque', 'The Scheduler Worker is not running') . "</bold>");
                $this->out("\t" . __d('cake_resque', 'But there is still <bold>{0}</bold> scheduled jobs left in its queue', $jobsCount));
                $this->out("\t<error>********************************</error>");
                $this->out("\n");
            }
        }
    }
1073356
public function onWorkerHandlerStatusMessage(WorkerHandlerStatus $msg)
    {
        $handlerId = $msg->getWorkerHandlerId();

        if (isset($this->workerHandlers[$handlerId])) {
            // Remove to make sure the handlers are ordered by time.
            unset($this->workerHandlers[$handlerId]);
        }

        $this->workerHandlers[$handlerId] = microtime(true);

        $requestId = $msg->getRequestId();

        if ($requestId === null) {
            return;
        }

        $this->getLogger()->debug('Storage has a result available for: '.$requestId.'.');

        $this->sendResponseToClients($requestId);
    }
1131378
protected function getStatistics() {
        if (!$this->_statistics) {
            $this->_statistics = new WorkflowStatisticsList(
                $this->version,
                $this->solution['workspaceSid'],
                $this->solution['sid']
            );
        }

        return $this->_statistics;
    }
601374
public function measuredJobs()
    {
        $classes = (array) $this->connection()->smembers('measured_jobs');

        return collect($classes)->map(function ($class) {
            return preg_match('/job:(.*)$/', $class, $matches) ? $matches[1] : $class;
        })->all();
    }
1013708
public static function checkWorkerStatusForWindows()
    {
        foreach(static::$_processForWindows as $process_data)
        {
            $process = $process_data[0];
            $start_file = $process_data[1];
            $timer_id = $process_data[2];
            $status = proc_get_status($process);
            if(isset($status['running']))
            {
                if(!$status['running'])
                {
                    static::safeEcho("process $start_file terminated and try to restart\n");
                    Timer::del($timer_id);
                    proc_close($process);
                    static::forkOneWorkerForWindows($start_file);
                }
            }
            else
            {
                static::safeEcho("proc_get_status fail\n");
            }
        }
    }
111674
public function collectStats()
    {
        Container::get('hooks')->fire('model.index.collect_stats_start');

        // Collect some statistics from the database
        if (!Container::get('cache')->isCached('users_info')) {
            Container::get('cache')->store('users_info', Cache::getUsersInfo());
        }

        $stats = Container::get('cache')->retrieve('users_info');

        $query = DB::forTable('forums')
            ->select_expr('SUM(num_topics)', 'total_topics')
            ->select_expr('SUM(num_posts)', 'total_posts');

        $query = Container::get('hooks')->fireDB('model.index.collect_stats_query', $query);

        $query = $query->find_one();

        $stats['total_topics'] = Utils::numberFormat((int)$query['total_topics']);
        $stats['total_posts'] = Utils::numberFormat((int)$query['total_posts']);

        if (User::get()->g_view_users == '1') {
            $stats['newest_user'] = '<a href="'.
                Router::pathFor('userProfile', ['id' => $stats['last_user']['id']]).'">'.
                Utils::escape($stats['last_user']['username']).'</a>';
        } else {
            $stats['newest_user'] = Utils::escape($stats['last_user']['username']);
        }
        $stats['total_users'] = Utils::numberFormat((int)$stats['total_users']);

        $stats = Container::get('hooks')->fire('model.index.collect_stats', $stats);

        return $stats;
    }
899054
public function status() : array
    {
        $this->reloadPids();

        $workerStatus = [];
        if (empty($this->pids)) {
            return $workerStatus;
        }

        $Client = new \GearmanClient();
        $Client->addServer($this->gearmanHost, $this->gearmanPort);
        $Client->setTimeout(1000);
        foreach ($this->pids as $pool => $workerPids) {
            $workerStatus[$pool] = [];
            foreach ($workerPids as $workerName => $workerPid) {
                // raises php warning on timeout so we need the "evil @" here...
                $workerStatus[$pool][$workerName] = false;
                $start = microtime(true);
                $pong = @$Client->doHigh('ping_'.$workerName, 'ping');
                if ($pong === 'pong') {
                    $jobinfo = @$Client->doHigh('jobinfo_'.$workerName, 'foo');
                    $jobinfo = json_decode($jobinfo, true);
                    $workerStatus[$pool][$workerName] = $jobinfo;
                    $pingtime = microtime(true) - $start;
                    $workerStatus[$pool][$workerName]['ping'] = $pingtime;
                    $workerStatus[$pool][$workerName]['status'] = 'idle';
                } else {
                    $workerStatus[$pool][$workerName]['jobs_total'] = 'n/a';
                    $workerStatus[$pool][$workerName]['avg_jobs_min'] = 'n/a';
                    $workerStatus[$pool][$workerName]['uptime_seconds'] = 0;
                    $workerStatus[$pool][$workerName]['ping'] = 0;
                    $workerStatus[$pool][$workerName]['status'] = 'busy';
                }
            }
        }

        return $workerStatus;
    }
1301804
public function onAdminInit()
    {
        if (!in_array($this->App->Config->§specs['§type'], ['theme', 'plugin'], true)) {
            return; // Not applicable; themes/plugins only.
        } // e.g., We don't collect stats for a simple MU plugin.

        if (!$this->c::appCore()->s::applyFilters('app_stats_enable', true)) {
            return; // Stats disabled via filter.
        }
        $last_posted = $this->lastPosted();

        if ($last_posted['time'] > $this->outdated_stats_time) {
            return; // Not time to post stats again yet.
        } // Must be 1+ week apart (enforced by remote API).

        $anonymous_stats = [
            'os'            => PHP_OS,
            'php_version'   => PHP_VERSION,
            'mysql_version' => $this->s::wpDb()->db_version(),

            'wp_version'      => WP_VERSION,
            'product_version' => $this->App::VERSION,
            'product'         => $this->App->Config->©brand['§product_slug'],
        ];
        wp_remote_post($this->s::coreBrandStatsUrl('/log'), [
                'blocking'  => false,
                'sslverify' => false,
                'body'      => $anonymous_stats,
        ]);
        $this->lastPosted(['time' => time()]); // Record last time.
    }
10868
public function index()
    {
        return [
            'jobsPerMinute' => app(MetricsRepository::class)->jobsProcessedPerMinute(),
            'processes' => $this->totalProcessCount(),
            'queueWithMaxRuntime' => app(MetricsRepository::class)->queueWithMaximumRuntime(),
            'queueWithMaxThroughput' => app(MetricsRepository::class)->queueWithMaximumThroughput(),
            'recentlyFailed' => app(JobRepository::class)->countRecentlyFailed(),
            'recentJobs' => app(JobRepository::class)->countRecent(),
            'status' => $this->currentStatus(),
            'wait' => collect(app(WaitTimeCalculator::class)->calculate())->take(1),
            'periods' => [
                'recentJobs' => config('horizon.trim.recent'),
                'recentlyFailed' => config('horizon.trim.failed'),
            ],
        ];
    }
824716
public function doneWorking()
    {
        $this->currentJob = null;
        $this->working = false;
        $this->resqueInstance->stats->incr('processed');
        $this->resqueInstance->stats->incr('processed:' . (string)$this);
        $this->resqueInstance->redis->del($this->workerName() . ':' . (string)$this);
    }
371484
protected function show_workers_queues_list($workers)
    {
        $this->line("<info>Your workers:</info>");
        foreach ($workers as $worker_file_name) {
            if (is_dir(getcwd() . '/' . $this->workers_dir . '/' . $worker_file_name))
                continue;
            $this->line('<comment>' . $this->remove_extension($worker_file_name) . '</comment>');
        }
    }
461784
public function index() {
		$this->loadModel('Queue.QueueProcesses');
		$status = $this->QueueProcesses->status();

		$current = $this->QueuedJobs->getLength();
		$pendingDetails = $this->QueuedJobs->getPendingStats();
		$new = 0;
		foreach ($pendingDetails as $pendingDetail) {
			if ($pendingDetail['fetched'] || $pendingDetail['failed']) {
				continue;
			}
			$new++;
		}

		$data = $this->QueuedJobs->getStats();

		$taskFinder = new TaskFinder();
		$tasks = $taskFinder->allAppAndPluginTasks();

		$servers = $this->QueueProcesses->find()->distinct(['server'])->find('list', ['keyField' => 'server', 'valueField' => 'server'])->toArray();
		$this->set(compact('new', 'current', 'data', 'pendingDetails', 'status', 'tasks', 'servers'));
		$this->helpers[] = 'Tools.Format';
		$this->helpers[] = 'Tools.Time';
		$this->helpers[] = 'Tools.Text';
	}
676458
public function work()
    {
        if (!$this->connect()) {
            $this->logger()->alert(sprintf('Worker unable to connect, exiting'));
            $this->dispatchEvent('Worker.job.connectionFailed');

            return false;
        }

        $jobClass = $this->engine->getJobClass();
        $time = microtime(true);
        while ($this->running) {
            if (is_int($this->maxRuntime) && $this->runtime >= $this->maxRuntime) {
                $this->logger()->debug('Max runtime reached, exiting');
                $this->dispatchEvent('Worker.maxRuntime');
                break;
            } elseif (is_int($this->maxIterations) && $this->iterations >= $this->maxIterations) {
                $this->logger()->debug('Max iterations reached, exiting');
                $this->dispatchEvent('Worker.maxIterations');
                break;
            }

            $this->runtime += microtime(true) - $time;
            $time = microtime(true);
            $this->iterations++;
            $item = $this->engine->pop($this->queue);
            $this->dispatchEvent('Worker.job.seen', ['item' => $item]);
            if (empty($item)) {
                $this->dispatchEvent('Worker.job.empty');
                sleep($this->interval);
                continue;
            }

            $success = false;
            $job = new $jobClass($item, $this->engine);
            if (!is_callable($item['class'])) {
                $this->logger()->alert('Invalid callable for job. Rejecting job from queue.');
                $job->reject();
                $this->dispatchEvent('Worker.job.invalid', ['job' => $job]);
                continue;
            }

            $this->dispatchEvent('Worker.job.start', ['job' => $job]);

            try {
                $success = $this->perform($item, $job);
            } catch (Exception $e) {
                $this->dispatchEvent('Worker.job.exception', [
                    'job' => $job,
                    'exception' => $e,
                ]);
            }

            if ($success) {
                $job->acknowledge();
                $this->dispatchEvent('Worker.job.success', ['job' => $job]);
                continue;
            }

            $this->logger()->info('Failed. Releasing job to queue');
            $job->release();
            $this->dispatchEvent('Worker.job.failure', ['job' => $job]);
        }

        return true;
    }
1176411
protected static function writeStatisticsToStatusFile()
    {
        // For master process.
        if (static::$_masterPid === posix_getpid()) {
            $all_worker_info = array();
            foreach(static::$_pidMap as $worker_id => $pid_array) {
                /** @var /Workerman/Worker $worker */
                $worker = static::$_workers[$worker_id];
                foreach($pid_array as $pid) {
                    $all_worker_info[$pid] = array('name' => $worker->name, 'listen' => $worker->getSocketName());
                }
            }

            file_put_contents(static::$_statisticsFile, json_encode($all_worker_info)."\n", FILE_APPEND);
            $loadavg = function_exists('sys_getloadavg') ? array_map('round', sys_getloadavg(), array(2)) : array('-', '-', '-');
            file_put_contents(static::$_statisticsFile,
                "----------------------------------------------GLOBAL STATUS----------------------------------------------------\n", FILE_APPEND);
            file_put_contents(static::$_statisticsFile,
                'Workerman version:' . static::VERSION . "          PHP version:" . PHP_VERSION . "\n", FILE_APPEND);
            file_put_contents(static::$_statisticsFile, 'start time:' . date('Y-m-d H:i:s',
                    static::$_globalStatistics['start_timestamp']) . '   run ' . floor((time() - static::$_globalStatistics['start_timestamp']) / (24 * 60 * 60)) . ' days ' . floor(((time() - static::$_globalStatistics['start_timestamp']) % (24 * 60 * 60)) / (60 * 60)) . " hours   \n",
                FILE_APPEND);
            $load_str = 'load average: ' . implode(", ", $loadavg);
            file_put_contents(static::$_statisticsFile,
                str_pad($load_str, 33) . 'event-loop:' . static::getEventLoopName() . "\n", FILE_APPEND);
            file_put_contents(static::$_statisticsFile,
                count(static::$_pidMap) . ' workers       ' . count(static::getAllWorkerPids()) . " processes\n",
                FILE_APPEND);
            file_put_contents(static::$_statisticsFile,
                str_pad('worker_name', static::$_maxWorkerNameLength) . " exit_status      exit_count\n", FILE_APPEND);
            foreach (static::$_pidMap as $worker_id => $worker_pid_array) {
                $worker = static::$_workers[$worker_id];
                if (isset(static::$_globalStatistics['worker_exit_info'][$worker_id])) {
                    foreach (static::$_globalStatistics['worker_exit_info'][$worker_id] as $worker_exit_status => $worker_exit_count) {
                        file_put_contents(static::$_statisticsFile,
                            str_pad($worker->name, static::$_maxWorkerNameLength) . " " . str_pad($worker_exit_status,
                                16) . " $worker_exit_count\n", FILE_APPEND);
                    }
                } else {
                    file_put_contents(static::$_statisticsFile,
                        str_pad($worker->name, static::$_maxWorkerNameLength) . " " . str_pad(0, 16) . " 0\n",
                        FILE_APPEND);
                }
            }
            file_put_contents(static::$_statisticsFile,
                "----------------------------------------------PROCESS STATUS---------------------------------------------------\n",
                FILE_APPEND);
            file_put_contents(static::$_statisticsFile,
                "pid\tmemory  " . str_pad('listening', static::$_maxSocketNameLength) . " " . str_pad('worker_name',
                    static::$_maxWorkerNameLength) . " connections " . str_pad('send_fail', 9) . " "
                . str_pad('timers', 8) . str_pad('total_request', 13) ." qps    status\n", FILE_APPEND);

            chmod(static::$_statisticsFile, 0722);

            foreach (static::getAllWorkerPids() as $worker_pid) {
                posix_kill($worker_pid, SIGUSR2);
            }
            return;
        }

        // For child processes.
        reset(static::$_workers);
        /** @var \Workerman\Worker $worker */
        $worker            = current(static::$_workers);
        $worker_status_str = posix_getpid() . "\t" . str_pad(round(memory_get_usage(true) / (1024 * 1024), 2) . "M", 7)
            . " " . str_pad($worker->getSocketName(), static::$_maxSocketNameLength) . " "
            . str_pad(($worker->name === $worker->getSocketName() ? 'none' : $worker->name), static::$_maxWorkerNameLength)
            . " ";
        $worker_status_str .= str_pad(ConnectionInterface::$statistics['connection_count'], 11)
            . " " .  str_pad(ConnectionInterface::$statistics['send_fail'], 9)
            . " " . str_pad(static::$globalEvent->getTimerCount(), 7)
            . " " . str_pad(ConnectionInterface::$statistics['total_request'], 13) . "\n";
        file_put_contents(static::$_statisticsFile, $worker_status_str, FILE_APPEND);
    }
1306951
protected function getRunningWorkerPids(){
        $workerPids = [];
        $workerPidsInRedis = $this->redisClientForStatics->keys(self::worker_status_monitoring_key_prefix.'*');
        if (!empty($workerPidsInRedis)) {
            foreach ($workerPidsInRedis as $workerPidKey){
                $workerPids[] = intval( str_replace( self::worker_status_monitoring_key_prefix , '' ,$workerPidKey ) );
            }
        }
        return $workerPids;
    }
740272
public function status() {
		$timeout = Config::defaultworkertimeout();
		$thresholdTime = (new FrozenTime())->subSeconds($timeout);

		$results = $this->find()
			->where(['modified >' => $thresholdTime])
			->orderDesc('modified')
			->enableHydration(false)
			->all()
			->toArray();

		if (!$results) {
			return [];
		}

		$count = count($results);
		$record = array_shift($results);
		/** @var \Cake\I18n\FrozenTime $time */
		$time = $record['modified'];

		return [
			'time' => $time,
			'workers' => $count,
		];
	}
576325
private function pull(): Worker
    {
        if (!$this->isRunning()) {
            throw new StatusError("The pool was shutdown");
        }

        do {
            if ($this->idleWorkers->isEmpty()) {
                if ($this->getWorkerCount() >= $this->maxSize) {
                    // All possible workers busy, so shift from head (will be pushed back onto tail below).
                    $worker = $this->busyQueue->shift();
                } else {
                    // Max worker count has not been reached, so create another worker.
                    $worker = $this->factory->create();
                    $this->workers->attach($worker, 0);
                    break;
                }
            } else {
                // Shift a worker off the idle queue.
                $worker = $this->idleWorkers->shift();
            }

            if ($worker->isRunning()) {
                break;
            }

            $this->workers->detach($worker);
        } while (true);

        $this->busyQueue->push($worker);
        $this->workers[$worker] += 1;

        return $worker;
    }