Optimize native handling further.

We now try to work within the boundaries given by the native. In the
case of Java natives, we work with byte arrays. With natives, always use
direct buffers.

However, the numbers do favor the natives, since they work with direct
byte buffers, without any copying. For the most part, this commit is
intended to improve the lives of Velocity users on Windows.

native/src/main/c/jni_zlib_deflate.c

@@ -85,7 +85,7 @@ Java_com_velocitypowered_natives_compression_NativeZlibDeflate_process(JNIEnv *e
     jint sourceLength,
     jlong destinationAddress,
     jint destinationLength,
-    jboolean flush)
+    jboolean finish)
 {
     z_stream* stream = (z_stream*) ctx;
     stream->next_in = (Bytef *) sourceAddress;
@@ -93,7 +93,7 @@ Java_com_velocitypowered_natives_compression_NativeZlibDeflate_process(JNIEnv *e
     stream->avail_in = sourceLength;
     stream->avail_out = destinationLength;
 
-    int res = deflate(stream, flush ? Z_FINISH : Z_NO_FLUSH);
+    int res = deflate(stream, finish ? Z_FINISH : Z_NO_FLUSH);
     switch (res) {
         case Z_STREAM_END:
             // The stream has ended.

native/src/main/java/com/velocitypowered/natives/compression/JavaVelocityCompressor.java

@@ -27,9 +27,14 @@ public class JavaVelocityCompressor implements VelocityCompressor {
   public void inflate(ByteBuf source, ByteBuf destination) throws DataFormatException {
     ensureNotDisposed();
 
-    byte[] inData = new byte[source.readableBytes()];
-    source.readBytes(inData);
-    inflater.setInput(inData);
+    if (source.hasArray()) {
+      inflater.setInput(source.array(), source.arrayOffset(), source.readableBytes());
+    } else {
+      byte[] inData = new byte[source.readableBytes()];
+      source.readBytes(inData);
+      inflater.setInput(inData);
+    }
+
     while (!inflater.finished()) {
       int read = inflater.inflate(buf);
       destination.writeBytes(buf, 0, read);
@@ -41,9 +46,13 @@ public class JavaVelocityCompressor implements VelocityCompressor {
   public void deflate(ByteBuf source, ByteBuf destination) throws DataFormatException {
     ensureNotDisposed();
 
-    byte[] inData = new byte[source.readableBytes()];
-    source.readBytes(inData);
-    deflater.setInput(inData);
+    if (source.hasArray()) {
+      deflater.setInput(source.array(), source.arrayOffset(), source.readableBytes());
+    } else {
+      byte[] inData = new byte[source.readableBytes()];
+      source.readBytes(inData);
+      deflater.setInput(inData);
+    }
     deflater.finish();
     while (!deflater.finished()) {
       int bytes = deflater.deflate(buf);

native/src/main/java/com/velocitypowered/natives/compression/NativeZlibDeflate.java

@@ -13,8 +13,7 @@ class NativeZlibDeflate {
   native long free(long ctx);
 
   native int process(long ctx, long sourceAddress, int sourceLength, long destinationAddress,
-      int destinationLength,
-      boolean flush);
+      int destinationLength, boolean finish);
 
   native void reset(long ctx);
 

native/src/main/java/com/velocitypowered/natives/encryption/JavaVelocityCipher.java

@@ -2,6 +2,7 @@ package com.velocitypowered.natives.encryption;
 
 import com.google.common.base.Preconditions;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import java.security.GeneralSecurityException;
 import javax.crypto.Cipher;
@@ -22,7 +23,7 @@ public class JavaVelocityCipher implements VelocityCipher {
       return new JavaVelocityCipher(false, key);
     }
   };
-  private static final int INITIAL_BUFFER_SIZE = 1024 * 16;
+  private static final int INITIAL_BUFFER_SIZE = 1024 * 8;
   private static final ThreadLocal<byte[]> inBufLocal = ThreadLocal.withInitial(
       () -> new byte[INITIAL_BUFFER_SIZE]);
 
@@ -40,12 +41,19 @@ public class JavaVelocityCipher implements VelocityCipher {
     ensureNotDisposed();
 
     int inBytes = source.readableBytes();
-    byte[] inBuf = slurp(source);
+    ByteBuf asHeapBuf = asHeapBuf(source);
 
     int outputSize = cipher.getOutputSize(inBytes);
-    byte[] outBuf = new byte[outputSize];
-    cipher.update(inBuf, 0, inBytes, outBuf);
-    destination.writeBytes(outBuf);
+    if (!destination.hasArray()) {
+      byte[] outBuf = new byte[outputSize];
+      cipher.update(asHeapBuf.array(), asHeapBuf.arrayOffset(), inBytes, outBuf);
+      destination.writeBytes(outBuf);
+    } else {
+      // If the destination we write to is an array, we can use the backing array directly.
+      destination.ensureWritable(outputSize);
+      destination.writerIndex(cipher.update(asHeapBuf.array(), asHeapBuf.arrayOffset(), inBytes,
+          destination.array(), destination.arrayOffset()));
+    }
   }
 
   @Override
@@ -53,14 +61,20 @@ public class JavaVelocityCipher implements VelocityCipher {
     ensureNotDisposed();
 
     int inBytes = source.readableBytes();
-    byte[] inBuf = slurp(source);
+    ByteBuf asHeapBuf = asHeapBuf(source);
 
     ByteBuf out = ctx.alloc().heapBuffer(cipher.getOutputSize(inBytes));
-    out.writerIndex(cipher.update(inBuf, 0, inBytes, out.array(), out.arrayOffset()));
+    out.writerIndex(cipher.update(asHeapBuf.array(), asHeapBuf.arrayOffset(), inBytes, out.array(),
+        out.arrayOffset()));
     return out;
   }
 
-  private static byte[] slurp(ByteBuf source) {
+  private static ByteBuf asHeapBuf(ByteBuf source) {
+    if (source.hasArray()) {
+      // If this byte buffer is backed by an array, we can just use this buffer directly.
+      return source;
+    }
+
     int inBytes = source.readableBytes();
     byte[] inBuf = inBufLocal.get();
     if (inBuf.length <= inBytes) {
@@ -68,7 +82,7 @@ public class JavaVelocityCipher implements VelocityCipher {
       inBufLocal.set(inBuf);
     }
     source.readBytes(inBuf, 0, inBytes);
-    return inBuf;
+    return Unpooled.wrappedBuffer(inBuf, 0, inBytes);
   }
 
   @Override

native/src/main/java/com/velocitypowered/natives/util/MoreByteBufUtils.java

@@ -30,4 +30,31 @@ public class MoreByteBufUtils {
     newBuf.writeBytes(buf);
     return newBuf;
   }
+
+  /**
+   * Creates a {@link ByteBuf} that will have the best performance with the specified
+   * {@code nativeStuff}.
+   *
+   * @param alloc the {@link ByteBufAllocator} to use
+   * @param nativeStuff the native we are working with
+   * @return a buffer compatible with the native
+   */
+  public static ByteBuf preferredBuffer(ByteBufAllocator alloc, Native nativeStuff) {
+    return nativeStuff.isNative() ? alloc.directBuffer() : alloc.heapBuffer();
+  }
+
+  /**
+   * Creates a {@link ByteBuf} that will have the best performance with the specified
+   * {@code nativeStuff}.
+   *
+   * @param alloc the {@link ByteBufAllocator} to use
+   * @param nativeStuff the native we are working with
+   * @param initialCapacity the initial capacity to allocate
+   * @return a buffer compatible with the native
+   */
+  public static ByteBuf preferredBuffer(ByteBufAllocator alloc, Native nativeStuff,
+      int initialCapacity) {
+    return nativeStuff.isNative() ? alloc.directBuffer(initialCapacity) : alloc
+        .heapBuffer(initialCapacity);
+  }
 }

native/src/test/java/com/velocitypowered/natives/compression/VelocityCompressorTest.java

@@ -9,7 +9,9 @@ import com.velocitypowered.natives.util.Natives;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
+import java.io.IOException;
 import java.util.Random;
+import java.util.function.Supplier;
 import java.util.zip.DataFormatException;
 import java.util.zip.Deflater;
 import org.junit.jupiter.api.BeforeAll;
@@ -18,9 +20,12 @@ import org.junit.jupiter.api.condition.EnabledOnOs;
 
 class VelocityCompressorTest {
 
+  private static byte[] TEST_DATA = new byte[1 << 14];
+
   @BeforeAll
-  static void checkNatives() {
+  static void checkNatives() throws IOException {
     Natives.compress.getLoadedVariant();
+    new Random(1).nextBytes(TEST_DATA);
   }
 
   @Test
@@ -31,25 +36,30 @@ class VelocityCompressorTest {
       compressor.dispose();
       fail("Loaded regular compressor");
     }
-    check(compressor);
+    check(compressor, () -> Unpooled.directBuffer(TEST_DATA.length));
   }
 
   @Test
-  void javaIntegrityCheck() throws DataFormatException {
+  void javaIntegrityCheckDirect() throws DataFormatException {
     VelocityCompressor compressor = JavaVelocityCompressor.FACTORY
         .create(Deflater.DEFAULT_COMPRESSION);
-    check(compressor);
+    check(compressor, () -> Unpooled.directBuffer(TEST_DATA.length));
   }
 
-  private void check(VelocityCompressor compressor) throws DataFormatException {
-    ByteBuf source = Unpooled.directBuffer();
-    ByteBuf dest = Unpooled.directBuffer();
-    ByteBuf decompressed = Unpooled.directBuffer();
+  @Test
+  void javaIntegrityCheckHeap() throws DataFormatException {
+    VelocityCompressor compressor = JavaVelocityCompressor.FACTORY
+        .create(Deflater.DEFAULT_COMPRESSION);
+    check(compressor, () -> Unpooled.buffer(TEST_DATA.length));
+  }
 
-    Random random = new Random(1);
-    byte[] randomBytes = new byte[1 << 16];
-    random.nextBytes(randomBytes);
-    source.writeBytes(randomBytes);
+  private void check(VelocityCompressor compressor, Supplier<ByteBuf> bufSupplier)
+      throws DataFormatException {
+    ByteBuf source = bufSupplier.get();
+    ByteBuf dest = bufSupplier.get();
+    ByteBuf decompressed = bufSupplier.get();
+
+    source.writeBytes(TEST_DATA);
 
     try {
       compressor.deflate(source, dest);

native/src/test/java/com/velocitypowered/natives/encryption/VelocityCipherTest.java

@@ -9,6 +9,7 @@ import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import java.security.GeneralSecurityException;
 import java.util.Random;
+import java.util.function.Supplier;
 import javax.crypto.spec.SecretKeySpec;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Disabled;
@@ -16,11 +17,16 @@ import org.junit.jupiter.api.Test;
 
 class VelocityCipherTest {
 
-  private static final int ENCRYPT_DATA_SIZE = 1 << 16;
+  private static final int ENCRYPT_DATA_SIZE = 1 << 14;
+  private static byte[] TEST_DATA = new byte[ENCRYPT_DATA_SIZE];;
+  private static final byte[] AES_KEY = new byte[16];
 
   @BeforeAll
   static void checkNatives() {
     Natives.cipher.getLoadedVariant();
+    Random random = new Random(1);
+    random.nextBytes(TEST_DATA);
+    random.nextBytes(AES_KEY);
   }
 
   @Test
@@ -30,30 +36,30 @@ class VelocityCipherTest {
     if (factory == JavaVelocityCipher.FACTORY) {
       fail("Loaded regular cipher");
     }
-    check(factory);
+    check(factory, Unpooled::directBuffer);
   }
 
   @Test
-  void javaIntegrityCheck() throws GeneralSecurityException {
-    check(JavaVelocityCipher.FACTORY);
+  void javaIntegrityCheckDirect() throws GeneralSecurityException {
+    check(JavaVelocityCipher.FACTORY, Unpooled::directBuffer);
   }
 
-  private void check(VelocityCipherFactory factory) throws GeneralSecurityException {
+  @Test
+  void javaIntegrityCheckHeap() throws GeneralSecurityException {
+    check(JavaVelocityCipher.FACTORY, Unpooled::buffer);
+  }
+
+  private void check(VelocityCipherFactory factory, Supplier<ByteBuf> bufSupplier)
+      throws GeneralSecurityException {
     // Generate a random 16-byte key.
-    Random random = new Random(1);
-    byte[] key = new byte[16];
-    random.nextBytes(key);
+    VelocityCipher decrypt = factory.forDecryption(new SecretKeySpec(AES_KEY, "AES"));
+    VelocityCipher encrypt = factory.forEncryption(new SecretKeySpec(AES_KEY, "AES"));
 
-    VelocityCipher decrypt = factory.forDecryption(new SecretKeySpec(key, "AES"));
-    VelocityCipher encrypt = factory.forEncryption(new SecretKeySpec(key, "AES"));
+    ByteBuf source = bufSupplier.get();
+    ByteBuf dest = bufSupplier.get();
+    ByteBuf decryptionBuf = bufSupplier.get();
 
-    ByteBuf source = Unpooled.directBuffer(ENCRYPT_DATA_SIZE);
-    ByteBuf dest = Unpooled.directBuffer(ENCRYPT_DATA_SIZE);
-    ByteBuf decryptionBuf = Unpooled.directBuffer(ENCRYPT_DATA_SIZE);
-
-    byte[] randomBytes = new byte[ENCRYPT_DATA_SIZE];
-    random.nextBytes(randomBytes);
-    source.writeBytes(randomBytes);
+    source.writeBytes(TEST_DATA);
 
     try {
       encrypt.process(source, dest);

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftCompressDecoder.java

@@ -1,9 +1,10 @@
 package com.velocitypowered.proxy.protocol.netty;
 
+import static com.velocitypowered.natives.util.MoreByteBufUtils.ensureCompatible;
+import static com.velocitypowered.natives.util.MoreByteBufUtils.preferredBuffer;
 import static com.velocitypowered.proxy.protocol.util.NettyPreconditions.checkFrame;
 
 import com.velocitypowered.natives.compression.VelocityCompressor;
-import com.velocitypowered.natives.util.MoreByteBufUtils;
 import com.velocitypowered.proxy.protocol.ProtocolUtils;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -35,9 +36,9 @@ public class MinecraftCompressDecoder extends MessageToMessageDecoder<ByteBuf> {
     checkFrame(expectedUncompressedSize >= threshold,
         "Uncompressed size %s is greater than threshold %s",
         expectedUncompressedSize, threshold);
-    ByteBuf compatibleIn = MoreByteBufUtils.ensureCompatible(ctx.alloc(), compressor, in);
-    ByteBuf uncompressed = ctx.alloc().directBuffer(Math.min(expectedUncompressedSize,
-        MAXIMUM_INITIAL_BUFFER_SIZE));
+    ByteBuf compatibleIn = ensureCompatible(ctx.alloc(), compressor, in);
+    int initialCapacity = Math.min(expectedUncompressedSize, MAXIMUM_INITIAL_BUFFER_SIZE);
+    ByteBuf uncompressed = preferredBuffer(ctx.alloc(), compressor, initialCapacity);
     try {
       compressor.inflate(compatibleIn, uncompressed);
       checkFrame(expectedUncompressedSize == uncompressed.readableBytes(),

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftCompressEncoder.java

@@ -38,11 +38,9 @@ public class MinecraftCompressEncoder extends MessageToByteEncoder<ByteBuf> {
   @Override
   protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, ByteBuf msg, boolean preferDirect)
       throws Exception {
-    if (msg.readableBytes() <= threshold) {
-      return ctx.alloc().directBuffer(msg.readableBytes() + 1);
-    }
-    // A reasonable assumption about compression savings
-    return ctx.alloc().directBuffer(msg.readableBytes() / 3);
+    int initialBufferSize = msg.readableBytes() <= threshold ? msg.readableBytes() + 1 :
+        msg.readableBytes() / 3;
+    return MoreByteBufUtils.preferredBuffer(ctx.alloc(), compressor, initialBufferSize);
   }
 
   @Override