stop in same thread

Summary: In a couple places in ServerBootstrap there is code that needs to run in the EB, but if we are already in the EB, it's fine to run it inline.  Maybe this should be a method on EB directly?  There is runInEventBaseThreadAndWait(), but it explicitly disallows this usage.

Test Plan: used in D1942242, deadlocks without this, since stop() was called in the same eb

Reviewed By: yfeldblum@fb.com

Subscribers: chalfant, doug, fugalh, folly-diffs@, jsedgwick, yfeldblum

FB internal diff: D1947581

Signature: t1:1947581:1427831021:d1d08ff9a4a00501d6be60670709fcb17af04134

folly/wangle/bootstrap/ServerBootstrap.cpp

@@ -26,7 +26,8 @@ void ServerWorkerPool::threadStarted(
   workers_.insert({h, worker});
 
   for(auto socket : *sockets_) {
-    socket->getEventBase()->runInEventBaseThreadAndWait([this, worker, socket](){
+    socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
+      [this, worker, socket](){
         socketFactory_->addAcceptCB(
           socket, worker.get(), worker->getEventBase());
     });
@@ -39,17 +40,16 @@ void ServerWorkerPool::threadStopped(
   CHECK(worker != workers_.end());
 
   for (auto& socket : *sockets_) {
-    folly::Baton<> barrier;
-    socket->getEventBase()->runInEventBaseThreadAndWait([&]() {
-      socketFactory_->removeAcceptCB(
-        socket, worker->second.get(), nullptr);
-      barrier.post();
+    socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
+      [&]() {
+        socketFactory_->removeAcceptCB(
+          socket, worker->second.get(), nullptr);
     });
-    barrier.wait();
   }
 
   if (!worker->second->getEventBase()->isInEventBaseThread()) {
-    worker->second->getEventBase()->runInEventBaseThreadAndWait([=]() {
+    worker->second->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
+      [=]() {
         worker->second->dropAllConnections();
       });
   } else {

folly/wangle/bootstrap/ServerBootstrap.h

@@ -177,7 +177,7 @@ class ServerBootstrap {
 
     // Startup all the threads
     workerFactory_->forEachWorker([this, socket](Acceptor* worker){
-      socket->getEventBase()->runInEventBaseThreadAndWait(
+      socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
         [this, worker, socket](){
           socketFactory_->addAcceptCB(socket, worker, worker->getEventBase());
       });
@@ -262,8 +262,9 @@ class ServerBootstrap {
     for (auto& socket : new_sockets) {
       // Startup all the threads
       workerFactory_->forEachWorker([this, socket](Acceptor* worker){
-        socket->getEventBase()->runInEventBaseThreadAndWait([this, worker, socket](){
-          socketFactory_->addAcceptCB(socket, worker, worker->getEventBase());
+        socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
+          [this, worker, socket](){
+            socketFactory_->addAcceptCB(socket, worker, worker->getEventBase());
         });
       });
 
@@ -276,12 +277,10 @@ class ServerBootstrap {
    */
   void stop() {
     for (auto socket : sockets_) {
-      folly::Baton<> barrier;
-      socket->getEventBase()->runInEventBaseThread([&]() mutable {
-        socketFactory_->stopSocket(socket);
-        barrier.post();
+      socket->getEventBase()->runImmediatelyOrRunInEventBaseThreadAndWait(
+        [&]() mutable {
+          socketFactory_->stopSocket(socket);
       });
-      barrier.wait();
     }
     sockets_.clear();
   }