Throttle: add a simpler throttle that just blocks above a threshold

This is convenient to use to turn synchronous calls into asynchronous
calls with a limited number of operations in flight concurrently.
It assumes the caller will wait for all operations to complete at the
end.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>

diff --git a/src/common/Throttle.cc b/src/common/Throttle.cc
index 9b1e5292ec221286b93658b2d9a79dfab75a837b..bfa1ae46040364cf02001f04a98282cf3c3ffa6b 100644 (file)
--- a/src/common/Throttle.cc
+++ b/src/common/Throttle.cc
@@ -1,3 +1,7 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <errno.h>
 
 #include "common/Throttle.h"
 #include "common/dout.h"
@@ -205,3 +209,43 @@ int64_t Throttle::put(int64_t c)
   }
   return count.read();
 }
+
+SimpleThrottle::SimpleThrottle(uint64_t max, bool ignore_enoent)
+  : m_lock("SimpleThrottle"),
+    m_max(max),
+    m_current(0),
+    m_ret(0),
+    m_ignore_enoent(ignore_enoent)
+{
+}
+
+SimpleThrottle::~SimpleThrottle()
+{
+  Mutex::Locker l(m_lock);
+  assert(m_current == 0);
+}
+
+void SimpleThrottle::start_op()
+{
+  Mutex::Locker l(m_lock);
+  while (m_max == m_current)
+    m_cond.Wait(m_lock);
+  ++m_current;
+}
+
+void SimpleThrottle::end_op(int r)
+{
+  Mutex::Locker l(m_lock);
+  --m_current;
+  if (r < 0 && !m_ret && (r != -ENOENT || m_ignore_enoent))
+    m_ret = r;
+  m_cond.Signal();
+}
+
+int SimpleThrottle::wait_for_ret()
+{
+  Mutex::Locker l(m_lock);
+  while (m_current > 0)
+    m_cond.Wait(m_lock);
+  return m_ret;
+}

diff --git a/src/common/Throttle.h b/src/common/Throttle.h
index a89783fdb77554067807a6af371642693cb28d4b..192a65a072f8ee1f5d4b7ddc64c832c660bff643 100644 (file)
--- a/src/common/Throttle.h
+++ b/src/common/Throttle.h
@@ -59,4 +59,41 @@ public:
 };
 
 
+/**
+ * @class SimpleThrottle
+ * This is a simple way to bound the number of concurrent operations.
+ *
+ * It tracks the first error encountered, and makes it available
+ * when all requests are complete. wait_for_ret() should be called
+ * before the instance is destroyed.
+ *
+ * Re-using the same instance isn't safe if you want to check each set
+ * of operations for errors, since the return value is not reset.
+ */
+class SimpleThrottle {
+public:
+  SimpleThrottle(uint64_t max, bool ignore_enoent);
+  ~SimpleThrottle();
+  void start_op();
+  void end_op(int r);
+  int wait_for_ret();
+private:
+  Mutex m_lock;
+  Cond m_cond;
+  uint64_t m_max;
+  uint64_t m_current;
+  int m_ret;
+  bool m_ignore_enoent;
+};
+
+class C_SimpleThrottle : public Context {
+public:
+  C_SimpleThrottle(SimpleThrottle *throttle) : m_throttle(throttle) {}
+  virtual void finish(int r) {
+    m_throttle->end_op(r);
+  }
+private:
+  SimpleThrottle *m_throttle;
+};
+
 #endif