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