diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java index dbba31083d9..cddeff9cf29 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java +++ b/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 diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java index 830763aa022..ad637f0156c 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java +++ b/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); } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SlowBufferedChannel.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SlowBufferedChannel.java index 13bc74e0c19..b70fc4d05bf 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SlowBufferedChannel.java +++ b/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); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java index 81e7c62af4a..0114dbc1253 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java +++ b/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);