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 9cd1bc051df85f2d0ad548aaa37a02f502002c19..df53bee4de2ef18a5d7adf2eaee088457e3b2d9a 100644 (file)
--- 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 353812e3d5decb638b2fdcf7a93e32d02b5270bd..f02e113a7b9febd0d7770a74f8eea7b613824c28 100644 (file)
--- 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