mexon/friendica
1
0
Fork 0
Code Pull requests Activity

More cooldown calls for worker processes

Browse source
This commit is contained in:
Michael Vogel 2022-09-21 21:03:07 +02:00
parent 3f3e1a2bf4
commit 0fa281247d
8 changed files with 58 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
bin/daemon.php
View file

@ -76,8 +76,8 @@ DI::config()->load();
if (empty(DI::config()->get('system', 'pidfile'))) {
die(<<<TXT
Please set system.pidfile in config/local.config.php. For example:
'system' => [
'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]);

62
src/Core/Worker.php
View file

@ -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;

2
src/Protocol/ActivityPub/Delivery.php
View file

@ -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'])) {

4
src/Worker/Notifier.php
View file

@ -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];

1
src/Worker/PollContacts.php
View file

@ -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);
}

1
src/Worker/UpdateContacts.php
View file

@ -62,6 +62,7 @@ class UpdateContacts
if (Worker::add(['priority' => PRIORITY_LOW, 'dont_fork' => true], "UpdateContact", $contact['id'])) {
++$count;
}
Worker::coolDown();
}
DBA::close($contacts);

1
src/Worker/UpdateGServers.php
View file

@ -72,6 +72,7 @@ class UpdateGServers
$count++;
}
}
Worker::coolDown();
}
DBA::close($gservers);
Logger::info('Updated servers', ['count' => $count]);

1
src/Worker/UpdateServerPeers.php
View file

@ -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]);
}