EventBase::runAfterDelay to throw an exception
[folly.git] / folly / io / async / EventBase.h
1 /*
2  * Copyright 2014 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    * Runs the given Cob at some time after the specified number of
387    * milliseconds.  (No guarantees exactly when.)
388    *
389    * Throws a std::system_error if an error occurs.
390    */
391   void runAfterDelay(
392       const Cob& c,
393       int milliseconds,
394       TimeoutManager::InternalEnum in = TimeoutManager::InternalEnum::NORMAL);
395
396   /**
397    * @see tryRunAfterDelay for more details
398    *
399    * @return  true iff the cob was successfully registered.
400    *
401    * */
402   bool tryRunAfterDelay(
403       const Cob& cob,
404       int milliseconds,
405       TimeoutManager::InternalEnum in = TimeoutManager::InternalEnum::NORMAL);
406
407   /**
408    * Set the maximum desired latency in us and provide a callback which will be
409    * called when that latency is exceeded.
410    * OBS: This functionality depends on time-measurement.
411    */
412   void setMaxLatency(int64_t maxLatency, const Cob& maxLatencyCob) {
413     assert(enableTimeMeasurement_);
414     maxLatency_ = maxLatency;
415     maxLatencyCob_ = maxLatencyCob;
416   }
417
418
419   /**
420    * Set smoothing coefficient for loop load average; # of milliseconds
421    * for exp(-1) (1/2.71828...) decay.
422    */
423   void setLoadAvgMsec(uint32_t ms);
424
425   /**
426    * reset the load average to a desired value
427    */
428   void resetLoadAvg(double value = 0.0);
429
430   /**
431    * Get the average loop time in microseconds (an exponentially-smoothed ave)
432    */
433   double getAvgLoopTime() const {
434     assert(enableTimeMeasurement_);
435     return avgLoopTime_.get();
436   }
437
438   /**
439     * check if the event base loop is running.
440    */
441   bool isRunning() const {
442     return loopThread_.load(std::memory_order_relaxed) != 0;
443   }
444
445   /**
446    * wait until the event loop starts (after starting the event loop thread).
447    */
448   void waitUntilRunning();
449
450   int getNotificationQueueSize() const;
451
452   void setMaxReadAtOnce(uint32_t maxAtOnce);
453
454   /**
455    * Verify that current thread is the EventBase thread, if the EventBase is
456    * running.
457    */
458   bool isInEventBaseThread() const {
459     auto tid = loopThread_.load(std::memory_order_relaxed);
460     return tid == 0 || pthread_equal(tid, pthread_self());
461   }
462
463   bool inRunningEventBaseThread() const {
464     return pthread_equal(
465       loopThread_.load(std::memory_order_relaxed), pthread_self());
466   }
467
468   // --------- interface to underlying libevent base ------------
469   // Avoid using these functions if possible.  These functions are not
470   // guaranteed to always be present if we ever provide alternative EventBase
471   // implementations that do not use libevent internally.
472   event_base* getLibeventBase() const { return evb_; }
473   static const char* getLibeventVersion();
474   static const char* getLibeventMethod();
475
476   /**
477    * only EventHandler/AsyncTimeout subclasses and ourselves should
478    * ever call this.
479    *
480    * This is used to mark the beginning of a new loop cycle by the
481    * first handler fired within that cycle.
482    *
483    */
484   bool bumpHandlingTime() override;
485
486   class SmoothLoopTime {
487    public:
488     explicit SmoothLoopTime(uint64_t timeInterval)
489       : expCoeff_(-1.0/timeInterval)
490       , value_(0.0)
491       , oldBusyLeftover_(0) {
492       VLOG(11) << "expCoeff_ " << expCoeff_ << " " << __PRETTY_FUNCTION__;
493     }
494
495     void setTimeInterval(uint64_t timeInterval);
496     void reset(double value = 0.0);
497
498     void addSample(int64_t idle, int64_t busy);
499
500     double get() const {
501       return value_;
502     }
503
504     void dampen(double factor) {
505       value_ *= factor;
506     }
507
508    private:
509     double  expCoeff_;
510     double  value_;
511     int64_t oldBusyLeftover_;
512   };
513
514   void setObserver(const std::shared_ptr<EventBaseObserver>& observer) {
515     assert(enableTimeMeasurement_);
516     observer_ = observer;
517   }
518
519   const std::shared_ptr<EventBaseObserver>& getObserver() {
520     return observer_;
521   }
522
523   /**
524    * Set the name of the thread that runs this event base.
525    */
526   void setName(const std::string& name);
527
528   /**
529    * Returns the name of the thread that runs this event base.
530    */
531   const std::string& getName();
532
533   /// Implements the Executor interface
534   void add(Cob fn) override {
535     // runInEventBaseThread() takes a const&,
536     // so no point in doing std::move here.
537     runInEventBaseThread(fn);
538   }
539
540   /// Implements the DrivableExecutor interface
541   void drive() override {
542     loopOnce();
543   }
544
545  private:
546
547   // TimeoutManager
548   void attachTimeoutManager(AsyncTimeout* obj,
549                             TimeoutManager::InternalEnum internal) override;
550
551   void detachTimeoutManager(AsyncTimeout* obj) override;
552
553   bool scheduleTimeout(AsyncTimeout* obj, std::chrono::milliseconds timeout)
554     override;
555
556   void cancelTimeout(AsyncTimeout* obj) override;
557
558   bool isInTimeoutManagerThread() override {
559     return isInEventBaseThread();
560   }
561
562   // Helper class used to short circuit runInEventBaseThread
563   class RunInLoopCallback : public LoopCallback {
564    public:
565     RunInLoopCallback(void (*fn)(void*), void* arg);
566     void runLoopCallback() noexcept;
567
568    private:
569     void (*fn_)(void*);
570     void* arg_;
571   };
572
573   /*
574    * Helper function that tells us whether we have already handled
575    * some event/timeout/callback in this loop iteration.
576    */
577   bool nothingHandledYet();
578
579   // --------- libevent callbacks (not for client use) ------------
580
581   static void runFunctionPtr(std::function<void()>* fn);
582
583   // small object used as a callback arg with enough info to execute the
584   // appropriate client-provided Cob
585   class CobTimeout : public AsyncTimeout {
586    public:
587     CobTimeout(EventBase* b, const Cob& c, TimeoutManager::InternalEnum in)
588         : AsyncTimeout(b, in), cob_(c) {}
589
590     virtual void timeoutExpired() noexcept;
591
592    private:
593     Cob cob_;
594
595    public:
596     typedef boost::intrusive::list_member_hook<
597       boost::intrusive::link_mode<boost::intrusive::auto_unlink> > ListHook;
598
599     ListHook hook;
600
601     typedef boost::intrusive::list<
602       CobTimeout,
603       boost::intrusive::member_hook<CobTimeout, ListHook, &CobTimeout::hook>,
604       boost::intrusive::constant_time_size<false> > List;
605   };
606
607   typedef LoopCallback::List LoopCallbackList;
608   class FunctionRunner;
609
610   bool loopBody(int flags = 0);
611
612   // executes any callbacks queued by runInLoop(); returns false if none found
613   bool runLoopCallbacks(bool setContext = true);
614
615   void initNotificationQueue();
616
617   CobTimeout::List pendingCobTimeouts_;
618
619   LoopCallbackList loopCallbacks_;
620   LoopCallbackList runBeforeLoopCallbacks_;
621   LoopCallbackList onDestructionCallbacks_;
622
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
629   // stop_ is set by terminateLoopSoon() and is used by the main loop
630   // to determine if it should exit
631   bool stop_;
632
633   // The ID of the thread running the main loop.
634   // 0 if loop is not running.
635   // Note: POSIX doesn't guarantee that 0 is an invalid pthread_t (or
636   // even that atomic<pthread_t> is valid), but that's how it is
637   // everywhere (at least on Linux, FreeBSD, and OSX).
638   std::atomic<pthread_t> loopThread_;
639
640   // pointer to underlying event_base class doing the heavy lifting
641   event_base* evb_;
642
643   // A notification queue for runInEventBaseThread() to use
644   // to send function requests to the EventBase thread.
645   std::unique_ptr<NotificationQueue<std::pair<void (*)(void*), void*>>> queue_;
646   std::unique_ptr<FunctionRunner> fnRunner_;
647
648   // limit for latency in microseconds (0 disables)
649   int64_t maxLatency_;
650
651   // exponentially-smoothed average loop time for latency-limiting
652   SmoothLoopTime avgLoopTime_;
653
654   // smoothed loop time used to invoke latency callbacks; differs from
655   // avgLoopTime_ in that it's scaled down after triggering a callback
656   // to reduce spamminess
657   SmoothLoopTime maxLatencyLoopTime_;
658
659   // callback called when latency limit is exceeded
660   Cob maxLatencyCob_;
661
662   // Enables/disables time measurements in loopBody(). if disabled, the
663   // following functionality that relies on time-measurement, will not
664   // be supported: avg loop time, observer and max latency.
665   const bool enableTimeMeasurement_;
666
667   // we'll wait this long before running deferred callbacks if the event
668   // loop is idle.
669   static const int kDEFAULT_IDLE_WAIT_USEC = 20000; // 20ms
670
671   // Wrap-around loop counter to detect beginning of each loop
672   uint64_t nextLoopCnt_;
673   uint64_t latestLoopCnt_;
674   uint64_t startWork_;
675
676   // Observer to export counters
677   std::shared_ptr<EventBaseObserver> observer_;
678   uint32_t observerSampleCount_;
679
680   // Name of the thread running this EventBase
681   std::string name_;
682 };
683
684 } // folly