X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FCore%2FWorker.php;h=723cd809da44d3230721afd1eb4e06cf5a17ffb7;hb=389685e09944e3823effd2e820ab2fd95d8b4b32;hp=f649e7773e3b8725129571b735996d974c62461a;hpb=af6dbc654f82225cfc647fe2072662acae388e47;p=friendica.git

diff --git a/src/Core/Worker.php b/src/Core/Worker.php
index f649e7773e..723cd809da 100644
--- a/src/Core/Worker.php
+++ b/src/Core/Worker.php
@@ -5,10 +5,10 @@
 namespace Friendica\Core;
 
 use Friendica\Database\DBA;
-use Friendica\Database\DBM;
 use Friendica\Model\Process;
 use Friendica\Util\DateTimeFormat;
 use Friendica\Util\Network;
+use Friendica\BaseObject;
 
 require_once 'include/dba.php';
 
@@ -63,7 +63,7 @@ class Worker
 		}
 
 		// Do we have too few memory?
-		if ($a->min_memory_reached()) {
+		if ($a->isMinMemoryReached()) {
 			logger('Pre check: Memory limit reached, quitting.', LOGGER_DEBUG);
 			return;
 		}
@@ -118,12 +118,14 @@ class Worker
 				// Count active workers and compare them with a maximum value that depends on the load
 				if (self::tooMuchWorkers()) {
 					logger('Active worker limit reached, quitting.', LOGGER_DEBUG);
+					Lock::release('worker');
 					return;
 				}
 
 				// Check free memory
-				if ($a->min_memory_reached()) {
+				if ($a->isMinMemoryReached()) {
 					logger('Memory limit reached, quitting.', LOGGER_DEBUG);
+					Lock::release('worker');
 					return;
 				}
 				Lock::release('worker');
@@ -151,7 +153,8 @@ class Worker
 	 */
 	private static function totalEntries()
 	{
-		return DBA::count('workerqueue', ["`executed` <= ? AND NOT `done`", NULL_DATE]);
+		return DBA::count('workerqueue', ["`executed` <= ? AND NOT `done` AND `next_try` < ?",
+			NULL_DATE, DateTimeFormat::utcNow()]);
 	}
 
 	/**
@@ -161,9 +164,9 @@ class Worker
 	 */
 	private static function highestPriority()
 	{
-		$condition = ["`executed` <= ? AND NOT `done`", NULL_DATE];
+		$condition = ["`executed` <= ? AND NOT `done` AND `next_try` < ?", NULL_DATE, DateTimeFormat::utcNow()];
 		$workerqueue = DBA::selectFirst('workerqueue', ['priority'], $condition, ['order' => ['priority']]);
-		if (DBM::is_result($workerqueue)) {
+		if (DBA::isResult($workerqueue)) {
 			return $workerqueue["priority"];
 		} else {
 			return 0;
@@ -179,7 +182,8 @@ class Worker
 	 */
 	private static function processWithPriorityActive($priority)
 	{
-		$condition = ["`priority` <= ? AND `executed` > ? AND NOT `done`", $priority, NULL_DATE];
+		$condition = ["`priority` <= ? AND `executed` > ? AND NOT `done` AND `next_try` < ?",
+			$priority, NULL_DATE, DateTimeFormat::utcNow()];
 		return DBA::exists('workerqueue', $condition);
 	}
 
@@ -239,7 +243,7 @@ class Worker
 			self::execFunction($queue, $include, $argv, true);
 
 			$stamp = (float)microtime(true);
-			if (DBA::update('workerqueue', ['done' => true], ['id' => $queue["id"]])) {
+			if (DBA::update('workerqueue', ['done' => true], ['id' => $queue['id']])) {
 				Config::set('system', 'last_worker_execution', DateTimeFormat::utcNow());
 			}
 			self::$db_duration = (microtime(true) - $stamp);
@@ -477,8 +481,8 @@ class Worker
 
 		if ($max == 0) {
 			// the maximum number of possible user connections can be a system variable
-			$r = DBA::fetch_first("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
-			if (DBM::is_result($r)) {
+			$r = DBA::fetchFirst("SHOW VARIABLES WHERE `variable_name` = 'max_user_connections'");
+			if (DBA::isResult($r)) {
 				$max = $r["Value"];
 			}
 			// Or it can be granted. This overrides the system variable
@@ -498,7 +502,7 @@ class Worker
 		// The processlist only shows entries of the current user
 		if ($max != 0) {
 			$r = DBA::p('SHOW PROCESSLIST');
-			$used = DBA::num_rows($r);
+			$used = DBA::numRows($r);
 			DBA::close($r);
 
 			logger("Connection usage (user values): ".$used."/".$max, LOGGER_DEBUG);
@@ -513,16 +517,16 @@ class Worker
 
 		// We will now check for the system values.
 		// This limit could be reached although the user limits are fine.
-		$r = DBA::fetch_first("SHOW VARIABLES WHERE `variable_name` = 'max_connections'");
-		if (!DBM::is_result($r)) {
+		$r = DBA::fetchFirst("SHOW VARIABLES WHERE `variable_name` = 'max_connections'");
+		if (!DBA::isResult($r)) {
 			return false;
 		}
 		$max = intval($r["Value"]);
 		if ($max == 0) {
 			return false;
 		}
-		$r = DBA::fetch_first("SHOW STATUS WHERE `variable_name` = 'Threads_connected'");
-		if (!DBM::is_result($r)) {
+		$r = DBA::fetchFirst("SHOW STATUS WHERE `variable_name` = 'Threads_connected'");
+		if (!DBA::isResult($r)) {
 			return false;
 		}
 		$used = intval($r["Value"]);
@@ -577,7 +581,7 @@ class Worker
 				// How long is the process already running?
 				$duration = (time() - strtotime($entry["executed"])) / 60;
 				if ($duration > $max_duration) {
-					logger("Worker process ".$entry["pid"]." (".implode(" ", $argv).") took more than ".$max_duration." minutes. It will be killed now.");
+					logger("Worker process ".$entry["pid"]." (".substr(json_encode($argv), 0, 50).") took more than ".$max_duration." minutes. It will be killed now.");
 					posix_kill($entry["pid"], SIGTERM);
 
 					// We killed the stale process.
@@ -617,7 +621,7 @@ class Worker
 		$active = self::activeWorkers();
 
 		// Decrease the number of workers at higher load
-		$load = current_load();
+		$load = System::currentLoad();
 		if ($load) {
 			$maxsysload = intval(Config::get("system", "maxloadavg", 50));
 
@@ -735,7 +739,7 @@ class Worker
 		);
 
 		// No active processes at all? Fine
-		if (!DBM::is_result($r)) {
+		if (!DBA::isResult($r)) {
 			return false;
 		}
 		$priorities = [];
@@ -804,7 +808,8 @@ class Worker
 			$result = DBA::select(
 				'workerqueue',
 				['id'],
-				["`executed` <= ? AND `priority` < ? AND NOT `done`", NULL_DATE, $highest_priority],
+				["`executed` <= ? AND `priority` < ? AND NOT `done` AND `next_try` < ?",
+				NULL_DATE, $highest_priority, DateTimeFormat::utcNow()],
 				['limit' => $limit, 'order' => ['priority', 'created']]
 			);
 
@@ -820,7 +825,8 @@ class Worker
 				$result = DBA::select(
 					'workerqueue',
 					['id'],
-					["`executed` <= ? AND `priority` > ? AND NOT `done`", NULL_DATE, $highest_priority],
+					["`executed` <= ? AND `priority` > ? AND NOT `done` AND `next_try` < ?",
+					NULL_DATE, $highest_priority, DateTimeFormat::utcNow()],
 					['limit' => $limit, 'order' => ['priority', 'created']]
 				);
 
@@ -839,7 +845,8 @@ class Worker
 			$result = DBA::select(
 				'workerqueue',
 				['id'],
-				["`executed` <= ? AND NOT `done`", NULL_DATE],
+				["`executed` <= ? AND NOT `done` AND `next_try` < ?",
+				NULL_DATE, DateTimeFormat::utcNow()],
 				['limit' => $limit, 'order' => ['priority', 'created']]
 			);
 
@@ -872,9 +879,9 @@ class Worker
 
 		// There can already be jobs for us in the queue.
 		$r = DBA::select('workerqueue', [], ['pid' => getmypid(), 'done' => false]);
-		if (DBM::is_result($r)) {
+		if (DBA::isResult($r)) {
 			self::$db_duration += (microtime(true) - $stamp);
-			return DBA::inArray($r);
+			return DBA::toArray($r);
 		}
 		DBA::close($r);
 
@@ -892,7 +899,7 @@ class Worker
 
 		if ($found) {
 			$r = DBA::select('workerqueue', [], ['pid' => getmypid(), 'done' => false]);
-			return DBA::inArray($r);
+			return DBA::toArray($r);
 		}
 		return false;
 	}
@@ -1010,13 +1017,11 @@ class Worker
 	 */
 	public static function spawnWorker($do_cron = false)
 	{
-		$args = ["bin/worker.php"];
+		$command = 'bin/worker.php';
 
-		if (!$do_cron) {
-			$args[] = "no_cron";
-		}
+		$args = ['no_cron' => !$do_cron];
 
-		get_app()->proc_run($args);
+		get_app()->proc_run($command, $args);
 
 		// after spawning we have to remove the flag.
 		if (Config::get('system', 'worker_daemon_mode', false)) {
@@ -1117,6 +1122,35 @@ class Worker
 		return true;
 	}
 
+	/**
+	 * Defers the current worker entry
+	 */
+	public static function defer()
+	{
+		if (empty(BaseObject::getApp()->queue)) {
+			return;
+		}
+
+		$queue = BaseObject::getApp()->queue;
+
+		$retrial = $queue['retrial'];
+		$id = $queue['id'];
+
+		if ($retrial > 14) {
+			logger('Id ' . $id . ' had been tried 14 times, it will be deleted now.', LOGGER_DEBUG);
+			DBA::delete('workerqueue', ['id' => $id]);
+		}
+
+		// Calculate the delay until the next trial
+		$delay = (($retrial + 3) ** 4) + (rand(1, 30) * ($retrial + 1));
+		$next = DateTimeFormat::utc('now + ' . $delay . ' seconds');
+
+		logger('Defer execution ' . $retrial . ' of id ' . $id . ' to ' . $next, LOGGER_DEBUG);
+
+		$fields = ['retrial' => $retrial + 1, 'next_try' => $next, 'executed' => NULL_DATE, 'pid' => 0];
+		DBA::update('workerqueue', $fields, ['id' => $id]);
+	}
+
 	/**
 	 * Log active processes into the "process" table
 	 *
@@ -1166,7 +1200,7 @@ class Worker
 		$row = DBA::selectFirst('worker-ipc', ['jobs'], ['key' => 1]);
 
 		// When we don't have a row, no job is running
-		if (!DBM::is_result($row)) {
+		if (!DBA::isResult($row)) {
 			return false;
 		}