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