Add REUSEPORT option to AsyncServerSocket

Summary:
Adds a reuse port option to AsyncServerSocket, so multiple sockets can bind to the same accept port.  Allows for multiple accept threads, so accepts can be greater, since there is no longer a single accept lock.

reuse port option is ifdefd, since this wouldn't build with some of the older kernels otherwise.

Postponed overnight

Test Plan: Builds.  Used in an upcoming diff.

Reviewed By: jsedgwick@fb.com

Subscribers: benj, trunkagent, doug, njormrod, folly-diffs@

FB internal diff: D1710600

Tasks: 5488516, 5788110

Signature: t1:1710600:1418066966:627e03857f9b5ff831f2922add08e90cc525c95c

folly/io/async/AsyncServerSocket.cpp

@@ -574,8 +574,18 @@ void AsyncServerSocket::setupSocket(int fd) {
     LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
   }
 
-  // Set keepalive as desired
+  // Set reuseport to support multiple accept threads
   int zero = 0;
+  if (reusePortEnabled_ &&
+      setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
+    LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
+               << strerror(errno);
+    folly::throwSystemError(errno,
+                            "failed to bind to async server socket: " +
+                            address.describe());
+  }
+
+  // Set keepalive as desired
   if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
                  (keepAliveEnabled_) ? &one : &zero, sizeof(int)) != 0) {
     LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: " <<

folly/io/async/AsyncServerSocket.h

@@ -30,6 +30,13 @@
 #include <stddef.h>
 #include <sys/socket.h>
 
+
+// Due to the way kernel headers are included, this may or may not be defined.
+// Number pulled from 3.10 kernel headers.
+#ifndef SO_REUSEPORT
+#define SO_REUSEPORT 15
+#endif
+
 namespace folly {
 
 /**
@@ -510,6 +517,36 @@ class AsyncServerSocket : public DelayedDestruction {
     return keepAliveEnabled_;
   }
 
+  /**
+   * Set whether or not SO_REUSEPORT should be enabled on the server socket,
+   * allowing multiple binds to the same port
+   */
+  void setReusePortEnabled(bool enabled) {
+    reusePortEnabled_ = enabled;
+
+    for (auto& handler : sockets_) {
+      if (handler.socket_ < 0) {
+        continue;
+      }
+
+      int val = (enabled) ? 1 : 0;
+      if (setsockopt(handler.socket_, SOL_SOCKET,
+                     SO_REUSEPORT, &val, sizeof(val)) != 0) {
+        LOG(ERROR) <<
+          "failed to set SO_REUSEPORT on async server socket " << errno;
+        folly::throwSystemError(errno,
+                                "failed to bind to async server socket");
+      }
+    }
+  }
+
+  /**
+   * Get whether or not SO_REUSEPORT is enabled on the server socket.
+   */
+  bool getReusePortEnabled_() const {
+    return reusePortEnabled_;
+  }
+
   /**
    * Set whether or not the socket should close during exec() (FD_CLOEXEC). By
    * default, this is enabled
@@ -675,6 +712,7 @@ class AsyncServerSocket : public DelayedDestruction {
   BackoffTimeout *backoffTimeout_;
   std::vector<CallbackInfo> callbacks_;
   bool keepAliveEnabled_;
+  bool reusePortEnabled_{false};
   bool closeOnExec_;
   ShutdownSocketSet* shutdownSocketSet_;
 };

folly/io/async/test/AsyncSocketTest.cpp

@@ -16,14 +16,17 @@
 #include <iostream>
 
 #include <folly/io/async/AsyncSocket.h>
+#include <folly/io/async/AsyncServerSocket.h>
 #include <folly/io/async/EventBase.h>
 
 #include <gtest/gtest.h>
 
+namespace folly {
+
 TEST(AsyncSocketTest, getSockOpt) {
-  folly::EventBase evb;
-  std::shared_ptr<folly::AsyncSocket> socket =
-    folly::AsyncSocket::newSocket(&evb, 0);
+  EventBase evb;
+  std::shared_ptr<AsyncSocket> socket =
+    AsyncSocket::newSocket(&evb, 0);
 
   int val;
   socklen_t len;
@@ -34,3 +37,31 @@ TEST(AsyncSocketTest, getSockOpt) {
 
   EXPECT_EQ(expectedRc, actualRc);
 }
+
+TEST(AsyncSocketTest, REUSEPORT) {
+  EventBase base;
+  auto serverSocket = AsyncServerSocket::newSocket(&base);
+  serverSocket->bind(0);
+  serverSocket->listen(0);
+  serverSocket->startAccepting();
+
+  try {
+    serverSocket->setReusePortEnabled(true);
+  } catch(...) {
+    LOG(INFO) << "Reuse port probably not supported";
+    return;
+  }
+
+  SocketAddress address;
+  serverSocket->getAddress(&address);
+  int port = address.getPort();
+
+  auto serverSocket2 = AsyncServerSocket::newSocket(&base);
+  serverSocket2->setReusePortEnabled(true);
+  serverSocket2->bind(port);
+  serverSocket2->listen(0);
+  serverSocket2->startAccepting();
+
+}
+
+} // namespace