Ensure every packet is ran on the same thread per player (#2473)

This removes a lot of concurrency checking that needs to be done, because there should be no way two packets can be handled at the same time.

connector/pom.xml

@@ -13,8 +13,8 @@
     <properties>
         <adventure.version>4.8.0</adventure.version>
         <fastutil.version>8.5.2</fastutil.version>
-        <jackson.version>2.10.2</jackson.version>
+        <jackson.version>2.12.4</jackson.version>
-        <netty.version>4.1.59.Final</netty.version>
+        <netty.version>4.1.66.Final</netty.version>
     </properties>
 
     <dependencies>
@@ -167,7 +167,7 @@
         <dependency>
             <groupId>com.github.GeyserMC</groupId>
             <artifactId>PacketLib</artifactId>
-            <version>25eb4c4</version>
+            <version>86c9c38</version>
             <scope>compile</scope>
             <exclusions>
                 <exclusion>

connector/src/main/java/org/geysermc/connector/network/ConnectorServerEventHandler.java

@@ -30,7 +30,9 @@ import com.nukkitx.protocol.bedrock.BedrockServerEventHandler;
 import com.nukkitx.protocol.bedrock.BedrockServerSession;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.DefaultEventLoopGroup;
 import io.netty.channel.socket.DatagramPacket;
+import io.netty.util.concurrent.DefaultThreadFactory;
 import org.geysermc.connector.GeyserConnector;
 import org.geysermc.connector.common.ping.GeyserPingInfo;
 import org.geysermc.connector.configuration.GeyserConfiguration;
@@ -56,6 +58,7 @@ public class ConnectorServerEventHandler implements BedrockServerEventHandler {
     private static final int MAGIC_RAKNET_LENGTH = 338;
 
     private final GeyserConnector connector;
+    private final DefaultEventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(new DefaultThreadFactory("Geyser player thread"));
 
     public ConnectorServerEventHandler(GeyserConnector connector) {
         this.connector = connector;
@@ -162,7 +165,7 @@ public class ConnectorServerEventHandler implements BedrockServerEventHandler {
     public void onSessionCreation(BedrockServerSession bedrockServerSession) {
         bedrockServerSession.setLogging(true);
         bedrockServerSession.setCompressionLevel(connector.getConfig().getBedrock().getCompressionLevel());
-        bedrockServerSession.setPacketHandler(new UpstreamPacketHandler(connector, new GeyserSession(connector, bedrockServerSession)));
+        bedrockServerSession.setPacketHandler(new UpstreamPacketHandler(connector, new GeyserSession(connector, bedrockServerSession, eventLoopGroup.next())));
         // Set the packet codec to default just in case we need to send disconnect packets.
         bedrockServerSession.setPacketCodec(BedrockProtocol.DEFAULT_BEDROCK_CODEC);
     }

connector/src/main/java/org/geysermc/connector/network/UpstreamPacketHandler.java

@@ -52,6 +52,11 @@ public class UpstreamPacketHandler extends LoggingPacketHandler {
         return PacketTranslatorRegistry.BEDROCK_TRANSLATOR.translate(packet.getClass(), packet, session);
     }
 
+    @Override
+    boolean defaultHandler(BedrockPacket packet) {
+        return translateAndDefault(packet);
+    }
+
     @Override
     public boolean handle(LoginPacket loginPacket) {
         BedrockPacketCodec packetCodec = BedrockProtocol.getBedrockCodec(loginPacket.getProtocolVersion());
@@ -156,7 +161,7 @@ public class UpstreamPacketHandler extends LoggingPacketHandler {
 
     @Override
     public boolean handle(ModalFormResponsePacket packet) {
-        session.getFormCache().handleResponse(packet);
+        session.getEventLoop().execute(() -> session.getFormCache().handleResponse(packet));
         return true;
     }
 
@@ -209,11 +214,6 @@ public class UpstreamPacketHandler extends LoggingPacketHandler {
         return translateAndDefault(packet);
     }
 
-    @Override
-    boolean defaultHandler(BedrockPacket packet) {
-        return translateAndDefault(packet);
-    }
-
     @Override
     public boolean handle(ResourcePackChunkRequestPacket packet) {
         ResourcePackChunkDataPacket data = new ResourcePackChunkDataPacket();

connector/src/main/java/org/geysermc/connector/network/session/GeyserSession.java

@@ -57,9 +57,9 @@ import com.nukkitx.protocol.bedrock.data.command.CommandPermission;
 import com.nukkitx.protocol.bedrock.data.entity.EntityData;
 import com.nukkitx.protocol.bedrock.data.entity.EntityFlag;
 import com.nukkitx.protocol.bedrock.packet.*;
+import io.netty.channel.EventLoop;
 import it.unimi.dsi.fastutil.ints.*;
 import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
-import it.unimi.dsi.fastutil.longs.Long2ObjectMaps;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap;
 import it.unimi.dsi.fastutil.objects.ObjectIterator;
@@ -102,7 +102,10 @@ import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 @Getter
@@ -110,6 +113,10 @@ public class GeyserSession implements CommandSender {
 
     private final GeyserConnector connector;
     private final UpstreamSession upstream;
+    /**
+     * The loop where all packets and ticking is processed to prevent concurrency issues.
+     */
+    private final EventLoop eventLoop;
     private TcpClientSession downstream;
     @Setter
     private AuthData authData;
@@ -158,11 +165,6 @@ public class GeyserSession implements CommandSender {
     @Getter(AccessLevel.NONE)
     private final AtomicInteger itemNetId = new AtomicInteger(2);
 
-    @Getter(AccessLevel.NONE)
-    private final Object inventoryLock = new Object();
-    @Getter(AccessLevel.NONE)
-    private CompletableFuture<Void> inventoryFuture;
-
     @Setter
     private ScheduledFuture<?> craftingGridFuture;
 
@@ -183,8 +185,8 @@ public class GeyserSession implements CommandSender {
     @Setter
     private ItemMappings itemMappings;
 
-    private final Map<Vector3i, SkullPlayerEntity> skullCache = new ConcurrentHashMap<>();
+    private final Map<Vector3i, SkullPlayerEntity> skullCache = new Object2ObjectOpenHashMap<>();
-    private final Long2ObjectMap<ClientboundMapItemDataPacket> storedMaps = Long2ObjectMaps.synchronize(new Long2ObjectOpenHashMap<>());
+    private final Long2ObjectMap<ClientboundMapItemDataPacket> storedMaps = new Long2ObjectOpenHashMap<>();
 
     /**
      * Stores the map between Java and Bedrock biome network IDs.
@@ -426,9 +428,10 @@ public class GeyserSession implements CommandSender {
 
     private MinecraftProtocol protocol;
 
-    public GeyserSession(GeyserConnector connector, BedrockServerSession bedrockServerSession) {
+    public GeyserSession(GeyserConnector connector, BedrockServerSession bedrockServerSession, EventLoop eventLoop) {
         this.connector = connector;
         this.upstream = new UpstreamSession(bedrockServerSession);
+        this.eventLoop = eventLoop;
 
         this.advancementsCache = new AdvancementsCache(this);
         this.bookEditCache = new BookEditCache(this);
@@ -447,7 +450,6 @@ public class GeyserSession implements CommandSender {
 
         this.playerInventory = new PlayerInventory();
         this.openInventory = null;
-        this.inventoryFuture = CompletableFuture.completedFuture(null);
         this.craftingRecipes = new Int2ObjectOpenHashMap<>();
         this.unlockedRecipes = new ObjectOpenHashSet<>();
         this.lastRecipeNetId = new AtomicInteger(1);
@@ -664,7 +666,7 @@ public class GeyserSession implements CommandSender {
         boolean floodgate = this.remoteAuthType == AuthType.FLOODGATE;
 
         // Start ticking
-        tickThread = connector.getGeneralThreadPool().scheduleAtFixedRate(this::tick, 50, 50, TimeUnit.MILLISECONDS);
+        tickThread = eventLoop.scheduleAtFixedRate(this::tick, 50, 50, TimeUnit.MILLISECONDS);
 
         downstream = new TcpClientSession(this.remoteAddress, this.remotePort, protocol);
         disableSrvResolving();
@@ -1095,39 +1097,6 @@ public class GeyserSession implements CommandSender {
         upstream.sendPacket(startGamePacket);
     }
 
-    /**
-     * Adds a new inventory task.
-     * Inventory tasks are executed one at a time, in order.
-     *
-     * @param task the task to run
-     */
-    public void addInventoryTask(Runnable task) {
-        synchronized (inventoryLock) {
-            inventoryFuture = inventoryFuture.thenRun(task).exceptionally(throwable -> {
-                GeyserConnector.getInstance().getLogger().error("Error processing inventory task", throwable.getCause());
-                return null;
-            });
-        }
-    }
-
-    /**
-     * Adds a new inventory task with a delay.
-     * The delay is achieved by scheduling with the Geyser general thread pool.
-     * Inventory tasks are executed one at a time, in order.
-     *
-     * @param task the delayed task to run
-     * @param delayMillis delay in milliseconds
-     */
-    public void addInventoryTask(Runnable task, long delayMillis) {
-        synchronized (inventoryLock) {
-            Executor delayedExecutor = command -> GeyserConnector.getInstance().getGeneralThreadPool().schedule(command, delayMillis, TimeUnit.MILLISECONDS);
-            inventoryFuture = inventoryFuture.thenRunAsync(task, delayedExecutor).exceptionally(throwable -> {
-                GeyserConnector.getInstance().getLogger().error("Error processing inventory task", throwable.getCause());
-                return null;
-            });
-        }
-    }
-
     /**
      * @return the next Bedrock item network ID to use for a new item
      */
@@ -1229,7 +1198,18 @@ public class GeyserSession implements CommandSender {
      * @param packet the java edition packet from MCProtocolLib
      */
     public void sendDownstreamPacket(Packet packet) {
-        if (downstream != null && (protocol.getSubProtocol().equals(SubProtocol.GAME) || packet.getClass() == LoginPluginResponsePacket.class)) {
+        if (!closed && this.downstream != null) {
+            EventLoop eventLoop = this.downstream.getChannel().eventLoop();
+            if (eventLoop.inEventLoop()) {
+                sendDownstreamPacket0(packet);
+            } else {
+                eventLoop.execute(() -> sendDownstreamPacket0(packet));
+            }
+        }
+    }
+
+    private void sendDownstreamPacket0(Packet packet) {
+        if (protocol.getSubProtocol().equals(SubProtocol.GAME) || packet.getClass() == LoginPluginResponsePacket.class) {
             downstream.send(packet);
         } else {
             connector.getLogger().debug("Tried to send downstream packet " + packet.getClass().getSimpleName() + " before connected to the server");

connector/src/main/java/org/geysermc/connector/network/session/cache/EntityCache.java

@@ -26,6 +26,8 @@
 package org.geysermc.connector.network.session.cache;
 
 import it.unimi.dsi.fastutil.longs.*;
+import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap;
+import it.unimi.dsi.fastutil.objects.ObjectArrayList;
 import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
 import lombok.Getter;
 import org.geysermc.connector.entity.Tickable;
@@ -44,15 +46,15 @@ public class EntityCache {
     private final GeyserSession session;
 
     @Getter
-    private Long2ObjectMap<Entity> entities = Long2ObjectMaps.synchronize(new Long2ObjectOpenHashMap<>());
+    private final Long2ObjectMap<Entity> entities = new Long2ObjectOpenHashMap<>();
     /**
      * A list of all entities that must be ticked.
      */
-    private final List<Tickable> tickableEntities = Collections.synchronizedList(new ArrayList<>());
+    private final List<Tickable> tickableEntities = new ObjectArrayList<>();
-    private Long2LongMap entityIdTranslations = Long2LongMaps.synchronize(new Long2LongOpenHashMap());
+    private final Long2LongMap entityIdTranslations = new Long2LongOpenHashMap();
-    private Map<UUID, PlayerEntity> playerEntities = Collections.synchronizedMap(new HashMap<>());
+    private final Map<UUID, PlayerEntity> playerEntities = new Object2ObjectOpenHashMap<>();
-    private Map<UUID, BossBar> bossBars = Collections.synchronizedMap(new HashMap<>());
+    private final Map<UUID, BossBar> bossBars = new Object2ObjectOpenHashMap<>();
-    private final Long2LongMap cachedPlayerEntityLinks = Long2LongMaps.synchronize(new Long2LongOpenHashMap());
+    private final Long2LongMap cachedPlayerEntityLinks = new Long2LongOpenHashMap();
 
     @Getter
     private final AtomicLong nextEntityId = new AtomicLong(2L);
@@ -156,13 +158,6 @@ public class EntityCache {
         bossBars.values().forEach(BossBar::updateBossBar);
     }
 
-    public void clear() {
-        entities = null;
-        entityIdTranslations = null;
-        playerEntities = null;
-        bossBars = null;
-    }
-
     public long getCachedPlayerEntityLink(long playerId) {
         return cachedPlayerEntityLinks.remove(playerId);
     }

connector/src/main/java/org/geysermc/connector/network/translators/PacketTranslatorRegistry.java

@@ -29,6 +29,7 @@ import com.github.steveice10.mc.protocol.packet.ingame.server.ServerPlayerListDa
 import com.github.steveice10.mc.protocol.packet.ingame.server.world.ServerUpdateLightPacket;
 import com.github.steveice10.packetlib.packet.Packet;
 import com.nukkitx.protocol.bedrock.BedrockPacket;
+import io.netty.channel.EventLoop;
 import it.unimi.dsi.fastutil.objects.ObjectArrayList;
 import org.geysermc.common.PlatformType;
 import org.geysermc.connector.GeyserConnector;
@@ -89,7 +90,12 @@ public class PacketTranslatorRegistry<T> {
             try {
                 PacketTranslator<P> translator = (PacketTranslator<P>) translators.get(clazz);
                 if (translator != null) {
+                    EventLoop eventLoop = session.getEventLoop();
+                    if (eventLoop.inEventLoop()) {
                         translator.translate(packet, session);
+                    } else {
+                        eventLoop.execute(() -> translator.translate(packet, session));
+                    }
                     return true;
                 } else {
                     if ((GeyserConnector.getInstance().getPlatformType() != PlatformType.STANDALONE || !(packet instanceof BedrockPacket)) && !IGNORED_PACKETS.contains(clazz)) {

connector/src/main/java/org/geysermc/connector/network/translators/bedrock/BedrockAnimateTranslator.java

@@ -48,7 +48,7 @@ public class BedrockAnimateTranslator extends PacketTranslator<AnimatePacket> {
         switch (packet.getAction()) {
             case SWING_ARM:
                 // Delay so entity damage can be processed first
-                session.getConnector().getGeneralThreadPool().schedule(() ->
+                session.getEventLoop().schedule(() ->
                         session.sendDownstreamPacket(new ClientPlayerSwingArmPacket(Hand.MAIN_HAND)),
                         25,
                         TimeUnit.MILLISECONDS

connector/src/main/java/org/geysermc/connector/network/translators/bedrock/BedrockContainerCloseTranslator.java

@@ -39,7 +39,6 @@ public class BedrockContainerCloseTranslator extends PacketTranslator<ContainerC
 
     @Override
     public void translate(ContainerClosePacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         byte windowId = packet.getId();
 
         //Client wants close confirmation
@@ -62,6 +61,5 @@ public class BedrockContainerCloseTranslator extends PacketTranslator<ContainerC
                 openInventory.setPending(false);
             }
         }
-        });
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/bedrock/BedrockInventoryTransactionTranslator.java

@@ -83,7 +83,6 @@ public class BedrockInventoryTransactionTranslator extends PacketTranslator<Inve
                     InventoryActionData containerAction = packet.getActions().get(1);
                     if (worldAction.getSource().getType() == InventorySource.Type.WORLD_INTERACTION
                             && worldAction.getSource().getFlag() == InventorySource.Flag.DROP_ITEM) {
-                        session.addInventoryTask(() -> {
                         if (session.getPlayerInventory().getHeldItemSlot() != containerAction.getSlot() ||
                                 session.getPlayerInventory().getItemInHand().isEmpty()) {
                             return;
@@ -102,7 +101,6 @@ public class BedrockInventoryTransactionTranslator extends PacketTranslator<Inve
                         } else {
                             session.getPlayerInventory().getItemInHand().sub(1);
                         }
-                        });
                     }
                 }
                 break;
@@ -222,7 +220,7 @@ public class BedrockInventoryTransactionTranslator extends PacketTranslator<Inve
                                     if (session.isSneaking() || blockState != BlockRegistries.JAVA_IDENTIFIERS.get("minecraft:crafting_table")) {
                                         // Delay the interaction in case the client doesn't intend to actually use the bucket
                                         // See BedrockActionTranslator.java
-                                        session.setBucketScheduledFuture(session.getConnector().getGeneralThreadPool().schedule(() -> {
+                                        session.setBucketScheduledFuture(session.getEventLoop().schedule(() -> {
                                             ClientPlayerUseItemPacket itemPacket = new ClientPlayerUseItemPacket(Hand.MAIN_HAND);
                                             session.sendDownstreamPacket(itemPacket);
                                         }, 5, TimeUnit.MILLISECONDS));

connector/src/main/java/org/geysermc/connector/network/translators/bedrock/BedrockItemStackRequestTranslator.java

@@ -45,6 +45,6 @@ public class BedrockItemStackRequestTranslator extends PacketTranslator<ItemStac
             return;
 
         InventoryTranslator translator = session.getInventoryTranslator();
-        session.addInventoryTask(() -> translator.translateRequests(session, inventory, packet.getRequests()));
+        translator.translateRequests(session, inventory, packet.getRequests());
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/bedrock/entity/BedrockEntityEventTranslator.java

@@ -37,6 +37,8 @@ import org.geysermc.connector.network.session.GeyserSession;
 import org.geysermc.connector.network.translators.PacketTranslator;
 import org.geysermc.connector.network.translators.Translator;
 
+import java.util.concurrent.TimeUnit;
+
 @Translator(packet = EntityEventPacket.class)
 public class BedrockEntityEventTranslator extends PacketTranslator<EntityEventPacket> {
 
@@ -48,12 +50,10 @@ public class BedrockEntityEventTranslator extends PacketTranslator<EntityEventPa
                 session.sendUpstreamPacket(packet);
                 return;
             case COMPLETE_TRADE:
-                session.addInventoryTask(() -> {
                 ClientSelectTradePacket selectTradePacket = new ClientSelectTradePacket(packet.getData());
                 session.sendDownstreamPacket(selectTradePacket);
-                });
 
-                session.addInventoryTask(() -> {
+                session.getEventLoop().schedule(() -> {
                     Entity villager = session.getPlayerEntity();
                     Inventory openInventory = session.getOpenInventory();
                     if (openInventory instanceof MerchantContainer) {
@@ -66,7 +66,7 @@ public class BedrockEntityEventTranslator extends PacketTranslator<EntityEventPa
                             villager.updateBedrockMetadata(session);
                         }
                     }
-                }, 100);
+                }, 100, TimeUnit.MILLISECONDS);
                 return;
         }
         session.getConnector().getLogger().debug("Did not translate incoming EntityEventPacket: " + packet.toString());

connector/src/main/java/org/geysermc/connector/network/translators/java/JavaRespawnTranslator.java

@@ -49,11 +49,9 @@ public class JavaRespawnTranslator extends PacketTranslator<ServerRespawnPacket>
         entity.setHealth(entity.getMaxHealth());
         entity.getAttributes().put(GeyserAttributeType.HEALTH, entity.createHealthAttribute());
 
-        session.addInventoryTask(() -> {
         session.setInventoryTranslator(InventoryTranslator.PLAYER_INVENTORY_TRANSLATOR);
         session.setOpenInventory(null);
         session.setClosingInventory(false);
-        });
 
         SetPlayerGameTypePacket playerGameTypePacket = new SetPlayerGameTypePacket();
         playerGameTypePacket.setGamemode(packet.getGamemode().ordinal());

connector/src/main/java/org/geysermc/connector/network/translators/java/entity/player/JavaPlayerChangeHeldItemTranslator.java

@@ -36,7 +36,6 @@ public class JavaPlayerChangeHeldItemTranslator extends PacketTranslator<ServerP
 
     @Override
     public void translate(ServerPlayerChangeHeldItemPacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         PlayerHotbarPacket hotbarPacket = new PlayerHotbarPacket();
         hotbarPacket.setContainerId(0);
         hotbarPacket.setSelectedHotbarSlot(packet.getSlot());
@@ -44,6 +43,5 @@ public class JavaPlayerChangeHeldItemTranslator extends PacketTranslator<ServerP
         session.sendUpstreamPacket(hotbarPacket);
 
         session.getPlayerInventory().setHeldItemSlot(packet.getSlot());
-        });
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/java/window/JavaCloseWindowTranslator.java

@@ -36,9 +36,8 @@ public class JavaCloseWindowTranslator extends PacketTranslator<ServerCloseWindo
 
     @Override
     public void translate(ServerCloseWindowPacket packet, GeyserSession session) {
-        session.addInventoryTask(() ->
         // Sometimes the server can request a window close of ID 0... when the window isn't even open
         // Don't confirm in this instance
-                InventoryUtils.closeInventory(session, packet.getWindowId(), (session.getOpenInventory() != null && session.getOpenInventory().getId() == packet.getWindowId())));
+        InventoryUtils.closeInventory(session, packet.getWindowId(), (session.getOpenInventory() != null && session.getOpenInventory().getId() == packet.getWindowId()));
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/java/window/JavaOpenWindowTranslator.java

@@ -31,17 +31,16 @@ import org.geysermc.connector.inventory.Inventory;
 import org.geysermc.connector.network.session.GeyserSession;
 import org.geysermc.connector.network.translators.PacketTranslator;
 import org.geysermc.connector.network.translators.Translator;
+import org.geysermc.connector.network.translators.chat.MessageTranslator;
 import org.geysermc.connector.network.translators.inventory.InventoryTranslator;
 import org.geysermc.connector.utils.InventoryUtils;
 import org.geysermc.connector.utils.LocaleUtils;
-import org.geysermc.connector.network.translators.chat.MessageTranslator;
 
 @Translator(packet = ServerOpenWindowPacket.class)
 public class JavaOpenWindowTranslator extends PacketTranslator<ServerOpenWindowPacket> {
 
     @Override
     public void translate(ServerOpenWindowPacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         if (packet.getWindowId() == 0) {
             return;
         }
@@ -73,6 +72,5 @@ public class JavaOpenWindowTranslator extends PacketTranslator<ServerOpenWindowP
 
         session.setInventoryTranslator(newTranslator);
         InventoryUtils.openInventory(session, newInventory);
-        });
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/java/window/JavaSetSlotTranslator.java

@@ -59,7 +59,6 @@ public class JavaSetSlotTranslator extends PacketTranslator<ServerSetSlotPacket>
 
     @Override
     public void translate(ServerSetSlotPacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         if (packet.getWindowId() == 255) { //cursor
             GeyserItemStack newItem = GeyserItemStack.from(packet.getItem());
             session.getPlayerInventory().setCursor(newItem, session);
@@ -79,7 +78,7 @@ public class JavaSetSlotTranslator extends PacketTranslator<ServerSetSlotPacket>
             if (session.getCraftingGridFuture() != null) {
                 session.getCraftingGridFuture().cancel(false);
             }
-                session.setCraftingGridFuture(session.getConnector().getGeneralThreadPool().schedule(() -> session.addInventoryTask(() -> updateCraftingGrid(session, packet, inventory, translator)), 150, TimeUnit.MILLISECONDS));
+            session.setCraftingGridFuture(session.getEventLoop().schedule(() -> updateCraftingGrid(session, packet, inventory, translator), 150, TimeUnit.MILLISECONDS));
 
             GeyserItemStack newItem = GeyserItemStack.from(packet.getItem());
             if (packet.getWindowId() == 0 && !(translator instanceof PlayerInventoryTranslator)) {
@@ -91,7 +90,6 @@ public class JavaSetSlotTranslator extends PacketTranslator<ServerSetSlotPacket>
                 translator.updateSlot(session, inventory, packet.getSlot());
             }
         }
-        });
     }
 
     private static void updateCraftingGrid(GeyserSession session, ServerSetSlotPacket packet, Inventory inventory, InventoryTranslator translator) {

connector/src/main/java/org/geysermc/connector/network/translators/java/window/JavaWindowItemsTranslator.java

@@ -39,7 +39,6 @@ public class JavaWindowItemsTranslator extends PacketTranslator<ServerWindowItem
 
     @Override
     public void translate(ServerWindowItemsPacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         Inventory inventory = InventoryUtils.getInventory(session, packet.getWindowId());
         if (inventory == null)
             return;
@@ -58,6 +57,5 @@ public class JavaWindowItemsTranslator extends PacketTranslator<ServerWindowItem
 
         session.getPlayerInventory().setCursor(GeyserItemStack.from(packet.getCarriedItem()), session);
         InventoryUtils.updateCursor(session);
-        });
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/java/window/JavaWindowPropertyTranslator.java

@@ -38,7 +38,6 @@ public class JavaWindowPropertyTranslator extends PacketTranslator<ServerWindowP
 
     @Override
     public void translate(ServerWindowPropertyPacket packet, GeyserSession session) {
-        session.addInventoryTask(() -> {
         Inventory inventory = InventoryUtils.getInventory(session, packet.getWindowId());
         if (inventory == null)
             return;
@@ -47,6 +46,5 @@ public class JavaWindowPropertyTranslator extends PacketTranslator<ServerWindowP
         if (translator != null) {
             translator.updateProperty(session, inventory, packet.getRawProperty(), packet.getValue());
         }
-        });
     }
 }

connector/src/main/java/org/geysermc/connector/network/translators/world/block/entity/SkullBlockEntityTranslator.java

@@ -147,7 +147,7 @@ public class SkullBlockEntityTranslator extends BlockEntityTranslator implements
             if (session.getUpstream().isInitialized()) {
                 player.spawnEntity(session);
 
-                SkullSkinManager.requestAndHandleSkin(player, session, (skin -> session.getConnector().getGeneralThreadPool().schedule(() -> {
+                SkullSkinManager.requestAndHandleSkin(player, session, (skin -> session.getEventLoop().schedule(() -> {
                     // Delay to minimize split-second "player" pop-in
                     player.getMetadata().getFlags().setFlag(EntityFlag.INVISIBLE, false);
                     player.updateBedrockMetadata(session);

connector/src/main/java/org/geysermc/connector/utils/InventoryUtils.java

@@ -37,7 +37,6 @@ import com.nukkitx.protocol.bedrock.data.inventory.ContainerId;
 import com.nukkitx.protocol.bedrock.data.inventory.ItemData;
 import com.nukkitx.protocol.bedrock.packet.InventorySlotPacket;
 import com.nukkitx.protocol.bedrock.packet.PlayerHotbarPacket;
-import org.geysermc.connector.GeyserConnector;
 import org.geysermc.connector.common.ChatColor;
 import org.geysermc.connector.inventory.Container;
 import org.geysermc.connector.inventory.GeyserItemStack;
@@ -80,8 +79,7 @@ public class InventoryUtils {
         if (translator != null) {
             translator.prepareInventory(session, inventory);
             if (translator instanceof DoubleChestInventoryTranslator && !((Container) inventory).isUsingRealBlock()) {
-                GeyserConnector.getInstance().getGeneralThreadPool().schedule(() ->
+                session.getEventLoop().schedule(() -> {
-                    session.addInventoryTask(() -> {
                     Inventory openInv = session.getOpenInventory();
                     if (openInv != null && openInv.getId() == inventory.getId()) {
                         translator.openInventory(session, inventory);
@@ -90,7 +88,7 @@ public class InventoryUtils {
                         // Presumably, this inventory is no longer relevant, and the client doesn't care about it
                         displayInventory(session, openInv);
                     }
-                }), 200, TimeUnit.MILLISECONDS);
+                }, 200, TimeUnit.MILLISECONDS);
             } else {
                 translator.openInventory(session, inventory);
                 translator.updateInventory(session, inventory);