X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FCore%2FWorker.php;h=037590c90d7783e025bdf492353114b34972bb73;hb=2c73377b3911800f5990b1ba3c090c3bc2ab5c69;hp=548547022476d91c8f75e7c421583467a8176efd;hpb=151c026a8aa5991c8f73b0974f479732c482cc45;p=friendica.git

diff --git a/src/Core/Worker.php b/src/Core/Worker.php
index 5485470224..037590c90d 100644
--- a/src/Core/Worker.php
+++ b/src/Core/Worker.php
@@ -5,10 +5,10 @@
 namespace Friendica\Core;
 
 use Friendica\BaseObject;
+use Friendica\Core;
 use Friendica\Database\DBA;
 use Friendica\Model\Process;
 use Friendica\Util\DateTimeFormat;
-use Friendica\Util\Logger\WorkerLogger;
 use Friendica\Util\Network;
 
 /**
@@ -187,7 +187,7 @@ class Worker
 	private static function deferredEntries()
 	{
 		$stamp = (float)microtime(true);
-		$count = DBA::count('workerqueue', ["NOT `done` AND `pid` = 0 AND `next_try` > ?", DateTimeFormat::utcNow()]);
+		$count = DBA::count('workerqueue', ["NOT `done` AND `pid` = 0 AND `retrial` > ?", 0]);
 		self::$db_duration += (microtime(true) - $stamp);
 		self::$db_duration_count += (microtime(true) - $stamp);
 		return $count;
@@ -378,10 +378,9 @@ class Worker
 
 		$argc = count($argv);
 
-		$logger = $a->getLogger();
-		$workerLogger = new WorkerLogger($logger, $funcname);
+		Logger::enableWorker($funcname);
 
-		$workerLogger ->info("Process start.", ['priority' => $queue["priority"], 'id' => $queue["id"]]);
+		Logger::info("Process start.", ['priority' => $queue["priority"], 'id' => $queue["id"]]);
 
 		$stamp = (float)microtime(true);
 
@@ -397,13 +396,13 @@ class Worker
 		unset($_SESSION);
 
 		// Set the workerLogger as new default logger
-		Logger::init($workerLogger);
 		if ($method_call) {
 			call_user_func_array(sprintf('Friendica\Worker\%s::execute', $funcname), $argv);
 		} else {
 			$funcname($argv, $argc);
 		}
-		Logger::init($logger);
+
+		Logger::disableWorker();
 
 		unset($a->queue);
 
@@ -423,7 +422,7 @@ class Worker
 		$rest    = round(max(0, $up_duration - (self::$db_duration + self::$lock_duration)), 2);
 		$exec    = round($duration, 2);
 
-		$logger->info('Performance:', ['state' => self::$state, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'lock' => $dblock, 'total' => $dbtotal, 'rest' => $rest, 'exec' => $exec]);
+		Logger::info('Performance:', ['state' => self::$state, 'count' => $dbcount, 'stat' => $dbstat, 'write' => $dbwrite, 'lock' => $dblock, 'total' => $dbtotal, 'rest' => $rest, 'exec' => $exec]);
 
 		self::$up_start = microtime(true);
 		self::$db_duration = 0;
@@ -433,23 +432,23 @@ class Worker
 		self::$lock_duration = 0;
 
 		if ($duration > 3600) {
-			$logger->info('Longer than 1 hour.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
+			Logger::info('Longer than 1 hour.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
 		} elseif ($duration > 600) {
-			$logger->info('Longer than 10 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
+			Logger::info('Longer than 10 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
 		} elseif ($duration > 300) {
-			$logger->info('Longer than 5 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
+			Logger::info('Longer than 5 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
 		} elseif ($duration > 120) {
-			$logger->info('Longer than 2 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
+			Logger::info('Longer than 2 minutes.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration/60, 3)]);
 		}
 
-		$workerLogger->info('Process done.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration, 3)]);
+		Logger::info('Process done.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'duration' => round($duration, 3)]);
 
 		$a->getProfiler()->saveLog($a->getLogger(), "ID " . $queue["id"] . ": " . $funcname);
 
 		$cooldown = Config::get("system", "worker_cooldown", 0);
 
 		if ($cooldown > 0) {
-			$logger->info('Cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
+			Logger::info('Cooldown.', ['priority' => $queue["priority"], 'id' => $queue["id"], 'cooldown' => $cooldown]);
 			sleep($cooldown);
 		}
 	}
@@ -688,10 +687,8 @@ class Worker
 					DBA::close($processes);
 				}
 				DBA::close($jobs);
-				$entries = $deferred + $waiting_processes;
 			} else {
-				$entries = self::totalEntries();
-				$waiting_processes = max(0, $entries - $deferred);
+				$waiting_processes =  self::totalEntries();
 				$stamp = (float)microtime(true);
 				$jobs = DBA::p("SELECT COUNT(*) AS `running`, `priority` FROM `process` INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `done` GROUP BY `priority` ORDER BY `priority`");
 				self::$db_duration += (microtime(true) - $stamp);
@@ -1086,7 +1083,9 @@ class Worker
 
 		$args = ['no_cron' => !$do_cron];
 
-		get_app()->proc_run($command, $args);
+		$a = get_app();
+		$process = new Core\Process($a->getLogger(), $a->getMode(), $a->getConfig(), $a->getBasePath());
+		$process->run($command, $args);
 
 		// after spawning we have to remove the flag.
 		if (Config::get('system', 'worker_daemon_mode', false)) {
@@ -1100,7 +1099,7 @@ class Worker
 	 * @param (integer|array) priority or parameter array, strings are deprecated and are ignored
 	 *
 	 * next args are passed as $cmd command line
-	 * or: Worker::add(PRIORITY_HIGH, "Notifier", "drop", $drop_id);
+	 * or: Worker::add(PRIORITY_HIGH, "Notifier", Delivery::DELETION, $drop_id);
 	 * or: Worker::add(array('priority' => PRIORITY_HIGH, 'dont_fork' => true), "CreateShadowEntry", $post_id);
 	 *
 	 * @return boolean "false" if proc_run couldn't be executed
@@ -1127,7 +1126,8 @@ class Worker
 		}
 
 		$priority = PRIORITY_MEDIUM;
-		$dont_fork = Config::get("system", "worker_dont_fork", false);
+		// Don't fork from frontend tasks by default
+		$dont_fork = Config::get("system", "worker_dont_fork", false) || !\get_app()->getMode()->isBackend();
 		$created = DateTimeFormat::utcNow();
 		$force_priority = false;
 
@@ -1194,6 +1194,32 @@ class Worker
 		return true;
 	}
 
+	/**
+	 * Returns the next retrial level for worker jobs.
+	 * This function will skip levels when jobs are older.
+	 *
+	 * @param array $queue Worker queue entry
+	 * @param integer $max_level maximum retrial level
+	 * @return integer the next retrial level value
+	 */
+	private static function getNextRetrial($queue, $max_level)
+	{
+		$created = strtotime($queue['created']);
+		$retrial_time = time() - $created;
+
+		$new_retrial = $queue['retrial'] + 1;
+		$total = 0;
+		for ($retrial = 0; $retrial <= $max_level + 1; ++$retrial) {
+			$delay = (($retrial + 3) ** 4) + (rand(1, 30) * ($retrial + 1));
+			$total += $delay;
+			if (($total < $retrial_time) && ($retrial > $queue['retrial'])) {
+				$new_retrial = $retrial;
+			}
+		}
+		Logger::info('New retrial for task', ['id' => $queue['id'], 'created' => $queue['created'], 'old' => $queue['retrial'], 'new' => $new_retrial]);
+		return $new_retrial;
+	}
+
 	/**
 	 * Defers the current worker entry
 	 */
@@ -1209,27 +1235,31 @@ class Worker
 		$id = $queue['id'];
 		$priority = $queue['priority'];
 
-		if ($retrial > 14) {
-			Logger::log('Id ' . $id . ' had been tried 14 times. We stop now.', Logger::DEBUG);
+		$max_level = Config::get('system', 'worker_defer_limit');
+
+		$new_retrial = self::getNextRetrial($queue, $max_level);
+
+		if ($new_retrial > $max_level) {
+			Logger::info('The task exceeded the maximum retry count', ['id' => $id, 'max_level' => $max_level, 'retrial' => $new_retrial]);
 			return;
 		}
 
 		// Calculate the delay until the next trial
-		$delay = (($retrial + 3) ** 4) + (rand(1, 30) * ($retrial + 1));
+		$delay = (($new_retrial + 2) ** 4) + (rand(1, 30) * ($new_retrial));
 		$next = DateTimeFormat::utc('now + ' . $delay . ' seconds');
 
-		if (($priority < PRIORITY_MEDIUM) && ($retrial > 2)) {
+		if (($priority < PRIORITY_MEDIUM) && ($new_retrial > 3)) {
 			$priority = PRIORITY_MEDIUM;
-		} elseif (($priority < PRIORITY_LOW) && ($retrial > 5)) {
+		} elseif (($priority < PRIORITY_LOW) && ($new_retrial > 6)) {
 			$priority = PRIORITY_LOW;
-		} elseif (($priority < PRIORITY_NEGLIGIBLE) && ($retrial > 7)) {
+		} elseif (($priority < PRIORITY_NEGLIGIBLE) && ($new_retrial > 8)) {
 			$priority = PRIORITY_NEGLIGIBLE;
 		}
 
-		Logger::log('Defer execution ' . $retrial . ' of id ' . $id . ' to ' . $next . ' - priority old/new: ' . $queue['priority'] . '/' . $priority, Logger::DEBUG);
+		Logger::info('Deferred task', ['id' => $id, 'retrial' => $new_retrial, 'next_execution' => $next, 'old_prio' => $queue['priority'], 'new_prio' => $priority]);
 
 		$stamp = (float)microtime(true);
-		$fields = ['retrial' => $retrial + 1, 'next_try' => $next, 'executed' => DBA::NULL_DATETIME, 'pid' => 0, 'priority' => $priority];
+		$fields = ['retrial' => $new_retrial, 'next_try' => $next, 'executed' => DBA::NULL_DATETIME, 'pid' => 0, 'priority' => $priority];
 		DBA::update('workerqueue', $fields, ['id' => $id]);
 		self::$db_duration += (microtime(true) - $stamp);
 		self::$db_duration_write += (microtime(true) - $stamp);