Wip async abstract backend

src/CMakeLists.txt

@@ -424,6 +424,8 @@ set(libcommon_files
   msg/async/Event.cc
   msg/async/EventEpoll.cc
   msg/async/EventSelect.cc
+  msg/async/Stack.cc
+  msg/async/PosixStack.cc
   msg/async/net_handler.cc
   ${xio_common_srcs}
   msg/msg_types.cc

src/common/ceph_context.cc

@@ -472,6 +472,7 @@ CephContext::CephContext(uint32_t module_type_, int init_flags_)
 {
   ceph_spin_init(&_service_thread_lock);
   ceph_spin_init(&_associated_objs_lock);
+  ceph_spin_init(&_fork_watchers_lock);
   ceph_spin_init(&_feature_lock);
   ceph_spin_init(&_cct_perf_lock);
 
@@ -575,6 +576,7 @@ CephContext::~CephContext()
 
   delete _conf;
   ceph_spin_destroy(&_service_thread_lock);
+  ceph_spin_destroy(&_fork_watchers_lock);
   ceph_spin_destroy(&_associated_objs_lock);
   ceph_spin_destroy(&_feature_lock);
   ceph_spin_destroy(&_cct_perf_lock);

src/common/ceph_context.h

@@ -177,6 +177,33 @@ class CephContext {
     return _set_gid_string;
   }
 
+  class ForkWatcher {
+   public:
+    virtual ~ForkWatcher() {}
+    virtual void handle_pre_fork() = 0;
+    virtual void handle_post_fork() = 0;
+  };
+
+  void register_fork_watcher(ForkWatcher *w) {
+    ceph_spin_lock(&_fork_watchers_lock);
+    _fork_watchers.push_back(w);
+    ceph_spin_unlock(&_fork_watchers_lock);
+  }
+
+  void notify_pre_fork() {
+    ceph_spin_lock(&_fork_watchers_lock);
+    for (auto &&t : _fork_watchers)
+      t->handle_pre_fork();
+    ceph_spin_unlock(&_fork_watchers_lock);
+  }
+
+  void notify_post_fork() {
+    ceph_spin_lock(&_fork_watchers_lock);
+    for (auto &&t : _fork_watchers)
+      t->handle_post_fork();
+    ceph_spin_unlock(&_fork_watchers_lock);
+  }
+
 private:
   struct SingletonWrapper : boost::noncopyable {
     virtual ~SingletonWrapper() {}
@@ -235,6 +262,9 @@ class CephContext {
   ceph_spinlock_t _associated_objs_lock;
   std::map<std::string, SingletonWrapper*> _associated_objs;
 
+  ceph_spinlock_t _fork_watchers_lock;
+  std::vector<ForkWatcher*> _fork_watchers;
+
   // crypto
   CryptoHandler *_crypto_none;
   CryptoHandler *_crypto_aes;

src/common/config_opts.h

@@ -197,8 +197,9 @@ OPTION(ms_inject_delay_probability, OPT_DOUBLE, 0) // range [0, 1]
 OPTION(ms_inject_internal_delays, OPT_DOUBLE, 0)   // seconds
 OPTION(ms_dump_on_send, OPT_BOOL, false)           // hexdump msg to log on send
 OPTION(ms_dump_corrupt_message_level, OPT_INT, 1)  // debug level to hexdump undecodeable messages at
-OPTION(ms_async_op_threads, OPT_INT, 3)            // number of worker processing threads for async messenger created on init
-OPTION(ms_async_max_op_threads, OPT_INT, 5)        // max number of worker processing threads for async messenger
+OPTION(ms_async_transport_type, OPT_STR, "posix")
+OPTION(ms_async_op_threads, OPT_U64, 3)            // number of worker processing threads for async messenger created on init
+OPTION(ms_async_max_op_threads, OPT_U64, 5)        // max number of worker processing threads for async messenger
 OPTION(ms_async_set_affinity, OPT_BOOL, true)
 // example: ms_async_affinity_cores = 0,1
 // The number of coreset is expected to equal to ms_async_op_threads, otherwise

src/global/global_init.cc

@@ -339,6 +339,7 @@ int global_init_prefork(CephContext *cct)
     return -1;
   }
 
+  cct->notify_pre_fork();
   // stop log thread
   cct->_log->flush();
   cct->_log->stop();
@@ -370,6 +371,7 @@ void global_init_postfork_start(CephContext *cct)
 {
   // restart log thread
   cct->_log->start();
+  cct->notify_post_fork();
 
   /* This is the old trick where we make file descriptors 0, 1, and possibly 2
    * point to /dev/null.

src/msg/Makefile.am

@@ -22,6 +22,8 @@ libmsg_la_SOURCES += \
 	msg/async/AsyncMessenger.cc \
 	msg/async/Event.cc \
 	msg/async/net_handler.cc \
+	msg/async/Stack.cc \
+	msg/async/PosixStack.cc \
 	msg/async/EventSelect.cc
 
 if LINUX
@@ -47,6 +49,8 @@ noinst_HEADERS += \
 	msg/async/Event.h \
 	msg/async/EventEpoll.h \
 	msg/async/EventSelect.h \
+	msg/async/Stack.h \
+	msg/async/PosixStack.h \
 	msg/async/net_handler.h
 
 if LINUX