X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FCore%2FWorker.php;h=8dd1ece8e5e41ada1cc0bfb4a9ed85de59ce219a;hb=06284e60073f374c1bd411e0bba6474a13c14f10;hp=21140b890e24a100fa1e4d46ce128b42f5eb3585;hpb=adb0b460ef8d2f133369b4b0732688ce4ab282bc;p=friendica.git

diff --git a/src/Core/Worker.php b/src/Core/Worker.php
index 21140b890e..8dd1ece8e5 100644
--- a/src/Core/Worker.php
+++ b/src/Core/Worker.php
@@ -1,6 +1,6 @@
 <?php
 /**
- * @copyright Copyright (C) 2020, Friendica
+ * @copyright Copyright (C) 2010-2021, the Friendica project
  *
  * @license GNU AGPL version 3 or any later version
  *
@@ -23,6 +23,7 @@ namespace Friendica\Core;
 
 use Friendica\App\Mode;
 use Friendica\Core;
+use Friendica\Core\Worker\Entity\Process;
 use Friendica\Database\DBA;
 use Friendica\DI;
 use Friendica\Util\DateTimeFormat;
@@ -51,29 +52,29 @@ class Worker
 	private static $last_update;
 	private static $state;
 	private static $daemon_mode = null;
+	/** @var Process */
+	private static $process;
 
 	/**
 	 * Processes the tasks that are in the workerqueue table
 	 *
 	 * @param boolean $run_cron Should the cron processes be executed?
+	 * @param Process $process  The current running process
 	 * @return void
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 */
-	public static function processQueue($run_cron = true)
+	public static function processQueue($run_cron, Process $process)
 	{
-		// Ensure that all "strtotime" operations do run timezone independent
-		date_default_timezone_set('UTC');
-
 		self::$up_start = microtime(true);
 
 		// At first check the maximum load. We shouldn't continue with a high load
-		if (DI::process()->isMaxLoadReached()) {
+		if (DI::system()->isMaxLoadReached()) {
 			Logger::notice('Pre check: maximum load reached, quitting.');
 			return;
 		}
 
 		// We now start the process. This is done after the load check since this could increase the load.
-		DI::process()->start();
+		self::$process = $process;
 
 		// Kill stale processes every 5 minutes
 		$last_cleanup = DI::config()->get('system', 'worker_last_cleaned', 0);
@@ -137,7 +138,7 @@ class Worker
 					}
 
 					// Check free memory
-					if (DI::process()->isMinMemoryReached()) {
+					if (DI::system()->isMinMemoryReached()) {
 						Logger::warning('Memory limit reached, quitting.');
 						DI::lock()->release(self::LOCK_WORKER);
 						return;
@@ -150,7 +151,7 @@ class Worker
 			// 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();
+				self::unclaimProcess($process);
 				if (self::isDaemonMode()) {
 					self::IPCSetJobState(true);
 				} else {
@@ -183,7 +184,7 @@ class Worker
 		}
 
 		// Do we have too few memory?
-		if (DI::process()->isMinMemoryReached()) {
+		if (DI::system()->isMinMemoryReached()) {
 			Logger::warning('Memory limit reached, quitting.');
 			return false;
 		}
@@ -195,7 +196,7 @@ class Worker
 		}
 
 		// Possibly there are too much database processes that block the system
-		if (DI::process()->isMaxProcessesReached()) {
+		if (DI::system()->isMaxProcessesReached()) {
 			Logger::warning('Maximum processes reached, quitting.');
 			return false;
 		}
@@ -280,6 +281,44 @@ class Worker
 		return DBA::exists('workerqueue', $condition);
 	}
 
+	/**
+	 * Checks if the given file is valid to be included
+	 *
+	 * @param mixed $file 
+	 * @return bool 
+	 */
+	private static function validateInclude(&$file)
+	{
+		$orig_file = $file;
+	
+		$file = realpath($file);
+	
+		if (strpos($file, getcwd()) !== 0) {
+			return false;
+		}
+	
+		$file = str_replace(getcwd() . "/", "", $file, $count);
+		if ($count != 1) {
+			return false;
+		}
+	
+		if ($orig_file !== $file) {
+			return false;
+		}
+	
+		$valid = false;
+		if (strpos($file, "include/") === 0) {
+			$valid = true;
+		}
+	
+		if (strpos($file, "addon/") === 0) {
+			$valid = true;
+		}
+	
+		// Simply return flag
+		return $valid;
+	}
+	
 	/**
 	 * Execute a worker entry
 	 *
@@ -299,7 +338,7 @@ class Worker
 		}
 
 		// Constantly check the number of parallel database processes
-		if (DI::process()->isMaxProcessesReached()) {
+		if (DI::system()->isMaxProcessesReached()) {
 			Logger::warning("Max processes reached for process", ['pid' => $mypid]);
 			return false;
 		}
@@ -363,7 +402,7 @@ class Worker
 			$include = "include/".$include.".php";
 		}
 
-		if (!validate_include($include)) {
+		if (!self::validateInclude($include)) {
 			Logger::warning("Include file is not valid", ['file' => $argv[0]]);
 			$stamp = (float)microtime(true);
 			DBA::delete('workerqueue', ['id' => $queue["id"]]);
@@ -446,7 +485,7 @@ class Worker
 			$queue['priority'] = PRIORITY_MEDIUM;
 		}
 
-		$a->queue = $queue;
+		$a->setQueue($queue);
 
 		$up_duration = microtime(true) - self::$up_start;
 
@@ -462,7 +501,7 @@ class Worker
 
 		Logger::disableWorker();
 
-		unset($a->queue);
+		$a->setQueue([]);
 
 		$duration = (microtime(true) - $stamp);
 
@@ -715,13 +754,10 @@ class Worker
 					}
 
 					$stamp = (float)microtime(true);
-					$jobs = DBA::p("SELECT COUNT(*) AS `jobs` FROM `workerqueue` WHERE `done` AND `executed` > UTC_TIMESTAMP() - INTERVAL ? MINUTE", $interval);
+					$jobs = DBA::count('workerqueue', ["`done` AND `executed` > UTC_TIMESTAMP() - INTERVAL ? MINUTE", $interval]);
 					self::$db_duration += (microtime(true) - $stamp);
 					self::$db_duration_stat += (microtime(true) - $stamp);
-					if ($job = DBA::fetch($jobs)) {
-						$jobs_per_minute[$interval] = number_format($job['jobs'] / $interval, 0);
-					}
-					DBA::close($jobs);
+					$jobs_per_minute[$interval] = number_format($jobs / $interval, 0);
 				}
 				$processlist = ' - jpm: '.implode('/', $jobs_per_minute);
 			}
@@ -742,15 +778,12 @@ class Worker
 				self::$db_duration_stat += (microtime(true) - $stamp);
 				while ($entry = DBA::fetch($jobs)) {
 					$stamp = (float)microtime(true);
-					$processes = DBA::p("SELECT COUNT(*) AS `running` FROM `workerqueue-view` WHERE `priority` = ?", $entry["priority"]);
+					$running = DBA::count('workerqueue-view', ['priority' => $entry["priority"]]);
 					self::$db_duration += (microtime(true) - $stamp);
 					self::$db_duration_stat += (microtime(true) - $stamp);
-					if ($process = DBA::fetch($processes)) {
-						$idle_workers -= $process["running"];
-						$waiting_processes += $entry["entries"];
-						$listitem[$entry["priority"]] = $entry["priority"].":".$process["running"]."/".$entry["entries"];
-					}
-					DBA::close($processes);
+					$idle_workers -= $running;
+					$waiting_processes += $entry["entries"];
+					$listitem[$entry["priority"]] = $entry["priority"] . ":" . $running . "/" . $entry["entries"];
 				}
 				DBA::close($jobs);
 			} else {
@@ -813,7 +846,7 @@ class Worker
 	private static function activeWorkers()
 	{
 		$stamp = (float)microtime(true);
-		$count = DBA::count('process', ['command' => 'Worker.php']);
+		$count = DI::process()->countCommand('Worker.php');
 		self::$db_duration += (microtime(true) - $stamp);
 		self::$db_duration_count += (microtime(true) - $stamp);
 		return $count;
@@ -831,7 +864,7 @@ class Worker
 		$stamp = (float)microtime(true);
 
 		$queues = DBA::p("SELECT `process`.`pid`, COUNT(`workerqueue`.`pid`) AS `entries` FROM `process`
-			LEFT JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `workerqueue`.`done` 
+			LEFT JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid` AND NOT `workerqueue`.`done`
 			GROUP BY `process`.`pid`");
 		while ($queue = DBA::fetch($queues)) {
 			$ids[$queue['pid']] = $queue['entries'];
@@ -1076,15 +1109,15 @@ class Worker
 	/**
 	 * Removes a workerqueue entry from the current process
 	 *
+	 * @param Process $process the process behind the workerqueue
+	 *
 	 * @return void
 	 * @throws \Exception
 	 */
-	public static function unclaimProcess()
+	public static function unclaimProcess(Process $process)
 	{
-		$mypid = getmypid();
-
 		$stamp = (float)microtime(true);
-		DBA::update('workerqueue', ['executed' => DBA::NULL_DATETIME, 'pid' => 0], ['pid' => $mypid, 'done' => false]);
+		DBA::update('workerqueue', ['executed' => DBA::NULL_DATETIME, 'pid' => 0], ['pid' => $process->pid, 'done' => false]);
 		self::$db_duration += (microtime(true) - $stamp);
 		self::$db_duration_write += (microtime(true) - $stamp);
 	}
@@ -1117,7 +1150,7 @@ class Worker
 	 */
 	private static function forkProcess(bool $do_cron)
 	{
-		if (DI::process()->isMinMemoryReached()) {
+		if (DI::system()->isMinMemoryReached()) {
 			Logger::warning('Memory limit reached - quitting');
 			return;
 		}
@@ -1147,26 +1180,25 @@ class Worker
 		}
 
 		// We now are in the new worker
-		$pid = getmypid();
-
 		DBA::connect();
-		/// @todo Reinitialize the logger to set a new process_id and uid
-		DI::process()->setPid($pid);
+
+		DI::flushLogger();
+		$process = DI::process()->create(getmypid(), basename(__FILE__));
 
 		$cycles = 0;
-		while (!self::IPCJobsExists($pid) && (++$cycles < 100)) {
+		while (!self::IPCJobsExists($process->pid) && (++$cycles < 100)) {
 			usleep(10000);
 		}
 
-		Logger::info('Worker spawned', ['pid' => $pid, 'wait_cycles' => $cycles]);
+		Logger::info('Worker spawned', ['pid' => $process->pid, 'wait_cycles' => $cycles]);
 
-		self::processQueue($do_cron);
+		self::processQueue($do_cron, $process);
 
-		self::unclaimProcess();
+		self::unclaimProcess($process);
 
-		self::IPCSetJobState(false, $pid);
-		DI::process()->end();
-		Logger::info('Worker ended', ['pid' => $pid]);
+		self::IPCSetJobState(false, $process->pid);
+		DI::process()->delete($process);
+		Logger::info('Worker ended', ['pid' => $process->pid]);
 		exit();
 	}
 
@@ -1182,9 +1214,7 @@ class Worker
 		if (self::isDaemonMode() && DI::config()->get('system', 'worker_fork')) {
 			self::forkProcess($do_cron);
 		} else {
-			$process = new Core\Process(DI::logger(), DI::mode(), DI::config(),
-				DI::modelProcess(), DI::app()->getBasePath(), getmypid());
-			$process->run('bin/worker.php', ['no_cron' => !$do_cron]);
+			DI::system()->run('bin/worker.php', ['no_cron' => !$do_cron]);
 		}
 		if (self::isDaemonMode()) {
 			self::IPCSetJobState(false);
@@ -1200,7 +1230,7 @@ class Worker
 	 * or: Worker::add(PRIORITY_HIGH, "Notifier", Delivery::DELETION, $drop_id);
 	 * or: Worker::add(array('priority' => PRIORITY_HIGH, 'dont_fork' => true), "Delivery", $post_id);
 	 *
-	 * @return boolean "false" if worker queue entry already existed or there had been an error
+	 * @return int "0" if worker queue entry already existed or there had been an error, otherwise the ID of the worker task
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @note $cmd and string args are surrounded with ""
 	 *
@@ -1208,19 +1238,17 @@ class Worker
 	 *    array $arr
 	 *
 	 */
-	public static function add($cmd)
+	public static function add(...$args)
 	{
-		$args = func_get_args();
-
 		if (!count($args)) {
-			return false;
+			return 0;
 		}
 
 		$arr = ['args' => $args, 'run_cmd' => true];
 
 		Hook::callAll("proc_run", $arr);
 		if (!$arr['run_cmd'] || !count($args)) {
-			return true;
+			return 1;
 		}
 
 		$priority = PRIORITY_MEDIUM;
@@ -1250,12 +1278,14 @@ class Worker
 			if (isset($run_parameter['force_priority'])) {
 				$force_priority = $run_parameter['force_priority'];
 			}
+		} else {
+			throw new \InvalidArgumentException('Priority number or task parameter array expected as first argument');
 		}
 
 		$command = array_shift($args);
 		$parameters = json_encode($args);
 		$found = DBA::exists('workerqueue', ['command' => $command, 'parameter' => $parameters, 'done' => false]);
-		$added = false;
+		$added = 0;
 
 		if (!in_array($priority, PRIORITIES)) {
 			Logger::warning('Invalid priority', ['priority' => $priority, 'command' => $command, 'callstack' => System::callstack(20)]);
@@ -1264,15 +1294,15 @@ class Worker
 
 		// Quit if there was a database error - a precaution for the update process to 3.5.3
 		if (DBA::errorNo() != 0) {
-			return false;
+			return 0;
 		}
 
 		if (!$found) {
-			$added = DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
-				'priority' => $priority, 'next_try' => $delayed]);
-			if (!$added) {
-				return false;
+			if (!DBA::insert('workerqueue', ['command' => $command, 'parameter' => $parameters, 'created' => $created,
+				'priority' => $priority, 'next_try' => $delayed])) {
+				return 0;
 			}
+			$added = DBA::lastInsertId();
 		} elseif ($force_priority) {
 			DBA::update('workerqueue', ['priority' => $priority], ['command' => $command, 'parameter' => $parameters, 'done' => false, 'pid' => 0]);
 		}
@@ -1351,12 +1381,12 @@ class Worker
 	 */
 	public static function defer()
 	{
-		if (empty(DI::app()->queue)) {
+		$queue = DI::app()->getQueue();
+
+		if (empty($queue)) {
 			return false;
 		}
 
-		$queue = DI::app()->queue;
-
 		$retrial = $queue['retrial'];
 		$id = $queue['id'];
 		$priority = $queue['priority'];
@@ -1542,8 +1572,7 @@ class Worker
 		Logger::notice('Starting new daemon process');
 		$command = 'bin/daemon.php';
 		$a = DI::app();
-		$process = new Core\Process(DI::logger(), DI::mode(), DI::config(), DI::modelProcess(), $a->getBasePath(), getmypid());
-		$process->run($command, ['start']);
+		DI::system()->run($command, ['start']);
 		Logger::notice('New daemon process started');
 	}
 
@@ -1587,7 +1616,7 @@ class Worker
 		} else {
 			Logger::info('We are outside the maintenance window', ['current' => date('H:i:s', $current), 'start' => date('H:i:s', $start), 'end' => date('H:i:s', $end)]);
 		}
-		
+
 		return $execute;
 	}
 }