]> git.mxchange.org Git - friendica.git/blobdiff - src/Core/Worker.php
Merge pull request #10557 from annando/delayed
[friendica.git] / src / Core / Worker.php
index 3083eacbfe8483c7b373924867c1df0400292d38..307451c05205eb7f811bccf4bcab7f160776dda9 100644 (file)
@@ -1,6 +1,6 @@
 <?php
 /**
- * @copyright Copyright (C) 2020, Friendica
+ * @copyright Copyright (C) 2010-2021, the Friendica project
  *
  * @license GNU AGPL version 3 or any later version
  *
@@ -21,6 +21,7 @@
 
 namespace Friendica\Core;
 
+use Friendica\App\Mode;
 use Friendica\Core;
 use Friendica\Database\DBA;
 use Friendica\DI;
@@ -49,6 +50,7 @@ class Worker
        private static $lock_duration = 0;
        private static $last_update;
        private static $state;
+       private static $daemon_mode = null;
 
        /**
         * Processes the tasks that are in the workerqueue table
@@ -95,6 +97,10 @@ class Worker
 
                // We fetch the next queue entry that is about to be executed
                while ($r = self::workerProcess()) {
+                       if (self::IPCJobsExists(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) {
@@ -103,7 +109,7 @@ class Worker
 
                                // The work will be done
                                if (!self::execute($entry)) {
-                                       Logger::info('Process execution failed, quitting.');
+                                       Logger::notice('Process execution failed, quitting.');
                                        return;
                                }
 
@@ -125,14 +131,14 @@ class Worker
                                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::info('Active worker limit reached, quitting.');
+                                               Logger::notice('Active worker limit reached, quitting.');
                                                DI::lock()->release(self::LOCK_WORKER);
                                                return;
                                        }
 
                                        // Check free memory
                                        if (DI::process()->isMinMemoryReached()) {
-                                               Logger::notice('Memory limit reached, quitting.');
+                                               Logger::warning('Memory limit reached, quitting.');
                                                DI::lock()->release(self::LOCK_WORKER);
                                                return;
                                        }
@@ -145,13 +151,17 @@ class Worker
                        if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
                                Logger::info('Process lifetime reached, respawning.');
                                self::unclaimProcess();
-                               self::spawnWorker();
+                               if (self::isDaemonMode()) {
+                                       self::IPCSetJobState(true);
+                               } else {
+                                       self::spawnWorker();
+                               }
                                return;
                        }
                }
 
                // Cleaning up. Possibly not needed, but it doesn't harm anything.
-               if (DI::config()->get('system', 'worker_daemon_mode', false)) {
+               if (self::isDaemonMode()) {
                        self::IPCSetJobState(false);
                }
                Logger::info("Couldn't select a workerqueue entry, quitting process", ['pid' => getmypid()]);
@@ -168,28 +178,28 @@ class Worker
        {
                // Count active workers and compare them with a maximum value that depends on the load
                if (self::tooMuchWorkers()) {
-                       Logger::info('Active worker limit reached, quitting.');
+                       Logger::notice('Active worker limit reached, quitting.');
                        return false;
                }
 
                // Do we have too few memory?
                if (DI::process()->isMinMemoryReached()) {
-                       Logger::notice('Memory limit reached, quitting.');
+                       Logger::warning('Memory limit reached, quitting.');
                        return false;
                }
 
                // Possibly there are too much database connections
                if (self::maxConnectionsReached()) {
-                       Logger::notice('Maximum connections reached, quitting.');
+                       Logger::warning('Maximum connections reached, quitting.');
                        return false;
                }
 
                // Possibly there are too much database processes that block the system
                if (DI::process()->isMaxProcessesReached()) {
-                       Logger::notice('Maximum processes reached, quitting.');
+                       Logger::warning('Maximum processes reached, quitting.');
                        return false;
                }
-               
+
                return true;
        }
 
@@ -290,17 +300,25 @@ class Worker
 
                // Constantly check the number of parallel database processes
                if (DI::process()->isMaxProcessesReached()) {
-                       Logger::notice("Max processes reached for process", ['pid' => $mypid]);
+                       Logger::warning("Max processes reached for process", ['pid' => $mypid]);
                        return false;
                }
 
                // Constantly check the number of available database connections to let the frontend be accessible at any time
                if (self::maxConnectionsReached()) {
-                       Logger::notice("Max connection reached for process", ['pid' => $mypid]);
+                       Logger::warning("Max connection reached for process", ['pid' => $mypid]);
                        return false;
                }
 
-               $argv = json_decode($queue["parameter"], true);
+               $argv = json_decode($queue['parameter'], true);
+               if (!is_array($argv)) {
+                       $argv = [];
+               }
+
+               if (!empty($queue['command'])) {
+                       array_unshift($argv, $queue['command']);
+               }
+
                if (empty($argv)) {
                        Logger::warning('Parameter is empty', ['queue' => $queue]);
                        return false;
@@ -407,6 +425,12 @@ class Worker
        {
                $a = DI::app();
 
+               $cooldown = DI::config()->get("system", "worker_cooldown", 0);
+               if ($cooldown > 0) {
+                       Logger::info('Pre execution cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
+                       sleep($cooldown);
+               }
+
                Logger::enableWorker($funcname);
 
                Logger::info("Process start.", ['priority' => $queue["priority"], 'id' => $queue["id"]]);
@@ -417,7 +441,12 @@ class Worker
                // For this reason the variables have to be initialized.
                DI::profiler()->reset();
 
-               $a->queue = $queue;
+               if (!in_array($queue['priority'], PRIORITIES)) {
+                       Logger::warning('Invalid priority', ['queue' => $queue, 'callstack' => System::callstack(20)]);
+                       $queue['priority'] = PRIORITY_MEDIUM;
+               }
+
+               $a->setQueue($queue);
 
                $up_duration = microtime(true) - self::$up_start;
 
@@ -433,7 +462,7 @@ class Worker
 
                Logger::disableWorker();
 
-               unset($a->queue);
+               $a->setQueue([]);
 
                $duration = (microtime(true) - $stamp);
 
@@ -474,10 +503,8 @@ class Worker
 
                DI::profiler()->saveLog(DI::logger(), "ID " . $queue["id"] . ": " . $funcname);
 
-               $cooldown = DI::config()->get("system", "worker_cooldown", 0);
-
                if ($cooldown > 0) {
-                       Logger::info('Cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
+                       Logger::info('Post execution cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
                        sleep($cooldown);
                }
        }
@@ -531,7 +558,7 @@ class Worker
                        $level = ($used / $max) * 100;
 
                        if ($level >= $maxlevel) {
-                               Logger::notice("Maximum level (".$maxlevel."%) of user connections reached: ".$used."/".$max);
+                               Logger::warning("Maximum level (".$maxlevel."%) of user connections reached: ".$used."/".$max);
                                return true;
                        }
                }
@@ -561,7 +588,7 @@ class Worker
                if ($level < $maxlevel) {
                        return false;
                }
-               Logger::notice("Maximum level (".$level."%) of system connections reached: ".$used."/".$max);
+               Logger::warning("Maximum level (".$level."%) of system connections reached: ".$used."/".$max);
                return true;
        }
 
@@ -576,7 +603,7 @@ class Worker
                $stamp = (float)microtime(true);
                $entries = DBA::select(
                        'workerqueue',
-                       ['id', 'pid', 'executed', 'priority', 'parameter'],
+                       ['id', 'pid', 'executed', 'priority', 'command', 'parameter'],
                        ['NOT `done` AND `pid` != 0'],
                        ['order' => ['priority', 'retrial', 'created']]
                );
@@ -603,17 +630,21 @@ class Worker
                                $max_duration_defaults = [PRIORITY_CRITICAL => 720, PRIORITY_HIGH => 10, PRIORITY_MEDIUM => 60, PRIORITY_LOW => 180, PRIORITY_NEGLIGIBLE => 720];
                                $max_duration = $max_duration_defaults[$entry["priority"]];
 
-                               $argv = json_decode($entry["parameter"], true);
-                               if (empty($argv)) {
+                               $argv = json_decode($entry['parameter'], true);
+                               if (!empty($entry['command'])) {
+                                       $command = $entry['command'];
+                               } elseif (!empty($argv)) {
+                                       $command = array_shift($argv);
+                               } else {
                                        return;
                                }
 
-                               $argv[0] = basename($argv[0]);
+                               $command = basename($command);
 
                                // How long is the process already running?
                                $duration = (time() - strtotime($entry["executed"])) / 60;
                                if ($duration > $max_duration) {
-                                       Logger::notice("Worker process ".$entry["pid"]." (".substr(json_encode($argv), 0, 50).") took more than ".$max_duration." minutes. It will be killed now.");
+                                       Logger::notice('Worker process took too much time - killed', ['duration' => number_format($duration, 3), 'max' => $max_duration, 'id' => $entry["id"], 'pid' => $entry["pid"], 'command' => $command]);
                                        posix_kill($entry["pid"], SIGTERM);
 
                                        // We killed the stale process.
@@ -636,7 +667,7 @@ class Worker
                                        self::$db_duration += (microtime(true) - $stamp);
                                        self::$db_duration_write += (microtime(true) - $stamp);
                                } else {
-                                       Logger::info('Process runtime is okay', ['pid' => $entry["pid"], 'duration' => $duration, 'max' => $max_duration, 'command' => substr(json_encode($argv), 0, 50)]);
+                                       Logger::info('Process runtime is okay', ['duration' => number_format($duration, 3), 'max' => $max_duration, 'id' => $entry["id"], 'pid' => $entry["pid"], 'command' => $command]);
                                }
                        }
                }
@@ -757,7 +788,7 @@ class Worker
                        // Are there fewer workers running as possible? Then fork a new one.
                        if (!DI::config()->get("system", "worker_dont_fork", false) && ($queues > ($active + 1)) && self::entriesExists()) {
                                Logger::info("There are fewer workers as possible, fork a new worker.", ['active' => $active, 'queues' => $queues]);
-                               if (DI::config()->get('system', 'worker_daemon_mode', false)) {
+                               if (self::isDaemonMode()) {
                                        self::IPCSetJobState(true);
                                } else {
                                        self::spawnWorker();
@@ -766,7 +797,7 @@ class Worker
                }
 
                // if there are too much worker, we don't spawn a new one.
-               if (DI::config()->get('system', 'worker_daemon_mode', false) && ($active > $queues)) {
+               if (self::isDaemonMode() && ($active > $queues)) {
                        self::IPCSetJobState(false);
                }
 
@@ -800,7 +831,7 @@ class Worker
                $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'];
@@ -848,12 +879,17 @@ class Worker
                $ids = [];
                $stamp = (float)microtime(true);
                $condition = ["`priority` = ? AND `pid` = 0 AND NOT `done` AND `next_try` < ?", $priority, DateTimeFormat::utcNow()];
-               $tasks = DBA::select('workerqueue', ['id', 'parameter'], $condition, ['limit' => $limit, 'order' => ['retrial', 'created']]);
+               $tasks = DBA::select('workerqueue', ['id', 'command', 'parameter'], $condition, ['limit' => $limit, 'order' => ['retrial', 'created']]);
                self::$db_duration += (microtime(true) - $stamp);
                while ($task = DBA::fetch($tasks)) {
                        $ids[] = $task['id'];
                        // Only continue that loop while we are storing commands that can be processed quickly
-                       $command = json_decode($task['parameter'])[0];
+                       if (!empty($task['command'])) {
+                               $command = $task['command'];
+                       } else {
+                               $command = json_decode($task['parameter'])[0];
+                       }
+
                        if (!in_array($command, self::FAST_COMMANDS)) {
                                break;
                        }
@@ -968,13 +1004,17 @@ class Worker
                if ($limit > 0) {
                        $stamp = (float)microtime(true);
                        $condition = ["`pid` = 0 AND NOT `done` AND `next_try` < ?", DateTimeFormat::utcNow()];
-                       $tasks = DBA::select('workerqueue', ['id', 'parameter'], $condition, ['limit' => $limit, 'order' => ['priority', 'retrial', 'created']]);
+                       $tasks = DBA::select('workerqueue', ['id', 'command', 'parameter'], $condition, ['limit' => $limit, 'order' => ['priority', 'retrial', 'created']]);
                        self::$db_duration += (microtime(true) - $stamp);
 
                        while ($task = DBA::fetch($tasks)) {
                                $ids[] = $task['id'];
                                // Only continue that loop while we are storing commands that can be processed quickly
-                               $command = json_decode($task['parameter'])[0];
+                               if (!empty($task['command'])) {
+                                       $command = $task['command'];
+                               } else {
+                                       $command = json_decode($task['parameter'])[0];
+                               }
                                if (!in_array($command, self::FAST_COMMANDS)) {
                                        break;
                                }
@@ -1050,112 +1090,84 @@ class Worker
        }
 
        /**
-        * Call the front end worker
+        * Runs the cron processes
         *
         * @return void
         * @throws \Friendica\Network\HTTPException\InternalServerErrorException
         */
-       public static function callWorker()
+       private static function runCron()
        {
-               if (!DI::config()->get("system", "frontend_worker")) {
-                       return;
-               }
+               Logger::info('Add cron entries');
+
+               // Check for spooled items
+               self::add(['priority' => PRIORITY_HIGH, 'force_priority' => true], 'SpoolPost');
 
-               $url = DI::baseUrl() . '/worker';
-               DI::httpRequest()->fetch($url, 1);
+               // Run the cron job that calls all other jobs
+               self::add(['priority' => PRIORITY_MEDIUM, 'force_priority' => true], 'Cron');
+
+               // Cleaning dead processes
+               self::killStaleWorkers();
        }
 
        /**
-        * Call the front end worker if there aren't any active
+        * Fork a child process
         *
+        * @param boolean $do_cron
         * @return void
-        * @throws \Friendica\Network\HTTPException\InternalServerErrorException
         */
-       public static function executeIfIdle()
+       private static function forkProcess(bool $do_cron)
        {
-               self::checkDaemonState();
-
-               if (!DI::config()->get("system", "frontend_worker")) {
+               if (DI::process()->isMinMemoryReached()) {
+                       Logger::warning('Memory limit reached - quitting');
                        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());
+               // Children inherit their parent's database connection.
+               // To avoid problems we disconnect and connect both parent and child
+               DBA::disconnect();
+               $pid = pcntl_fork();
+               if ($pid == -1) {
+                       DBA::connect();
+                       Logger::warning('Could not spawn worker');
+                       return;
+               } elseif ($pid) {
+                       // The parent process continues here
+                       DBA::connect();
 
-                       // Do we have enough running workers? Then we quit here.
-                       if (self::tooMuchWorkers()) {
-                               // Cleaning dead processes
-                               self::killStaleWorkers();
-                               DI::modelProcess()->deleteInactive();
+                       self::IPCSetJobState(true, $pid);
+                       Logger::info('Spawned new worker', ['pid' => $pid]);
 
-                               return;
+                       $cycles = 0;
+                       while (self::IPCJobsExists($pid) && (++$cycles < 100)) {
+                               usleep(10000);
                        }
 
-                       self::runCron();
-
-                       Logger::info('Call worker');
-                       self::spawnWorker();
+                       Logger::info('Spawned worker is ready', ['pid' => $pid, 'wait_cycles' => $cycles]);
                        return;
                }
 
-               // We cannot execute background processes.
-               // We now run the processes from the frontend.
-               // This won't work with long running processes.
-               self::runCron();
+               // We now are in the new worker
+               $pid = getmypid();
 
-               self::clearProcesses();
+               DBA::connect();
+               /// @todo Reinitialize the logger to set a new process_id and uid
+               DI::process()->setPid($pid);
 
-               $workers = self::activeWorkers();
-
-               if ($workers == 0) {
-                       self::callWorker();
+               $cycles = 0;
+               while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
+                       usleep(10000);
                }
-       }
 
-       /**
-        * 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);
+               Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
 
-               /// @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);
-       }
+               self::processQueue($do_cron);
 
-       /**
-        * Runs the cron processes
-        *
-        * @return void
-        * @throws \Friendica\Network\HTTPException\InternalServerErrorException
-        */
-       private static function runCron()
-       {
-               Logger::info('Add cron entries');
+               self::unclaimProcess();
 
-               // Check for spooled items
-               self::add(['priority' => PRIORITY_HIGH, 'force_priority' => true], 'SpoolPost');
-
-               // Run the cron job that calls all other jobs
-               self::add(['priority' => PRIORITY_MEDIUM, 'force_priority' => true], 'Cron');
-
-               // Cleaning dead processes
-               self::killStaleWorkers();
+               self::IPCSetJobState(false, $pid);
+               DI::process()->end();
+               Logger::info('Worker ended', ['pid' => $pid]);
+               exit();
        }
 
        /**
@@ -1167,16 +1179,14 @@ class Worker
         */
        public static function spawnWorker($do_cron = false)
        {
-               $command = 'bin/worker.php';
-
-               $args = ['no_cron' => !$do_cron];
-
-               $a = DI::app();
-               $process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::modelProcess(), $a->getBasePath(), getmypid());
-               $process->run($command, $args);
-
-               // after spawning we have to remove the flag.
-               if (DI::config()->get('system', 'worker_daemon_mode', false)) {
+               if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) {
+                       self::forkProcess($do_cron);
+               } 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);
                }
        }
@@ -1190,7 +1200,7 @@ class Worker
         * 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 ""
         *
@@ -1203,20 +1213,21 @@ class Worker
                $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;
                // Don't fork from frontend tasks by default
                $dont_fork = DI::config()->get("system", "worker_dont_fork", false) || !DI::mode()->isBackend();
                $created = DateTimeFormat::utcNow();
+               $delayed = DBA::NULL_DATETIME;
                $force_priority = false;
 
                $run_parameter = array_shift($args);
@@ -1224,6 +1235,9 @@ class Worker
                if (is_int($run_parameter)) {
                        $priority = $run_parameter;
                } elseif (is_array($run_parameter)) {
+                       if (isset($run_parameter['delayed'])) {
+                               $delayed = $run_parameter['delayed'];
+                       }
                        if (isset($run_parameter['priority'])) {
                                $priority = $run_parameter['priority'];
                        }
@@ -1238,26 +1252,33 @@ class Worker
                        }
                }
 
+               $command = array_shift($args);
                $parameters = json_encode($args);
-               $found = DBA::exists('workerqueue', ['parameter' => $parameters, 'done' => false]);
-               $added = false;
+               $found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
+               $added = 0;
+
+               if (!in_array($priority, PRIORITIES)) {
+                       Logger::warning('Invalid priority', ['priority' => $priority, 'command' => $command, 'callstack' => System::callstack(20)]);
+                       $priority = PRIORITY_MEDIUM;
+               }
 
                // 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', ['parameter' => $parameters, 'created' => $created, 'priority' => $priority]);
-                       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], ['parameter' => $parameters, 'done' => false, 'pid' => 0]);
+                       DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
                }
 
                // Set the IPC flag to ensure an immediate process execution via daemon
-               if (DI::config()->get('system', 'worker_daemon_mode', false)) {
+               if (self::isDaemonMode()) {
                        self::IPCSetJobState(true);
                }
 
@@ -1282,7 +1303,7 @@ class Worker
                }
 
                // Quit on daemon mode
-               if (DI::config()->get('system', 'worker_daemon_mode', false)) {
+               if (self::isDaemonMode()) {
                        return $added;
                }
 
@@ -1292,6 +1313,11 @@ class Worker
                return $added;
        }
 
+       public static function countWorkersByCommand(string $command)
+       {
+               return DBA::count('workerqueue', ['done' => false, 'pid' => 0, 'command' => $command]);
+       }
+
        /**
         * Returns the next retrial level for worker jobs.
         * This function will skip levels when jobs are older.
@@ -1314,7 +1340,7 @@ class Worker
                                $new_retrial = $retrial;
                        }
                }
-               Logger::info('New retrial for task', ['id' => $queue['id'], 'created' => $queue['created'], 'old' => $queue['retrial'], 'new' => $new_retrial]);
+               Logger::notice('New retrial for task', ['id' => $queue['id'], 'created' => $queue['created'], 'old' => $queue['retrial'], 'new' => $new_retrial]);
                return $new_retrial;
        }
 
@@ -1325,12 +1351,12 @@ class Worker
         */
        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'];
@@ -1340,7 +1366,7 @@ class Worker
                $new_retrial = self::getNextRetrial($queue, $max_level);
 
                if ($new_retrial > $max_level) {
-                       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]);
+                       Logger::notice('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;
                }
 
@@ -1371,12 +1397,27 @@ class Worker
         * 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($jobs)
+       public static function IPCSetJobState(bool $jobs, int $key = 0)
        {
                $stamp = (float)microtime(true);
-               DBA::update('worker-ipc', ['jobs' => $jobs], ['key' => 1], 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 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);
        }
@@ -1384,13 +1425,14 @@ class Worker
        /**
         * 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
@@ -1401,6 +1443,51 @@ class Worker
                return (bool)$row['jobs'];
        }
 
+       /**
+        * Checks if the worker is running in the daemon mode.
+        *
+        * @return boolean
+        */
+       public static function isDaemonMode()
+       {
+               if (!is_null(self::$daemon_mode)) {
+                       return self::$daemon_mode;
+               }
+
+               if (DI::mode()->getExecutor() == Mode::DAEMON) {
+                       return true;
+               }
+
+               $daemon_mode = DI::config()->get('system', 'worker_daemon_mode', false, true);
+               if ($daemon_mode) {
+                       return $daemon_mode;
+               }
+
+               if (!function_exists('pcntl_fork')) {
+                       self::$daemon_mode = false;
+                       return false;
+               }
+
+               $pidfile = DI::config()->get('system', 'pidfile');
+               if (empty($pidfile)) {
+                       // No pid file, no daemon
+                       self::$daemon_mode = false;
+                       return false;
+               }
+
+               if (!is_readable($pidfile)) {
+                       // No pid file. We assume that the daemon had been intentionally stopped.
+                       self::$daemon_mode = false;
+                       return false;
+               }
+
+               $pid = intval(file_get_contents($pidfile));
+               $running = posix_kill($pid, 0);
+
+               self::$daemon_mode = $running;
+               return $running;
+       }
+
        /**
         * Test if the daemon is running. If not, it will be started
         *
@@ -1452,12 +1539,12 @@ class Worker
         */
        private static function spawnDaemon()
        {
-               Logger::info('Starting new daemon process');
+               Logger::notice('Starting new daemon process');
                $command = 'bin/daemon.php';
                $a = DI::app();
                $process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::modelProcess(), $a->getBasePath(), getmypid());
                $process->run($command, ['start']);
-               Logger::info('New daemon process started');
+               Logger::notice('New daemon process started');
        }
 
        /**
@@ -1500,7 +1587,7 @@ class Worker
                } 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;
        }
 }