Reapply "Run plugin messages on a DH thread"

This reverts commit ff3145336d.

core/src/main/java/com/seibel/distanthorizons/core/api/internal/ClientApi.java

@@ -33,6 +33,8 @@ import com.seibel.distanthorizons.core.render.DhApiRenderProxy;
 import com.seibel.distanthorizons.core.render.renderer.*;
 import com.seibel.distanthorizons.core.util.TimerUtil;
 import com.seibel.distanthorizons.core.util.objects.Pair;
+import com.seibel.distanthorizons.core.util.threading.PriorityTaskPicker;
+import com.seibel.distanthorizons.core.util.threading.ThreadPoolUtil;
 import com.seibel.distanthorizons.core.wrapperInterfaces.minecraft.IMinecraftRenderWrapper;
 import com.seibel.distanthorizons.coreapi.DependencyInjection.ApiEventInjector;
 import com.seibel.distanthorizons.core.config.Config;
@@ -58,6 +60,8 @@ import org.lwjgl.glfw.GLFW;
 import java.io.File;
 import java.util.*;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
 
 /**
  * This holds the methods that should be called
@@ -329,10 +333,27 @@ public class ClientApi
 	 */
 	public void pluginMessageReceived(@NotNull AbstractNetworkMessage message)
 	{
-		NetworkSession networkSession = this.pluginChannelApi.networkSession;
-		if (networkSession != null)
+		@Nullable ThreadPoolExecutor executor = ThreadPoolUtil.networkClientHandlerExecutor();
+		if (executor == null)
 		{
-			networkSession.tryHandleMessage(message);
+			LOGGER.warn("warn");
+			return;
+		}
+		
+		try
+		{
+			executor.execute(() ->
+			{
+				NetworkSession networkSession = this.pluginChannelApi.networkSession;
+				if (networkSession != null)
+				{
+					networkSession.tryHandleMessage(message);
+				}
+			});
+		}
+		catch (RejectedExecutionException e)
+		{
+			LOGGER.warn("Plugin message executor rejected");
 		}
 	}
 	

core/src/main/java/com/seibel/distanthorizons/core/multiplayer/client/AbstractFullDataNetworkRequestQueue.java

@@ -24,6 +24,7 @@ import com.seibel.distanthorizons.core.render.renderer.IDebugRenderable;
 import com.seibel.distanthorizons.core.sql.dto.FullDataSourceV2DTO;
 import com.seibel.distanthorizons.core.util.LodUtil;
 import com.seibel.distanthorizons.core.util.ratelimiting.SupplierBasedRateLimiter;
+import com.seibel.distanthorizons.core.util.threading.ThreadPoolUtil;
 import com.seibel.distanthorizons.core.world.DhApiWorldProxy;
 import com.seibel.distanthorizons.core.wrapperInterfaces.minecraft.IMinecraftClientWrapper;
 
@@ -217,11 +218,18 @@ public abstract class AbstractFullDataNetworkRequestQueue implements IDebugRende
 				FullDataSourceResponseMessage.class
 		);
 		requestTask.networkDataSourceFuture = dataSourceNetworkFuture;
-		dataSourceNetworkFuture.handle((FullDataSourceResponseMessage response, Throwable throwable) ->
+		
+		Executor networkCompressionExecutor = ThreadPoolUtil.getNetworkCompressionExecutor();
+		if (networkCompressionExecutor == null)
+		{
+			return;
+		}
+		
+		dataSourceNetworkFuture.handleAsync((FullDataSourceResponseMessage response, Throwable throwable) ->
 		{
 			this.handleNetResponse(requestTask, response, throwable);
 			return null;
-		});
+		}, networkCompressionExecutor);
 	}
 	private void handleNetResponse(NetRequestTask requestTask, FullDataSourceResponseMessage response, Throwable throwable)
 	{

core/src/main/java/com/seibel/distanthorizons/core/util/threading/ThreadPoolUtil.java

@@ -69,10 +69,14 @@ public class ThreadPoolUtil
 	@Nullable
 	public static ThreadPoolExecutor getBeaconCullingExecutor() { return beaconCullingThreadPool; }
 	
+	// The main distinction between these thread pools is that one for compression has multiple threads and client handler is single-threaded
 	private static PriorityTaskPicker.Executor networkCompressionThreadPool;
 	@Nullable
 	public static PriorityTaskPicker.Executor getNetworkCompressionExecutor() { return networkCompressionThreadPool; }
 	
+	private static ThreadPoolExecutor networkClientHandlerThreadPool;
+	@Nullable
+	public static ThreadPoolExecutor networkClientHandlerExecutor() { return networkClientHandlerThreadPool; }
 	
 	
 	public static final String FULL_DATA_MIGRATION_THREAD_NAME = "Full Data Migration";
@@ -103,7 +107,8 @@ public class ThreadPoolUtil
 		}
 		taskPicker = new PriorityTaskPicker();
 		
-		networkCompressionThreadPool = taskPicker.createExecutor("Network");
+		networkCompressionThreadPool = taskPicker.createExecutor("Network Compression");
+		networkClientHandlerThreadPool = ThreadUtil.makeSingleThreadPool("Network Client Handler");
 		fileHandlerThreadPool = taskPicker.createExecutor("IO");
 		renderSectionLoadThreadPool = taskPicker.createExecutor("Render Loader");
 		chunkToLodBuilderThreadPool = taskPicker.createExecutor("LOD Builder");
@@ -133,6 +138,7 @@ public class ThreadPoolUtil
 	public static void shutdownThreadPools()
 	{
 		// standalone threads
+		networkClientHandlerThreadPool.shutdownNow();
 		taskPicker.shutdownNow();
 		beaconCullingThreadPool.shutdown();
 		fullDataMigrationThreadPool.shutdown();