Increase IO direct byte buffers to 256KB (#37283)

Currently we read and write 64KB at a time in the nio libraries. As a
single byte buffer per event loop thread does not consume much memory,
there is little reason to not increase it further. This commit increases
the buffer to 256KB but still limits a single write to 64KB. The write
limit could be increased, but too high of a write limit will lead to
copying more data (if all the data is not flushed and needs to be copied
on the next call). This is something to explore in the future.

libs/nio/src/main/java/org/elasticsearch/nio/NioSelector.java

@@ -69,7 +69,7 @@ public class NioSelector implements Closeable {
     public NioSelector(EventHandler eventHandler, Selector selector) {
         this.selector = selector;
         this.eventHandler = eventHandler;
-        this.ioBuffer = ByteBuffer.allocateDirect(1 << 16);
+        this.ioBuffer = ByteBuffer.allocateDirect(1 << 18);
     }
 
     /**

libs/nio/src/main/java/org/elasticsearch/nio/SocketChannelContext.java

@@ -294,9 +294,14 @@ public abstract class SocketChannelContext extends ChannelContext<SocketChannel>
         }
     }
 
+    // Currently we limit to 64KB. This is a trade-off which means more syscalls, in exchange for less
+    // copying.
+    private final int WRITE_LIMIT = 1 << 16;
+
     protected int flushToChannel(ByteBuffer buffer) throws IOException {
         int initialPosition = buffer.position();
         ByteBuffer ioBuffer = getSelector().getIoBuffer();
+        ioBuffer.limit(Math.min(WRITE_LIMIT, ioBuffer.limit()));
         copyBytes(buffer, ioBuffer);
         ioBuffer.flip();
         int bytesWritten;
@@ -318,6 +323,7 @@ public abstract class SocketChannelContext extends ChannelContext<SocketChannel>
         int totalBytesFlushed = 0;
         while (continueFlush) {
             ioBuffer.clear();
+            ioBuffer.limit(Math.min(WRITE_LIMIT, ioBuffer.limit()));
             int j = 0;
             ByteBuffer[] buffers = flushOperation.getBuffersToWrite();
             while (j < buffers.length && ioBuffer.remaining() > 0) {