From bc09fc8741ee1e5b49246eaa26e2a0cb6e61f26d Mon Sep 17 00:00:00 2001
From: nathan <nathan@dorkbox.com>
Date: Mon, 29 Feb 2016 21:24:56 +0100
Subject: [PATCH] Serialization is now ThreadLocal. Changed data compression
 from snappy -> LZ4

---
 Dorkbox-Network/Dorkbox-Network.iml           |   2 +-
 .../KryoCryptoSerializationManager.java       | 870 +++++-------------
 .../dorkbox/network/connection/KryoExtra.java | 425 ++++++++-
 .../rmi/InvocationHandlerSerializer.java      |  42 +
 .../rmi/InvocationResultSerializer.java       |  30 +
 .../network/rmi/RemoteInvocationHandler.java  |  12 +-
 .../dorkbox/network/rmi/RmiGlobalTest.java    |   1 -
 .../network/rmi/RmiSendObjectTest.java        |   2 -
 8 files changed, 685 insertions(+), 699 deletions(-)
 create mode 100644 Dorkbox-Network/src/dorkbox/network/rmi/InvocationHandlerSerializer.java
 create mode 100644 Dorkbox-Network/src/dorkbox/network/rmi/InvocationResultSerializer.java

diff --git a/Dorkbox-Network/Dorkbox-Network.iml b/Dorkbox-Network/Dorkbox-Network.iml
index 0d6d7185..2b804a58 100644
--- a/Dorkbox-Network/Dorkbox-Network.iml
+++ b/Dorkbox-Network/Dorkbox-Network.iml
@@ -46,10 +46,10 @@
     <orderEntry type="library" name="netty-all (4.1.0)" level="application" />
     <orderEntry type="library" name="reflectasm" level="application" />
     <orderEntry type="library" name="objenesis (2.1)" level="application" />
-    <orderEntry type="library" name="JCTools (1.4 alpha MTAQ)" level="application" />
     <orderEntry type="library" name="junit-4.12" level="application" />
     <orderEntry type="library" name="bouncyCastle bcpkix-jdk15on (1.53)" level="application" />
     <orderEntry type="library" name="bouncyCastle bcprov-jdk15on (1.53)" level="application" />
+    <orderEntry type="library" name="lz4-1.3.0" level="application" />
   </component>
   <component name="org.twodividedbyzero.idea.findbugs">
     <option name="_basePreferences">
diff --git a/Dorkbox-Network/src/dorkbox/network/connection/KryoCryptoSerializationManager.java b/Dorkbox-Network/src/dorkbox/network/connection/KryoCryptoSerializationManager.java
index 950fcdb6..4c4f1aea 100644
--- a/Dorkbox-Network/src/dorkbox/network/connection/KryoCryptoSerializationManager.java
+++ b/Dorkbox-Network/src/dorkbox/network/connection/KryoCryptoSerializationManager.java
@@ -29,17 +29,15 @@ import com.esotericsoftware.kryo.serializers.FieldSerializer;
 import com.esotericsoftware.kryo.util.MapReferenceResolver;
 import dorkbox.network.connection.ping.PingMessage;
 import dorkbox.network.rmi.CachedMethod;
+import dorkbox.network.rmi.InvocationHandlerSerializer;
+import dorkbox.network.rmi.InvocationResultSerializer;
 import dorkbox.network.rmi.InvokeMethod;
 import dorkbox.network.rmi.InvokeMethodResult;
 import dorkbox.network.rmi.InvokeMethodSerializer;
-import dorkbox.network.rmi.RemoteInvocationHandler;
 import dorkbox.network.rmi.RemoteObjectSerializer;
 import dorkbox.network.rmi.RmiRegistration;
 import dorkbox.network.util.CryptoSerializationManager;
-import dorkbox.util.crypto.CryptoAES;
-import dorkbox.util.objectPool.ObjectPool;
-import dorkbox.util.objectPool.ObjectPoolFactory;
-import dorkbox.util.objectPool.PoolableObject;
+import dorkbox.util.Property;
 import dorkbox.util.serialization.ArraysAsListSerializer;
 import dorkbox.util.serialization.EccPrivateKeySerializer;
 import dorkbox.util.serialization.EccPublicKeySerializer;
@@ -49,66 +47,46 @@ import dorkbox.util.serialization.IesWithCipherParametersSerializer;
 import dorkbox.util.serialization.IgnoreSerialization;
 import dorkbox.util.serialization.UnmodifiableCollectionsSerializer;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
-import io.netty.handler.codec.compression.CompressionException;
-import io.netty.handler.codec.compression.SnappyAccess;
 import org.bouncycastle.crypto.params.ECPrivateKeyParameters;
 import org.bouncycastle.crypto.params.ECPublicKeyParameters;
 import org.bouncycastle.crypto.params.IESParameters;
 import org.bouncycastle.crypto.params.IESWithCipherParameters;
-import org.jctools.util.Pow2;
 import org.slf4j.Logger;
 
 import java.io.IOException;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Proxy;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.zip.DataFormatException;
-import java.util.zip.Deflater;
-import java.util.zip.Inflater;
+import java.util.List;
 
 /**
- * Threads reading/writing, it messes up a single instance.
- * it is possible to use a single kryo with the use of synchronize, however - that defeats the point of multi-threaded
+ * Threads reading/writing, it messes up a single instance. it is possible to use a single kryo with the use of synchronize, however - that
+ * defeats the point of multi-threaded
  */
 @SuppressWarnings({"unused", "StaticNonFinalField"})
 public
 class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
     private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(KryoCryptoSerializationManager.class);
-    private static final int capacity = Pow2.roundToPowerOfTwo(Runtime.getRuntime()
-                                                                      .availableProcessors() * 32);
 
-    private static final boolean ENABLE_SNAPPY = false;
-
-    /**
-     * The minimum amount that we'll consider actually attempting to compress.
-     * This value is preamble + the minimum length our Snappy service will
-     * compress (instead of just emitting a literal).
-     */
-    private static final int MIN_COMPRESSIBLE_LENGTH = 18;
-
-    /**
-     * bit masks
-     */
-    private static final int compression = 1;
-    private static final int crypto = 1 << 1;
+    // round to the nearest power of 2
+    private static final int capacity = 1 << (32 - Integer.numberOfLeadingZeros((Runtime.getRuntime().availableProcessors() * 2) - 1));
 
     // compression options
+    @Property
     static final int compressionLevel = 6;
-    private static final ByteBuf NULL_BUFFER = null;
+
 
     /**
      * Specify if we want KRYO to use unsafe memory for serialization, or to use the ASM backend. Unsafe memory use is WAY faster, but is
      * limited to the "same endianess" on all endpoints, and unsafe DOES NOT work on android.
      */
+    @Property
     public static boolean useUnsafeMemory = false;
 
     private static final String OBJECT_ID = "objectID";
-    private boolean initialized = false;
 
     /**
      * The default serialization manager. This is static, since serialization must be consistent within the JVM. This can be changed.
@@ -148,273 +126,86 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
         // extra serializers
         //noinspection ArraysAsListWithZeroOrOneArgument
-        serializationManager.register(Arrays.asList("").getClass(), new ArraysAsListSerializer());
+        serializationManager.register(Arrays.asList("")
+                                            .getClass(), new ArraysAsListSerializer());
 
         UnmodifiableCollectionsSerializer.registerSerializers(serializationManager);
 
         return serializationManager;
     }
 
-    @SuppressWarnings("unused")
-    private static
-    void compress(ByteBuf inputBuffer, ByteBuf outputBuffer, int length, Deflater compress) {
 
-        byte[] in = new byte[inputBuffer.readableBytes()];
-        inputBuffer.readBytes(in);
+    static class ClassSerializer {
+        final Class<?> clazz;
+        final Serializer<?> serializer;
 
-        compress.reset();
-        compress.setInput(in);
-        compress.finish();
+        public
+        ClassSerializer(final Class<?> clazz, final Serializer<?> serializer) {
+            this.clazz = clazz;
+            this.serializer = serializer;
+        }
+    }
+    static class ClassSerializer2 {
+        final Class<?> clazz;
+        final Serializer<?> serializer;
+        final int id;
 
-        byte[] out = new byte[1024];
-        int numBytes = out.length;
-        while (numBytes == out.length) {
-            numBytes = compress.deflate(out, 0, out.length);
-            outputBuffer.writeBytes(out, 0, numBytes);
+        public
+        ClassSerializer2(final Class<?> clazz, final Serializer<?> serializer, final int id) {
+            this.clazz = clazz;
+            this.serializer = serializer;
+            this.id = id;
+        }
+    }
+    static class RemoteClass<Iface, Impl extends Iface> {
+        private final Class<Iface> ifaceClass;
+        private final Class<Impl> implClass;
+
+        public
+        RemoteClass(final Class<Iface> ifaceClass, final Class<Impl> implClass) {
+            this.ifaceClass = ifaceClass;
+            this.implClass = implClass;
         }
     }
 
+    private boolean initialized = false;
 
-    private static
-    void decompress(ByteBuf inputBuffer, ByteBuf outputBuffer, Inflater decompress) throws IOException {
-        byte[] in = new byte[inputBuffer.readableBytes()];
-        inputBuffer.readBytes(in);
+    private final ThreadLocal<KryoExtra> kryoThreadLocal;
 
-        decompress.reset();
-        decompress.setInput(in);
+    // used by operations performed during kryo initialization, which are by default package access (since it's an anon-inner class
+    final List<Class<?>> classesToRegister = new ArrayList<Class<?>>();
+    final List<ClassSerializer> classSerializerToRegister = new ArrayList<ClassSerializer>();
+    final List<ClassSerializer2> classSerializer2ToRegister = new ArrayList<ClassSerializer2>();
+    final List<RemoteClass> remoteClassToRegister = new ArrayList<RemoteClass>();
 
-        byte[] out = new byte[1024];
-        int numBytes = out.length;
-        while (numBytes == out.length) {
-            try {
-                numBytes = decompress.inflate(out, 0, out.length);
-            } catch (DataFormatException e) {
-                logger.error("Error inflating data.", e);
-                throw new IOException(e.getCause());
-            }
+    boolean shouldInitRMI = false;
+    InvokeMethodSerializer methodSerializer = null;
+    Serializer<Object> invocationSerializer = null;
+    RemoteObjectSerializer remoteObjectSerializer;
 
-            outputBuffer.writeBytes(out, 0, numBytes);
-        }
-    }
 
-    private static
-    void snappyCompress(ByteBuf inputBuffer, ByteBuf outputBuffer, int length, SnappyAccess snappy) {
-        // compress the tempBuffer (which has our object serialized inside it)
-
-        // If we have lots of available data, break it up into smaller chunks
-        int dataLength = length;
-        while (true) {
-            final int lengthIdx = outputBuffer.writerIndex() + 1;
-            if (dataLength < MIN_COMPRESSIBLE_LENGTH) {
-                ByteBuf slice = inputBuffer.readSlice(dataLength);
-                writeUnencodedChunk(slice, outputBuffer, dataLength);
-                break;
-            }
-
-            outputBuffer.writeInt(0);
-            if (dataLength > Short.MAX_VALUE) {
-                ByteBuf slice = inputBuffer.readSlice(Short.MAX_VALUE);
-                calculateAndWriteChecksum(slice, outputBuffer);
-                snappy.encode(slice, outputBuffer, Short.MAX_VALUE);
-                setChunkLength(outputBuffer, lengthIdx);
-                dataLength -= Short.MAX_VALUE;
-            }
-            else {
-                ByteBuf slice = inputBuffer.readSlice(dataLength);
-                calculateAndWriteChecksum(slice, outputBuffer);
-                snappy.encode(slice, outputBuffer, dataLength);
-                setChunkLength(outputBuffer, lengthIdx);
-                break;
-            }
-        }
-    }
-
-    private static
-    void snappyDecompress(ByteBuf inputBuffer, ByteBuf outputBuffer, SnappyAccess snappy) throws IOException {
-        try {
-            int idx = inputBuffer.readerIndex();
-            final int inSize = inputBuffer.writerIndex() - idx;
-            if (inSize < 4) {
-                // We need to be at least able to read the chunk type identifier (one byte),
-                // and the length of the chunk (3 bytes) in order to proceed
-                return;
-            }
-
-            final int chunkTypeVal = inputBuffer.getUnsignedByte(idx);
-            final ChunkType chunkType = mapChunkType((byte) chunkTypeVal);
-            final int chunkLength = ByteBufUtil.swapMedium(inputBuffer.getUnsignedMedium(idx + 1));
-
-            switch (chunkType) {
-                case RESERVED_SKIPPABLE:
-                    if (inSize < 4 + chunkLength) {
-                        // TODO: Don't keep skippable bytes
-                        return;
-                    }
-
-                    inputBuffer.skipBytes(4 + chunkLength);
-                    break;
-                case RESERVED_UNSKIPPABLE:
-                    // The spec mandates that reserved unskippable chunks must immediately
-                    // return an error, as we must assume that we cannot decode the stream
-                    // correctly
-                    throw new CompressionException("Found reserved unskippable chunk type: 0x" + Integer.toHexString(chunkTypeVal));
-                case UNCOMPRESSED_DATA:
-                    if (chunkLength > 65536 + 4) {
-                        throw new CompressionException("Received UNCOMPRESSED_DATA larger than 65540 bytes");
-                    }
-
-                    if (inSize < 4 + chunkLength) {
-                        return;
-                    }
-
-                    inputBuffer.skipBytes(4);
-                {
-                    int checksum = ByteBufUtil.swapInt(inputBuffer.readInt());
-                    validateChecksum(checksum, inputBuffer, inputBuffer.readerIndex(), chunkLength - 4);
-                    outputBuffer.writeBytes(inputBuffer, chunkLength - 4);
-                }
-                break;
-                case COMPRESSED_DATA:
-                    if (inSize < 4 + chunkLength) {
-                        return;
-                    }
-
-                    inputBuffer.skipBytes(4);
-                {
-                    int checksum = ByteBufUtil.swapInt(inputBuffer.readInt());
-                    int oldWriterIndex = inputBuffer.writerIndex();
-                    int uncompressedStart = outputBuffer.writerIndex();
-                    try {
-                        inputBuffer.writerIndex(inputBuffer.readerIndex() + chunkLength - 4);
-                        snappy.decode(inputBuffer, outputBuffer);
-                    } finally {
-                        inputBuffer.writerIndex(oldWriterIndex);
-                    }
-                    int uncompressedLength = outputBuffer.writerIndex() - uncompressedStart;
-                    validateChecksum(checksum, outputBuffer, uncompressedStart, uncompressedLength);
-                }
-                snappy.reset();
-                break;
-            }
-        } catch (Exception e) {
-            throw new IOException("Unable to decompress SNAPPY data!! " + e.getMessage());
-        }
-    }
 
     /**
-     * Decodes the chunk type from the type tag byte.
-     *
-     * @param type The tag byte extracted from the stream
-     * @return The appropriate {@link ChunkType}, defaulting to {@link ChunkType#RESERVED_UNSKIPPABLE}
-     */
-    static
-    ChunkType mapChunkType(byte type) {
-        if (type == 0) {
-            return ChunkType.COMPRESSED_DATA;
-        }
-        else if (type == 1) {
-            return ChunkType.UNCOMPRESSED_DATA;
-        }
-        else if ((type & 0x80) == 0x80) {
-            return ChunkType.RESERVED_SKIPPABLE;
-        }
-        else {
-            return ChunkType.RESERVED_UNSKIPPABLE;
-        }
-    }
-
-    /**
-     * Computes the CRC32 checksum of the supplied data, performs the "mask" operation
-     * on the computed checksum, and then compares the resulting masked checksum to the
-     * supplied checksum.
-     *
-     * @param expectedChecksum The checksum decoded from the stream to compare against
-     * @param data             The input data to calculate the CRC32 checksum of
-     * @throws CompressionException If the calculated and supplied checksums do not match
-     */
-    static
-    void validateChecksum(int expectedChecksum, ByteBuf data) {
-        validateChecksum(expectedChecksum, data, data.readerIndex(), data.readableBytes());
-    }
-
-    /**
-     * Computes the CRC32 checksum of the supplied data, performs the "mask" operation
-     * on the computed checksum, and then compares the resulting masked checksum to the
-     * supplied checksum.
-     *
-     * @param expectedChecksum The checksum decoded from the stream to compare against
-     * @param data             The input data to calculate the CRC32 checksum of
-     * @throws CompressionException If the calculated and supplied checksums do not match
-     */
-    static
-    void validateChecksum(int expectedChecksum, ByteBuf data, int offset, int length) {
-        final int actualChecksum = SnappyAccess.calculateChecksum(data, offset, length);
-        if (actualChecksum != expectedChecksum) {
-            throw new CompressionException("mismatching checksum: " + Integer.toHexString(actualChecksum) +
-                                           " (expected: " + Integer.toHexString(expectedChecksum) + ')');
-        }
-    }
-
-    private static
-    void writeUnencodedChunk(ByteBuf in, ByteBuf out, int dataLength) {
-        out.writeByte(1);
-        writeChunkLength(out, dataLength + 4);
-        calculateAndWriteChecksum(in, out);
-        out.writeBytes(in, dataLength);
-    }
-
-    private static
-    void setChunkLength(ByteBuf out, int lengthIdx) {
-        int chunkLength = out.writerIndex() - lengthIdx - 3;
-        if (chunkLength >>> 24 != 0) {
-            throw new CompressionException("compressed data too large: " + chunkLength);
-        }
-        out.setMedium(lengthIdx, ByteBufUtil.swapMedium(chunkLength));
-    }
-
-    /**
-     * Writes the 2-byte chunk length to the output buffer.
-     *
-     * @param out         The buffer to write to
-     * @param chunkLength The length to write
-     */
-    private static
-    void writeChunkLength(ByteBuf out, int chunkLength) {
-        out.writeMedium(ByteBufUtil.swapMedium(chunkLength));
-    }
-
-    /**
-     * Calculates and writes the 4-byte checksum to the output buffer
-     *
-     * @param slice The data to calculate the checksum for
-     * @param out   The output buffer to write the checksum to
-     */
-    private static
-    void calculateAndWriteChecksum(ByteBuf slice, ByteBuf out) {
-        out.writeInt(ByteBufUtil.swapInt(SnappyAccess.calculateChecksum(slice)));
-    }
-
-    final ObjectPool<Kryo> pool;
-
-    /**
-     * @param references           If true, each appearance of an object in the graph after the first is stored as an integer ordinal.
-     *                             When set to true, {@link MapReferenceResolver} is used. This enables references to the same object and
-     *                             cyclic graphs to be serialized, but typically adds overhead of one byte per object. (should be true)
-     *                             <p/>
-     * @param registrationRequired If true, an exception is thrown when an unregistered class is encountered.
-     *                             <p/>
-     *                             If false, when an unregistered class is encountered, its fully qualified class name will be serialized
-     *                             and the {@link Kryo#addDefaultSerializer(Class, Class) default serializer} for the class used to
-     *                             serialize the object. Subsequent appearances of the class within the same object graph are serialized
-     *                             as an int id.
-     *                             <p/>
-     *                             Registered classes are serialized as an int id, avoiding the overhead of serializing the class name,
-     *                             but have the drawback of needing to know the classes to be serialized up front.
-     *                             <p/>
-     * @param factory              Sets the serializer factory to use when no {@link Kryo#addDefaultSerializer(Class, Class) default
-     *                             serializers} match an object's type. Default is {@link ReflectionSerializerFactory} with
-     *                             {@link FieldSerializer}. @see Kryo#newDefaultSerializer(Class)
-     *                             <p/>
+     * @param references
+     *                 If true, each appearance of an object in the graph after the first is stored as an integer ordinal. When set to true,
+     *                 {@link MapReferenceResolver} is used. This enables references to the same object and cyclic graphs to be serialized,
+     *                 but typically adds overhead of one byte per object. (should be true)
+     *                 <p>
+     * @param registrationRequired
+     *                 If true, an exception is thrown when an unregistered class is encountered.
+     *                 <p>
+     *                 If false, when an unregistered class is encountered, its fully qualified class name will be serialized and the {@link
+     *                 Kryo#addDefaultSerializer(Class, Class) default serializer} for the class used to serialize the object. Subsequent
+     *                 appearances of the class within the same object graph are serialized as an int id.
+     *                 <p>
+     *                 Registered classes are serialized as an int id, avoiding the overhead of serializing the class name, but have the
+     *                 drawback of needing to know the classes to be serialized up front.
+     *                 <p>
+     * @param factory
+     *                 Sets the serializer factory to use when no {@link Kryo#addDefaultSerializer(Class, Class) default serializers} match
+     *                 an object's type. Default is {@link ReflectionSerializerFactory} with {@link FieldSerializer}. @see
+     *                 Kryo#newDefaultSerializer(Class)
+     *                 <p>
      */
     public
     KryoCryptoSerializationManager(final boolean references, final boolean registrationRequired, final SerializerFactory factory) {
@@ -423,264 +214,193 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
         // This pool wil also pre-populate, so that we have the hit on startup, instead of on access
         // this is also so that our register methods can correctly register with all of the kryo instances
-        pool = ObjectPoolFactory.create(new PoolableObject<Kryo>() {
+        kryoThreadLocal = new ThreadLocal<KryoExtra>() {
             @Override
-            public
-            Kryo create() {
-                KryoExtra kryo = new KryoExtra();
+            protected
+            KryoExtra initialValue() {
+                synchronized (KryoCryptoSerializationManager.this) {
+                    KryoExtra kryo = new KryoExtra();
 
-                // we HAVE to pre-allocate the KRYOs
-                boolean useAsm = !useUnsafeMemory;
+                    // we HAVE to pre-allocate the KRYOs
+                    boolean useAsm = !useUnsafeMemory;
 
-                kryo.setAsmEnabled(useAsm);
-                kryo.setRegistrationRequired(registrationRequired);
+                    kryo.setAsmEnabled(useAsm);
+                    kryo.setRegistrationRequired(registrationRequired);
 
-                kryo.setReferences(references);
+                    kryo.setReferences(references);
 
-                if (factory != null) {
-                    kryo.setDefaultSerializer(factory);
+                    for (Class<?> clazz : classesToRegister) {
+                        kryo.register(clazz);
+                    }
+
+                    for (ClassSerializer classSerializer : classSerializerToRegister) {
+                        kryo.register(classSerializer.clazz, classSerializer.serializer);
+                    }
+
+                    for (ClassSerializer2 classSerializer : classSerializer2ToRegister) {
+                        kryo.register(classSerializer.clazz, classSerializer.serializer, classSerializer.id);
+                    }
+
+                    if (shouldInitRMI) {
+                        kryo.register(Class.class);
+                        kryo.register(RmiRegistration.class);
+                        kryo.register(InvokeMethod.class, methodSerializer);
+                        kryo.register(Object[].class);
+
+                        // This has to be PER KRYO,
+                        InvocationResultSerializer resultSerializer = new InvocationResultSerializer(kryo);
+                        resultSerializer.removeField(OBJECT_ID);
+
+                        kryo.register(InvokeMethodResult.class, resultSerializer);
+                        kryo.register(InvocationHandler.class, invocationSerializer);
+
+                        for (RemoteClass remoteClass : remoteClassToRegister) {
+                            kryo.register(remoteClass.implClass, remoteObjectSerializer);
+
+                            // After all common registrations, register OtherObjectImpl only on the server using the remote object interface ID.
+                            // This causes OtherObjectImpl to be serialized as OtherObject.
+                            int otherObjectID = kryo.getRegistration(remoteClass.implClass).getId();
+
+                            // this overrides the 'otherObjectID' with the specified class/serializer, so that when we WRITE this ID, the impl is READ
+                            kryo.register(remoteClass.ifaceClass, remoteObjectSerializer, otherObjectID);
+
+                            // we have to save the fact that we might have overridden methods
+                            CachedMethod.registerOverridden(remoteClass.ifaceClass, remoteClass.implClass);
+                        }
+                    }
+
+                    if (factory != null) {
+                        kryo.setDefaultSerializer(factory);
+                    }
+
+                    return kryo;
                 }
-
-                return kryo;
             }
-        }, capacity);
+        };
     }
 
     /**
-     * If the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is false, it is
-     * automatically registered using the {@link Kryo#addDefaultSerializer(Class, Class) default serializer}.
+     * If the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is false, it is automatically registered using the
+     * {@link Kryo#addDefaultSerializer(Class, Class) default serializer}.
      *
-     * @throws IllegalArgumentException if the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is true.
+     * @throws IllegalArgumentException  if the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is true.
      * @see ClassResolver#getRegistration(Class)
      */
     @Override
     public synchronized
     Registration getRegistration(Class<?> clazz) {
-        Kryo kryo = null;
-        Registration r = null;
-
-        try {
-            kryo = this.pool.take();
-            r = kryo.getRegistration(clazz);
-        } catch (InterruptedException e) {
-            final String msg = "Interrupted during getRegistration()";
-            logger.error(msg);
-        } finally {
-            if (kryo != null) {
-                this.pool.release(kryo);
-            }
-        }
-
-        return r;
+        return kryoThreadLocal.get().getRegistration(clazz);
     }
 
     /**
-     * Registers the class using the lowest, next available integer ID and the
-     * {@link Kryo#getDefaultSerializer(Class) default serializer}. If the class
-     * is already registered, the existing entry is updated with the new
-     * serializer. Registering a primitive also affects the corresponding
-     * primitive wrapper.
-     * <p/>
-     * Because the ID assigned is affected by the IDs registered before it, the
-     * order classes are registered is important when using this method. The
-     * order must be the same at deserialization as it was for serialization.
+     * Registers the class using the lowest, next available integer ID and the {@link Kryo#getDefaultSerializer(Class) default serializer}.
+     * If the class is already registered, the existing entry is updated with the new serializer. Registering a primitive also affects the
+     * corresponding primitive wrapper.
+     * <p>
+     * Because the ID assigned is affected by the IDs registered before it, the order classes are registered is important when using this
+     * method. The order must be the same at deserialization as it was for serialization.
      */
     @Override
     public synchronized
     void register(Class<?> clazz) {
         if (initialized) {
-            throw new RuntimeException("Cannot register classes after initialization.");
+            logger.warn("Serialization manager already initialized. Ignoring duplicate register(Class) call.");
         }
 
-        Kryo kryo;
-        try {
-            for (int i = 0; i < capacity; i++) {
-                kryo = this.pool.take();
-                kryo.register(clazz);
-                this.pool.release(kryo);
-            }
-        } catch (InterruptedException e) {
-            throw new RuntimeException("Thread interrupted during kryo pool take/release cycle!", e);
-        }
+        classesToRegister.add(clazz);
     }
 
     /**
-     * Registers the class using the lowest, next available integer ID and the
-     * specified serializer. If the class is already registered, the existing
-     * entry is updated with the new serializer. Registering a primitive also
-     * affects the corresponding primitive wrapper.
-     * <p/>
-     * Because the ID assigned is affected by the IDs registered before it, the
-     * order classes are registered is important when using this method. The
-     * order must be the same at deserialization as it was for serialization.
+     * Registers the class using the lowest, next available integer ID and the specified serializer. If the class is already registered, the
+     * existing entry is updated with the new serializer. Registering a primitive also affects the corresponding primitive wrapper.
+     * <p>
+     * Because the ID assigned is affected by the IDs registered before it, the order classes are registered is important when using this
+     * method. The order must be the same at deserialization as it was for serialization.
      */
     @Override
     public synchronized
     void register(Class<?> clazz, Serializer<?> serializer) {
         if (initialized) {
-            throw new RuntimeException("Cannot register classes after initialization.");
+            logger.warn("Serialization manager already initialized. Ignoring duplicate register(Class, Serializer) call.");
+            return;
         }
 
-        Kryo kryo;
-        try {
-            for (int i = 0; i < capacity; i++) {
-                kryo = this.pool.take();
-                kryo.register(clazz, serializer);
-                this.pool.release(kryo);
-            }
-        } catch (InterruptedException e) {
-            throw new RuntimeException("Thread interrupted during kryo pool take/release cycle!", e);
-        }
+        classSerializerToRegister.add(new ClassSerializer(clazz, serializer));
     }
 
     /**
-     * Registers the class using the specified ID and serializer. If the ID is
-     * already in use by the same type, the old entry is overwritten. If the ID
-     * is already in use by a different type, a {@link KryoException} is thrown.
-     * Registering a primitive also affects the corresponding primitive wrapper.
-     * <p/>
+     * Registers the class using the specified ID and serializer. If the ID is already in use by the same type, the old entry is
+     * overwritten. If the ID is already in use by a different type, a {@link KryoException} is thrown. Registering a primitive also affects
+     * the corresponding primitive wrapper.
+     * <p>
      * IDs must be the same at deserialization as they were for serialization.
      *
-     * @param id Must be >= 0. Smaller IDs are serialized more efficiently. IDs
-     *           0-8 are used by default for primitive types and String, but
-     *           these IDs can be repurposed.
+     * @param id Must be >= 0. Smaller IDs are serialized more efficiently. IDs 0-8 are used by default for primitive types and
+     *           String, but these IDs can be repurposed.
      */
     @Override
     public synchronized
     void register(Class<?> clazz, Serializer<?> serializer, int id) {
         if (initialized) {
-            throw new RuntimeException("Cannot register classes after initialization.");
+            logger.warn("Serialization manager already initialized. Ignoring duplicate register(Class, Serializer, int) call.");
+            return;
         }
 
-        Kryo kryo;
-        try {
-            for (int i = 0; i < capacity; i++) {
-                kryo = this.pool.take();
-                kryo.register(clazz, serializer, id);
-                this.pool.release(kryo);
-            }
-        } catch (InterruptedException e) {
-            throw new RuntimeException("Thread interrupted during kryo pool take/release cycle!", e);
-        }
+        classSerializer2ToRegister.add(new ClassSerializer2(clazz, serializer, id));
     }
 
     /**
-     * Objects that we want to use RMI with must be accessed via an interface. This method configures the serialization of an
-     * implementation to be serialized via the defined interface, as a RemoteObject (ie: proxy object). If the implementation
-     * class is ALREADY registered, then it's registration will be overwritten by this one
+     * Objects that we want to use RMI with must be accessed via an interface. This method configures the serialization of an implementation
+     * to be serialized via the defined interface, as a RemoteObject (ie: proxy object). If the implementation class is ALREADY registered,
+     * then it's registration will be overwritten by this one
      *
      * @param ifaceClass The interface used to access the remote object
-     * @param implClass  The implementation class of the interface
+     * @param implClass The implementation class of the interface
      */
     @Override
     public synchronized
     <Iface, Impl extends Iface> void registerRemote(final Class<Iface> ifaceClass, final Class<Impl> implClass) {
-        register(implClass, new RemoteObjectSerializer<Impl>());
-
-        // After all common registrations, register OtherObjectImpl only on the server using the remote object interface ID.
-        // This causes OtherObjectImpl to be serialized as OtherObject.
-        int otherObjectID = getRegistration(implClass).getId();
-
-        // this overrides the 'otherObjectID' with the specified class/serializer, so that when we WRITE this ID, the impl is READ
-        register(ifaceClass, new RemoteObjectSerializer<Impl>(), otherObjectID);
-
-        // we have to save the fact that we might have overridden methods
-        CachedMethod.registerOverridden(ifaceClass, implClass);
+        remoteClassToRegister.add(new RemoteClass<Iface, Impl>(ifaceClass, implClass));
     }
 
     /**
-     * Necessary to register classes for RMI, only called once when the RMI bridge is created.
+     * Necessary to register classes for RMI, only called once if/when the RMI bridge is created.
      */
     @Override
     public synchronized
     void initRmiSerialization() {
         if (initialized) {
-            // already initialized.
+            logger.warn("Serialization manager already initialized. Ignoring duplicate initRmiSerialization() call.");
             return;
         }
 
-        InvokeMethodSerializer methodSerializer = new InvokeMethodSerializer();
-        Serializer<Object> invocationSerializer = new Serializer<Object>() {
-            @Override
-            public
-            void write(Kryo kryo, Output output, Object object) {
-                RemoteInvocationHandler handler = (RemoteInvocationHandler) Proxy.getInvocationHandler(object);
-                output.writeInt(handler.objectID, true);
-            }
+        methodSerializer = new InvokeMethodSerializer();
+        invocationSerializer = new InvocationHandlerSerializer(logger);
+        remoteObjectSerializer = new RemoteObjectSerializer();
 
-            @Override
-            @SuppressWarnings({"unchecked", "AutoBoxing"})
-            public
-            Object read(Kryo kryo, Input input, Class<Object> type) {
-                int objectID = input.readInt(true);
-
-                KryoExtra kryoExtra = (KryoExtra) kryo;
-                Object object = kryoExtra.connection.getImplementationObject(objectID);
-
-                if (object == null) {
-                    logger.error("Unknown object ID in RMI ObjectSpace: {}", objectID);
-                }
-                return object;
-            }
-        };
-
-        Kryo kryo;
-        try {
-            for (int i = 0; i < capacity; i++) {
-                kryo = this.pool.take();
-
-                kryo.register(Class.class);
-                kryo.register(RmiRegistration.class);
-                kryo.register(InvokeMethod.class, methodSerializer);
-                kryo.register(Object[].class);
-
-                FieldSerializer<InvokeMethodResult> resultSerializer = new FieldSerializer<InvokeMethodResult>(kryo, InvokeMethodResult.class) {
-                    @Override
-                    public
-                    void write(Kryo kryo, Output output, InvokeMethodResult result) {
-                        super.write(kryo, output, result);
-                        output.writeInt(result.objectID, true);
-                    }
-
-                    @Override
-                    public
-                    InvokeMethodResult read(Kryo kryo, Input input, Class<InvokeMethodResult> type) {
-                        InvokeMethodResult result = super.read(kryo, input, type);
-                        result.objectID = input.readInt(true);
-                        return result;
-                    }
-                };
-                resultSerializer.removeField(OBJECT_ID);
-                kryo.register(InvokeMethodResult.class, resultSerializer);
-                kryo.register(InvocationHandler.class, invocationSerializer);
-
-                this.pool.release(kryo);
-            }
-        } catch (InterruptedException e) {
-            throw new RuntimeException("Thread interrupted during kryo pool take/release cycle!", e);
-        }
+        shouldInitRMI = true;
     }
 
     /**
      * Called when initialization is complete. This is to prevent (and recognize) out-of-order class/serializer registration.
      */
     @Override
-    public
+    public synchronized
     void finishInit() {
         initialized = true;
     }
 
     @Override
-    public
+    public synchronized
     boolean initialized() {
         return initialized;
     }
 
     /**
      * Waits until a kryo is available to write, using CAS operations to prevent having to synchronize.
-     * <p/>
+     * <p>
      * No crypto and no sequence number
-     * <p/>
+     * <p>
      * There is a small speed penalty if there were no kryo's available to use.
      */
     @Override
@@ -691,7 +411,7 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
     /**
      * Reads an object from the buffer.
-     * <p/>
+     * <p>
      * No crypto and no sequence number
      *
      * @param length should ALWAYS be the length of the expected object!
@@ -699,7 +419,7 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
     @Override
     public final
     Object read(final ByteBuf buffer, final int length) throws IOException {
-        return read0(null, buffer, length, false);
+        return read0(null, buffer, length);
     }
 
     /**
@@ -708,11 +428,10 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
     @Override
     public
     void writeFullClassAndObject(final Logger logger, final Output output, final Object value) throws IOException {
-        Kryo kryo = null;
-        boolean prev = false;
+        Kryo kryo = kryoThreadLocal.get();
+        boolean prev;
 
         try {
-            kryo = this.pool.take();
             prev = kryo.isRegistrationRequired();
             kryo.setRegistrationRequired(false);
 
@@ -723,22 +442,21 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
                 logger.error(msg, ex);
             }
             throw new IOException(msg, ex);
-        } finally {
-            if (kryo != null) {
-                kryo.setRegistrationRequired(prev);
-                this.pool.release(kryo);
-            }
         }
+
+        kryo.setRegistrationRequired(prev);
     }
 
+    /**
+     * Returns a class read from the input
+     */
     @Override
     public
     Object readFullClassAndObject(final Logger logger, final Input input) throws IOException {
-        Kryo kryo = null;
+        Kryo kryo = kryoThreadLocal.get();
         boolean prev = false;
 
         try {
-            kryo = this.pool.take();
             prev = kryo.isRegistrationRequired();
             kryo.setRegistrationRequired(false);
 
@@ -750,23 +468,14 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
             }
             throw new IOException(msg, ex);
         } finally {
-            if (kryo != null) {
-                kryo.setRegistrationRequired(prev);
-                this.pool.release(kryo);
-            }
+            kryo.setRegistrationRequired(prev);
         }
     }
 
     @Override
     public
-    Kryo take() throws InterruptedException {
-        return this.pool.take();
-    }
-
-    @Override
-    public
-    void release(final Kryo kryo) {
-        this.pool.release(kryo);
+    Kryo getKryo() {
+        return kryoThreadLocal.get();
     }
 
     /**
@@ -776,12 +485,12 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
     boolean isEncrypted(final ByteBuf buffer) {
         // read off the magic byte
         byte magicByte = buffer.getByte(buffer.readerIndex());
-        return (magicByte & crypto) == crypto;
+        return (magicByte & KryoExtra.crypto) == KryoExtra.crypto;
     }
 
     /**
      * Waits until a kryo is available to write, using CAS operations to prevent having to synchronize.
-     * <p/>
+     * <p>
      * There is a small speed penalty if there were no kryo's available to use.
      */
     @Override
@@ -792,7 +501,7 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
     /**
      * Waits until a kryo is available to write, using CAS operations to prevent having to synchronize.
-     * <p/>
+     * <p>
      * There is a small speed penalty if there were no kryo's available to use.
      */
     @Override
@@ -803,221 +512,69 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
 
     /**
      * Reads an object from the buffer.
-     * <p/>
+     * <p>
      * Crypto + sequence number
      *
      * @param connection can be NULL
-     * @param length     should ALWAYS be the length of the expected object!
+     * @param length should ALWAYS be the length of the expected object!
      */
     @Override
     public final
     Object readWithCryptoTcp(final ConnectionImpl connection, final ByteBuf buffer, final int length) throws IOException {
-        return read0(connection, buffer, length, true);
+        return read0(connection, buffer, length);
     }
 
     /**
      * Reads an object from the buffer.
-     * <p/>
+     * <p>
      * Crypto + sequence number
      *
      * @param connection can be NULL
-     * @param length     should ALWAYS be the length of the expected object!
+     * @param length should ALWAYS be the length of the expected object!
      */
     @Override
     public final
     Object readWithCryptoUdp(final ConnectionImpl connection, final ByteBuf buffer, final int length) throws IOException {
-        return read0(connection, buffer, length, true);
+        return read0(connection, buffer, length);
     }
 
     /**
      * @param doCrypto true if we want to perform crypto on this data.
      */
     @SuppressWarnings("unchecked")
+
     private
     void write0(final ConnectionImpl connection, final ByteBuf buffer, final Object message, final boolean doCrypto) throws IOException {
-        final KryoExtra kryo = (KryoExtra) this.pool.takeUninterruptibly();
-        Logger logger2 = logger;
-
-        if (kryo == null) {
-            final String msg = "Unable to serialize buffer. Kryo was null (likely interrupted during pool.take())";
-            logger2.error(msg);
-            return;
-        }
+        final KryoExtra kryo = kryoThreadLocal.get();
 
         try {
-            byte magicByte = (byte) 0x00000000;
-
-            ByteBuf bufferWithData = kryo.tmpBuffer1;
-            ByteBuf bufferTempData = kryo.tmpBuffer2;
-
-            bufferWithData.clear();
-            bufferTempData.clear();
-
-            // write the object to the TEMP buffer! this will be compressed with snappy
-            kryo.output.setBuffer(bufferWithData);
-
-            // connection will ALWAYS be of type Connection or NULL.
-            // used by RMI/some serializers to determine which connection wrote this object
-            // NOTE: this is only valid in the context of this thread, which RMI stuff is accessed in -- so this is SAFE for RMI
-            kryo.connection = connection;
-
-            kryo.writeClassAndObject(kryo.output, message);
-
-            // save off how much data the object took + the length of the (possible) sequence.
-            int length = bufferWithData.writerIndex();  // it started at ZERO (since it's written to the temp buffer.
-
-            // snappy compression
-            //noinspection StatementWithEmptyBody
-            if (length > MIN_COMPRESSIBLE_LENGTH) {
-                if (ENABLE_SNAPPY) {
-                    snappyCompress(bufferWithData, bufferTempData, length, kryo.snappy);
-                }
-                else {
-                    compress(bufferWithData, bufferTempData, length, kryo.deflater);
-                }
-
-                // check to make sure that it was WORTH compressing, like what I had before
-                int compressedLength = bufferTempData.readableBytes();
-                if (compressedLength < length) {
-                    // specify we compressed data
-                    magicByte |= compression;
-
-                    length = compressedLength;
-
-                    // swap buffers
-                    ByteBuf tmp = bufferWithData;
-                    bufferWithData = bufferTempData;
-                    bufferTempData = tmp;
-                }
-                else {
-                    // "copy" (do nothing)
-                    bufferWithData.readerIndex(0); // have to reset the reader
-                }
-            }
-            else {
-                // "copy" (do nothing)
-            }
-
-            // AES CRYPTO
-            if (doCrypto && connection != null) {
-                if (logger2.isTraceEnabled()) {
-                    logger2.trace("Encrypting data with - AES {}", connection);
-                }
-
-                CryptoAES.encrypt(kryo.aesEngine, connection.getCryptoParameters(), bufferWithData, bufferTempData, length, logger);
-
-                // swap buffers
-                ByteBuf tmp = bufferWithData;
-                bufferWithData = bufferTempData;
-                bufferTempData = tmp;
-                bufferTempData.clear();
-
-                // only needed for server UDP connections to determine if the data is encrypted or not.
-                magicByte |= crypto;
-            }
-
-
-            // write out the "magic" byte.
-            buffer.writeByte(magicByte); // leave space for the magicByte
-
-            // transfer the tmpBuffer (if necessary) back into the "primary" buffer.
-            buffer.writeBytes(bufferWithData);
-
+            kryo.write(connection, buffer, message, doCrypto);
         } catch (Exception ex) {
             final String msg = "Unable to serialize buffer";
-            logger2.error(msg, ex);
+            logger.error(msg, ex);
             throw new IOException(msg, ex);
         } finally {
             // release resources
-            kryo.output.setBuffer(NULL_BUFFER);
-
-            // don't forget the clear the temp buffers!
-            kryo.tmpBuffer1.clear();
-            kryo.tmpBuffer2.clear();
-
-            this.pool.release(kryo);
+            kryo.releaseWrite();
         }
     }
 
-    /**
-     * @param doCrypto true if crypto was used for this data.
-     */
     @SuppressWarnings({"unchecked", "UnnecessaryLocalVariable"})
     private
-    Object read0(final ConnectionImpl connection, final ByteBuf buffer, final int length, final boolean doCrypto) throws IOException {
-        final KryoExtra kryo = (KryoExtra) this.pool.takeUninterruptibly();
-        Logger logger2 = logger;
+    Object read0(final ConnectionImpl connection, final ByteBuf buffer, final int length) throws IOException {
+        final KryoExtra kryo = kryoThreadLocal.get();
 
-        if (kryo == null) {
-            final String msg = "Unable to deserialize buffer. Kryo was null (likely interrupted during pool.take())";
-            logger2.error(msg);
-            return null;
-        }
-
-
-        int originalStartPos = 0;
+        int originalStartPos = buffer.readerIndex();
 
         ////////////////
-        // Note: we CANNOT write BACK to "buffer" since there could be additional data on it!
+        // Note: we CANNOT write BACK to the buffer since there could be additional data on it!
         ////////////////
         try {
-            // read off the magic byte
-            originalStartPos = buffer.readerIndex();
-            byte magicByte = buffer.readByte();
-
-            ByteBuf bufferWithData = buffer;
-            ByteBuf bufferTempData = kryo.tmpBuffer1;
-
-            // AES CRYPTO STUFF
-            if (doCrypto) {
-                if ((magicByte & crypto) != crypto) {
-                    throw new IOException("Unable to perform crypto when data does not use crypto!");
-                }
-
-                if (logger2.isTraceEnabled()) {
-                    logger2.trace("Decrypting data with - AES " + connection);
-                }
-
-                // length-1 to adjust for the magic byte
-                CryptoAES.decrypt(kryo.aesEngine, connection.getCryptoParameters(), bufferWithData, bufferTempData, length - 1, logger);
-
-                // correct which buffers are used
-                bufferWithData = bufferTempData;
-                bufferTempData = kryo.tmpBuffer2;
-            }
-
-            // did we compress it??
-            //noinspection StatementWithEmptyBody
-            if ((magicByte & compression) == compression) {
-                if (ENABLE_SNAPPY) {
-                    snappyDecompress(bufferWithData, bufferTempData, kryo.snappy);
-                }
-                else {
-                    decompress(bufferWithData, bufferTempData, kryo.inflater);
-                }
-
-                // correct which buffers are used
-                bufferWithData = bufferTempData;
-            }
-            else {
-                // "copy" (do nothing)
-            }
-
-            // read the object from the buffer.
-            kryo.input.setBuffer(bufferWithData);
-
-
-            // connection will ALWAYS be of type IConnection or NULL.
-            // used by RMI/some serializers to determine which connection read this object
-            // NOTE: this is only valid in the context of this thread, which RMI stuff is accessed in -- so this is SAFE for RMI
-            kryo.connection = connection;
-
-
-            Object object = kryo.readClassAndObject(kryo.input);
+            Object object = kryo.read(connection, buffer, length);
             return object;
         } catch (Exception ex) {
             final String msg = "Unable to deserialize buffer";
-            logger2.error(msg, ex);
+            logger.error(msg, ex);
             throw new IOException(msg, ex);
         } finally {
             // make sure the end of the buffer is in the correct spot.
@@ -1026,22 +583,7 @@ class KryoCryptoSerializationManager implements CryptoSerializationManager {
             buffer.readerIndex(originalStartPos + length);
 
             // release resources
-            kryo.input.setBuffer(NULL_BUFFER);
-
-            // don't forget the clear the temp buffers!
-            kryo.tmpBuffer1.clear();
-            kryo.tmpBuffer2.clear();
-
-            this.pool.release(kryo);
+            kryo.releaseRead();
         }
     }
-
-    // @formatter:off
-    private enum ChunkType {
-        COMPRESSED_DATA,
-        UNCOMPRESSED_DATA,
-        RESERVED_UNSKIPPABLE,
-        RESERVED_SKIPPABLE
-    }
-    // @formatter:on
 }
diff --git a/Dorkbox-Network/src/dorkbox/network/connection/KryoExtra.java b/Dorkbox-Network/src/dorkbox/network/connection/KryoExtra.java
index 6e2b06be..7635ea45 100644
--- a/Dorkbox-Network/src/dorkbox/network/connection/KryoExtra.java
+++ b/Dorkbox-Network/src/dorkbox/network/connection/KryoExtra.java
@@ -18,43 +18,428 @@ package dorkbox.network.connection;
 import com.esotericsoftware.kryo.Kryo;
 import dorkbox.network.pipeline.ByteBufInput;
 import dorkbox.network.pipeline.ByteBufOutput;
-import dorkbox.util.crypto.bouncycastle.GCMBlockCipher_ByteBuf;
+import dorkbox.util.bytes.BigEndian;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.handler.codec.compression.SnappyAccess;
+import net.jpountz.lz4.LZ4Compressor;
+import net.jpountz.lz4.LZ4Factory;
+import net.jpountz.lz4.LZ4FastDecompressor;
 import org.bouncycastle.crypto.engines.AESFastEngine;
+import org.bouncycastle.crypto.modes.GCMBlockCipher;
 
-import java.util.zip.Deflater;
-import java.util.zip.Inflater;
+import java.io.IOException;
 
 public
 class KryoExtra extends Kryo {
-    final ByteBufOutput output;
-    final ByteBufInput input;
 
-    final Inflater inflater;
-    final Deflater deflater;
-    final SnappyAccess snappy;
+    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(KryoExtra.class);
+    private static final ByteBuf nullByteBuf = (ByteBuf) null;
 
-    final ByteBuf tmpBuffer1;
-    final ByteBuf tmpBuffer2;
-    final GCMBlockCipher_ByteBuf aesEngine;
+    // this is the minimum size that LZ4 can compress. Anything smaller than this will result in an output LARGER than the input.
+    private static final int LZ4_COMPRESSION_MIN_SIZE = 32;
+
+    /**
+     * bit masks
+     */
+    private static final byte compress = (byte) 1;
+    static final byte crypto = (byte) (1 << 1);
+
+    private static LZ4Factory factory = LZ4Factory.fastestInstance();
+
+    // for kryo serialization
+    private final ByteBufInput reader = new ByteBufInput();
+    private final ByteBufOutput writer = new ByteBufOutput();
 
     // not thread safe
     public ConnectionImpl connection;
 
+    private final GCMBlockCipher aesEngine;
+
+    // writing data
+    private final ByteBuf tempBuffer = Unpooled.buffer(EndPoint.udpMaxSize);
+    private LZ4Compressor compressor = factory.fastCompressor();
+
+    private int compressOutputLength = -1;
+    private byte[] compressOutput;
+
+    private int cryptoOutputLength = -1;
+    private byte[] cryptoOutput;
+
+
+
+
+    // reading data
+    private LZ4FastDecompressor decompressor = factory.fastDecompressor();
+
+    private int decryptOutputLength = -1;
+    private byte[] decryptOutput;
+    private ByteBuf decryptBuf;
+
+    private int decompressOutputLength = -1;
+    private byte[] decompressOutput;
+    private ByteBuf decompressBuf;
+
+//    private static XXHashFactory hashFactory = XXHashFactory.fastestInstance();
 
     public
     KryoExtra() {
-        this.snappy = new SnappyAccess();
-        this.deflater = new Deflater(KryoCryptoSerializationManager.compressionLevel, true);
-        this.inflater = new Inflater(true);
+//        final Checksum checksumIn;
+//        final Checksum checksumOut;
+//        if (OS.is64bit()) {
+//            long seed = 0x3f79759cb27bf824L; // this has to stay the same
+//            checksumIn = hashFactory.newStreamingHash64(seed)
+//                                    .asChecksum();
+//            checksumOut = hashFactory.newStreamingHash64(seed)
+//                                     .asChecksum();
+//
+//        }
+//        else {
+//            int seed = 0xF51BA30E; // this has to stay the same
+//            checksumIn = hashFactory.newStreamingHash32(seed)
+//                                    .asChecksum();
+//            checksumOut = hashFactory.newStreamingHash32(seed)
+//                                     .asChecksum();
+//        }
+//
 
-        this.input = new ByteBufInput();
-        this.output = new ByteBufOutput();
 
-        this.tmpBuffer1 = Unpooled.buffer(1024);
-        this.tmpBuffer2 = Unpooled.buffer(1024);
-        this.aesEngine = new GCMBlockCipher_ByteBuf(new AESFastEngine());
+//        > Our data is partitionaned into 64 blocks which are compressed separetely.
+//
+//        64 bytes blocks ? This would be much too small.
+//
+//
+//        > Would a larger block size improve the compression ratio?
+//
+//        Probably.
+//        When compressing data using independent blocks, you can observe compression ratio gains up to 1 MB blocks.
+//        Beyond that, benefits become less ans less visible.
+//
+//
+//        > Is there an optimal block size?
+//
+//        For independent blocks, 64KB is considered optimal "small" size.
+//        Between 4 KB & 64 KB, it is "very small", but still manageable.
+//        Anything below 4 KB is starting to miss a lost of compression opportunity. Ratio will plummet.
+//
+//
+//        > Does anybody have recommendations on how to improve the compression ration?
+//
+//        Anytime you input data consists of
+//        tables with fixed length cells,
+//                        blosc should be tested : if offers big opportunities for compression savings.
+
+
+//        ➜  leveldb git:(lz4) ✗ ./db_bench
+//        LevelDB:    version 1.7
+//        Keys:       16 bytes each
+//        Values:     100 bytes each (50 bytes after compression)
+//        Entries:    1000000
+//        RawSize:    110.6 MB (estimated)
+//        FileSize:   62.9 MB (estimated)
+//                        ------------------------------------------------
+//        snappycomp   :       7.534 micros/op;  518.5 MB/s (output: 55.1%)
+//        snappyuncomp :       1.391 micros/op; 2808.1 MB/s
+//        lz4comp      :       6.210 micros/op;  629.0 MB/s (output: 55.4%)
+//        lz4uncomp    :       0.641 micros/op; 6097.9 MB/s
+
+        this.aesEngine = new GCMBlockCipher(new AESFastEngine());
+    }
+
+    public
+    void write(final ConnectionImpl connection, final ByteBuf buffer, final Object message, final boolean doCrypto) throws IOException {
+        // do we compress + crypto the data?  (don't always need it, specifically during connection INIT)
+        if (connection == null || !doCrypto) {
+            // magic byte
+            buffer.writeByte(0);
+
+            // write the object to the NORMAL output buffer!
+            writer.setBuffer(buffer);
+
+            // connection will ALWAYS be of type Connection or NULL.
+            // used by RMI/some serializers to determine which connection wrote this object
+            // NOTE: this is only valid in the context of this thread, which RMI stuff is accessed in -- so this is SAFE for RMI
+            this.connection = connection;
+            writeClassAndObject(writer, message);
+        } else {
+            byte magicByte = (byte) 0x00000000;
+
+            ByteBuf objectOutputBuffer = this.tempBuffer;
+            objectOutputBuffer.clear(); // always have to reset everything
+
+            // write the object to a TEMP buffer! this will be compressed
+            writer.setBuffer(objectOutputBuffer);
+
+            // connection will ALWAYS be of type Connection or NULL.
+            // used by RMI/some serializers to determine which connection wrote this object
+            // NOTE: this is only valid in the context of this thread, which RMI stuff is accessed in -- so this is SAFE for RMI
+            this.connection = connection;
+
+            writeClassAndObject(writer, message);
+
+            // save off how much data the object took + magic byte
+            int length = objectOutputBuffer.writerIndex();
+
+
+            // NOTE: compression and encryption MUST work with byte[] because they use JNI!
+            // Realistically, it is impossible to get the backing arrays out of a Heap Buffer once they are resized and begin to use
+            // sliced. It's lame that there is a "double copy" of bytes here, but I don't know how to avoid it...
+            // see:   https://stackoverflow.com/questions/19296386/netty-java-getting-data-from-bytebuf
+
+            byte[] inputArray;
+            int inputOffset;
+
+            //noinspection Duplicates
+            if (objectOutputBuffer.hasArray()) {
+                // Even if a ByteBuf has a backing array (i.e. buf.hasArray() returns true), the following isn't necessarily true because
+                // the buffer might be a slice of other buffer or a pooled buffer:
+
+                if (objectOutputBuffer.array()[0] == objectOutputBuffer.getByte(0) &&
+                    objectOutputBuffer.array().length == objectOutputBuffer.capacity()) {
+
+                    // we can use it...
+                    inputArray = objectOutputBuffer.array();
+                    inputOffset = objectOutputBuffer.arrayOffset();
+                } else {
+                    // we can NOT use it.
+                    inputArray = new byte[length];
+                    objectOutputBuffer.getBytes(objectOutputBuffer.readerIndex(), inputArray);
+                    inputOffset = 0;
+                }
+            } else {
+                inputArray = new byte[length];
+                objectOutputBuffer.getBytes(objectOutputBuffer.readerIndex(), inputArray);
+                inputOffset = 0;
+            }
+
+            if (length > LZ4_COMPRESSION_MIN_SIZE) {
+                if (logger.isTraceEnabled()) {
+                    logger.trace("Compressing data {}", connection);
+                }
+
+                byte[] compressOutput = this.compressOutput;
+
+                int maxCompressedLength = compressor.maxCompressedLength(length);
+
+                // add 4 so there is room to write the compressed size to the buffer
+                int maxCompressedLengthWithOffset = maxCompressedLength + 4;
+
+                // lazy initialize the compression output buffer
+                if (maxCompressedLengthWithOffset > compressOutputLength) {
+                    compressOutputLength = maxCompressedLengthWithOffset;
+                    compressOutput = new byte[maxCompressedLengthWithOffset];
+                    this.compressOutput = compressOutput;
+                }
+
+
+                // LZ4 compress. output offset 4 to leave room for length of tempOutput data
+                int compressedLength = compressor.compress(inputArray, inputOffset, length, compressOutput, 4, maxCompressedLength);
+
+                // bytes can now be written to, because our compressed data is stored in a temp array.
+                // ONLY do this if our compressed length is LESS than our uncompressed length.
+                if (compressedLength < length) {
+                    // now write the ORIGINAL (uncompressed) length to the front of the byte array. This is so we can use the FAST version
+                    BigEndian.Int_.toBytes(length, compressOutput);
+
+                    magicByte |= compress;
+
+                    // corrected length
+                    length = compressedLength + 4; // +4 for the uncompressed size bytes
+
+                    // correct input.  compression output is now encryption input
+                    inputArray = compressOutput;
+                    inputOffset = 0;
+                }
+            }
+
+
+            if (logger.isTraceEnabled()) {
+                logger.trace("AES encrypting data {}", connection);
+            }
+
+
+            final GCMBlockCipher aes = this.aesEngine;
+
+            // TODO: AES IV must be a NONCE! can use an initial IV, and then have a counter or something with a hash of counter+initialIV to make the new IV
+            aes.reset();
+            aes.init(true, connection.getCryptoParameters());
+
+            byte[] cryptoOutput = this.cryptoOutput;
+
+            // lazy initialize the crypto output buffer
+            int cryptoSize = aes.getOutputSize(length);
+
+            // 'output' is the temp byte array
+            if (cryptoSize > cryptoOutputLength) {
+                cryptoOutputLength = cryptoSize;
+                cryptoOutput = new byte[cryptoSize];
+                this.cryptoOutput = cryptoOutput;
+            }
+
+            //System.err.println("out orig: " + length + "  first/last " + inputArray[inputOffset] + " " + inputArray[length-1]);
+            int actualLength = aes.processBytes(inputArray, inputOffset, length, cryptoOutput, 0);
+
+            try {
+                // authentication tag for GCM
+                actualLength += aes.doFinal(cryptoOutput, actualLength);
+            } catch (Exception e) {
+                throw new IOException("Unable to AES encrypt the data", e);
+            }
+
+            magicByte |= crypto;
+
+            // write out the "magic" byte.
+            buffer.writeByte(magicByte);
+
+            // have to copy over the orig data, because we used the temp buffer
+            buffer.writeBytes(cryptoOutput, 0, actualLength);
+        }
+    }
+
+    @SuppressWarnings("Duplicates")
+    public
+    Object read(final ConnectionImpl connection, final ByteBuf buffer, int length) throws IOException {
+        // we cannot use the buffer as a "temp" storage, because there is other data on it
+
+        // read off the magic byte
+        final byte magicByte = buffer.readByte();
+        ByteBuf inputBuf = buffer;
+
+
+        // compression can ONLY happen if it's ALSO crypto'd
+        if ((magicByte & crypto) == crypto) {
+            // have to adjust for the magic byte
+            length -= 1;
+
+            // it's ALWAYS encrypted, but MAYBE compressed
+            final boolean isCompressed = (magicByte & compress) == compress;
+
+
+            if (logger.isTraceEnabled()) {
+                logger.trace("AES decrypting data {}", connection);
+            }
+
+
+            // NOTE: compression and encryption MUST work with byte[] because they use JNI!
+            // Realistically, it is impossible to get the backing arrays out of a Heap Buffer once they are resized and begin to use
+            // sliced. It's lame that there is a "double copy" of bytes here, but I don't know how to avoid it...
+            // see:   https://stackoverflow.com/questions/19296386/netty-java-getting-data-from-bytebuf
+
+            byte[] inputArray;
+            int inputOffset;
+
+            if (inputBuf.hasArray()) {
+                // Even if a ByteBuf has a backing array (i.e. buf.hasArray() returns true), the following isn't necessarily true because
+                // the buffer might be a slice of other buffer or a pooled buffer:
+
+                if (inputBuf.array()[0] == inputBuf.getByte(0) &&
+                    inputBuf.array().length == inputBuf.capacity()) {
+
+                    // we can use it...
+                    inputArray = inputBuf.array();
+                    inputOffset = inputBuf.arrayOffset();
+                } else {
+                    // we can NOT use it.
+                    inputArray = new byte[length];
+                    inputBuf.getBytes(inputBuf.readerIndex(), inputArray);
+                    inputOffset = 0;
+                }
+            } else {
+                inputArray = new byte[length];
+                inputBuf.getBytes(inputBuf.readerIndex(), inputArray);
+                inputOffset = 0;
+            }
+
+
+            final GCMBlockCipher aes = this.aesEngine;
+            aes.reset();
+            aes.init(false, connection.getCryptoParameters());
+
+            int cryptoSize = aes.getOutputSize(length);
+
+            // lazy initialize the decrypt output buffer
+            byte[] decryptOutputArray = this.decryptOutput;
+            if (cryptoSize > decryptOutputLength) {
+                decryptOutputLength = cryptoSize;
+                decryptOutputArray = new byte[cryptoSize];
+                this.decryptOutput = decryptOutputArray;
+
+                decryptBuf = Unpooled.wrappedBuffer(decryptOutputArray);
+            }
+
+            int decryptedLength = aes.processBytes(inputArray, inputOffset, length, decryptOutputArray, 0);
+
+            try {
+                // authentication tag for GCM (since we are DECRYPTING, the original 'actualLength' is the correct one.)
+                decryptedLength += aes.doFinal(decryptOutputArray, decryptedLength);
+            } catch (Exception e) {
+                throw new IOException("Unable to AES decrypt the data", e);
+            }
+
+            if (isCompressed) {
+                inputArray = decryptOutputArray;
+                inputOffset = 4; // because 4 bytes for the decompressed size
+
+                // get the decompressed length (at the beginning of the array)
+                final int uncompressedLength = BigEndian.Int_.from(inputArray);
+
+                byte[] decompressOutputArray = this.decompressOutput;
+                if (uncompressedLength > decompressOutputLength) {
+                    decompressOutputLength = uncompressedLength;
+                    decompressOutputArray = new byte[uncompressedLength];
+                    this.decompressOutput = decompressOutputArray;
+
+                    decompressBuf = Unpooled.wrappedBuffer(decompressOutputArray);  // so we can read via kryo
+                }
+
+                // LZ4 decompress, requires the size of the ORIGINAL length (because we use the FAST decompressor
+                decompressor.decompress(inputArray, inputOffset, decompressOutputArray, 0, uncompressedLength);
+
+                decompressBuf.setIndex(0, uncompressedLength);
+                inputBuf = decompressBuf;
+            }
+            else {
+                decryptBuf.setIndex(0, decryptedLength);
+                inputBuf = decryptBuf;
+            }
+        }
+
+
+        // read the object from the buffer.
+        reader.setBuffer(inputBuf);
+
+        // connection will ALWAYS be of type IConnection or NULL.
+        // used by RMI/some serializers to determine which connection read this object
+        // NOTE: this is only valid in the context of this thread, which RMI stuff is accessed in -- so this is SAFE for RMI
+        this.connection = connection;
+
+        return readClassAndObject(reader);
+    }
+
+    public final
+    void releaseWrite() {
+        // release/reset connection resources
+        writer.setBuffer(nullByteBuf);
+        connection = null;
+    }
+
+    public final
+    void releaseRead() {
+        // release/reset connection resources
+        reader.setBuffer(nullByteBuf);
+        connection = null;
+    }
+
+    @Override
+    protected
+    void finalize() throws Throwable {
+        if (decompressBuf != null) {
+            decompressBuf.release();
+        }
+
+        if (decryptBuf != null) {
+            decryptBuf.release();
+        }
+
+        super.finalize();
     }
 }
diff --git a/Dorkbox-Network/src/dorkbox/network/rmi/InvocationHandlerSerializer.java b/Dorkbox-Network/src/dorkbox/network/rmi/InvocationHandlerSerializer.java
new file mode 100644
index 00000000..6d22539a
--- /dev/null
+++ b/Dorkbox-Network/src/dorkbox/network/rmi/InvocationHandlerSerializer.java
@@ -0,0 +1,42 @@
+package dorkbox.network.rmi;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import dorkbox.network.connection.KryoExtra;
+import org.slf4j.Logger;
+
+import java.lang.reflect.Proxy;
+
+public
+class InvocationHandlerSerializer extends Serializer<Object> {
+    private final org.slf4j.Logger logger;
+
+    public
+    InvocationHandlerSerializer(final Logger logger) {
+        this.logger = logger;
+    }
+
+    @Override
+    public
+    void write(Kryo kryo, Output output, Object object) {
+        RemoteInvocationHandler handler = (RemoteInvocationHandler) Proxy.getInvocationHandler(object);
+        output.writeInt(handler.objectID, true);
+    }
+
+    @Override
+    @SuppressWarnings({"unchecked", "AutoBoxing"})
+    public
+    Object read(Kryo kryo, Input input, Class<Object> type) {
+        int objectID = input.readInt(true);
+
+        KryoExtra kryoExtra = (KryoExtra) kryo;
+        Object object = kryoExtra.connection.getImplementationObject(objectID);
+
+        if (object == null) {
+            logger.error("Unknown object ID in RMI ObjectSpace: {}", objectID);
+        }
+        return object;
+    }
+}
diff --git a/Dorkbox-Network/src/dorkbox/network/rmi/InvocationResultSerializer.java b/Dorkbox-Network/src/dorkbox/network/rmi/InvocationResultSerializer.java
new file mode 100644
index 00000000..07ef69aa
--- /dev/null
+++ b/Dorkbox-Network/src/dorkbox/network/rmi/InvocationResultSerializer.java
@@ -0,0 +1,30 @@
+package dorkbox.network.rmi;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.serializers.FieldSerializer;
+import dorkbox.network.connection.KryoExtra;
+
+public
+class InvocationResultSerializer extends FieldSerializer<InvokeMethodResult> {
+    public
+    InvocationResultSerializer(final KryoExtra kryo) {
+        super(kryo, InvokeMethodResult.class);
+    }
+
+    @Override
+    public
+    void write(Kryo kryo, Output output, InvokeMethodResult result) {
+        super.write(kryo, output, result);
+        output.writeInt(result.objectID, true);
+    }
+
+    @Override
+    public
+    InvokeMethodResult read(Kryo kryo, Input input, Class<InvokeMethodResult> type) {
+        InvokeMethodResult result = super.read(kryo, input, type);
+        result.objectID = input.readInt(true);
+        return result;
+    }
+}
diff --git a/Dorkbox-Network/src/dorkbox/network/rmi/RemoteInvocationHandler.java b/Dorkbox-Network/src/dorkbox/network/rmi/RemoteInvocationHandler.java
index c432dbe3..908ed140 100644
--- a/Dorkbox-Network/src/dorkbox/network/rmi/RemoteInvocationHandler.java
+++ b/Dorkbox-Network/src/dorkbox/network/rmi/RemoteInvocationHandler.java
@@ -35,7 +35,6 @@
 package dorkbox.network.rmi;
 
 
-import com.esotericsoftware.kryo.Kryo;
 import dorkbox.network.connection.Connection;
 import dorkbox.network.connection.EndPoint;
 import dorkbox.network.connection.ListenerRaw;
@@ -204,17 +203,8 @@ class RemoteInvocationHandler implements InvocationHandler {
         invokeMethod.args = args;
 
 
-        // thread safe access.
-        final Kryo kryo = serializationManager.take();
-        if (kryo == null) {
-            String msg = "Interrupted during kryo pool.take()";
-            logger1.error(msg);
-            return msg;
-        }
-
         // which method do we access?
-        CachedMethod[] cachedMethods = CachedMethod.getMethods(kryo, method.getDeclaringClass());
-        serializationManager.release(kryo);
+        CachedMethod[] cachedMethods = CachedMethod.getMethods(serializationManager.getKryo(), method.getDeclaringClass());
         for (int i = 0, n = cachedMethods.length; i < n; i++) {
             CachedMethod cachedMethod = cachedMethods[i];
             Method checkMethod = cachedMethod.origMethod;
diff --git a/Dorkbox-Network/test/dorkbox/network/rmi/RmiGlobalTest.java b/Dorkbox-Network/test/dorkbox/network/rmi/RmiGlobalTest.java
index 066eca24..9bf16a21 100644
--- a/Dorkbox-Network/test/dorkbox/network/rmi/RmiGlobalTest.java
+++ b/Dorkbox-Network/test/dorkbox/network/rmi/RmiGlobalTest.java
@@ -197,7 +197,6 @@ class RmiGlobalTest extends BaseTest {
 
                       stopEndPoints(2000);
                   }
-
               });
 
 
diff --git a/Dorkbox-Network/test/dorkbox/network/rmi/RmiSendObjectTest.java b/Dorkbox-Network/test/dorkbox/network/rmi/RmiSendObjectTest.java
index 89880fb0..1d8d24e9 100644
--- a/Dorkbox-Network/test/dorkbox/network/rmi/RmiSendObjectTest.java
+++ b/Dorkbox-Network/test/dorkbox/network/rmi/RmiSendObjectTest.java
@@ -78,8 +78,6 @@ class RmiSendObjectTest extends BaseTest {
                           @Override
                           public
                           void run() {
-
-
                               TestObject test = null;
                               try {
                                   test = connection.createProxyObject(TestObjectImpl.class);