Allow deferring functions to the epoll(2) thread

Summary:
In the current gloo design, `Loop` uses `epoll` via file descriptors to handle events.
For listening port, the mesh connection currently leads to O(mn) port usage where m is the num of ranks per host and n is the total num of ranks.

In later diffs of the stack we'll introduce a listener role to remove this quadratic port usage problem as TCP is a 4-tuple so we can reuse the same local port as long as the remote address is different.

To do that we'll introduce a listener role in the mesh connection. The listener will accept all incoming connections, and then allow the caller (pair) to handle the post rendezvous socket events. We would still use `Pair` to abstract away the mesh conn details but in order for the listening side of a pair to register a callback after connection establishment, we introduce the `defer()` function such that `Pair` can register the callback to be fired in next epoll tick.

Credit to original author Pieter Noordhuis (pietern)

This diff basically just resolves the conflicts.

Differential Revision: D45434482

fbshipit-source-id: c2bb99653820b0e5ad7403ec9bbdd1d616e2d50f

Author: xunnanxu

gloo/transport/tcp/loop.cc

@@ -8,6 +8,7 @@
 
 #include <gloo/transport/tcp/loop.h>
 
+#include <fcntl.h>
 #include <string.h>
 #include <unistd.h>
 
@@ -40,10 +41,69 @@ namespace gloo {
 namespace transport {
 namespace tcp {
 
+Deferrables::Deferrables() {
+  std::array<int, 2> fds;
+  auto rv = pipe2(fds.data(), O_NONBLOCK);
+  GLOO_ENFORCE_NE(rv, -1, "pipe: ", strerror(errno));
+  rfd_ = fds[0];
+  wfd_ = fds[1];
+}
+
+Deferrables::~Deferrables() {
+  close(rfd_);
+  close(wfd_);
+}
+
+void Deferrables::defer(function_t fn) {
+  std::lock_guard<std::mutex> guard(mutex_);
+  functions_.push_back(std::move(fn));
+
+  // Write byte to pipe to make epoll(2) wake up.
+  if (!triggered_) {
+    for (;;) {
+      char byte = 0;
+      auto rv = write(wfd_, &byte, sizeof(byte));
+      if (rv == -1 && errno == EINTR) {
+        continue;
+      }
+      GLOO_ENFORCE_NE(rv, -1, "write: ", strerror(errno));
+      break;
+    }
+    triggered_ = true;
+  }
+}
+
+void Deferrables::handleEvents(int /* unused */) {
+  decltype(functions_) localFunctions;
+
+  {
+    std::lock_guard<std::mutex> guard(mutex_);
+    std::swap(localFunctions, functions_);
+
+    // Read byte from pipe to drain it.
+    for (;;) {
+      char byte = 0;
+      auto rv = read(rfd_, &byte, sizeof(byte));
+      if (rv == -1 && errno == EINTR) {
+        continue;
+      }
+      GLOO_ENFORCE_NE(rv, -1, "read: ", strerror(errno));
+      break;
+    }
+    triggered_ = false;
+  }
+
+  // Execute deferred functions.
+  for (auto fn : localFunctions) {
+    fn();
+  }
+}
+
 Loop::Loop() {
   fd_ = epoll_create(1);
   GLOO_ENFORCE_NE(fd_, -1, "epoll_create: ", strerror(errno));
   loop_.reset(new std::thread(&Loop::run, this));
+  registerDescriptor(deferrables_.rfd_, EPOLLIN, &deferrables_);
 }
 
 Loop::~Loop() {
@@ -81,6 +141,10 @@ void Loop::unregisterDescriptor(int fd, Handler* h) {
   }
 }
 
+void Loop::defer(std::function<void()> fn) {
+  deferrables_.defer(std::move(fn));
+}
+
 void Loop::run() {
   std::array<struct epoll_event, capacity_> events;
   int nfds;

gloo/transport/tcp/loop.h

@@ -10,6 +10,8 @@
 
 #include <atomic>
 #include <condition_variable>
+#include <functional>
+#include <list>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -32,6 +34,33 @@ class Handler {
   virtual void handleEvents(int events) = 0;
 };
 
+class Loop;
+
+// Functions can be deferred to the epoll(2) thread through the this
+// class. It uses readability of a pipe to wake up the event loop.
+class Deferrables final : public Handler {
+public:
+  using function_t = std::function<void()>;
+
+  Deferrables();
+
+  ~Deferrables() override;
+
+  void defer(function_t fn);
+
+  void handleEvents(int events) override;
+
+private:
+  int rfd_;
+  int wfd_;
+
+  std::mutex mutex_;
+  std::list<function_t> functions_;
+  bool triggered_{false};
+
+  friend class Loop;
+};
+
 class Loop final : public std::enable_shared_from_this<Loop> {
  public:
   explicit Loop();
@@ -42,13 +71,16 @@ class Loop final : public std::enable_shared_from_this<Loop> {
 
   void unregisterDescriptor(int fd, Handler *h);
 
+  void defer(std::function<void()> fn);
+
   void run();
 
  private:
   static constexpr auto capacity_ = 64;
 
   int fd_{-1};
   std::atomic<bool> done_{false};
+  Deferrables deferrables_;
   std::unique_ptr<std::thread> loop_;
 
   std::mutex m_;