base/message_loop/message_pump_libevent.cc

   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