folly/io/async/EventBase.cpp

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