Write journal entry length prefix and payload in a single BufferedChannel write

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java

@@ -115,35 +115,68 @@ public synchronized void close() throws IOException {
      * @throws IOException if a write operation fails.
      */
     public void write(ByteBuf src) throws IOException {
-        int copied = 0;
         boolean shouldForceWrite = false;
         synchronized (this) {
-            int len = src.readableBytes();
-            while (copied < len) {
-                int bytesToCopy = Math.min(src.readableBytes() - copied, writeBuffer.writableBytes());
-                writeBuffer.writeBytes(src, src.readerIndex() + copied, bytesToCopy);
-                copied += bytesToCopy;
+            int copied = copyIntoWriteBuffer(src);
+            shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
+        }
+        if (shouldForceWrite) {
+            forceWrite(false);
+        }
+    }
 
-                // if we have run out of buffer space, we should flush to the
-                // file
-                if (!writeBuffer.isWritable()) {
-                    flush();
-                }
-            }
-            position += copied;
-            if (doRegularFlushes) {
-                unpersistedBytes.addAndGet(copied);
-                if (unpersistedBytes.get() >= unpersistedBytesBound) {
-                    flush();
-                    shouldForceWrite = true;
-                }
-            }
+    /**
+     * Write all the data in src1 and src2, in order, to the {@link FileChannel}, taking the
+     * lock only once. This avoids the overhead of two lock acquisitions when writing an
+     * entry preceded by its length prefix.
+     *
+     * @param src1 The first source buffer which contains the data to be written.
+     * @param src2 The second source buffer which contains the data to be written.
+     * @throws IOException if a write operation fails.
+     */
+    public void write(ByteBuf src1, ByteBuf src2) throws IOException {
+        boolean shouldForceWrite = false;
+        synchronized (this) {
+            int copied = copyIntoWriteBuffer(src1);
+            copied += copyIntoWriteBuffer(src2);
+            shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
         }
         if (shouldForceWrite) {
             forceWrite(false);
         }
     }
 
+    // must be called while holding the lock on this instance
+    private int copyIntoWriteBuffer(ByteBuf src) throws IOException {
+        int copied = 0;
+        int len = src.readableBytes();
+        while (copied < len) {
+            int bytesToCopy = Math.min(src.readableBytes() - copied, writeBuffer.writableBytes());
+            writeBuffer.writeBytes(src, src.readerIndex() + copied, bytesToCopy);
+            copied += bytesToCopy;
+
+            // if we have run out of buffer space, we should flush to the
+            // file
+            if (!writeBuffer.isWritable()) {
+                flush();
+            }
+        }
+        return copied;
+    }
+
+    // must be called while holding the lock on this instance
+    private boolean updatePositionAndFlushIfNeeded(int copied) throws IOException {
+        position += copied;
+        if (doRegularFlushes) {
+            unpersistedBytes.addAndGet(copied);
+            if (unpersistedBytes.get() >= unpersistedBytesBound) {
+                flush();
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Get the position where the next write operation will begin writing from.
      * @return

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java

@@ -1232,8 +1232,7 @@ public void run() {
                     // preAlloc based on size
                     logFile.preAllocIfNeeded(4 + entrySize);
 
-                    bc.write(lenBuff);
-                    bc.write(qe.entry);
+                    bc.write(lenBuff, qe.entry);
                     memoryLimitController.releaseMemory(qe.entry.readableBytes());
                     ReferenceCountUtil.release(qe.entry);
                 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SlowBufferedChannel.java

@@ -63,6 +63,12 @@ public synchronized void write(ByteBuf src) throws IOException {
         super.write(src);
     }
 
+    @Override
+    public synchronized void write(ByteBuf src1, ByteBuf src2) throws IOException {
+        delayMs(addDelay);
+        super.write(src1, src2);
+    }
+
     @Override
     public void flush() throws IOException {
         delayMs(flushDelay);

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java

@@ -24,12 +24,14 @@
 import static org.junit.Assert.assertThrows;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.channels.FileChannel;
+import java.nio.file.Files;
 import java.util.Random;
 import org.junit.Assert;
 import org.junit.Test;
@@ -129,6 +131,44 @@ public void testBufferedChannel(int byteBufLength, int numOfWrites, int unpersis
         fileChannel.close();
     }
 
+    @Test
+    public void testWriteTwoBuffers() throws Exception {
+        File newLogFile = File.createTempFile("test", "log");
+        newLogFile.deleteOnExit();
+        FileChannel fileChannel = new RandomAccessFile(newLogFile, "rw").getChannel();
+
+        // small write capacity so the pairs below cross the buffer boundary at varied offsets,
+        // including a payload equal to the capacity and one larger than it
+        int writeCapacity = 64;
+        BufferedChannel logChannel = new BufferedChannel(UnpooledByteBufAllocator.DEFAULT, fileChannel,
+                writeCapacity, INTERNAL_BUFFER_READ_CAPACITY, 0);
+
+        int[] payloadSizes = { 25, 31, 60, 3, 64, 128, 1, 41 };
+        ByteBuf expected = Unpooled.buffer();
+        ByteBuf lenBuf = Unpooled.buffer(4);
+
+        for (int payloadSize : payloadSizes) {
+            ByteBuf payload = generateEntry(payloadSize);
+            lenBuf.clear();
+            lenBuf.writeInt(payloadSize);
+
+            expected.writeBytes(lenBuf.slice());
+            expected.writeBytes(payload.slice());
+
+            logChannel.write(lenBuf, payload);
+        }
+
+        Assert.assertEquals(expected.readableBytes(), logChannel.position());
+
+        logChannel.flush();
+
+        byte[] fileBytes = Files.readAllBytes(newLogFile.toPath());
+        Assert.assertArrayEquals(ByteBufUtil.getBytes(expected), fileBytes);
+
+        logChannel.close();
+        fileChannel.close();
+    }
+
     @Test
     public void testBufferedChannelReadWhenDestBufSizeExceedsReadLength() throws IOException {
         doTestBufferedChannelReadThrowing(100, 60);