<?php
/**
- * @copyright Copyright (C) 2020, Friendica
+ * @copyright Copyright (C) 2010-2021, the Friendica project
*
* @license GNU AGPL version 3 or any later version
*
// We fetch the next queue entry that is about to be executed
while ($r = self::workerProcess()) {
if (self::IPCJobsExists(getmypid())) {
- self::IPCSetJobState(false, getmypid());
+ self::IPCDeleteJobState(getmypid());
}
+
// Don't refetch when a worker fetches tasks for multiple workers
$refetched = DI::config()->get('system', 'worker_multiple_fetch');
foreach ($r as $entry) {
}
$argv = json_decode($queue['parameter'], true);
+ if (!is_array($argv)) {
+ $argv = [];
+ }
+
if (!empty($queue['command'])) {
array_unshift($argv, $queue['command']);
}
$queue['priority'] = PRIORITY_MEDIUM;
}
- $a->queue = $queue;
+ $a->setQueue($queue);
$up_duration = microtime(true) - self::$up_start;
Logger::disableWorker();
- unset($a->queue);
+ $a->setQueue([]);
$duration = (microtime(true) - $stamp);
$stamp = (float)microtime(true);
$queues = DBA::p("SELECT `process`.`pid`, COUNT(`workerqueue`.`pid`) AS `entries` FROM `process`
- LEFT JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `workerqueue`.`done`
+ LEFT JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `workerqueue`.`done`
GROUP BY `process`.`pid`");
while ($queue = DBA::fetch($queues)) {
$ids[$queue['pid']] = $queue['entries'];
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', ['pid' => $pid]);
+
self::IPCSetJobState(true, $pid);
+ Logger::info('Spawned new worker', ['pid' => $pid]);
$cycles = 0;
while (self::IPCJobsExists($pid) && (++$cycles < 100)) {
}
// We now are in the new worker
+ $pid = getmypid();
+
DBA::connect();
- Logger::info('Worker spawned', ['pid' => getmypid()]);
+ /// @todo Reinitialize the logger to set a new process_id and uid
+ DI::process()->setPid($pid);
$cycles = 0;
while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
usleep(10000);
}
- Logger::info('Parent is ready', ['pid' => getmypid(), 'wait_cycles' => $cycles]);
+ Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
self::processQueue($do_cron);
self::unclaimProcess();
- self::IPCSetJobState(false, getmypid());
+ self::IPCSetJobState(false, $pid);
DI::process()->end();
- Logger::info('Worker ended', ['pid' => getmypid()]);
+ Logger::info('Worker ended', ['pid' => $pid]);
exit();
}
* or: Worker::add(PRIORITY_HIGH, "Notifier", Delivery::DELETION, $drop_id);
* or: Worker::add(array('priority' => PRIORITY_HIGH, 'dont_fork' => true), "Delivery", $post_id);
*
- * @return boolean "false" if worker queue entry already existed or there had been an error
+ * @return int "0" if worker queue entry already existed or there had been an error, otherwise the ID of the worker task
* @throws \Friendica\Network\HTTPException\InternalServerErrorException
* @note $cmd and string args are surrounded with ""
*
$args = func_get_args();
if (!count($args)) {
- return false;
+ return 0;
}
$arr = ['args' => $args, 'run_cmd' => true];
Hook::callAll("proc_run", $arr);
if (!$arr['run_cmd'] || !count($args)) {
- return true;
+ return 1;
}
$priority = PRIORITY_MEDIUM;
$command = array_shift($args);
$parameters = json_encode($args);
$found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
- $added = false;
+ $added = 0;
if (!in_array($priority, PRIORITIES)) {
Logger::warning('Invalid priority', ['priority' => $priority, 'command' => $command, 'callstack' => System::callstack(20)]);
// Quit if there was a database error - a precaution for the update process to 3.5.3
if (DBA::errorNo() != 0) {
- return false;
+ return 0;
}
if (!$found) {
- $added = DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
- 'priority' => $priority, 'next_try' => $delayed]);
- if (!$added) {
- return false;
+ if (!DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
+ 'priority' => $priority, 'next_try' => $delayed])) {
+ return 0;
}
+ $added = DBA::lastInsertId();
} elseif ($force_priority) {
DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
}
*/
public static function defer()
{
- if (empty(DI::app()->queue)) {
+ $queue = DI::app()->getQueue();
+
+ if (empty($queue)) {
return false;
}
- $queue = DI::app()->queue;
-
$retrial = $queue['retrial'];
$id = $queue['id'];
$priority = $queue['priority'];
* 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)
self::$db_duration_write += (microtime(true) - $stamp);
}
+ /**
+ * Delete a key entry
+ *
+ * @param int $key Key number
+ * @throws \Exception
+ */
+ public static function IPCDeleteJobState(int $key)
+ {
+ $stamp = (float)microtime(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
*/
} else {
Logger::info('We are outside the maintenance window', ['current' => date('H:i:s', $current), 'start' => date('H:i:s', $start), 'end' => date('H:i:s', $end)]);
}
-
+
return $execute;
}
}