add pmmp#5805

could be buggy according to pr but wtv

Author: Joshy3282

README.md

@@ -48,6 +48,7 @@ Additions
 > https://github.com/pmmp/PocketMine-MP/pull/5581
 > https://github.com/pmmp/PocketMine-MP/pull/5583
 > https://github.com/pmmp/PocketMine-MP/pull/5800
+> https://github.com/pmmp/PocketMine-MP/pull/5805
 
 
 ## What is this?

resources/pocketmine.yml

@@ -85,11 +85,10 @@ network:
   batch-threshold: 256
   #Compression level used when sending batched packets. Higher = more CPU, less bandwidth usage
   compression-level: 6
-  #Use AsyncTasks for compression during the main game session. Increases latency, but may reduce main thread load
-  async-compression: false
-  #Threshold for async compression, in bytes. Only packets larger than this will be compressed asynchronously
-  #Due to large overhead of AsyncTask, async compression isn't worth it except for large packets
-  async-compression-threshold: 10000
+  #Max threads to use for packet compression. If disabled, compression will be done on the main thread.
+  #Set to 0 to disable, or "auto" to try to detect the number of available CPU cores.
+  #Higher values will allow using more CPU cores, but will also increase memory usage.
+  compression-threads: auto
   #Experimental. Use UPnP to automatically port forward
   upnp-forwarding: false
   #Maximum size in bytes of packets sent over the network (default 1492 bytes). Packets larger than this will be

src/Server.php

@@ -52,6 +52,7 @@
 use pocketmine\network\mcpe\compression\CompressBatchPromise;
 use pocketmine\network\mcpe\compression\CompressBatchTask;
 use pocketmine\network\mcpe\compression\Compressor;
+use pocketmine\network\mcpe\compression\CompressorWorkerPool;
 use pocketmine\network\mcpe\compression\ZlibCompressor;
 use pocketmine\network\mcpe\convert\TypeConverter;
 use pocketmine\network\mcpe\encryption\EncryptionContext;
@@ -210,8 +211,6 @@ class Server{
 	private const TICKS_PER_TPS_OVERLOAD_WARNING = 5 * self::TARGET_TICKS_PER_SECOND;
 	private const TICKS_PER_STATS_REPORT = 300 * self::TARGET_TICKS_PER_SECOND;
 
-	private const DEFAULT_ASYNC_COMPRESSION_THRESHOLD = 10_000;
-
 	private static ?Server $instance = null;
 
 	private TimeTrackingSleeperHandler $tickSleeper;
@@ -269,8 +268,13 @@ class Server{
 	private bool $onlineMode = true;
 
 	private Network $network;
-	private bool $networkCompressionAsync = true;
-	private int $networkCompressionAsyncThreshold = self::DEFAULT_ASYNC_COMPRESSION_THRESHOLD;
+
+	private int $networkCompressionThreads;
+	/**
+	 * @var CompressorWorkerPool[]
+	 * @phpstan-var array<int, CompressorWorkerPool>
+	 */
+	private array $networkCompressionThreadPools = [];
 
 	private Language $language;
 	private bool $forceLanguage = false;
@@ -909,11 +913,13 @@ public function __construct(
 			}
 			ZlibCompressor::setInstance(new ZlibCompressor($netCompressionLevel, $netCompressionThreshold, ZlibCompressor::DEFAULT_MAX_DECOMPRESSION_SIZE));
 
-			$this->networkCompressionAsync = $this->configGroup->getPropertyBool(Yml::NETWORK_ASYNC_COMPRESSION, true);
-			$this->networkCompressionAsyncThreshold = max(
-				$this->configGroup->getPropertyInt(Yml::NETWORK_ASYNC_COMPRESSION_THRESHOLD, self::DEFAULT_ASYNC_COMPRESSION_THRESHOLD),
-				$netCompressionThreshold ?? self::DEFAULT_ASYNC_COMPRESSION_THRESHOLD
-			);
+			// todo maybe add toggle to use this or legacy network compression
+			$netCompressionThreads = $this->configGroup->getPropertyString(Yml::NETWORK_COMPRESSION_THREADS, "auto");
+			if($netCompressionThreads === "auto"){
+				$this->networkCompressionThreads = max(1, Utils::getCoreCount() - 2);
+			}else{
+				$this->networkCompressionThreads = max(0, (int) $netCompressionThreads);
+			}
 
 			EncryptionContext::$ENABLED = $this->configGroup->getPropertyBool(Yml::NETWORK_ENABLE_ENCRYPTION, true);
 
@@ -1011,7 +1017,7 @@ public function __construct(
 
 			$this->queryInfo = new QueryInfo($this);
 
-			$playerDataProviderType = $this->configGroup->getPropertyString("player.default-data-format", "datfile");
+			$playerDataProviderType = $this->configGroup->getPropertyString(Yml::PLAYER_DEFAULT_DATA_FORMAT, "datfile");
 			$path = Path::join($this->dataPath, "players");
 			if ($playerDataProviderType === "datfile") {
 				$this->playerDataProvider = new DatFilePlayerDataProvider($path);
@@ -1364,6 +1370,17 @@ public function broadcastTitle(string $title, string $subtitle = "", int $fadeIn
 		return count($recipients);
 	}
 
+	private function getNetworkCompressionWorkerPool(Compressor $compressor) : CompressorWorkerPool{
+		$compressorId = spl_object_id($compressor);
+		$workerPool = $this->networkCompressionThreadPools[$compressorId] ?? null;
+		if ($workerPool === null) {
+			$this->logger->debug("Creating new worker pool for compressor " . get_class($compressor) . "#" . $compressorId);
+			$workerPool = $this->networkCompressionThreadPools[$compressorId] = new CompressorWorkerPool($this->networkCompressionThreads, $compressor, $this->tickSleeper);
+		}
+
+		return $workerPool;
+	}
+
 	/**
 	 * @internal
 	 * Promises to compress the given batch buffer using the selected compressor, optionally on a separate thread.
@@ -1382,26 +1399,22 @@ public function prepareBatch(string $buffer, Compressor $compressor, ?bool $sync
 		try{
 			$timings->startTiming();
 
-			$threshold = $compressor->getCompressionThreshold();
-			if($threshold === null || strlen($buffer) < $compressor->getCompressionThreshold()){
-				$compressionType = CompressionAlgorithm::NONE;
-				$compressed = $buffer;
-
-			}else{
-				$sync ??= !$this->networkCompressionAsync;
+			if($sync === null){
+				$threshold = $compressor->getCompressionThreshold();
+				$sync = $threshold === null || strlen($buffer) < $threshold;
+			}
 
-				if(!$sync && strlen($buffer) >= $this->networkCompressionAsyncThreshold){
-					$promise = new CompressBatchPromise();
-					$task = new CompressBatchTask($buffer, $promise, $compressor);
-					$this->asyncPool->submitTask($task);
-					return $promise;
-				}
+			if(!$sync && $this->networkCompressionThreads > 0){
+				$workerPool = $this->getNetworkCompressionWorkerPool($compressor);
 
-				$compressionType = $compressor->getNetworkId();
-				$compressed = $compressor->compress($buffer);
+				//TODO: we really want to be submitting all sessions' buffers in one go to maximize performance
+				$promise = $workerPool->submit($buffer);
+			}else{
+				$promise = new CompressBatchPromise();
+				$promise->resolve($compressor->compress($buffer));
 			}
 
-			return chr($compressionType) . $compressed;
+			return $promise;
 		}finally{
 			$timings->stopTiming();
 		}
@@ -1518,6 +1531,10 @@ public function forceShutdown() : void{
 					$this->network->unregisterInterface($interface);
 				}
 			}
+			foreach($this->networkCompressionThreadPools as $pool){
+				$this->logger->debug("Shutting down network compression thread pool for compressor " . get_class($pool->getCompressor()) . "#" . spl_object_id($pool->getCompressor()));
+				$pool->shutdown();
+			}
 		}catch(\Throwable $e){
 			$this->logger->logException($e);
 			$this->logger->emergency("Crashed while crashing, killing process");

src/YmlServerProperties.php

@@ -94,12 +94,14 @@ private function __construct(){
 	public const NETWORK_ASYNC_COMPRESSION_THRESHOLD = 'network.async-compression-threshold';
 	public const NETWORK_BATCH_THRESHOLD = 'network.batch-threshold';
 	public const NETWORK_COMPRESSION_LEVEL = 'network.compression-level';
+	public const NETWORK_COMPRESSION_THREADS = 'network.compression-threads';
 	public const NETWORK_ENABLE_ENCRYPTION = 'network.enable-encryption';
 	public const NETWORK_MAX_MTU_SIZE = 'network.max-mtu-size';
 	public const NETWORK_UPNP_FORWARDING = 'network.upnp-forwarding';
 	public const PLAYER = 'player';
 	public const PLAYER_SAVE_PLAYER_DATA = 'player.save-player-data';
 	public const PLAYER_VERIFY_XUID = 'player.verify-xuid';
+	public const PLAYER_DEFAULT_DATA_FORMAT = 'player.default-data-format';
 	public const PLUGINS = 'plugins';
 	public const PLUGINS_LEGACY_DATA_DIR = 'plugins.legacy-data-dir';
 	public const SETTINGS = 'settings';