Merge pull request #7754 from annando/aria

[friendica.git] / src / Core / Worker.php

diff --git a/src/Core/Worker.php b/src/Core/Worker.php
index 709424c5ad318afefefa3bc8cdcdf7a29453936c..c64b0ebc6bdaa681e631e46e25dddb8526aa5bf2 100644 (file)
--- a/src/Core/Worker.php
+++ b/src/Core/Worker.php
@@ -5,6 +5,7 @@
 namespace Friendica\Core;
 
 use Friendica\BaseObject;
+use Friendica\Core;
 use Friendica\Database\DBA;
 use Friendica\Model\Process;
 use Friendica\Util\DateTimeFormat;
@@ -670,7 +671,7 @@ class Worker
                                $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)) {
@@ -700,6 +701,8 @@ class Worker
                                DBA::close($jobs);
                        }
 
+                       $waiting_processes -= $deferred;
+
                        $listitem[0] = "0:" . max(0, $idle_workers);
 
                        $processlist .= ' ('.implode(', ', $listitem).')';
@@ -1082,7 +1085,9 @@ class Worker
 
                $args = ['no_cron' => !$do_cron];
 
-               get_app()->proc_run($command, $args);
+               $a = get_app();
+               $process = new Core\Process($a->getLogger(), $a->getMode(), $a->getConfig(), $a->getBasePath());
+               $process->run($command, $args);
 
                // after spawning we have to remove the flag.
                if (Config::get('system', 'worker_daemon_mode', false)) {
@@ -1099,7 +1104,7 @@ class Worker
         * 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 ""
         *
@@ -1124,7 +1129,7 @@ class Worker
 
                $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) || !\get_app()->getMode()->isBackend();
                $created = DateTimeFormat::utcNow();
                $force_priority = false;
 
@@ -1149,6 +1154,7 @@ class Worker
 
                $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) {
@@ -1156,19 +1162,22 @@ class Worker
                }
 
                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
@@ -1176,19 +1185,19 @@ class Worker
                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;
        }
 
        /**
@@ -1219,11 +1228,12 @@ class Worker
 
        /**
         * Defers the current worker entry
+        * @return boolean had the entry been deferred?
         */
        public static function defer()
        {
                if (empty(BaseObject::getApp()->queue)) {
-                       return;
+                       return false;
                }
 
                $queue = BaseObject::getApp()->queue;
@@ -1237,8 +1247,8 @@ 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, '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
@@ -1253,13 +1263,15 @@ class Worker
                        $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;
        }
 
        /**