Create nio-transport plugin for NioTransport (#27949)

This is related to #27260. This commit moves the NioTransport from
:test:framework to a new nio-transport plugin. Additionally, supporting
tcp decoding classes are moved to this plugin. Generic byte reading and
writing contexts are moved to the nio library.

Additionally, this commit adds a basic MockNioTransport to
:test:framework that is a TcpTransport implementation for testing that
is driven by nio.

Author: Tim-Brooks

core/src/main/java/org/elasticsearch/common/bytes/BytesReference.java

@@ -27,6 +27,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.function.ToIntBiFunction;
 
 /**
@@ -148,6 +150,37 @@ public static byte[] toBytes(BytesReference reference) {
         return BytesRef.deepCopyOf(bytesRef).bytes;
     }
 
+    /**
+     * Returns an array of byte buffers from the given BytesReference.
+     */
+    public static ByteBuffer[] toByteBuffers(BytesReference reference) {
+        BytesRefIterator byteRefIterator = reference.iterator();
+        BytesRef r;
+        try {
+            ArrayList<ByteBuffer> buffers = new ArrayList<>();
+            while ((r = byteRefIterator.next()) != null) {
+                buffers.add(ByteBuffer.wrap(r.bytes, r.offset, r.length));
+            }
+            return buffers.toArray(new ByteBuffer[buffers.size()]);
+
+        } catch (IOException e) {
+            // this is really an error since we don't do IO in our bytesreferences
+            throw new AssertionError("won't happen", e);
+        }
+    }
+
+    /**
+     * Returns BytesReference composed of the provided ByteBuffers.
+     */
+    public static BytesReference fromByteBuffers(ByteBuffer[] buffers) {
+        ByteBufferReference[] references = new ByteBufferReference[buffers.length];
+        for (int i = 0; i < references.length; ++i) {
+            references[i] = new ByteBufferReference(buffers[i]);
+        }
+
+        return new CompositeBytesReference(references);
+    }
+
     @Override
     public int compareTo(final BytesReference other) {
         return compareIterators(this, other, (a, b) -> a.compareTo(b));

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

@@ -46,12 +46,17 @@
 public interface TcpChannel extends Releasable {
 
     /**
-     * Closes the channel. This might be an asynchronous process. There is notguarantee that the channel
+     * Closes the channel. This might be an asynchronous process. There is no guarantee that the channel
      * will be closed when this method returns. Use the {@link #addCloseListener(ActionListener)} method
      * to implement logic that depends on knowing when the channel is closed.
      */
     void close();
 
+    /**
+     * This returns the profile for this channel.
+     */
+    String getProfile();
+
     /**
      * Adds a listener that will be executed when the channel is closed. If the channel is still open when
      * this listener is added, the listener will be executed by the thread that eventually closes the
@@ -86,6 +91,13 @@ public interface TcpChannel extends Releasable {
      */
     InetSocketAddress getLocalAddress();
 
+    /**
+     * Returns the remote address for this channel. Can be null if channel does not have a remote address.
+     *
+     * @return the remote address of this channel.
+     */
+    InetSocketAddress getRemoteAddress();
+
     /**
      * Sends a tcp message to the channel. The listener will be executed once the send process has been
      * completed.

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -184,8 +184,12 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
     public static final Setting.AffixSetting<Integer> PUBLISH_PORT_PROFILE = affixKeySetting("transport.profiles.", "publish_port",
         key -> intSetting(key, -1, -1, Setting.Property.NodeScope));
 
-    private static final long NINETY_PER_HEAP_SIZE = (long) (JvmInfo.jvmInfo().getMem().getHeapMax().getBytes() * 0.9);
+    // This is the number of bytes necessary to read the message size
+    public static final int BYTES_NEEDED_FOR_MESSAGE_SIZE = TcpHeader.MARKER_BYTES_SIZE + TcpHeader.MESSAGE_LENGTH_SIZE;
     public static final int PING_DATA_SIZE = -1;
+    private static final long NINETY_PER_HEAP_SIZE = (long) (JvmInfo.jvmInfo().getMem().getHeapMax().getBytes() * 0.9);
+    private static final BytesReference EMPTY_BYTES_REFERENCE = new BytesArray(new byte[0]);
+
     private final CircuitBreakerService circuitBreakerService;
     // package visibility for tests
     protected final ScheduledPing scheduledPing;
@@ -317,8 +321,7 @@ public String executor() {
     public class ScheduledPing extends AbstractLifecycleRunnable {
 
         /**
-         * The magic number (must be lower than 0) for a ping message. This is handled
-         * specifically in {@link TcpTransport#validateMessageHeader}.
+         * The magic number (must be lower than 0) for a ping message.
          */
         private final BytesReference pingHeader;
         final CounterMetric successfulPings = new CounterMetric();
@@ -1210,7 +1213,7 @@ private void sendResponse(Version nodeVersion, TcpChannel channel, final Transpo
      * @param length          the payload length in bytes
      * @see TcpHeader
      */
-    final BytesReference buildHeader(long requestId, byte status, Version protocolVersion, int length) throws IOException {
+    private BytesReference buildHeader(long requestId, byte status, Version protocolVersion, int length) throws IOException {
         try (BytesStreamOutput headerOutput = new BytesStreamOutput(TcpHeader.HEADER_SIZE)) {
             headerOutput.setVersion(protocolVersion);
             TcpHeader.writeHeader(headerOutput, requestId, status, protocolVersion, length);
@@ -1247,76 +1250,135 @@ private BytesReference buildMessage(long requestId, byte status, Version nodeVer
     }
 
     /**
-     * Validates the first N bytes of the message header and returns <code>false</code> if the message is
-     * a ping message and has no payload ie. isn't a real user level message.
+     * Consumes bytes that are available from network reads. This method returns the number of bytes consumed
+     * in this call.
      *
-     * @throws IllegalStateException    if the message is too short, less than the header or less that the header plus the message size
-     * @throws HttpOnTransportException if the message has no valid header and appears to be a HTTP message
-     * @throws IllegalArgumentException if the message is greater that the maximum allowed frame size. This is dependent on the available
-     *                                  memory.
+     * @param channel the channel read from
+     * @param bytesReference the bytes available to consume
+     * @return the number of bytes consumed
+     * @throws StreamCorruptedException if the message header format is not recognized
+     * @throws TcpTransport.HttpOnTransportException if the message header appears to be a HTTP message
+     * @throws IllegalArgumentException if the message length is greater that the maximum allowed frame size.
+     *                                  This is dependent on the available memory.
      */
-    public static boolean validateMessageHeader(BytesReference buffer) throws IOException {
-        final int sizeHeaderLength = TcpHeader.MARKER_BYTES_SIZE + TcpHeader.MESSAGE_LENGTH_SIZE;
-        if (buffer.length() < sizeHeaderLength) {
-            throw new IllegalStateException("message size must be >= to the header size");
-        }
-        int offset = 0;
-        if (buffer.get(offset) != 'E' || buffer.get(offset + 1) != 'S') {
-            // special handling for what is probably HTTP
-            if (bufferStartsWith(buffer, offset, "GET ") ||
-                bufferStartsWith(buffer, offset, "POST ") ||
-                bufferStartsWith(buffer, offset, "PUT ") ||
-                bufferStartsWith(buffer, offset, "HEAD ") ||
-                bufferStartsWith(buffer, offset, "DELETE ") ||
-                bufferStartsWith(buffer, offset, "OPTIONS ") ||
-                bufferStartsWith(buffer, offset, "PATCH ") ||
-                bufferStartsWith(buffer, offset, "TRACE ")) {
-
-                throw new HttpOnTransportException("This is not a HTTP port");
+    public int consumeNetworkReads(TcpChannel channel, BytesReference bytesReference) throws IOException {
+        BytesReference message = decodeFrame(bytesReference);
+
+        if (message == null) {
+            return 0;
+        } else if (message.length() == 0) {
+            // This is a ping and should not be handled.
+            return BYTES_NEEDED_FOR_MESSAGE_SIZE;
+        } else {
+            try {
+                messageReceived(message, channel);
+            } catch (Exception e) {
+                onException(channel, e);
             }
+            return message.length() + BYTES_NEEDED_FOR_MESSAGE_SIZE;
+        }
+    }
 
-            // we have 6 readable bytes, show 4 (should be enough)
-            throw new StreamCorruptedException("invalid internal transport message format, got ("
-                + Integer.toHexString(buffer.get(offset) & 0xFF) + ","
-                + Integer.toHexString(buffer.get(offset + 1) & 0xFF) + ","
-                + Integer.toHexString(buffer.get(offset + 2) & 0xFF) + ","
-                + Integer.toHexString(buffer.get(offset + 3) & 0xFF) + ")");
+    /**
+     * Attempts to a decode a message from the provided bytes. If a full message is not available, null is
+     * returned. If the message is a ping, an empty {@link BytesReference} will be returned.
+     *
+     * @param networkBytes the will be read
+     * @return the message decoded
+     * @throws StreamCorruptedException if the message header format is not recognized
+     * @throws TcpTransport.HttpOnTransportException if the message header appears to be a HTTP message
+     * @throws IllegalArgumentException if the message length is greater that the maximum allowed frame size.
+     *                                  This is dependent on the available memory.
+     */
+    public static BytesReference decodeFrame(BytesReference networkBytes) throws IOException {
+        int messageLength = readMessageLength(networkBytes);
+        if (messageLength == -1) {
+            return null;
+        } else {
+            int totalLength = messageLength + BYTES_NEEDED_FOR_MESSAGE_SIZE;
+            if (totalLength > networkBytes.length()) {
+                return null;
+            } else if (totalLength == 6) {
+                return EMPTY_BYTES_REFERENCE;
+            } else {
+                return networkBytes.slice(BYTES_NEEDED_FOR_MESSAGE_SIZE, messageLength);
+            }
         }
+    }
 
-        final int dataLen;
-        try (StreamInput input = buffer.streamInput()) {
-            input.skip(TcpHeader.MARKER_BYTES_SIZE);
-            dataLen = input.readInt();
-            if (dataLen == PING_DATA_SIZE) {
-                // discard the messages we read and continue, this is achieved by skipping the bytes
-                // and returning null
-                return false;
+    /**
+     * Validates the first 6 bytes of the message header and returns the length of the message. If 6 bytes
+     * are not available, it returns -1.
+     *
+     * @param networkBytes the will be read
+     * @return the length of the message
+     * @throws StreamCorruptedException if the message header format is not recognized
+     * @throws TcpTransport.HttpOnTransportException if the message header appears to be a HTTP message
+     * @throws IllegalArgumentException if the message length is greater that the maximum allowed frame size.
+     *                                  This is dependent on the available memory.
+     */
+    public static int readMessageLength(BytesReference networkBytes) throws IOException {
+        if (networkBytes.length() < BYTES_NEEDED_FOR_MESSAGE_SIZE) {
+            return -1;
+        } else {
+            return readHeaderBuffer(networkBytes);
+        }
+    }
+
+    private static int readHeaderBuffer(BytesReference headerBuffer) throws IOException {
+        if (headerBuffer.get(0) != 'E' || headerBuffer.get(1) != 'S') {
+            if (appearsToBeHTTP(headerBuffer)) {
+                throw new TcpTransport.HttpOnTransportException("This is not a HTTP port");
             }
+
+            throw new StreamCorruptedException("invalid internal transport message format, got ("
+                + Integer.toHexString(headerBuffer.get(0) & 0xFF) + ","
+                + Integer.toHexString(headerBuffer.get(1) & 0xFF) + ","
+                + Integer.toHexString(headerBuffer.get(2) & 0xFF) + ","
+                + Integer.toHexString(headerBuffer.get(3) & 0xFF) + ")");
+        }
+        final int messageLength;
+        try (StreamInput input = headerBuffer.streamInput()) {
+            input.skip(TcpHeader.MARKER_BYTES_SIZE);
+            messageLength = input.readInt();
         }
 
-        if (dataLen <= 0) {
-            throw new StreamCorruptedException("invalid data length: " + dataLen);
+        if (messageLength == TcpTransport.PING_DATA_SIZE) {
+            // This is a ping
+            return 0;
         }
-        // safety against too large frames being sent
-        if (dataLen > NINETY_PER_HEAP_SIZE) {
-            throw new IllegalArgumentException("transport content length received [" + new ByteSizeValue(dataLen) + "] exceeded ["
-                + new ByteSizeValue(NINETY_PER_HEAP_SIZE) + "]");
+
+        if (messageLength <= 0) {
+            throw new StreamCorruptedException("invalid data length: " + messageLength);
         }
 
-        if (buffer.length() < dataLen + sizeHeaderLength) {
-            throw new IllegalStateException("buffer must be >= to the message size but wasn't");
+        if (messageLength > NINETY_PER_HEAP_SIZE) {
+            throw new IllegalArgumentException("transport content length received [" + new ByteSizeValue(messageLength) + "] exceeded ["
+                + new ByteSizeValue(NINETY_PER_HEAP_SIZE) + "]");
         }
-        return true;
+
+        return messageLength;
+    }
+
+    private static boolean appearsToBeHTTP(BytesReference headerBuffer) {
+        return bufferStartsWith(headerBuffer, "GET") ||
+            bufferStartsWith(headerBuffer, "POST") ||
+            bufferStartsWith(headerBuffer, "PUT") ||
+            bufferStartsWith(headerBuffer, "HEAD") ||
+            bufferStartsWith(headerBuffer, "DELETE") ||
+            // Actually 'OPTIONS'. But we are only guaranteed to have read six bytes at this point.
+            bufferStartsWith(headerBuffer, "OPTION") ||
+            bufferStartsWith(headerBuffer, "PATCH") ||
+            bufferStartsWith(headerBuffer, "TRACE");
     }
 
-    private static boolean bufferStartsWith(BytesReference buffer, int offset, String method) {
+    private static boolean bufferStartsWith(BytesReference buffer, String method) {
         char[] chars = method.toCharArray();
         for (int i = 0; i < chars.length; i++) {
-            if (buffer.get(offset + i) != chars[i]) {
+            if (buffer.get(i) != chars[i]) {
                 return false;
             }
         }
-
         return true;
     }
 
@@ -1343,8 +1405,10 @@ public HttpOnTransportException(StreamInput in) throws IOException {
     /**
      * This method handles the message receive part for both request and responses
      */
-    public final void messageReceived(BytesReference reference, TcpChannel channel, String profileName,
-                                      InetSocketAddress remoteAddress, int messageLengthBytes) throws IOException {
+    public final void messageReceived(BytesReference reference, TcpChannel channel) throws IOException {
+        String profileName = channel.getProfile();
+        InetSocketAddress remoteAddress = channel.getRemoteAddress();
+        int messageLengthBytes = reference.length();
         final int totalMessageSize = messageLengthBytes + TcpHeader.MARKER_BYTES_SIZE + TcpHeader.MESSAGE_LENGTH_SIZE;
         readBytesMetric.inc(totalMessageSize);
         // we have additional bytes to read, outside of the header