*/
namespace Friendica\Core;
-use Friendica\BaseObject;
+use Friendica\Core;
use Friendica\Database\DBA;
+use Friendica\DI;
use Friendica\Model\Process;
use Friendica\Util\DateTimeFormat;
use Friendica\Util\Network;
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;
}
}
// Check free memory
- if ($a->isMinMemoryReached()) {
+ if (DI::process()->isMinMemoryReached()) {
Logger::log('Memory limit reached, quitting.', Logger::DEBUG);
Lock::release('worker');
return;
*/
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;
}
// 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);
$waiting_processes = 0;
// Now adding all processes with workerqueue entries
$stamp = (float)microtime(true);
- $jobs = DBA::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` WHERE NOT `done` AND `next_try` < ? GROUP BY `priority`", DateTimeFormat::utcNow());
+ $jobs = DBA::p("SELECT COUNT(*) AS `entries`, `priority` FROM `workerqueue` WHERE NOT `done` GROUP BY `priority`");
self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_stat += (microtime(true) - $stamp);
while ($entry = DBA::fetch($jobs)) {
DBA::close($jobs);
}
+ $waiting_processes -= $deferred;
+
$listitem[0] = "0:" . max(0, $idle_workers);
$processlist .= ' ('.implode(', ', $listitem).')';
return;
}
- $url = System::baseUrl()."/worker";
+ $url = DI::baseUrl() . '/worker';
Network::fetchUrl($url, false, 1);
}
$args = ['no_cron' => !$do_cron];
- get_app()->proc_run($command, $args);
+ $a = get_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.
if (Config::get('system', 'worker_daemon_mode', false)) {
* or: Worker::add(PRIORITY_HIGH, "Notifier", Delivery::DELETION, $drop_id);
* or: Worker::add(array('priority' => PRIORITY_HIGH, 'dont_fork' => true), "CreateShadowEntry", $post_id);
*
- * @return boolean "false" if proc_run couldn't be executed
+ * @return boolean "false" if worker queue entry already existed or there had been an error
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
* @note $cmd and string args are surrounded with ""
*
$priority = PRIORITY_MEDIUM;
// Don't fork from frontend tasks by default
- $dont_fork = Config::get("system", "worker_dont_fork", false) || !\get_app()->isBackend();
+ $dont_fork = Config::get("system", "worker_dont_fork", false) || !DI::mode()->isBackend();
$created = DateTimeFormat::utcNow();
$force_priority = false;
$parameters = json_encode($args);
$found = DBA::exists('workerqueue', ['parameter' => $parameters, 'done' => false]);
+ $added = false;
// Quit if there was a database error - a precaution for the update process to 3.5.3
if (DBA::errorNo() != 0) {
}
if (!$found) {
- DBA::insert('workerqueue', ['parameter' => $parameters, 'created' => $created, 'priority' => $priority]);
+ $added = DBA::insert('workerqueue', ['parameter' => $parameters, 'created' => $created, 'priority' => $priority]);
+ if (!$added) {
+ return false;
+ }
} elseif ($force_priority) {
DBA::update('workerqueue', ['priority' => $priority], ['parameter' => $parameters, 'done' => false, 'pid' => 0]);
}
// Should we quit and wait for the worker to be called as a cronjob?
if ($dont_fork) {
- return true;
+ return $added;
}
// If there is a lock then we don't have to check for too much worker
if (!Lock::acquire('worker', 0)) {
- return true;
+ return $added;
}
// If there are already enough workers running, don't fork another one
Lock::release('worker');
if ($quit) {
- return true;
+ return $added;
}
// We tell the daemon that a new job entry exists
if (Config::get('system', 'worker_daemon_mode', false)) {
// We don't have to set the IPC flag - this is done in "tooMuchWorkers"
- return true;
+ return $added;
}
// Now call the worker to execute the jobs that we just added to the queue
self::spawnWorker();
- return true;
+ return $added;
}
/**
/**
* Defers the current worker entry
+ * @return boolean had the entry been deferred?
*/
public static function defer()
{
- if (empty(BaseObject::getApp()->queue)) {
- return;
+ if (empty(DI::app()->queue)) {
+ return false;
}
- $queue = BaseObject::getApp()->queue;
+ $queue = DI::app()->queue;
$retrial = $queue['retrial'];
$id = $queue['id'];
$new_retrial = self::getNextRetrial($queue, $max_level);
if ($new_retrial > $max_level) {
- Logger::info('The task exceeded the maximum retry count', ['id' => $id, 'max_level' => $max_level, 'retrial' => $new_retrial]);
- return;
+ Logger::info('The task exceeded the maximum retry count', ['id' => $id, 'created' => $queue['created'], 'old_prio' => $queue['priority'], 'old_retrial' => $queue['retrial'], 'max_level' => $max_level, 'retrial' => $new_retrial]);
+ return false;
}
// Calculate the delay until the next trial
$priority = PRIORITY_NEGLIGIBLE;
}
- Logger::info('Deferred task', ['id' => $id, 'retrial' => $new_retrial, 'next_execution' => $next, 'old_prio' => $queue['priority'], 'new_prio' => $priority]);
+ Logger::info('Deferred task', ['id' => $id, 'retrial' => $new_retrial, 'created' => $queue['created'], 'next_execution' => $next, 'old_prio' => $queue['priority'], 'new_prio' => $priority]);
$stamp = (float)microtime(true);
$fields = ['retrial' => $new_retrial, 'next_try' => $next, 'executed' => DBA::NULL_DATETIME, 'pid' => 0, 'priority' => $priority];
DBA::update('workerqueue', $fields, ['id' => $id]);
self::$db_duration += (microtime(true) - $stamp);
self::$db_duration_write += (microtime(true) - $stamp);
+
+ return true;
}
/**
projects / friendica.git / blobdiff