Provide an iOS message pump for IO implementation.

base/message_pump_io_ios.cc

-// 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();
-  }
-}
-
-bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() {
-  event* e = ReleaseEvent();
-  if (e == NULL)
+      watcher_(NULL) {
+}
+
+MessagePumpIOSForIO::FileDescriptorWatcher::~FileDescriptorWatcher() {
+  StopWatchingFileDescriptor();
+}
+
+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);
-}
-
-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;
+  return true;
+}
+
+void MessagePumpIOSForIO::FileDescriptorWatcher::Init(
+    CFFileDescriptorRef fdref,
+    CFOptionFlags callback_types,
+    CFRunLoopSourceRef fd_source,
+    bool is_persistent) {
+  DCHECK(fdref);
+  DCHECK(!fdref_);
+
+  is_persistent_ = is_persistent;
+  fdref_.reset(fdref);
+  callback_types_ = callback_types;
+  fd_source_.reset(fd_source);
+}
+
+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();
-}
-
-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,
-                                              Mode mode,
-                                              FileDescriptorWatcher *controller,
-                                              Watcher *delegate) {
+MessagePumpIOSForIO::MessagePumpIOSForIO() {
+}
+
+MessagePumpIOSForIO::~MessagePumpIOSForIO() {
+}
+
+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);
+
   // 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;
-  }
-  if ((mode & WATCH_WRITE) != 0) {
-    event_mask |= EV_WRITE;
-  }
-
-  scoped_ptr<event> evt(controller->ReleaseEvent());
-  if (evt.get() == NULL) {
-    // Ownership is transferred to the controller.
-    evt.reset(new event);
+  CFFileDescriptorContext source_context = {0};
+  source_context.info = controller;
+
+  CFOptionFlags callback_types = 0;
+  if (mode & WATCH_READ) {
+    callback_types |= kCFFileDescriptorReadCallBack;
+  }
+  if (mode & WATCH_WRITE) {
+    callback_types |= kCFFileDescriptorWriteCallBack;
+  }
+
+  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;
+    }
+
+    CFFileDescriptorEnableCallBacks(scoped_fdref, callback_types);
+
+    // 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 {
-    // 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()) != 0) {
-    return false;
-  }
-
-  // Add this socket to the list of monitored sockets.
-  if (event_add(evt.get(), NULL) != 0) {
-    return false;
-  }
-
-  // Transfer ownership of evt to controller.
-  controller->Init(evt.release());
+    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;
+    }
+
+    // Combine old/new event masks.
+    CFFileDescriptorDisableCallBacks(fdref, controller->callback_types_);
+    controller->callback_types_ |= callback_types;
+    CFFileDescriptorEnableCallBacks(fdref, controller->callback_types_);
+  }
 
   controller->set_watcher(delegate);
   controller->set_pump(this);
 
   return true;
 }
 
-void MessagePumpLibevent::AddIOObserver(IOObserver *obs) {
+void MessagePumpIOSForIO::RemoveRunLoopSource(CFRunLoopSourceRef source) {
+  CFRunLoopRemoveSource(run_loop(), source, kCFRunLoopCommonModes);
+}
+
+void MessagePumpIOSForIO::AddIOObserver(IOObserver *obs) {
   io_observers_.AddObserver(obs);
 }
 
-void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) {
+void MessagePumpIOSForIO::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() {
+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) {
-  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) {
+void MessagePumpIOSForIO::HandleFdIOEvent(CFFileDescriptorRef fdref,
+                                          CFOptionFlags callback_types,
+                                          void* context) {
+  // Ensure that |fdref| will remain live for the duration of this function

[wtc, 2012/11/29 20:30:34]

Can you get the retain count of |fdref| on entry to this
function?

If the retain count is only 1, then |controller| is the sole
owner of |fdref|, and it is necessary for this function to
retain |fdref| to ensure it remains live.

If the retain count is > 1, then the OS must have retained
|fdref| as we suspected, and our use of scoped_fdref is
defensive programming.

[blundell, 2012/11/30 11:19:52]

The retain count is 2, so in fact the system is retaining |fdref| in practice.
Nonetheless, I would like to avoid relying on this undocumented behavior.

On 2012/11/29 20:30:34, wtc wrote:

[wtc, 2012/11/30 15:14:20]

In https://chromiumcodereview.appspot.com/11412101/#msg16,
you reported that "the |fdref| passed in to
|MessagePumpIOSForIO::HandleFdIOEvent| is in fact owned by
the |controller| and thus goes away if the controller goes
away". If the retain count is 2, |fdref| should not go away
when the controller goes away. Can you find out exactly what
happened when you ran the DeleteWatcher test? Perhaps the
test crashed when calling controller->OnFileCanReadWithoutBlocking
rather than when reading controller->is_persistent?

As an experiment, if you remove the scoped_fdref variable,
do the tests still pass?
Agreed. I just wanted to fully understand this code. Thanks.

+  // call even if |controller| is deleted or |StopWatchingFileDescriptor()| is
+  // called, either of which will cause |fdref| to be released.
+  mac::ScopedCFTypeRef<CFFileDescriptorRef> scoped_fdref(
+      fdref, base::scoped_policy::RETAIN);
+  int fd = CFFileDescriptorGetNativeDescriptor(scoped_fdref);
+
+  FileDescriptorWatcher* controller =
+      static_cast<FileDescriptorWatcher*>(context);
+  DCHECK_EQ(scoped_fdref, controller->fdref_);
+
+  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 only if the file descriptor has not been
+  // invalidated in the write callback. As |FileDescriptorWatcher| invalidates
+  // its file descriptor on destruction, the file descriptor being valid also
+  // guarantees that |controller| has not been deleted.
+  if (callback_types & kCFFileDescriptorReadCallBack &&
+      CFFileDescriptorIsValid(scoped_fdref)) {
+    DCHECK_EQ(scoped_fdref, controller->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 the file descriptor is still
+  // valid and the controller is persistent.
+  if (CFFileDescriptorIsValid(scoped_fdref) && controller->is_persistent_) {
+    DCHECK_EQ(scoped_fdref, controller->fdref_);
+    CFFileDescriptorEnableCallBacks(fdref, callback_types);
+  }
 }
 
 }  // namespace base