Move message_pump to base/message_loop.

base/message_pump_libevent.cc

Index: base/message_pump_libevent.cc
diff --git a/base/message_pump_libevent.cc b/base/message_pump_libevent.cc
deleted file mode 100644
index 364f44a5f70eb2119d36174afb3023ae08cf733e..0000000000000000000000000000000000000000
--- a/base/message_pump_libevent.cc
+++ /dev/null
@@ -1,379 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "base/message_pump_libevent.h"
-
-#include <errno.h>
-#include <fcntl.h>
-#include <unistd.h>
-
-#include "base/auto_reset.h"
-#include "base/compiler_specific.h"
-#include "base/logging.h"
-#if defined(OS_MACOSX)
-#include "base/mac/scoped_nsautorelease_pool.h"
-#endif
-#include "base/memory/scoped_ptr.h"
-#include "base/observer_list.h"
-#include "base/posix/eintr_wrapper.h"
-#include "base/time.h"
-#include "third_party/libevent/event.h"
-
-#if defined(OS_MACOSX)
-#include "base/mac/scoped_nsautorelease_pool.h"
-#endif
-
-// Lifecycle of struct event
-// Libevent uses two main data structures:
-// struct event_base (of which there is one per message pump), and
-// struct event (of which there is roughly one per socket).
-// The socket's struct event is created in
-// MessagePumpLibevent::WatchFileDescriptor(),
-// is owned by the FileDescriptorWatcher, and is destroyed in
-// StopWatchingFileDescriptor().
-// It is moved into and out of lists in struct event_base by
-// the libevent functions event_add() and event_del().
-//
-// TODO(dkegel):
-// At the moment bad things happen if a FileDescriptorWatcher
-// is active after its MessagePumpLibevent has been destroyed.
-// See MessageLoopTest.FileDescriptorWatcherOutlivesMessageLoop
-// Not clear yet whether that situation occurs in practice,
-// but if it does, we need to fix it.
-
-namespace base {
-
-// Return 0 on success
-// Too small a function to bother putting in a library?
-static int SetNonBlocking(int fd) {
-  int flags = fcntl(fd, F_GETFL, 0);
-  if (flags == -1)
-    flags = 0;
-  return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
-}
-
-MessagePumpLibevent::FileDescriptorWatcher::FileDescriptorWatcher()
-    : event_(NULL),
-      pump_(NULL),
-      watcher_(NULL),
-      weak_factory_(this) {
-}
-
-MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() {
-  if (event_) {
-    StopWatchingFileDescriptor();
-  }
-}
-
-bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() {
-  event* e = ReleaseEvent();
-  if (e == NULL)
-    return true;
-
-  // event_del() is a no-op if the event isn't active.
-  int rv = event_del(e);
-  delete e;
-  pump_ = NULL;
-  watcher_ = NULL;
-  return (rv == 0);
-}
-
-void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e) {
-  DCHECK(e);
-  DCHECK(!event_);
-
-  event_ = e;
-}
-
-event *MessagePumpLibevent::FileDescriptorWatcher::ReleaseEvent() {
-  struct event *e = event_;
-  event_ = NULL;
-  return e;
-}
-
-void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanReadWithoutBlocking(
-    int fd, MessagePumpLibevent* pump) {
-  // Since OnFileCanWriteWithoutBlocking() gets called first, it can stop
-  // watching the file descriptor.
-  if (!watcher_)
-    return;
-  pump->WillProcessIOEvent();
-  watcher_->OnFileCanReadWithoutBlocking(fd);
-  pump->DidProcessIOEvent();
-}
-
-void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking(
-    int fd, MessagePumpLibevent* pump) {
-  DCHECK(watcher_);
-  pump->WillProcessIOEvent();
-  watcher_->OnFileCanWriteWithoutBlocking(fd);
-  pump->DidProcessIOEvent();
-}
-
-MessagePumpLibevent::MessagePumpLibevent()
-    : keep_running_(true),
-      in_run_(false),
-      processed_io_events_(false),
-      event_base_(event_base_new()),
-      wakeup_pipe_in_(-1),
-      wakeup_pipe_out_(-1) {
-  if (!Init())
-     NOTREACHED();
-}
-
-MessagePumpLibevent::~MessagePumpLibevent() {
-  DCHECK(wakeup_event_);
-  DCHECK(event_base_);
-  event_del(wakeup_event_);
-  delete wakeup_event_;
-  if (wakeup_pipe_in_ >= 0) {
-    if (HANDLE_EINTR(close(wakeup_pipe_in_)) < 0)
-      DPLOG(ERROR) << "close";
-  }
-  if (wakeup_pipe_out_ >= 0) {
-    if (HANDLE_EINTR(close(wakeup_pipe_out_)) < 0)
-      DPLOG(ERROR) << "close";
-  }
-  event_base_free(event_base_);
-}
-
-bool MessagePumpLibevent::WatchFileDescriptor(int fd,
-                                              bool persistent,
-                                              int mode,
-                                              FileDescriptorWatcher *controller,
-                                              Watcher *delegate) {
-  DCHECK_GE(fd, 0);
-  DCHECK(controller);
-  DCHECK(delegate);
-  DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE);
-  // WatchFileDescriptor should be called on the pump thread. It is not
-  // threadsafe, and your watcher may never be registered.
-  DCHECK(watch_file_descriptor_caller_checker_.CalledOnValidThread());
-
-  int event_mask = persistent ? EV_PERSIST : 0;
-  if (mode & WATCH_READ) {
-    event_mask |= EV_READ;
-  }
-  if (mode & WATCH_WRITE) {
-    event_mask |= EV_WRITE;
-  }
-
-  scoped_ptr<event> evt(controller->ReleaseEvent());
-  if (evt.get() == NULL) {
-    // Ownership is transferred to the controller.
-    evt.reset(new event);
-  } else {
-    // Make sure we don't pick up any funky internal libevent masks.
-    int old_interest_mask = evt.get()->ev_events &
-        (EV_READ | EV_WRITE | EV_PERSIST);
-
-    // Combine old/new event masks.
-    event_mask |= old_interest_mask;
-
-    // Must disarm the event before we can reuse it.
-    event_del(evt.get());
-
-    // It's illegal to use this function to listen on 2 separate fds with the
-    // same |controller|.
-    if (EVENT_FD(evt.get()) != fd) {
-      NOTREACHED() << "FDs don't match" << EVENT_FD(evt.get()) << "!=" << fd;
-      return false;
-    }
-  }
-
-  // Set current interest mask and message pump for this event.
-  event_set(evt.get(), fd, event_mask, OnLibeventNotification, controller);
-
-  // Tell libevent which message pump this socket will belong to when we add it.
-  if (event_base_set(event_base_, evt.get())) {
-    return false;
-  }
-
-  // Add this socket to the list of monitored sockets.
-  if (event_add(evt.get(), NULL)) {
-    return false;
-  }
-
-  // Transfer ownership of evt to controller.
-  controller->Init(evt.release());
-
-  controller->set_watcher(delegate);
-  controller->set_pump(this);
-
-  return true;
-}
-
-void MessagePumpLibevent::AddIOObserver(IOObserver *obs) {
-  io_observers_.AddObserver(obs);
-}
-
-void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) {
-  io_observers_.RemoveObserver(obs);
-}
-
-// Tell libevent to break out of inner loop.
-static void timer_callback(int fd, short events, void *context)
-{
-  event_base_loopbreak((struct event_base *)context);
-}
-
-// Reentrant!
-void MessagePumpLibevent::Run(Delegate* delegate) {
-  DCHECK(keep_running_) << "Quit must have been called outside of Run!";
-  base::AutoReset<bool> auto_reset_in_run(&in_run_, true);
-
-  // event_base_loopexit() + EVLOOP_ONCE is leaky, see http://crbug.com/25641.
-  // Instead, make our own timer and reuse it on each call to event_base_loop().
-  scoped_ptr<event> timer_event(new event);
-
-  for (;;) {
-#if defined(OS_MACOSX)
-    mac::ScopedNSAutoreleasePool autorelease_pool;
-#endif
-
-    bool did_work = delegate->DoWork();
-    if (!keep_running_)
-      break;
-
-    event_base_loop(event_base_, EVLOOP_NONBLOCK);
-    did_work |= processed_io_events_;
-    processed_io_events_ = false;
-    if (!keep_running_)
-      break;
-
-    did_work |= delegate->DoDelayedWork(&delayed_work_time_);
-    if (!keep_running_)
-      break;
-
-    if (did_work)
-      continue;
-
-    did_work = delegate->DoIdleWork();
-    if (!keep_running_)
-      break;
-
-    if (did_work)
-      continue;
-
-    // EVLOOP_ONCE tells libevent to only block once,
-    // but to service all pending events when it wakes up.
-    if (delayed_work_time_.is_null()) {
-      event_base_loop(event_base_, EVLOOP_ONCE);
-    } else {
-      TimeDelta delay = delayed_work_time_ - TimeTicks::Now();
-      if (delay > TimeDelta()) {
-        struct timeval poll_tv;
-        poll_tv.tv_sec = delay.InSeconds();
-        poll_tv.tv_usec = delay.InMicroseconds() % Time::kMicrosecondsPerSecond;
-        event_set(timer_event.get(), -1, 0, timer_callback, event_base_);
-        event_base_set(event_base_, timer_event.get());
-        event_add(timer_event.get(), &poll_tv);
-        event_base_loop(event_base_, EVLOOP_ONCE);
-        event_del(timer_event.get());
-      } else {
-        // It looks like delayed_work_time_ indicates a time in the past, so we
-        // need to call DoDelayedWork now.
-        delayed_work_time_ = TimeTicks();
-      }
-    }
-  }
-
-  keep_running_ = true;
-}
-
-void MessagePumpLibevent::Quit() {
-  DCHECK(in_run_);
-  // Tell both libevent and Run that they should break out of their loops.
-  keep_running_ = false;
-  ScheduleWork();
-}
-
-void MessagePumpLibevent::ScheduleWork() {
-  // Tell libevent (in a threadsafe way) that it should break out of its loop.
-  char buf = 0;
-  int nwrite = HANDLE_EINTR(write(wakeup_pipe_in_, &buf, 1));
-  DCHECK(nwrite == 1 || errno == EAGAIN)
-      << "[nwrite:" << nwrite << "] [errno:" << errno << "]";
-}
-
-void MessagePumpLibevent::ScheduleDelayedWork(
-    const TimeTicks& delayed_work_time) {
-  // We know that we can't be blocked on Wait right now since this method can
-  // only be called on the same thread as Run, so we only need to update our
-  // record of how long to sleep when we do sleep.
-  delayed_work_time_ = delayed_work_time;
-}
-
-void MessagePumpLibevent::WillProcessIOEvent() {
-  FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent());
-}
-
-void MessagePumpLibevent::DidProcessIOEvent() {
-  FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent());
-}
-
-bool MessagePumpLibevent::Init() {
-  int fds[2];
-  if (pipe(fds)) {
-    DLOG(ERROR) << "pipe() failed, errno: " << errno;
-    return false;
-  }
-  if (SetNonBlocking(fds[0])) {
-    DLOG(ERROR) << "SetNonBlocking for pipe fd[0] failed, errno: " << errno;
-    return false;
-  }
-  if (SetNonBlocking(fds[1])) {
-    DLOG(ERROR) << "SetNonBlocking for pipe fd[1] failed, errno: " << errno;
-    return false;
-  }
-  wakeup_pipe_out_ = fds[0];
-  wakeup_pipe_in_ = fds[1];
-
-  wakeup_event_ = new event;
-  event_set(wakeup_event_, wakeup_pipe_out_, EV_READ | EV_PERSIST,
-            OnWakeup, this);
-  event_base_set(event_base_, wakeup_event_);
-
-  if (event_add(wakeup_event_, 0))
-    return false;
-  return true;
-}
-
-// static
-void MessagePumpLibevent::OnLibeventNotification(int fd, short flags,
-                                                 void* context) {
-  base::WeakPtr<FileDescriptorWatcher> controller =
-      static_cast<FileDescriptorWatcher*>(context)->weak_factory_.GetWeakPtr();
-  DCHECK(controller.get());
-
-  MessagePumpLibevent* pump = controller->pump();
-  pump->processed_io_events_ = true;
-
-  if (flags & EV_WRITE) {
-    controller->OnFileCanWriteWithoutBlocking(fd, pump);
-  }
-  // Check |controller| in case it's been deleted in
-  // controller->OnFileCanWriteWithoutBlocking().
-  if (controller.get() && flags & EV_READ) {
-    controller->OnFileCanReadWithoutBlocking(fd, pump);
-  }
-}
-
-// Called if a byte is received on the wakeup pipe.
-// static
-void MessagePumpLibevent::OnWakeup(int socket, short flags, void* context) {
-  base::MessagePumpLibevent* that =
-              static_cast<base::MessagePumpLibevent*>(context);
-  DCHECK(that->wakeup_pipe_out_ == socket);
-
-  // Remove and discard the wakeup byte.
-  char buf;
-  int nread = HANDLE_EINTR(read(socket, &buf, 1));
-  DCHECK_EQ(nread, 1);
-  that->processed_io_events_ = true;
-  // Tell libevent to break out of inner loop.
-  event_base_loopbreak(that->event_base_);
-}
-
-}  // namespace base