Provide an iOS message pump for IO implementation.

base/message_pump_io_ios.cc

Index: base/message_pump_io_ios.cc
diff --git a/base/message_pump_libevent.cc b/base/message_pump_io_ios.cc
similarity index 14%
copy from base/message_pump_libevent.cc
copy to base/message_pump_io_ios.cc
index 66acd86e6e0abef5559600f615d7c0c5b5217adf..61f366240adf8d2923c51359fb689b8fb8e356bb 100644
--- a/base/message_pump_libevent.cc
+++ b/base/message_pump_io_ios.cc
@@ -1,383 +1,203 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Copyright 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"
-#if defined(USE_SYSTEM_LIBEVENT)
-#include <event.h>
-#else
-#include "third_party/libevent/event.h"
-#endif
-
-#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.
+#include "base/message_pump_io_ios.h"
 
 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),
+MessagePumpIOSForIO::FileDescriptorWatcher::FileDescriptorWatcher()
+    : is_persistent_(false),
+      fdref_(NULL),
+      callback_types_(0),
+      fd_source_(NULL),
       pump_(NULL),
       watcher_(NULL),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
 }
 
-MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() {
-  if (event_) {
-    StopWatchingFileDescriptor();
-  }
+MessagePumpIOSForIO::FileDescriptorWatcher::~FileDescriptorWatcher() {
+  StopWatchingFileDescriptor();
 }
 
-bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() {
-  event* e = ReleaseEvent();
-  if (e == NULL)
+bool MessagePumpIOSForIO::FileDescriptorWatcher::StopWatchingFileDescriptor() {
+  if (fdref_ == NULL)
     return true;
 
-  // event_del() is a no-op if the event isn't active.
-  int rv = event_del(e);
-  delete e;
+  CFFileDescriptorDisableCallBacks(fdref_, callback_types_);
+  pump_->RemoveRunLoopSource(fd_source_);
+  fd_source_.reset();
+  fdref_.reset();
+  callback_types_ = 0;
   pump_ = NULL;
   watcher_ = NULL;
-  return (rv == 0);
+  return true;
 }
 
-void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e) {
-  DCHECK(e);
-  DCHECK(!event_);
+void MessagePumpIOSForIO::FileDescriptorWatcher::Init(
+    CFFileDescriptorRef fdref,
+    CFOptionFlags callback_types,
+    CFRunLoopSourceRef fd_source,
+    bool is_persistent) {
+  DCHECK(fdref);
+  DCHECK(!fdref_);
 
-  event_ = e;
+  is_persistent_ = is_persistent;
+  fdref_.reset(fdref);
+  callback_types_ = callback_types;
+  fd_source_.reset(fd_source);
 }
 
-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;
+void MessagePumpIOSForIO::FileDescriptorWatcher::OnFileCanReadWithoutBlocking(
+    int fd,
+    MessagePumpIOSForIO* pump) {
+  DCHECK(callback_types_ & kCFFileDescriptorReadCallBack);
   pump->WillProcessIOEvent();
   watcher_->OnFileCanReadWithoutBlocking(fd);
   pump->DidProcessIOEvent();
 }
 
-void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking(
-    int fd, MessagePumpLibevent* pump) {
-  DCHECK(watcher_);
+void MessagePumpIOSForIO::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking(
+    int fd,
+    MessagePumpIOSForIO* pump) {
+  DCHECK(callback_types_ & kCFFileDescriptorWriteCallBack);
   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();
+MessagePumpIOSForIO::MessagePumpIOSForIO() {
 }
 
-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_);
+MessagePumpIOSForIO::~MessagePumpIOSForIO() {
 }
 
-bool MessagePumpLibevent::WatchFileDescriptor(int fd,
-                                              bool persistent,
-                                              Mode mode,
-                                              FileDescriptorWatcher *controller,
-                                              Watcher *delegate) {
+bool MessagePumpIOSForIO::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);
+  DCHECK(mode == WATCH_READ || mode == WATCH_WRITE ||
+         mode == (WATCH_READ | WATCH_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) != 0) {
-    event_mask |= EV_READ;
+  CFFileDescriptorContext source_context = {0};
+  source_context.info = controller;
+
+  CFOptionFlags callback_types = 0;
+  if (mode & WATCH_READ) {
+    callback_types |= kCFFileDescriptorReadCallBack;
   }
-  if ((mode & WATCH_WRITE) != 0) {
-    event_mask |= EV_WRITE;
+  if (mode & WATCH_WRITE) {
+    callback_types |= kCFFileDescriptorWriteCallBack;
   }
 
-  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);
+  CFFileDescriptorRef fdref = controller->fdref_;
+  if (fdref == NULL) {
+    base::mac::ScopedCFTypeRef<CFFileDescriptorRef> scoped_fdref(
+        CFFileDescriptorCreate(kCFAllocatorDefault, fd, false, HandleFdIOEvent,
+                               &source_context));
+    if (scoped_fdref == NULL) {
+      NOTREACHED() << "CFFileDescriptorCreate failed";
+      return false;
+    }
 
-    // Combine old/new event masks.
-    event_mask |= old_interest_mask;
+    CFFileDescriptorEnableCallBacks(scoped_fdref, callback_types);
 
-    // Must disarm the event before we can reuse it.
-    event_del(evt.get());
+    // TODO(wtc): what should the 'order' argument be?
+    base::mac::ScopedCFTypeRef<CFRunLoopSourceRef> scoped_fd_source(
+        CFFileDescriptorCreateRunLoopSource(kCFAllocatorDefault,
+                                            scoped_fdref,
+                                            0));
+    if (scoped_fd_source == NULL) {
+      NOTREACHED() << "CFFileDescriptorCreateRunLoopSource failed";
+      return false;
+    }
+    CFRunLoopAddSource(run_loop(), scoped_fd_source, kCFRunLoopCommonModes);
 
+    // Transfer ownership of scoped_fdref and fd_source to controller.
+    controller->Init(scoped_fdref.release(), callback_types,
+                     scoped_fd_source.release(), persistent);
+  } else {
     // 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;
+    if (CFFileDescriptorGetNativeDescriptor(fdref) != fd) {
+      NOTREACHED() << "FDs don't match: "
+                   << CFFileDescriptorGetNativeDescriptor(fdref)
+                   << " != " << fd;
+      return false;
+    }
+    if (persistent != controller->is_persistent_) {
+      NOTREACHED() << "persistent doesn't match";
       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()) != 0) {
-    return false;
-  }
 
-  // Add this socket to the list of monitored sockets.
-  if (event_add(evt.get(), NULL) != 0) {
-    return false;
+    // Combine old/new event masks.
+    CFFileDescriptorDisableCallBacks(fdref, controller->callback_types_);
+    controller->callback_types_ |= callback_types;
+    CFFileDescriptorEnableCallBacks(fdref, controller->callback_types_);
   }
 
-  // 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);
+void MessagePumpIOSForIO::RemoveRunLoopSource(CFRunLoopSourceRef source) {
+  CFRunLoopRemoveSource(run_loop(), source, kCFRunLoopCommonModes);
 }
 
-// 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 MessagePumpIOSForIO::AddIOObserver(IOObserver *obs) {
+  io_observers_.AddObserver(obs);
 }
 
-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 MessagePumpIOSForIO::RemoveIOObserver(IOObserver *obs) {
+  io_observers_.RemoveObserver(obs);
 }
 
-void MessagePumpLibevent::WillProcessIOEvent() {
+void MessagePumpIOSForIO::WillProcessIOEvent() {
   FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent());
 }
 
-void MessagePumpLibevent::DidProcessIOEvent() {
+void MessagePumpIOSForIO::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) {
+void MessagePumpIOSForIO::HandleFdIOEvent(CFFileDescriptorRef fdref,
+                                          CFOptionFlags callback_types,
+                                          void* context) {
+  int fd = CFFileDescriptorGetNativeDescriptor(fdref);

[blundell, 2012/11/27 10:58:43]

The changes to message_pump_io_ios.{mm, cc} from patch 6 to patch 7 are to this
function:
- Maintaining |controller| as a weak pointer
- Guarding the read callback against the controller having been deleted/the fd
invalidated
- Explicitly checking for deletion of the controller before re-enabling the
callback
- Changing comments accordingly with the above changes

+
   base::WeakPtr<FileDescriptorWatcher> controller =
       static_cast<FileDescriptorWatcher*>(context)->weak_factory_.GetWeakPtr();
-  DCHECK(controller.get());
+  DCHECK_EQ(fdref, controller->fdref_);
 
-  MessagePumpLibevent* pump = controller->pump();
-  pump->processed_io_events_ = true;
-
-  if (flags & EV_WRITE) {
+  MessagePumpIOSForIO* pump = controller->pump();
+  if (callback_types & kCFFileDescriptorWriteCallBack) {
     controller->OnFileCanWriteWithoutBlocking(fd, pump);
   }
-  // Check |controller| in case it's been deleted in
-  // controller->OnFileCanWriteWithoutBlocking().
-  if (controller.get() && flags & EV_READ) {
+  // Perform the read callback if the controller has not been deleted and/or
+  // the file descriptor invalidated in the write callback.
+  if (callback_types & kCFFileDescriptorReadCallBack &&
+      controller.get() && CFFileDescriptorIsValid(fdref)) {
     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_);
+  // Re-enable callbacks after the read/write if (1) the controller has not
+  // been deleted, (2) the file descriptor has not been invalidated, and (3)
+  // the file descriptor watcher is persistent.
+  if (controller.get() && controller->is_persistent_ &&
+      CFFileDescriptorIsValid(fdref))

[wtc, 2012/11/26 22:17:42]

Can you print the retain count of fdref at strategic points,
in particular before and after the CFFileDescriptorEnableCallBacks
and CFFileDescriptorDisableCallBacks calls? If those calls
do not change the retain count of fdref, then I think this
CFFileDescriptorIsValid(fdref) call here is wrong and should
be removed.

Note: I remember running into a destroyed 'controller' at
this point. I was running this code on Mac OS X though
(I manually modified Chrome to use this message pump on
Mac OS X). I don't know what has changed since then.

[blundell, 2012/11/27 10:58:43]

|CFFileDescriptorIsValid()| checks whether |CFFileDescriptorInvalidate()| has
been called (e.g., via a call to
|FileDescriptorWatcher::StopWatchingFileDescriptor()|. If |fdref| has been
invalidated, then I would think that we shouldn't re-enable the callbacks. Note
that if we omit the check for validity before performing the read callback, the
|StopWatcher| unittest fails because |StopWatchingFileDescriptor()| resets
internal state and hence the DCHECK that we are listening for read callbacks
goes off. Does that make sense?

However, your comment made me realize that there's still a problem with this
code: if |StopWatchingFileDescriptor()| is called, |controller->fdref_| will be
reset, and thus we shouldn't access |fdref| anymore either. In the newest patch,
changed the code to check |controller->fdref_|.

On 2012/11/26 22:17:42, wtc wrote:

[wtc, 2012/11/28 00:25:25]

Yes, that makes sense. (My brain is not in its optimal state
right now, so I am not confident that I fully understand
this code. But your explanation makes sense.)

+    CFFileDescriptorEnableCallBacks(fdref, callback_types);
 }
 
 }  // namespace base