OLD | NEW |
| (Empty) |
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 #include "base/message_loop/message_pump_libevent.h" | |
6 | |
7 #include <errno.h> | |
8 #include <fcntl.h> | |
9 #include <unistd.h> | |
10 | |
11 #include "base/auto_reset.h" | |
12 #include "base/compiler_specific.h" | |
13 #include "base/logging.h" | |
14 #include "base/memory/scoped_ptr.h" | |
15 #include "base/observer_list.h" | |
16 #include "base/posix/eintr_wrapper.h" | |
17 #include "base/time.h" | |
18 #include "third_party/libevent/event.h" | |
19 | |
20 #if defined(OS_MACOSX) | |
21 #include "base/mac/scoped_nsautorelease_pool.h" | |
22 #endif | |
23 | |
24 // Lifecycle of struct event | |
25 // Libevent uses two main data structures: | |
26 // struct event_base (of which there is one per message pump), and | |
27 // struct event (of which there is roughly one per socket). | |
28 // The socket's struct event is created in | |
29 // MessagePumpLibevent::WatchFileDescriptor(), | |
30 // is owned by the FileDescriptorWatcher, and is destroyed in | |
31 // StopWatchingFileDescriptor(). | |
32 // It is moved into and out of lists in struct event_base by | |
33 // the libevent functions event_add() and event_del(). | |
34 // | |
35 // TODO(dkegel): | |
36 // At the moment bad things happen if a FileDescriptorWatcher | |
37 // is active after its MessagePumpLibevent has been destroyed. | |
38 // See MessageLoopTest.FileDescriptorWatcherOutlivesMessageLoop | |
39 // Not clear yet whether that situation occurs in practice, | |
40 // but if it does, we need to fix it. | |
41 | |
42 namespace base { | |
43 | |
44 // Return 0 on success | |
45 // Too small a function to bother putting in a library? | |
46 static int SetNonBlocking(int fd) { | |
47 int flags = fcntl(fd, F_GETFL, 0); | |
48 if (flags == -1) | |
49 flags = 0; | |
50 return fcntl(fd, F_SETFL, flags | O_NONBLOCK); | |
51 } | |
52 | |
53 MessagePumpLibevent::FileDescriptorWatcher::FileDescriptorWatcher() | |
54 : event_(NULL), | |
55 pump_(NULL), | |
56 watcher_(NULL), | |
57 weak_factory_(this) { | |
58 } | |
59 | |
60 MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() { | |
61 if (event_) { | |
62 StopWatchingFileDescriptor(); | |
63 } | |
64 } | |
65 | |
66 bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() { | |
67 event* e = ReleaseEvent(); | |
68 if (e == NULL) | |
69 return true; | |
70 | |
71 // event_del() is a no-op if the event isn't active. | |
72 int rv = event_del(e); | |
73 delete e; | |
74 pump_ = NULL; | |
75 watcher_ = NULL; | |
76 return (rv == 0); | |
77 } | |
78 | |
79 void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e) { | |
80 DCHECK(e); | |
81 DCHECK(!event_); | |
82 | |
83 event_ = e; | |
84 } | |
85 | |
86 event *MessagePumpLibevent::FileDescriptorWatcher::ReleaseEvent() { | |
87 struct event *e = event_; | |
88 event_ = NULL; | |
89 return e; | |
90 } | |
91 | |
92 void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanReadWithoutBlocking( | |
93 int fd, MessagePumpLibevent* pump) { | |
94 // Since OnFileCanWriteWithoutBlocking() gets called first, it can stop | |
95 // watching the file descriptor. | |
96 if (!watcher_) | |
97 return; | |
98 pump->WillProcessIOEvent(); | |
99 watcher_->OnFileCanReadWithoutBlocking(fd); | |
100 pump->DidProcessIOEvent(); | |
101 } | |
102 | |
103 void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking( | |
104 int fd, MessagePumpLibevent* pump) { | |
105 DCHECK(watcher_); | |
106 pump->WillProcessIOEvent(); | |
107 watcher_->OnFileCanWriteWithoutBlocking(fd); | |
108 pump->DidProcessIOEvent(); | |
109 } | |
110 | |
111 MessagePumpLibevent::MessagePumpLibevent() | |
112 : keep_running_(true), | |
113 in_run_(false), | |
114 processed_io_events_(false), | |
115 event_base_(event_base_new()), | |
116 wakeup_pipe_in_(-1), | |
117 wakeup_pipe_out_(-1) { | |
118 if (!Init()) | |
119 NOTREACHED(); | |
120 } | |
121 | |
122 MessagePumpLibevent::~MessagePumpLibevent() { | |
123 DCHECK(wakeup_event_); | |
124 DCHECK(event_base_); | |
125 event_del(wakeup_event_); | |
126 delete wakeup_event_; | |
127 if (wakeup_pipe_in_ >= 0) { | |
128 if (HANDLE_EINTR(close(wakeup_pipe_in_)) < 0) | |
129 DPLOG(ERROR) << "close"; | |
130 } | |
131 if (wakeup_pipe_out_ >= 0) { | |
132 if (HANDLE_EINTR(close(wakeup_pipe_out_)) < 0) | |
133 DPLOG(ERROR) << "close"; | |
134 } | |
135 event_base_free(event_base_); | |
136 } | |
137 | |
138 bool MessagePumpLibevent::WatchFileDescriptor(int fd, | |
139 bool persistent, | |
140 int mode, | |
141 FileDescriptorWatcher *controller, | |
142 Watcher *delegate) { | |
143 DCHECK_GE(fd, 0); | |
144 DCHECK(controller); | |
145 DCHECK(delegate); | |
146 DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE); | |
147 // WatchFileDescriptor should be called on the pump thread. It is not | |
148 // threadsafe, and your watcher may never be registered. | |
149 DCHECK(watch_file_descriptor_caller_checker_.CalledOnValidThread()); | |
150 | |
151 int event_mask = persistent ? EV_PERSIST : 0; | |
152 if (mode & WATCH_READ) { | |
153 event_mask |= EV_READ; | |
154 } | |
155 if (mode & WATCH_WRITE) { | |
156 event_mask |= EV_WRITE; | |
157 } | |
158 | |
159 scoped_ptr<event> evt(controller->ReleaseEvent()); | |
160 if (evt.get() == NULL) { | |
161 // Ownership is transferred to the controller. | |
162 evt.reset(new event); | |
163 } else { | |
164 // Make sure we don't pick up any funky internal libevent masks. | |
165 int old_interest_mask = evt.get()->ev_events & | |
166 (EV_READ | EV_WRITE | EV_PERSIST); | |
167 | |
168 // Combine old/new event masks. | |
169 event_mask |= old_interest_mask; | |
170 | |
171 // Must disarm the event before we can reuse it. | |
172 event_del(evt.get()); | |
173 | |
174 // It's illegal to use this function to listen on 2 separate fds with the | |
175 // same |controller|. | |
176 if (EVENT_FD(evt.get()) != fd) { | |
177 NOTREACHED() << "FDs don't match" << EVENT_FD(evt.get()) << "!=" << fd; | |
178 return false; | |
179 } | |
180 } | |
181 | |
182 // Set current interest mask and message pump for this event. | |
183 event_set(evt.get(), fd, event_mask, OnLibeventNotification, controller); | |
184 | |
185 // Tell libevent which message pump this socket will belong to when we add it. | |
186 if (event_base_set(event_base_, evt.get())) { | |
187 return false; | |
188 } | |
189 | |
190 // Add this socket to the list of monitored sockets. | |
191 if (event_add(evt.get(), NULL)) { | |
192 return false; | |
193 } | |
194 | |
195 // Transfer ownership of evt to controller. | |
196 controller->Init(evt.release()); | |
197 | |
198 controller->set_watcher(delegate); | |
199 controller->set_pump(this); | |
200 | |
201 return true; | |
202 } | |
203 | |
204 void MessagePumpLibevent::AddIOObserver(IOObserver *obs) { | |
205 io_observers_.AddObserver(obs); | |
206 } | |
207 | |
208 void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) { | |
209 io_observers_.RemoveObserver(obs); | |
210 } | |
211 | |
212 // Tell libevent to break out of inner loop. | |
213 static void timer_callback(int fd, short events, void *context) | |
214 { | |
215 event_base_loopbreak((struct event_base *)context); | |
216 } | |
217 | |
218 // Reentrant! | |
219 void MessagePumpLibevent::Run(Delegate* delegate) { | |
220 DCHECK(keep_running_) << "Quit must have been called outside of Run!"; | |
221 AutoReset<bool> auto_reset_in_run(&in_run_, true); | |
222 | |
223 // event_base_loopexit() + EVLOOP_ONCE is leaky, see http://crbug.com/25641. | |
224 // Instead, make our own timer and reuse it on each call to event_base_loop(). | |
225 scoped_ptr<event> timer_event(new event); | |
226 | |
227 for (;;) { | |
228 #if defined(OS_MACOSX) | |
229 mac::ScopedNSAutoreleasePool autorelease_pool; | |
230 #endif | |
231 | |
232 bool did_work = delegate->DoWork(); | |
233 if (!keep_running_) | |
234 break; | |
235 | |
236 event_base_loop(event_base_, EVLOOP_NONBLOCK); | |
237 did_work |= processed_io_events_; | |
238 processed_io_events_ = false; | |
239 if (!keep_running_) | |
240 break; | |
241 | |
242 did_work |= delegate->DoDelayedWork(&delayed_work_time_); | |
243 if (!keep_running_) | |
244 break; | |
245 | |
246 if (did_work) | |
247 continue; | |
248 | |
249 did_work = delegate->DoIdleWork(); | |
250 if (!keep_running_) | |
251 break; | |
252 | |
253 if (did_work) | |
254 continue; | |
255 | |
256 // EVLOOP_ONCE tells libevent to only block once, | |
257 // but to service all pending events when it wakes up. | |
258 if (delayed_work_time_.is_null()) { | |
259 event_base_loop(event_base_, EVLOOP_ONCE); | |
260 } else { | |
261 TimeDelta delay = delayed_work_time_ - TimeTicks::Now(); | |
262 if (delay > TimeDelta()) { | |
263 struct timeval poll_tv; | |
264 poll_tv.tv_sec = delay.InSeconds(); | |
265 poll_tv.tv_usec = delay.InMicroseconds() % Time::kMicrosecondsPerSecond; | |
266 event_set(timer_event.get(), -1, 0, timer_callback, event_base_); | |
267 event_base_set(event_base_, timer_event.get()); | |
268 event_add(timer_event.get(), &poll_tv); | |
269 event_base_loop(event_base_, EVLOOP_ONCE); | |
270 event_del(timer_event.get()); | |
271 } else { | |
272 // It looks like delayed_work_time_ indicates a time in the past, so we | |
273 // need to call DoDelayedWork now. | |
274 delayed_work_time_ = TimeTicks(); | |
275 } | |
276 } | |
277 } | |
278 | |
279 keep_running_ = true; | |
280 } | |
281 | |
282 void MessagePumpLibevent::Quit() { | |
283 DCHECK(in_run_); | |
284 // Tell both libevent and Run that they should break out of their loops. | |
285 keep_running_ = false; | |
286 ScheduleWork(); | |
287 } | |
288 | |
289 void MessagePumpLibevent::ScheduleWork() { | |
290 // Tell libevent (in a threadsafe way) that it should break out of its loop. | |
291 char buf = 0; | |
292 int nwrite = HANDLE_EINTR(write(wakeup_pipe_in_, &buf, 1)); | |
293 DCHECK(nwrite == 1 || errno == EAGAIN) | |
294 << "[nwrite:" << nwrite << "] [errno:" << errno << "]"; | |
295 } | |
296 | |
297 void MessagePumpLibevent::ScheduleDelayedWork( | |
298 const TimeTicks& delayed_work_time) { | |
299 // We know that we can't be blocked on Wait right now since this method can | |
300 // only be called on the same thread as Run, so we only need to update our | |
301 // record of how long to sleep when we do sleep. | |
302 delayed_work_time_ = delayed_work_time; | |
303 } | |
304 | |
305 void MessagePumpLibevent::WillProcessIOEvent() { | |
306 FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent()); | |
307 } | |
308 | |
309 void MessagePumpLibevent::DidProcessIOEvent() { | |
310 FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent()); | |
311 } | |
312 | |
313 bool MessagePumpLibevent::Init() { | |
314 int fds[2]; | |
315 if (pipe(fds)) { | |
316 DLOG(ERROR) << "pipe() failed, errno: " << errno; | |
317 return false; | |
318 } | |
319 if (SetNonBlocking(fds[0])) { | |
320 DLOG(ERROR) << "SetNonBlocking for pipe fd[0] failed, errno: " << errno; | |
321 return false; | |
322 } | |
323 if (SetNonBlocking(fds[1])) { | |
324 DLOG(ERROR) << "SetNonBlocking for pipe fd[1] failed, errno: " << errno; | |
325 return false; | |
326 } | |
327 wakeup_pipe_out_ = fds[0]; | |
328 wakeup_pipe_in_ = fds[1]; | |
329 | |
330 wakeup_event_ = new event; | |
331 event_set(wakeup_event_, wakeup_pipe_out_, EV_READ | EV_PERSIST, | |
332 OnWakeup, this); | |
333 event_base_set(event_base_, wakeup_event_); | |
334 | |
335 if (event_add(wakeup_event_, 0)) | |
336 return false; | |
337 return true; | |
338 } | |
339 | |
340 // static | |
341 void MessagePumpLibevent::OnLibeventNotification(int fd, short flags, | |
342 void* context) { | |
343 WeakPtr<FileDescriptorWatcher> controller = | |
344 static_cast<FileDescriptorWatcher*>(context)->weak_factory_.GetWeakPtr(); | |
345 DCHECK(controller.get()); | |
346 | |
347 MessagePumpLibevent* pump = controller->pump(); | |
348 pump->processed_io_events_ = true; | |
349 | |
350 if (flags & EV_WRITE) { | |
351 controller->OnFileCanWriteWithoutBlocking(fd, pump); | |
352 } | |
353 // Check |controller| in case it's been deleted in | |
354 // controller->OnFileCanWriteWithoutBlocking(). | |
355 if (controller.get() && flags & EV_READ) { | |
356 controller->OnFileCanReadWithoutBlocking(fd, pump); | |
357 } | |
358 } | |
359 | |
360 // Called if a byte is received on the wakeup pipe. | |
361 // static | |
362 void MessagePumpLibevent::OnWakeup(int socket, short flags, void* context) { | |
363 MessagePumpLibevent* that = static_cast<MessagePumpLibevent*>(context); | |
364 DCHECK(that->wakeup_pipe_out_ == socket); | |
365 | |
366 // Remove and discard the wakeup byte. | |
367 char buf; | |
368 int nread = HANDLE_EINTR(read(socket, &buf, 1)); | |
369 DCHECK_EQ(nread, 1); | |
370 that->processed_io_events_ = true; | |
371 // Tell libevent to break out of inner loop. | |
372 event_base_loopbreak(that->event_base_); | |
373 } | |
374 | |
375 } // namespace base | |
OLD | NEW |