msg/async/Event: add Poller to allow event thread spin on Poller::poll

Signed-off-by: Haomai Wang <haomai@xsky.com>

diff --git a/src/msg/async/Event.cc b/src/msg/async/Event.cc
index a2e7f8833baa7d1e37b61c367e9cb0264674c164..1c35e9daa8d54b16451f0c64d26fcc2c5584b9b2 100644 (file)
--- a/src/msg/async/Event.cc
+++ b/src/msg/async/Event.cc
@@ -53,6 +53,42 @@ class C_handle_notify : public EventCallback {
 #undef dout_prefix
 #define dout_prefix _event_prefix(_dout)
 
+/**
+ * Construct a Poller.
+ *
+ * \param center
+ *      EventCenter object through which the poller will be invoked (defaults
+ *      to the global #RAMCloud::center object).
+ * \param pollerName
+ *      Human readable name that can be printed out in debugging messages
+ *      about the poller. The name of the superclass is probably sufficient
+ *      for most cases.
+ */
+EventCenter::Poller::Poller(EventCenter* center, const string& name)
+    : owner(center), poller_name(name), slot(owner->pollers.size())
+{
+  owner->pollers.push_back(this);
+}
+
+/**
+ * Destroy a Poller.
+ */
+EventCenter::Poller::~Poller()
+{
+  // Erase this Poller from the vector by overwriting it with the
+  // poller that used to be the last one in the vector.
+  //
+  // Note: this approach is reentrant (it is safe to delete a
+  // poller from a poller callback, which means that the poll
+  // method is in the middle of scanning the list of all pollers;
+  // the worst that will happen is that the poller that got moved
+  // may not be invoked in the current scan).
+  owner->pollers[slot] = owner->pollers.back();
+  owner->pollers[slot]->slot = slot;
+  owner->pollers.pop_back();
+  slot = -1;
+}
+
 ostream& EventCenter::_event_prefix(std::ostream *_dout)
 {
   return *_dout << "Event(" << this << " nevent=" << nevent
@@ -261,8 +297,11 @@ void EventCenter::delete_time_event(uint64_t id)
 
 void EventCenter::wakeup()
 {
-  ldout(cct, 2) << __func__ << dendl;
+  // No need to wake up since we never sleep
+  if (!pollers.empty())
+    return ;
 
+  ldout(cct, 2) << __func__ << dendl;
   char buf = 'c';
   // wake up "event_wait"
   int n = write(notify_send_fd, &buf, sizeof(buf));
@@ -304,8 +343,9 @@ int EventCenter::process_events(int timeout_microseconds)
   bool trigger_time = false;
   auto now = clock_type::now();
 
-  // If exists external events, don't block
-  if (external_num_events.load()) {
+  bool blocking = pollers.empty() && !external_num_events.load() ? &tv : nullptr;
+  // If exists external events or poller, don't block
+  if (!blocking) {
     tv.tv_sec = 0;
     tv.tv_usec = 0;
   } else {
@@ -374,6 +414,12 @@ int EventCenter::process_events(int timeout_microseconds)
       numevents++;
     }
   }
+
+  if (!numevents && !blocking) {
+    for (uint32_t i = 0; i < pollers.size(); i++)
+      numevents += pollers[i]->poll();
+  }
+
   return numevents;
 }
 

diff --git a/src/msg/async/Event.h b/src/msg/async/Event.h
index d887c5420ce6af4ddfcd6f8e7c2c08177f27cd01..21159fac68e4d8480e2e32e9e38a33ab934c0c0f 100644 (file)
--- a/src/msg/async/Event.h
+++ b/src/msg/async/Event.h
@@ -112,6 +112,44 @@ class EventCenter {
     TimeEvent(): id(0), time_cb(NULL) {}
   };
 
+ public:
+  /**
+     * A Poller object is invoked once each time through the dispatcher's
+     * inner polling loop.
+     */
+  class Poller {
+   public:
+    explicit Poller(EventCenter* center, const string& pollerName);
+    virtual ~Poller();
+
+    /**
+     * This method is defined by a subclass and invoked once by the
+     * center during each pass through its inner polling loop.
+     *
+     * \return
+     *      1 means that this poller did useful work during this call.
+     *      0 means that the poller found no work to do.
+     */
+    virtual int poll() = 0;
+
+   private:
+    /// The EventCenter object that owns this Poller.  NULL means the
+    /// EventCenter has been deleted.
+    EventCenter* owner;
+
+    /// Human-readable string name given to the poller to make it
+    /// easy to identify for debugging. For most pollers just passing
+    /// in the subclass name probably makes sense.
+    string poller_name;
+
+    /// Index of this Poller in EventCenter::pollers.  Allows deletion
+    /// without having to scan all the entries in pollers. -1 means
+    /// this poller isn't currently in EventCenter::pollers (happens
+    /// after EventCenter::reset).
+    int slot;
+  };
+
+ private:
   CephContext *cct;
   int nevent;
   // Used only to external event
@@ -122,6 +160,10 @@ class EventCenter {
   vector<FileEvent> file_events;
   EventDriver *driver;
   std::multimap<clock_type::time_point, TimeEvent> time_events;
+  // Keeps track of all of the pollers currently defined.  We don't
+  // use an intrusive list here because it isn't reentrant: we need
+  // to add/remove elements while the center is traversing the list.
+  std::vector<Poller*> pollers;
   std::map<uint64_t, std::multimap<clock_type::time_point, TimeEvent>::iterator> event_map;
   uint64_t time_event_next_id;
   int notify_receive_fd;