Use Netty ByteBuf Bulk Operations for Faster Deserialization (#40158) (#40339)

* Use bulk methods to read numbers faster from byte buffers

Author: original-brownbear

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/ByteBufStreamInput.java

@@ -28,7 +28,7 @@
 import java.io.IOException;
 
 /**
- * A Netty {@link io.netty.buffer.ByteBuf} based {@link org.elasticsearch.common.io.stream.StreamInput}.
+ * A Netty {@link ByteBuf} based {@link StreamInput}.
  */
 class ByteBufStreamInput extends StreamInput {
 
@@ -109,6 +109,39 @@ public int read(byte[] b, int off, int len) throws IOException {
         return len;
     }
 
+    @Override
+    public short readShort() throws IOException {
+        try {
+            return buffer.readShort();
+        } catch (IndexOutOfBoundsException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        try {
+            return buffer.readInt();
+        } catch (IndexOutOfBoundsException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        try {
+            return buffer.readLong();
+        } catch (IndexOutOfBoundsException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
     @Override
     public void reset() throws IOException {
         buffer.resetReaderIndex();

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/ByteBufUtils.java

@@ -226,6 +226,39 @@ public int read(byte[] b, int off, int len) throws IOException {
             return len;
         }
 
+        @Override
+        public short readShort() throws IOException {
+            try {
+                return buffer.readShort();
+            } catch (IndexOutOfBoundsException ex) {
+                EOFException eofException = new EOFException();
+                eofException.initCause(ex);
+                throw eofException;
+            }
+        }
+
+        @Override
+        public int readInt() throws IOException {
+            try {
+                return buffer.readInt();
+            } catch (IndexOutOfBoundsException ex) {
+                EOFException eofException = new EOFException();
+                eofException.initCause(ex);
+                throw eofException;
+            }
+        }
+
+        @Override
+        public long readLong() throws IOException {
+            try {
+                return buffer.readLong();
+            } catch (IndexOutOfBoundsException ex) {
+                EOFException eofException = new EOFException();
+                eofException.initCause(ex);
+                throw eofException;
+            }
+        }
+
         @Override
         public void reset() throws IOException {
             buffer.resetReaderIndex();

server/src/main/java/org/elasticsearch/common/io/stream/ByteBufferStreamInput.java

@@ -20,6 +20,7 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
 
 public class ByteBufferStreamInput extends StreamInput {
@@ -76,6 +77,39 @@ public void readBytes(byte[] b, int offset, int len) throws IOException {
         buffer.get(b, offset, len);
     }
 
+    @Override
+    public short readShort() throws IOException {
+        try {
+            return buffer.getShort();
+        } catch (BufferUnderflowException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
+    @Override
+    public int readInt() throws IOException {
+        try {
+            return buffer.getInt();
+        } catch (BufferUnderflowException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
+    @Override
+    public long readLong() throws IOException {
+        try {
+            return buffer.getLong();
+        } catch (BufferUnderflowException ex) {
+            EOFException eofException = new EOFException();
+            eofException.initCause(ex);
+            throw eofException;
+        }
+    }
+
     @Override
     public void reset() throws IOException {
         buffer.reset();

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -415,7 +415,6 @@ public String readString() throws IOException {
         return spare.toString();
     }
 
-
     public final float readFloat() throws IOException {
         return Float.intBitsToFloat(readInt());
     }