This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new d476a7b9e2 Write journal entry length prefix and payload in a single
BufferedChannel write (#4833)
d476a7b9e2 is described below
commit d476a7b9e2adf317fee614913740b3677ae295df
Author: Matteo Merli <[email protected]>
AuthorDate: Tue Jul 14 08:25:47 2026 -0700
Write journal entry length prefix and payload in a single BufferedChannel
write (#4833)
---
.../apache/bookkeeper/bookie/BufferedChannel.java | 73 ++++++++++++++++------
.../java/org/apache/bookkeeper/bookie/Journal.java | 3 +-
.../bookkeeper/bookie/SlowBufferedChannel.java | 6 ++
.../bookkeeper/bookie/BufferedChannelTest.java | 40 ++++++++++++
4 files changed, 100 insertions(+), 22 deletions(-)
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 dbba31083d..cddeff9cf2 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 class BufferedChannel extends BufferedReadChannel
implements Closeable {
* @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 830763aa02..ad637f0156 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 class Journal implements CheckpointSource {
// 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 13bc74e0c1..b70fc4d05b 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 class SlowBufferedChannel extends BufferedChannel {
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 81e7c62af4..0114dbc125 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 @@ package org.apache.bookkeeper.bookie;
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 class BufferedChannelTest {
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);
