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)
+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) {
+void MessagePumpIOSForIO::HandleFdIOEvent(CFFileDescriptorRef fdref,
+                                          CFOptionFlags callback_types,
+                                          void* context) {
+  int fd = CFFileDescriptorGetNativeDescriptor(fdref);
+
   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) {
+  DCHECK_EQ(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 (1) the controller has not been deleted
+  // within the write callback and (2) |StopWatchingFileDescriptor()| has not
+  // been called within the write callback. The latter call having been made is
+  // indicated by |controller->fdref_| no longer being equal to |fdref| (NOTE:
+  // It's not enough to check for whether |controller->fdref_| is not NULL, as
+  // another call to |FileDescriptorWatcher::Init()| may have been made from

[wtc, 2012/11/28 23:59:47]

Nit: this comment should mention the FileDescriptorWatcher::Init() call
uses the same controller to watch a different fd.

[blundell, 2012/11/29 15:01:04]

Changed the comment entirely (see below discussion).

On 2012/11/28 23:59:47, wtc wrote:

+  // within the callback).
+  if (callback_types & kCFFileDescriptorReadCallBack &&
+      controller.get() && controller->fdref_ == fdref) {

[blundell, 2012/11/28 16:56:55]

Made this change because I saw while testing downstream that it was possible for
|StopWatchingFileDescriptor| and then |Init| to be successively called from
within a callback, so that |controller->fdref_| wouldn't be NULL but nonetheless
|fdref| would have been invalidated.

[wtc, 2012/11/28 23:59:47]

I confirm this can happen during TCPClientSocket::Connect.
Is that where you observed this behavior?

If a server has multiple IP addresses, TCPClientSocket::Connect
will try connecting to the IP addresses of the server one
by one. If the connection to the first IP address failed,
TCPClientSocket will call StopWatchingFileDescriptor on
the first (Unix) socket descriptor and open a new (Unix)
socket to connect to the next IP address, while reusing
the same |controller|.

Hmm... Perhaps this is why I used to check if fdref is still
valid at the end of this function. From patch set 1 of this
CL:

  CHECK_GT(CFGetRetainCount(fdref), 0);
  if (CFFileDescriptorIsValid(fdref) && persistent)
    CFFileDescriptorEnableCallBacks(fdref, callback_types);

I am worried that the controller->fdref_ == fdref test is
still not robust in theory because the OS could reuse the
fdref value. I think we should at least bring back the
assertion
  CHECK_GT(CFGetRetainCount(fdref), 0);
to ensure that fdref (the pointer value) could not have
been freed and immediately reused. Perhaps this function
should acquire a reference to fdref to ensure that it is
not freed during this function, if we are not sure whether
that is guaranteed by the OS.

[blundell, 2012/11/29 15:01:04]

After your comments and Mark's comments about the growing spaghetti nature of
the code, I went back to the old approach of checking
|CFFileDescriptorIsValid(fdref)| directly together with an added retain of
|fdref| to make sure it doesn't go away from under us. I searched through
documentation on |CFFileDescriptorRef| and could not find anything indicating
whether the system guarantees that it retains |fdref| before passing it in to
this callback, so I think that it is best to be defensive here.

On 2012/11/28 23:59:47, wtc wrote:

+    DCHECK(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) |StopWatchingFileDescriptor()| has not been called, and
+  // (3) the file descriptor watcher is persistent.
+  if (controller.get() && controller->fdref_ == fdref &&
+      controller->is_persistent_) {
+    DCHECK(CFFileDescriptorIsValid(fdref));
+    CFFileDescriptorEnableCallBacks(fdref, callback_types);
+  }
 }
 
 }  // namespace base