Paper/CraftBukkit-Patches/0105-Implement-Threaded-Bulk-Chunk-Compression-and-Cachin.patch

From cf130aa4f7b9f3dae460f43d761870222b408fcd Mon Sep 17 00:00:00 2001
From: md_5 <git@md-5.net>
Date: Tue, 28 Jan 2014 20:32:07 +1100
Subject: [PATCH] Implement Threaded Bulk Chunk Compression and Caching


diff --git a/src/main/java/net/minecraft/server/EntityPlayer.java b/src/main/java/net/minecraft/server/EntityPlayer.java
index 9b853a9..a4c8843 100644
--- a/src/main/java/net/minecraft/server/EntityPlayer.java
+++ b/src/main/java/net/minecraft/server/EntityPlayer.java
@@ -228,6 +228,7 @@ public class EntityPlayer extends EntityHuman implements ICrafting {
             }
 
             if (!arraylist.isEmpty()) {
+                java.util.Collections.sort( arraylist, new PlayerChunkMap.ChunkComparator( this ) ); // Spigot - sort a final time before sending
                 this.playerConnection.sendPacket(new PacketPlayOutMapChunkBulk(arraylist));
                 Iterator iterator2 = arraylist1.iterator();
 
diff --git a/src/main/java/net/minecraft/server/PacketPlayOutMapChunkBulk.java b/src/main/java/net/minecraft/server/PacketPlayOutMapChunkBulk.java
index fc92026..484d727 100644
--- a/src/main/java/net/minecraft/server/PacketPlayOutMapChunkBulk.java
+++ b/src/main/java/net/minecraft/server/PacketPlayOutMapChunkBulk.java
@@ -12,9 +12,9 @@ public class PacketPlayOutMapChunkBulk extends Packet {
     private int[] b;
     private int[] c;
     private int[] d;
-    private byte[] buffer;
-    private byte[][] inflatedBuffers;
-    private int size;
+    public byte[] buffer; // Spigot
+    public byte[][] inflatedBuffers; // Spigot
+    public int size; // Spigot
     private boolean h;
     private byte[] buildBuffer = new byte[0]; // CraftBukkit - remove static
     // CraftBukkit start
diff --git a/src/main/java/net/minecraft/server/PlayerChunkMap.java b/src/main/java/net/minecraft/server/PlayerChunkMap.java
index ae4ca63..962e902 100644
--- a/src/main/java/net/minecraft/server/PlayerChunkMap.java
+++ b/src/main/java/net/minecraft/server/PlayerChunkMap.java
@@ -279,8 +279,27 @@ public class PlayerChunkMap {
         return playermanager.d;
     }
 
+    // Spigot Start
+    public static class ChunkComparator implements java.util.Comparator<Chunk>
+    {
+
+        private final ChunkCoordComparator base;
+
+        public ChunkComparator(EntityPlayer player)
+        {
+            this.base = new ChunkCoordComparator( player );
+        }
+
+        @Override
+        public int compare(Chunk o1, Chunk o2)
+        {
+            return base.compare( o1.l(), o2.l() );
+        }
+
+    }
     // CraftBukkit start - Sorter to load nearby chunks first
-    private static class ChunkCoordComparator implements java.util.Comparator<ChunkCoordIntPair> {
+    public static class ChunkCoordComparator implements java.util.Comparator<ChunkCoordIntPair> {
+        // Spigot End
         private int x;
         private int z;
 
diff --git a/src/main/java/net/minecraft/server/ServerConnectionChannel.java b/src/main/java/net/minecraft/server/ServerConnectionChannel.java
index fb95be4..a382235 100644
--- a/src/main/java/net/minecraft/server/ServerConnectionChannel.java
+++ b/src/main/java/net/minecraft/server/ServerConnectionChannel.java
@@ -1,14 +1,25 @@
 package net.minecraft.server;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder; // Spigot
 import net.minecraft.util.io.netty.channel.Channel;
 import net.minecraft.util.io.netty.channel.ChannelException;
 import net.minecraft.util.io.netty.channel.ChannelInitializer;
 import net.minecraft.util.io.netty.channel.ChannelOption;
 import net.minecraft.util.io.netty.handler.timeout.ReadTimeoutHandler;
+// Spigot Start
+import net.minecraft.util.io.netty.util.concurrent.DefaultEventExecutorGroup;
+import net.minecraft.util.io.netty.util.concurrent.EventExecutorGroup;
+import org.spigotmc.ChunkCompressor;
+import org.spigotmc.SpigotConfig;
+// Spigot End
 
 class ServerConnectionChannel extends ChannelInitializer {
 
     final ServerConnection a;
+    // Spigot Start
+    private static final EventExecutorGroup threadPool = new DefaultEventExecutorGroup( SpigotConfig.compressionThreads, new ThreadFactoryBuilder().setNameFormat( "Chunk Compressor #%d" ).setDaemon( true ).build() );
+    private static final ChunkCompressor chunkCompressor = new ChunkCompressor();
+    // Spigot End
 
     ServerConnectionChannel(ServerConnection serverconnection) {
         this.a = serverconnection;
@@ -28,6 +39,7 @@ class ServerConnectionChannel extends ChannelInitializer {
         }
 
         channel.pipeline().addLast("timeout", new ReadTimeoutHandler(30)).addLast("legacy_query", new LegacyPingHandler(this.a)).addLast("splitter", new PacketSplitter()).addLast("decoder", new PacketDecoder()).addLast("prepender", new PacketPrepender()).addLast("encoder", new PacketEncoder());
+        channel.pipeline().addLast( threadPool, "compressor", chunkCompressor ); // Spigot
         NetworkManager networkmanager = new NetworkManager(false);
 
         ServerConnection.a(this.a).add(networkmanager);
diff --git a/src/main/java/org/spigotmc/ChunkCompressor.java b/src/main/java/org/spigotmc/ChunkCompressor.java
new file mode 100644
index 0000000..78560c6
--- /dev/null
+++ b/src/main/java/org/spigotmc/ChunkCompressor.java
@@ -0,0 +1,87 @@
+package org.spigotmc;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.logging.Level;
+import java.util.zip.CRC32;
+import net.minecraft.server.PacketPlayOutMapChunkBulk;
+import net.minecraft.util.io.netty.channel.ChannelHandler;
+import net.minecraft.util.io.netty.channel.ChannelHandlerContext;
+import net.minecraft.util.io.netty.channel.ChannelOutboundHandlerAdapter;
+import net.minecraft.util.io.netty.channel.ChannelPromise;
+import org.bukkit.Bukkit;
+
+@ChannelHandler.Sharable
+public class ChunkCompressor extends ChannelOutboundHandlerAdapter
+{
+
+    private final LinkedHashMap<Long, byte[]> cache = new LinkedHashMap<Long, byte[]>( 16, 0.75f, true ); // Defaults, order by access
+    private volatile int cacheSize;
+
+    @Override
+    public synchronized void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception
+    {
+        try
+        {
+            long start = System.currentTimeMillis();
+            boolean cached = false;
+            if ( msg instanceof PacketPlayOutMapChunkBulk )
+            {
+                PacketPlayOutMapChunkBulk chunk = (PacketPlayOutMapChunkBulk) msg;
+                // Here we assign a hash to the chunk based on its contents. CRC32 is fast and sufficient for use here.
+                CRC32 crc = new CRC32();
+                for ( byte[] c : chunk.inflatedBuffers )
+                {
+                    crc.update( c );
+                }
+                long hash = crc.getValue();
+
+                byte[] deflated = cache.get( hash );
+                if ( deflated != null )
+                {
+                    chunk.buffer = deflated;
+                    chunk.size = deflated.length;
+                    cached = true;
+                } else
+                {
+                    chunk.compress(); // Compress the chunk
+                    byte[] buffer = Arrays.copyOf( chunk.buffer, chunk.size ); // Resize the array to correct sizing
+
+                    Iterator<byte[]> iter = cache.values().iterator(); // Grab a single iterator reference
+                    // Whilst this next entry is too big for us, and we have stuff to remove
+                    while ( cacheSize + buffer.length > org.spigotmc.SpigotConfig.chunkCacheBytes && iter.hasNext() )
+                    {
+                        cacheSize -= iter.next().length; // Update size table
+                        iter.remove(); // Remove it alltogether
+                    }
+                    cacheSize += buffer.length; // Update size table
+                    cache.put( hash, buffer ); // Pop the new one in the cache
+                }
+                // System.out.println( "Time: " + ( System.currentTimeMillis() - start ) + " " + cached + " " + cacheSize );
+            }
+        } catch ( Throwable t )
+        {
+            Bukkit.getServer().getLogger().log( Level.WARNING, "Error compressing or caching chunk", t );
+        }
+
+        super.write( ctx, msg, promise );
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
+    {
+        // In short, there isn't actually anything wrong with the async chunk compressor, it just accidentally caused console logging of errors which were previously ignored.\
+        // This commit restores that behaviour
+
+        // You may be asking yourself why we are completely ignoring any errors which come this far down the pipeline.
+        // The answer is quite simple:
+        // Mojang did it
+        // The default Mojang pipeline doesn't have any ChannelOutboundHandlerAdapter or similar instances, and thus nothing to handle exceptionCaught
+        // So when a channel.write() or channel.flush() fails, the error message is actually just passed straight to the future provided.
+        // It is then subsequently discarded, the channel closed, and no one except the user was any the wiser it actually happened!
+        // Unfortunately for us, the default exceptionCaught in this class sends a blaring warning to the server admins indicating that it couldn't send a packet to a disconnected user!
+        // We don't care about these warnings, if we did something wrong to disconnect the user, it is already logged in the proper location, as are broken sockets
+        // tl;dr no need to blare warnings on each write to a broken socket
+    }
+}
diff --git a/src/main/java/org/spigotmc/SpigotConfig.java b/src/main/java/org/spigotmc/SpigotConfig.java
index 769ef2a..fb944e7 100755
--- a/src/main/java/org/spigotmc/SpigotConfig.java
+++ b/src/main/java/org/spigotmc/SpigotConfig.java
@@ -266,4 +266,16 @@ public class SpigotConfig
     {
         playerShuffle = getInt( "settings.player-shuffle", 0 );
     }
+
+    public static int compressionThreads = 1; // Stupid unit tests
+    public static int chunkCacheBytes;
+    private static void chunkStuff()
+    {
+        compressionThreads = getInt( "settings.compression-threads", 4 );
+        Bukkit.getLogger().log( Level.INFO, "Using {0} threads for async chunk compression", compressionThreads );
+
+        chunkCacheBytes = getInt( "settings.compressed-chunk-cache", 64 ) << 20;
+        Bukkit.getLogger().log( Level.INFO, "Reserving {0} bytes for compressed chunk cache", chunkCacheBytes );
+
+    }
 }
-- 
1.8.3.2