Add logs for TFO succeded

[folly.git] / folly / io / async / EventBase.cpp

diff --git a/folly/io/async/EventBase.cpp b/folly/io/async/EventBase.cpp
index 6f7e9a27425e264be6539998aaa73d87a55fdd28..a4cbfa2055724c6c94166ffa98f4db3f2e6734db 100644 (file)
--- a/folly/io/async/EventBase.cpp
+++ b/folly/io/async/EventBase.cpp
@@ -22,12 +22,12 @@
 
 #include <folly/ThreadName.h>
 #include <folly/io/async/NotificationQueue.h>
+#include <folly/portability/Unistd.h>
 
 #include <condition_variable>
 #include <fcntl.h>
 #include <mutex>
 #include <pthread.h>
-#include <unistd.h>
 
 namespace {
 
@@ -195,6 +195,14 @@ EventBase::EventBase(event_base* evb, bool enableTimeMeasurement)
 }
 
 EventBase::~EventBase() {
+  // Keep looping until all keep-alive handles are released. Each keep-alive
+  // handle signals that some external code will still schedule some work on
+  // this EventBase (so it's not safe to destroy it).
+  while (loopKeepAliveCount_ > 0) {
+    applyLoopKeepAlive();
+    loopOnce();
+  }
+
   // Call all destruction callbacks, before we start cleaning up our state.
   while (!onDestructionCallbacks_.empty()) {
     LoopCallback* callback = &onDestructionCallbacks_.front();
@@ -215,7 +223,7 @@ EventBase::~EventBase() {
     delete &runBeforeLoopCallbacks_.front();
   }
 
-  (void) runLoopCallbacks(false);
+  (void)runLoopCallbacks();
 
   if (!fnRunner_->consumeUntilDrained()) {
     LOG(ERROR) << "~EventBase(): Unable to drain notification queue";
@@ -228,12 +236,6 @@ EventBase::~EventBase() {
     event_base_free(evb_);
   }
 
-  while (!runAfterDrainCallbacks_.empty()) {
-    LoopCallback* callback = &runAfterDrainCallbacks_.front();
-    runAfterDrainCallbacks_.pop_front();
-    callback->runLoopCallback();
-  }
-
   {
     std::lock_guard<std::mutex> lock(localStorageMutex_);
     for (auto storage : localStorageToDtor_) {
@@ -295,6 +297,19 @@ bool EventBase::loopOnce(int flags) {
 
 bool EventBase::loopBody(int flags) {
   VLOG(5) << "EventBase(): Starting loop.";
+
+  DCHECK(!invokingLoop_)
+      << "Your code just tried to loop over an event base from inside another "
+      << "event base loop. Since libevent is not reentrant, this leads to "
+      << "undefined behavior in opt builds. Please fix immediately. For the "
+      << "common case of an inner function that needs to do some synchronous "
+      << "computation on an event-base, replace getEventBase() by a new, "
+      << "stack-allocated EvenBase.";
+  invokingLoop_ = true;
+  SCOPE_EXIT {
+    invokingLoop_ = false;
+  };
+
   int res = 0;
   bool ranLoopCallbacks;
   bool blocking = !(flags & EVLOOP_NONBLOCK);
@@ -319,6 +334,7 @@ bool EventBase::loopBody(int flags) {
   }
 
   while (!stop_.load(std::memory_order_acquire)) {
+    applyLoopKeepAlive();
     ++nextLoopCnt_;
 
     // Run the before loop callbacks
@@ -425,18 +441,32 @@ bool EventBase::loopBody(int flags) {
   return true;
 }
 
-void EventBase::loopForever() {
-  // Update the notification queue event to treat it as a normal (non-internal)
-  // event.  The notification queue event always remains installed, and the main
-  // loop won't exit with it installed.
-  fnRunner_->stopConsuming();
-  fnRunner_->startConsuming(this, queue_.get());
-
-  bool ret = loop();
+void EventBase::applyLoopKeepAlive() {
+  if (loopKeepAliveActive_ && loopKeepAliveCount_ == 0) {
+    // Restore the notification queue internal flag
+    fnRunner_->stopConsuming();
+    fnRunner_->startConsumingInternal(this, queue_.get());
+    loopKeepAliveActive_ = false;
+  } else if (!loopKeepAliveActive_ && loopKeepAliveCount_ > 0) {
+    // Update the notification queue event to treat it as a normal
+    // (non-internal) event.  The notification queue event always remains
+    // installed, and the main loop won't exit with it installed.
+    fnRunner_->stopConsuming();
+    fnRunner_->startConsuming(this, queue_.get());
+    loopKeepAliveActive_ = true;
+  }
+}
 
-  // Restore the notification queue internal flag
-  fnRunner_->stopConsuming();
-  fnRunner_->startConsumingInternal(this, queue_.get());
+void EventBase::loopForever() {
+  bool ret;
+  {
+    SCOPE_EXIT {
+      applyLoopKeepAlive();
+    };
+    // Make sure notification queue events are treated as normal events.
+    auto keepAlive = loopKeepAlive();
+    ret = loop();
+  }
 
   if (!ret) {
     folly::throwSystemError("error in EventBase::loopForever()");
@@ -513,13 +543,6 @@ void EventBase::runInLoop(Func cob, bool thisIteration) {
   }
 }
 
-void EventBase::runAfterDrain(Func cob) {
-  auto callback = new FunctionLoopCallback(std::move(cob));
-  std::lock_guard<std::mutex> lg(runAfterDrainCallbacksMutex_);
-  callback->cancelLoopCallback();
-  runAfterDrainCallbacks_.push_back(*callback);
-}
-
 void EventBase::runOnDestruction(LoopCallback* callback) {
   std::lock_guard<std::mutex> lg(onDestructionCallbacksMutex_);
   callback->cancelLoopCallback();
@@ -620,7 +643,7 @@ bool EventBase::tryRunAfterDelay(
   return true;
 }
 
-bool EventBase::runLoopCallbacks(bool setContext) {
+bool EventBase::runLoopCallbacks() {
   if (!loopCallbacks_.empty()) {
     bumpHandlingTime();
     // Swap the loopCallbacks_ list with a temporary list on our stack.
@@ -638,9 +661,7 @@ bool EventBase::runLoopCallbacks(bool setContext) {
     while (!currentCallbacks.empty()) {
       LoopCallback* callback = &currentCallbacks.front();
       currentCallbacks.pop_front();
-      if (setContext) {
-        RequestContext::setContext(callback->context_);
-      }
+      folly::RequestContextScopeGuard rctx(callback->context_);
       callback->runLoopCallback();
     }