]> 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 504a7b9f548aaa59bc5d4648caad3de92d35e17b..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
  *
@@ -311,6 +311,10 @@ class Worker
                }
 
                $argv = json_decode($queue['parameter'], true);
+               if (!is_array($argv)) {
+                       $argv = [];
+               }
+
                if (!empty($queue['command'])) {
                        array_unshift($argv, $queue['command']);
                }
@@ -442,7 +446,7 @@ class Worker
                        $queue['priority'] = PRIORITY_MEDIUM;
                }
 
-               $a->queue = $queue;
+               $a->setQueue($queue);
 
                $up_duration = microtime(true) - self::$up_start;
 
@@ -458,7 +462,7 @@ class Worker
 
                Logger::disableWorker();
 
-               unset($a->queue);
+               $a->setQueue([]);
 
                $duration = (microtime(true) - $stamp);
 
@@ -827,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'];
@@ -1085,95 +1089,6 @@ class Worker
                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
         *
@@ -1285,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 ""
         *
@@ -1298,14 +1213,14 @@ 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;
@@ -1340,7 +1255,7 @@ class Worker
                $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)]);
@@ -1349,15 +1264,15 @@ class Worker
 
                // 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]);
                }
@@ -1436,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'];
@@ -1672,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;
        }
 }