*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
#include <folly/ThreadName.h>
#include <folly/io/async/NotificationQueue.h>
#include <boost/static_assert.hpp>
#include <folly/ThreadName.h>
#include <folly/io/async/NotificationQueue.h>
#include <boost/static_assert.hpp>
class EventBase::FunctionRunner
: public NotificationQueue<std::pair<void (*)(void*), void*>>::Consumer {
public:
class EventBase::FunctionRunner
: public NotificationQueue<std::pair<void (*)(void*), void*>>::Consumer {
public:
// In libevent2, internal events do not break the loop.
// Most users would expect loop(), followed by runInEventBaseThread(),
// In libevent2, internal events do not break the loop.
// Most users would expect loop(), followed by runInEventBaseThread(),
, nextLoopCnt_(-40) // Early wrap-around so bugs will manifest soon
, latestLoopCnt_(nextLoopCnt_)
, startWork_(0)
, observer_(nullptr)
, nextLoopCnt_(-40) // Early wrap-around so bugs will manifest soon
, latestLoopCnt_(nextLoopCnt_)
, startWork_(0)
, observer_(nullptr)
, nextLoopCnt_(-40) // Early wrap-around so bugs will manifest soon
, latestLoopCnt_(nextLoopCnt_)
, startWork_(0)
, observer_(nullptr)
, nextLoopCnt_(-40) // Early wrap-around so bugs will manifest soon
, latestLoopCnt_(nextLoopCnt_)
, startWork_(0)
, observer_(nullptr)
if (UNLIKELY(evb_ == nullptr)) {
LOG(ERROR) << "EventBase(): Pass nullptr as event base.";
throw std::invalid_argument("EventBase(): event base cannot be nullptr");
}
initNotificationQueue();
if (UNLIKELY(evb_ == nullptr)) {
LOG(ERROR) << "EventBase(): Pass nullptr as event base.";
throw std::invalid_argument("EventBase(): event base cannot be nullptr");
}
initNotificationQueue();
// Set smoothing coefficient for loop load average; input is # of milliseconds
// for exp(-1) decay.
void EventBase::setLoadAvgMsec(uint32_t ms) {
// Set smoothing coefficient for loop load average; input is # of milliseconds
// for exp(-1) decay.
void EventBase::setLoadAvgMsec(uint32_t ms) {
- auto prev = std::chrono::steady_clock::now();
- int64_t idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
- std::chrono::steady_clock::now().time_since_epoch()).count();
+ if (enableTimeMeasurement_) {
+ prev = std::chrono::steady_clock::now();
+ idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
+ std::chrono::steady_clock::now().time_since_epoch()).count();
+ }
- int64_t busy = std::chrono::duration_cast<std::chrono::microseconds>(
- std::chrono::steady_clock::now().time_since_epoch()).count() - startWork_;
- int64_t idle = startWork_ - idleStart;
+ if (enableTimeMeasurement_) {
+ busy = std::chrono::duration_cast<std::chrono::microseconds>(
+ std::chrono::steady_clock::now().time_since_epoch()).count() -
+ startWork_;
+ idle = startWork_ - idleStart;
- avgLoopTime_.addSample(idle, busy);
- maxLatencyLoopTime_.addSample(idle, busy);
+ avgLoopTime_.addSample(idle, busy);
+ maxLatencyLoopTime_.addSample(idle, busy);
- if (observer_) {
- if (observerSampleCount_++ == observer_->getSampleRate()) {
- observerSampleCount_ = 0;
- observer_->loopSample(busy, idle);
+ if (observer_) {
+ if (observerSampleCount_++ == observer_->getSampleRate()) {
+ observerSampleCount_ = 0;
+ observer_->loopSample(busy, idle);
+ }
- VLOG(11) << "EventBase " << this << " did not timeout "
- " loop time guess: " << busy + idle <<
- " idle time: " << idle <<
- " busy time: " << busy <<
- " avgLoopTime: " << avgLoopTime_.get() <<
- " maxLatencyLoopTime: " << maxLatencyLoopTime_.get() <<
- " maxLatency_: " << maxLatency_ <<
- " nothingHandledYet(): "<< nothingHandledYet();
-
- // see if our average loop time has exceeded our limit
- if ((maxLatency_ > 0) &&
- (maxLatencyLoopTime_.get() > double(maxLatency_))) {
- maxLatencyCob_();
- // back off temporarily -- don't keep spamming maxLatencyCob_
- // if we're only a bit over the limit
- maxLatencyLoopTime_.dampen(0.9);
- }
+ VLOG(11) << "EventBase " << this << " did not timeout "
+ " loop time guess: " << busy + idle <<
+ " idle time: " << idle <<
+ " busy time: " << busy <<
+ " avgLoopTime: " << avgLoopTime_.get() <<
+ " maxLatencyLoopTime: " << maxLatencyLoopTime_.get() <<
+ " maxLatency_: " << maxLatency_ <<
+ " nothingHandledYet(): "<< nothingHandledYet();
+
+ // see if our average loop time has exceeded our limit
+ if ((maxLatency_ > 0) &&
+ (maxLatencyLoopTime_.get() > double(maxLatency_))) {
+ maxLatencyCob_();
+ // back off temporarily -- don't keep spamming maxLatencyCob_
+ // if we're only a bit over the limit
+ maxLatencyLoopTime_.dampen(0.9);
+ }
- // Our loop run did real work; reset the idle timer
- idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
- std::chrono::steady_clock::now().time_since_epoch()).count();
+ // Our loop run did real work; reset the idle timer
+ idleStart = std::chrono::duration_cast<std::chrono::microseconds>(
+ std::chrono::steady_clock::now().time_since_epoch()).count();
+ } else {
+ VLOG(11) << "EventBase " << this << " did not timeout "
+ " time measurement is disabled "
+ " nothingHandledYet(): "<< nothingHandledYet();
+ }
// If the event loop indicate that there were no more events, and
// we also didn't have any loop callbacks to run, there is nothing left to
// If the event loop indicate that there were no more events, and
// we also didn't have any loop callbacks to run, there is nothing left to
- VLOG(5) << "EventBase " << this << " loop time: " <<
- getTimeDelta(&prev).count();
+ if (enableTimeMeasurement_) {
+ VLOG(5) << "EventBase " << this << " loop time: " <<
+ getTimeDelta(&prev).count();
+ }
if (inRunningEventBaseThread()) {
LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
<< "allowed";
return false;
}
if (inRunningEventBaseThread()) {
LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
<< "allowed";
return false;
}
-bool EventBase::runInEventBaseThreadAndWait(const Cob& fn) {
- if (inRunningEventBaseThread()) {
- LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
- << "allowed";
- return false;
+bool EventBase::runImmediatelyOrRunInEventBaseThreadAndWait(const Cob& fn) {
+ if (isInEventBaseThread()) {
+ fn();
+ return true;
+ } else {
+ return runInEventBaseThreadAndWait(fn);
- Baton<> ready;
- runInEventBaseThread([&] {
- fn();
- ready.post();
- });
- ready.wait();
-
- return true;
+void EventBase::runAfterDelay(const Cob& cob,
+ int milliseconds,
+ TimeoutManager::InternalEnum in) {
+ if (!tryRunAfterDelay(cob, milliseconds, in)) {
+ folly::throwSystemError(
+ "error in EventBase::runAfterDelay(), failed to schedule timeout");
+ }
-bool EventBase::runAfterDelay(const Cob& cob,
- int milliseconds,
- TimeoutManager::InternalEnum in) {
+bool EventBase::tryRunAfterDelay(const Cob& cob,
+ int milliseconds,
+ TimeoutManager::InternalEnum in) {
CobTimeout* timeout = new CobTimeout(this, cob, in);
if (!timeout->scheduleTimeout(milliseconds)) {
delete timeout;
return false;
}
CobTimeout* timeout = new CobTimeout(this, cob, in);
if (!timeout->scheduleTimeout(milliseconds)) {
delete timeout;
return false;
}