folly/io/async/EventBase.cpp

   1 /*
   2  * Copyright 2016 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #ifndef __STDC_FORMAT_MACROS
  18 #define __STDC_FORMAT_MACROS
  19 #endif
  20
  21 #include <folly/io/async/EventBase.h>
  22
  23 #include <folly/ThreadName.h>
  24 #include <folly/io/async/NotificationQueue.h>
  25 #include <folly/portability/Unistd.h>
  26
  27 #include <condition_variable>
  28 #include <fcntl.h>
  29 #include <mutex>
  30 #include <pthread.h>
  31
  32 namespace {
  33
  34 using folly::EventBase;
  35
  36 class FunctionLoopCallback : public EventBase::LoopCallback {
  37  public:
  38   explicit FunctionLoopCallback(EventBase::Func&& function)
  39       : function_(std::move(function)) {}
  40
  41   void runLoopCallback() noexcept override {
  42     function_();
  43     delete this;
  44   }
  45
  46  private:
  47   EventBase::Func function_;
  48 };
  49 }
  50
  51 namespace folly {
  52
  53 /*
  54  * EventBase::FunctionRunner
  55  */
  56
  57 class EventBase::FunctionRunner
  58     : public NotificationQueue<EventBase::Func>::Consumer {
  59  public:
  60   void messageAvailable(Func&& msg) override {
  61     // In libevent2, internal events do not break the loop.
  62     // Most users would expect loop(), followed by runInEventBaseThread(),
  63     // to break the loop and check if it should exit or not.
  64     // To have similar bejaviour to libevent1.4, tell the loop to break here.
  65     // Note that loop() may still continue to loop, but it will also check the
  66     // stop_ flag as well as runInLoop callbacks, etc.
  67     event_base_loopbreak(getEventBase()->evb_);
  68
  69     if (!msg) {
  70       // terminateLoopSoon() sends a null message just to
  71       // wake up the loop.  We can ignore these messages.
  72       return;
  73     }
  74
  75     // The function should never throw an exception, because we have no
  76     // way of knowing what sort of error handling to perform.
  77     //
  78     // If it does throw, log a message and abort the program.
  79     try {
  80       msg();
  81     } catch (const std::exception& ex) {
  82       LOG(ERROR) << "runInEventBaseThread() function threw a "
  83                  << typeid(ex).name() << " exception: " << ex.what();
  84       abort();
  85     } catch (...) {
  86       LOG(ERROR) << "runInEventBaseThread() function threw an exception";
  87       abort();
  88     }
  89   }
  90 };
  91
  92 /*
  93  * EventBase::CobTimeout methods
  94  */
  95
  96 void EventBase::CobTimeout::timeoutExpired() noexcept {
  97   // For now, we just swallow any exceptions that the callback threw.
  98   try {
  99     cob_();
 100   } catch (const std::exception& ex) {
 101     LOG(ERROR) << "EventBase::runAfterDelay() callback threw "
 102                << typeid(ex).name() << " exception: " << ex.what();
 103   } catch (...) {
 104     LOG(ERROR) << "EventBase::runAfterDelay() callback threw non-exception "
 105                << "type";
 106   }
 107
 108   // The CobTimeout object was allocated on the heap by runAfterDelay(),
 109   // so delete it now that the it has fired.
 110   delete this;
 111 }
 112
 113
 114 // The interface used to libevent is not thread-safe.  Calls to
 115 // event_init() and event_base_free() directly modify an internal
 116 // global 'current_base', so a mutex is required to protect this.
 117 //
 118 // event_init() should only ever be called once.  Subsequent calls
 119 // should be made to event_base_new().  We can recognise that
 120 // event_init() has already been called by simply inspecting current_base.
 121 static std::mutex libevent_mutex_;
 122
 123 /*
 124  * EventBase methods
 125  */
 126
 127 EventBase::EventBase(bool enableTimeMeasurement)
 128   : runOnceCallbacks_(nullptr)
 129   , stop_(false)
 130   , loopThread_()
 131   , queue_(nullptr)
 132   , fnRunner_(nullptr)
 133   , maxLatency_(0)
 134   , avgLoopTime_(2000000)
 135   , maxLatencyLoopTime_(avgLoopTime_)
 136   , enableTimeMeasurement_(enableTimeMeasurement)
 137   , nextLoopCnt_(-40)       // Early wrap-around so bugs will manifest soon
 138   , latestLoopCnt_(nextLoopCnt_)
 139   , startWork_(0)
 140   , observer_(nullptr)
 141   , observerSampleCount_(0)
 142   , executionObserver_(nullptr) {
 143   struct event ev;
 144   {
 145     std::lock_guard<std::mutex> lock(libevent_mutex_);
 146
 147     // The value 'current_base' (libevent 1) or
 148     // 'event_global_current_base_' (libevent 2) is filled in by event_set(),
 149     // allowing examination of its value without an explicit reference here.
 150     // If ev.ev_base is NULL, then event_init() must be called, otherwise
 151     // call event_base_new().
 152     event_set(&ev, 0, 0, nullptr, nullptr);
 153     if (!ev.ev_base) {
 154       evb_ = event_init();
 155     }
 156   }
 157
 158   if (ev.ev_base) {
 159     evb_ = event_base_new();
 160   }
 161
 162   if (UNLIKELY(evb_ == nullptr)) {
 163     LOG(ERROR) << "EventBase(): Failed to init event base.";
 164     folly::throwSystemError("error in EventBase::EventBase()");
 165   }
 166   VLOG(5) << "EventBase(): Created.";
 167   initNotificationQueue();
 168   RequestContext::saveContext();
 169 }
 170
 171 // takes ownership of the event_base
 172 EventBase::EventBase(event_base* evb, bool enableTimeMeasurement)
 173   : runOnceCallbacks_(nullptr)
 174   , stop_(false)
 175   , loopThread_()
 176   , evb_(evb)
 177   , queue_(nullptr)
 178   , fnRunner_(nullptr)
 179   , maxLatency_(0)
 180   , avgLoopTime_(2000000)
 181   , maxLatencyLoopTime_(avgLoopTime_)
 182   , enableTimeMeasurement_(enableTimeMeasurement)
 183   , nextLoopCnt_(-40)       // Early wrap-around so bugs will manifest soon
 184   , latestLoopCnt_(nextLoopCnt_)
 185   , startWork_(0)
 186   , observer_(nullptr)
 187   , observerSampleCount_(0)
 188   , executionObserver_(nullptr) {
 189   if (UNLIKELY(evb_ == nullptr)) {
 190     LOG(ERROR) << "EventBase(): Pass nullptr as event base.";
 191     throw std::invalid_argument("EventBase(): event base cannot be nullptr");
 192   }
 193   initNotificationQueue();
 194   RequestContext::saveContext();
 195 }
 196
 197 EventBase::~EventBase() {
 198   // Keep looping until all keep-alive handles are released. Each keep-alive
 199   // handle signals that some external code will still schedule some work on
 200   // this EventBase (so it's not safe to destroy it).
 201   while (loopKeepAliveCount_ > 0) {
 202     applyLoopKeepAlive();
 203     loopOnce();
 204   }
 205
 206   // Call all destruction callbacks, before we start cleaning up our state.
 207   while (!onDestructionCallbacks_.empty()) {
 208     LoopCallback* callback = &onDestructionCallbacks_.front();
 209     onDestructionCallbacks_.pop_front();
 210     callback->runLoopCallback();
 211   }
 212
 213   // Delete any unfired callback objects, so that we don't leak memory
 214   // (Note that we don't fire them.  The caller is responsible for cleaning up
 215   // its own data structures if it destroys the EventBase with unfired events
 216   // remaining.)
 217   while (!pendingCobTimeouts_.empty()) {
 218     CobTimeout* timeout = &pendingCobTimeouts_.front();
 219     delete timeout;
 220   }
 221
 222   while (!runBeforeLoopCallbacks_.empty()) {
 223     delete &runBeforeLoopCallbacks_.front();
 224   }
 225
 226   (void) runLoopCallbacks(false);
 227
 228   if (!fnRunner_->consumeUntilDrained()) {
 229     LOG(ERROR) << "~EventBase(): Unable to drain notification queue";
 230   }
 231
 232   // Stop consumer before deleting NotificationQueue
 233   fnRunner_->stopConsuming();
 234   {
 235     std::lock_guard<std::mutex> lock(libevent_mutex_);
 236     event_base_free(evb_);
 237   }
 238
 239   while (!runAfterDrainCallbacks_.empty()) {
 240     LoopCallback* callback = &runAfterDrainCallbacks_.front();
 241     runAfterDrainCallbacks_.pop_front();
 242     callback->runLoopCallback();
 243   }
 244
 245   {
 246     std::lock_guard<std::mutex> lock(localStorageMutex_);
 247     for (auto storage : localStorageToDtor_) {
 248       storage->onEventBaseDestruction(*this);
 249     }
 250   }
 251   VLOG(5) << "EventBase(): Destroyed.";
 252 }
 253
 254 int EventBase::getNotificationQueueSize() const {
 255   return queue_->size();
 256 }
 257
 258 void EventBase::setMaxReadAtOnce(uint32_t maxAtOnce) {
 259   fnRunner_->setMaxReadAtOnce(maxAtOnce);
 260 }
 261
 262 // Set smoothing coefficient for loop load average; input is # of milliseconds
 263 // for exp(-1) decay.
 264 void EventBase::setLoadAvgMsec(uint32_t ms) {
 265   assert(enableTimeMeasurement_);
 266   uint64_t us = 1000 * ms;
 267   if (ms > 0) {
 268     maxLatencyLoopTime_.setTimeInterval(us);
 269     avgLoopTime_.setTimeInterval(us);
 270   } else {
 271     LOG(ERROR) << "non-positive arg to setLoadAvgMsec()";
 272   }
 273 }
 274
 275 void EventBase::resetLoadAvg(double value) {
 276   assert(enableTimeMeasurement_);
 277   avgLoopTime_.reset(value);
 278   maxLatencyLoopTime_.reset(value);
 279 }
 280
 281 static std::chrono::milliseconds
 282 getTimeDelta(std::chrono::steady_clock::time_point* prev) {
 283   auto result = std::chrono::steady_clock::now() - *prev;
 284   *prev = std::chrono::steady_clock::now();
 285
 286   return std::chrono::duration_cast<std::chrono::milliseconds>(result);
 287 }
 288
 289 void EventBase::waitUntilRunning() {
 290   while (!isRunning()) {
 291     sched_yield();
 292   }
 293 }
 294
 295 // enters the event_base loop -- will only exit when forced to
 296 bool EventBase::loop() {
 297   return loopBody();
 298 }
 299
 300 bool EventBase::loopOnce(int flags) {
 301   return loopBody(flags | EVLOOP_ONCE);
 302 }
 303
 304 bool EventBase::loopBody(int flags) {
 305   VLOG(5) << "EventBase(): Starting loop.";
 306
 307   DCHECK(!invokingLoop_)
 308       << "Your code just tried to loop over an event base from inside another "
 309       << "event base loop. Since libevent is not reentrant, this leads to "
 310       << "undefined behavior in opt builds. Please fix immediately. For the "
 311       << "common case of an inner function that needs to do some synchronous "
 312       << "computation on an event-base, replace getEventBase() by a new, "
 313       << "stack-allocated EvenBase.";
 314   invokingLoop_ = true;
 315   SCOPE_EXIT {
 316     invokingLoop_ = false;
 317   };
 318
 319   int res = 0;
 320   bool ranLoopCallbacks;
 321   bool blocking = !(flags & EVLOOP_NONBLOCK);
 322   bool once = (flags & EVLOOP_ONCE);
 323
 324   // time-measurement variables.
 325   std::chrono::steady_clock::time_point prev;
 326   int64_t idleStart = 0;
 327   int64_t busy;
 328   int64_t idle;
 329
 330   loopThread_.store(pthread_self(), std::memory_order_release);
 331
 332   if (!name_.empty()) {
 333     setThreadName(name_);
 334   }
 335
 336   if (enableTimeMeasurement_) {
 337     prev = std::chrono::steady_clock::now();
 338     idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
 339       std::chrono::steady_clock::now().time_since_epoch()).count();
 340   }
 341
 342   while (!stop_.load(std::memory_order_acquire)) {
 343     applyLoopKeepAlive();
 344     ++nextLoopCnt_;
 345
 346     // Run the before loop callbacks
 347     LoopCallbackList callbacks;
 348     callbacks.swap(runBeforeLoopCallbacks_);
 349
 350     while(!callbacks.empty()) {
 351       auto* item = &callbacks.front();
 352       callbacks.pop_front();
 353       item->runLoopCallback();
 354     }
 355
 356     // nobody can add loop callbacks from within this thread if
 357     // we don't have to handle anything to start with...
 358     if (blocking && loopCallbacks_.empty()) {
 359       res = event_base_loop(evb_, EVLOOP_ONCE);
 360     } else {
 361       res = event_base_loop(evb_, EVLOOP_ONCE | EVLOOP_NONBLOCK);
 362     }
 363
 364     ranLoopCallbacks = runLoopCallbacks();
 365
 366     if (enableTimeMeasurement_) {
 367       busy = std::chrono::duration_cast<std::chrono::microseconds>(
 368         std::chrono::steady_clock::now().time_since_epoch()).count() -
 369         startWork_;
 370       idle = startWork_ - idleStart;
 371
 372       avgLoopTime_.addSample(idle, busy);
 373       maxLatencyLoopTime_.addSample(idle, busy);
 374
 375       if (observer_) {
 376         if (observerSampleCount_++ == observer_->getSampleRate()) {
 377           observerSampleCount_ = 0;
 378           observer_->loopSample(busy, idle);
 379         }
 380       }
 381
 382       VLOG(11) << "EventBase "  << this         << " did not timeout "
 383         " loop time guess: "    << busy + idle  <<
 384         " idle time: "          << idle         <<
 385         " busy time: "          << busy         <<
 386         " avgLoopTime: "        << avgLoopTime_.get() <<
 387         " maxLatencyLoopTime: " << maxLatencyLoopTime_.get() <<
 388         " maxLatency_: "        << maxLatency_ <<
 389         " notificationQueueSize: " << getNotificationQueueSize() <<
 390         " nothingHandledYet(): "<< nothingHandledYet();
 391
 392       // see if our average loop time has exceeded our limit
 393       if ((maxLatency_ > 0) &&
 394           (maxLatencyLoopTime_.get() > double(maxLatency_))) {
 395         maxLatencyCob_();
 396         // back off temporarily -- don't keep spamming maxLatencyCob_
 397         // if we're only a bit over the limit
 398         maxLatencyLoopTime_.dampen(0.9);
 399       }
 400
 401       // Our loop run did real work; reset the idle timer
 402       idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
 403         std::chrono::steady_clock::now().time_since_epoch()).count();
 404     } else {
 405       VLOG(11) << "EventBase " << this << " did not timeout";
 406     }
 407
 408     // If the event loop indicate that there were no more events, and
 409     // we also didn't have any loop callbacks to run, there is nothing left to
 410     // do.
 411     if (res != 0 && !ranLoopCallbacks) {
 412       // Since Notification Queue is marked 'internal' some events may not have
 413       // run.  Run them manually if so, and continue looping.
 414       //
 415       if (getNotificationQueueSize() > 0) {
 416         fnRunner_->handlerReady(0);
 417       } else {
 418         break;
 419       }
 420     }
 421
 422     if (enableTimeMeasurement_) {
 423       VLOG(5) << "EventBase " << this << " loop time: " <<
 424         getTimeDelta(&prev).count();
 425     }
 426
 427     if (once) {
 428       break;
 429     }
 430   }
 431   // Reset stop_ so loop() can be called again
 432   stop_ = false;
 433
 434   if (res < 0) {
 435     LOG(ERROR) << "EventBase: -- error in event loop, res = " << res;
 436     return false;
 437   } else if (res == 1) {
 438     VLOG(5) << "EventBase: ran out of events (exiting loop)!";
 439   } else if (res > 1) {
 440     LOG(ERROR) << "EventBase: unknown event loop result = " << res;
 441     return false;
 442   }
 443
 444   loopThread_.store({}, std::memory_order_release);
 445
 446   VLOG(5) << "EventBase(): Done with loop.";
 447   return true;
 448 }
 449
 450 void EventBase::applyLoopKeepAlive() {
 451   if (loopKeepAliveActive_ && loopKeepAliveCount_ == 0) {
 452     // Restore the notification queue internal flag
 453     fnRunner_->stopConsuming();
 454     fnRunner_->startConsumingInternal(this, queue_.get());
 455     loopKeepAliveActive_ = false;
 456   } else if (!loopKeepAliveActive_ && loopKeepAliveCount_ > 0) {
 457     // Update the notification queue event to treat it as a normal
 458     // (non-internal) event.  The notification queue event always remains
 459     // installed, and the main loop won't exit with it installed.
 460     fnRunner_->stopConsuming();
 461     fnRunner_->startConsuming(this, queue_.get());
 462     loopKeepAliveActive_ = true;
 463   }
 464 }
 465
 466 void EventBase::loopForever() {
 467   bool ret;
 468   {
 469     SCOPE_EXIT {
 470       applyLoopKeepAlive();
 471     };
 472     // Make sure notification queue events are treated as normal events.
 473     auto keepAlive = loopKeepAlive();
 474     ret = loop();
 475   }
 476
 477   if (!ret) {
 478     folly::throwSystemError("error in EventBase::loopForever()");
 479   }
 480 }
 481
 482 void EventBase::bumpHandlingTime() {
 483   if (!enableTimeMeasurement_) {
 484     return;
 485   }
 486
 487   VLOG(11) << "EventBase " << this << " " << __PRETTY_FUNCTION__ <<
 488     " (loop) latest " << latestLoopCnt_ << " next " << nextLoopCnt_;
 489   if (nothingHandledYet()) {
 490     latestLoopCnt_ = nextLoopCnt_;
 491     // set the time
 492     startWork_ = std::chrono::duration_cast<std::chrono::microseconds>(
 493                      std::chrono::steady_clock::now().time_since_epoch())
 494                      .count();
 495
 496     VLOG(11) << "EventBase " << this << " " << __PRETTY_FUNCTION__
 497              << " (loop) startWork_ " << startWork_;
 498   }
 499 }
 500
 501 void EventBase::terminateLoopSoon() {
 502   VLOG(5) << "EventBase(): Received terminateLoopSoon() command.";
 503
 504   // Set stop to true, so the event loop will know to exit.
 505   // TODO: We should really use an atomic operation here with a release
 506   // barrier.
 507   stop_ = true;
 508
 509   // Call event_base_loopbreak() so that libevent will exit the next time
 510   // around the loop.
 511   event_base_loopbreak(evb_);
 512
 513   // If terminateLoopSoon() is called from another thread,
 514   // the EventBase thread might be stuck waiting for events.
 515   // In this case, it won't wake up and notice that stop_ is set until it
 516   // receives another event.  Send an empty frame to the notification queue
 517   // so that the event loop will wake up even if there are no other events.
 518   //
 519   // We don't care about the return value of trySendFrame().  If it fails
 520   // this likely means the EventBase already has lots of events waiting
 521   // anyway.
 522   try {
 523     queue_->putMessage(nullptr);
 524   } catch (...) {
 525     // We don't care if putMessage() fails.  This likely means
 526     // the EventBase already has lots of events waiting anyway.
 527   }
 528 }
 529
 530 void EventBase::runInLoop(LoopCallback* callback, bool thisIteration) {
 531   DCHECK(isInEventBaseThread());
 532   callback->cancelLoopCallback();
 533   callback->context_ = RequestContext::saveContext();
 534   if (runOnceCallbacks_ != nullptr && thisIteration) {
 535     runOnceCallbacks_->push_back(*callback);
 536   } else {
 537     loopCallbacks_.push_back(*callback);
 538   }
 539 }
 540
 541 void EventBase::runInLoop(Func cob, bool thisIteration) {
 542   DCHECK(isInEventBaseThread());
 543   auto wrapper = new FunctionLoopCallback(std::move(cob));
 544   wrapper->context_ = RequestContext::saveContext();
 545   if (runOnceCallbacks_ != nullptr && thisIteration) {
 546     runOnceCallbacks_->push_back(*wrapper);
 547   } else {
 548     loopCallbacks_.push_back(*wrapper);
 549   }
 550 }
 551
 552 void EventBase::runAfterDrain(Func cob) {
 553   auto callback = new FunctionLoopCallback(std::move(cob));
 554   std::lock_guard<std::mutex> lg(runAfterDrainCallbacksMutex_);
 555   callback->cancelLoopCallback();
 556   runAfterDrainCallbacks_.push_back(*callback);
 557 }
 558
 559 void EventBase::runOnDestruction(LoopCallback* callback) {
 560   std::lock_guard<std::mutex> lg(onDestructionCallbacksMutex_);
 561   callback->cancelLoopCallback();
 562   onDestructionCallbacks_.push_back(*callback);
 563 }
 564
 565 void EventBase::runBeforeLoop(LoopCallback* callback) {
 566   DCHECK(isInEventBaseThread());
 567   callback->cancelLoopCallback();
 568   runBeforeLoopCallbacks_.push_back(*callback);
 569 }
 570
 571 bool EventBase::runInEventBaseThread(Func fn) {
 572   // Send the message.
 573   // It will be received by the FunctionRunner in the EventBase's thread.
 574
 575   // We try not to schedule nullptr callbacks
 576   if (!fn) {
 577     LOG(ERROR) << "EventBase " << this
 578                << ": Scheduling nullptr callbacks is not allowed";
 579     return false;
 580   }
 581
 582   // Short-circuit if we are already in our event base
 583   if (inRunningEventBaseThread()) {
 584     runInLoop(std::move(fn));
 585     return true;
 586
 587   }
 588
 589   try {
 590     queue_->putMessage(std::move(fn));
 591   } catch (const std::exception& ex) {
 592     LOG(ERROR) << "EventBase " << this << ": failed to schedule function "
 593                << "for EventBase thread: " << ex.what();
 594     return false;
 595   }
 596
 597   return true;
 598 }
 599
 600 bool EventBase::runInEventBaseThreadAndWait(Func fn) {
 601   if (inRunningEventBaseThread()) {
 602     LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
 603                << "allowed";
 604     return false;
 605   }
 606
 607   bool ready = false;
 608   std::mutex m;
 609   std::condition_variable cv;
 610   runInEventBaseThread([&] {
 611       SCOPE_EXIT {
 612         std::unique_lock<std::mutex> l(m);
 613         ready = true;
 614         cv.notify_one();
 615         // We cannot release the lock before notify_one, because a spurious
 616         // wakeup in the waiting thread may lead to cv and m going out of scope
 617         // prematurely.
 618       };
 619       fn();
 620   });
 621   std::unique_lock<std::mutex> l(m);
 622   cv.wait(l, [&] { return ready; });
 623
 624   return true;
 625 }
 626
 627 bool EventBase::runImmediatelyOrRunInEventBaseThreadAndWait(Func fn) {
 628   if (isInEventBaseThread()) {
 629     fn();
 630     return true;
 631   } else {
 632     return runInEventBaseThreadAndWait(std::move(fn));
 633   }
 634 }
 635
 636 void EventBase::runAfterDelay(
 637     Func cob,
 638     uint32_t milliseconds,
 639     TimeoutManager::InternalEnum in) {
 640   if (!tryRunAfterDelay(std::move(cob), milliseconds, in)) {
 641     folly::throwSystemError(
 642       "error in EventBase::runAfterDelay(), failed to schedule timeout");
 643   }
 644 }
 645
 646 bool EventBase::tryRunAfterDelay(
 647     Func cob,
 648     uint32_t milliseconds,
 649     TimeoutManager::InternalEnum in) {
 650   CobTimeout* timeout = new CobTimeout(this, std::move(cob), in);
 651   if (!timeout->scheduleTimeout(milliseconds)) {
 652     delete timeout;
 653     return false;
 654   }
 655   pendingCobTimeouts_.push_back(*timeout);
 656   return true;
 657 }
 658
 659 bool EventBase::runLoopCallbacks(bool setContext) {
 660   if (!loopCallbacks_.empty()) {
 661     bumpHandlingTime();
 662     // Swap the loopCallbacks_ list with a temporary list on our stack.
 663     // This way we will only run callbacks scheduled at the time
 664     // runLoopCallbacks() was invoked.
 665     //
 666     // If any of these callbacks in turn call runInLoop() to schedule more
 667     // callbacks, those new callbacks won't be run until the next iteration
 668     // around the event loop.  This prevents runInLoop() callbacks from being
 669     // able to start file descriptor and timeout based events.
 670     LoopCallbackList currentCallbacks;
 671     currentCallbacks.swap(loopCallbacks_);
 672     runOnceCallbacks_ = &currentCallbacks;
 673
 674     while (!currentCallbacks.empty()) {
 675       LoopCallback* callback = &currentCallbacks.front();
 676       currentCallbacks.pop_front();
 677       if (setContext) {
 678         RequestContext::setContext(callback->context_);
 679       }
 680       callback->runLoopCallback();
 681     }
 682
 683     runOnceCallbacks_ = nullptr;
 684     return true;
 685   }
 686   return false;
 687 }
 688
 689 void EventBase::initNotificationQueue() {
 690   // Infinite size queue
 691   queue_.reset(new NotificationQueue<Func>());
 692
 693   // We allocate fnRunner_ separately, rather than declaring it directly
 694   // as a member of EventBase solely so that we don't need to include
 695   // NotificationQueue.h from EventBase.h
 696   fnRunner_.reset(new FunctionRunner());
 697
 698   // Mark this as an internal event, so event_base_loop() will return if
 699   // there are no other events besides this one installed.
 700   //
 701   // Most callers don't care about the internal notification queue used by
 702   // EventBase.  The queue is always installed, so if we did count the queue as
 703   // an active event, loop() would never exit with no more events to process.
 704   // Users can use loopForever() if they do care about the notification queue.
 705   // (This is useful for EventBase threads that do nothing but process
 706   // runInEventBaseThread() notifications.)
 707   fnRunner_->startConsumingInternal(this, queue_.get());
 708 }
 709
 710 void EventBase::SmoothLoopTime::setTimeInterval(uint64_t timeInterval) {
 711   expCoeff_ = -1.0/timeInterval;
 712   VLOG(11) << "expCoeff_ " << expCoeff_ << " " << __PRETTY_FUNCTION__;
 713 }
 714
 715 void EventBase::SmoothLoopTime::reset(double value) {
 716   value_ = value;
 717 }
 718
 719 void EventBase::SmoothLoopTime::addSample(int64_t idle, int64_t busy) {
 720     /*
 721      * Position at which the busy sample is considered to be taken.
 722      * (Allows to quickly skew our average without editing much code)
 723      */
 724     enum BusySamplePosition {
 725       RIGHT = 0,  // busy sample placed at the end of the iteration
 726       CENTER = 1, // busy sample placed at the middle point of the iteration
 727       LEFT = 2,   // busy sample placed at the beginning of the iteration
 728     };
 729
 730   // See http://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average
 731   // and D676020 for more info on this calculation.
 732   VLOG(11) << "idle " << idle << " oldBusyLeftover_ " << oldBusyLeftover_ <<
 733               " idle + oldBusyLeftover_ " << idle + oldBusyLeftover_ <<
 734               " busy " << busy << " " << __PRETTY_FUNCTION__;
 735   idle += oldBusyLeftover_ + busy;
 736   oldBusyLeftover_ = (busy * BusySamplePosition::CENTER) / 2;
 737   idle -= oldBusyLeftover_;
 738
 739   double coeff = exp(idle * expCoeff_);
 740   value_ *= coeff;
 741   value_ += (1.0 - coeff) * busy;
 742 }
 743
 744 bool EventBase::nothingHandledYet() const noexcept {
 745   VLOG(11) << "latest " << latestLoopCnt_ << " next " << nextLoopCnt_;
 746   return (nextLoopCnt_ != latestLoopCnt_);
 747 }
 748
 749 void EventBase::attachTimeoutManager(AsyncTimeout* obj,
 750                                       InternalEnum internal) {
 751
 752   struct event* ev = obj->getEvent();
 753   assert(ev->ev_base == nullptr);
 754
 755   event_base_set(getLibeventBase(), ev);
 756   if (internal == AsyncTimeout::InternalEnum::INTERNAL) {
 757     // Set the EVLIST_INTERNAL flag
 758     event_ref_flags(ev) |= EVLIST_INTERNAL;
 759   }
 760 }
 761
 762 void EventBase::detachTimeoutManager(AsyncTimeout* obj) {
 763   cancelTimeout(obj);
 764   struct event* ev = obj->getEvent();
 765   ev->ev_base = nullptr;
 766 }
 767
 768 bool EventBase::scheduleTimeout(AsyncTimeout* obj,
 769                                  TimeoutManager::timeout_type timeout) {
 770   assert(isInEventBaseThread());
 771   // Set up the timeval and add the event
 772   struct timeval tv;
 773   tv.tv_sec = timeout.count() / 1000LL;
 774   tv.tv_usec = (timeout.count() % 1000LL) * 1000LL;
 775
 776   struct event* ev = obj->getEvent();
 777   if (event_add(ev, &tv) < 0) {
 778     LOG(ERROR) << "EventBase: failed to schedule timeout: " << strerror(errno);
 779     return false;
 780   }
 781
 782   return true;
 783 }
 784
 785 void EventBase::cancelTimeout(AsyncTimeout* obj) {
 786   assert(isInEventBaseThread());
 787   struct event* ev = obj->getEvent();
 788   if (EventUtil::isEventRegistered(ev)) {
 789     event_del(ev);
 790   }
 791 }
 792
 793 void EventBase::setName(const std::string& name) {
 794   assert(isInEventBaseThread());
 795   name_ = name;
 796
 797   if (isRunning()) {
 798     setThreadName(loopThread_.load(std::memory_order_relaxed),
 799                   name_);
 800   }
 801 }
 802
 803 const std::string& EventBase::getName() {
 804   assert(isInEventBaseThread());
 805   return name_;
 806 }
 807
 808 const char* EventBase::getLibeventVersion() { return event_get_version(); }
 809 const char* EventBase::getLibeventMethod() { return event_get_method(); }
 810
 811 } // folly