WBM-Initiated Flushes (#155) - CR Updates #5

memtable/write_buffer_manager.cc

@@ -100,8 +100,7 @@ size_t WriteBufferManager::ReserveMemWithCache(size_t mem) {
   // lock-free solution if it ends up with a performance bottleneck.
   std::lock_guard<std::mutex> lock(cache_res_mgr_mu_);
 
-  size_t old_mem_used = memory_used_.load(std::memory_order_relaxed);
-  size_t new_mem_used = old_mem_used + mem;
+  size_t new_mem_used = memory_used_.load(std::memory_order_relaxed) + mem;
   memory_used_.store(new_mem_used, std::memory_order_relaxed);
   Status s = cache_res_mgr_->UpdateCacheReservation(new_mem_used);
 
@@ -427,18 +426,20 @@ void WriteBufferManager::InitiateFlushesThread() {
           // increment & decrement cancel each other with respect to the recalc
           ++num_running_flushes_;
           --num_flushes_to_initiate_;
+
+          // TODO: Use a weak-pointer for the registered initiators. That would
+          // allow us to release the flushes_initiators_mu_ mutex before calling
+          // the callback (which may take a long time).
           was_flush_initiated = initiator.cb(kMinFlushSizes[iter]);
-          if (!was_flush_initiated) {
-            // No flush was initiated => undo the counters update
-            --num_running_flushes_;
-            ++num_flushes_to_initiate_;
-          }
         }
 
-        if (was_flush_initiated) {
-          num_repeated_failures_to_initiate = 0U;
-        } else {
+        if (!was_flush_initiated) {
+          // No flush was initiated => undo the counters update
+          --num_running_flushes_;
+          ++num_flushes_to_initiate_;
           ++num_repeated_failures_to_initiate;
+        } else {
+          num_repeated_failures_to_initiate = 0U;
         }
       }
       ++iter;