1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4 
5 #ifndef BASE_MESSAGE_LOOP_MESSAGE_PUMP_LIBEVENT_H_
6 #define BASE_MESSAGE_LOOP_MESSAGE_PUMP_LIBEVENT_H_
7 
8 #include "base/compiler_specific.h"
9 #include "base/macros.h"
10 #include "base/message_loop/message_pump.h"
11 #include "base/observer_list.h"
12 #include "base/threading/thread_checker.h"
13 #include "base/time/time.h"
14 
15 // Declare structs we need from libevent.h rather than including it
16 struct event_base;
17 struct event;
18 
19 namespace base {
20 
21 // Class to monitor sockets and issue callbacks when sockets are ready for I/O
22 // TODO(dkegel): add support for background file IO somehow
23 class BASE_EXPORT MessagePumpLibevent : public MessagePump {
24  public:
25   class IOObserver {
26    public:
IOObserver()27     IOObserver() {}
28 
29     // An IOObserver is an object that receives IO notifications from the
30     // MessagePump.
31     //
32     // NOTE: An IOObserver implementation should be extremely fast!
33     virtual void WillProcessIOEvent() = 0;
34     virtual void DidProcessIOEvent() = 0;
35 
36    protected:
~IOObserver()37     virtual ~IOObserver() {}
38   };
39 
40   // Used with WatchFileDescriptor to asynchronously monitor the I/O readiness
41   // of a file descriptor.
42   class Watcher {
43    public:
44     // Called from MessageLoop::Run when an FD can be read from/written to
45     // without blocking
46     virtual void OnFileCanReadWithoutBlocking(int fd) = 0;
47     virtual void OnFileCanWriteWithoutBlocking(int fd) = 0;
48 
49    protected:
~Watcher()50     virtual ~Watcher() {}
51   };
52 
53   // Object returned by WatchFileDescriptor to manage further watching.
54   class FileDescriptorWatcher {
55    public:
56     FileDescriptorWatcher();
57     ~FileDescriptorWatcher();  // Implicitly calls StopWatchingFileDescriptor.
58 
59     // NOTE: These methods aren't called StartWatching()/StopWatching() to
60     // avoid confusion with the win32 ObjectWatcher class.
61 
62     // Stop watching the FD, always safe to call.  No-op if there's nothing
63     // to do.
64     bool StopWatchingFileDescriptor();
65 
66    private:
67     friend class MessagePumpLibevent;
68     friend class MessagePumpLibeventTest;
69 
70     // Called by MessagePumpLibevent, ownership of |e| is transferred to this
71     // object.
72     void Init(event* e);
73 
74     // Used by MessagePumpLibevent to take ownership of event_.
75     event* ReleaseEvent();
76 
set_pump(MessagePumpLibevent * pump)77     void set_pump(MessagePumpLibevent* pump) { pump_ = pump; }
pump()78     MessagePumpLibevent* pump() const { return pump_; }
79 
set_watcher(Watcher * watcher)80     void set_watcher(Watcher* watcher) { watcher_ = watcher; }
81 
82     void OnFileCanReadWithoutBlocking(int fd, MessagePumpLibevent* pump);
83     void OnFileCanWriteWithoutBlocking(int fd, MessagePumpLibevent* pump);
84 
85     event* event_;
86     MessagePumpLibevent* pump_;
87     Watcher* watcher_;
88     // If this pointer is non-NULL, the pointee is set to true in the
89     // destructor.
90     bool* was_destroyed_;
91 
92     DISALLOW_COPY_AND_ASSIGN(FileDescriptorWatcher);
93   };
94 
95   enum Mode {
96     WATCH_READ = 1 << 0,
97     WATCH_WRITE = 1 << 1,
98     WATCH_READ_WRITE = WATCH_READ | WATCH_WRITE
99   };
100 
101   MessagePumpLibevent();
102   ~MessagePumpLibevent() override;
103 
104   // Have the current thread's message loop watch for a a situation in which
105   // reading/writing to the FD can be performed without blocking.
106   // Callers must provide a preallocated FileDescriptorWatcher object which
107   // can later be used to manage the lifetime of this event.
108   // If a FileDescriptorWatcher is passed in which is already attached to
109   // an event, then the effect is cumulative i.e. after the call |controller|
110   // will watch both the previous event and the new one.
111   // If an error occurs while calling this method in a cumulative fashion, the
112   // event previously attached to |controller| is aborted.
113   // Returns true on success.
114   // Must be called on the same thread the message_pump is running on.
115   // TODO(dkegel): switch to edge-triggered readiness notification
116   bool WatchFileDescriptor(int fd,
117                            bool persistent,
118                            int mode,
119                            FileDescriptorWatcher *controller,
120                            Watcher *delegate);
121 
122   void AddIOObserver(IOObserver* obs);
123   void RemoveIOObserver(IOObserver* obs);
124 
125   // MessagePump methods:
126   void Run(Delegate* delegate) override;
127   void Quit() override;
128   void ScheduleWork() override;
129   void ScheduleDelayedWork(const TimeTicks& delayed_work_time) override;
130 
131  private:
132   friend class MessagePumpLibeventTest;
133 
134   void WillProcessIOEvent();
135   void DidProcessIOEvent();
136 
137   // Risky part of constructor.  Returns true on success.
138   bool Init();
139 
140   // Called by libevent to tell us a registered FD can be read/written to.
141   static void OnLibeventNotification(int fd, short flags,
142                                      void* context);
143 
144   // Unix pipe used to implement ScheduleWork()
145   // ... callback; called by libevent inside Run() when pipe is ready to read
146   static void OnWakeup(int socket, short flags, void* context);
147 
148   // This flag is set to false when Run should return.
149   bool keep_running_;
150 
151   // This flag is set when inside Run.
152   bool in_run_;
153 
154   // This flag is set if libevent has processed I/O events.
155   bool processed_io_events_;
156 
157   // The time at which we should call DoDelayedWork.
158   TimeTicks delayed_work_time_;
159 
160   // Libevent dispatcher.  Watches all sockets registered with it, and sends
161   // readiness callbacks when a socket is ready for I/O.
162   event_base* event_base_;
163 
164   // ... write end; ScheduleWork() writes a single byte to it
165   int wakeup_pipe_in_;
166   // ... read end; OnWakeup reads it and then breaks Run() out of its sleep
167   int wakeup_pipe_out_;
168   // ... libevent wrapper for read end
169   event* wakeup_event_;
170 
171   ObserverList<IOObserver> io_observers_;
172   ThreadChecker watch_file_descriptor_caller_checker_;
173   DISALLOW_COPY_AND_ASSIGN(MessagePumpLibevent);
174 };
175 
176 }  // namespace base
177 
178 #endif  // BASE_MESSAGE_LOOP_MESSAGE_PUMP_LIBEVENT_H_
179