Improve DefaultEntryLogger read performance.

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

@@ -86,7 +86,7 @@ public BufferedChannel(ByteBufAllocator allocator, FileChannel fc, int capacity,
 
     public BufferedChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity,
             long unpersistedBytesBound) throws IOException {
-        super(fc, readCapacity);
+        super(fc, readCapacity, false);
         this.writeCapacity = writeCapacity;
         this.position = fc.position();
         this.writeBufferStartPosition.set(position);

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

@@ -30,7 +30,7 @@
 /**
  * A Buffered channel without a write buffer. Only reads are buffered.
  */
-public class BufferedReadChannel extends BufferedChannelBase  {
+public class BufferedReadChannel extends BufferedChannelBase {
 
     // The capacity of the read buffer.
     protected final int readCapacity;
@@ -43,9 +43,12 @@ public class BufferedReadChannel extends BufferedChannelBase  {
 
     long invocationCount = 0;
     long cacheHitCount = 0;
+    private long fileSize = -1;
+    private final boolean sealed;
 
-    public BufferedReadChannel(FileChannel fileChannel, int readCapacity) {
+    public BufferedReadChannel(FileChannel fileChannel, int readCapacity, boolean sealed) {
         super(fileChannel);
+        this.sealed = sealed;
         this.readCapacity = readCapacity;
         this.readBuffer = Unpooled.buffer(readCapacity);
     }
@@ -64,10 +67,22 @@ public int read(ByteBuf dest, long pos) throws IOException {
         return read(dest, pos, dest.writableBytes());
     }
 
+    @Override
+    public long size() throws IOException {
+        if (sealed) {
+            if (fileSize == -1) {
+                fileSize = validateAndGetFileChannel().size();
+            }
+            return fileSize;
+        } else {
+            return validateAndGetFileChannel().size();
+        }
+    }
+
     public synchronized int read(ByteBuf dest, long pos, int length) throws IOException {
         invocationCount++;
         long currentPosition = pos;
-        long eof = validateAndGetFileChannel().size();
+        long eof = size();
         // return -1 if the given position is greater than or equal to the file's current size.
         if (pos >= eof) {
             return -1;

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

@@ -81,6 +81,8 @@
 public class DefaultEntryLogger implements EntryLogger {
     private static final Logger LOG = LoggerFactory.getLogger(DefaultEntryLogger.class);
 
+    public static final String NAME = "DEFAULT";
+
     @VisibleForTesting
     static final int UNINITIALIZED_LOG_ID = -0xDEAD;
 
@@ -575,6 +577,10 @@ private long readLastLogId(File f) {
         }
     }
 
+    void clearCompactingLogId() {
+        entryLoggerAllocator.clearCompactingLogId();
+    }
+
     /**
      * Flushes all rotated log channels. After log channels are flushed,
      * move leastUnflushedLogId ptr to current logId.
@@ -583,6 +589,11 @@ void checkpoint() throws IOException {
         entryLogManager.checkpoint();
     }
 
+    @Override
+    public String name() {
+        return NAME;
+    }
+
     @Override
     public void flush() throws IOException {
         entryLogManager.flush();
@@ -898,7 +909,7 @@ private BufferedReadChannel getChannelForLogId(long entryLogId) throws IOExcepti
         }
         // We set the position of the write buffer of this buffered channel to Long.MAX_VALUE
         // so that there are no overlaps with the write buffer while reading
-        fc = new BufferedReadChannel(newFc, conf.getReadBufferBytes());
+        fc = new BufferedReadChannel(newFc, conf.getReadBufferBytes(), entryLoggerAllocator.isSealed(entryLogId));
         putInReadChannels(entryLogId, fc);
         return fc;
     }

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

@@ -63,6 +63,8 @@ class EntryLoggerAllocator {
     private final boolean entryLogPreAllocationEnabled;
     private final ByteBufAllocator byteBufAllocator;
     final ByteBuf logfileHeader = Unpooled.buffer(DefaultEntryLogger.LOGFILE_HEADER_SIZE);
+    private long writingLogId = -1;
+    private long writingCompactingLogId = -1;
 
     EntryLoggerAllocator(ServerConfiguration conf, LedgerDirsManager ledgerDirsManager,
                          DefaultEntryLogger.RecentEntryLogsStatus recentlyCreatedEntryLogsStatus, long logId,
@@ -90,21 +92,28 @@ synchronized long getPreallocatedLogId() {
         return preallocatedLogId;
     }
 
+    public boolean isSealed(long logId) {
+        return logId != writingLogId && logId != writingCompactingLogId;
+    }
+
     BufferedLogChannel createNewLog(File dirForNextEntryLog) throws IOException {
         synchronized (createEntryLogLock) {
             BufferedLogChannel bc;
-            if (!entryLogPreAllocationEnabled){
+            if (!entryLogPreAllocationEnabled) {
                 // create a new log directly
                 bc = allocateNewLog(dirForNextEntryLog);
+                writingLogId = bc.getLogId();
                 return bc;
             } else {
                 // allocate directly to response request
-                if (null == preallocation){
+                if (null == preallocation) {
                     bc = allocateNewLog(dirForNextEntryLog);
+                    writingLogId = bc.getLogId();
                 } else {
                     // has a preallocated entry log
                     try {
                         bc = preallocation.get();
+                        writingLogId = bc.getLogId();
                     } catch (ExecutionException ee) {
                         if (ee.getCause() instanceof IOException) {
                             throw (IOException) (ee.getCause());
@@ -115,7 +124,7 @@ BufferedLogChannel createNewLog(File dirForNextEntryLog) throws IOException {
                         throw new IOException("Task to allocate a new entry log is cancelled.", ce);
                     } catch (InterruptedException ie) {
                         Thread.currentThread().interrupt();
-                        throw new IOException("Intrrupted when waiting a new entry log to be allocated.", ie);
+                        throw new IOException("Interrupted when waiting a new entry log to be allocated.", ie);
                     }
                 }
                 // preallocate a new log in background upon every call
@@ -127,10 +136,16 @@ BufferedLogChannel createNewLog(File dirForNextEntryLog) throws IOException {
 
     BufferedLogChannel createNewLogForCompaction(File dirForNextEntryLog) throws IOException {
         synchronized (createCompactionLogLock) {
-            return allocateNewLog(dirForNextEntryLog, COMPACTING_SUFFIX);
+            BufferedLogChannel bc = allocateNewLog(dirForNextEntryLog, COMPACTING_SUFFIX);
+            writingCompactingLogId = bc.getLogId();
+            return bc;
         }
     }
 
+    void clearCompactingLogId() {
+        writingCompactingLogId = -1;
+    }
+
     private synchronized BufferedLogChannel allocateNewLog(File dirForNextEntryLog) throws IOException {
         return allocateNewLog(dirForNextEntryLog, ".log");
     }

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

@@ -30,6 +30,13 @@
  */
 public class ReadOnlyDefaultEntryLogger extends DefaultEntryLogger {
 
+    public static final String NAME = "READ_ONLY_DEFAULT";
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
     public ReadOnlyDefaultEntryLogger(ServerConfiguration conf) throws IOException {
         super(conf);
     }

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

@@ -199,6 +199,7 @@ boolean complete() {
                 LOG.info("No valid entry is found in entry log after scan, removing entry log now.");
                 logRemovalListener.removeEntryLog(metadata.getEntryLogId());
                 compactionLog.abort();
+                compactingLogWriteDone();
                 return false;
             }
             return true;
@@ -209,6 +210,13 @@ void abort() {
             offsets.clear();
             // since we haven't flushed yet, we only need to delete the unflushed compaction file.
             compactionLog.abort();
+            compactingLogWriteDone();
+        }
+    }
+
+    private void compactingLogWriteDone() {
+        if (DefaultEntryLogger.NAME.equals(entryLogger.name())) {
+            ((DefaultEntryLogger) entryLogger).clearCompactingLogId();
         }
     }
 
@@ -241,6 +249,8 @@ boolean complete() throws IOException {
             } catch (IOException ioe) {
                 LOG.warn("Error marking compaction as done", ioe);
                 return false;
+            } finally {
+                compactingLogWriteDone();
             }
         }
 
@@ -249,6 +259,7 @@ void abort() {
             offsets.clear();
             // remove compaction log file and its hardlink
             compactionLog.abort();
+            compactingLogWriteDone();
         }
     }
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/EntryLogger.java

@@ -129,4 +129,6 @@ default EntryLogMetadata getEntryLogMetadata(long entryLogId) throws IOException
      * @return false if the entrylog doesn't exist.
      */
     boolean removeEntryLog(long entryLogId);
+
+    String name();
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectEntryLogger.java

@@ -63,6 +63,8 @@
  * DirectEntryLogger.
  */
 public class DirectEntryLogger implements EntryLogger {
+
+    public static final String NAME = "direct";
     private final Slogger slog;
     private final File ledgerDir;
     private final EntryLogIds ids;
@@ -380,6 +382,11 @@ public boolean removeEntryLog(long entryLogId) {
         return result;
     }
 
+    @Override
+    public String name() {
+        return NAME;
+    }
+
     @Override
     public void scanEntryLog(long entryLogId, EntryLogScanner scanner) throws IOException {
         checkArgument(entryLogId < Integer.MAX_VALUE, "Entry log id must be an int [%d]", entryLogId);