Leaf/leaf-server/minecraft-patches/features/0125-Rewrite-queue-on-Connection.flushQueue.patch

From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: Taiyou06 <kaandindar21@gmail.com>
Date: Fri, 21 Feb 2025 15:52:42 +0100
Subject: [PATCH] Rewrite queue on Connection.flushQueue


diff --git a/net/minecraft/network/Connection.java b/net/minecraft/network/Connection.java
index 7b78c0af4a83bd39a5bc2d6554cc677bd4c0c822..eb9d99d657ae01c36af7cf4a75e206b917a14c39 100644
--- a/net/minecraft/network/Connection.java
+++ b/net/minecraft/network/Connection.java
@@ -35,6 +35,7 @@ import io.netty.handler.timeout.TimeoutException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
+import java.util.ArrayDeque;
 import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.RejectedExecutionException;
@@ -85,7 +86,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
     private static final ProtocolInfo<ServerHandshakePacketListener> INITIAL_PROTOCOL = HandshakeProtocols.SERVERBOUND;
     private final PacketFlow receiving;
     private volatile boolean sendLoginDisconnect = true;
-    private final Queue<WrappedConsumer> pendingActions = Queues.newConcurrentLinkedQueue(); // Paper - Optimize network
+    private final Queue<WrappedConsumer> pendingActions = new ArrayDeque<>(); // Paper - Optimize network // Leaf - Rewrite queue on Connection.flushQueue
     public Channel channel;
     public SocketAddress address;
     // Spigot start
@@ -541,10 +542,11 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
         if (io.papermc.paper.util.MCUtil.isMainThread()) {
             return this.processQueue();
         } else if (this.isPending) {
-            // Should only happen during login/status stages
-            synchronized (this.pendingActions) {
-                return this.processQueue();
-            }
+            // Leaf start - Rewrite queue on Connection.flushQueue
+            // Submit to the event loop to ensure thread confinement
+            this.channel.eventLoop().execute(this::processQueue);
+            return false;
+            // Leaf end - Rewrite queue on Connection.flushQueue
         }
         return false;
     }
@@ -554,29 +556,19 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
             return true;
         }
 
-        // If we are on main, we are safe here in that nothing else should be processing queue off main anymore
-        // But if we are not on main due to login/status, the parent is synchronized on packetQueue
-        final java.util.Iterator<WrappedConsumer> iterator = this.pendingActions.iterator();
-        while (iterator.hasNext()) {
-            final WrappedConsumer queued = iterator.next(); // poll -> peek
-
-            // Fix NPE (Spigot bug caused by handleDisconnection())
-            if (queued == null) {
-                return true;
-            }
-
-            if (queued.isConsumed()) {
-                continue;
-            }
-
+        // Leaf start - Rewrite queue on Connection.flushQueue
+        WrappedConsumer queued;
+        while ((queued = this.pendingActions.poll()) != null) {
             if (queued instanceof PacketSendAction packetSendAction) {
                 final Packet<?> packet = packetSendAction.packet;
                 if (!packet.isReady()) {
+                    // Re-add to the front and exit
+                    this.pendingActions.add(queued);
                     return false;
                 }
             }
 
-            iterator.remove();
+            // Leaf end - Rewrite queue on Connection.flushQueue
             if (queued.tryMarkConsumed()) {
                 queued.accept(this);
             }