Generalize asynchronous events

CONTRIBUTING.md

@@ -58,6 +58,7 @@ PocketMine-MP has three primary branches of development.
 | Deprecating API classes, methods or constants                                                            |    ❌     |               ✔️                |      ✔️      |
 | Adding optional parameters to an API method                                                              |    ❌     |               ✔️                |      ✔️      |
 | Changing API behaviour                                                                                   |    ❌     | 🟡 Only if backwards-compatible |      ✔️      |
+| Changing an event from sync to async or vice versa                                                       |    ❌     |                ❌                |      ✔️      |
 | Removal of API                                                                                           |    ❌     |                ❌                |      ✔️      |
 | Backwards-incompatible API change (e.g. renaming a method)                                               |    ❌     |                ❌                |      ✔️      |
 | Backwards-incompatible internals change (e.g. changing things in `pocketmine\network\mcpe`)              |    ❌     |               ✔️                |      ✔️      |

src/Server.php

@@ -39,6 +39,7 @@
 use pocketmine\data\bedrock\BedrockDataFiles;
 use pocketmine\entity\EntityDataHelper;
 use pocketmine\entity\Location;
+use pocketmine\event\AsyncHandlerListManager;
 use pocketmine\event\HandlerListManager;
 use pocketmine\event\player\PlayerCreationEvent;
 use pocketmine\event\player\PlayerDataSaveEvent;
@@ -1557,6 +1558,7 @@ public function forceShutdown() : void{
 
 			$this->logger->debug("Removing event handlers");
 			HandlerListManager::global()->unregisterAll();
+			AsyncHandlerListManager::global()->unregisterAll();
 
 			if(isset($this->asyncPool)){
 				$this->logger->debug("Shutting down async task worker pool");

src/event/AsyncEvent.php

@@ -0,0 +1,136 @@
+<?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\event;
+
+use pocketmine\promise\Promise;
+use pocketmine\promise\PromiseResolver;
+use pocketmine\timings\Timings;
+use pocketmine\utils\Utils;
+use function count;
+use function spl_object_id;
+
+/**
+ * This class is used to permit asynchronous event handling.
+ *
+ * When an event is called asynchronously, the event handlers are called by priority level.
+ * When all the promises of a priority level have been resolved, the next priority level is called.
+ */
+abstract class AsyncEvent{
+	/** @var array<int, int> $handlersCallState */
+	private static array $handlersCallState = [];
+	private const MAX_CONCURRENT_CALLS = 1000; //max number of concurrent calls to a single handler
+
+	/**
+	 * @phpstan-return Promise<static>
+	 */
+	final public function call() : Promise{
+		$timings = Timings::getAsyncEventTimings($this);
+		$timings->startTiming();
+
+		try{
+			/** @phpstan-var PromiseResolver<static> $globalResolver */
+			$globalResolver = new PromiseResolver();
+
+			$handlers = AsyncHandlerListManager::global()->getHandlersFor(static::class);
+			if(count($handlers) > 0){
+				$this->processRemainingHandlers($handlers, fn() => $globalResolver->resolve($this), $globalResolver->reject(...));
+			}else{
+				$globalResolver->resolve($this);
+			}
+
+			return $globalResolver->getPromise();
+		}finally{
+			$timings->stopTiming();
+		}
+	}
+
+	/**
+	 * @param AsyncRegisteredListener[] $handlers
+	 * @phpstan-param array<int, AsyncRegisteredListener> $handlers
+	 * @phpstan-param \Closure() : void $resolve
+	 * @phpstan-param \Closure() : void $reject
+	 */
+	private function processRemainingHandlers(array $handlers, \Closure $resolve, \Closure $reject) : void{
+		$currentPriority = null;
+		$awaitPromises = [];
+		foreach($handlers as $k => $handler){
+			$priority = $handler->getPriority();
+			if(count($awaitPromises) > 0 && $currentPriority !== null && $currentPriority !== $priority){
+				//wait for concurrent promises from previous priority to complete
+				break;
+			}
+
+			$currentPriority = $priority;
+			if(!isset(self::$handlersCallState[$handlerId = spl_object_id($handler)])){
+				self::$handlersCallState[$handlerId] = 0;
+			}
+			if(self::$handlersCallState[$handlerId] >= self::MAX_CONCURRENT_CALLS){
+				throw new \RuntimeException("Concurrent call limit reached for handler " .
+					Utils::getNiceClosureName($handler->getHandler()) . "(" . Utils::getNiceClassName($this) . ")" .
+					" (max: " . self::MAX_CONCURRENT_CALLS . ")");
+			}
+			$removeCallback = static function() use ($handlerId) : void{
+				--self::$handlersCallState[$handlerId];
+			};
+			if($handler->canBeCalledConcurrently()){
+				unset($handlers[$k]);
+				++self::$handlersCallState[$handlerId];
+				$promise = $handler->callAsync($this);
+				if($promise !== null){
+					$promise->onCompletion($removeCallback, $removeCallback);
+					$awaitPromises[] = $promise;
+				}else{
+					$removeCallback();
+				}
+			}else{
+				if(count($awaitPromises) > 0){
+					//wait for concurrent promises to complete
+					break;
+				}
+
+				unset($handlers[$k]);
+				++self::$handlersCallState[$handlerId];
+				$promise = $handler->callAsync($this);
+				if($promise !== null){
+					$promise->onCompletion($removeCallback, $removeCallback);
+					$promise->onCompletion(
+						onSuccess: fn() => $this->processRemainingHandlers($handlers, $resolve, $reject),
+						onFailure: $reject
+					);
+					return;
+				}
+				$removeCallback();
+			}
+		}
+
+		if(count($awaitPromises) > 0){
+			Promise::all($awaitPromises)->onCompletion(
+				onSuccess: fn() => $this->processRemainingHandlers($handlers, $resolve, $reject),
+				onFailure: $reject
+			);
+		}else{
+			$resolve();
+		}
+	}
+}

src/event/AsyncHandlerListManager.php

@@ -0,0 +1,58 @@
+<?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\event;
+
+use function uasort;
+
+/**
+ * @phpstan-extends BaseHandlerListManager<AsyncEvent, AsyncRegisteredListener>
+ */
+final class AsyncHandlerListManager extends BaseHandlerListManager{
+	private static ?self $globalInstance = null;
+
+	public static function global() : self{
+		return self::$globalInstance ?? (self::$globalInstance = new self());
+	}
+
+	protected function getBaseEventClass() : string{
+		return AsyncEvent::class;
+	}
+
+	/**
+	 * @phpstan-param array<int, AsyncRegisteredListener> $listeners
+	 * @phpstan-return array<int, AsyncRegisteredListener>
+	 */
+	private static function sortSamePriorityHandlers(array $listeners) : array{
+		uasort($listeners, function(AsyncRegisteredListener $left, AsyncRegisteredListener $right) : int{
+			//Promise::all() can be used more efficiently if concurrent handlers are grouped together.
+			//It's not important whether they are grouped before or after exclusive handlers.
+			return $left->canBeCalledConcurrently() <=> $right->canBeCalledConcurrently();
+		});
+		return $listeners;
+	}
+
+	protected function createHandlerList(string $event, ?HandlerList $parentList, RegisteredListenerCache $handlerCache) : HandlerList{
+		return new HandlerList($event, $parentList, $handlerCache, self::sortSamePriorityHandlers(...));
+	}
+}

src/event/AsyncRegisteredListener.php

@@ -0,0 +1,60 @@
+<?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\event;
+
+use pocketmine\plugin\Plugin;
+use pocketmine\promise\Promise;
+use pocketmine\timings\TimingsHandler;
+
+class AsyncRegisteredListener extends BaseRegisteredListener{
+	public function __construct(
+		\Closure $handler,
+		int $priority,
+		Plugin $plugin,
+		bool $handleCancelled,
+		private bool $exclusiveCall,
+		TimingsHandler $timings
+	){
+		parent::__construct($handler, $priority, $plugin, $handleCancelled, $timings);
+	}
+
+	/**
+	 * @phpstan-return Promise<null>|null
+	 */
+	public function callAsync(AsyncEvent $event) : ?Promise{
+		if($event instanceof Cancellable && $event->isCancelled() && !$this->isHandlingCancelled()){
+			return null;
+		}
+		$this->timings->startTiming();
+		try{
+			return ($this->handler)($event);
+		}finally{
+			$this->timings->stopTiming();
+		}
+	}
+
+	public function canBeCalledConcurrently() : bool{
+		return !$this->exclusiveCall;
+	}
+}