First look at separating chunk sending from Level

src/pocketmine/Player.php

@@ -96,7 +96,6 @@ use pocketmine\nbt\tag\CompoundTag;
 use pocketmine\nbt\tag\DoubleTag;
 use pocketmine\nbt\tag\IntTag;
 use pocketmine\nbt\tag\ListTag;
-use pocketmine\network\mcpe\CompressBatchPromise;
 use pocketmine\network\mcpe\NetworkSession;
 use pocketmine\network\mcpe\protocol\AnimatePacket;
 use pocketmine\network\mcpe\protocol\BookEditPacket;
@@ -954,12 +953,7 @@ class Player extends Human implements CommandSender, ChunkLoader, ChunkListener,
 		$level = $level ?? $this->level;
 		$index = Level::chunkHash($x, $z);
 		if(isset($this->usedChunks[$index])){
-			foreach($level->getChunkEntities($x, $z) as $entity){
+			$this->networkSession->stopUsingChunk($x, $z);
-				if($entity !== $this){
-					$entity->despawnFrom($this);
-				}
-			}
-
 			unset($this->usedChunks[$index]);
 		}
 		$level->unregisterChunkLoader($this, $x, $z);
@@ -967,39 +961,24 @@ class Player extends Human implements CommandSender, ChunkLoader, ChunkListener,
 		unset($this->loadQueue[$index]);
 	}
 
-	public function sendChunk(int $x, int $z, CompressBatchPromise $promise){
+	public function onChunkReady(int $x, int $z){
 		if(!$this->isConnected()){
 			return;
 		}
 
+		assert(isset($this->usedChunks[Level::chunkHash($x, $z)]));
 		$this->usedChunks[Level::chunkHash($x, $z)] = true;
 
-		$this->networkSession->queueCompressed($promise);
+		$spawn = ++$this->spawnChunkLoadCount === $this->spawnThreshold;
+		$this->networkSession->startUsingChunk($x, $z, $spawn);
 
-		if($this->spawned){
+		if($spawn){
-			foreach($this->level->getChunkEntities($x, $z) as $entity){
+			//TODO: not sure this should be here
-				if($entity !== $this and !$entity->isClosed() and $entity->isAlive()){
-					$entity->spawnTo($this);
-				}
-			}
-		}elseif(++$this->spawnChunkLoadCount >= $this->spawnThreshold){
-			$this->spawnChunkLoadCount = -1;
 			$this->spawned = true;
-
-			foreach($this->usedChunks as $index => $c){
-				Level::getXZ($index, $chunkX, $chunkZ);
-				foreach($this->level->getChunkEntities($chunkX, $chunkZ) as $entity){
-					if($entity !== $this and !$entity->isClosed() and $entity->isAlive() and !$entity->isFlaggedForDespawn()){
-						$entity->spawnTo($this);
-					}
-				}
-			}
-
-			$this->networkSession->onTerrainReady();
 		}
 	}
 
-	protected function sendNextChunk(){
+	protected function requestChunks(){
 		if(!$this->isConnected()){
 			return;
 		}
@@ -1146,7 +1125,7 @@ class Player extends Human implements CommandSender, ChunkLoader, ChunkListener,
 		}
 
 		if(count($this->loadQueue) > 0){
-			$this->sendNextChunk();
+			$this->requestChunks();
 		}
 	}
 

src/pocketmine/level/Level.php

@@ -72,8 +72,6 @@ use pocketmine\metadata\Metadatable;
 use pocketmine\metadata\MetadataValue;
 use pocketmine\nbt\tag\ListTag;
 use pocketmine\nbt\tag\StringTag;
-use pocketmine\network\mcpe\ChunkRequestTask;
-use pocketmine\network\mcpe\CompressBatchPromise;
 use pocketmine\network\mcpe\protocol\BlockEntityDataPacket;
 use pocketmine\network\mcpe\protocol\ClientboundPacket;
 use pocketmine\network\mcpe\protocol\LevelEventPacket;
@@ -148,9 +146,6 @@ class Level implements ChunkManager, Metadatable{
 	/** @var Block[][] */
 	private $blockCache = [];
 
-	/** @var CompressBatchPromise[] */
-	private $chunkCache = [];
-
 	/** @var int */
 	private $sendTimeTicker = 0;
 
@@ -221,8 +216,6 @@ class Level implements ChunkManager, Metadatable{
 
 	/** @var Player[][] */
 	private $chunkSendQueue = [];
-	/** @var ChunkRequestTask[] */
-	private $chunkSendTasks = [];
 
 	/** @var bool[] */
 	private $chunkPopulationQueue = [];
@@ -843,7 +836,6 @@ class Level implements ChunkManager, Metadatable{
 					if(empty($blocks)){ //blocks can be set normally and then later re-set with direct send
 						continue;
 					}
-					unset($this->chunkCache[$index]);
 					Level::getXZ($index, $chunkX, $chunkZ);
 					if(count($blocks) > 512){
 						$chunk = $this->getChunk($chunkX, $chunkZ);
@@ -854,8 +846,6 @@ class Level implements ChunkManager, Metadatable{
 						$this->sendBlocks($this->getChunkPlayers($chunkX, $chunkZ), $blocks);
 					}
 				}
-			}else{
-				$this->chunkCache = [];
 			}
 
 			$this->changedBlocks = [];
@@ -960,7 +950,6 @@ class Level implements ChunkManager, Metadatable{
 
 	public function clearCache(bool $force = false){
 		if($force){
-			$this->chunkCache = [];
 			$this->blockCache = [];
 		}else{
 			$count = 0;
@@ -974,10 +963,6 @@ class Level implements ChunkManager, Metadatable{
 		}
 	}
 
-	public function clearChunkCache(int $chunkX, int $chunkZ){
-		unset($this->chunkCache[Level::chunkHash($chunkX, $chunkZ)]);
-	}
-
 	public function getRandomTickedBlocks() : \SplFixedArray{
 		return $this->randomTickBlocks;
 	}
@@ -2319,17 +2304,13 @@ class Level implements ChunkManager, Metadatable{
 		$this->chunks[$chunkHash] = $chunk;
 
 		unset($this->blockCache[$chunkHash]);
-		unset($this->chunkCache[$chunkHash]);
 		unset($this->changedBlocks[$chunkHash]);
-		if(isset($this->chunkSendTasks[$chunkHash])){ //invalidate pending caches
-			$this->chunkSendTasks[$chunkHash]->cancelRun();
-			unset($this->chunkSendTasks[$chunkHash]);
-		}
 		$chunk->setChanged();
 
 		if(!$this->isChunkInUse($chunkX, $chunkZ)){
 			$this->unloadChunkRequest($chunkX, $chunkZ);
 		}
+
 		foreach($this->getChunkListeners($chunkX, $chunkZ) as $listener){
 			$listener->onChunkChanged($chunk);
 		}
@@ -2419,13 +2400,11 @@ class Level implements ChunkManager, Metadatable{
 		$this->chunkSendQueue[$index][spl_object_id($player)] = $player;
 	}
 
-	private function sendCachedChunk(int $x, int $z){
+	private function onChunkReady(int $x, int $z){
 		if(isset($this->chunkSendQueue[$index = Level::chunkHash($x, $z)])){
 			foreach($this->chunkSendQueue[$index] as $player){
 				/** @var Player $player */
-				if($player->isConnected() and isset($player->usedChunks[$index])){
+				$player->onChunkReady($x, $z);
-					$player->sendChunk($x, $z, $this->chunkCache[$index]);
-				}
 			}
 			unset($this->chunkSendQueue[$index]);
 		}
@@ -2438,55 +2417,13 @@ class Level implements ChunkManager, Metadatable{
 			foreach($this->chunkSendQueue as $index => $players){
 				Level::getXZ($index, $x, $z);
 
-				if(isset($this->chunkSendTasks[$index])){
-					//Not ready for sending yet
-					continue;
-				}
-
-				if(isset($this->chunkCache[$index])){
-					$this->sendCachedChunk($x, $z);
-					continue;
-				}
-
 				$this->timings->syncChunkSendPrepareTimer->startTiming();
 
 				$chunk = $this->chunks[$index] ?? null;
-				if(!($chunk instanceof Chunk)){
+				if($chunk === null or !$chunk->isGenerated() or !$chunk->isPopulated()){
 					throw new ChunkException("Invalid Chunk sent");
 				}
-				assert($chunk->getX() === $x and $chunk->getZ() === $z, "Chunk coordinate mismatch: expected $x $z, but chunk has coordinates " . $chunk->getX() . " " . $chunk->getZ() . ", did you forget to clone a chunk before setting?");
+				$this->onChunkReady($x, $z);
-
-				/*
-				 * we don't send promises directly to the players here because unresolved promises of chunk sending
-				 * would slow down the sending of other packets, especially if a chunk takes a long time to prepare.
-				 */
-
-				$promise = new CompressBatchPromise();
-				$promise->onResolve(function(CompressBatchPromise $promise) use ($x, $z, $index): void{
-					if(!$this->closed){
-						$this->timings->syncChunkSendTimer->startTiming();
-
-						unset($this->chunkSendTasks[$index]);
-
-						$this->chunkCache[$index] = $promise;
-						$this->sendCachedChunk($x, $z);
-						if(!$this->server->getMemoryManager()->canUseChunkCache()){
-							unset($this->chunkCache[$index]);
-						}
-
-						$this->timings->syncChunkSendTimer->stopTiming();
-					}else{
-						$this->server->getLogger()->debug("Dropped prepared chunk $x $z due to world not loaded");
-					}
-				});
-				$this->server->getAsyncPool()->submitTask($task = new ChunkRequestTask($x, $z, $chunk, $promise, function() use($index, $x, $z){
-					if(isset($this->chunkSendTasks[$index])){
-						unset($this->chunkSendTasks[$index]);
-						$this->server->getLogger()->error("Failed to prepare chunk $x $z for sending, retrying");
-					}
-				}));
-				$this->chunkSendTasks[$index] = $task;
-
 				$this->timings->syncChunkSendPrepareTimer->stopTiming();
 			}
 
@@ -2577,7 +2514,9 @@ class Level implements ChunkManager, Metadatable{
 		if(isset($this->chunks[$hash = Level::chunkHash($chunkX, $chunkZ)])){
 			$this->chunks[$hash]->removeTile($tile);
 		}
-		$this->clearChunkCache($chunkX, $chunkZ);
+		foreach($this->getChunkListeners($chunkX, $chunkZ) as $listener){
+			$listener->onBlockChanged($tile);
+		}
 	}
 
 	/**
@@ -2712,11 +2651,9 @@ class Level implements ChunkManager, Metadatable{
 		}
 
 		unset($this->chunks[$chunkHash]);
-		unset($this->chunkCache[$chunkHash]);
 		unset($this->blockCache[$chunkHash]);
 		unset($this->changedBlocks[$chunkHash]);
 		unset($this->chunkSendQueue[$chunkHash]);
-		unset($this->chunkSendTasks[$chunkHash]);
 
 		$this->timings->doChunkUnload->stopTiming();
 

src/pocketmine/network/mcpe/ChunkCache.php

@@ -0,0 +1,231 @@
+<?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\network\mcpe;
+
+use pocketmine\level\ChunkListener;
+use pocketmine\level\format\Chunk;
+use pocketmine\level\Level;
+use pocketmine\math\Vector3;
+use function spl_object_id;
+use function strlen;
+
+/**
+ * This class is used by the current MCPE protocol system to store cached chunk packets for fast resending.
+ *
+ * TODO: make MemoryManager aware of this so the cache can be destroyed when memory is low
+ * TODO: this needs a hook for world unloading
+ */
+class ChunkCache implements ChunkListener{
+	/** @var self[] */
+	private static $instances = [];
+
+	/**
+	 * Fetches the ChunkCache instance for the given world. This lazily creates cache systems as needed.
+	 *
+	 * @param Level $world
+	 *
+	 * @return ChunkCache
+	 */
+	public static function getInstance(Level $world) : self{
+		return self::$instances[spl_object_id($world)] ?? (self::$instances[spl_object_id($world)] = new self($world));
+	}
+
+	/** @var Level */
+	private $world;
+
+	/** @var CompressBatchPromise[] */
+	private $caches = [];
+
+	/** @var int */
+	private $hits = 0;
+	/** @var int */
+	private $misses = 0;
+
+	/**
+	 * @param Level $world
+	 */
+	private function __construct(Level $world){
+		$this->world = $world;
+	}
+
+	/**
+	 * Requests asynchronous preparation of the chunk at the given coordinates.
+	 *
+	 * @param int $chunkX
+	 * @param int $chunkZ
+	 *
+	 * @return CompressBatchPromise a promise of resolution which will contain a compressed chunk packet.
+	 */
+	public function request(int $chunkX, int $chunkZ) : CompressBatchPromise{
+		$this->world->registerChunkListener($this, $chunkX, $chunkZ);
+		$chunkHash = Level::chunkHash($chunkX, $chunkZ);
+
+		if(isset($this->caches[$chunkHash])){
+			++$this->hits;
+			return $this->caches[$chunkHash];
+		}
+
+		++$this->misses;
+
+		$this->world->timings->syncChunkSendPrepareTimer->startTiming();
+		try{
+			$this->caches[$chunkHash] = new CompressBatchPromise();
+
+			$this->world->getServer()->getAsyncPool()->submitTask(
+				new ChunkRequestTask(
+					$chunkX,
+					$chunkZ,
+					$this->world->getChunk($chunkX, $chunkZ),
+					$this->caches[$chunkHash],
+					function() use($chunkX, $chunkZ){
+						$this->world->getServer()->getLogger()->error("Failed preparing chunk for " . $this->world->getDisplayName() . " chunk $chunkX $chunkZ, retrying");
+
+						$this->restartPendingRequest($chunkX, $chunkZ);
+					}
+				)
+			);
+
+			return $this->caches[$chunkHash];
+		}finally{
+			$this->world->timings->syncChunkSendPrepareTimer->stopTiming();
+		}
+	}
+
+	private function destroy(int $chunkX, int $chunkZ) : bool{
+		$chunkHash = Level::chunkHash($chunkX, $chunkZ);
+		$existing = $this->caches[$chunkHash] ?? null;
+		unset($this->caches[$chunkHash]);
+
+		return $existing !== null;
+	}
+
+	/**
+	 * Restarts an async request for an unresolved chunk.
+	 *
+	 * @param int $chunkX
+	 * @param int $chunkZ
+	 *
+	 * @throws \InvalidArgumentException
+	 */
+	private function restartPendingRequest(int $chunkX, int $chunkZ) : void{
+		$chunkHash = Level::chunkHash($chunkX, $chunkZ);
+		$existing = $this->caches[$chunkHash];
+		if($existing === null or $existing->hasResult()){
+			throw new \InvalidArgumentException("Restart can only be applied to unresolved promises");
+		}
+		$existing->cancel();
+		unset($this->caches[$chunkHash]);
+
+		$this->request($chunkX, $chunkZ)->onResolve(...$existing->getResolveCallbacks());
+	}
+
+	/**
+	 * @param int $chunkX
+	 * @param int $chunkZ
+	 *
+	 * @throws \InvalidArgumentException
+	 */
+	private function destroyOrRestart(int $chunkX, int $chunkZ) : void{
+		$cache = $this->caches[Level::chunkHash($chunkX, $chunkZ)] ?? null;
+		if($cache !== null){
+			if(!$cache->hasResult()){
+				//some requesters are waiting for this chunk, so their request needs to be fulfilled
+				$this->restartPendingRequest($chunkX, $chunkZ);
+			}else{
+				//dump the cache, it'll be regenerated the next time it's requested
+				$this->destroy($chunkX, $chunkZ);
+			}
+		}
+	}
+
+	/**
+	 * @see ChunkListener::onChunkChanged()
+	 * @param Chunk $chunk
+	 */
+	public function onChunkChanged(Chunk $chunk) : void{
+		//FIXME: this gets fired for stuff that doesn't change terrain related things (like lighting updates)
+		$this->destroyOrRestart($chunk->getX(), $chunk->getZ());
+	}
+
+	/**
+	 * @see ChunkListener::onBlockChanged()
+	 * @param Vector3 $block
+	 */
+	public function onBlockChanged(Vector3 $block) : void{
+		//FIXME: requesters will still receive this chunk after it's been dropped, but we can't mark this for a simple
+		//sync here because it can spam the worker pool
+		$this->destroy($block->getFloorX() >> 4, $block->getFloorZ() >> 4);
+	}
+
+	/**
+	 * @see ChunkListener::onChunkUnloaded()
+	 * @param Chunk $chunk
+	 */
+	public function onChunkUnloaded(Chunk $chunk) : void{
+		$this->destroy($chunk->getX(), $chunk->getZ());
+		$this->world->unregisterChunkListener($this, $chunk->getX(), $chunk->getZ());
+	}
+
+	/**
+	 * @see ChunkListener::onChunkLoaded()
+	 * @param Chunk $chunk
+	 */
+	public function onChunkLoaded(Chunk $chunk) : void{
+		//NOOP
+	}
+
+	/**
+	 * @see ChunkListener::onChunkPopulated()
+	 * @param Chunk $chunk
+	 */
+	public function onChunkPopulated(Chunk $chunk) : void{
+		//NOOP - we also receive this in onChunkChanged, so we don't care here
+	}
+
+	/**
+	 * Returns the number of bytes occupied by the cache data in this cache. This does not include the size of any
+	 * promises referenced by the cache.
+	 *
+	 * @return int
+	 */
+	public function calculateCacheSize() : int{
+		$result = 0;
+		foreach($this->caches as $cache){
+			if($cache->hasResult()){
+				$result += strlen($cache->getResult());
+			}
+		}
+		return $result;
+	}
+
+	/**
+	 * Returns the percentage of requests to the cache which resulted in a cache hit.
+	 *
+	 * @return float
+	 */
+	public function getHitPercentage() : float{
+		$total = $this->hits + $this->misses;
+		return $total > 0 ? $this->hits / $total : 0.0;
+	}
+}

src/pocketmine/network/mcpe/CompressBatchPromise.php

@@ -23,6 +23,9 @@ declare(strict_types=1);
 
 namespace pocketmine\network\mcpe;
 
+use pocketmine\utils\Utils;
+use function array_push;
+
 class CompressBatchPromise{
 	/** @var callable[] */
 	private $callbacks = [];
@@ -30,26 +33,45 @@ class CompressBatchPromise{
 	/** @var string|null */
 	private $result = null;
 
-	public function onResolve(callable $callback) : void{
+	/** @var bool */
+	private $cancelled = false;
+
+	public function onResolve(callable ...$callbacks) : void{
+		$this->checkCancelled();
+		foreach($callbacks as $callback){
+			Utils::validateCallableSignature(function(CompressBatchPromise $promise){}, $callback);
+		}
 		if($this->result !== null){
-			$callback($this);
+			foreach($callbacks as $callback){
+				$callback($this);
+			}
 		}else{
-			$this->callbacks[] = $callback;
+			array_push($this->callbacks, ...$callbacks);
 		}
 	}
 
 	public function resolve(string $result) : void{
-		if($this->result !== null){
+		if(!$this->cancelled){
-			throw new \InvalidStateException("Cannot resolve promise more than once");
+			if($this->result !== null){
+				throw new \InvalidStateException("Cannot resolve promise more than once");
+			}
+			$this->result = $result;
+			foreach($this->callbacks as $callback){
+				$callback($this);
+			}
+			$this->callbacks = [];
 		}
-		$this->result = $result;
+	}
-		foreach($this->callbacks as $callback){
+
-			$callback($this);
+	/**
-		}
+	 * @return callable[]
-		$this->callbacks = [];
+	 */
+	public function getResolveCallbacks() : array{
+		return $this->callbacks;
 	}
 
 	public function getResult() : string{
+		$this->checkCancelled();
 		if($this->result === null){
 			throw new \InvalidStateException("Promise has not yet been resolved");
 		}
@@ -59,4 +81,24 @@ class CompressBatchPromise{
 	public function hasResult() : bool{
 		return $this->result !== null;
 	}
+
+	/**
+	 * @return bool
+	 */
+	public function isCancelled() : bool{
+		return $this->cancelled;
+	}
+
+	public function cancel() : void{
+		if($this->hasResult()){
+			throw new \InvalidStateException("Cannot cancel a resolved promise");
+		}
+		$this->cancelled = true;
+	}
+
+	private function checkCancelled() : void{
+		if($this->cancelled){
+			throw new \InvalidArgumentException("Promise has been cancelled");
+		}
+	}
 }

src/pocketmine/network/mcpe/NetworkSession.php

@@ -744,6 +744,43 @@ class NetworkSession{
 		return $this->sendDataPacket($pk);
 	}
 
+	public function startUsingChunk(int $chunkX, int $chunkZ, bool $spawn = false) : void{
+		ChunkCache::getInstance($this->player->getLevel())->request($chunkX, $chunkZ)->onResolve(
+
+			//this callback may be called synchronously or asynchronously, depending on whether the promise is resolved yet
+			function(CompressBatchPromise $promise) use($chunkX, $chunkZ, $spawn){
+				if(!$this->isConnected()){
+					return;
+				}
+				$this->player->level->timings->syncChunkSendTimer->startTiming();
+				try{
+					$this->queueCompressed($promise);
+
+					foreach($this->player->getLevel()->getChunkEntities($chunkX, $chunkZ) as $entity){
+						if($entity !== $this->player and !$entity->isClosed() and !$entity->isFlaggedForDespawn()){
+							$entity->spawnTo($this->player);
+						}
+					}
+
+					if($spawn){
+						//TODO: potential race condition during chunk sending could cause this to be called too early
+						$this->onTerrainReady();
+					}
+				}finally{
+					$this->player->level->timings->syncChunkSendTimer->stopTiming();
+				}
+			}
+		);
+	}
+
+	public function stopUsingChunk(int $chunkX, int $chunkZ) : void{
+		foreach($this->player->getLevel()->getChunkEntities($chunkX, $chunkZ) as $entity){
+			if($entity !== $this->player){
+				$entity->despawnFrom($this->player);
+			}
+		}
+	}
+
 	public function tick() : bool{
 		if($this->handler instanceof LoginSessionHandler){
 			if(time() >= $this->connectTime + 10){