clones/friendica
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Let the worker run for an hour in daemon mode

This commit is contained in:
Michael 2021-01-04 09:20:44 +00:00
parent 74d7d7e164
commit 69c7e9af20
2 changed files with 92 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
bin/daemon.php
View File

@ -224,8 +224,10 @@ while (true) {
usleep($sleep); usleep($sleep);
$pid = pcntl_waitpid(-1, $status, WNOHANG); $pid = pcntl_waitpid(-1, $status, WNOHANG);
Logger::info('Checked children status via pcntl_waitpid', ['pid' => $pid, 'status' => $status]); if ($pid > 0) {
Logger::info('Children quit via pcntl_waitpid', ['pid' => $pid, 'status' => $status]);
}
$timeout = ($seconds >= $wait_interval); $timeout = ($seconds >= $wait_interval);
} while (!$timeout && !Worker::IPCJobsExists()); } while (!$timeout && !Worker::IPCJobsExists());

142
src/Core/Worker.php
View File

@ -94,67 +94,101 @@ class Worker
$last_check = $starttime = time(); $last_check = $starttime = time();
self::$state = self::STATE_STARTUP; self::$state = self::STATE_STARTUP;
$wait_interval = self::isDaemonMode() ? 360 : 10;
$start = time();
// We fetch the next queue entry that is about to be executed do {
while ($r = self::workerProcess()) { // We fetch the next queue entry that is about to be executed
// Don't refetch when a worker fetches tasks for multiple workers while ($r = self::workerProcess()) {
$refetched = DI::config()->get('system', 'worker_multiple_fetch'); // Don't refetch when a worker fetches tasks for multiple workers
foreach ($r as $entry) { $refetched = DI::config()->get('system', 'worker_multiple_fetch');
// Assure that the priority is an integer value foreach ($r as $entry) {
$entry['priority'] = (int)$entry['priority']; // Assure that the priority is an integer value
$entry['priority'] = (int)$entry['priority'];
// The work will be done // The work will be done
if (!self::execute($entry)) { if (!self::execute($entry)) {
Logger::notice('Process execution failed, quitting.'); Logger::notice('Process execution failed, quitting.');
return;
}
// Trying to fetch new processes - but only once when successful
if (!$refetched && DI::lock()->acquire(self::LOCK_PROCESS, 0)) {
self::findWorkerProcesses();
DI::lock()->release(self::LOCK_PROCESS);
self::$state = self::STATE_REFETCH;
$refetched = true;
} else {
self::$state = self::STATE_SHORT_LOOP;
}
}
// To avoid the quitting of multiple workers only one worker at a time will execute the check
if ((time() > $last_check + 5) && !self::getWaitingJobForPID()) {
self::$state = self::STATE_LONG_LOOP;
if (DI::lock()->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.');
DI::lock()->release(self::LOCK_WORKER);
return;
}
// Check free memory
if (DI::process()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.');
DI::lock()->release(self::LOCK_WORKER);
return;
}
DI::lock()->release(self::LOCK_WORKER);
}
$last_check = time();
}
// Quit the worker once every cron interval
if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
Logger::info('Process lifetime reached, respawning.');
self::unclaimProcess();
if (self::isDaemonMode()) {
self::IPCSetJobState(true);
} else {
self::spawnWorker();
}
return;
}
$start = time();
}
$seconds = (time() - $start);
// logarithmic wait time calculation.
$arg = (($seconds + 1) / ($wait_interval / 9)) + 1;
$sleep = min(1000000, round(log10($arg) * 1000000, 0));
usleep($sleep);
$timeout = ($seconds >= $wait_interval);
Logger::info('Timeout', ['timeout' => $timeout, 'seconds' => $seconds, 'sleep' => $sleep]);
if (!$timeout) {
if (DI::process()->isMaxLoadReached()) {
Logger::notice('maximum load reached, quitting.');
return; return;
} }
// Trying to fetch new processes - but only once when successful // Kill stale processes every 5 minutes
if (!$refetched && DI::lock()->acquire(self::LOCK_PROCESS, 0)) { $last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0);
self::findWorkerProcesses(); if (time() > ($last_cleanup + 300)) {
DI::lock()->release(self::LOCK_PROCESS); DI::config()->set('system', 'worker_last_cleaned', time());
self::$state = self::STATE_REFETCH; self::killStaleWorkers();
$refetched = true;
} else {
self::$state = self::STATE_SHORT_LOOP;
} }
// Check if the system is ready
if (!self::isReady()) {
return;
}
} }
} while (!$timeout);
// To avoid the quitting of multiple workers only one worker at a time will execute the check
if ((time() > $last_check + 5) && !self::getWaitingJobForPID()) {
self::$state = self::STATE_LONG_LOOP;
if (DI::lock()->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.');
DI::lock()->release(self::LOCK_WORKER);
return;
}
// Check free memory
if (DI::process()->isMinMemoryReached()) {
Logger::warning('Memory limit reached, quitting.');
DI::lock()->release(self::LOCK_WORKER);
return;
}
DI::lock()->release(self::LOCK_WORKER);
}
$last_check = time();
}
// Quit the worker once every cron interval
if (time() > ($starttime + (DI::config()->get('system', 'cron_interval') * 60))) {
Logger::info('Process lifetime reached, respawning.');
self::unclaimProcess();
if (self::isDaemonMode()) {
self::IPCSetJobState(true);
} else {
self::spawnWorker();
}
return;
}
}
// Cleaning up. Possibly not needed, but it doesn't harm anything. // Cleaning up. Possibly not needed, but it doesn't harm anything.
if (self::isDaemonMode()) { if (self::isDaemonMode()) {