X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FCore%2FWorker.php;h=c84b59e2b7324c5ec6838c3962706e3f306c54c9;hb=34521c228bd69609fa4f475bb2e2e826723fcc16;hp=cbe6645294fe16b6eb289e5c10d16145aadbc6a3;hpb=bacf901d0e0d7a1f3f9a46ef3b13545688f8fd62;p=friendica.git diff --git a/src/Core/Worker.php b/src/Core/Worker.php index cbe6645294..c84b59e2b7 100644 --- a/src/Core/Worker.php +++ b/src/Core/Worker.php @@ -1,6 +1,6 @@ acquire(self::LOCK_WORKER, 0)) { // Count active workers and compare them with a maximum value that depends on the load if (self::tooMuchWorkers()) { - Logger::notice('Active worker limit reached, quitting.'); + Logger::info('Active worker limit reached, quitting.'); DI::lock()->release(self::LOCK_WORKER); return; } @@ -188,7 +188,7 @@ class Worker { // Count active workers and compare them with a maximum value that depends on the load if (self::tooMuchWorkers()) { - Logger::notice('Active worker limit reached, quitting.'); + Logger::info('Active worker limit reached, quitting.'); return false; } @@ -331,7 +331,7 @@ class Worker $mypid = getmypid(); // Quit when in maintenance - if (DI::config()->get('system', 'maintenance', false, true)) { + if (DI::config()->get('system', 'maintenance', false)) { Logger::notice('Maintenance mode - quit process', ['pid' => $mypid]); return false; } @@ -362,7 +362,7 @@ class Worker return false; } - // Check for existance and validity of the include file + // Check for existence and validity of the include file $include = $argv[0]; if (method_exists(sprintf('Friendica\Worker\%s', $include), 'execute')) { @@ -511,7 +511,7 @@ class Worker while ($load = System::getLoadAvg($processes_cooldown != 0)) { if (($load_cooldown > 0) && ($load['average1'] > $load_cooldown)) { if (!$sleeping) { - Logger::notice('Load induced pre execution cooldown.', ['max' => $load_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); + Logger::info('Load induced pre execution cooldown.', ['max' => $load_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); $sleeping = true; } sleep(1); @@ -519,7 +519,7 @@ class Worker } if (($processes_cooldown > 0) && ($load['scheduled'] > $processes_cooldown)) { if (!$sleeping) { - Logger::notice('Process induced pre execution cooldown.', ['max' => $processes_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); + Logger::info('Process induced pre execution cooldown.', ['max' => $processes_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); $sleeping = true; } sleep(1); @@ -529,7 +529,7 @@ class Worker } if ($sleeping) { - Logger::notice('Cooldown ended.', ['max-load' => $load_cooldown, 'max-processes' => $processes_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); + Logger::info('Cooldown ended.', ['max-load' => $load_cooldown, 'max-processes' => $processes_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); } } @@ -574,7 +574,7 @@ class Worker // No need to defer a worker queue entry if the arguments are invalid Logger::notice('Wrong worker arguments', ['class' => $funcname, 'argv' => $argv, 'queue' => $queue, 'message' => $e->getMessage()]); } catch (\Throwable $e) { - Logger::error('Uncaught exception in worker execution', ['class' => get_class($e), 'message' => $e->getMessage(), 'code' => $e->getCode(), 'file' => $e->getFile() . ':' . $e->getLine(), 'trace' => $e->getTraceAsString()]); + Logger::error('Uncaught exception in worker execution', ['class' => get_class($e), 'message' => $e->getMessage(), 'code' => $e->getCode(), 'file' => $e->getFile() . ':' . $e->getLine(), 'trace' => $e->getTraceAsString(), 'previous' => $e->getPrevious()]); Worker::defer(); } } else { @@ -590,7 +590,7 @@ class Worker /* With these values we can analyze how effective the worker is. * The database and rest time should be low since this is the unproductive time. * The execution time is the productive time. - * By changing parameters like the maximum number of workers we can check the effectivness. + * By changing parameters like the maximum number of workers we can check the effectiveness. */ $dbtotal = round(self::$db_duration, 2); $dbread = round(self::$db_duration - (self::$db_duration_count + self::$db_duration_write + self::$db_duration_stat), 2); @@ -601,7 +601,7 @@ class Worker $rest = round(max(0, $up_duration - (self::$db_duration + self::$lock_duration)), 2); $exec = round($duration, 2); - Logger::info('Performance:', ['state' => self::$state, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'lock' => $dblock, 'total' => $dbtotal, 'rest' => $rest, 'exec' => $exec]); + Logger::info('Performance:', ['function' => $funcname, 'state' => self::$state, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'lock' => $dblock, 'total' => $dbtotal, 'rest' => $rest, 'exec' => $exec]); self::coolDown(); @@ -622,7 +622,7 @@ class Worker Logger::info('Longer than 2 minutes.', ['priority' => $queue['priority'], 'id' => $queue['id'], 'duration' => round($duration/60, 3)]); } - Logger::info('Process done.', ['priority' => $queue['priority'], 'id' => $queue['id'], 'duration' => round($duration, 3)]); + Logger::info('Process done.', ['function' => $funcname, 'priority' => $queue['priority'], 'retrial' => $queue['retrial'], 'id' => $queue['id'], 'duration' => round($duration, 3)]); DI::profiler()->saveLog(DI::logger(), 'ID ' . $queue['id'] . ': ' . $funcname); } @@ -662,16 +662,24 @@ class Worker DBA::close($r); } + $stamp = (float)microtime(true); + $used = 0; + $sleep = 0; + $data = DBA::p("SHOW PROCESSLIST"); + while ($row = DBA::fetch($data)) { + if ($row['Command'] != 'Sleep') { + ++$used; + } else { + ++$sleep; + } + } + DBA::close($data); + self::$db_duration += (microtime(true) - $stamp); + // If $max is set we will use the processlist to determine the current number of connections // The processlist only shows entries of the current user if ($max != 0) { - $stamp = (float)microtime(true); - $r = DBA::p('SHOW PROCESSLIST'); - self::$db_duration += (microtime(true) - $stamp); - $used = DBA::numRows($r); - DBA::close($r); - - Logger::info('Connection usage (user values)', ['usage' => $used, 'max' => $max]); + Logger::info('Connection usage (user values)', ['working' => $used, 'sleeping' => $sleep, 'max' => $max]); $level = ($used / $max) * 100; @@ -695,11 +703,11 @@ class Worker if (!DBA::isResult($r)) { return false; } - $used = intval($r['Value']); + $used = max($used, intval($r['Value'])) - $sleep; if ($used == 0) { return false; } - Logger::info('Connection usage (system values)', ['used' => $used, 'max' => $max]); + Logger::info('Connection usage (system values)', ['working' => $used, 'sleeping' => $sleep, 'max' => $max]); $level = $used / $max * 100; @@ -814,7 +822,7 @@ class Worker } } - Logger::notice('Load: ' . $load . '/' . $maxsysload . ' - processes: ' . $deferred . '/' . $active . '/' . $waiting_processes . $processlist . ' - maximum: ' . $queues . '/' . $maxqueues); + Logger::info('Load: ' . $load . '/' . $maxsysload . ' - processes: ' . $deferred . '/' . $active . '/' . $waiting_processes . $processlist . ' - maximum: ' . $queues . '/' . $maxqueues); // 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() && !self::systemLimitReached()) { @@ -877,7 +885,7 @@ class Worker /** * Returns waiting jobs for the current process id * - * @return array|bool waiting workerqueue jobs or FALSE on failture + * @return array|bool waiting workerqueue jobs or FALSE on failure * @throws \Exception */ private static function getWaitingJobForPID() @@ -1264,7 +1272,7 @@ class Worker $command = array_shift($args); $parameters = json_encode($args); - $found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]); + $queue = DBA::selectFirst('workerqueue', ['id', 'priority'], ['command' => $command, 'parameter' => $parameters, 'done' => false]); $added = 0; if (!is_int($priority) || !in_array($priority, self::PRIORITIES)) { @@ -1277,14 +1285,17 @@ class Worker return 0; } - if (!$found) { + if (empty($queue)) { 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]); + $ret = DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]); + if ($ret && ($priority != $queue['priority'])) { + $added = $queue['id']; + } } // Set the IPC flag to ensure an immediate process execution via daemon @@ -1312,8 +1323,8 @@ class Worker return $added; } - // Quit on daemon mode - if (Worker\Daemon::isMode()) { + // Quit on daemon mode, except the priority is critical (like for db updates) + if (Worker\Daemon::isMode() && $priority !== self::PRIORITY_CRITICAL) { return $added; } @@ -1354,6 +1365,17 @@ class Worker return $new_retrial; } + /** + * Get the number of retrials for the current worker task + * + * @return integer + */ + public static function getRetrial(): int + { + $queue = DI::app()->getQueue(); + return $queue['retrial'] ?? 0; + } + /** * Defers the current worker entry * @@ -1411,7 +1433,7 @@ class Worker */ public static function isInMaintenanceWindow(bool $check_last_execution = false): bool { - // Calculate the seconds of the start end end of the maintenance window + // Calculate the seconds of the start and end of the maintenance window $start = strtotime(DI::config()->get('system', 'maintenance_start')) % 86400; $end = strtotime(DI::config()->get('system', 'maintenance_end')) % 86400;