Reduce blocking in ServerConnection

src/main/java/io/akarin/server/core/AkarinAsyncScheduler.java

@@ -1,9 +1,12 @@
 package io.akarin.server.core;
 
+import java.util.List;
+
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import net.minecraft.server.MinecraftServer;
+import net.minecraft.server.NetworkManager;
 
 public class AkarinAsyncScheduler extends Thread {
     private final static Logger LOGGER = LogManager.getLogger("Akarin");
@@ -32,6 +35,14 @@ public class AkarinAsyncScheduler extends Thread {
         while (server.isRunning()) {
             long currentLoop = System.currentTimeMillis();
             
+            List<NetworkManager> networkManagers = server.getServerConnection().getNetworkManagers();
+            if (!networkManagers.isEmpty()) {
+                synchronized (networkManagers) {
+                    for (NetworkManager player : networkManagers)
+                        player.sendPacketQueue();
+                }
+            }
+            
             try {
                 long sleepFixed = STD_TICK_TIME - (System.currentTimeMillis() - currentLoop);
                 Thread.sleep(sleepFixed);

src/main/java/net/minecraft/server/IntHashMap.java

@@ -78,7 +78,7 @@ public class IntHashMap<V> {
         this.a(j, i, v0, k);
         */
         map.put(i, v0);
-     // Akarin end
+        // Akarin end
     }
 
     @Deprecated // Akarin

src/main/java/net/minecraft/server/NetworkManager.java

@@ -250,7 +250,9 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet<?>> {
     private final void dispatchPacket(Packet<?> packet, @Nullable GenericFutureListener<? extends Future<? super Void>> genericFutureListener) { this.b(packet, genericFutureListener); } // Paper - OBFHELPER
     private final void b(Packet<?> packet, @Nullable GenericFutureListener<? extends Future<? super Void>> genericfuturelistener) {
         if (!packet.canDispatchImmediately()) {
-            this.pendingChunkQueue.add((PacketPlayOutMapChunk) packet);
+            synchronized (this) {
+                this.pendingChunkQueue.add((PacketPlayOutMapChunk) packet);
+            }
             return;
         }
         // Akarin end
@@ -303,16 +305,18 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet<?>> {
     }
 
     // Paper start - Async-Anti-Xray - Stop dispatching further packets and return false if the peeked packet is a chunk packet which is not ready
-    private boolean sendPacketQueue() { return this.o(); } // OBFHELPER // void -> boolean
+    public boolean sendPacketQueue() { return this.o(); } // OBFHELPER // void -> boolean // Akarin - public
     private boolean o() { // void -> boolean
-        if (this.channel != null && this.channel.isOpen()) {
+        if (this.channel != null && this.channel.isOpen() && this.channel.isRegistered() && !this.pendingChunkQueue.isEmpty()) {
             // Akarin start
-            Iterator<PacketPlayOutMapChunk> iterator = this.pendingChunkQueue.iterator();
-            while (iterator.hasNext()) {
-                PacketPlayOutMapChunk packet = iterator.next();
-                if (packet.isReady()) {
-                    this.dispatchPacket(packet, null);
-                    iterator.remove();
+            synchronized (this) {
+                Iterator<PacketPlayOutMapChunk> iterator = this.pendingChunkQueue.iterator();
+                while (iterator.hasNext()) {
+                    PacketPlayOutMapChunk packet = iterator.next();
+                    if (packet.isReady()) {
+                        this.dispatchPacket(packet, null);
+                        iterator.remove();
+                    }
                 }
             }
             /*
@@ -348,7 +352,7 @@ public class NetworkManager extends SimpleChannelInboundHandler<Packet<?>> {
     // Paper end
 
     public void a() {
-        this.o();
+        //this.o(); // Akarin - move to scheduler
         if (this.packetListener instanceof ITickable) {
             ((ITickable) this.packetListener).tick();
         }

src/main/java/net/minecraft/server/ServerConnection.java

@@ -45,14 +45,20 @@ public class ServerConnection {
     private final MinecraftServer e;
     public volatile boolean c;
     private final List<ChannelFuture> f = Collections.synchronizedList(Lists.newArrayList());
-    private final List<NetworkManager> g = Collections.synchronizedList(Lists.newArrayList());
+    private final List<NetworkManager> g = Lists.newArrayList(); public final List<NetworkManager> getNetworkManagers() { return this.g; }// Akarin
     // Paper start - prevent blocking on adding a new network manager while the server is ticking
     private final List<NetworkManager> pending = Collections.synchronizedList(Lists.<NetworkManager>newArrayList());
+    // Akarin start
+    private final List<NetworkManager> pendingRemoval = Lists.newArrayList();
     private void addPending() {
-        synchronized (pending) {
+        // Akarin start
+        /*
+        synchronized (this.pending) {
             this.g.addAll(pending); // Paper - OBFHELPER - List of network managers
             pending.clear();
         }
+        */
+        // Akarin end
     }
     // Paper end
 
@@ -95,7 +101,7 @@ public class ServerConnection {
                     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);
 
-                    pending.add(networkmanager); // Paper
+                    //pending.add(networkmanager); // Paper // Akarin
                     channel.pipeline().addLast("packet_handler", networkmanager);
                     networkmanager.setPacketListener(new HandshakeListener(ServerConnection.this.e, networkmanager));
                 }
@@ -123,8 +129,12 @@ public class ServerConnection {
         List list = this.g;
 
         synchronized (this.g) {
+            // Akarin start
+            this.g.removeAll(pendingRemoval);
+        } {
+            // Akarin end
             // Spigot Start
-            addPending(); // Paper
+            //addPending(); // Paper // Akarin
             // This prevents players from 'gaming' the server, and strategically relogging to increase their position in the tick order
             if ( org.spigotmc.SpigotConfig.playerShuffle > 0 && MinecraftServer.currentTick % org.spigotmc.SpigotConfig.playerShuffle == 0 )
             {
@@ -162,7 +172,7 @@ public class ServerConnection {
                         // Fix a race condition where a NetworkManager could be unregistered just before connection.
                         if (networkmanager.preparing) continue;
                         // Spigot End
-                        iterator.remove();
+                        pendingRemoval.add(networkmanager); // Akarin
                         networkmanager.handleDisconnection();
                     }
                 }