From: Haomai Wang <haomai@xsky.com>
Date: Wed, 25 May 2016 01:45:52 +0000 (+0800)
Subject: Event: add submit_to apis
X-Git-Tag: ses5-milestone5~574^2~12
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=c33753e97e1e7bc86408b6b69812f4a620cd2fcb;p=ceph.git

Event: add submit_to apis

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

diff --git a/src/msg/async/Event.cc b/src/msg/async/Event.cc
index 9cd1bc051df8..df53bee4de2e 100644
--- a/src/msg/async/Event.cc
+++ b/src/msg/async/Event.cc
@@ -59,6 +59,7 @@ ostream& EventCenter::_event_prefix(std::ostream *_dout)
                 << " time_id=" << time_event_next_id << ").";
 }
 
+EventCenter *EventCenter::centers[MAX_EVENTCENTER];
 thread_local EventCenter* local_center = nullptr;
 
 int EventCenter::init(int n, unsigned i)
@@ -142,7 +143,7 @@ EventCenter::~EventCenter()
 
 void EventCenter::set_owner()
 {
-  local_center = this;
+  centers[id] = local_center = this;
 }
 
 int EventCenter::create_file_event(int fd, int mask, EventCallbackRef ctxt)
diff --git a/src/msg/async/Event.h b/src/msg/async/Event.h
index 353812e3d5de..f02e113a7b9f 100644
--- a/src/msg/async/Event.h
+++ b/src/msg/async/Event.h
@@ -91,6 +91,10 @@ inline EventCenter* center() {
  */
 class EventCenter {
   using clock_type = ceph::coarse_mono_clock;
+  // should be enough;
+  static const int MAX_EVENTCENTER = 24;
+  static EventCenter *centers[MAX_EVENTCENTER];
+
   struct FileEvent {
     int mask;
     EventCallbackRef read_cb;
@@ -159,6 +163,53 @@ class EventCenter {
   inline bool in_thread() const {
     return local_center == this;
   }
+
+ private:
+  template <typename func>
+  class C_submit_event : public EventCallback {
+    std::mutex lock;
+    std::condition_variable cond;
+    bool done = false;
+    func f;
+    bool nonwait;
+   public:
+    C_submit_event(func &&_f, bool nw)
+      : f(std::move(_f)), nonwait(nw) {}
+    void do_request(int id) {
+      f();
+      lock.lock();
+      cond.notify_all();
+      done = true;
+      lock.unlock();
+      if (nonwait)
+        delete this;
+    }
+    void wait() {
+      assert(!nonwait);
+      std::unique_lock<std::mutex> l(lock);
+      while (!done)
+        cond.wait(l);
+    }
+  };
+
+ public:
+  template <typename func>
+  static void submit_to(int i, func &&f, bool nowait = false) {
+    assert(i < MAX_EVENTCENTER);
+    EventCenter *c = centers[i];
+    if (c->in_thread()) {
+      f();
+      return ;
+    }
+    if (nowait) {
+      C_submit_event<func> *event = new C_submit_event<func>(std::move(f), true);
+      c->dispatch_event_external(event);
+    } else {
+      C_submit_event<func> event(std::move(f), false);
+      c->dispatch_event_external(&event);
+      event.wait();
+    }
+  };
 };
 
 #endif