0010-Optimize-Network-Manager-and-add-advanced-packet-sup.patch

From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: hpfxd <me@hpfxd.com>
Date: Fri, 29 Oct 2021 06:38:18 -0400
Subject: [PATCH] Optimize Network Manager and add advanced packet support

Adds ability for 1 packet to bundle other packets to follow it
Adds ability for a packet to delay sending more packets until a state is ready.

Removes synchronization from sending packets
Removes processing packet queue off of main thread
  - for the few cases where it is allowed, order is not necessary nor
    should it even be happening concurrently in first place (handshaking/login/status)

Ensures packets sent asynchronously are dispatched on main thread

This helps ensure safety for ProtocolLib as packet listeners
are commonly accessing world state. This will allow you to schedule
a packet to be sent async, but itll be dispatched sync for packet
listeners to process.

This should solve some deadlock risks

Also adds Netty Channel Flush Consolidation to reduce the amount of flushing

Also avoids spamming closed channel exception by rechecking closed state in dispatch
and then catch exceptions and close if they fire.

diff --git a/src/main/java/net/minecraft/server/NetworkManager.java b/src/main/java/net/minecraft/server/NetworkManager.java
index fcf54e3a138f8408d7b85f087be03ca8cedb45f9..067793583980d102f8310f6197a52d2f126d15d4 100644
--- a/src/main/java/net/minecraft/server/NetworkManager.java
+++ b/src/main/java/net/minecraft/server/NetworkManager.java
@@ -62,7 +62,7 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
     };
     private final EnumProtocolDirection h;
     private final Queue<NetworkManager.QueuedPacket> i = Queues.newConcurrentLinkedQueue();
-    private final ReentrantReadWriteLock j = new ReentrantReadWriteLock();
+    //private final ReentrantReadWriteLock j = new ReentrantReadWriteLock(); // PandaSpigot - Remove packet queue locking
     public Channel channel;
     // Spigot Start // PAIL
     public SocketAddress l;
@@ -74,6 +74,11 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
     private IChatBaseComponent n;
     private boolean o;
     private boolean p;
+    // PandaSpigot start - Optimize network
+    public boolean isPending = true;
+    public boolean queueImmunity = false;
+    public EnumProtocol protocol;
+    // PandaSpigot end
 
     public NetworkManager(EnumProtocolDirection enumprotocoldirection) {
         this.h = enumprotocoldirection;
@@ -96,6 +101,7 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
     }
 
     public void a(EnumProtocol enumprotocol) {
+        this.protocol = enumprotocol; // PandaSpigot
         this.channel.attr(NetworkManager.c).set(enumprotocol);
         this.channel.config().setAutoRead(true);
         NetworkManager.g.debug("Enabled auto read");
@@ -135,38 +141,87 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
         this.m = packetlistener;
     }
 
-    public void handle(Packet packet) {
-        if (this.g()) {
-            this.m();
-            this.a(packet, (GenericFutureListener[]) null);
+    // PandaSpigot start
+    public EntityPlayer getPlayer() {
+        if (this.m instanceof PlayerConnection) {
+            return ((PlayerConnection) this.m).player;
         } else {
-            this.j.writeLock().lock();
-
-            try {
-                this.i.add(new NetworkManager.QueuedPacket(packet, (GenericFutureListener[]) null));
-            } finally {
-                this.j.writeLock().unlock();
+            return null;
+        }
+    }
+    private static class InnerUtil { // Attempt to hide these methods from ProtocolLib so it doesn't accidently pick them up.
+        private static java.util.List<Packet> buildExtraPackets(Packet packet) {
+            java.util.List<Packet> extra = packet.getExtraPackets();
+            if (extra == null || extra.isEmpty()) {
+               return null;
             }
+            java.util.List<Packet> ret = new java.util.ArrayList<>(1 + extra.size());
+            buildExtraPackets0(extra, ret);
+            return ret;
         }
 
+        private static void buildExtraPackets0(java.util.List<Packet> extraPackets, java.util.List<Packet> into) {
+            for (Packet extra : extraPackets) {
+                into.add(extra);
+                java.util.List<Packet> extraExtra = extra.getExtraPackets();
+                if (extraExtra != null && !extraExtra.isEmpty()) {
+                    buildExtraPackets0(extraExtra, into);
+                }
+            }
+        }
+        private static boolean canSendImmediate(NetworkManager networkManager, Packet<?> packet) {
+            return networkManager.isPending || networkManager.protocol != EnumProtocol.PLAY ||
+                    packet instanceof PacketPlayOutKeepAlive ||
+                    packet instanceof PacketPlayOutChat ||
+                    packet instanceof PacketPlayOutTabComplete ||
+                    packet instanceof PacketPlayOutTitle;
+        }
     }
+    // PandaSpigot end
 
+    public void handle(Packet packet) {
+        this.a(packet, null, (GenericFutureListener<? extends Future<? super Void>>) null); // PandaSpigot
+    }
     public void a(Packet packet, GenericFutureListener<? extends Future<? super Void>> genericfuturelistener, GenericFutureListener<? extends Future<? super Void>>... agenericfuturelistener) {
-        if (this.g()) {
-            this.m();
-            this.a(packet, (GenericFutureListener[]) ArrayUtils.add(agenericfuturelistener, 0, genericfuturelistener));
+        // PandaSpigot start - handle oversized packets better
+        GenericFutureListener<? extends Future<? super Void>>[] listeners = null;
+        if (genericfuturelistener != null || agenericfuturelistener != null) { // cannot call ArrayUtils.add with both null arguments
+            listeners = ArrayUtils.add(agenericfuturelistener, 0, genericfuturelistener);
+        }
+        boolean connected = this.isConnected();
+        if (!connected && !preparing) {
+            return; // Do nothing
+        }
+        packet.onPacketDispatch(getPlayer());
+        if (connected && (InnerUtil.canSendImmediate(this, packet) || (
+            MinecraftServer.getServer().isMainThread() && packet.isReady() && this.i.isEmpty() &&
+                        (packet.getExtraPackets() == null || packet.getExtraPackets().isEmpty())
+        ))) {
+            this.dispatchPacket(packet, listeners);
+            return;
+        }
+        // write the packets to the queue, then flush - antixray hooks there already
+        java.util.List<Packet> extraPackets = InnerUtil.buildExtraPackets(packet);
+        boolean hasExtraPackets = extraPackets != null && !extraPackets.isEmpty();
+        if (!hasExtraPackets) {
+            this.i.add(new NetworkManager.QueuedPacket(packet, listeners));
         } else {
-            this.j.writeLock().lock();
+            java.util.List<NetworkManager.QueuedPacket> packets = new java.util.ArrayList<>(1 + extraPackets.size());
+            packets.add(new NetworkManager.QueuedPacket(packet, (GenericFutureListener<? extends Future<? super Void>>) null)); // delay the future listener until the end of the extra packets
 
-            try {
-                this.i.add(new NetworkManager.QueuedPacket(packet, (GenericFutureListener[]) ArrayUtils.add(agenericfuturelistener, 0, genericfuturelistener)));
-            } finally {
-                this.j.writeLock().unlock();
+            for (int i = 0, len = extraPackets.size(); i < len;) {
+                Packet extra = extraPackets.get(i);
+                boolean end = ++i == len;
+                packets.add(new NetworkManager.QueuedPacket(extra, end ? listeners : null)); // append listener to the end
             }
+            this.i.addAll(packets);
         }
 
+        this.sendPacketQueue();
+        // PandaSpigot end
     }
 
+    private void dispatchPacket(final Packet packet, final GenericFutureListener<? extends Future<? super Void>>[] agenericfuturelistener) { this.a(packet, agenericfuturelistener); } // PandaSpigot - OBFHELPER
     private void a(final Packet packet, final GenericFutureListener<? extends Future<? super Void>>[] agenericfuturelistener) {
         final EnumProtocol enumprotocol = EnumProtocol.a(packet);
         final EnumProtocol enumprotocol1 = (EnumProtocol) this.channel.attr(NetworkManager.c).get();
@@ -176,18 +231,38 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
             this.channel.config().setAutoRead(false);
         }
 
+        EntityPlayer player = getPlayer(); // PandaSpigot
         if (this.channel.eventLoop().inEventLoop()) {
             if (enumprotocol != enumprotocol1) {
                 this.a(enumprotocol);
             }
 
+            // PandaSpigot start
+            if (!isConnected()) {
+                packet.onPacketDispatchFinish(player, null);
+                return;
+            }
+            try {
+            // PandaSpigot end
             ChannelFuture channelfuture = this.channel.writeAndFlush(packet);
 
             if (agenericfuturelistener != null) {
                 channelfuture.addListeners(agenericfuturelistener);
             }
 
+            // PandaSpigot start
+            if (packet.hasFinishListener()) {
+                channelfuture.addListener((ChannelFutureListener) channelFuture -> packet.onPacketDispatchFinish(player, channelFuture));
+            }
+            // PandaSpigot end
             channelfuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+            // PandaSpigot start
+            } catch (Exception e) {
+                g.error("NetworkException: " + player, e);
+                close(new ChatMessage("disconnect.genericReason", "Internal Exception: " + e.getMessage()));
+                packet.onPacketDispatchFinish(player, null);
+            }
+            // PandaSpigot end
         } else {
             this.channel.eventLoop().execute(new Runnable() {
                 public void run() {
@@ -195,35 +270,77 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
                         NetworkManager.this.a(enumprotocol);
                     }
 
+                // PandaSpigot start
+                if (!isConnected()) {
+                    packet.onPacketDispatchFinish(player, null);
+                    return;
+                }
+                try {
+                // PandaSpigot end
                     ChannelFuture channelfuture = NetworkManager.this.channel.writeAndFlush(packet);
 
                     if (agenericfuturelistener != null) {
                         channelfuture.addListeners(agenericfuturelistener);
                     }
 
+                // PandaSpigot start
+                if (packet.hasFinishListener()) {
+                    channelfuture.addListener((ChannelFutureListener) channelFuture -> packet.onPacketDispatchFinish(player, channelFuture));
+                }
+                // PandaSpigot end
                     channelfuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+                // PandaSpigot start
+                } catch (Exception e) {
+                    g.error("NetworkException: " + player, e);
+                    close(new ChatMessage("disconnect.genericReason", "Internal Exception: " + e.getMessage()));
+                    packet.onPacketDispatchFinish(player, null);
+                }
+                // PandaSpigot end
                 }
             });
         }
 
     }
 
-    private void m() {
-        if (this.channel != null && this.channel.isOpen()) {
-            this.j.readLock().lock();
-
-            try {
-                while (!this.i.isEmpty()) {
-                    NetworkManager.QueuedPacket networkmanager_queuedpacket = (NetworkManager.QueuedPacket) this.i.poll();
-
-                    this.a(networkmanager_queuedpacket.a, networkmanager_queuedpacket.b);
-                }
-            } finally {
-                this.j.readLock().unlock();
+    // PandaSpigot start - rewrite this to be safer if ran off main thread
+    private boolean sendPacketQueue() { return this.m(); } // OBFHELPER // void -> boolean
+    private boolean m() { // void -> boolean
+        if (!isConnected()) {
+            return true;
+        }
+        if (MinecraftServer.getServer().isMainThread()) {
+            return processQueue();
+        } else if (isPending) {
+            // Should only happen during login/status stages
+            synchronized (this.i) {
+                return this.processQueue();
+            }
+        }
+        return false;
+    }
+    private boolean processQueue() {
+        if (this.i.isEmpty()) 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
+        java.util.Iterator<QueuedPacket> iterator = this.i.iterator();
+        while (iterator.hasNext()) {
+            NetworkManager.QueuedPacket queued = iterator.next(); // poll -> peek
+            // Fix NPE (Spigot bug caused by handleDisconnection())
+            if (queued == null) {
+                return true;
             }
 
+            Packet packet = queued.getPacket();
+            if (!packet.isReady()) {
+                return false;
+            } else {
+                iterator.remove();
+                this.dispatchPacket(packet, queued.getGenericFutureListeners());
+            }
         }
+        return true;
     }
+    // PandaSpigot end
 
     public void a() {
         this.m();
@@ -238,9 +355,21 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
         return this.l;
     }
 
+    // PandaSpigot start
+    public void clearPacketQueue() {
+        EntityPlayer player = getPlayer();
+        i.forEach(queuedPacket -> {
+            Packet packet = queuedPacket.getPacket();
+            if (packet.hasFinishListener()) {
+                packet.onPacketDispatchFinish(player, null);
+            }
+        });
+        i.clear();
+    } // PandaSpigot end
     public void close(IChatBaseComponent ichatbasecomponent) {
         // Spigot Start
         this.preparing = false;
+        clearPacketQueue(); // PandaSpigot
         // Spigot End
         if (this.channel.isOpen()) {
             this.channel.close(); // We can't wait as this may be called from an event loop.
@@ -259,6 +388,7 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
         this.channel.pipeline().addBefore("prepender", "encrypt", new PacketEncrypter(MinecraftEncryption.a(1, secretkey)));
     }
 
+    public boolean isConnected() { return this.g(); } // PandaSpigot - OBFHELPER
     public boolean g() {
         return this.channel != null && this.channel.isOpen();
     }
@@ -313,9 +443,9 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
                 } else if (this.getPacketListener() != null) {
                     this.getPacketListener().a(new ChatComponentText("Disconnected"));
                 }
-                this.i.clear(); // Free up packet queue.
+                clearPacketQueue(); // PandaSpigot
             } else {
-                NetworkManager.g.warn("handleDisconnection() called twice");
+                //NetworkManager.g.warn("handleDisconnection() called twice"); // PandaSpigot - Do not log useless message
             }
 
         }
@@ -327,8 +457,8 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet> {
 
     static class QueuedPacket {
 
-        private final Packet a;
-        private final GenericFutureListener<? extends Future<? super Void>>[] b;
+        private final Packet a; private final Packet getPacket() { return this.a; } // PandaSpigot - OBFHELPER
+        private final GenericFutureListener<? extends Future<? super Void>>[] b; private final GenericFutureListener<? extends Future<? super Void>>[] getGenericFutureListeners() { return this.b; } // PandaSpigot - OBFHELPER
 
         public QueuedPacket(Packet packet, GenericFutureListener<? extends Future<? super Void>>... agenericfuturelistener) {
             this.a = packet;
diff --git a/src/main/java/net/minecraft/server/Packet.java b/src/main/java/net/minecraft/server/Packet.java
index fdc142b75b78289f01a9e9fed6f36432624735db..bb18dac7a3554a8a6cb93d24bc539901a4d9a89b 100644
--- a/src/main/java/net/minecraft/server/Packet.java
+++ b/src/main/java/net/minecraft/server/Packet.java
@@ -9,4 +9,36 @@ public interface Packet<T extends PacketListener> {
     void b(PacketDataSerializer packetdataserializer) throws IOException;
 
     void a(T t0);
+    // PandaSpigot start
+    /**
+     * @param player {@code null} if not at {@link EnumProtocol#PLAY} yet.
+     */
+    default void onPacketDispatch(EntityPlayer player) {}
+
+    /**
+     * @param player {@code null} if not at {@link EnumProtocol#PLAY} yet.
+     * @param future Can be {@code null} if packet was cancelled.
+     */
+    default void onPacketDispatchFinish(EntityPlayer player, io.netty.channel.ChannelFuture future) {}
+
+    /**
+     * @return Whether {@link #onPacketDispatchFinish(EntityPlayer, io.netty.channel.ChannelFuture)} should
+     * be called after this packet has been dispatched.
+     */
+    default boolean hasFinishListener() { return false; }
+
+    /**
+     * Checks whether this packet is ready to be sent.
+     * <p>
+     * If this returns {@code false}, the packet will be added to the queue, and checked every tick until it is ready.
+     *
+     * @return Whether this packet is ready.
+     */
+    default boolean isReady() { return true; }
+
+    /**
+     * @return A list of extra packets to be sent after this packet.
+     */
+    default java.util.List<Packet<?>> getExtraPackets() { return null; }
+    // PandaSpigot end
 }
diff --git a/src/main/java/net/minecraft/server/ServerConnection.java b/src/main/java/net/minecraft/server/ServerConnection.java
index 180b1432e8fbbc516ecfb60ba2da4cd319d85a97..e390ceef529d98c10bc467ed5517e1ab17b336e0 100644
--- a/src/main/java/net/minecraft/server/ServerConnection.java
+++ b/src/main/java/net/minecraft/server/ServerConnection.java
@@ -63,10 +63,12 @@ public class ServerConnection {
     private final List<NetworkManager> h = Collections.synchronizedList(Lists.<NetworkManager>newArrayList());
     // PandaSpigot start - Prevent blocking on adding a new network manager while the server is ticking
     private final java.util.Queue<NetworkManager> pending = new java.util.concurrent.ConcurrentLinkedQueue<>();
+    private static final boolean disableFlushConsolidation = Boolean.getBoolean("Paper.disableFlushConsolidate"); // PandaSpigot
     private void addPending() {
         NetworkManager manager = null;
         while ((manager = pending.poll()) != null) {
             h.add(manager);
+            manager.isPending = false;
         }
     }
     // PandaSpigot end
@@ -112,6 +114,7 @@ public class ServerConnection {
                         ;
                     }
 
+                    if (!disableFlushConsolidation) channel.pipeline().addFirst(new io.netty.handler.flush.FlushConsolidationHandler()); // PandaSpigot
                     channel.pipeline().addLast("timeout", new ReadTimeoutHandler(30)).addLast("legacy_query", new LegacyPingHandler(ServerConnection.this)).addLast("splitter", new PacketSplitter()).addLast("decoder", new PacketDecoder(EnumProtocolDirection.SERVERBOUND)).addLast("prepender", new PacketPrepender()).addLast("encoder", new PacketEncoder(EnumProtocolDirection.CLIENTBOUND));
                     NetworkManager networkmanager = new NetworkManager(EnumProtocolDirection.SERVERBOUND);