msg/async/rdma: make RDMADispatch thread spawn after forking

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

diff --git a/src/msg/async/rdma/RDMAStack.cc b/src/msg/async/rdma/RDMAStack.cc
index 3b2c6cc84b9527597ef5538c8ca11e2b5b54ded0..3fdb1fc5591d321880d550e02e7c8cf15d266e9a 100644 (file)
--- a/src/msg/async/rdma/RDMAStack.cc
+++ b/src/msg/async/rdma/RDMAStack.cc
@@ -75,10 +75,12 @@ RDMAStack::RDMAStack(CephContext *cct, const string &t): NetworkStack(cct, t)
 
 void RDMAWorker::initialize()
 {
-  dispatcher = stack->get_dispatcher();
-  notify_fd = dispatcher->register_worker(this);
-  center.create_file_event(notify_fd, EVENT_READABLE, tx_handler);
-  memory_manager = infiniband->get_memory_manager();
+  if (!dispatcher) {
+    dispatcher = stack->get_dispatcher();
+    notify_fd = dispatcher->register_worker(this);
+    center.create_file_event(notify_fd, EVENT_READABLE, tx_handler);
+    memory_manager = infiniband->get_memory_manager();
+  }
 }
 
 int RDMAWorker::reserve_message_buffer(RDMAConnectedSocketImpl *o, std::vector<Chunk*> &c, size_t bytes)

diff --git a/src/msg/async/rdma/RDMAStack.h b/src/msg/async/rdma/RDMAStack.h
index dc5bc1a44e563415a37a04aa2e511998922f6883..eef848ce6b9f39c9099c9b558a41d2da03c59cef 100644 (file)
--- a/src/msg/async/rdma/RDMAStack.h
+++ b/src/msg/async/rdma/RDMAStack.h
@@ -34,7 +34,7 @@ class RDMAServerSocketImpl;
 class RDMAStack;
 class RDMAWorker;
 
-class RDMADispatcher {
+class RDMADispatcher : public CephContext::ForkWatcher {
   typedef Infiniband::MemoryManager::Chunk Chunk;
   typedef Infiniband::QueuePair QueuePair;
 
@@ -92,6 +92,7 @@ class RDMADispatcher {
     rx_cq = ib->create_comp_queue(rx_cc);
     assert(rx_cq);
     t = std::thread(&RDMADispatcher::polling, this);
+    cct->register_fork_watcher(this);
   }
   virtual ~RDMADispatcher();
   void handle_async_event();
@@ -155,6 +156,14 @@ class RDMADispatcher {
   }
   Infiniband::CompletionQueue* get_rx_cq() const { return rx_cq; }
   void notify_pending_workers();
+  virtual void handle_pre_fork() override {
+    done = true;
+    t.join();
+    done = false;
+  }
+  virtual void handle_post_fork() override {
+    t = std::thread(&RDMADispatcher::polling, this);
+  }
 };
 
 
@@ -169,7 +178,7 @@ class RDMAWorker : public Worker {
   EventCallbackRef tx_handler;
   MemoryManager *memory_manager;
   std::list<RDMAConnectedSocketImpl*> pending_sent_conns;
-  RDMADispatcher* dispatcher;
+  RDMADispatcher* dispatcher = nullptr;
   int notify_fd = -1;
   Mutex lock;
   std::vector<ibv_wc> wc;