Improved threading and resource usage, new defaults

src/pocketmine/scheduler/AsyncPool.php

@@ -0,0 +1,109 @@
+<?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
+
+ *
+ *
+*/
+
+namespace pocketmine\scheduler;
+
+use pocketmine\Server;
+
+class AsyncPool{
+
+	/** @var Server */
+	private $server;
+
+	protected $size;
+
+	/** @var AsyncTask[] */
+	private $tasks = [];
+	/** @var int[] */
+	private $taskWorkers = [];
+
+	/** @var AsyncWorker[] */
+	private $workers = [];
+	/** @var int[] */
+	private $workerUsage = [];
+
+	public function __construct(Server $server, $size){
+		$this->server = $server;
+		$this->size = (int) $size;
+
+		for($i = 0; $i < $this->size; ++$i){
+			$this->workerUsage[$i] = 0;
+			$this->workers[$i] = new AsyncWorker();
+			$this->workers[$i]->start();
+		}
+	}
+
+	public function submitTask(AsyncTask $task){
+		if(isset($this->tasks[$task->getTaskId()]) or $task->isGarbage()){
+			return;
+		}
+
+		$this->tasks[$task->getTaskId()] = $task;
+
+		$selectedWorker = mt_rand(0, $this->size - 1);
+		$selectedTasks = $this->workerUsage[$selectedWorker];
+		for($i = 0; $i < $this->size; ++$i){
+			if($this->workerUsage[$i] < $selectedTasks){
+				$selectedWorker = $i;
+				$selectedTasks = $this->workerUsage[$i];
+			}
+		}
+
+		$this->workers[$selectedWorker]->stack($task);
+		$this->workerUsage[$selectedWorker]++;
+		$this->taskWorkers[$task->getTaskId()] = $selectedWorker;
+	}
+
+	private function removeTask(AsyncTask $task){
+		if(isset($this->taskWorkers[$task->getTaskId()])){
+			$this->workers[$this->taskWorkers[$task->getTaskId()]]->unstack($task);
+		}
+
+		unset($this->tasks[$task->getTaskId()]);
+		unset($this->taskWorkers[$task->getTaskId()]);
+	}
+
+	public function removeTasks(){
+		foreach($this->tasks as $task){
+			$this->removeTask($task);
+		}
+
+		for($i = 0; $i < $this->size; ++$i){
+			$this->workerUsage[$i] = 0;
+		}
+
+		$this->taskWorkers = [];
+		$this->tasks = [];
+	}
+
+	public function collectTasks(){
+		foreach($this->tasks as $task){
+			if($task->isGarbage()){
+
+				$task->onCompletion($this->server);
+
+				$this->workerUsage[$this->taskWorkers[$task->getTaskId()]]--;
+				unset($this->tasks[$task->getTaskId()]);
+				unset($this->taskWorkers[$task->getTaskId()]);
+			}
+		}
+	}
+}

src/pocketmine/scheduler/AsyncTask.php

@@ -30,23 +30,25 @@ use pocketmine\Server;
  */
 abstract class AsyncTask extends \Collectable{
 
-	private $finished = null;
 	private $result = null;
+	/** @var int */
 	private $taskId = null;
 
 	public function run(){
-		$this->finished = false;
 		$this->result = null;
 
 		$this->onRun();
-		$this->finished = true;
+
+		$this->setGarbage();
 	}
 
 	/**
+	 * @deprecated
+	 *
 	 * @return bool
 	 */
 	public function isFinished(){
-		return $this->finished === true;
+		return $this->isGarbage();
 	}
 
 	/**

src/pocketmine/scheduler/AsyncWorker.php

@@ -34,6 +34,11 @@ class AsyncWorker extends Worker{
 		}
 		$autoloader = new \BaseClassLoader();
 		$autoloader->addPath(\pocketmine\PATH . "src");
+		$autoloader->addPath(\pocketmine\PATH . "src" . DIRECTORY_SEPARATOR . "spl");
 		$autoloader->register(true);
 	}
+
+	public function start($options = PTHREADS_INHERIT_NONE){
+		parent::start(PTHREADS_INHERIT_CONSTANTS | PTHREADS_INHERIT_FUNCTIONS);
+	}
 }

src/pocketmine/scheduler/SendUsageTask.php

@@ -37,9 +37,4 @@ class SendUsageTask extends AsyncTask{
 	public function onRun(){
 		Utils::postURL($this->endpoint, unserialize($this->data));
 	}
-
-	public function onCompletion(Server $server){
-
-	}
-
 }

src/pocketmine/scheduler/ServerScheduler.php

@@ -42,14 +42,9 @@ class ServerScheduler{
 	 */
 	protected $tasks = [];
 
-	/** @var \Pool */
+	/** @var AsyncPool */
 	protected $asyncPool;
 
-	/** @var AsyncTask[] */
-	protected $asyncTaskStorage = [];
-
-	protected $asyncTasks = 0;
-
 	/** @var int */
 	private $ids = 1;
 
@@ -58,7 +53,7 @@ class ServerScheduler{
 
 	public function __construct(){
 		$this->queue = new ReversePriorityQueue();
-		$this->asyncPool = new \Pool(self::$WORKERS, AsyncWorker::class);
+		$this->asyncPool = new AsyncPool(Server::getInstance(), self::$WORKERS);
 	}
 
 	/**
@@ -80,9 +75,7 @@ class ServerScheduler{
 	public function scheduleAsyncTask(AsyncTask $task){
 		$id = $this->nextId();
 		$task->setTaskId($id);
-		$this->asyncPool->submit($task);
-		$this->asyncTaskStorage[$task->getTaskId()] = $task;
-		++$this->asyncTasks;
+		$this->asyncPool->submitTask($task);
 	}
 
 	/**
@@ -144,11 +137,9 @@ class ServerScheduler{
 			$task->cancel();
 		}
 		$this->tasks = [];
-		$this->asyncTaskStorage = [];
-		//$this->asyncPool->shutdown();
-		$this->asyncTasks = 0;
+		$this->asyncPool->removeTasks();
 		$this->queue = new ReversePriorityQueue();
-		$this->asyncPool = new \Pool(self::$WORKERS, AsyncWorker::class);
+		$this->ids = 1;
 	}
 
 	/**
@@ -237,7 +228,8 @@ class ServerScheduler{
 					$task->run($this->currentTick);
 				}catch(\Exception $e){
 					Server::getInstance()->getLogger()->critical("Could not execute task " . $task->getTaskName() . ": " . $e->getMessage());
-					if(($logger = Server::getInstance()->getLogger()) instanceof MainLogger){
+					$logger = Server::getInstance()->getLogger();
+					if($logger instanceof MainLogger){
 						$logger->logException($e);
 					}
 				}
@@ -252,26 +244,7 @@ class ServerScheduler{
 			}
 		}
 
-		if($this->asyncTasks > 0){ //Garbage collector
-			$this->asyncPool->collect([$this, "collectAsyncTask"]);
-
-			if($this->asyncTasks > 0){
-				foreach($this->asyncTaskStorage as $asyncTask){
-					$this->collectAsyncTask($asyncTask);
-				}
-			}
-		}
-	}
-
-	public function collectAsyncTask(AsyncTask $task){
-		if($task->isFinished() and !$task->isGarbage()){
-			--$this->asyncTasks;
-			$task->onCompletion(Server::getInstance());
-			$task->setGarbage();
-			unset($this->asyncTaskStorage[$task->getTaskId()]);
-		}
-
-		return $task->isGarbage();
+		$this->asyncPool->collectTasks();
 	}
 
 	private function isReady($currentTicks){