Merge pull request #298 from cruzdb/inflight-throttle

log: throttle number of inflight ios

src/include/zlog/options.h

@@ -35,6 +35,8 @@ struct Options {
   uint32_t stripe_width = 10;
   uint32_t stripe_slots = 5;
 
+  uint32_t max_inflight_ops = 1024;
+
   ///////////////////////////////////////////////////////////////////
 
   // number of I/O threads

src/libzlog/log_impl.cc

@@ -32,6 +32,7 @@ LogImpl::LogImpl(std::shared_ptr<Backend> backend,
   hoid(hoid),
   prefix(prefix),
   striper(this, secret),
+  num_inflight_ops_(0),
   options(opts)
 {
   assert(!name.empty());
@@ -465,7 +466,21 @@ int LogImpl::trimAsync(uint64_t position, std::function<void(int)> cb)
 
 void LogImpl::queue_op(std::unique_ptr<LogOp> op)
 {
-  std::lock_guard<std::mutex> lk(lock);
+  std::unique_lock<std::mutex> lk(lock);
+
+  if (num_inflight_ops_ >= options.max_inflight_ops) {
+    std::condition_variable cond;
+    queue_op_waiters_.emplace_front(false, &cond);
+    auto it = queue_op_waiters_.begin();
+    cond.wait(lk, [&] {
+      assert(it->second == &cond);
+      return it->first;
+    });
+    queue_op_waiters_.erase(it);
+  }
+
+  num_inflight_ops_++;
+
   pending_ops_.emplace_back(std::move(op));
   finishers_cond_.notify_all();
 }
@@ -499,6 +514,14 @@ void LogImpl::finisher_entry_()
       int ret = op->run();
       op->callback(ret);
     }
+
+    std::lock_guard<std::mutex> lk(lock);
+    assert(num_inflight_ops_ > 0);
+    num_inflight_ops_--;
+    if (!queue_op_waiters_.empty()) {
+      queue_op_waiters_.back().first = true;
+      queue_op_waiters_.back().second->notify_one();
+    }
   }
 }
 

src/libzlog/log_impl.h

@@ -212,6 +212,10 @@ class LogImpl : public Log {
   uint64_t exclusive_position;
   bool exclusive_empty;
 
+  uint32_t num_inflight_ops_;
+  std::list<std::pair<bool,
+    std::condition_variable*>> queue_op_waiters_;
+
   const Options options;
 };