diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogMetadata.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogMetadata.java index 17cf58fb52c..b4f91e09420 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogMetadata.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogMetadata.java @@ -35,10 +35,13 @@ public class EntryLogMetadata { private final ConcurrentLongLongHashMap ledgersMap; public EntryLogMetadata(long logId) { + this(logId, 1); + } + public EntryLogMetadata(long logId, int concurrencyLevel) { this.entryLogId = logId; totalSize = remainingSize = 0; - ledgersMap = new ConcurrentLongLongHashMap(256, 1); + ledgersMap = new ConcurrentLongLongHashMap(256, concurrencyLevel); } public void addLedgerSize(long ledgerId, long size) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java index 49a9ca40594..1fa4fcdbfe8 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java @@ -89,9 +89,13 @@ static class BufferedLogChannel extends BufferedChannel { public BufferedLogChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity, long logId, File logFile, long unpersistedBytesBound) throws IOException { + this(allocator, fc, writeCapacity, readCapacity, logId, logFile, unpersistedBytesBound, 1); + } + public BufferedLogChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity, + long logId, File logFile, long unpersistedBytesBound, int logMetaMapConcurrency) throws IOException { super(allocator, fc, writeCapacity, readCapacity, unpersistedBytesBound); this.logId = logId; - this.entryLogMetadata = new EntryLogMetadata(logId); + this.entryLogMetadata = new EntryLogMetadata(logId, logMetaMapConcurrency); this.logFile = logFile; } public long getLogId() { @@ -229,11 +233,13 @@ private static class Header { final int version; final long ledgersMapOffset; final int ledgersCount; + final long createTimestamp; - Header(int version, long ledgersMapOffset, int ledgersCount) { + Header(int version, long ledgersMapOffset, int ledgersCount, long createTimestamp) { this.version = version; this.ledgersMapOffset = ledgersMapOffset; this.ledgersCount = ledgersCount; + this.createTimestamp = createTimestamp; } } @@ -254,6 +260,7 @@ private static class Header { static final int HEADER_VERSION_POSITION = 4; static final int LEDGERS_MAP_OFFSET_POSITION = HEADER_VERSION_POSITION + 4; + static final int CREATE_TIMESTAMP_POSITION = LEDGERS_MAP_OFFSET_POSITION + 8 + 4; /** * Ledgers map is composed of multiple parts that can be split into separated entries. Each of them is composed of: @@ -357,6 +364,8 @@ public EntryLogger(ServerConfiguration conf, // this header buffer is cleared before writing it into the new logChannel. logfileHeader.writeBytes("BKLO".getBytes(UTF_8)); logfileHeader.writeInt(HEADER_CURRENT_VERSION); + logfileHeader.writerIndex(CREATE_TIMESTAMP_POSITION); + logfileHeader.writeLong(System.currentTimeMillis()); logfileHeader.writerIndex(LOGFILE_HEADER_SIZE); // Find the largest logId @@ -902,11 +911,16 @@ private Header getHeaderForLogId(long entryLogId) throws IOException { long ledgersMapOffset = headers.readLong(); int ledgersCount = headers.readInt(); - return new Header(headerVersion, ledgersMapOffset, ledgersCount); + long createTimestamp = headers.readLong(); + return new Header(headerVersion, ledgersMapOffset, ledgersCount, createTimestamp); } finally { headers.release(); } } + public long getCreateTimestampForLogId(long entryLogId) throws IOException { + Header header = getHeaderForLogId(entryLogId); + return header.createTimestamp; + } private BufferedReadChannel getChannelForLogId(long entryLogId) throws IOException { BufferedReadChannel fc = getFromChannels(entryLogId); @@ -1087,7 +1101,7 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce // There can be multiple entries containing the various components of the serialized ledgers map long offset = header.ledgersMapOffset; - EntryLogMetadata meta = new EntryLogMetadata(entryLogId); + EntryLogMetadata meta = new EntryLogMetadata(entryLogId, conf.getEntryLogLedgerMapConcurrency()); final int maxMapSize = LEDGERS_MAP_HEADER_SIZE + LEDGERS_MAP_ENTRY_SIZE * LEDGERS_MAP_MAX_BATCH_SIZE; ByteBuf ledgersMap = allocator.directBuffer(maxMapSize); @@ -1151,7 +1165,7 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce } private EntryLogMetadata extractEntryLogMetadataByScanning(long entryLogId) throws IOException { - final EntryLogMetadata meta = new EntryLogMetadata(entryLogId); + final EntryLogMetadata meta = new EntryLogMetadata(entryLogId, conf.getEntryLogLedgerMapConcurrency()); // Read through the entry log file and extract the entry log meta scanEntryLog(entryLogId, new EntryLogScanner() { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java index 5b05e64cb8d..5aa9e12f4d1 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java @@ -84,6 +84,8 @@ class EntryLoggerAllocator { // this header buffer is cleared before writing it into the new logChannel. logfileHeader.writeBytes("BKLO".getBytes(UTF_8)); logfileHeader.writeInt(EntryLogger.HEADER_CURRENT_VERSION); + logfileHeader.writerIndex(EntryLogger.CREATE_TIMESTAMP_POSITION); + logfileHeader.writeLong(System.currentTimeMillis()); logfileHeader.writerIndex(EntryLogger.LOGFILE_HEADER_SIZE); } @@ -167,7 +169,12 @@ private synchronized BufferedLogChannel allocateNewLog(File dirForNextEntryLog, FileChannel channel = new RandomAccessFile(newLogFile, "rw").getChannel(); BufferedLogChannel logChannel = new BufferedLogChannel(byteBufAllocator, channel, conf.getWriteBufferBytes(), - conf.getReadBufferBytes(), preallocatedLogId, newLogFile, conf.getFlushIntervalInBytes()); + conf.getReadBufferBytes(), preallocatedLogId, newLogFile, conf.getFlushIntervalInBytes(), + conf.getEntryLogLedgerMapConcurrency()); + //update create timestamp + logfileHeader.writerIndex(EntryLogger.CREATE_TIMESTAMP_POSITION); + logfileHeader.writeLong(System.currentTimeMillis()); + logfileHeader.writerIndex(EntryLogger.LOGFILE_HEADER_SIZE); logfileHeader.readerIndex(0); logChannel.write(logfileHeader); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java index 05649b426b2..a56a7b5191c 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java @@ -613,6 +613,10 @@ protected Map extractMetaFromEntryLogs(Map extractMetaFromEntryLogs(Map 0) { + long createTimestamp = entryLogger.getCreateTimestampForLogId(entryLogId); + if (createTimestamp > 0 && System.currentTimeMillis() - createTimestamp + < conf.getExtractMetaDelayTimeSeconds() * 1000) { + return false; + } + } + } catch (Exception ex) { + LOG.warn("Failed to get create timestamp from: {}.log : {}", entryLogId, ex.getMessage()); + } + return true; + } + CompactableLedgerStorage getLedgerStorage() { return ledgerStorage; } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index 94904befb40..8807b6ee55f 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -149,8 +149,10 @@ public SingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager le log.info("Creating single directory db ledger storage on {}", baseDir); this.writeCacheMaxSize = writeCacheSize; - this.writeCache = new WriteCache(allocator, writeCacheMaxSize / 2); - this.writeCacheBeingFlushed = new WriteCache(allocator, writeCacheMaxSize / 2); + this.writeCache = new WriteCache(allocator, writeCacheMaxSize / 2, + conf.getFlushEntrySortBufferInitSize()); + this.writeCacheBeingFlushed = new WriteCache(allocator, writeCacheMaxSize / 2, + conf.getFlushEntrySortBufferInitSize()); this.checkpointSource = checkpointSource; diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/WriteCache.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/WriteCache.java index ac58e8eacac..13b7fceed78 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/WriteCache.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/WriteCache.java @@ -84,10 +84,15 @@ public interface EntryConsumer { public WriteCache(ByteBufAllocator allocator, long maxCacheSize) { // Default maxSegmentSize set to 1Gb - this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024); + this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024, 0); } - public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentSize) { + public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int entrySortBufferInitSize) { + // Default maxSegmentSize set to 1Gb + this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024, entrySortBufferInitSize); + } + + public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentSize, int entrySortBufferInitSize) { checkArgument(maxSegmentSize > 0); long alignedMaxSegmentSize = alignToPowerOfTwo(maxSegmentSize); @@ -110,6 +115,11 @@ public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentS int lastSegmentSize = (int) (maxCacheSize % maxSegmentSize); cacheSegments[segmentsCount - 1] = Unpooled.directBuffer(lastSegmentSize, lastSegmentSize); + + if (entrySortBufferInitSize > 8) { + int bufferSize = entrySortBufferInitSize / 8; + sortedEntries = new long[bufferSize]; + } } public void clear() { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java index 8fffdde0d7e..38dee135e64 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java @@ -115,12 +115,14 @@ public class ServerConfiguration extends AbstractConfigurationDefault is 0. Greater than 1, will pre-allocated buffer to reduce the probability of allocation failure. + * + * @return entry sorted buffer init size + */ + + public int getFlushEntrySortBufferInitSize() { + return this.getInt(FLUSH_ENTRY_SORT_BUFFER_INIT_SIZE, 0); + } + + /** + * Set flush entry sorted buffer init size + * + * @param initSize + * @return server configuration + */ + public ServerConfiguration SetFlushEntrySortBufferInitSize(int initSize) { + this.setProperty(FLUSH_ENTRY_SORT_BUFFER_INIT_SIZE, Integer.toString(initSize)); + return this; + } /** * Get bookie death watch interval. @@ -3561,6 +3607,28 @@ public ServerConfiguration setNumOfMemtableFlushThreads(int numOfMemtableFlushTh return this; } + /** + * Get ledger map concurrency of entry log. It should be set greater than 1 to avoid allocating a large contiguous + * memory, if there are many ledgers in a entry log. + * + * @return ledger map concurrency. + */ + public int getEntryLogLedgerMapConcurrency() { + return this.getInt(ENTRY_LOG_LEDGER_MAP_CONCURRENCY, 1); + } + + /** + * Set ledger map concurrency of entry log. + * + * @param concurrency + * map concurrency level + * @return client configuration. + */ + public ServerConfiguration setEntryLogLedgerMapConcurrency(int concurrency) { + this.setProperty(ENTRY_LOG_LEDGER_MAP_CONCURRENCY, Integer.toString(concurrency)); + return this; + } + /* * in entryLogPerLedger feature, this specifies the time, once this duration * has elapsed after the entry's last access, that entry should be diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java index 861c363bf54..b44916e7c31 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java @@ -29,6 +29,7 @@ import com.google.common.collect.Sets; import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.Unpooled; import io.netty.buffer.UnpooledByteBufAllocator; @@ -1824,4 +1825,50 @@ public void testSwappingEntryLogManager(boolean initialEntryLogPerLedgerEnabled, } } + + /** + * testcase for createtimestamp of header + */ + @Test + public void testCreatTimestamp() throws Exception { + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setLedgerDirNames(createAndGetLedgerDirs(1)); + LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), + new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); + + long timestamp = System.currentTimeMillis(); + EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager); + EntryLogManagerForSingleEntryLog entryLogManager = + (EntryLogManagerForSingleEntryLog) entryLogger.getEntryLogManager(); + + long entry0Position = entryLogger.addEntry(0L, generateEntry(0, 1)); + long entry1Position = entryLogger.addEntry(1L, generateEntry(1, 1)); + long entry2Position = entryLogger.addEntry(2L, generateEntry(2, 1)); + + BufferedLogChannel bufferedLogChannel = entryLogManager.getCurrentLogForLedger(0); + + entryLogger.flush(); + bufferedLogChannel.appendLedgersMap(); + + // Allocate buffer to read (version, ledgersMapOffset, ledgerCount) + ByteBuf headers = ByteBufAllocator.DEFAULT.buffer(1024); + bufferedLogChannel.read(headers, 0); + + // Skip marker string "BKLO" + headers.readInt(); + headers.readInt(); + + headers.readLong(); + int ledgersCount = headers.readInt(); + long createTimestampFile = headers.readLong(); + + //verify headers + assertEquals(3, ledgersCount); + assertTrue(timestamp - createTimestampFile < 4); + + //verify entry + assertEquals(0, generateEntry(0, 1).compareTo(entryLogger.readEntry(0, 1, entry0Position))); + assertEquals(0, generateEntry(1, 1).compareTo(entryLogger.readEntry(1, 1, entry1Position))); + assertEquals(0, generateEntry(2, 1).compareTo(entryLogger.readEntry(2, 1, entry2Position))); + } } diff --git a/conf/bk_server.conf b/conf/bk_server.conf index e2abc5f0e69..31871d48e21 100755 --- a/conf/bk_server.conf +++ b/conf/bk_server.conf @@ -470,6 +470,9 @@ ledgerDirectories=/tmp/bk-data # I/O. Flushing too frequently may also affect performance negatively. # flushEntrylogBytes=0 +# Default is 0. Greater than 1, will pre-allocated buffer to reduce the probability of allocation failure +# flushEntrySortBufferInitSize=0 + # The number of bytes we should use as capacity for BufferedReadChannel. Default is 512 bytes. # readBufferSizeBytes=512 @@ -485,6 +488,10 @@ ledgerDirectories=/tmp/bk-data # In the case of multipleentrylogs, multiple threads can be used to flush the memtable # numOfMemtableFlushThreads=8 +# It should be set greater than 1 to avoid allocating a large contiguous memory, +# if there are a large number of ledgers in a entry log. +# entryLogLedgerMapConcurrency=1 + # in entryLogPerLedger feature, the time duration used for lastaccess eviction policy for cache # entrylogMapAccessExpiryTimeInSeconds=300 @@ -588,6 +595,12 @@ ledgerDirectories=/tmp/bk-data # True if the bookie should double check readMetadata prior to gc # verifyMetadataOnGC=false +# If the ledgers count is very lardge, the memory of EntryLogMetadata of all entry log files will also +# be very large. If the retention time of all the entry is the same, we do not need to extract the entry +# logs in retention time to load meta to memory. It will reduce memory usage. The delay time can be set +# to the retention time. +# extractMetaDelayTimeSeconds=0 + ############################################################################# ## Disk utilization #############################################################################