<?php
/**
- * @copyright Copyright (C) 2020, Friendica
+ * @copyright Copyright (C) 2010-2021, the Friendica project
*
* @license GNU AGPL version 3 or any later version
*
$last_check = $starttime = time();
self::$state = self::STATE_STARTUP;
- $wait_interval = self::isDaemonMode() ? 360 : 10;
- $start = time();
-
- do {
- // We fetch the next queue entry that is about to be executed
- while ($r = self::workerProcess()) {
- // Don't refetch when a worker fetches tasks for multiple workers
- $refetched = DI::config()->get('system', 'worker_multiple_fetch');
- foreach ($r as $entry) {
- // Assure that the priority is an integer value
- $entry['priority'] = (int)$entry['priority'];
-
- // The work will be done
- if (!self::execute($entry)) {
- Logger::notice('Process execution failed, quitting.');
- return;
- }
- // Trying to fetch new processes - but only once when successful
- if (!$refetched && DI::lock()->acquire(self::LOCK_PROCESS, 0)) {
- self::findWorkerProcesses();
- DI::lock()->release(self::LOCK_PROCESS);
- self::$state = self::STATE_REFETCH;
- $refetched = true;
- } else {
- self::$state = self::STATE_SHORT_LOOP;
- }
- }
+ // We fetch the next queue entry that is about to be executed
+ while ($r = self::workerProcess()) {
+ if (self::IPCJobsExists(getmypid())) {
+ self::IPCDeleteJobState(getmypid());
+ }
- // To avoid the quitting of multiple workers only one worker at a time will execute the check
- if ((time() > $last_check + 5) && !self::getWaitingJobForPID()) {
- self::$state = self::STATE_LONG_LOOP;
-
- if (DI::lock()->acquire(self::LOCK_WORKER, 0)) {
- // Count active workers and compare them with a maximum value that depends on the load
- if (self::tooMuchWorkers()) {
- Logger::notice('Active worker limit reached, quitting.');
- DI::lock()->release(self::LOCK_WORKER);
- return;
- }
-
- // Check free memory
- if (DI::process()->isMinMemoryReached()) {
- Logger::warning('Memory limit reached, quitting.');
- DI::lock()->release(self::LOCK_WORKER);
- return;
- }
- DI::lock()->release(self::LOCK_WORKER);
- }
- $last_check = time();
- }
+ // Don't refetch when a worker fetches tasks for multiple workers
+ $refetched = DI::config()->get('system', 'worker_multiple_fetch');
+ foreach ($r as $entry) {
+ // Assure that the priority is an integer value
+ $entry['priority'] = (int)$entry['priority'];
- // Quit the worker once every cron interval
- if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
- Logger::info('Process lifetime reached, respawning.');
- self::unclaimProcess();
- if (self::isDaemonMode()) {
- self::IPCSetJobState(true);
- } else {
- self::spawnWorker();
- }
+ // The work will be done
+ if (!self::execute($entry)) {
+ Logger::notice('Process execution failed, quitting.');
return;
}
- $start = time();
- }
- $seconds = (time() - $start);
+ // Trying to fetch new processes - but only once when successful
+ if (!$refetched && DI::lock()->acquire(self::LOCK_PROCESS, 0)) {
+ self::findWorkerProcesses();
+ DI::lock()->release(self::LOCK_PROCESS);
+ self::$state = self::STATE_REFETCH;
+ $refetched = true;
+ } else {
+ self::$state = self::STATE_SHORT_LOOP;
+ }
+ }
- // logarithmic wait time calculation.
- $arg = (($seconds + 1) / ($wait_interval / 9)) + 1;
- $sleep = min(1000000, round(log10($arg) * 1000000, 0));
- usleep($sleep);
+ // To avoid the quitting of multiple workers only one worker at a time will execute the check
+ if ((time() > $last_check + 5) && !self::getWaitingJobForPID()) {
+ self::$state = self::STATE_LONG_LOOP;
- $timeout = ($seconds >= $wait_interval);
- Logger::info('Timeout', ['timeout' => $timeout, 'seconds' => $seconds, 'sleep' => $sleep]);
+ if (DI::lock()->acquire(self::LOCK_WORKER, 0)) {
+ // Count active workers and compare them with a maximum value that depends on the load
+ if (self::tooMuchWorkers()) {
+ Logger::notice('Active worker limit reached, quitting.');
+ DI::lock()->release(self::LOCK_WORKER);
+ return;
+ }
- if (!$timeout) {
- if (DI::process()->isMaxLoadReached()) {
- Logger::notice('maximum load reached, quitting.');
- return;
+ // Check free memory
+ if (DI::process()->isMinMemoryReached()) {
+ Logger::warning('Memory limit reached, quitting.');
+ DI::lock()->release(self::LOCK_WORKER);
+ return;
+ }
+ DI::lock()->release(self::LOCK_WORKER);
}
+ $last_check = time();
+ }
- // Kill stale processes every 5 minutes
- $last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0);
- if (time() > ($last_cleanup + 300)) {
- DI::config()->set('system', 'worker_last_cleaned', time());
- self::killStaleWorkers();
+ // Quit the worker once every cron interval
+ if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
+ Logger::info('Process lifetime reached, respawning.');
+ self::unclaimProcess();
+ if (self::isDaemonMode()) {
+ self::IPCSetJobState(true);
+ } else {
+ self::spawnWorker();
}
-
- // Check if the system is ready
- if (!self::isReady()) {
- return;
- }
+ return;
}
- } while (!$timeout);
+ }
// Cleaning up. Possibly not needed, but it doesn't harm anything.
if (self::isDaemonMode()) {
}
$argv = json_decode($queue['parameter'], true);
+ if (!is_array($argv)) {
+ $argv = [];
+ }
+
if (!empty($queue['command'])) {
array_unshift($argv, $queue['command']);
}
self::$db_duration_write += (microtime(true) - $stamp);
}
- /**
- * Call the front end worker
- *
- * @return void
- * @throws \Friendica\Network\HTTPException\InternalServerErrorException
- */
- public static function callWorker()
- {
- if (!DI::config()->get("system", "frontend_worker")) {
- return;
- }
-
- $url = DI::baseUrl() . '/worker';
- DI::httpRequest()->fetch($url, 1);
- }
-
- /**
- * Call the front end worker if there aren't any active
- *
- * @return void
- * @throws \Friendica\Network\HTTPException\InternalServerErrorException
- */
- public static function executeIfIdle()
- {
- self::checkDaemonState();
-
- if (!DI::config()->get("system", "frontend_worker")) {
- return;
- }
-
- // Do we have "proc_open"? Then we can fork the worker
- if (function_exists("proc_open")) {
- // When was the last time that we called the worker?
- // Less than one minute? Then we quit
- if ((time() - DI::config()->get("system", "worker_started")) < 60) {
- return;
- }
-
- DI::config()->set("system", "worker_started", time());
-
- // Do we have enough running workers? Then we quit here.
- if (self::tooMuchWorkers()) {
- // Cleaning dead processes
- self::killStaleWorkers();
- DI::modelProcess()->deleteInactive();
-
- return;
- }
-
- self::runCron();
-
- Logger::info('Call worker');
- self::spawnWorker();
- return;
- }
-
- // We cannot execute background processes.
- // We now run the processes from the frontend.
- // This won't work with long running processes.
- self::runCron();
-
- self::clearProcesses();
-
- $workers = self::activeWorkers();
-
- if ($workers == 0) {
- self::callWorker();
- }
- }
-
- /**
- * Removes long running worker processes
- *
- * @return void
- * @throws \Friendica\Network\HTTPException\InternalServerErrorException
- */
- public static function clearProcesses()
- {
- $timeout = DI::config()->get("system", "frontend_worker_timeout", 10);
-
- /// @todo We should clean up the corresponding workerqueue entries as well
- $stamp = (float)microtime(true);
- $condition = ["`created` < ? AND `command` = 'worker.php'",
- DateTimeFormat::utc("now - ".$timeout." minutes")];
- DBA::delete('process', $condition);
- self::$db_duration = (microtime(true) - $stamp);
- self::$db_duration_write += (microtime(true) - $stamp);
- }
-
/**
* Runs the cron processes
*
} elseif ($pid) {
// The parent process continues here
DBA::connect();
- Logger::info('Spawned new worker', ['cron' => $do_cron, 'pid' => $pid]);
+
+ self::IPCSetJobState(true, $pid);
+ Logger::info('Spawned new worker', ['pid' => $pid]);
+
+ $cycles = 0;
+ while (self::IPCJobsExists($pid) && (++$cycles < 100)) {
+ usleep(10000);
+ }
+
+ Logger::info('Spawned worker is ready', ['pid' => $pid, 'wait_cycles' => $cycles]);
return;
}
// We now are in the new worker
+ $pid = getmypid();
+
DBA::connect();
- Logger::info('Worker spawned', ['cron' => $do_cron, 'pid' => getmypid()]);
+ /// @todo Reinitialize the logger to set a new process_id and uid
+ DI::process()->setPid($pid);
- DI::process()->start();
+ $cycles = 0;
+ while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
+ usleep(10000);
+ }
+
+ Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
self::processQueue($do_cron);
self::unclaimProcess();
+ self::IPCSetJobState(false, $pid);
DI::process()->end();
- Logger::info('Worker ended', ['cron' => $do_cron, 'pid' => getmypid()]);
-
- DBA::disconnect();
-/*
- $php = '/usr/bin/php';
- $param = ['bin/worker.php'];
- if ($do_cron) {
- $param[] = 'no_cron';
- }
- pcntl_exec($php, $param);
- Logger::warning('Error calling worker', ['cron' => $do_cron, 'pid' => getmypid()]);
-*/
+ Logger::info('Worker ended', ['pid' => $pid]);
exit();
}
*/
public static function spawnWorker($do_cron = false)
{
- if (self::isDaemonMode()) {
+ if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) {
self::forkProcess($do_cron);
- self::IPCSetJobState(false);
} else {
$process = new Core\Process(DI::logger(), DI::mode(), DI::config(),
DI::modelProcess(), DI::app()->getBasePath(), getmypid());
$process->run('bin/worker.php', ['no_cron' => !$do_cron]);
}
+ if (self::isDaemonMode()) {
+ self::IPCSetJobState(false);
+ }
}
/**
* Set the flag if some job is waiting
*
* @param boolean $jobs Is there a waiting job?
+ * @param int $key Key number
+ * @throws \Exception
+ */
+ public static function IPCSetJobState(bool $jobs, int $key = 0)
+ {
+ $stamp = (float)microtime(true);
+ DBA::replace('worker-ipc', ['jobs' => $jobs, 'key' => $key]);
+ self::$db_duration += (microtime(true) - $stamp);
+ self::$db_duration_write += (microtime(true) - $stamp);
+ }
+
+ /**
+ * Delete a key entry
+ *
+ * @param int $key Key number
* @throws \Exception
*/
- public static function IPCSetJobState($jobs)
+ public static function IPCDeleteJobState(int $key)
{
$stamp = (float)microtime(true);
- DBA::update('worker-ipc', ['jobs' => $jobs], ['key' => 1], true);
+ DBA::delete('worker-ipc', ['key' => $key]);
self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_write += (microtime(true) - $stamp);
}
/**
* Checks if some worker job waits to be executed
*
+ * @param int $key Key number
* @return bool
* @throws \Exception
*/
- public static function IPCJobsExists()
+ public static function IPCJobsExists(int $key = 0)
{
$stamp = (float)microtime(true);
- $row = DBA::selectFirst('worker-ipc', ['jobs'], ['key' => 1]);
+ $row = DBA::selectFirst('worker-ipc', ['jobs'], ['key' => $key]);
self::$db_duration += (microtime(true) - $stamp);
// When we don't have a row, no job is running
projects / friendica.git / blobdiff