Merge branch 'resource-pack-ack-receipts' into minor-next

src/network/mcpe/NetworkSession.php

@@ -101,6 +101,8 @@ use pocketmine\player\Player;
 use pocketmine\player\PlayerInfo;
 use pocketmine\player\UsedChunkStatus;
 use pocketmine\player\XboxLivePlayerInfo;
+use pocketmine\promise\Promise;
+use pocketmine\promise\PromiseResolver;
 use pocketmine\Server;
 use pocketmine\timings\Timings;
 use pocketmine\utils\AssumptionFailedError;
@@ -159,15 +161,24 @@ class NetworkSession{
 
 	/** @var string[] */
 	private array $sendBuffer = [];
-
 	/**
-	 * @var \SplQueue|CompressBatchPromise[]|string[]
-	 * @phpstan-var \SplQueue<CompressBatchPromise|string>
+	 * @var PromiseResolver[]
+	 * @phpstan-var list<PromiseResolver<true>>
 	 */
+	private array $sendBufferAckPromises = [];
+
+	/** @phpstan-var \SplQueue<array{CompressBatchPromise|string, list<PromiseResolver<true>>}> */
 	private \SplQueue $compressedQueue;
 	private bool $forceAsyncCompression = true;
 	private bool $enableCompression = false; //disabled until handshake completed
 
+	private int $nextAckReceiptId = 0;
+	/**
+	 * @var PromiseResolver[][]
+	 * @phpstan-var array<int, list<PromiseResolver<true>>>
+	 */
+	private array $ackPromisesByReceiptId = [];
+
 	private ?InventoryManager $invManager = null;
 
 	/**
@@ -465,7 +476,23 @@ class NetworkSession{
 		}
 	}
 
-	public function sendDataPacket(ClientboundPacket $packet, bool $immediate = false) : bool{
+	public function handleAckReceipt(int $receiptId) : void{
+		if(!$this->connected){
+			return;
+		}
+		if(isset($this->ackPromisesByReceiptId[$receiptId])){
+			$promises = $this->ackPromisesByReceiptId[$receiptId];
+			unset($this->ackPromisesByReceiptId[$receiptId]);
+			foreach($promises as $promise){
+				$promise->resolve(true);
+			}
+		}
+	}
+
+	/**
+	 * @phpstan-param PromiseResolver<true>|null $ackReceiptResolver
+	 */
+	private function sendDataPacketInternal(ClientboundPacket $packet, bool $immediate, ?PromiseResolver $ackReceiptResolver) : bool{
 		if(!$this->connected){
 			return false;
 		}
@@ -488,6 +515,9 @@ class NetworkSession{
 				$packets = [$packet];
 			}
 
+			if($ackReceiptResolver !== null){
+				$this->sendBufferAckPromises[] = $ackReceiptResolver;
+			}
 			foreach($packets as $evPacket){
 				$this->addToSendBuffer(self::encodePacketTimed(PacketSerializer::encoder(), $evPacket));
 			}
@@ -501,6 +531,23 @@ class NetworkSession{
 		}
 	}
 
+	public function sendDataPacket(ClientboundPacket $packet, bool $immediate = false) : bool{
+		return $this->sendDataPacketInternal($packet, $immediate, null);
+	}
+
+	/**
+	 * @phpstan-return Promise<true>
+	 */
+	public function sendDataPacketWithReceipt(ClientboundPacket $packet, bool $immediate = false) : Promise{
+		$resolver = new PromiseResolver();
+
+		if(!$this->sendDataPacketInternal($packet, $immediate, $resolver)){
+			$resolver->reject();
+		}
+
+		return $resolver->getPromise();
+	}
+
 	/**
 	 * @internal
 	 */
@@ -542,7 +589,9 @@ class NetworkSession{
 					$batch = $stream->getBuffer();
 				}
 				$this->sendBuffer = [];
-				$this->queueCompressedNoBufferFlush($batch, $immediate);
+				$ackPromises = $this->sendBufferAckPromises;
+				$this->sendBufferAckPromises = [];
+				$this->queueCompressedNoBufferFlush($batch, $immediate, $ackPromises);
 			}finally{
 				Timings::$playerNetworkSend->stopTiming();
 			}
@@ -569,22 +618,27 @@ class NetworkSession{
 		}
 	}
 
-	private function queueCompressedNoBufferFlush(CompressBatchPromise|string $batch, bool $immediate = false) : void{
+	/**
+	 * @param PromiseResolver[] $ackPromises
+	 *
+	 * @phpstan-param list<PromiseResolver<true>> $ackPromises
+	 */
+	private function queueCompressedNoBufferFlush(CompressBatchPromise|string $batch, bool $immediate = false, array $ackPromises = []) : void{
 		Timings::$playerNetworkSend->startTiming();
 		try{
 			if(is_string($batch)){
 				if($immediate){
 					//Skips all queues
-					$this->sendEncoded($batch, true);
+					$this->sendEncoded($batch, true, $ackPromises);
 				}else{
-					$this->compressedQueue->enqueue($batch);
+					$this->compressedQueue->enqueue([$batch, $ackPromises]);
 					$this->flushCompressedQueue();
 				}
 			}elseif($immediate){
 				//Skips all queues
-				$this->sendEncoded($batch->getResult(), true);
+				$this->sendEncoded($batch->getResult(), true, $ackPromises);
 			}else{
-				$this->compressedQueue->enqueue($batch);
+				$this->compressedQueue->enqueue([$batch, $ackPromises]);
 				$batch->onResolve(function() : void{
 					if($this->connected){
 						$this->flushCompressedQueue();
@@ -601,14 +655,14 @@ class NetworkSession{
 		try{
 			while(!$this->compressedQueue->isEmpty()){
 				/** @var CompressBatchPromise|string $current */
-				$current = $this->compressedQueue->bottom();
+				[$current, $ackPromises] = $this->compressedQueue->bottom();
 				if(is_string($current)){
 					$this->compressedQueue->dequeue();
-					$this->sendEncoded($current);
+					$this->sendEncoded($current, false, $ackPromises);
 
 				}elseif($current->hasResult()){
 					$this->compressedQueue->dequeue();
-					$this->sendEncoded($current->getResult());
+					$this->sendEncoded($current->getResult(), false, $ackPromises);
 
 				}else{
 					//can't send any more queued until this one is ready
@@ -620,13 +674,24 @@ class NetworkSession{
 		}
 	}
 
-	private function sendEncoded(string $payload, bool $immediate = false) : void{
+	/**
+	 * @param PromiseResolver[] $ackPromises
+	 * @phpstan-param list<PromiseResolver<true>> $ackPromises
+	 */
+	private function sendEncoded(string $payload, bool $immediate, array $ackPromises) : void{
 		if($this->cipher !== null){
 			Timings::$playerNetworkSendEncrypt->startTiming();
 			$payload = $this->cipher->encrypt($payload);
 			Timings::$playerNetworkSendEncrypt->stopTiming();
 		}
-		$this->sender->send($payload, $immediate);
+
+		if(count($ackPromises) > 0){
+			$ackReceiptId = $this->nextAckReceiptId++;
+			$this->ackPromisesByReceiptId[$ackReceiptId] = $ackPromises;
+		}else{
+			$ackReceiptId = null;
+		}
+		$this->sender->send($payload, $immediate, $ackReceiptId);
 	}
 
 	/**
@@ -646,6 +711,19 @@ class NetworkSession{
 			$this->setHandler(null);
 			$this->connected = false;
 
+			$ackPromisesByReceiptId = $this->ackPromisesByReceiptId;
+			$this->ackPromisesByReceiptId = [];
+			foreach($ackPromisesByReceiptId as $resolvers){
+				foreach($resolvers as $resolver){
+					$resolver->reject();
+				}
+			}
+			$sendBufferAckPromises = $this->sendBufferAckPromises;
+			$this->sendBufferAckPromises = [];
+			foreach($sendBufferAckPromises as $resolver){
+				$resolver->reject();
+			}
+
 			$this->logger->info($this->server->getLanguage()->translate(KnownTranslationFactory::pocketmine_network_session_close($reason)));
 		}
 	}

src/network/mcpe/PacketSender.php

@@ -28,7 +28,7 @@ interface PacketSender{
 	/**
 	 * Pushes a packet into the channel to be processed.
 	 */
-	public function send(string $payload, bool $immediate) : void;
+	public function send(string $payload, bool $immediate, ?int $receiptId) : void;
 
 	/**
 	 * Closes the channel, terminating the connection.

src/network/mcpe/handler/ResourcePacksPacketHandler.php

@@ -51,7 +51,13 @@ use function substr;
  * packs to the client.
  */
 class ResourcePacksPacketHandler extends PacketHandler{
-	private const PACK_CHUNK_SIZE = 128 * 1024; //128KB
+	private const PACK_CHUNK_SIZE = 256 * 1024; //256KB
+
+	/**
+	 * Larger values allow downloading more chunks at the same time, increasing download speed, but the client may choke
+	 * and cause the download speed to drop (due to ACKs taking too long to arrive).
+	 */
+	private const MAX_CONCURRENT_CHUNK_REQUESTS = 1;
 
 	/**
 	 * @var ResourcePack[]
@@ -62,6 +68,11 @@ class ResourcePacksPacketHandler extends PacketHandler{
 	/** @var bool[][] uuid => [chunk index => hasSent] */
 	private array $downloadedChunks = [];
 
+	/** @phpstan-var \SplQueue<array{ResourcePack, int}> */
+	private \SplQueue $requestQueue;
+
+	private int $activeRequests = 0;
+
 	/**
 	 * @param ResourcePack[] $resourcePackStack
 	 * @param string[]       $encryptionKeys    pack UUID => key, leave unset for any packs that are not encrypted
@@ -77,6 +88,7 @@ class ResourcePacksPacketHandler extends PacketHandler{
 		private bool $mustAccept,
 		private \Closure $completionCallback
 	){
+		$this->requestQueue = new \SplQueue();
 		foreach($resourcePackStack as $pack){
 			$this->resourcePacksById[$pack->getPackId()] = $pack;
 		}
@@ -197,8 +209,37 @@ class ResourcePacksPacketHandler extends PacketHandler{
 			$this->downloadedChunks[$packId][$packet->chunkIndex] = true;
 		}
 
-		$this->session->sendDataPacket(ResourcePackChunkDataPacket::create($packId, $packet->chunkIndex, $offset, $pack->getPackChunk($offset, self::PACK_CHUNK_SIZE)));
+		$this->requestQueue->enqueue([$pack, $packet->chunkIndex]);
+		$this->processChunkRequestQueue();
 
 		return true;
 	}
+
+	private function processChunkRequestQueue() : void{
+		if($this->activeRequests >= self::MAX_CONCURRENT_CHUNK_REQUESTS || $this->requestQueue->isEmpty()){
+			return;
+		}
+		/**
+		 * @var ResourcePack $pack
+		 * @var int          $chunkIndex
+		 */
+		[$pack, $chunkIndex] = $this->requestQueue->dequeue();
+
+		$packId = $pack->getPackId();
+		$offset = $chunkIndex * self::PACK_CHUNK_SIZE;
+		$chunkData = $pack->getPackChunk($offset, self::PACK_CHUNK_SIZE);
+		$this->activeRequests++;
+		$this->session
+			->sendDataPacketWithReceipt(ResourcePackChunkDataPacket::create($packId, $chunkIndex, $offset, $chunkData))
+			->onCompletion(
+				function() : void{
+					$this->activeRequests--;
+					$this->processChunkRequestQueue();
+				},
+				function() : void{
+					//this may have been rejected because of a disconnection - this will do nothing in that case
+					$this->disconnectWithError("Plugin interrupted sending of resource packs");
+				}
+			);
+	}
 }

src/network/mcpe/raklib/RakLibInterface.php

@@ -252,7 +252,9 @@ class RakLibInterface implements ServerEventListener, AdvancedNetworkInterface{
 	}
 
 	public function onPacketAck(int $sessionId, int $identifierACK) : void{
-
+		if(isset($this->sessions[$sessionId])){
+			$this->sessions[$sessionId]->handleAckReceipt($identifierACK);
+		}
 	}
 
 	public function setName(string $name) : void{
@@ -289,12 +291,13 @@ class RakLibInterface implements ServerEventListener, AdvancedNetworkInterface{
 		$this->network->getBandwidthTracker()->add($bytesSentDiff, $bytesReceivedDiff);
 	}
 
-	public function putPacket(int $sessionId, string $payload, bool $immediate = true) : void{
+	public function putPacket(int $sessionId, string $payload, bool $immediate = true, ?int $receiptId = null) : void{
 		if(isset($this->sessions[$sessionId])){
 			$pk = new EncapsulatedPacket();
 			$pk->buffer = self::MCPE_RAKNET_PACKET_ID . $payload;
 			$pk->reliability = PacketReliability::RELIABLE_ORDERED;
 			$pk->orderChannel = 0;
+			$pk->identifierACK = $receiptId;
 
 			$this->interface->sendEncapsulated($sessionId, $pk, $immediate);
 		}

src/network/mcpe/raklib/RakLibPacketSender.php

@@ -33,9 +33,9 @@ class RakLibPacketSender implements PacketSender{
 		private RakLibInterface $handler
 	){}
 
-	public function send(string $payload, bool $immediate) : void{
+	public function send(string $payload, bool $immediate, ?int $receiptId) : void{
 		if(!$this->closed){
-			$this->handler->putPacket($this->sessionId, $payload, $immediate);
+			$this->handler->putPacket($this->sessionId, $payload, $immediate, $receiptId);
 		}
 	}