]> git.mxchange.org Git - friendica.git/blobdiff - src/Core/Worker.php
Formatting logger to new style
[friendica.git] / src / Core / Worker.php
index c8a4c038bbc44f78f5979171cee2cfc71ab3e0de..29d825050f67920b56aadcb3bb529ee4f8fec4b0 100644 (file)
@@ -8,6 +8,7 @@ use Friendica\BaseObject;
 use Friendica\Database\DBA;
 use Friendica\Model\Process;
 use Friendica\Util\DateTimeFormat;
+use Friendica\Util\Logger\WorkerLogger;
 use Friendica\Util\Network;
 
 /**
@@ -40,6 +41,9 @@ class Worker
        {
                $a = \get_app();
 
+               // Ensure that all "strtotime" operations do run timezone independent
+               date_default_timezone_set('UTC');
+
                self::$up_start = microtime(true);
 
                // At first check the maximum load. We shouldn't continue with a high load
@@ -89,15 +93,8 @@ class Worker
 
                $starttime = time();
 
-               $entries = 0;
-               $deferred = 0;
-
                // We fetch the next queue entry that is about to be executed
-               while ($r = self::workerProcess($passing_slow, $entries, $deferred)) {
-                       // When we are processing jobs with a lower priority, we don't refetch new jobs
-                       // Otherwise fast jobs could wait behind slow ones and could be blocked.
-                       $refetched = $passing_slow;
-
+               while ($r = self::workerProcess()) {
                        foreach ($r as $entry) {
                                // Assure that the priority is an integer value
                                $entry['priority'] = (int)$entry['priority'];
@@ -109,20 +106,16 @@ class Worker
                                }
 
                                // If possible we will fetch new jobs for this worker
-                               if (!$refetched) {
-                                       $entries = self::totalEntries();
-                                       $deferred = self::deferredEntries();
-                                       if (Lock::acquire('worker_process', 0)) {
-                                               $refetched = self::findWorkerProcesses($passing_slow, $entries, $deferred);
-                                               Lock::release('worker_process');
-                                       }
+                               if (!self::getWaitingJobForPID() && Lock::acquire('worker_process', 0)) {
+                                       self::findWorkerProcesses();
+                                       Lock::release('worker_process');
                                }
                        }
 
                        // To avoid the quitting of multiple workers only one worker at a time will execute the check
                        if (Lock::acquire('worker', 0)) {
                                // Count active workers and compare them with a maximum value that depends on the load
-                               if (self::tooMuchWorkers($entries, $deferred)) {
+                               if (self::tooMuchWorkers()) {
                                        Logger::log('Active worker limit reached, quitting.', Logger::DEBUG);
                                        Lock::release('worker');
                                        return;
@@ -363,51 +356,33 @@ class Worker
        {
                $a = \get_app();
 
-               $mypid = getmypid();
-
                $argc = count($argv);
 
-               $new_process_id = System::processID("wrk");
+               $logger = $a->getLogger();
+               $workerLogger = new WorkerLogger($logger, $funcname);
 
-               Logger::log("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]." - Process PID: ".$new_process_id);
+               $workerLogger ->info("Process start.", ['priority' => $queue["priority"], 'id' => $queue["id"]]);
 
                $stamp = (float)microtime(true);
 
                // We use the callstack here to analyze the performance of executed worker entries.
                // For this reason the variables have to be initialized.
-               if (Config::get("system", "profiler")) {
-                       $a->performance["start"] = microtime(true);
-                       $a->performance["database"] = 0;
-                       $a->performance["database_write"] = 0;
-                       $a->performance["cache"] = 0;
-                       $a->performance["cache_write"] = 0;
-                       $a->performance["network"] = 0;
-                       $a->performance["file"] = 0;
-                       $a->performance["rendering"] = 0;
-                       $a->performance["parser"] = 0;
-                       $a->performance["marktime"] = 0;
-                       $a->performance["markstart"] = microtime(true);
-                       $a->callstack = [];
-               }
-
-               // For better logging create a new process id for every worker call
-               // But preserve the old one for the worker
-               $old_process_id = $a->process_id;
-               $a->process_id = $new_process_id;
-               $a->queue = $queue;
+               $a->getProfiler()->reset();
 
                $up_duration = microtime(true) - self::$up_start;
 
                // Reset global data to avoid interferences
                unset($_SESSION);
 
+               // Set the workerLogger as new default logger
+               Logger::init($workerLogger);
                if ($method_call) {
                        call_user_func_array(sprintf('Friendica\Worker\%s::execute', $funcname), $argv);
                } else {
                        $funcname($argv, $argc);
                }
+               Logger::init($logger);
 
-               $a->process_id = $old_process_id;
                unset($a->queue);
 
                $duration = (microtime(true) - $stamp);
@@ -417,16 +392,15 @@ class Worker
                 * The execution time is the productive time.
                 * By changing parameters like the maximum number of workers we can check the effectivness.
                */
-               Logger::log(
-                       'DB: '.number_format(self::$db_duration - (self::$db_duration_count + self::$db_duration_write + self::$db_duration_stat), 4).
-                       ' - DB-Count: '.number_format(self::$db_duration_count, 4).
-                       ' - DB-Stat: '.number_format(self::$db_duration_stat, 4).
-                       ' - DB-Write: '.number_format(self::$db_duration_write, 4).
-                       ' - Lock: '.number_format(self::$lock_duration, 4).
-                       ' - Rest: '.number_format(max(0, $up_duration - (self::$db_duration + self::$lock_duration)), 4).
-                       ' - Execution: '.number_format($duration, 4),
-                       Logger::DEBUG
-               );
+               $dbtotal = number_format(self::$db_duration - (self::$db_duration_count + self::$db_duration_write + self::$db_duration_stat), 4);
+               $dbcount = number_format(self::$db_duration_count, 4);
+               $dbstat  = number_format(self::$db_duration_stat, 4);
+               $dbwrite = number_format(self::$db_duration_write, 4);
+               $dblock  = number_format(self::$lock_duration, 4);
+               $rest    = number_format(max(0, $up_duration - (self::$db_duration + self::$lock_duration)), 4);
+               $exec    = number_format($duration, 4);
+
+               $workerLogger->info('Performance log.', ['total' => $dbtotal, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'block' => $dblock, 'rest' => $rest, 'exec' => $exec]);
 
                self::$up_start = microtime(true);
                self::$db_duration = 0;
@@ -436,92 +410,23 @@ class Worker
                self::$lock_duration = 0;
 
                if ($duration > 3600) {
-                       Logger::log("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 1 hour (".round($duration/60, 3).")", Logger::DEBUG);
+                       $workerLogger->info('Longer than 1 hour.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' =>   round($duration/60, 3)]);
                } elseif ($duration > 600) {
-                       Logger::log("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 10 minutes (".round($duration/60, 3).")", Logger::DEBUG);
+                       $workerLogger->info('Longer than 10 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' =>   round($duration/60, 3)]);
                } elseif ($duration > 300) {
-                       Logger::log("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 5 minutes (".round($duration/60, 3).")", Logger::DEBUG);
+                       $workerLogger->info('Longer than 5 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' =>   round($duration/60, 3)]);
                } elseif ($duration > 120) {
-                       Logger::log("Prio ".$queue["priority"].": ".$queue["parameter"]." - longer than 2 minutes (".round($duration/60, 3).")", Logger::DEBUG);
+                       $workerLogger->info('Longer than 2 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' =>   round($duration/60, 3)]);
                }
 
-               Logger::log("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - done in ".$duration." seconds. Process PID: ".$new_process_id);
+               $workerLogger->info('Process done. ', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' =>   number_format($duration, 4)]);
 
-               // Write down the performance values into the log
-               if (Config::get("system", "profiler")) {
-                       $duration = microtime(true)-$a->performance["start"];
-
-                       $o = '';
-                       if (Config::get("rendertime", "callstack")) {
-                               if (isset($a->callstack["database"])) {
-                                       $o .= "\nDatabase Read:\n";
-                                       foreach ($a->callstack["database"] as $func => $time) {
-                                               $time = round($time, 3);
-                                               if ($time > 0) {
-                                                       $o .= $func.": ".$time."\n";
-                                               }
-                                       }
-                               }
-                               if (isset($a->callstack["database_write"])) {
-                                       $o .= "\nDatabase Write:\n";
-                                       foreach ($a->callstack["database_write"] as $func => $time) {
-                                               $time = round($time, 3);
-                                               if ($time > 0) {
-                                                       $o .= $func.": ".$time."\n";
-                                               }
-                                       }
-                               }
-                               if (isset($a->callstack["dache"])) {
-                                       $o .= "\nCache Read:\n";
-                                       foreach ($a->callstack["dache"] as $func => $time) {
-                                               $time = round($time, 3);
-                                               if ($time > 0) {
-                                                       $o .= $func.": ".$time."\n";
-                                               }
-                                       }
-                               }
-                               if (isset($a->callstack["dache_write"])) {
-                                       $o .= "\nCache Write:\n";
-                                       foreach ($a->callstack["dache_write"] as $func => $time) {
-                                               $time = round($time, 3);
-                                               if ($time > 0) {
-                                                       $o .= $func.": ".$time."\n";
-                                               }
-                                       }
-                               }
-                               if (isset($a->callstack["network"])) {
-                                       $o .= "\nNetwork:\n";
-                                       foreach ($a->callstack["network"] as $func => $time) {
-                                               $time = round($time, 3);
-                                               if ($time > 0) {
-                                                       $o .= $func.": ".$time."\n";
-                                               }
-                                       }
-                               }
-                       }
-
-                       Logger::log(
-                               "ID ".$queue["id"].": ".$funcname.": ".sprintf(
-                                       "DB: %s/%s, Cache: %s/%s, Net: %s, I/O: %s, Other: %s, Total: %s".$o,
-                                       number_format($a->performance["database"] - $a->performance["database_write"], 2),
-                                       number_format($a->performance["database_write"], 2),
-                                       number_format($a->performance["cache"], 2),
-                                       number_format($a->performance["cache_write"], 2),
-                                       number_format($a->performance["network"], 2),
-                                       number_format($a->performance["file"], 2),
-                                       number_format($duration - ($a->performance["database"]
-                                               + $a->performance["cache"] + $a->performance["cache_write"]
-                                               + $a->performance["network"] + $a->performance["file"]), 2),
-                                       number_format($duration, 2)
-                               ),
-                               Logger::DEBUG
-                       );
-               }
+               $a->getProfiler()->saveLog($a->getLogger(), "ID " . $queue["id"] . ": " . $funcname);
 
                $cooldown = Config::get("system", "worker_cooldown", 0);
 
                if ($cooldown > 0) {
-                       Logger::log("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." - in cooldown for ".$cooldown." seconds");
+                       $workerLogger->info('Cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
                        sleep($cooldown);
                }
        }
@@ -684,13 +589,10 @@ class Worker
        /**
         * @brief Checks if the number of active workers exceeds the given limits
         *
-        * @param integer $entries Total number of queue entries
-        * @param integer $deferred Number of deferred queue entries
-        *
         * @return bool Are there too much workers running?
         * @throws \Friendica\Network\HTTPException\InternalServerErrorException
         */
-       public static function tooMuchWorkers($entries = 0, $deferred = 0)
+       private static function tooMuchWorkers()
        {
                $queues = Config::get("system", "worker_queues", 4);
 
@@ -718,6 +620,12 @@ class Worker
                                $intervals = explode(',', Config::get('system', 'worker_jpm_range'));
                                $jobs_per_minute = [];
                                foreach ($intervals as $interval) {
+                                       if ($interval == 0) {
+                                               continue;
+                                       } else {
+                                               $interval = (int)$interval;
+                                       }
+
                                        $stamp = (float)microtime(true);
                                        $jobs = DBA::p("SELECT COUNT(*) AS `jobs` FROM `workerqueue` WHERE `done` AND `executed` > UTC_TIMESTAMP() - INTERVAL ? MINUTE", $interval);
                                        self::$db_duration += (microtime(true) - $stamp);
@@ -735,7 +643,10 @@ class Worker
 
                        $idle_workers = $active;
 
+                       $deferred = self::deferredEntries();
+
                        if (Config::get('system', 'worker_debug')) {
+                               $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());
@@ -748,15 +659,20 @@ class Worker
                                        self::$db_duration_stat += (microtime(true) - $stamp);
                                        if ($process = DBA::fetch($processes)) {
                                                $idle_workers -= $process["running"];
+                                               $waiting_processes += $entry["entries"];
                                                $listitem[$entry["priority"]] = $entry["priority"].":".$process["running"]."/".$entry["entries"];
                                        }
                                        DBA::close($processes);
                                }
                                DBA::close($jobs);
+                               $entries = $deferred + $waiting_processes;
                        } else {
+                               $entries = self::totalEntries();
+                               $waiting_processes = max(0, $entries - $deferred);
                                $stamp = (float)microtime(true);
                                $jobs = DBA::p("SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `done` GROUP BY `priority` ORDER BY `priority`");
                                self::$db_duration += (microtime(true) - $stamp);
+                               self::$db_duration_stat += (microtime(true) - $stamp);
 
                                while ($entry = DBA::fetch($jobs)) {
                                        $idle_workers -= $entry["running"];
@@ -769,11 +685,6 @@ class Worker
 
                        $processlist .= ' ('.implode(', ', $listitem).')';
 
-                       if (empty($deferred) && empty($entries)) {
-                               $deferred = self::deferredEntries();
-                               $entries = max(self::totalEntries() - $deferred, 0);
-                       }
-
                        if (Config::get("system", "worker_fastlane", false) && ($queues > 0) && self::entriesExists() && ($active >= $queues)) {
                                $top_priority = self::highestPriority();
                                $high_running = self::processWithPriorityActive($top_priority);
@@ -784,7 +695,7 @@ class Worker
                                }
                        }
 
-                       Logger::log("Load: " . $load ."/" . $maxsysload . " - processes: " . $deferred . "/" . $active . "/" . ($entries - $deferred) . $processlist . " - maximum: " . $queues . "/" . $maxqueues, Logger::DEBUG);
+                       Logger::log("Load: " . $load ."/" . $maxsysload . " - processes: " . $deferred . "/" . $active . "/" . $waiting_processes . $processlist . " - maximum: " . $queues . "/" . $maxqueues, Logger::DEBUG);
 
                        // Are there fewer workers running as possible? Then fork a new one.
                        if (!Config::get("system", "worker_dont_fork", false) && ($queues > ($active + 1)) && ($entries > 1)) {
@@ -820,192 +731,180 @@ class Worker
        }
 
        /**
-        * @brief Check if we should pass some slow processes
+        * @brief Returns waiting jobs for the current process id
         *
-        * When the active processes of the highest priority are using more than 2/3
-        * of all processes, we let pass slower processes.
-        *
-        * @param string $highest_priority Returns the currently highest priority
-        * @return bool We let pass a slower process than $highest_priority
+        * @return array waiting workerqueue jobs
         * @throws \Exception
         */
-       private static function passingSlow(&$highest_priority)
+       private static function getWaitingJobForPID()
        {
-               $highest_priority = 0;
-
                $stamp = (float)microtime(true);
-               $r = DBA::p(
-                       "SELECT `priority`
-                               FROM `process`
-                               INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `done`"
-               );
+               $r = DBA::select('workerqueue', [], ['pid' => getmypid(), 'done' => false]);
                self::$db_duration += (microtime(true) - $stamp);
-
-               // No active processes at all? Fine
-               if (!DBA::isResult($r)) {
-                       return false;
-               }
-               $priorities = [];
-               while ($line = DBA::fetch($r)) {
-                       $priorities[] = $line["priority"];
+               if (DBA::isResult($r)) {
+                       return DBA::toArray($r);
                }
                DBA::close($r);
 
-               // Should not happen
-               if (count($priorities) == 0) {
-                       return false;
-               }
-               $highest_priority = min($priorities);
+               return false;
+       }
 
-               // The highest process is already the slowest one?
-               // Then we quit
-               if ($highest_priority == PRIORITY_NEGLIGIBLE) {
-                       return false;
+       /**
+        * @brief Returns the next jobs that should be executed
+        *
+        * @return array array with next jobs
+        * @throws \Exception
+        */
+       private static function nextProcess()
+       {
+               $priority = self::nextPriority();
+               if (empty($priority)) {
+                       Logger::info('No tasks found');
+                       return [];
                }
-               $high = 0;
-               foreach ($priorities as $priority) {
-                       if ($priority == $highest_priority) {
-                               ++$high;
-                       }
+
+               if ($priority <= PRIORITY_MEDIUM) {
+                       $limit = Config::get('system', 'worker_fetch_limit', 1);
+               } else {
+                       $limit = 1;
                }
-               Logger::log("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, Logger::DEBUG);
-               $passing_slow = (($high/count($priorities)) > (2/3));
 
-               if ($passing_slow) {
-                       Logger::log("Passing slower processes than priority ".$highest_priority, Logger::DEBUG);
+               $ids = [];
+               $stamp = (float)microtime(true);
+               $condition = ["`priority` = ? AND `pid` = 0 AND NOT `done` AND `next_try` < ?", $priority, DateTimeFormat::utcNow()];
+               $tasks = DBA::select('workerqueue', ['id'], $condition, ['limit' => $limit, 'order' => ['created']]);
+               self::$db_duration += (microtime(true) - $stamp);
+               while ($task = DBA::fetch($tasks)) {
+                       $ids[] = $task['id'];
                }
-               return $passing_slow;
+               DBA::close($tasks);
+
+               Logger::info('Found:', ['id' => $ids, 'priority' => $priority]);
+               return $ids;
        }
 
        /**
-        * @brief Find and claim the next worker process for us
+        * @brief Returns the priority of the next workerqueue job
         *
-        * @param boolean $passing_slow Returns if we had passed low priority processes
-        * @param integer $entries Total number of queue entries
-        * @param integer $deferred Number of deferred queue entries
-        * @return boolean Have we found something?
-        * @throws \Friendica\Network\HTTPException\InternalServerErrorException
+        * @return string priority
+        * @throws \Exception
         */
-       private static function findWorkerProcesses(&$passing_slow, $entries, $deferred)
+       private static function nextPriority()
        {
-               $mypid = getmypid();
+               $waiting = [];
+               $priorities = [PRIORITY_CRITICAL, PRIORITY_HIGH, PRIORITY_MEDIUM, PRIORITY_LOW, PRIORITY_NEGLIGIBLE];
+               foreach ($priorities as $priority) {
+                       $stamp = (float)microtime(true);
+                       if (DBA::exists('workerqueue', ["`priority` = ? AND `pid` = 0 AND NOT `done` AND `next_try` < ?", $priority, DateTimeFormat::utcNow()])) {
+                               $waiting[$priority] = true;
+                       }
+                       self::$db_duration += (microtime(true) - $stamp);
+               }
 
-               // Check if we should pass some low priority process
-               $highest_priority = 0;
-               $found = false;
-               $passing_slow = false;
+               if (!empty($waiting[PRIORITY_CRITICAL])) {
+                       return PRIORITY_CRITICAL;
+               }
 
-               // The higher the number of parallel workers, the more we prefetch to prevent concurring access
-               // We decrease the limit with the number of entries left in the queue
-               $worker_queues = Config::get("system", "worker_queues", 4);
-               $queue_length = Config::get('system', 'worker_fetch_limit', 1);
-               $lower_job_limit = $worker_queues * $queue_length * 2;
-               $entries = max($entries - $deferred, 0);
+               $running = [];
+               $running_total = 0;
+               $stamp = (float)microtime(true);
+               $processes = DBA::p("SELECT COUNT(DISTINCT(`process`.`pid`)) AS `running`, `priority` FROM `process`
+                       INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`
+                       WHERE NOT `done` GROUP BY `priority`");
+               self::$db_duration += (microtime(true) - $stamp);
+               while ($process = DBA::fetch($processes)) {
+                       $running[$process['priority']] = $process['running'];
+                       $running_total += $process['running'];
+               }
+               DBA::close($processes);
 
-               // Now do some magic
+               foreach ($priorities as $priority) {
+                       if (!empty($waiting[$priority]) && empty($running[$priority])) {
+                               Logger::info('No running worker found with priority {priority} - assigning it.', ['priority' => $priority]);
+                               return $priority;
+                       }
+               }
+
+               $active = max(self::activeWorkers(), $running_total);
+               $priorities = max(count($waiting), count($running));
                $exponent = 2;
-               $slope = $queue_length / pow($lower_job_limit, $exponent);
-               $limit = min($queue_length, ceil($slope * pow($entries, $exponent)));
 
-               Logger::log('Deferred: ' . $deferred . ' - Total: ' . $entries . ' - Maximum: ' . $queue_length . ' - jobs per queue: ' . $limit, Logger::DEBUG);
-               $ids = [];
-               if (self::passingSlow($highest_priority)) {
-                       // Are there waiting processes with a higher priority than the currently highest?
-                       $stamp = (float)microtime(true);
-                       $result = DBA::select(
-                               'workerqueue',
-                               ['id'],
-                               ["`pid` = 0 AND `priority` < ? AND NOT `done` AND `next_try` < ?",
-                               $highest_priority, DateTimeFormat::utcNow()],
-                               ['limit' => $limit, 'order' => ['priority', 'created']]
-                       );
-                       self::$db_duration += (microtime(true) - $stamp);
+               $total = 0;
+               for ($i = 1; $i <= $priorities; ++$i) {
+                       $total += pow($i, $exponent);
+               }
 
-                       while ($id = DBA::fetch($result)) {
-                               $ids[] = $id["id"];
+               $limit = [];
+               for ($i = 1; $i <= $priorities; ++$i) {
+                       $limit[$priorities - $i] = max(1, round($active * (pow($i, $exponent) / $total)));
+               }
+
+               $i = 0;
+               foreach ($running as $priority => $workers) {
+                       if ($workers < $limit[$i++]) {
+                               Logger::info('Priority {priority} has got {workers} workers out of a limit of {limit}', ['priority' => $priority, 'workers' => $workers, 'limit' => $limit[$i - 1]]);
+                               return $priority;
                        }
-                       DBA::close($result);
+               }
 
-                       $found = (count($ids) > 0);
+               if (!empty($waiting)) {
+                       $priority =  array_shift(array_keys($waiting));
+                       Logger::info('No underassigned priority found, now taking the highest priority.', ['priority' => $priority]);
+                       return $priority;
+               }
 
-                       if (!$found) {
-                               // Give slower processes some processing time
-                               $stamp = (float)microtime(true);
-                               $result = DBA::select(
-                                       'workerqueue',
-                                       ['id'],
-                                       ["`pid` = 0 AND `priority` > ? AND NOT `done` AND `next_try` < ?",
-                                       $highest_priority, DateTimeFormat::utcNow()],
-                                       ['limit' => $limit, 'order' => ['priority', 'created']]
-                               );
-                               self::$db_duration += (microtime(true) - $stamp);
+               return false;
+       }
 
-                               while ($id = DBA::fetch($result)) {
-                                       $ids[] = $id["id"];
-                               }
-                               DBA::close($result);
+       /**
+        * @brief Find and claim the next worker process for us
+        *
+        * @return boolean Have we found something?
+        * @throws \Friendica\Network\HTTPException\InternalServerErrorException
+        */
+       private static function findWorkerProcesses()
+       {
+               $mypid = getmypid();
 
-                               $found = (count($ids) > 0);
-                               $passing_slow = $found;
-                       }
-               }
+               $ids = self::nextProcess();
 
-               // If there is no result (or we shouldn't pass lower processes) we check without priority limit
-               if (!$found) {
+               // If there is no result we check without priority limit
+               if (empty($ids)) {
                        $stamp = (float)microtime(true);
-                       $result = DBA::select(
-                               'workerqueue',
-                               ['id'],
-                               ["`pid` = 0 AND NOT `done` AND `next_try` < ?",
-                               DateTimeFormat::utcNow()],
-                               ['limit' => $limit, 'order' => ['priority', 'created']]
-                       );
+                       $condition = ["`pid` = 0 AND NOT `done` AND `next_try` < ?", DateTimeFormat::utcNow()];
+                       $result = DBA::select('workerqueue', ['id'], $condition, ['limit' => 1, 'order' => ['priority', 'created']]);
                        self::$db_duration += (microtime(true) - $stamp);
 
                        while ($id = DBA::fetch($result)) {
                                $ids[] = $id["id"];
                        }
                        DBA::close($result);
-
-                       $found = (count($ids) > 0);
                }
 
-               if ($found) {
+               if (!empty($ids)) {
                        $stamp = (float)microtime(true);
-                       $condition = "`id` IN (".substr(str_repeat("?, ", count($ids)), 0, -2).") AND `pid` = 0 AND NOT `done`";
-                       array_unshift($ids, $condition);
-                       DBA::update('workerqueue', ['executed' => DateTimeFormat::utcNow(), 'pid' => $mypid], $ids);
+                       $condition = ['id' => $ids, 'done' => false, 'pid' => 0];
+                       DBA::update('workerqueue', ['executed' => DateTimeFormat::utcNow(), 'pid' => $mypid], $condition);
                        self::$db_duration += (microtime(true) - $stamp);
                        self::$db_duration_write += (microtime(true) - $stamp);
                }
 
-               return $found;
+               return !empty($ids);
        }
 
        /**
         * @brief Returns the next worker process
         *
-        * @param boolean $passing_slow Returns if we had passed low priority processes
-        * @param integer $entries Returns total number of queue entries
-        * @param integer $deferred Returns number of deferred queue entries
-        *
         * @return string SQL statement
         * @throws \Friendica\Network\HTTPException\InternalServerErrorException
         */
-       public static function workerProcess(&$passing_slow, &$entries, &$deferred)
+       public static function workerProcess()
        {
                // There can already be jobs for us in the queue.
-               $stamp = (float)microtime(true);
-               $r = DBA::select('workerqueue', [], ['pid' => getmypid(), 'done' => false]);
-               self::$db_duration += (microtime(true) - $stamp);
-               if (DBA::isResult($r)) {
-                       return DBA::toArray($r);
+               $waiting = self::getWaitingJobForPID();
+               if (!empty($waiting)) {
+                       return $waiting;
                }
-               DBA::close($r);
-
-               // Counting the rows outside the lock reduces the lock time
-               $entries = self::totalEntries();
-               $deferred = self::deferredEntries();
 
                $stamp = (float)microtime(true);
                if (!Lock::acquire('worker_process')) {
@@ -1013,7 +912,7 @@ class Worker
                }
                self::$lock_duration += (microtime(true) - $stamp);
 
-               $found = self::findWorkerProcesses($passing_slow, $entries, $deferred);
+               $found = self::findWorkerProcesses();
 
                Lock::release('worker_process');