Merge branch 'master' of https://github.com/eBay/Jungle

cmake/CodeCoverage.cmake

@@ -162,7 +162,7 @@ function(SETUP_TARGET_FOR_COVERAGE)
         COMMAND ${LCOV_PATH} --directory . --capture --output-file ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info
         # add baseline counters
         COMMAND ${LCOV_PATH} -a ${PROJECT_BINARY_DIR}/${Coverage_NAME}.base -a ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info --output-file ${PROJECT_BINARY_DIR}/${Coverage_NAME}.total
-        COMMAND ${LCOV_PATH} --remove ${Coverage_NAME}.total ${COVERAGE_EXCLUDES} --output-file ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info.cleaned
+        COMMAND ${LCOV_PATH} --ignore-errors unused --remove ${Coverage_NAME}.total ${COVERAGE_EXCLUDES} --output-file ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info.cleaned
         COMMAND ${GENHTML_PATH} -o ${PROJECT_BINARY_DIR}/${Coverage_NAME} ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info.cleaned
         #COMMAND ${CMAKE_COMMAND} -E remove ${Coverage_NAME}.base ${Coverage_NAME}.info ${Coverage_NAME}.total ${PROJECT_BINARY_DIR}/${Coverage_NAME}.info.cleaned
 

include/libjungle/db_config.h

@@ -104,6 +104,7 @@ class DBConfig {
     DBConfig()
         : allowOverwriteSeqNum(false)
         , logSectionOnly(false)
+        , autoLogFlush(true)
         , truncateInconsecutiveLogs(true)
         , logFileTtl_sec(0)
         , maxKeepingMemtables(0)
@@ -143,6 +144,8 @@ class DBConfig {
         , fastIndexScan(false)
         , seqLoadingDelayFactor(0)
         , safeMode(false)
+        , serializeMultiThreadedLogFlush(false)
+        , skipManifestSync(false)
     {
         tableSizeRatio.push_back(2.5);
         levelSizeRatio.push_back(10.0);
@@ -178,11 +181,21 @@ class DBConfig {
      */
     bool allowOverwriteSeqNum;
 
-    /*
+    /**
      * Disable table section and use logging part only.
      */
     bool logSectionOnly;
 
+    /**
+     * If it is normal DB instance (`readOnly = false` and `logSectionOnly = false`),
+     * background flusher thread will automatically flush logs to L0 tables.
+     *
+     * WARNING:
+     *   If it is set to `false`, users should manually call `flushLogs()`.
+     *   The more unflushed logs, the more memory consumption.
+     */
+    bool autoLogFlush;
+
     /*
      * (Only when `logSectionOnly == true`)
      * Truncate tail logs if they are inconsecutive,
@@ -575,6 +588,25 @@ class DBConfig {
      * real production environment.
      */
     bool safeMode;
+
+    /**
+     * If `true`, `sync` and `flushLogs` calls by multiple threads will be serialized,
+     * and executed one by one.
+     * If `false`, only one thread will execute `sync` and `flushLogs` calls, while
+     * the other concurrent threads will get `OPERATION_IN_PROGRESS` status.
+     */
+    bool serializeMultiThreadedLogFlush;
+
+    /**
+     * [EXPERIMENTAL]
+     * If `true`, when `sync()` is invoked, only the actual log files will be synced,
+     * not the manifest file. The manifest file is synced when 1) a new log file is
+     * added, or 2) the DB is closed.
+     *
+     * Even without syncing the manifest file, Jungle can recover the last synced
+     * data by scanning the log files.
+     */
+    bool skipManifestSync;
 };
 
 class GlobalConfig {

include/libjungle/params.h

@@ -205,6 +205,7 @@ struct DebugParams {
         , newLogBatchCb(nullptr)
         , getLogFileInfoBySeqCb(nullptr)
         , logFlushCb(nullptr)
+        , syncCb(nullptr)
         , forceMerge(false)
         {}
 
@@ -299,6 +300,12 @@ struct DebugParams {
      */
     std::function< void(const GenericCbParams&) > logFlushCb;
 
+    /**
+     * Callback function that will be invoked at the beginning log sync
+     * (reading memtable data and writing them into log files).
+     */
+    std::function< void(const GenericCbParams&) > syncCb;
+
     /**
      * If true, merge will proceed the task even with the small number
      * of tables in the level.

include/libjungle/sized_buf.h

@@ -16,6 +16,7 @@ limitations under the License.
 
 #pragma once
 
+#include <cstdint>
 #include <iostream>
 #include <sstream>
 #include <string>

src/cmd_handler.cc

@@ -291,7 +291,7 @@ std::string CmdHandler::hDumpKv(DBWrap* target_dbw,
         ss << "[" << count << "]" << std::endl;
         ss << "key: " << HexDump::toString(rec_out.kv.key) << std::endl;
         ss << "sequence number: " << rec_out.seqNum << std::endl;
-        ss << "type: " << rec_out.type << std::endl;
+        ss << "type: " << (int)rec_out.type << std::endl;
         ss << "meta: " << HexDump::toString(rec_out.meta) << std::endl;
         if (tokens[0] == "getmeta") {
             ss << "value: " << rec_out.kv.value.size << " bytes" << std::endl;

src/generic_bitmap.h

@@ -20,6 +20,7 @@ limitations under the License.
 
 #pragma once
 
+#include <cstdlib>
 #include <mutex>
 #include <thread>
 

src/log_file.cc

@@ -401,7 +401,7 @@ Status LogFile::getPrefix(const uint64_t chk,
     return Status();
 }
 
-Status LogFile::flushMemTable(uint64_t upto) {
+Status LogFile::flushMemTable(uint64_t upto, uint64_t& flushed_seq_out) {
    touch();
    // Skip unnecessary flushing
    if (immutable && !fHandle && isSynced()) {
@@ -509,7 +509,7 @@ Status LogFile::flushMemTable(uint64_t upto) {
 
     RwSerializer rws(fOps, fHandle, true);
 
-    TC( mTable->flush(rws, upto) );
+    TC( mTable->flush(rws, upto, flushed_seq_out) );
     TC( mTable->appendFlushMarker(rws) );
 
     TC( fOps->flush(fHandle) );

src/log_file.h

@@ -93,7 +93,7 @@ class LogFile {
                      bool allow_flushed_log,
                      bool allow_tombstone);
 
-    Status flushMemTable(uint64_t upto = NOT_INITIALIZED);
+    Status flushMemTable(uint64_t upto, uint64_t& flushed_seq_out);
 
     Status purgeMemTable();
 

src/log_manifest.cc

@@ -480,8 +480,9 @@ Status LogManifest::clone(const std::string& dst_path) {
 Status LogManifest::store(bool call_fsync) {
     if (mFileName.empty() || !fOps) return Status::NOT_INITIALIZED;
 
-    if (call_fsync) {
-        // `fsync` is required: calls by multiple threads should be serialized.
+    if (call_fsync || logMgr->getDbConfig()->serializeMultiThreadedLogFlush) {
+        // `fsync` is required, or serialize option is on:
+        // calls by multiple threads should be serialized.
         std::unique_lock<std::mutex> l(mFileWriteLock);
         return storeInternal(call_fsync);
     } else {
@@ -600,7 +601,7 @@ Status LogManifest::storeInternal(bool call_fsync) {
     if (need_truncate) {
         fOps->ftruncate(mFile, ss.pos());
     }
-    
+
     bool backup_done = false;
     if (call_fsync) {
         s = fOps->fsync(mFile);