X-Git-Url: http://plrg.eecs.uci.edu/git/?a=blobdiff_plain;f=folly%2Fio%2Fasync%2FAsyncServerSocket.cpp;h=985f3d3fefea0a230bb2d1cc3b12c85c8864c6e0;hb=dc1c3dcc04baa70029d16351d4424af4a7ef4385;hp=ea2a46d9025a282e723b74137c8f584d66711467;hpb=ef2fa20f75bca8b00d4fd8a12e75ce14a0a30040;p=folly.git

diff --git a/folly/io/async/AsyncServerSocket.cpp b/folly/io/async/AsyncServerSocket.cpp
index ea2a46d9..985f3d3f 100644
--- a/folly/io/async/AsyncServerSocket.cpp
+++ b/folly/io/async/AsyncServerSocket.cpp
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Facebook, Inc.
+ * Copyright 2017 Facebook, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,17 +20,22 @@
 
 #include <folly/io/async/AsyncServerSocket.h>
 
+#include <folly/FileUtil.h>
+#include <folly/Portability.h>
+#include <folly/SocketAddress.h>
+#include <folly/String.h>
+#include <folly/detail/SocketFastOpen.h>
 #include <folly/io/async/EventBase.h>
 #include <folly/io/async/NotificationQueue.h>
-#include <folly/SocketAddress.h>
+#include <folly/portability/Fcntl.h>
+#include <folly/portability/Sockets.h>
+#include <folly/portability/Unistd.h>
 
 #include <errno.h>
 #include <string.h>
-#include <unistd.h>
-#include <fcntl.h>
 #include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/tcp.h>
+
+namespace fsp = folly::portability::sockets;
 
 namespace folly {
 
@@ -85,11 +90,14 @@ void AsyncServerSocket::RemoteAcceptor::stop(
 }
 
 void AsyncServerSocket::RemoteAcceptor::messageAvailable(
-  QueueMessage&& msg) {
-
+    QueueMessage&& msg) noexcept {
   switch (msg.type) {
     case MessageType::MSG_NEW_CONN:
     {
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onConnectionDequeuedByAcceptorCallback(
+            msg.fd, msg.address);
+      }
       callback_->connectionAccepted(msg.fd, msg.address);
       break;
     }
@@ -115,13 +123,12 @@ void AsyncServerSocket::RemoteAcceptor::messageAvailable(
  */
 class AsyncServerSocket::BackoffTimeout : public AsyncTimeout {
  public:
-  BackoffTimeout(AsyncServerSocket* socket)
-    : AsyncTimeout(socket->getEventBase()),
-      socket_(socket) {}
+  // Disallow copy, move, and default constructors.
+  BackoffTimeout(BackoffTimeout&&) = delete;
+  explicit BackoffTimeout(AsyncServerSocket* socket)
+      : AsyncTimeout(socket->getEventBase()), socket_(socket) {}
 
-  virtual void timeoutExpired() noexcept {
-    socket_->backoffTimeoutExpired();
-  }
+  void timeoutExpired() noexcept override { socket_->backoffTimeoutExpired(); }
 
  private:
   AsyncServerSocket* socket_;
@@ -175,23 +182,29 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) {
     VLOG(10) << "AsyncServerSocket::stopAccepting " << this <<
               handler.socket_;
   }
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
-  // When destroy is called, unregister and close the socket immediately
+  // When destroy is called, unregister and close the socket immediately.
   accepting_ = false;
 
-  for (auto& handler : sockets_) {
+  // Close the sockets in reverse order as they were opened to avoid
+  // the condition where another process concurrently tries to open
+  // the same port, succeed to bind the first socket but fails on the
+  // second because it hasn't been closed yet.
+  for (; !sockets_.empty(); sockets_.pop_back()) {
+    auto& handler = sockets_.back();
     handler.unregisterHandler();
     if (shutdownSocketSet_) {
       shutdownSocketSet_->close(handler.socket_);
     } else if (shutdownFlags >= 0) {
-      result = ::shutdown(handler.socket_, shutdownFlags);
+      result = shutdownNoInt(handler.socket_, shutdownFlags);
       pendingCloseSockets_.push_back(handler.socket_);
     } else {
-      ::close(handler.socket_);
+      closeNoInt(handler.socket_);
     }
   }
-  sockets_.clear();
 
   // Destroy the backoff timout.  This will cancel it if it is running.
   delete backoffTimeout_;
@@ -208,7 +221,14 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) {
   for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
        it != callbacksCopy.end();
        ++it) {
-    it->consumer->stop(it->eventBase, it->callback);
+    // consumer may not be set if we are running in primary event base
+    if (it->consumer) {
+      DCHECK(it->eventBase);
+      it->consumer->stop(it->eventBase, it->callback);
+    } else {
+      DCHECK(it->callback);
+      it->callback->acceptStopped();
+    }
   }
 
   return result;
@@ -216,8 +236,8 @@ int AsyncServerSocket::stopAccepting(int shutdownFlags) {
 
 void AsyncServerSocket::destroy() {
   stopAccepting();
-  for (auto s: pendingCloseSockets_) {
-    ::close(s);
+  for (auto s : pendingCloseSockets_) {
+    closeNoInt(s);
   }
   // Then call DelayedDestruction::destroy() to take care of
   // whether or not we need immediate or delayed destruction
@@ -226,7 +246,7 @@ void AsyncServerSocket::destroy() {
 
 void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
   assert(eventBase_ == nullptr);
-  assert(eventBase->isInEventBaseThread());
+  eventBase->dcheckIsInEventBaseThread();
 
   eventBase_ = eventBase;
   for (auto& handler : sockets_) {
@@ -236,7 +256,7 @@ void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
 
 void AsyncServerSocket::detachEventBase() {
   assert(eventBase_ != nullptr);
-  assert(eventBase_->isInEventBaseThread());
+  eventBase_->dcheckIsInEventBaseThread();
   assert(!accepting_);
 
   eventBase_ = nullptr;
@@ -246,7 +266,9 @@ void AsyncServerSocket::detachEventBase() {
 }
 
 void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   if (sockets_.size() > 0) {
     throw std::invalid_argument(
@@ -262,9 +284,15 @@ void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
     SocketAddress address;
     address.setFromLocalAddress(fd);
 
-    setupSocket(fd);
-    sockets_.push_back(
-      ServerEventHandler(eventBase_, fd, this, address.getFamily()));
+#if __linux__
+    if (noTransparentTls_) {
+      // Ignore return value, errors are ok
+      setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
+    }
+#endif
+
+    setupSocket(fd, address.getFamily());
+    sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
     sockets_.back().changeHandlerFD(fd);
   }
 }
@@ -280,24 +308,33 @@ void AsyncServerSocket::bindSocket(
   sockaddr_storage addrStorage;
   address.getAddress(&addrStorage);
   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
-  if (::bind(fd, saddr, address.getActualSize()) != 0) {
+
+  if (fsp::bind(fd, saddr, address.getActualSize()) != 0) {
     if (!isExistingSocket) {
-      ::close(fd);
+      closeNoInt(fd);
     }
     folly::throwSystemError(errno,
         "failed to bind to async server socket: " +
         address.describe());
   }
 
+#if __linux__
+  if (noTransparentTls_) {
+    // Ignore return value, errors are ok
+    setsockopt(fd, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
+  }
+#endif
+
   // If we just created this socket, update the EventHandler and set socket_
   if (!isExistingSocket) {
-    sockets_.push_back(
-      ServerEventHandler(eventBase_, fd, this, address.getFamily()));
+    sockets_.emplace_back(eventBase_, fd, this, address.getFamily());
   }
 }
 
 void AsyncServerSocket::bind(const SocketAddress& address) {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   // useExistingSocket() may have been called to initialize socket_ already.
   // However, in the normal case we need to create a new socket now.
@@ -345,38 +382,38 @@ void AsyncServerSocket::bind(
 }
 
 void AsyncServerSocket::bind(uint16_t port) {
-  struct addrinfo hints, *res, *res0;
+  struct addrinfo hints, *res0;
   char sport[sizeof("65536")];
 
   memset(&hints, 0, sizeof(hints));
   hints.ai_family = AF_UNSPEC;
   hints.ai_socktype = SOCK_STREAM;
-  hints.ai_flags = AI_PASSIVE;
+  hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
   snprintf(sport, sizeof(sport), "%u", port);
 
-  if (getaddrinfo(nullptr, sport, &hints, &res0)) {
+  // On Windows the value we need to pass to bind to all available
+  // addresses is an empty string. Everywhere else, it's nullptr.
+  constexpr const char* kWildcardNode = kIsWindows ? "" : nullptr;
+  if (getaddrinfo(kWildcardNode, sport, &hints, &res0)) {
     throw std::invalid_argument(
                               "Attempted to bind address to socket with "
                               "bad getaddrinfo");
   }
 
-  folly::ScopeGuard guard = folly::makeGuard([&]{
-      freeaddrinfo(res0);
-    });
-  DCHECK(&guard);
+  SCOPE_EXIT { freeaddrinfo(res0); };
 
-  for (res = res0; res; res = res->ai_next) {
-    int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+  auto setupAddress = [&] (struct addrinfo* res) {
+    int s = fsp::socket(res->ai_family, res->ai_socktype, res->ai_protocol);
     // IPv6/IPv4 may not be supported by the kernel
     if (s < 0 && errno == EAFNOSUPPORT) {
-      continue;
+      return;
     }
-    CHECK(s);
+    CHECK_GE(s, 0);
 
     try {
-      setupSocket(s);
+      setupSocket(s, res->ai_family);
     } catch (...) {
-      ::close(s);
+      closeNoInt(s);
       throw;
     }
 
@@ -386,19 +423,87 @@ void AsyncServerSocket::bind(uint16_t port) {
                             &v6only, sizeof(v6only)));
     }
 
+    // Bind to the socket
+    if (fsp::bind(s, res->ai_addr, socklen_t(res->ai_addrlen)) != 0) {
+      folly::throwSystemError(
+          errno,
+          "failed to bind to async server socket for port ",
+          SocketAddress::getPortFrom(res->ai_addr),
+          " family ",
+          SocketAddress::getFamilyNameFrom(res->ai_addr, "<unknown>"));
+    }
+
+#if __linux__
+    if (noTransparentTls_) {
+      // Ignore return value, errors are ok
+      setsockopt(s, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
+    }
+#endif
+
     SocketAddress address;
     address.setFromLocalAddress(s);
 
-    sockets_.push_back(
-      ServerEventHandler(eventBase_, s, this, address.getFamily()));
+    sockets_.emplace_back(eventBase_, s, this, address.getFamily());
+  };
 
-    // Bind to the socket
-    if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) {
-      folly::throwSystemError(
-        errno,
-        "failed to bind to async server socket for port");
+  const int kNumTries = 25;
+  for (int tries = 1; true; tries++) {
+    // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
+    // should return IPv6 first and then IPv4 addresses, but glibc's
+    // getaddrinfo(nullptr) with AI_PASSIVE returns:
+    // - 0.0.0.0 (IPv4-only)
+    // - :: (IPv6+IPv4) in this order
+    // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
+    for (struct addrinfo* res = res0; res; res = res->ai_next) {
+      if (res->ai_family == AF_INET6) {
+        setupAddress(res);
+      }
+    }
+
+    // If port == 0, then we should try to bind to the same port on ipv4 and
+    // ipv6.  So if we did bind to ipv6, figure out that port and use it.
+    if (sockets_.size() == 1 && port == 0) {
+      SocketAddress address;
+      address.setFromLocalAddress(sockets_.back().socket_);
+      snprintf(sport, sizeof(sport), "%u", address.getPort());
+      freeaddrinfo(res0);
+      CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
     }
+
+    try {
+      for (struct addrinfo* res = res0; res; res = res->ai_next) {
+        if (res->ai_family != AF_INET6) {
+          setupAddress(res);
+        }
+      }
+    } catch (const std::system_error&) {
+      // If we can't bind to the same port on ipv4 as ipv6 when using
+      // port=0 then we will retry again before giving up after
+      // kNumTries attempts.  We do this by closing the sockets that
+      // were opened, then restarting from scratch.
+      if (port == 0 && !sockets_.empty() && tries != kNumTries) {
+        for (const auto& socket : sockets_) {
+          if (socket.socket_ <= 0) {
+            continue;
+          } else if (shutdownSocketSet_) {
+            shutdownSocketSet_->close(socket.socket_);
+          } else {
+            closeNoInt(socket.socket_);
+          }
+        }
+        sockets_.clear();
+        snprintf(sport, sizeof(sport), "%u", port);
+        freeaddrinfo(res0);
+        CHECK_EQ(0, getaddrinfo(nullptr, sport, &hints, &res0));
+        continue;
+      }
+
+      throw;
+    }
+
+    break;
   }
+
   if (sockets_.size() == 0) {
     throw std::runtime_error(
         "did not bind any async server socket for port");
@@ -406,11 +511,13 @@ void AsyncServerSocket::bind(uint16_t port) {
 }
 
 void AsyncServerSocket::listen(int backlog) {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   // Start listening
   for (auto& handler : sockets_) {
-    if (::listen(handler.socket_, backlog) == -1) {
+    if (fsp::listen(handler.socket_, backlog) == -1) {
       folly::throwSystemError(errno,
                                     "failed to listen on async server socket");
     }
@@ -419,10 +526,10 @@ void AsyncServerSocket::listen(int backlog) {
 
 void AsyncServerSocket::getAddress(SocketAddress* addressReturn) const {
   CHECK(sockets_.size() >= 1);
-  if (sockets_.size() > 1) {
-    VLOG(2) << "Warning: getAddress can return multiple addresses, " <<
-      "but getAddress was called, so only returning the first";
-  }
+  VLOG_IF(2, sockets_.size() > 1)
+    << "Warning: getAddress() called and multiple addresses available ("
+    << sockets_.size() << "). Returning only the first one.";
+
   addressReturn->setFromLocalAddress(sockets_[0].socket_);
 }
 
@@ -440,18 +547,31 @@ std::vector<SocketAddress> AsyncServerSocket::getAddresses()
 void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
                                            EventBase *eventBase,
                                            uint32_t maxAtOnce) {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   // If this is the first accept callback and we are supposed to be accepting,
   // start accepting once the callback is installed.
   bool runStartAccepting = accepting_ && callbacks_.empty();
 
+  callbacks_.emplace_back(callback, eventBase);
+
+  SCOPE_SUCCESS {
+    // If this is the first accept callback and we are supposed to be accepting,
+    // start accepting.
+    if (runStartAccepting) {
+      startAccepting();
+    }
+  };
+
   if (!eventBase) {
-    eventBase = eventBase_; // Run in AsyncServerSocket's eventbase
+    // Run in AsyncServerSocket's eventbase; notify that we are
+    // starting to accept connections
+    callback->acceptStarted();
+    return;
   }
 
-  callbacks_.push_back(CallbackInfo(callback, eventBase));
-
   // Start the remote acceptor.
   //
   // It would be nice if we could avoid starting the remote acceptor if
@@ -463,7 +583,7 @@ void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
   // callback more efficiently without having to use a notification queue.
   RemoteAcceptor* acceptor = nullptr;
   try {
-    acceptor = new RemoteAcceptor(callback);
+    acceptor = new RemoteAcceptor(callback, connectionEventCallback_);
     acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
   } catch (...) {
     callbacks_.pop_back();
@@ -471,17 +591,13 @@ void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
     throw;
   }
   callbacks_.back().consumer = acceptor;
-
-  // If this is the first accept callback and we are supposed to be accepting,
-  // start accepting.
-  if (runStartAccepting) {
-    startAccepting();
-  }
 }
 
 void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
                                               EventBase *eventBase) {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   // Find the matching AcceptCallback.
   // We just do a simple linear search; we don't expect removeAcceptCallback()
@@ -523,7 +639,16 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
     }
   }
 
-  info.consumer->stop(info.eventBase, info.callback);
+  if (info.consumer) {
+    // consumer could be nullptr is we run callbacks in primary event
+    // base
+    DCHECK(info.eventBase);
+    info.consumer->stop(info.eventBase, info.callback);
+  } else {
+    // callback invoked in the primary event base, just call directly
+    DCHECK(info.callback);
+    callback->acceptStopped();
+  }
 
   // If we are supposed to be accepting but the last accept callback
   // was removed, unregister for events until a callback is added.
@@ -535,7 +660,9 @@ void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
 }
 
 void AsyncServerSocket::startAccepting() {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
 
   accepting_ = true;
   if (callbacks_.empty()) {
@@ -553,7 +680,9 @@ void AsyncServerSocket::startAccepting() {
 }
 
 void AsyncServerSocket::pauseAccepting() {
-  assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
+  if (eventBase_) {
+    eventBase_->dcheckIsInEventBaseThread();
+  }
   accepting_ = false;
   for (auto& handler : sockets_) {
    handler. unregisterHandler();
@@ -566,26 +695,21 @@ void AsyncServerSocket::pauseAccepting() {
 }
 
 int AsyncServerSocket::createSocket(int family) {
-  int fd = socket(family, SOCK_STREAM, 0);
+  int fd = fsp::socket(family, SOCK_STREAM, 0);
   if (fd == -1) {
     folly::throwSystemError(errno, "error creating async server socket");
   }
 
   try {
-    setupSocket(fd);
+    setupSocket(fd, family);
   } catch (...) {
-    ::close(fd);
+    closeNoInt(fd);
     throw;
   }
   return fd;
 }
 
-void AsyncServerSocket::setupSocket(int fd) {
-  // Get the address family
-  SocketAddress address;
-  address.setFromLocalAddress(fd);
-  auto family = address.getFamily();
-
+void AsyncServerSocket::setupSocket(int fd, int family) {
   // Put the socket in non-blocking mode
   if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
     folly::throwSystemError(errno,
@@ -605,9 +729,15 @@ void AsyncServerSocket::setupSocket(int fd) {
       setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
     LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
                << strerror(errno);
+#ifdef WIN32
+    folly::throwSystemError(errno, "failed to bind to the async server socket");
+#else
+    SocketAddress address;
+    address.setFromLocalAddress(fd);
     folly::throwSystemError(errno,
                             "failed to bind to async server socket: " +
                             address.describe());
+#endif
   }
 
   // Set keepalive as desired
@@ -636,13 +766,22 @@ void AsyncServerSocket::setupSocket(int fd) {
   }
 #endif
 
+#if FOLLY_ALLOW_TFO
+  if (tfo_ && detail::tfo_enable(fd, tfoMaxQueueSize_) != 0) {
+    // This isn't a fatal error; just log an error message and continue
+    LOG(WARNING) << "failed to set TCP_FASTOPEN on async server socket: "
+                 << folly::errnoStr(errno);
+  }
+#endif
+
   if (shutdownSocketSet_) {
     shutdownSocketSet_->add(fd);
   }
 }
 
-void AsyncServerSocket::handlerReady(
-  uint16_t events, int fd, sa_family_t addressFamily) noexcept {
+void AsyncServerSocket::handlerReady(uint16_t /* events */,
+                                     int fd,
+                                     sa_family_t addressFamily) noexcept {
   assert(!callbacks_.empty());
   DestructorGuard dg(this);
 
@@ -670,10 +809,14 @@ void AsyncServerSocket::handlerReady(
 
     address.setFromSockaddr(saddr, addrLen);
 
+    if (clientSocket >= 0 && connectionEventCallback_) {
+      connectionEventCallback_->onConnectionAccepted(clientSocket, address);
+    }
+
     std::chrono::time_point<std::chrono::steady_clock> nowMs =
       std::chrono::steady_clock::now();
-    int64_t timeSinceLastAccept = std::max(
-      int64_t(0),
+    auto timeSinceLastAccept = std::max<int64_t>(
+      0,
       nowMs.time_since_epoch().count() -
       lastAccepTimestamp_.time_since_epoch().count());
     lastAccepTimestamp_ = nowMs;
@@ -684,7 +827,11 @@ void AsyncServerSocket::handlerReady(
       } else if (rand() > acceptRate_ * RAND_MAX) {
         ++numDroppedConnections_;
         if (clientSocket >= 0) {
-          ::close(clientSocket);
+          closeNoInt(clientSocket);
+          if (connectionEventCallback_) {
+            connectionEventCallback_->onConnectionDropped(clientSocket,
+                                                          address);
+          }
         }
         continue;
       }
@@ -708,15 +855,21 @@ void AsyncServerSocket::handlerReady(
       } else {
         dispatchError("accept() failed", errno);
       }
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onConnectionAcceptError(errno);
+      }
       return;
     }
 
 #ifndef SOCK_NONBLOCK
     // Explicitly set the new connection to non-blocking mode
     if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
-      ::close(clientSocket);
+      closeNoInt(clientSocket);
       dispatchError("failed to set accepted socket to non-blocking mode",
                     errno);
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onConnectionDropped(clientSocket, address);
+      }
       return;
     }
 #endif
@@ -743,6 +896,7 @@ void AsyncServerSocket::dispatchSocket(int socket,
     return;
   }
 
+  const SocketAddress addr(address);
   // Create a message to send over the notification queue
   QueueMessage msg;
   msg.type = MessageType::MSG_NEW_CONN;
@@ -752,9 +906,14 @@ void AsyncServerSocket::dispatchSocket(int socket,
   // Loop until we find a free queue to write to
   while (true) {
     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onConnectionEnqueuedForAcceptorCallback(
+            socket,
+            addr);
+      }
       // Success! return.
       return;
-   }
+    }
 
     // We couldn't add to queue.  Fall through to below
 
@@ -778,7 +937,10 @@ void AsyncServerSocket::dispatchSocket(int socket,
       // even accept new messages.
       LOG(ERROR) << "failed to dispatch newly accepted socket:"
                  << " all accept callback queues are full";
-      ::close(socket);
+      closeNoInt(socket);
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onConnectionDropped(socket, addr);
+      }
       return;
     }
 
@@ -828,12 +990,15 @@ void AsyncServerSocket::enterBackoff() {
   if (backoffTimeout_ == nullptr) {
     try {
       backoffTimeout_ = new BackoffTimeout(this);
-    } catch (const std::bad_alloc& ex) {
+    } catch (const std::bad_alloc&) {
       // Man, we couldn't even allocate the timer to re-enable accepts.
       // We must be in pretty bad shape.  Don't pause accepting for now,
       // since we won't be able to re-enable ourselves later.
       LOG(ERROR) << "failed to allocate AsyncServerSocket backoff"
                  << " timer; unable to temporarly pause accepting";
+      if (connectionEventCallback_) {
+        connectionEventCallback_->onBackoffError();
+      }
       return;
     }
   }
@@ -851,6 +1016,9 @@ void AsyncServerSocket::enterBackoff() {
   if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
     LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
                << "unable to temporarly pause accepting";
+    if (connectionEventCallback_) {
+      connectionEventCallback_->onBackoffError();
+    }
     return;
   }
 
@@ -860,6 +1028,9 @@ void AsyncServerSocket::enterBackoff() {
   for (auto& handler : sockets_) {
     handler.unregisterHandler();
   }
+  if (connectionEventCallback_) {
+    connectionEventCallback_->onBackoffStarted();
+  }
 }
 
 void AsyncServerSocket::backoffTimeoutExpired() {
@@ -868,10 +1039,14 @@ void AsyncServerSocket::backoffTimeoutExpired() {
   // the backoff timeout.
   assert(accepting_);
   // We can't be detached from the EventBase without being paused
-  assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread());
+  assert(eventBase_ != nullptr);
+  eventBase_->dcheckIsInEventBaseThread();
 
   // If all of the callbacks were removed, we shouldn't re-enable accepts
   if (callbacks_.empty()) {
+    if (connectionEventCallback_) {
+      connectionEventCallback_->onBackoffEnded();
+    }
     return;
   }
 
@@ -890,6 +1065,9 @@ void AsyncServerSocket::backoffTimeoutExpired() {
       abort();
     }
   }
+  if (connectionEventCallback_) {
+    connectionEventCallback_->onBackoffEnded();
+  }
 }