Do not block Translog add on file write (#63374)

Currently a TranslogWriter add operation is synchronized. This operation
adds the bytes to the file output stream buffer and issues a write
system call if the buffer is filled. This happens every 8KB which means
that we routinely block other add calls on system writes.

This commit modifies the add operation to simply place the operation in
an array list. The array list if flushed when the sync call occurs or
when 1MB is buffered.

Author: Tim-Brooks

server/src/main/java/org/elasticsearch/common/io/DiskIoBufferPool.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.io;
+
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.threadpool.ThreadPool;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+public class DiskIoBufferPool {
+
+    public static final int BUFFER_SIZE = StrictMath.toIntExact(ByteSizeValue.parseBytesSizeValue(
+        System.getProperty("es.disk_io.direct.buffer.size", "64KB"), "es.disk_io.direct.buffer.size").getBytes());
+    public static final int HEAP_BUFFER_SIZE = 8 * 1024;
+
+    private static final ThreadLocal<ByteBuffer> ioBufferPool = ThreadLocal.withInitial(() -> {
+        if (isWriteOrFlushThread()) {
+            return ByteBuffer.allocateDirect(BUFFER_SIZE);
+        } else {
+            return ByteBuffer.allocate(HEAP_BUFFER_SIZE);
+        }
+    });
+
+    public static ByteBuffer getIoBuffer() {
+        ByteBuffer ioBuffer = ioBufferPool.get();
+        ioBuffer.clear();
+        return ioBuffer;
+    }
+
+    private static boolean isWriteOrFlushThread() {
+        String threadName = Thread.currentThread().getName();
+        for (String s : Arrays.asList(
+            "[" + ThreadPool.Names.WRITE + "]",
+            "[" + ThreadPool.Names.FLUSH + "]",
+            "[" + ThreadPool.Names.SYSTEM_WRITE + "]")) {
+            if (threadName.contains(s)) {
+                return true;
+            }
+        }
+        return false;
+    }
+}

server/src/main/java/org/elasticsearch/common/util/concurrent/ReleasableLock.java

@@ -58,6 +58,19 @@ public ReleasableLock acquire() throws EngineException {
         return this;
     }
 
+    /**
+     * Try acquiring lock, returning null if unable.
+     */
+    public ReleasableLock tryAcquire() {
+        boolean locked = lock.tryLock();
+        if (locked) {
+            assert addCurrentThread();
+            return this;
+        } else {
+            return null;
+        }
+    }
+
     /**
      * Try acquiring lock, returning null if unable to acquire lock within timeout.
      */

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -28,13 +28,13 @@
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.lucene.uid.Versions;
-import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
 import org.elasticsearch.core.internal.io.IOUtils;
@@ -521,6 +521,7 @@ TranslogWriter createWriter(long fileGeneration, long initialMinTranslogGen, lon
      */
     public Location add(final Operation operation) throws IOException {
         final ReleasableBytesStreamOutput out = new ReleasableBytesStreamOutput(bigArrays);
+        boolean successfullySerialized = false;
         try {
             final long start = out.position();
             out.skip(Integer.BYTES);
@@ -530,8 +531,9 @@ public Location add(final Operation operation) throws IOException {
             out.seek(start);
             out.writeInt(operationSize);
             out.seek(end);
-            final BytesReference bytes = out.bytes();
-            try (ReleasableLock ignored = readLock.acquire()) {
+            successfullySerialized = true;
+            try (ReleasableBytesReference bytes = new ReleasableBytesReference(out.bytes(), out);
+                 ReleasableLock ignored = readLock.acquire()) {
                 ensureOpen();
                 if (operation.primaryTerm() > current.getPrimaryTerm()) {
                     assert false :
@@ -549,7 +551,9 @@ public Location add(final Operation operation) throws IOException {
             closeOnTragicEvent(ex);
             throw new TranslogException(shardId, "Failed to write operation [" + operation + "]", ex);
         } finally {
-            Releasables.close(out);
+            if (successfullySerialized == false) {
+                Releasables.close(out);
+            }
         }
     }
 
@@ -1907,7 +1911,7 @@ public static String createEmptyTranslog(final Path location,
         Checkpoint.write(channelFactory, checkpointFile, checkpoint, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
         IOUtils.fsync(checkpointFile, false);
         final TranslogWriter writer = TranslogWriter.create(shardId, uuid, generation, translogFile, channelFactory,
-            new ByteSizeValue(10), minTranslogGeneration, initialGlobalCheckpoint,
+            TranslogConfig.DEFAULT_BUFFER_SIZE, minTranslogGeneration, initialGlobalCheckpoint,
             () -> {
                 throw new UnsupportedOperationException();
             }, () -> {

server/src/main/java/org/elasticsearch/index/translog/TranslogConfig.java

@@ -34,7 +34,7 @@
  */
 public final class TranslogConfig {
 
-    public static final ByteSizeValue DEFAULT_BUFFER_SIZE = new ByteSizeValue(8, ByteSizeUnit.KB);
+    public static final ByteSizeValue DEFAULT_BUFFER_SIZE = new ByteSizeValue(1, ByteSizeUnit.MB);
     private final BigArrays bigArrays;
     private final IndexSettings indexSettings;
     private final ShardId shardId;