X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FCore%2FWorker.php;h=296d40b5046e182ba86cb219f8384f272f170314;hb=3792e2819d8b2f7fe96a0260a0f280712b225f74;hp=4539dcc84372a36fcc5ae2eac2791254e8ace480;hpb=e1216e3a6079d20f0e525193f500ae2141a29ffd;p=friendica.git

diff --git a/src/Core/Worker.php b/src/Core/Worker.php
index 4539dcc843..296d40b504 100644
--- a/src/Core/Worker.php
+++ b/src/Core/Worker.php
@@ -4,15 +4,11 @@
  */
 namespace Friendica\Core;
 
-use Friendica\Core\Addon;
-use Friendica\Core\Config;
-use Friendica\Core\System;
+use dba;
 use Friendica\Database\DBM;
 use Friendica\Model\Process;
 use Friendica\Util\DateTimeFormat;
-use Friendica\Util\Lock;
 use Friendica\Util\Network;
-use dba;
 
 require_once 'include/dba.php';
 
@@ -53,11 +49,6 @@ class Worker
 		// We now start the process. This is done after the load check since this could increase the load.
 		self::startProcess();
 
-		// The daemon doesn't need to fork new workers anymore, since we started a process
-		if (Config::get('system', 'worker_daemon_mode', false)) {
-			self::IPCSetJobState(false);
-		}
-
 		// Kill stale processes every 5 minutes
 		$last_cleanup = Config::get('system', 'worker_last_cleaned', 0);
 		if (time() > ($last_cleanup + 300)) {
@@ -113,16 +104,16 @@ class Worker
 				}
 
 				// If possible we will fetch new jobs for this worker
-				if (!$refetched && Lock::set('worker_process', 0)) {
+				if (!$refetched && Lock::acquire('worker_process', 0)) {
 					$stamp = (float)microtime(true);
 					$refetched = self::findWorkerProcesses($passing_slow);
 					self::$db_duration += (microtime(true) - $stamp);
-					Lock::remove('worker_process');
+					Lock::release('worker_process');
 				}
 			}
 
 			// To avoid the quitting of multiple workers only one worker at a time will execute the check
-			if (Lock::set('worker', 0)) {
+			if (Lock::acquire('worker', 0)) {
 				$stamp = (float)microtime(true);
 				// Count active workers and compare them with a maximum value that depends on the load
 				if (self::tooMuchWorkers()) {
@@ -135,7 +126,7 @@ class Worker
 					logger('Memory limit reached, quitting.', LOGGER_DEBUG);
 					return;
 				}
-				Lock::remove('worker');
+				Lock::release('worker');
 				self::$db_duration += (microtime(true) - $stamp);
 			}
 
@@ -145,10 +136,12 @@ class Worker
 				return;
 			}
 		}
+
+		// Cleaning up. Possibly not needed, but it doesn't harm anything.
 		if (Config::get('system', 'worker_daemon_mode', false)) {
 			self::IPCSetJobState(false);
 		}
-		logger("Couldn't select a workerqueue entry, quitting.", LOGGER_DEBUG);
+		logger("Couldn't select a workerqueue entry, quitting process " . getmypid() . ".", LOGGER_DEBUG);
 	}
 
 	/**
@@ -158,12 +151,7 @@ class Worker
 	 */
 	private static function totalEntries()
 	{
-		$s = dba::fetch_first("SELECT COUNT(*) AS `total` FROM `workerqueue` WHERE `executed` <= ? AND NOT `done`", NULL_DATE);
-		if (DBM::is_result($s)) {
-			return $s["total"];
-		} else {
-			return 0;
-		}
+		return dba::count('workerqueue', ["`executed` <= ? AND NOT `done`", NULL_DATE]);
 	}
 
 	/**
@@ -321,7 +309,7 @@ class Worker
 
 		$argc = count($argv);
 
-		$new_process_id = uniqid("wrk", true);
+		$new_process_id = System::processID("wrk");
 
 		logger("Process ".$mypid." - Prio ".$queue["priority"]." - ID ".$queue["id"].": ".$funcname." ".$queue["parameter"]." - Process PID: ".$new_process_id);
 
@@ -633,12 +621,15 @@ class Worker
 		if ($load) {
 			$maxsysload = intval(Config::get("system", "maxloadavg", 50));
 
-			$maxworkers = $queues;
+			/* Default exponent 3 causes queues to rapidly decrease as load increases.
+			 * If you have 20 max queues at idle, then you get only 5 queues at 37.1% of $maxsysload.
+			 * For some environments, this rapid decrease is not needed.
+			 * With exponent 1, you could have 20 max queues at idle and 13 at 37% of $maxsysload.
+			 */
+			$exponent = intval(Config::get('system', 'worker_load_exponent', 3));
+			$slope = pow(max(0, $maxsysload - $load) / $maxsysload, $exponent);
+			$queues = intval(ceil($slope * $maxqueues));
 
-			// Some magical mathemathics to reduce the workers
-			$exponent = 3;
-			$slope = $maxworkers / pow($maxsysload, $exponent);
-			$queues = ceil($slope * pow(max(0, $maxsysload - $load), $exponent));
 			$processlist = '';
 
 			if (Config::get('system', 'worker_debug')) {
@@ -698,11 +689,20 @@ class Worker
 			// Are there fewer workers running as possible? Then fork a new one.
 			if (!Config::get("system", "worker_dont_fork", false) && ($queues > ($active + 1)) && ($entries > 1)) {
 				logger("Active workers: ".$active."/".$queues." Fork a new worker.", LOGGER_DEBUG);
-				self::spawnWorker();
+				if (Config::get('system', 'worker_daemon_mode', false)) {
+					self::IPCSetJobState(true);
+				} else {
+					self::spawnWorker();
+				}
 			}
 		}
 
-		return $active >= $queues;
+		// if there are too much worker, we don't spawn a new one.
+		if (Config::get('system', 'worker_daemon_mode', false) && ($active > $queues)) {
+			self::IPCSetJobState(false);
+		}
+
+		return $active > $queues;
 	}
 
 	/**
@@ -712,9 +712,7 @@ class Worker
 	 */
 	private static function activeWorkers()
 	{
-		$workers = dba::fetch_first("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'Worker.php'");
-
-		return $workers["processes"];
+		return dba::count('process', ['command' => 'Worker.php']);
 	}
 
 	/**
@@ -859,11 +857,6 @@ class Worker
 			dba::update('workerqueue', ['executed' => DateTimeFormat::utcNow(), 'pid' => $mypid], $ids);
 		}
 
-		// The daemon doesn't need to fork new workers anymore, since we are inside the worker
-		if (Config::get('system', 'worker_daemon_mode', false)) {
-			self::IPCSetJobState(false);
-		}
-
 		return $found;
 	}
 
@@ -886,7 +879,7 @@ class Worker
 		dba::close($r);
 
 		$stamp = (float)microtime(true);
-		if (!Lock::set('worker_process')) {
+		if (!Lock::acquire('worker_process')) {
 			return false;
 		}
 		self::$lock_duration = (microtime(true) - $stamp);
@@ -895,7 +888,7 @@ class Worker
 		$found = self::findWorkerProcesses($passing_slow);
 		self::$db_duration += (microtime(true) - $stamp);
 
-		Lock::remove('worker_process');
+		Lock::release('worker_process');
 
 		if ($found) {
 			$r = dba::select('workerqueue', [], ['pid' => getmypid(), 'done' => false]);
@@ -972,9 +965,9 @@ class Worker
 
 		self::clearProcesses();
 
-		$workers = dba::fetch_first("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'worker.php'");
+		$workers = self::activeWorkers();
 
-		if ($workers["processes"] == 0) {
+		if ($workers == 0) {
 			self::callWorker();
 		}
 	}
@@ -1024,6 +1017,11 @@ class Worker
 		}
 
 		get_app()->proc_run($args);
+
+		// after spawning we have to remove the flag.
+		if (Config::get('system', 'worker_daemon_mode', false)) {
+			self::IPCSetJobState(false);
+		}
 	}
 
 	/**
@@ -1094,25 +1092,25 @@ class Worker
 			return true;
 		}
 
-		// We tell the daemon that a new job entry exists
-		if (Config::get('system', 'worker_daemon_mode', false)) {
-			self::IPCSetJobState(true);
-			return true;
-		}
-
 		// If there is a lock then we don't have to check for too much worker
-		if (!Lock::set('worker', 0)) {
+		if (!Lock::acquire('worker', 0)) {
 			return true;
 		}
 
 		// If there are already enough workers running, don't fork another one
 		$quit = self::tooMuchWorkers();
-		Lock::remove('worker');
+		Lock::release('worker');
 
 		if ($quit) {
 			return true;
 		}
 
+		// We tell the daemon that a new job entry exists
+		if (Config::get('system', 'worker_daemon_mode', false)) {
+			// We don't have to set the IPC flag - this is done in "tooMuchWorkers"
+			return true;
+		}
+
 		// Now call the worker to execute the jobs that we just added to the queue
 		self::spawnWorker();