Add dedicated flusher for syncOnly (#170)

* Add dedicated flusher for syncOnly * Resolve comments * Make it extendable * Fix typo * Remove unnecessary comment * Fix include order --------- Co-authored-by: Zexi Liu <[email protected]>
eBay · Aug 7, 2024 · fb7904f · fb7904f
1 parent 7648451
commit fb7904f
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 3 deletions.
diff --git a/include/libjungle/db_config.h b/include/libjungle/db_config.h
@@ -582,6 +582,7 @@ class GlobalConfig {
  GlobalConfig()
  : globalLogPath("./")
  , numFlusherThreads(1)
+ , numDedicatedFlusherForAsyncReqs(0)
  , flusherSleepDuration_ms(500)
  , flusherMinRecordsToTrigger(65536)
  , flusherMinLogFilesToTrigger(16)
@@ -592,8 +593,7 @@ class GlobalConfig {
  , fdbCacheSize(0)
  , numTableWriters(8)
  , memTableFlushBufferSize(32768)
- , shutdownLogger(true)
- {}
+ , shutdownLogger(true) {}
 
  /**
  * Path where Jungle's global log will be located.
@@ -605,6 +605,12 @@ class GlobalConfig {
  */
  size_t numFlusherThreads;
 
+ /**
+ * Create dedicated flushers for async request. Only effective when it is
+ * not 0 and `numFlusherThreads` > `numDedicatedFlusherForAsyncReqs`.
+ */
+ size_t numDedicatedFlusherForAsyncReqs;
+
  /**
  * Fluhser thread sleep time in ms.
  */

diff --git a/src/db_mgr.cc b/src/db_mgr.cc
@@ -23,6 +23,8 @@ limitations under the License.
 #include "internal_helper.h"
 #include "log_reclaimer.h"
 
+#include "libjungle/db_config.h"
+
 #include <set>
 
 #include _MACRO_TO_STR(LOGGER_H)
@@ -86,9 +88,17 @@ void DBMgr::initInternal(const GlobalConfig& config) {
 
  printGlobalConfig();
 
+ bool dedicated_async_flusher =
+ config.numDedicatedFlusherForAsyncReqs
+ && config.numFlusherThreads > config.numDedicatedFlusherForAsyncReqs;
  for (size_t ii=0; ii<config.numFlusherThreads; ++ii) {
  std::string t_name = "flusher_" + std::to_string(ii);
  Flusher* flusher = new Flusher(t_name, config);
+ // If dedicated flusher is enabled, only the first
+ // `numDedicatedFlusherForAsyncReqs` flushers will handle async reqs.
+ if (dedicated_async_flusher && ii >= config.numDedicatedFlusherForAsyncReqs) {
+ flusher->handleAsyncReqs = false;
+ }
  wMgr->addWorker(flusher);
  flusher->run();
  }

diff --git a/src/flusher.cc b/src/flusher.cc
@@ -87,6 +87,7 @@ Flusher::Flusher(const std::string& _w_name,
 {
  workerName = _w_name;
  gConfig = _config;
+ handleAsyncReqs = true;
  FlusherOptions options;
  options.sleepDuration_ms = gConfig.flusherSleepDuration_ms;
  options.worker = this;
@@ -105,7 +106,11 @@ void Flusher::work(WorkerOptions* opt_base) {
 
  DB* target_db = nullptr;
 
- FlusherQueueElem* elem = dbm->flusherQueue()->pop();
+ FlusherQueueElem* elem = nullptr;
+ if (handleAsyncReqs) {
+ elem = dbm->flusherQueue()->pop();
+ }
+
  if (elem) {
  // User assigned work check if it is already closed.
  std::lock_guard<std::mutex> l(dbm->dbMapLock);

diff --git a/src/flusher.h b/src/flusher.h
@@ -16,6 +16,7 @@ limitations under the License.
 
 #pragma once
 
+#include "internal_helper.h"
 #include "worker_mgr.h"
 
 #include <libjungle/jungle.h>
@@ -82,6 +83,7 @@ class Flusher : public WorkerBase {
 
  GlobalConfig gConfig;
  size_t lastCheckedFileIndex;
+ bool handleAsyncReqs;
 };