From 0fa281247d9707bc448ebb853f9051eba3e4304e Mon Sep 17 00:00:00 2001 From: Michael Vogel Date: Wed, 21 Sep 2022 21:03:07 +0200 Subject: [PATCH] More cooldown calls for worker processes --- bin/daemon.php | 5 ++- src/Core/Worker.php | 62 +++++++++++++++++++-------- src/Protocol/ActivityPub/Delivery.php | 2 + src/Worker/Notifier.php | 4 ++ src/Worker/PollContacts.php | 1 + src/Worker/UpdateContacts.php | 1 + src/Worker/UpdateGServers.php | 1 + src/Worker/UpdateServerPeers.php | 1 + 8 files changed, 58 insertions(+), 19 deletions(-) diff --git a/bin/daemon.php b/bin/daemon.php index 880115d0d..f1eb077ad 100755 --- a/bin/daemon.php +++ b/bin/daemon.php @@ -76,8 +76,8 @@ DI::config()->load(); if (empty(DI::config()->get('system', 'pidfile'))) { die(<< [ + + 'system' => [ 'pidfile' => '/path/to/daemon.pid', ], TXT @@ -199,6 +199,7 @@ while (true) { } if ($do_cron || (!DI::system()->isMaxLoadReached() && Worker::entriesExists() && Worker::isReady())) { + Worker::coolDown(); Worker::spawnWorker($do_cron); } else { Logger::info('Cool down for 5 seconds', ['pid' => $pid]); diff --git a/src/Core/Worker.php b/src/Core/Worker.php index 830e02983..4496caf92 100644 --- a/src/Core/Worker.php +++ b/src/Core/Worker.php @@ -444,6 +444,47 @@ class Worker return true; } + /** + * Slow the execution down if the system load is too high + * + * @return void + */ + public static function 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 * @@ -460,26 +501,11 @@ class Worker $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']]); + Logger::notice('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 +553,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; diff --git a/src/Protocol/ActivityPub/Delivery.php b/src/Protocol/ActivityPub/Delivery.php index 9aa45d041..49a2b5aa4 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 7921097d4..b236abc19 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 b61f5aebb..b4312ef0c 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 31426e79f..7e352a27c 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 e53754584..108482eae 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 09c88499e..8c916504e 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]); }