diff --git a/src/Core/System.php b/src/Core/System.php index 55b4bb8cf9..0c1e499d93 100644 --- a/src/Core/System.php +++ b/src/Core/System.php @@ -442,17 +442,24 @@ class System */ public static function getLoadAvg(): array { - $content = file_get_contents('/proc/loadavg'); + $content = @file_get_contents('/proc/loadavg'); if (empty($content)) { $content = shell_exec('cat /proc/loadavg'); } - if (empty($content)) { - return []; + if (empty($content) || !preg_match("#([.\d]+)\s([.\d]+)\s([.\d]+)\s(\d+)/(\d+)#", $content, $matches)) { + $load_arr = sys_getloadavg(); + if (empty($load_arr)) { + return []; + } + return [ + 'average1' => $load_arr[0], + 'average5' => $load_arr[1], + 'average15' => $load_arr[2], + 'runnable' => 0, + 'scheduled' => 0 + ]; } - if (!preg_match("#([.\d]+)\s([.\d]+)\s([.\d]+)\s(\d+)/(\d+)#", $content, $matches)) { - return []; - } return [ 'average1' => (float)$matches[1], 'average5' => (float)$matches[2], diff --git a/src/Core/Worker.php b/src/Core/Worker.php index 830e029832..036f065d42 100644 --- a/src/Core/Worker.php +++ b/src/Core/Worker.php @@ -143,7 +143,7 @@ class Worker } // Quit the worker once every cron interval - if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) { + if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60)) && !self::systemLimitReached()) { Logger::info('Process lifetime reached, respawning.'); self::unclaimProcess($process); if (Worker\Daemon::isMode()) { @@ -444,6 +444,87 @@ class Worker return true; } + /** + * Checks if system limits are reached. + * + * @return boolean + */ + private static function systemLimitReached(): bool + { + $load_cooldown = DI::config()->get('system', 'worker_load_cooldown'); + $processes_cooldown = DI::config()->get('system', 'worker_processes_cooldown'); + + if (($load_cooldown == 0) && ($processes_cooldown == 0)) { + return false; + } + + $load = System::getLoadAvg(); + if (empty($load)) { + return false; + } + + if (($load_cooldown > 0) && ($load['average1'] > $load_cooldown)) { + return true; + } + + if (($processes_cooldown > 0) && ($load['scheduled'] > $processes_cooldown)) { + return true; + } + + return false; + } + + /** + * Slow the execution down if the system load is too high + * + * @return void + */ + public static function coolDown() + { + $cooldown = DI::config()->get('system', 'worker_cooldown', 0); + if ($cooldown > 0) { + Logger::debug('Wait for cooldown.', ['cooldown' => $cooldown]); + if ($cooldown < 1) { + usleep($cooldown * 1000000); + } else { + sleep($cooldown); + } + } + + $load_cooldown = DI::config()->get('system', 'worker_load_cooldown'); + $processes_cooldown = DI::config()->get('system', 'worker_processes_cooldown'); + + if (($load_cooldown == 0) && ($processes_cooldown == 0)) { + return; + } + + $sleeping = false; + + while ($load = System::getLoadAvg()) { + 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)]); + $sleeping = true; + } + sleep(1); + continue; + } + 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)]); + $sleeping = true; + } + sleep(1); + continue; + } + break; + } + + if ($sleeping) { + Logger::notice('Cooldown ended.', ['max-load' => $load_cooldown, 'max-processes' => $processes_cooldown, 'load' => $load, 'called-by' => System::callstack(1)]); + } + } + /** * Execute a function from the queue * @@ -458,28 +539,7 @@ class Worker { $a = DI::app(); - $cooldown = DI::config()->get('system', 'worker_cooldown', 0); - if ($cooldown > 0) { - Logger::debug('Pre execution cooldown.', ['cooldown' => $cooldown, 'id' => $queue['id'], 'priority' => $queue['priority'], 'command' => $queue['command']]); - sleep($cooldown); - } - - $load_cooldown = DI::config()->get('system', 'worker_load_cooldown'); - $processes_cooldown = DI::config()->get('system', 'worker_processes_cooldown'); - - while ((($load_cooldown > 0) || ($processes_cooldown > 0)) && ($load = System::getLoadAvg())) { - if (($load_cooldown > 0) && ($load['average1'] > $load_cooldown)) { - Logger::debug('Load induced pre execution cooldown.', ['max' => $load_cooldown, 'load' => $load, 'id' => $queue['id'], 'priority' => $queue['priority'], 'command' => $queue['command']]); - sleep(1); - continue; - } - if (($processes_cooldown > 0) && ($load['scheduled'] > $processes_cooldown)) { - Logger::debug('Process induced pre execution cooldown.', ['max' => $processes_cooldown, 'load' => $load, 'id' => $queue['id'], 'priority' => $queue['priority'], 'command' => $queue['command']]); - sleep(1); - continue; - } - break; - } + self::coolDown(); Logger::enableWorker($funcname); @@ -527,6 +587,8 @@ class Worker Logger::info('Performance:', ['state' => self::$state, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'lock' => $dblock, 'total' => $dbtotal, 'rest' => $rest, 'exec' => $exec]); + self::coolDown(); + self::$up_start = microtime(true); self::$db_duration = 0; self::$db_duration_count = 0; @@ -547,11 +609,6 @@ class Worker Logger::info('Process done.', ['priority' => $queue['priority'], 'id' => $queue['id'], 'duration' => round($duration, 3)]); DI::profiler()->saveLog(DI::logger(), 'ID ' . $queue['id'] . ': ' . $funcname); - - if ($cooldown > 0) { - Logger::info('Post execution cooldown.', ['priority' => $queue['priority'], 'id' => $queue['id'], 'cooldown' => $cooldown]); - sleep($cooldown); - } } /** @@ -744,7 +801,7 @@ class Worker Logger::notice('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()) { + if (!DI::config()->get('system', 'worker_dont_fork', false) && ($queues > ($active + 1)) && self::entriesExists() && !self::systemLimitReached()) { Logger::info('There are fewer workers as possible, fork a new worker.', ['active' => $active, 'queues' => $queues]); if (Worker\Daemon::isMode()) { Worker\IPC::SetJobState(true); @@ -1222,7 +1279,7 @@ class Worker Worker\Daemon::checkState(); // Should we quit and wait for the worker to be called as a cronjob? - if ($dont_fork) { + if ($dont_fork || self::systemLimitReached()) { return $added; } diff --git a/src/Protocol/ActivityPub/Delivery.php b/src/Protocol/ActivityPub/Delivery.php index f66febb82d..3b8e539e9e 100644 --- a/src/Protocol/ActivityPub/Delivery.php +++ b/src/Protocol/ActivityPub/Delivery.php @@ -22,6 +22,7 @@ namespace Friendica\Protocol\ActivityPub; use Friendica\Core\Logger; +use Friendica\Core\Worker; use Friendica\Database\DBA; use Friendica\DI; use Friendica\Model\Contact; @@ -55,6 +56,7 @@ class Delivery Logger::notice('Inbox delivery has a server failure', ['inbox' => $inbox]); $serverfail = true; } + Worker::coolDown(); } if ($serverfail || (!$result['success'] && !$result['drop'])) { diff --git a/src/Worker/Notifier.php b/src/Worker/Notifier.php index 76eb260a08..3b4e1478c0 100644 --- a/src/Worker/Notifier.php +++ b/src/Worker/Notifier.php @@ -573,6 +573,7 @@ class Notifier if (Worker::add($deliver_options, 'Delivery', $cmd, $post_uriid, (int)$contact['id'], $sender_uid)) { $delivery_queue_count++; } + Worker::coolDown(); } return $delivery_queue_count; } @@ -695,6 +696,7 @@ class Notifier Logger::info('Account removal via ActivityPub', ['uid' => $self_user_id, 'inbox' => $inbox]); Worker::add(['priority' => PRIORITY_NEGLIGIBLE, 'created' => $created, 'dont_fork' => true], 'APDelivery', Delivery::REMOVAL, 0, $inbox, $self_user_id, $receivers); + Worker::coolDown(); } return true; @@ -818,6 +820,7 @@ class Notifier $delivery_queue_count++; } } + Worker::coolDown(); } // We deliver posts to relay servers slightly delayed to priorize the direct delivery @@ -833,6 +836,7 @@ class Notifier $delivery_queue_count++; } } + Worker::coolDown(); } return ['count' => $delivery_queue_count, 'contacts' => $contacts]; diff --git a/src/Worker/PollContacts.php b/src/Worker/PollContacts.php index b61f5aebb8..b4312ef0c3 100644 --- a/src/Worker/PollContacts.php +++ b/src/Worker/PollContacts.php @@ -81,6 +81,7 @@ class PollContacts Logger::notice("Polling " . $contact["network"] . " " . $contact["id"] . " " . $contact['priority'] . " " . $contact["nick"] . " " . $contact["name"]); Worker::add(['priority' => $priority, 'dont_fork' => true, 'force_priority' => true], 'OnePoll', (int)$contact['id']); + Worker::coolDown(); } DBA::close($contacts); } diff --git a/src/Worker/UpdateContacts.php b/src/Worker/UpdateContacts.php index 31426e79f2..7e352a27ca 100644 --- a/src/Worker/UpdateContacts.php +++ b/src/Worker/UpdateContacts.php @@ -62,6 +62,7 @@ class UpdateContacts if (Worker::add(['priority' => PRIORITY_LOW, 'dont_fork' => true], "UpdateContact", $contact['id'])) { ++$count; } + Worker::coolDown(); } DBA::close($contacts); diff --git a/src/Worker/UpdateGServers.php b/src/Worker/UpdateGServers.php index e537545846..108482eaec 100644 --- a/src/Worker/UpdateGServers.php +++ b/src/Worker/UpdateGServers.php @@ -72,6 +72,7 @@ class UpdateGServers $count++; } } + Worker::coolDown(); } DBA::close($gservers); Logger::info('Updated servers', ['count' => $count]); diff --git a/src/Worker/UpdateServerPeers.php b/src/Worker/UpdateServerPeers.php index 09c88499ec..8c916504ed 100644 --- a/src/Worker/UpdateServerPeers.php +++ b/src/Worker/UpdateServerPeers.php @@ -63,6 +63,7 @@ class UpdateServerPeers // This endpoint doesn't offer the schema. So we assume that it is HTTPS. GServer::add('https://' . $peer); ++$added; + Worker::coolDown(); } Logger::info('Server peer update ended', ['total' => $total, 'added' => $added, 'url' => $url]); } diff --git a/static/defaults.config.php b/static/defaults.config.php index fb6727491b..b02f537745 100644 --- a/static/defaults.config.php +++ b/static/defaults.config.php @@ -615,7 +615,7 @@ return [ // Minimum for this config value is 1. Maximum is 64 as the resulting profile URL mustn't be longer than 255 chars. 'username_max_length' => 48, - // worker_cooldown (Integer) + // worker_cooldown (Float) // Cooldown period in seconds before each worker function call. 'worker_cooldown' => 0,