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