EventBase.h 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810
  1. /*
  2. * Copyright 2014-present 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. #pragma once
  17. #include <atomic>
  18. #include <cerrno>
  19. #include <cmath>
  20. #include <cstdlib>
  21. #include <functional>
  22. #include <list>
  23. #include <memory>
  24. #include <mutex>
  25. #include <queue>
  26. #include <set>
  27. #include <stack>
  28. #include <unordered_map>
  29. #include <unordered_set>
  30. #include <utility>
  31. #include <boost/intrusive/list.hpp>
  32. #include <glog/logging.h>
  33. #include <folly/Executor.h>
  34. #include <folly/Function.h>
  35. #include <folly/Portability.h>
  36. #include <folly/ScopeGuard.h>
  37. #include <folly/executors/DrivableExecutor.h>
  38. #include <folly/executors/IOExecutor.h>
  39. #include <folly/executors/ScheduledExecutor.h>
  40. #include <folly/executors/SequencedExecutor.h>
  41. #include <folly/experimental/ExecutionObserver.h>
  42. #include <folly/io/async/AsyncTimeout.h>
  43. #include <folly/io/async/HHWheelTimer.h>
  44. #include <folly/io/async/Request.h>
  45. #include <folly/io/async/TimeoutManager.h>
  46. #include <folly/portability/Event.h>
  47. #include <folly/synchronization/CallOnce.h>
  48. namespace folly {
  49. using Cob = Func; // defined in folly/Executor.h
  50. template <typename MessageT>
  51. class NotificationQueue;
  52. namespace detail {
  53. class EventBaseLocalBase;
  54. class EventBaseLocalBaseBase {
  55. public:
  56. virtual void onEventBaseDestruction(EventBase& evb) = 0;
  57. virtual ~EventBaseLocalBaseBase() = default;
  58. };
  59. } // namespace detail
  60. template <typename T>
  61. class EventBaseLocal;
  62. class EventBaseObserver {
  63. public:
  64. virtual ~EventBaseObserver() = default;
  65. virtual uint32_t getSampleRate() const = 0;
  66. virtual void loopSample(int64_t busyTime, int64_t idleTime) = 0;
  67. };
  68. // Helper class that sets and retrieves the EventBase associated with a given
  69. // request via RequestContext. See Request.h for that mechanism.
  70. class RequestEventBase : public RequestData {
  71. public:
  72. static EventBase* get() {
  73. auto data = dynamic_cast<RequestEventBase*>(
  74. RequestContext::get()->getContextData(kContextDataName));
  75. if (!data) {
  76. return nullptr;
  77. }
  78. return data->eb_;
  79. }
  80. static void set(EventBase* eb) {
  81. RequestContext::get()->setContextData(
  82. kContextDataName,
  83. std::unique_ptr<RequestEventBase>(new RequestEventBase(eb)));
  84. }
  85. bool hasCallback() override {
  86. return false;
  87. }
  88. private:
  89. explicit RequestEventBase(EventBase* eb) : eb_(eb) {}
  90. EventBase* eb_;
  91. static constexpr const char* kContextDataName{"EventBase"};
  92. };
  93. class VirtualEventBase;
  94. /**
  95. * This class is a wrapper for all asynchronous I/O processing functionality
  96. *
  97. * EventBase provides a main loop that notifies EventHandler callback objects
  98. * when I/O is ready on a file descriptor, and notifies AsyncTimeout objects
  99. * when a specified timeout has expired. More complex, higher-level callback
  100. * mechanisms can then be built on top of EventHandler and AsyncTimeout.
  101. *
  102. * A EventBase object can only drive an event loop for a single thread. To
  103. * take advantage of multiple CPU cores, most asynchronous I/O servers have one
  104. * thread per CPU, and use a separate EventBase for each thread.
  105. *
  106. * In general, most EventBase methods may only be called from the thread
  107. * running the EventBase's loop. There are a few exceptions to this rule, for
  108. * methods that are explicitly intended to allow communication with a
  109. * EventBase from other threads. When it is safe to call a method from
  110. * another thread it is explicitly listed in the method comments.
  111. */
  112. class EventBase : private boost::noncopyable,
  113. public TimeoutManager,
  114. public DrivableExecutor,
  115. public IOExecutor,
  116. public SequencedExecutor,
  117. public ScheduledExecutor {
  118. public:
  119. using Func = folly::Function<void()>;
  120. /**
  121. * A callback interface to use with runInLoop()
  122. *
  123. * Derive from this class if you need to delay some code execution until the
  124. * next iteration of the event loop. This allows you to schedule code to be
  125. * invoked from the top-level of the loop, after your immediate callers have
  126. * returned.
  127. *
  128. * If a LoopCallback object is destroyed while it is scheduled to be run in
  129. * the next loop iteration, it will automatically be cancelled.
  130. */
  131. class LoopCallback
  132. : public boost::intrusive::list_base_hook<
  133. boost::intrusive::link_mode<boost::intrusive::auto_unlink>> {
  134. public:
  135. virtual ~LoopCallback() = default;
  136. virtual void runLoopCallback() noexcept = 0;
  137. void cancelLoopCallback() {
  138. context_.reset();
  139. unlink();
  140. }
  141. bool isLoopCallbackScheduled() const {
  142. return is_linked();
  143. }
  144. private:
  145. typedef boost::intrusive::
  146. list<LoopCallback, boost::intrusive::constant_time_size<false>>
  147. List;
  148. // EventBase needs access to LoopCallbackList (and therefore to hook_)
  149. friend class EventBase;
  150. friend class VirtualEventBase;
  151. std::shared_ptr<RequestContext> context_;
  152. };
  153. class FunctionLoopCallback : public LoopCallback {
  154. public:
  155. explicit FunctionLoopCallback(Func&& function)
  156. : function_(std::move(function)) {}
  157. void runLoopCallback() noexcept override {
  158. function_();
  159. delete this;
  160. }
  161. private:
  162. Func function_;
  163. };
  164. // Like FunctionLoopCallback, but saves one allocation. Use with caution.
  165. //
  166. // The caller is responsible for maintaining the lifetime of this callback
  167. // until after the point at which the contained function is called.
  168. class StackFunctionLoopCallback : public LoopCallback {
  169. public:
  170. explicit StackFunctionLoopCallback(Func&& function)
  171. : function_(std::move(function)) {}
  172. void runLoopCallback() noexcept override {
  173. Func(std::move(function_))();
  174. }
  175. private:
  176. Func function_;
  177. };
  178. /**
  179. * Create a new EventBase object.
  180. *
  181. * Same as EventBase(true), which constructs an EventBase that measures time.
  182. */
  183. EventBase() : EventBase(true) {}
  184. /**
  185. * Create a new EventBase object.
  186. *
  187. * @param enableTimeMeasurement Informs whether this event base should measure
  188. * time. Disabling it would likely improve
  189. * performance, but will disable some features
  190. * that relies on time-measurement, including:
  191. * observer, max latency and avg loop time.
  192. */
  193. explicit EventBase(bool enableTimeMeasurement);
  194. /**
  195. * Create a new EventBase object that will use the specified libevent
  196. * event_base object to drive the event loop.
  197. *
  198. * The EventBase will take ownership of this event_base, and will call
  199. * event_base_free(evb) when the EventBase is destroyed.
  200. *
  201. * @param enableTimeMeasurement Informs whether this event base should measure
  202. * time. Disabling it would likely improve
  203. * performance, but will disable some features
  204. * that relies on time-measurement, including:
  205. * observer, max latency and avg loop time.
  206. */
  207. explicit EventBase(event_base* evb, bool enableTimeMeasurement = true);
  208. ~EventBase() override;
  209. /**
  210. * Runs the event loop.
  211. *
  212. * loop() will loop waiting for I/O or timeouts and invoking EventHandler
  213. * and AsyncTimeout callbacks as their events become ready. loop() will
  214. * only return when there are no more events remaining to process, or after
  215. * terminateLoopSoon() has been called.
  216. *
  217. * loop() may be called again to restart event processing after a previous
  218. * call to loop() or loopForever() has returned.
  219. *
  220. * Returns true if the loop completed normally (if it processed all
  221. * outstanding requests, or if terminateLoopSoon() was called). If an error
  222. * occurs waiting for events, false will be returned.
  223. */
  224. bool loop();
  225. /**
  226. * Same as loop(), but doesn't wait for all keep-alive tokens to be released.
  227. */
  228. [[deprecated("This should only be used in legacy unit tests")]] bool
  229. loopIgnoreKeepAlive();
  230. /**
  231. * Wait for some events to become active, run them, then return.
  232. *
  233. * When EVLOOP_NONBLOCK is set in flags, the loop won't block if there
  234. * are not any events to process.
  235. *
  236. * This is useful for callers that want to run the loop manually.
  237. *
  238. * Returns the same result as loop().
  239. */
  240. bool loopOnce(int flags = 0);
  241. /**
  242. * Runs the event loop.
  243. *
  244. * loopForever() behaves like loop(), except that it keeps running even if
  245. * when there are no more user-supplied EventHandlers or AsyncTimeouts
  246. * registered. It will only return after terminateLoopSoon() has been
  247. * called.
  248. *
  249. * This is useful for callers that want to wait for other threads to call
  250. * runInEventBaseThread(), even when there are no other scheduled events.
  251. *
  252. * loopForever() may be called again to restart event processing after a
  253. * previous call to loop() or loopForever() has returned.
  254. *
  255. * Throws a std::system_error if an error occurs.
  256. */
  257. void loopForever();
  258. /**
  259. * Causes the event loop to exit soon.
  260. *
  261. * This will cause an existing call to loop() or loopForever() to stop event
  262. * processing and return, even if there are still events remaining to be
  263. * processed.
  264. *
  265. * It is safe to call terminateLoopSoon() from another thread to cause loop()
  266. * to wake up and return in the EventBase loop thread. terminateLoopSoon()
  267. * may also be called from the loop thread itself (for example, a
  268. * EventHandler or AsyncTimeout callback may call terminateLoopSoon() to
  269. * cause the loop to exit after the callback returns.) If the loop is not
  270. * running, this will cause the next call to loop to terminate soon after
  271. * starting. If a loop runs out of work (and so terminates on its own)
  272. * concurrently with a call to terminateLoopSoon(), this may cause a race
  273. * condition.
  274. *
  275. * Note that the caller is responsible for ensuring that cleanup of all event
  276. * callbacks occurs properly. Since terminateLoopSoon() causes the loop to
  277. * exit even when there are pending events present, there may be remaining
  278. * callbacks present waiting to be invoked. If the loop is later restarted
  279. * pending events will continue to be processed normally, however if the
  280. * EventBase is destroyed after calling terminateLoopSoon() it is the
  281. * caller's responsibility to ensure that cleanup happens properly even if
  282. * some outstanding events are never processed.
  283. */
  284. void terminateLoopSoon();
  285. /**
  286. * Adds the given callback to a queue of things run after the current pass
  287. * through the event loop completes. Note that if this callback calls
  288. * runInLoop() the new callback won't be called until the main event loop
  289. * has gone through a cycle.
  290. *
  291. * This method may only be called from the EventBase's thread. This
  292. * essentially allows an event handler to schedule an additional callback to
  293. * be invoked after it returns.
  294. *
  295. * Use runInEventBaseThread() to schedule functions from another thread.
  296. *
  297. * The thisIteration parameter makes this callback run in this loop
  298. * iteration, instead of the next one, even if called from a
  299. * runInLoop callback (normal io callbacks that call runInLoop will
  300. * always run in this iteration). This was originally added to
  301. * support detachEventBase, as a user callback may have called
  302. * terminateLoopSoon(), but we want to make sure we detach. Also,
  303. * detachEventBase almost always must be called from the base event
  304. * loop to ensure the stack is unwound, since most users of
  305. * EventBase are not thread safe.
  306. *
  307. * Ideally we would not need thisIteration, and instead just use
  308. * runInLoop with loop() (instead of terminateLoopSoon).
  309. */
  310. void runInLoop(LoopCallback* callback, bool thisIteration = false);
  311. /**
  312. * Convenience function to call runInLoop() with a folly::Function.
  313. *
  314. * This creates a LoopCallback object to wrap the folly::Function, and invoke
  315. * the folly::Function when the loop callback fires. This is slightly more
  316. * expensive than defining your own LoopCallback, but more convenient in
  317. * areas that aren't too performance sensitive.
  318. *
  319. * This method may only be called from the EventBase's thread. This
  320. * essentially allows an event handler to schedule an additional callback to
  321. * be invoked after it returns.
  322. *
  323. * Use runInEventBaseThread() to schedule functions from another thread.
  324. */
  325. void runInLoop(Func c, bool thisIteration = false);
  326. /**
  327. * Adds the given callback to a queue of things run before destruction
  328. * of current EventBase.
  329. *
  330. * This allows users of EventBase that run in it, but don't control it,
  331. * to be notified before EventBase gets destructed.
  332. *
  333. * Note: will be called from the thread that invoked EventBase destructor,
  334. * before the final run of loop callbacks.
  335. */
  336. void runOnDestruction(LoopCallback* callback);
  337. /**
  338. * Adds a callback that will run immediately *before* the event loop.
  339. * This is very similar to runInLoop(), but will not cause the loop to break:
  340. * For example, this callback could be used to get loop times.
  341. */
  342. void runBeforeLoop(LoopCallback* callback);
  343. /**
  344. * Run the specified function in the EventBase's thread.
  345. *
  346. * This method is thread-safe, and may be called from another thread.
  347. *
  348. * If runInEventBaseThread() is called when the EventBase loop is not
  349. * running, the function call will be delayed until the next time the loop is
  350. * started.
  351. *
  352. * If runInEventBaseThread() returns true the function has successfully been
  353. * scheduled to run in the loop thread. However, if the loop is terminated
  354. * (and never later restarted) before it has a chance to run the requested
  355. * function, the function will be run upon the EventBase's destruction.
  356. *
  357. * If two calls to runInEventBaseThread() are made from the same thread, the
  358. * functions will always be run in the order that they were scheduled.
  359. * Ordering between functions scheduled from separate threads is not
  360. * guaranteed.
  361. *
  362. * @param fn The function to run. The function must not throw any
  363. * exceptions.
  364. * @param arg An argument to pass to the function.
  365. *
  366. * @return Returns true if the function was successfully scheduled, or false
  367. * if there was an error scheduling the function.
  368. */
  369. template <typename T>
  370. bool runInEventBaseThread(void (*fn)(T*), T* arg);
  371. /**
  372. * Run the specified function in the EventBase's thread
  373. *
  374. * This version of runInEventBaseThread() takes a folly::Function object.
  375. * Note that this may be less efficient than the version that takes a plain
  376. * function pointer and void* argument, if moving the function is expensive
  377. * (e.g., if it wraps a lambda which captures some values with expensive move
  378. * constructors).
  379. *
  380. * If the loop is terminated (and never later restarted) before it has a
  381. * chance to run the requested function, the function will be run upon the
  382. * EventBase's destruction.
  383. *
  384. * The function must not throw any exceptions.
  385. */
  386. bool runInEventBaseThread(Func fn);
  387. /*
  388. * Like runInEventBaseThread, but the caller waits for the callback to be
  389. * executed.
  390. */
  391. template <typename T>
  392. bool runInEventBaseThreadAndWait(void (*fn)(T*), T* arg);
  393. /*
  394. * Like runInEventBaseThread, but the caller waits for the callback to be
  395. * executed.
  396. */
  397. bool runInEventBaseThreadAndWait(Func fn);
  398. /*
  399. * Like runInEventBaseThreadAndWait, except if the caller is already in the
  400. * event base thread, the functor is simply run inline.
  401. */
  402. template <typename T>
  403. bool runImmediatelyOrRunInEventBaseThreadAndWait(void (*fn)(T*), T* arg);
  404. /*
  405. * Like runInEventBaseThreadAndWait, except if the caller is already in the
  406. * event base thread, the functor is simply run inline.
  407. */
  408. bool runImmediatelyOrRunInEventBaseThreadAndWait(Func fn);
  409. /**
  410. * Set the maximum desired latency in us and provide a callback which will be
  411. * called when that latency is exceeded.
  412. * OBS: This functionality depends on time-measurement.
  413. */
  414. void setMaxLatency(std::chrono::microseconds maxLatency, Func maxLatencyCob) {
  415. assert(enableTimeMeasurement_);
  416. maxLatency_ = maxLatency;
  417. maxLatencyCob_ = std::move(maxLatencyCob);
  418. }
  419. /**
  420. * Set smoothing coefficient for loop load average; # of milliseconds
  421. * for exp(-1) (1/2.71828...) decay.
  422. */
  423. void setLoadAvgMsec(std::chrono::milliseconds ms);
  424. /**
  425. * reset the load average to a desired value
  426. */
  427. void resetLoadAvg(double value = 0.0);
  428. /**
  429. * Get the average loop time in microseconds (an exponentially-smoothed ave)
  430. */
  431. double getAvgLoopTime() const {
  432. assert(enableTimeMeasurement_);
  433. return avgLoopTime_.get();
  434. }
  435. /**
  436. * check if the event base loop is running.
  437. */
  438. bool isRunning() const {
  439. return loopThread_.load(std::memory_order_relaxed) != std::thread::id();
  440. }
  441. /**
  442. * wait until the event loop starts (after starting the event loop thread).
  443. */
  444. void waitUntilRunning();
  445. size_t getNotificationQueueSize() const;
  446. void setMaxReadAtOnce(uint32_t maxAtOnce);
  447. /**
  448. * Verify that current thread is the EventBase thread, if the EventBase is
  449. * running.
  450. */
  451. bool isInEventBaseThread() const {
  452. auto tid = loopThread_.load(std::memory_order_relaxed);
  453. return tid == std::thread::id() || tid == std::this_thread::get_id();
  454. }
  455. bool inRunningEventBaseThread() const {
  456. return loopThread_.load(std::memory_order_relaxed) ==
  457. std::this_thread::get_id();
  458. }
  459. /**
  460. * Equivalent to CHECK(isInEventBaseThread()) (and assert/DCHECK for
  461. * dcheckIsInEventBaseThread), but it prints more information on
  462. * failure.
  463. */
  464. void checkIsInEventBaseThread() const;
  465. void dcheckIsInEventBaseThread() const {
  466. if (kIsDebug) {
  467. checkIsInEventBaseThread();
  468. }
  469. }
  470. HHWheelTimer& timer() {
  471. if (!wheelTimer_) {
  472. wheelTimer_ = HHWheelTimer::newTimer(this);
  473. }
  474. return *wheelTimer_.get();
  475. }
  476. // --------- interface to underlying libevent base ------------
  477. // Avoid using these functions if possible. These functions are not
  478. // guaranteed to always be present if we ever provide alternative EventBase
  479. // implementations that do not use libevent internally.
  480. event_base* getLibeventBase() const {
  481. return evb_;
  482. }
  483. static const char* getLibeventVersion();
  484. static const char* getLibeventMethod();
  485. /**
  486. * only EventHandler/AsyncTimeout subclasses and ourselves should
  487. * ever call this.
  488. *
  489. * This is used to mark the beginning of a new loop cycle by the
  490. * first handler fired within that cycle.
  491. *
  492. */
  493. void bumpHandlingTime() final;
  494. class SmoothLoopTime {
  495. public:
  496. explicit SmoothLoopTime(std::chrono::microseconds timeInterval)
  497. : expCoeff_(-1.0 / timeInterval.count()), value_(0.0) {
  498. VLOG(11) << "expCoeff_ " << expCoeff_ << " " << __PRETTY_FUNCTION__;
  499. }
  500. void setTimeInterval(std::chrono::microseconds timeInterval);
  501. void reset(double value = 0.0);
  502. void addSample(
  503. std::chrono::microseconds total,
  504. std::chrono::microseconds busy);
  505. double get() const {
  506. // Add the outstanding buffered times linearly, to avoid
  507. // expensive exponentiation
  508. auto lcoeff = buffer_time_.count() * -expCoeff_;
  509. return value_ * (1.0 - lcoeff) + lcoeff * busy_buffer_.count();
  510. }
  511. void dampen(double factor) {
  512. value_ *= factor;
  513. }
  514. private:
  515. double expCoeff_;
  516. double value_;
  517. std::chrono::microseconds buffer_time_{0};
  518. std::chrono::microseconds busy_buffer_{0};
  519. std::size_t buffer_cnt_{0};
  520. static constexpr std::chrono::milliseconds buffer_interval_{10};
  521. };
  522. void setObserver(const std::shared_ptr<EventBaseObserver>& observer) {
  523. assert(enableTimeMeasurement_);
  524. observer_ = observer;
  525. }
  526. const std::shared_ptr<EventBaseObserver>& getObserver() {
  527. return observer_;
  528. }
  529. /**
  530. * Setup execution observation/instrumentation for every EventHandler
  531. * executed in this EventBase.
  532. *
  533. * @param executionObserver EventHandle's execution observer.
  534. */
  535. void setExecutionObserver(ExecutionObserver* observer) {
  536. executionObserver_ = observer;
  537. }
  538. /**
  539. * Gets the execution observer associated with this EventBase.
  540. */
  541. ExecutionObserver* getExecutionObserver() {
  542. return executionObserver_;
  543. }
  544. /**
  545. * Set the name of the thread that runs this event base.
  546. */
  547. void setName(const std::string& name);
  548. /**
  549. * Returns the name of the thread that runs this event base.
  550. */
  551. const std::string& getName();
  552. /// Implements the Executor interface
  553. void add(Cob fn) override {
  554. // runInEventBaseThread() takes a const&,
  555. // so no point in doing std::move here.
  556. runInEventBaseThread(std::move(fn));
  557. }
  558. /// Implements the DrivableExecutor interface
  559. void drive() override {
  560. ++loopKeepAliveCount_;
  561. SCOPE_EXIT {
  562. --loopKeepAliveCount_;
  563. };
  564. loopOnce();
  565. }
  566. // Implements the ScheduledExecutor interface
  567. void scheduleAt(Func&& fn, TimePoint const& timeout) override;
  568. // TimeoutManager
  569. void attachTimeoutManager(
  570. AsyncTimeout* obj,
  571. TimeoutManager::InternalEnum internal) final;
  572. void detachTimeoutManager(AsyncTimeout* obj) final;
  573. bool scheduleTimeout(AsyncTimeout* obj, TimeoutManager::timeout_type timeout)
  574. final;
  575. void cancelTimeout(AsyncTimeout* obj) final;
  576. bool isInTimeoutManagerThread() final {
  577. return isInEventBaseThread();
  578. }
  579. // Returns a VirtualEventBase attached to this EventBase. Can be used to
  580. // pass to APIs which expect VirtualEventBase. This VirtualEventBase will be
  581. // destroyed together with the EventBase.
  582. //
  583. // Any number of VirtualEventBases instances may be independently constructed,
  584. // which are backed by this EventBase. This method should be only used if you
  585. // don't need to manage the life time of the VirtualEventBase used.
  586. folly::VirtualEventBase& getVirtualEventBase();
  587. /// Implements the IOExecutor interface
  588. EventBase* getEventBase() override;
  589. protected:
  590. bool keepAliveAcquire() override {
  591. if (inRunningEventBaseThread()) {
  592. loopKeepAliveCount_++;
  593. } else {
  594. loopKeepAliveCountAtomic_.fetch_add(1, std::memory_order_relaxed);
  595. }
  596. return true;
  597. }
  598. void keepAliveRelease() override {
  599. if (!inRunningEventBaseThread()) {
  600. return add([this] { loopKeepAliveCount_--; });
  601. }
  602. loopKeepAliveCount_--;
  603. }
  604. private:
  605. void applyLoopKeepAlive();
  606. ssize_t loopKeepAliveCount();
  607. /*
  608. * Helper function that tells us whether we have already handled
  609. * some event/timeout/callback in this loop iteration.
  610. */
  611. bool nothingHandledYet() const noexcept;
  612. typedef LoopCallback::List LoopCallbackList;
  613. class FunctionRunner;
  614. bool loopBody(int flags = 0, bool ignoreKeepAlive = false);
  615. // executes any callbacks queued by runInLoop(); returns false if none found
  616. bool runLoopCallbacks();
  617. void initNotificationQueue();
  618. // should only be accessed through public getter
  619. HHWheelTimer::UniquePtr wheelTimer_;
  620. LoopCallbackList loopCallbacks_;
  621. LoopCallbackList runBeforeLoopCallbacks_;
  622. LoopCallbackList onDestructionCallbacks_;
  623. // This will be null most of the time, but point to currentCallbacks
  624. // if we are in the middle of running loop callbacks, such that
  625. // runInLoop(..., true) will always run in the current loop
  626. // iteration.
  627. LoopCallbackList* runOnceCallbacks_;
  628. // stop_ is set by terminateLoopSoon() and is used by the main loop
  629. // to determine if it should exit
  630. std::atomic<bool> stop_;
  631. // The ID of the thread running the main loop.
  632. // std::thread::id{} if loop is not running.
  633. std::atomic<std::thread::id> loopThread_;
  634. // pointer to underlying event_base class doing the heavy lifting
  635. event_base* evb_;
  636. // A notification queue for runInEventBaseThread() to use
  637. // to send function requests to the EventBase thread.
  638. std::unique_ptr<NotificationQueue<Func>> queue_;
  639. std::unique_ptr<FunctionRunner> fnRunner_;
  640. ssize_t loopKeepAliveCount_{0};
  641. std::atomic<ssize_t> loopKeepAliveCountAtomic_{0};
  642. bool loopKeepAliveActive_{false};
  643. // limit for latency in microseconds (0 disables)
  644. std::chrono::microseconds maxLatency_;
  645. // exponentially-smoothed average loop time for latency-limiting
  646. SmoothLoopTime avgLoopTime_;
  647. // smoothed loop time used to invoke latency callbacks; differs from
  648. // avgLoopTime_ in that it's scaled down after triggering a callback
  649. // to reduce spamminess
  650. SmoothLoopTime maxLatencyLoopTime_;
  651. // callback called when latency limit is exceeded
  652. Func maxLatencyCob_;
  653. // Enables/disables time measurements in loopBody(). if disabled, the
  654. // following functionality that relies on time-measurement, will not
  655. // be supported: avg loop time, observer and max latency.
  656. const bool enableTimeMeasurement_;
  657. // Wrap-around loop counter to detect beginning of each loop
  658. std::size_t nextLoopCnt_;
  659. std::size_t latestLoopCnt_;
  660. std::chrono::steady_clock::time_point startWork_;
  661. // Prevent undefined behavior from invoking event_base_loop() reentrantly.
  662. // This is needed since many projects use libevent-1.4, which lacks commit
  663. // b557b175c00dc462c1fce25f6e7dd67121d2c001 from
  664. // https://github.com/libevent/libevent/.
  665. bool invokingLoop_{false};
  666. // Observer to export counters
  667. std::shared_ptr<EventBaseObserver> observer_;
  668. uint32_t observerSampleCount_;
  669. // EventHandler's execution observer.
  670. ExecutionObserver* executionObserver_;
  671. // Name of the thread running this EventBase
  672. std::string name_;
  673. // allow runOnDestruction() to be called from any threads
  674. std::mutex onDestructionCallbacksMutex_;
  675. // see EventBaseLocal
  676. friend class detail::EventBaseLocalBase;
  677. template <typename T>
  678. friend class EventBaseLocal;
  679. std::unordered_map<std::size_t, std::shared_ptr<void>> localStorage_;
  680. std::unordered_set<detail::EventBaseLocalBaseBase*> localStorageToDtor_;
  681. folly::once_flag virtualEventBaseInitFlag_;
  682. std::unique_ptr<VirtualEventBase> virtualEventBase_;
  683. };
  684. template <typename T>
  685. bool EventBase::runInEventBaseThread(void (*fn)(T*), T* arg) {
  686. return runInEventBaseThread([=] { fn(arg); });
  687. }
  688. template <typename T>
  689. bool EventBase::runInEventBaseThreadAndWait(void (*fn)(T*), T* arg) {
  690. return runInEventBaseThreadAndWait([=] { fn(arg); });
  691. }
  692. template <typename T>
  693. bool EventBase::runImmediatelyOrRunInEventBaseThreadAndWait(
  694. void (*fn)(T*),
  695. T* arg) {
  696. return runImmediatelyOrRunInEventBaseThreadAndWait([=] { fn(arg); });
  697. }
  698. } // namespace folly