/**
* @file src/Core/Worker.php
*
- * @brief Contains the class for the worker background job processing
+ * Contains the class for the worker background job processing
*/
/**
- * @brief Worker methods
+ * Worker methods
*/
class Worker
{
private static $state;
/**
- * @brief Processes the tasks that are in the workerqueue table
+ * Processes the tasks that are in the workerqueue table
*
* @param boolean $run_cron Should the cron processes be executed?
* @return void
*/
public static function processQueue($run_cron = true)
{
- $a = \get_app();
-
// Ensure that all "strtotime" operations do run timezone independent
date_default_timezone_set('UTC');
self::$up_start = microtime(true);
// At first check the maximum load. We shouldn't continue with a high load
- if ($a->isMaxLoadReached()) {
+ if (DI::process()->isMaxLoadReached()) {
Logger::log('Pre check: maximum load reached, quitting.', Logger::DEBUG);
return;
}
}
// Do we have too few memory?
- if ($a->isMinMemoryReached()) {
+ if (DI::process()->isMinMemoryReached()) {
Logger::log('Pre check: Memory limit reached, quitting.', Logger::DEBUG);
return;
}
}
// Possibly there are too much database processes that block the system
- if ($a->isMaxProcessesReached()) {
+ if (DI::process()->isMaxProcessesReached()) {
Logger::log('Pre check: maximum processes reached, quitting.', Logger::DEBUG);
return;
}
}
// Trying to fetch new processes - but only once when successful
- if (!$refetched && Lock::acquire('worker_process', 0)) {
+ if (!$refetched && DI::lock()->acquire('worker_process', 0)) {
self::findWorkerProcesses();
- Lock::release('worker_process');
+ DI::lock()->release('worker_process');
self::$state = self::STATE_REFETCH;
$refetched = true;
} else {
if (!self::getWaitingJobForPID()) {
self::$state = self::STATE_LONG_LOOP;
- if (Lock::acquire('worker', 0)) {
+ if (DI::lock()->acquire('worker', 0)) {
// Count active workers and compare them with a maximum value that depends on the load
if (self::tooMuchWorkers()) {
Logger::log('Active worker limit reached, quitting.', Logger::DEBUG);
- Lock::release('worker');
+ DI::lock()->release('worker');
return;
}
// Check free memory
- if ($a->isMinMemoryReached()) {
+ if (DI::process()->isMinMemoryReached()) {
Logger::log('Memory limit reached, quitting.', Logger::DEBUG);
- Lock::release('worker');
+ DI::lock()->release('worker');
return;
}
- Lock::release('worker');
+ DI::lock()->release('worker');
}
}
}
/**
- * @brief Check if non executed tasks do exist in the worker queue
+ * Check if non executed tasks do exist in the worker queue
*
* @return boolean Returns "true" if tasks are existing
* @throws \Exception
}
/**
- * @brief Returns the number of deferred entries in the worker queue
+ * Returns the number of deferred entries in the worker queue
*
* @return integer Number of deferred entries in the worker queue
* @throws \Exception
}
/**
- * @brief Returns the number of non executed entries in the worker queue
+ * Returns the number of non executed entries in the worker queue
*
* @return integer Number of non executed entries in the worker queue
* @throws \Exception
}
/**
- * @brief Returns the highest priority in the worker queue that isn't executed
+ * Returns the highest priority in the worker queue that isn't executed
*
* @return integer Number of active worker processes
* @throws \Exception
}
/**
- * @brief Returns if a process with the given priority is running
+ * Returns if a process with the given priority is running
*
* @param integer $priority The priority that should be checked
*
}
/**
- * @brief Execute a worker entry
+ * Execute a worker entry
*
* @param array $queue Workerqueue entry
*
*/
public static function execute($queue)
{
- $a = \get_app();
-
$mypid = getmypid();
// Quit when in maintenance
}
// Constantly check the number of parallel database processes
- if ($a->isMaxProcessesReached()) {
+ if (DI::process()->isMaxProcessesReached()) {
Logger::log("Max processes reached for process ".$mypid, Logger::DEBUG);
return false;
}
}
/**
- * @brief Execute a function from the queue
+ * Execute a function from the queue
*
* @param array $queue Workerqueue entry
* @param string $funcname name of the function
*/
private static function execFunction($queue, $funcname, $argv, $method_call)
{
- $a = \get_app();
+ $a = DI::app();
$argc = count($argv);
// We use the callstack here to analyze the performance of executed worker entries.
// For this reason the variables have to be initialized.
- $a->getProfiler()->reset();
+ DI::profiler()->reset();
$a->queue = $queue;
Logger::info('Process done.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration, 3)]);
- $a->getProfiler()->saveLog($a->getLogger(), "ID " . $queue["id"] . ": " . $funcname);
+ DI::profiler()->saveLog(DI::logger(), "ID " . $queue["id"] . ": " . $funcname);
$cooldown = Config::get("system", "worker_cooldown", 0);
}
/**
- * @brief Checks if the number of database connections has reached a critical limit.
+ * Checks if the number of database connections has reached a critical limit.
*
* @return bool Are more than 3/4 of the maximum connections used?
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
}
/**
- * @brief fix the queue entry if the worker process died
+ * fix the queue entry if the worker process died
* @return void
* @throws \Exception
*/
}
/**
- * @brief Checks if the number of active workers exceeds the given limits
+ * Checks if the number of active workers exceeds the given limits
*
* @return bool Are there too much workers running?
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
}
/**
- * @brief Returns the number of active worker processes
+ * Returns the number of active worker processes
*
* @return integer Number of active worker processes
* @throws \Exception
}
/**
- * @brief Returns waiting jobs for the current process id
+ * Returns waiting jobs for the current process id
*
* @return array waiting workerqueue jobs
* @throws \Exception
}
/**
- * @brief Returns the next jobs that should be executed
+ * Returns the next jobs that should be executed
*
* @return array array with next jobs
* @throws \Exception
}
/**
- * @brief Returns the priority of the next workerqueue job
+ * Returns the priority of the next workerqueue job
*
* @return string priority
* @throws \Exception
}
/**
- * @brief Find and claim the next worker process for us
+ * Find and claim the next worker process for us
*
* @return boolean Have we found something?
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
}
/**
- * @brief Returns the next worker process
+ * Returns the next worker process
*
* @return string SQL statement
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
}
$stamp = (float)microtime(true);
- if (!Lock::acquire('worker_process')) {
+ if (!DI::lock()->acquire('worker_process')) {
return false;
}
self::$lock_duration += (microtime(true) - $stamp);
$found = self::findWorkerProcesses();
- Lock::release('worker_process');
+ DI::lock()->release('worker_process');
if ($found) {
$stamp = (float)microtime(true);
}
/**
- * @brief Removes a workerqueue entry from the current process
+ * Removes a workerqueue entry from the current process
* @return void
* @throws \Exception
*/
}
/**
- * @brief Call the front end worker
+ * Call the front end worker
* @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
*/
return;
}
- $url = System::baseUrl()."/worker";
+ $url = DI::baseUrl() . '/worker';
Network::fetchUrl($url, false, 1);
}
/**
- * @brief Call the front end worker if there aren't any active
+ * Call the front end worker if there aren't any active
* @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
*/
}
/**
- * @brief Removes long running worker processes
+ * Removes long running worker processes
* @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
*/
}
/**
- * @brief Runs the cron processes
+ * Runs the cron processes
* @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
*/
}
/**
- * @brief Spawns a new worker
+ * Spawns a new worker
* @param bool $do_cron
* @return void
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
$args = ['no_cron' => !$do_cron];
- $a = get_app();
- $process = new Core\Process($a->getLogger(), $a->getMode(), $a->getConfig(), $a->getBasePath());
+ $a = DI::app();
+ $process = new Core\Process(DI::logger(), DI::mode(), DI::config(), $a->getBasePath());
$process->run($command, $args);
// after spawning we have to remove the flag.
}
/**
- * @brief Adds tasks to the worker queue
+ * Adds tasks to the worker queue
*
* @param (integer|array) priority or parameter array, strings are deprecated and are ignored
*
$priority = PRIORITY_MEDIUM;
// Don't fork from frontend tasks by default
- $dont_fork = Config::get("system", "worker_dont_fork", false) || !\get_app()->getMode()->isBackend();
+ $dont_fork = Config::get("system", "worker_dont_fork", false) || !DI::mode()->isBackend();
$created = DateTimeFormat::utcNow();
$force_priority = false;
}
// If there is a lock then we don't have to check for too much worker
- if (!Lock::acquire('worker', 0)) {
+ if (!DI::lock()->acquire('worker', 0)) {
return $added;
}
// If there are already enough workers running, don't fork another one
$quit = self::tooMuchWorkers();
- Lock::release('worker');
+ DI::lock()->release('worker');
if ($quit) {
return $added;
/**
* Log active processes into the "process" table
- *
- * @brief Log active processes into the "process" table
*/
public static function startProcess()
{
/**
* Remove the active process from the "process" table
*
- * @brief Remove the active process from the "process" table
* @return bool
* @throws \Exception
*/
/**
* Set the flag if some job is waiting
*
- * @brief Set the flag if some job is waiting
* @param boolean $jobs Is there a waiting job?
* @throws \Exception
*/
/**
* Checks if some worker job waits to be executed
*
- * @brief Checks if some worker job waits to be executed
* @return bool
* @throws \Exception
*/
projects / friendica.git / blobdiff