Merge pull request #2071 from PocketMine/pthreads-fix

Removed pthreads workarounds, new pthreads version

.travis.yml

@@ -9,7 +9,7 @@ branches:
 before_script:
  - mkdir plugins
  - wget -O plugins/DevTools.phar https://github.com/PocketMine/DevTools/releases/download/v1.9.0/DevTools_v1.9.0.phar
- - pecl install channel://pecl.php.net/pthreads-2.0.7
+ - pecl install channel://pecl.php.net/pthreads-2.0.8
  - pecl install channel://pecl.php.net/weakref-0.2.4
  - echo | pecl install channel://pecl.php.net/yaml-1.1.1
 

src/pocketmine/PocketMine.php

@@ -279,8 +279,8 @@ namespace pocketmine {
 	if(substr_count($pthreads_version, ".") < 2){
 		$pthreads_version = "0.$pthreads_version";
 	}
-	if(version_compare($pthreads_version, "2.0.4") < 0){
+	if(version_compare($pthreads_version, "2.0.8") < 0){
-		$logger->critical("pthreads >= 2.0.4 is required, while you have $pthreads_version.");
+		$logger->critical("pthreads >= 2.0.8 is required, while you have $pthreads_version.");
 		++$errors;
 	}
 

src/pocketmine/Server.php

@@ -129,7 +129,6 @@ class Server{
 	private $nextTick = 0;
 	private $tickAverage = [20,20,20,20,20];
 	private $useAverage = [20,20,20,20,20];
-	private $inTick = false;
 
 	/** @var \AttachableThreadedLogger */
 	private $logger;
@@ -1061,12 +1060,8 @@ class Server{
 		if($generator !== null and class_exists($generator) and is_subclass_of($generator, "pocketmine\\level\\generator\\Generator")){
 			$generator = new $generator($options);
 		}else{
-			if(strtoupper($this->getLevelType()) == "FLAT"){
-				$generator = Generator::getGenerator("flat");
 			$options["preset"] = $this->getConfigString("generator-settings", "");
-			}else{
+			$generator = Generator::getGenerator($this->getLevelType());
-				$generator = Generator::getGenerator("normal");
-			}
 		}
 
 		if(($provider = LevelProviderManager::getProviderByName($providerName = $this->getProperty("level-settings.default-format", "mcregion"))) === null){
@@ -1844,7 +1839,7 @@ class Server{
 
 		$this->getScheduler()->scheduleRepeatingTask(new CallbackTask([$this, "checkTicks"]), 20 * 5);
 
-		$this->logger->info("Default game type: " . self::getGamemodeString($this->getGamemode())); //TODO: string name
+		$this->logger->info("Default game type: " . self::getGamemodeString($this->getGamemode()));
 
 		$this->logger->info("Done (" . round(microtime(true) - \pocketmine\START_TIME, 3) . 's)! For help, type "help" or "?"');
 
@@ -1941,17 +1936,8 @@ class Server{
 	private function tickProcessor(){
 		$lastLoop = 0;
 		while($this->isRunning){
-			++$lastLoop;
+			$this->tick();
-
+			usleep((int) max(1, ($this->nextTick - microtime(true)) * 1000000));
-			if(($ticks = $this->tick()) !== true){
-				if($lastLoop > 2 and $lastLoop < 16){
-					usleep(1000);
-				}elseif($lastLoop < 128){
-					usleep(2000);
-				}else{
-					usleep(10000);
-				}
-			}
 		}
 	}
 
@@ -2032,8 +2018,7 @@ class Server{
 	/**
 	 * Tries to execute a server tick
 	 */
-	public function tick(){
+	private function tick(){
-		if($this->inTick === false){
 		$tickTime = microtime(true);
 		if($tickTime < $this->nextTick){
 			return false;
@@ -2041,12 +2026,10 @@ class Server{
 
 		Timings::$serverTickTimer->startTiming();
 
-			$this->inTick = true; //Fix race conditions
 		++$this->tickCounter;
 
 		$this->checkConsole();
 
-			//TODO: move this to tick
 		Timings::$connectionTimer->startTiming();
 		foreach($this->interfaces as $interface){
 			$interface->process();
@@ -2056,6 +2039,7 @@ class Server{
 		Timings::$schedulerTimer->startTiming();
 		$this->scheduler->mainThreadHeartbeat($this->tickCounter);
 		Timings::$schedulerTimer->stopTiming();
+
 		$this->checkTickUpdates($this->tickCounter);
 
 		if(($this->tickCounter & 0b1111) === 0){
@@ -2086,7 +2070,4 @@ class Server{
 		return true;
 	}
 
-		return false;
-	}
-
 }

src/pocketmine/Thread.php

@@ -29,6 +29,6 @@ abstract class Thread extends \Thread{
 	public final function start($options = PTHREADS_INHERIT_ALL){
 		ThreadManager::getInstance()->add($this);
 
-		return parent::start($options & ~PTHREADS_INHERIT_CLASSES);
+		return parent::start($options);
 	}
 }

src/pocketmine/Worker.php

@@ -29,6 +29,6 @@ abstract class Worker extends \Worker{
 	public final function start($options = PTHREADS_INHERIT_ALL){
 		ThreadManager::getInstance()->add($this);
 
-		return parent::start($options & ~PTHREADS_INHERIT_CLASSES);
+		return parent::start($options);
 	}
 }

src/pocketmine/level/generator/GenerationChunkManager.php

@@ -44,14 +44,13 @@ class GenerationChunkManager implements ChunkManager{
 	protected $changes = [];
 
 	public function __construct(GenerationManager $manager, $levelID, $seed, $class, array $options){
-		if(!is_subclass_of($class, "pocketmine\\level\\generator\\Generator")){
+		if(!class_exists($class, true) or !is_subclass_of($class, "pocketmine\\level\\generator\\Generator")){
-			throw new \Exception("Class is not a subclass of Generator");
+			throw new \Exception("Class $class does not exists or is not a subclass of Generator");
 		}
 
 		$this->levelID = $levelID;
 		$this->seed = $seed;
 		$this->manager = $manager;
-
 		$this->generator = new $class($options);
 		$this->generator->init($this, new Random($seed));
 	}
@@ -75,10 +74,15 @@ class GenerationChunkManager implements ChunkManager{
 	 * @param $chunkZ
 	 *
 	 * @return FullChunk
+	 *
+	 * @throws \Exception
 	 */
 	public function getChunk($chunkX, $chunkZ){
 		$index = Level::chunkHash($chunkX, $chunkZ);
 		$chunk = !isset($this->chunks[$index]) ? $this->requestChunk($chunkX, $chunkZ) : $this->chunks[$index];
+		if($chunk === null){
+			throw new \Exception("null chunk received");
+		}
 		$this->changes[$index] = $chunk;
 
 		return $chunk;
@@ -106,14 +110,14 @@ class GenerationChunkManager implements ChunkManager{
 				unset($this->changes[$index]);
 			}
 		}
-
-		gc_collect_cycles();
 	}
 
 	public function generateChunk($chunkX, $chunkZ){
+		try{
 			$this->getChunk($chunkX, $chunkZ);
 			$this->generator->generateChunk($chunkX, $chunkZ);
 			$this->setChunkGenerated($chunkX, $chunkZ);
+		}catch(\Exception $e){}
 	}
 
 	public function populateChunk($chunkX, $chunkZ){
@@ -134,21 +138,36 @@ class GenerationChunkManager implements ChunkManager{
 	}
 
 	public function isChunkGenerated($chunkX, $chunkZ){
+		try{
 			return $this->getChunk($chunkX, $chunkZ)->isGenerated();
+		}catch(\Exception $e){
+			return false;
+		}
 	}
 
 	public function isChunkPopulated($chunkX, $chunkZ){
+		try{
 			return $this->getChunk($chunkX, $chunkZ)->isPopulated();
+		}catch(\Exception $e){
+			return false;
+		}
 	}
 
 	public function setChunkGenerated($chunkX, $chunkZ){
 		$chunk = $this->getChunk($chunkX, $chunkZ);
 		$chunk->setGenerated(true);
+		try{
+			$chunk = $this->getChunk($chunkX, $chunkZ);
+			$chunk->setGenerated(true);
+		}catch(\Exception $e){}
 	}
 
 	public function setChunkPopulated($chunkX, $chunkZ){
+
+		try{
 			$chunk = $this->getChunk($chunkX, $chunkZ);
 			$chunk->setPopulated(true);
+		}catch(\Exception $e){}
 	}
 
 	protected function requestChunk($chunkX, $chunkZ){
@@ -181,7 +200,11 @@ class GenerationChunkManager implements ChunkManager{
 	 * @return int 0-255
 	 */
 	public function getBlockIdAt($x, $y, $z){
+		try{
 			return $this->getChunk($x >> 4, $z >> 4)->getBlockId($x & 0x0f, $y & 0x7f, $z & 0x0f);
+		}catch(\Exception $e){
+			return 0;
+		}
 	}
 
 	/**
@@ -193,7 +216,9 @@ class GenerationChunkManager implements ChunkManager{
 	 * @param int $id 0-255
 	 */
 	public function setBlockIdAt($x, $y, $z, $id){
+		try{
 			$this->getChunk($x >> 4, $z >> 4)->setBlockId($x & 0x0f, $y & 0x7f, $z & 0x0f, $id & 0xff);
+		}catch(\Exception $e){}
 	}
 
 	/**
@@ -206,7 +231,11 @@ class GenerationChunkManager implements ChunkManager{
 	 * @return int 0-15
 	 */
 	public function getBlockDataAt($x, $y, $z){
+		try{
 			return $this->getChunk($x >> 4, $z >> 4)->getBlockData($x & 0x0f, $y & 0x7f, $z & 0x0f);
+		}catch(\Exception $e){
+			return 0;
+		}
 	}
 
 	/**
@@ -218,7 +247,9 @@ class GenerationChunkManager implements ChunkManager{
 	 * @param int $data 0-15
 	 */
 	public function setBlockDataAt($x, $y, $z, $data){
+		try{
 			$this->getChunk($x >> 4, $z >> 4)->setBlockData($x & 0x0f, $y & 0x7f, $z & 0x0f, $data & 0x0f);
+		}catch(\Exception $e){}
 	}
 
 	public function shutdown(){

src/pocketmine/level/generator/GenerationManager.php

@@ -89,8 +89,9 @@ class GenerationManager{
 	 */
 	const PACKET_SHUTDOWN = 0xff;
 
+	/** @var GenerationThread */
+	protected $thread;
 
-	protected $socket;
 	/** @var \Logger */
 	protected $logger;
 	/** @var \ClassLoader */
@@ -110,17 +111,18 @@ class GenerationManager{
 	protected $shutdown = false;
 
 	/**
-	 * @param resource     $socket
+	 * @param GenerationThread $thread
 	 * @param \Logger          $logger
 	 * @param \ClassLoader     $loader
 	 */
-	public function __construct($socket, \Logger $logger, \ClassLoader $loader){
+	public function __construct(GenerationThread $thread, \Logger $logger, \ClassLoader $loader){
-		$this->socket = $socket;
+		$this->thread = $thread;
 		$this->logger = $logger;
 		$this->loader = $loader;
 		$chunkX = $chunkZ = null;
 
 		while($this->shutdown !== true){
+			try{
 				if(count($this->requestQueue) > 0){
 					foreach($this->requestQueue as $levelID => $chunks){
 						if(count($chunks) === 0){
@@ -134,6 +136,9 @@ class GenerationManager{
 				}else{
 					$this->readPacket();
 				}
+			}catch(\Exception $e){
+				$this->logger->warning("[Generator Thread] Exception: ".$e->getMessage() . " on file \"".$e->getFile()."\" line ".$e->getLine());
+			}
 		}
 	}
 
@@ -204,7 +209,7 @@ class GenerationManager{
 	public function requestChunk($levelID, $chunkX, $chunkZ){
 		$this->needsChunk[$levelID] = [$chunkX, $chunkZ];
 		$binary = chr(self::PACKET_REQUEST_CHUNK) . Binary::writeInt($levelID) . Binary::writeInt($chunkX) . Binary::writeInt($chunkZ);
-		@socket_write($this->socket, Binary::writeInt(strlen($binary)) . $binary);
+		$this->thread->pushThreadToMainPacket($binary);
 
 		do{
 			$this->readPacket();
@@ -221,29 +226,11 @@ class GenerationManager{
 
 	public function sendChunk($levelID, FullChunk $chunk){
 		$binary = chr(self::PACKET_SEND_CHUNK) . Binary::writeInt($levelID) . chr(strlen($class = get_class($chunk))) . $class . $chunk->toBinary();
-		@socket_write($this->socket, Binary::writeInt(strlen($binary)) . $binary);
+		$this->thread->pushThreadToMainPacket($binary);
-	}
-
-	protected function socketRead($len){
-		$buffer = "";
-		while(strlen($buffer) < $len){
-			$buffer .= @socket_read($this->socket, $len - strlen($buffer));
-		}
-
-		return $buffer;
 	}
 
 	protected function readPacket(){
-		$len = $this->socketRead(4);
+		if(strlen($packet = $this->thread->readMainToThreadPacket()) > 0){
-		if(($len = Binary::readInt($len)) <= 0){
-			$this->shutdown = true;
-			$this->getLogger()->critical("Generation Thread found a stream error, shutting down");
-
-			return;
-		}
-
-		$packet = $this->socketRead($len);
-
 			$pid = ord($packet{0});
 			$offset = 1;
 			if($pid === self::PACKET_REQUEST_CHUNK){
@@ -290,7 +277,12 @@ class GenerationManager{
 				$this->levels = [];
 
 				$this->shutdown = true;
-			socket_close($this->socket);
+			}
+		}elseif(count($this->thread->getInternalQueue()) === 0){
+			$this->thread->synchronized(function(){
+				$this->thread->wait(50000);
+			});
+
 		}
 	}
 

src/pocketmine/level/generator/GenerationRequestManager.php

@@ -28,7 +28,6 @@ use pocketmine\utils\Binary;
 
 class GenerationRequestManager{
 
-	protected $socket;
 	/** @var Server */
 	protected $server;
 	/** @var GenerationThread */
@@ -40,7 +39,6 @@ class GenerationRequestManager{
 	public function __construct(Server $server){
 		$this->server = $server;
 		$this->generationThread = new GenerationThread($server->getLogger(), $server->getLoader());
-		$this->socket = $this->generationThread->getExternalSocket();
 	}
 
 	/**
@@ -52,7 +50,7 @@ class GenerationRequestManager{
 		$buffer = chr(GenerationManager::PACKET_OPEN_LEVEL) . Binary::writeInt($level->getID()) . Binary::writeInt($level->getSeed()) .
 			Binary::writeShort(strlen($generator)) . $generator . serialize($options);
 
-		@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+		$this->generationThread->pushMainToThreadPacket($buffer);
 	}
 
 	/**
@@ -60,31 +58,22 @@ class GenerationRequestManager{
 	 */
 	public function closeLevel(Level $level){
 		$buffer = chr(GenerationManager::PACKET_CLOSE_LEVEL) . Binary::writeInt($level->getID());
-		@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+		$this->generationThread->pushMainToThreadPacket($buffer);
 	}
 
 	public function addNamespace($namespace, $path){
 		$buffer = chr(GenerationManager::PACKET_ADD_NAMESPACE) . Binary::writeShort(strlen($namespace)) . $namespace . $path;
-		@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+		$this->generationThread->pushMainToThreadPacket($buffer);
-	}
-
-	protected function socketRead($len){
-		$buffer = "";
-		while(strlen($buffer) < $len){
-			$buffer .= @socket_read($this->socket, $len - strlen($buffer));
-		}
-
-		return $buffer;
 	}
 
 	protected function sendChunk($levelID, FullChunk $chunk){
-		$binary = chr(GenerationManager::PACKET_SEND_CHUNK) . Binary::writeInt($levelID) . chr(strlen($class = get_class($chunk))) . $class . $chunk->toBinary();
+		$buffer = chr(GenerationManager::PACKET_SEND_CHUNK) . Binary::writeInt($levelID) . chr(strlen($class = get_class($chunk))) . $class . $chunk->toBinary();
-		@socket_write($this->socket, Binary::writeInt(strlen($binary)) . $binary);
+		$this->generationThread->pushMainToThreadPacket($buffer);
 	}
 
 	public function requestChunk(Level $level, $chunkX, $chunkZ){
 		$buffer = chr(GenerationManager::PACKET_REQUEST_CHUNK) . Binary::writeInt($level->getID()) . Binary::writeInt($chunkX) . Binary::writeInt($chunkZ);
-		@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+		$this->generationThread->pushMainToThreadPacket($buffer);
 	}
 
 	protected function handleRequest($levelID, $chunkX, $chunkZ){
@@ -97,7 +86,7 @@ class GenerationRequestManager{
 			}
 		}else{
 			$buffer = chr(GenerationManager::PACKET_CLOSE_LEVEL) . Binary::writeInt($levelID);
-			@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+			$this->generationThread->pushMainToThreadPacket($buffer);
 		}
 	}
 
@@ -106,17 +95,12 @@ class GenerationRequestManager{
 			$level->generateChunkCallback($chunk->getX(), $chunk->getZ(), $chunk);
 		}else{
 			$buffer = chr(GenerationManager::PACKET_CLOSE_LEVEL) . Binary::writeInt($levelID);
-			@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+			$this->generationThread->pushMainToThreadPacket($buffer);
 		}
 	}
 
 	public function handlePackets(){
-		if(($len = @socket_read($this->socket, 4)) !== false and $len !== ""){
+		while(strlen($packet = $this->generationThread->readThreadToMainPacket()) > 0){
-			if(strlen($len) < 4){
-				$len .= $this->socketRead(4 - strlen($len));
-			}
-
-			$packet = $this->socketRead(Binary::readInt($len));
 			$pid = ord($packet{0});
 			$offset = 1;
 
@@ -145,7 +129,7 @@ class GenerationRequestManager{
 
 	public function shutdown(){
 		$buffer = chr(GenerationManager::PACKET_SHUTDOWN);
-		@socket_write($this->socket, Binary::writeInt(strlen($buffer)) . $buffer);
+		$this->generationThread->pushMainToThreadPacket($buffer);
 		$this->generationThread->join();
 	}
 

src/pocketmine/level/generator/GenerationThread.php

@@ -32,15 +32,42 @@ class GenerationThread extends Thread{
 	/** @var \ThreadedLogger */
 	protected $logger;
 
-	protected $externalSocket;
+	/** @var \Threaded */
-	protected $internalSocket;
+	protected $externalQueue;
+	/** @var \Threaded */
+	protected $internalQueue;
 
-	public function getExternalSocket(){
+	/**
-		return $this->externalSocket;
+	 * @return \Threaded
+	 */
+	public function getInternalQueue(){
+		return $this->internalQueue;
 	}
 
-	public function getInternalSocket(){
+	/**
-		return $this->internalSocket;
+	 * @return \Threaded
+	 */
+	public function getExternalQueue(){
+		return $this->externalQueue;
+	}
+
+	public function pushMainToThreadPacket($str){
+		$this->internalQueue[] = $str;
+		$this->synchronized(function(){
+			$this->notify();
+		});
+	}
+
+	public function readMainToThreadPacket(){
+		return $this->internalQueue->shift();
+	}
+
+	public function pushThreadToMainPacket($str){
+		$this->externalQueue[] = $str;
+	}
+
+	public function readThreadToMainPacket(){
+		return $this->externalQueue->shift();
 	}
 
 	/**
@@ -57,19 +84,8 @@ class GenerationThread extends Thread{
 		$this->addDependency($loadPaths, new \ReflectionClass($this->loader));
 		$this->loadPaths = array_reverse($loadPaths);
 
-		$sockets = [];
+		$this->externalQueue = new \Threaded();
-		if(!socket_create_pair((strtoupper(substr(PHP_OS, 0, 3)) === 'WIN' ? AF_INET : AF_UNIX), SOCK_STREAM, 0, $sockets)){
+		$this->internalQueue = new \Threaded();
-			throw new \Exception("Could not create IPC sockets. Reason: " . socket_strerror(socket_last_error()));
-		}
-
-		$this->internalSocket = $sockets[0];
-		socket_set_block($this->internalSocket); //IMPORTANT!
-		@socket_set_option($this->internalSocket, SOL_SOCKET, SO_SNDBUF, 1024 * 1024 * 2);
-		@socket_set_option($this->internalSocket, SOL_SOCKET, SO_RCVBUF, 1024 * 1024 * 2);
-		$this->externalSocket = $sockets[1];
-		socket_set_nonblock($this->externalSocket);
-		@socket_set_option($this->externalSocket, SOL_SOCKET, SO_SNDBUF, 1024 * 1024 * 2);
-		@socket_set_option($this->externalSocket, SOL_SOCKET, SO_RCVBUF, 1024 * 1024 * 2);
 
 		$this->start();
 	}
@@ -90,14 +106,15 @@ class GenerationThread extends Thread{
 
 	public function run(){
 		error_reporting(-1);
+		gc_enable();
 		//Load removed dependencies, can't use require_once()
 		foreach($this->loadPaths as $name => $path){
-			if(!class_exists($name, false) and !class_exists($name, false)){
+			if(!class_exists($name, false) and !interface_exists($name, false)){
 				require($path);
 			}
 		}
 		$this->loader->register();
 
-		$generationManager = new GenerationManager($this->getInternalSocket(), $this->getLogger(), $this->loader);
+		$generationManager = new GenerationManager($this, $this->getLogger(), $this->loader);
 	}
 }

src/pocketmine/plugin/PluginManager.php

@@ -436,6 +436,7 @@ class PluginManager{
 	 */
 	public function unsubscribeFromPermission($permission, Permissible $permissible){
 		if(isset($this->permSubs[$permission])){
+			$this->permSubs[$permission][spl_object_hash($permissible)]->release();
 			unset($this->permSubs[$permission][spl_object_hash($permissible)]);
 			if(count($this->permSubs[$permission]) === 0){
 				unset($this->permSubs[$permission]);

src/pocketmine/scheduler/AsyncWorker.php

@@ -26,8 +26,12 @@ use pocketmine\Worker;
 class AsyncWorker extends Worker{
 
 	public function run(){
+		if(!interface_exists("ClassLoader", false)){
 			require(\pocketmine\PATH . "src/spl/ClassLoader.php");
+		}
+		if(!class_exists("BaseClassLoader", false)){
 			require(\pocketmine\PATH . "src/spl/BaseClassLoader.php");
+		}
 		$autoloader = new \BaseClassLoader();
 		$autoloader->addPath(\pocketmine\PATH . "src");
 		$autoloader->register(true);

src/pocketmine/utils/Random.php

@@ -46,7 +46,7 @@ class Random{
 	 * @param int $seed Integer to be used as seed.
 	 */
 	public function setSeed($seed){
-		$this->seed = crc32(Binary::writeInt($seed));
+		$this->seed = crc32(pack("N", $seed));
 	}
 
 	/**
@@ -64,14 +64,14 @@ class Random{
 	 * @return int
 	 */
 	public function nextSignedInt(){
-		$t = crc32(Binary::writeInt($this->seed));
+		$t = crc32(pack("N", $this->seed));
 		$this->seed ^= $t;
 
-		if($t > 2147483647){
+		if(PHP_INT_SIZE === 8){
-			$t -= 4294967296;
+			return $t << 32 >> 32;
+		}else{
+			return $t;
 		}
-
-		return (int) $t;
 	}
 
 	/**
@@ -105,11 +105,11 @@ class Random{
 	 * Returns a random integer between $start and $end
 	 *
 	 * @param int $start default 0
-	 * @param int $end   default PHP_INT_MAX
+	 * @param int $end   default 0x7fffffff
 	 *
 	 * @return int
 	 */
-	public function nextRange($start = 0, $end = PHP_INT_MAX){
+	public function nextRange($start = 0, $end = 0x7fffffff){
 		return $start + ($this->nextInt() % ($end + 1 - $start));
 	}
 

src/raklib

@@ -1 +1 @@
-Subproject commit 6fad38f761d9ee204839eb557503645799b2ec99
+Subproject commit c32adbd6f3e06e4002fce6987651e421b784e997