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