PocketMine-MP/src/scheduler/AsyncPool.php

<?php

/*
 *
 *  ____            _        _   __  __ _                  __  __ ____
 * |  _ \ ___   ___| | _____| |_|  \/  (_)_ __   ___      |  \/  |  _ \
 * | |_) / _ \ / __| |/ / _ \ __| |\/| | | '_ \ / _ \_____| |\/| | |_) |
 * |  __/ (_) | (__|   <  __/ |_| |  | | | | | |  __/_____| |  | |  __/
 * |_|   \___/ \___|_|\_\___|\__|_|  |_|_|_| |_|\___|     |_|  |_|_|
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * @author PocketMine Team
 * @link http://www.pocketmine.net/
 *
 *
 */

declare(strict_types=1);

namespace pocketmine\scheduler;

use pmmp\thread\Thread as NativeThread;
use pocketmine\snooze\SleeperHandler;
use pocketmine\thread\log\ThreadSafeLogger;
use pocketmine\thread\ThreadCrashException;
use pocketmine\thread\ThreadSafeClassLoader;
use pocketmine\timings\Timings;
use pocketmine\utils\AssumptionFailedError;
use pocketmine\utils\Utils;
use function array_keys;
use function array_map;
use function assert;
use function count;
use function get_class;
use function spl_object_id;
use function time;
use const PHP_INT_MAX;

/**
 * Manages general-purpose worker threads used for processing asynchronous tasks, and the tasks submitted to those
 * workers.
 */
class AsyncPool{
	private const WORKER_START_OPTIONS = NativeThread::INHERIT_INI | NativeThread::INHERIT_COMMENTS;

	/**
	 * @var AsyncPoolWorkerEntry[]
	 * @phpstan-var array<int, AsyncPoolWorkerEntry>
	 */
	private array $workers = [];

	/**
	 * @var \Closure[]
	 * @phpstan-var (\Closure(int $workerId) : void)[]
	 */
	private array $workerStartHooks = [];

	public function __construct(
		protected int $size,
		private int $workerMemoryLimit,
		private ThreadSafeClassLoader $classLoader,
		private ThreadSafeLogger $logger,
		private SleeperHandler $eventLoop
	){}

	/**
	 * Returns the maximum size of the pool. Note that there may be less active workers than this number.
	 */
	public function getSize() : int{
		return $this->size;
	}

	/**
	 * Increases the maximum size of the pool to the specified amount. This does not immediately start new workers.
	 */
	public function increaseSize(int $newSize) : void{
		if($newSize > $this->size){
			$this->size = $newSize;
		}
	}

	/**
	 * Registers a Closure callback to be fired whenever a new worker is started by the pool.
	 * The signature should be `function(int $worker) : void`
	 *
	 * This function will call the hook for every already-running worker.
	 *
	 * @phpstan-param \Closure(int $workerId) : void $hook
	 */
	public function addWorkerStartHook(\Closure $hook) : void{
		Utils::validateCallableSignature(function(int $worker) : void{}, $hook);
		$this->workerStartHooks[spl_object_id($hook)] = $hook;
		foreach($this->workers as $i => $worker){
			$hook($i);
		}
	}

	/**
	 * Removes a previously-registered callback listening for workers being started.
	 *
	 * @phpstan-param \Closure(int $workerId) : void $hook
	 */
	public function removeWorkerStartHook(\Closure $hook) : void{
		unset($this->workerStartHooks[spl_object_id($hook)]);
	}

	/**
	 * Returns an array of IDs of currently running workers.
	 *
	 * @return int[]
	 */
	public function getRunningWorkers() : array{
		return array_keys($this->workers);
	}

	/**
	 * Fetches the worker with the specified ID, starting it if it does not exist, and firing any registered worker
	 * start hooks.
	 */
	private function getWorker(int $workerId) : AsyncPoolWorkerEntry{
		if(!isset($this->workers[$workerId])){
			$sleeperEntry = $this->eventLoop->addNotifier(function() use ($workerId) : void{
				$this->collectTasksFromWorker($workerId);
			});
			$this->workers[$workerId] = new AsyncPoolWorkerEntry(new AsyncWorker($this->logger, $workerId, $this->workerMemoryLimit, $sleeperEntry), $sleeperEntry->getNotifierId());
			$this->workers[$workerId]->worker->setClassLoaders([$this->classLoader]);
			$this->workers[$workerId]->worker->start(self::WORKER_START_OPTIONS);

			foreach($this->workerStartHooks as $hook){
				$hook($workerId);
			}
		}else{
			$this->checkCrashedWorker($workerId, null);
		}

		return $this->workers[$workerId];
	}

	/**
	 * Submits an AsyncTask to an arbitrary worker.
	 */
	public function submitTaskToWorker(AsyncTask $task, int $worker) : void{
		if($worker < 0 || $worker >= $this->size){
			throw new \InvalidArgumentException("Invalid worker $worker");
		}
		if($task->isSubmitted()){
			throw new \InvalidArgumentException("Cannot submit the same AsyncTask instance more than once");
		}

		$task->setSubmitted();

		$this->getWorker($worker)->submit($task);
	}

	/**
	 * Selects a worker ID to run a task.
	 *
	 * - if an idle worker is found, it will be selected
	 * - else, if the worker pool is not full, a new worker will be selected
	 * - else, the worker with the smallest backlog is chosen.
	 */
	public function selectWorker() : int{
		$worker = null;
		$minUsage = PHP_INT_MAX;
		foreach($this->workers as $i => $entry){
			if(($usage = $entry->tasks->count()) < $minUsage){
				$worker = $i;
				$minUsage = $usage;
				if($usage === 0){
					break;
				}
			}
		}
		if($worker === null || ($minUsage > 0 && count($this->workers) < $this->size)){
			//select a worker to start on the fly
			for($i = 0; $i < $this->size; ++$i){
				if(!isset($this->workers[$i])){
					$worker = $i;
					break;
				}
			}
		}

		assert($worker !== null);
		return $worker;
	}

	/**
	 * Submits an AsyncTask to the worker with the least load. If all workers are busy and the pool is not full, a new
	 * worker may be started.
	 */
	public function submitTask(AsyncTask $task) : int{
		if($task->isSubmitted()){
			throw new \InvalidArgumentException("Cannot submit the same AsyncTask instance more than once");
		}

		$worker = $this->selectWorker();
		$this->submitTaskToWorker($task, $worker);
		return $worker;
	}

	private function checkCrashedWorker(int $workerId, ?AsyncTask $crashedTask) : void{
		$entry = $this->workers[$workerId];
		if($entry->worker->isTerminated()){
			if($crashedTask === null){
				foreach($entry->tasks as $task){
					if($task->isTerminated()){
						$crashedTask = $task;
						break;
					}elseif(!$task->isFinished()){
						break;
					}
				}
			}
			$info = $entry->worker->getCrashInfo();
			if($info !== null){
				if($crashedTask !== null){
					$message = "Worker $workerId crashed while running task " . get_class($crashedTask) . "#" . spl_object_id($crashedTask);
				}else{
					$message = "Worker $workerId crashed while doing unknown work";
				}
				throw new ThreadCrashException($message, $info);
			}else{
				throw new \RuntimeException("Worker $workerId crashed for unknown reason");
			}
		}
	}

	/**
	 * Collects finished and/or crashed tasks from the workers, firing their on-completion hooks where appropriate.
	 *
	 * @throws \ReflectionException
	 * @return bool whether there are tasks left to be collected
	 */
	public function collectTasks() : bool{
		foreach($this->workers as $workerId => $entry){
			$this->collectTasksFromWorker($workerId);
		}

		//we check this in a second loop, because task collection could have caused new tasks to be added to the queues
		foreach($this->workers as $entry){
			if(!$entry->tasks->isEmpty()){
				return true;
			}
		}
		return false;
	}

	public function collectTasksFromWorker(int $worker) : bool{
		if(!isset($this->workers[$worker])){
			throw new \InvalidArgumentException("No such worker $worker");
		}
		$queue = $this->workers[$worker]->tasks;
		$more = false;
		while(!$queue->isEmpty()){
			/** @var AsyncTask $task */
			$task = $queue->bottom();
			if($task->isFinished()){ //make sure the task actually executed before trying to collect
				$queue->dequeue();

				if($task->isTerminated()){
					$this->checkCrashedWorker($worker, $task);
					throw new AssumptionFailedError("checkCrashedWorker() should have thrown an exception, making this unreachable");
				}else{
					/*
					 * It's possible for a task to submit a progress update and then finish before the progress
					 * update is detected by the parent thread, so here we consume any missed updates.
					 *
					 * When this happens, it's possible for a progress update to arrive between the previous
					 * checkProgressUpdates() call and the next isGarbage() call, causing progress updates to be
					 * lost. Thus, it's necessary to do one last check here to make sure all progress updates have
					 * been consumed before completing.
					 */
					$this->checkTaskProgressUpdates($task);
					Timings::getAsyncTaskCompletionTimings($task)->time(function() use ($task) : void{
						$task->onCompletion();
					});
				}
			}else{
				$this->checkTaskProgressUpdates($task);
				$more = true;
				break; //current task is still running, skip to next worker
			}
		}
		$this->workers[$worker]->worker->collect();
		return $more;
	}

	/**
	 * Returns an array of worker ID => task queue size
	 *
	 * @return int[]
	 * @phpstan-return array<int, int>
	 */
	public function getTaskQueueSizes() : array{
		return array_map(function(AsyncPoolWorkerEntry $entry) : int{ return $entry->tasks->count(); }, $this->workers);
	}

	public function shutdownUnusedWorkers() : int{
		$ret = 0;
		$time = time();
		foreach($this->workers as $i => $entry){
			if($entry->lastUsed + 300 < $time && $entry->tasks->isEmpty()){
				$entry->worker->quit();
				$this->eventLoop->removeNotifier($entry->sleeperNotifierId);
				unset($this->workers[$i]);
				$ret++;
			}
		}

		return $ret;
	}

	/**
	 * Cancels all pending tasks and shuts down all the workers in the pool.
	 */
	public function shutdown() : void{
		while($this->collectTasks()){
			//NOOP
		}

		foreach($this->workers as $worker){
			$worker->worker->quit();
			$this->eventLoop->removeNotifier($worker->sleeperNotifierId);
		}
		$this->workers = [];
	}

	private function checkTaskProgressUpdates(AsyncTask $task) : void{
		Timings::getAsyncTaskProgressUpdateTimings($task)->time(function() use ($task) : void{
			$task->checkProgressUpdates();
		});
	}
}