| 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 |
|---|
Chromium Code Reviews
Issue 16897005:
Revert 206507 "Move message_pump to base/message_loop." (Closed)
Base URL: svn://svn.chromium.org/chrome/