Mojo: Armed Watchers

mojo/edk/system/watcher_dispatcher.cc

Index: mojo/edk/system/watcher_dispatcher.cc
diff --git a/mojo/edk/system/watcher_dispatcher.cc b/mojo/edk/system/watcher_dispatcher.cc
new file mode 100644
index 0000000000000000000000000000000000000000..93ee18b073e80e4af52d77ac557a363563ad4292
--- /dev/null
+++ b/mojo/edk/system/watcher_dispatcher.cc
@@ -0,0 +1,374 @@
+// Copyright 2017 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 "mojo/edk/system/watcher_dispatcher.h"
+
+#include <limits>
+#include <map>
+
+#include "base/macros.h"
+#include "base/memory/ptr_util.h"
+#include "mojo/edk/system/watch.h"
+
+namespace mojo {
+namespace edk {
+
+// Helper to track contexts relevant to a specific handle. This is used to avoid
+// redundant lookups when notifying interested watches about a handle change.
+//
+// Not thread-safe: Should only be accessed while the dispatcher holds its lock,
+// with the exception that during watch removal these may be moved out of the
+// WatcherDispatcher and safely used without locking.
+class WatcherDispatcher::WatchedHandle {
+ public:
+  struct WatchEntry {
+    explicit WatchEntry(const scoped_refptr<Watch>& watch) : watch(watch) {}
+    ~WatchEntry() {}
+
+    // Indicates whether the last computed MojoResult (combinatation of
+    // watched signals and current signal state) indicates that a notification
+    // should happen when our watcher is armed.
+    //
+    // |MOJO_RESULT_CANCELLED| is not included in the set of "ready" results
+    // because cancellation is a one-time event that always fires immediately.
+    bool ready() const {
+      return last_known_result != MOJO_RESULT_SHOULD_WAIT &&
+             last_known_result != MOJO_RESULT_CANCELLED;
+    }
+
+    scoped_refptr<Watch> watch;
+    MojoResult last_known_result = MOJO_RESULT_UNKNOWN;
+  };
+
+  using WatchMap = std::map<uintptr_t, std::unique_ptr<WatchEntry>>;
+
+  explicit WatchedHandle(const scoped_refptr<Dispatcher>& dispatcher)
+      : dispatcher_(dispatcher) {}
+  ~WatchedHandle() {}
+
+  void AddWatch(uintptr_t context, const scoped_refptr<Watch>& watch) {
+    auto result = watches_.insert(
+        std::make_pair(context, base::MakeUnique<WatchEntry>(watch)));
+    DCHECK(result.second);
+  }
+
+  std::unique_ptr<WatchEntry> RemoveWatch(uintptr_t context) {
+    auto it = watches_.find(context);
+    DCHECK(it != watches_.end());
+
+    std::unique_ptr<WatchEntry> entry = std::move(it->second);
+    watches_.erase(it);
+    return entry;
+  }
+
+  bool NotifyState(const HandleSignalsState& state,
+                   bool allowed_to_call_callback) {
+    last_known_signals_state_ = state;
+    bool handle_ready = false;
+    for (auto& entry : watches_) {
+      MojoResult rv =
+          entry.second->watch->NotifyState(state, allowed_to_call_callback);
+      if (rv != MOJO_RESULT_SHOULD_WAIT && rv != MOJO_RESULT_CANCELLED)
+        handle_ready |= true;
+      entry.second->last_known_result = rv;
+    }
+    return handle_ready;
+  }
+
+  void CancelAllWatches() {
+    for (auto& entry : watches_)
+      entry.second->watch->Cancel();
+  }
+
+  // Determines if any watch on this handle is ready.
+  bool IsAnyWatchReady() const {
+    for (auto& entry : watches_)
+      if (entry.second->ready())
+        return true;
+    return false;
+  }
+
+  const WatchMap& watches() const { return watches_; }
+
+  // Stores at most |max_contexts| elements at the output locations, regarding
+  // ready contexts associated with this handle. Returns the number of elements
+  // actually stored.
+  //
+  // This is used to implement WatcherDispatcher::Arm()'s already-satisfied/
+  // already-unsatisfiable failure mode by filling in a user-provided buffer
+  // with as much information as we can fit.
+  uint32_t StoreReadyContextState(uint32_t max_contexts,
+                                  uintptr_t* out_contexts,
+                                  MojoResult* out_results,
+                                  MojoHandleSignalsState* out_states) {
+    uint32_t num_contexts = 0;
+    for (auto& entry : watches_) {
+      if (!max_contexts)
+        return num_contexts;
+
+      if (entry.second->last_known_result == MOJO_RESULT_OK ||
+          entry.second->last_known_result == MOJO_RESULT_FAILED_PRECONDITION) {
+        out_contexts[num_contexts] = entry.first;
+        out_results[num_contexts] = entry.second->last_known_result;
+        out_states[num_contexts] =
+            static_cast<MojoHandleSignalsState>(last_known_signals_state_);
+        num_contexts++;
+        max_contexts--;
+      }
+    }
+
+    return num_contexts;
+  }
+
+ private:
+  // The dispatcher to which all these Watch instances belong.
+  const scoped_refptr<Dispatcher> dispatcher_;
+
+  // A ref to every Watch actively watching |dispatcher_| on behalf of our
+  // owning watcher.
+  WatchMap watches_;
+
+  // The last known state of |dispatcher_|. This is tracked for quick access in
+  // StoreReadyContextState().
+  HandleSignalsState last_known_signals_state_ = {0, 0};
+
+  DISALLOW_COPY_AND_ASSIGN(WatchedHandle);
+};
+
+WatcherDispatcher::WatcherDispatcher(MojoWatcherCallback callback)
+    : callback_(callback) {}
+
+void WatcherDispatcher::NotifyHandleState(Dispatcher* dispatcher,
+                                          const HandleSignalsState& state) {
+  base::AutoLock lock(lock_);
+  auto it = watched_handles_.find(dispatcher);
+  if (it == watched_handles_.end())
+    return;
+
+  // Maybe fire a notification to one or more watches assoicated with the
+  // dispatcher, provided we're armed and at least one of the watches cares
+  // about the new state.
+  if (it->second->NotifyState(state, armed_)) {
+    ready_handles_.insert(it->second.get());
+
+    // If we were armed, we also notified watchers. Ensure we're disarmed now.
+    armed_ = false;
+  } else {
+    ready_handles_.erase(it->second.get());
+  }
+}
+
+void WatcherDispatcher::NotifyHandleClosed(Dispatcher* dispatcher) {
+  std::unique_ptr<WatchedHandle> handle;
+  {
+    base::AutoLock lock(lock_);
+    auto it = watched_handles_.find(dispatcher);
+    if (it == watched_handles_.end())
+      return;
+
+    handle = std::move(it->second);
+
+    // Wipe out all watches associated with the closed dispatcher.
+    for (auto& entry : handle->watches())
+      watches_.erase(entry.first);
+    ready_handles_.erase(handle.get());
+    watched_handles_.erase(it);
+  }
+
+  // NOTE: It's important that this is called outside of |lock_| since it
+  // acquires internal Watch locks.
+  handle->CancelAllWatches();
+}
+
+void WatcherDispatcher::InvokeWatchCallback(
+    uintptr_t context,
+    MojoResult result,
+    const HandleSignalsState& state,
+    MojoWatcherNotificationFlags flags) {
+  {
+    // We avoid holding the lock during dispatch. It's OK for notification
+    // callbacks to close this watcher, and it's OK for notifications to race
+    // with closure, if for example the watcher is closed from another thread
+    // between this test and the invocation of |callback_| below.
+    //
+    // Because cancellation synchronously blocks all future notifications, and
+    // because notifications themselves are mutually exclusive for any given
+    // context, we still guarantee that a single MOJO_RESULT_CANCELLED result
+    // is the last notification received for any given context.
+    //
+    // This guarantee is sufficient to make safe, synchronized, per-context
+    // state management possible in user code.
+    base::AutoLock lock(lock_);
+    if (closed_ && result != MOJO_RESULT_CANCELLED)
+      return;
+  }
+
+  callback_(context, result, static_cast<MojoHandleSignalsState>(state), flags);
+}
+
+Dispatcher::Type WatcherDispatcher::GetType() const {
+  return Type::WATCHER;
+}
+
+MojoResult WatcherDispatcher::Close() {
+  // We swap out all the watched handle information onto the stack so we can
+  // call into their dispatchers without our own lock held.
+  std::map<uintptr_t, scoped_refptr<Watch>> watches;
+  {
+    base::AutoLock lock(lock_);
+    DCHECK(!closed_);
+    closed_ = true;
+    std::swap(watches, watches_);
+    watched_handles_.clear();
+  }
+
+  // Remove all refs from our watched dispatchers and fire cancellations.
+  for (auto& entry : watches) {
+    entry.second->dispatcher()->RemoveWatcherRef(this, entry.first);
+    entry.second->Cancel();
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+MojoResult WatcherDispatcher::WatchDispatcher(
+    scoped_refptr<Dispatcher> dispatcher,
+    MojoHandleSignals signals,
+    uintptr_t context) {
+  // NOTE: Because it's critical to avoid acquiring any other dispatcher locks
+  // while |lock_| is held, we defer adding oursevles to the dispatcher until
+  // after we've updated all our own relevant state and released |lock_|.
+  {
+    base::AutoLock lock(lock_);
+    if (watches_.find(context) != watches_.end())
+      return MOJO_RESULT_ALREADY_EXISTS;
+
+    auto it = watched_handles_.find(dispatcher.get());
+    if (it == watched_handles_.end()) {
+      auto result = watched_handles_.insert(std::make_pair(
+          dispatcher.get(), base::MakeUnique<WatchedHandle>(dispatcher)));
+      DCHECK(result.second);
+      it = result.first;
+    }
+
+    scoped_refptr<Watch> watch = new Watch(this, dispatcher, context, signals);
+    watches_.insert({context, watch});
+    it->second->AddWatch(context, watch);
+  }
+
+  MojoResult rv = dispatcher->AddWatcherRef(this, context);
+  if (rv != MOJO_RESULT_OK) {
+    // Oops. This was not a valid handle to watch. Undo the above work and
+    // fail gracefully. Note that other watches may have been added for the
+    // handle since we last held the lock, so we can't necessarily just wipe
+    // out the |watched_handles_| entry.
+    base::AutoLock lock(lock_);
+    watches_.erase(context);
+
+    // Also note that it's possible for someone to have closed this
+    // WatcherDispatcher since we last held |lock_|, so the entry might already
+    // be gone.
+    auto it = watched_handles_.find(dispatcher.get());
+    if (it != watched_handles_.end()) {
+      it->second->RemoveWatch(context);
+      if (it->second->watches().empty())
+        watched_handles_.erase(dispatcher.get());
+    }
+
+    return rv;
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+MojoResult WatcherDispatcher::CancelWatch(uintptr_t context) {
+  // We may remove the last stored ref to the Watch below, so we retain
+  // a reference on the stack.
+  scoped_refptr<Watch> watch;
+  {
+    base::AutoLock lock(lock_);
+    auto it = watches_.find(context);
+    if (it == watches_.end())
+      return MOJO_RESULT_NOT_FOUND;
+    watch = it->second;
+    watches_.erase(it);
+  }
+
+  // Mark the watch as cancelled so no further notifications get through.
+  watch->Cancel();
+
+  // We remove the watcher ref for this context before updating any more
+  // internal watcher state, ensuring that we don't receiving further
+  // notifications for this context.
+  watch->dispatcher()->RemoveWatcherRef(this, context);
+
+  {
+    base::AutoLock lock(lock_);
+    auto handle_it = watched_handles_.find(watch->dispatcher().get());
+    DCHECK(handle_it != watched_handles_.end());
+
+    std::unique_ptr<WatchedHandle::WatchEntry> entry =
+        handle_it->second->RemoveWatch(context);
+
+    if (handle_it->second->watches().empty()) {
+      // This was the only remaining watch we had for its dispatcher, so we
+      // can wipe out all knowledge of that dispatcher.
+      ready_handles_.erase(handle_it->second.get());
+      watched_handles_.erase(handle_it);
+    } else if (entry->ready() && !handle_it->second->IsAnyWatchReady()) {
+      // There are other watches for this dispatcher, but this was the only one
+      // keeping it in a "ready" state. Remove the handle from the ready-set.
+      ready_handles_.erase(handle_it->second.get());
+    }
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+MojoResult WatcherDispatcher::Arm(
+    uint32_t* num_ready_contexts,
+    uintptr_t* ready_contexts,
+    MojoResult* ready_results,
+    MojoHandleSignalsState* ready_signals_states) {
+  base::AutoLock lock(lock_);
+  if (num_ready_contexts &&
+      (!ready_contexts || !ready_results || !ready_signals_states)) {
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  }
+
+  if (watched_handles_.empty())
+    return MOJO_RESULT_NOT_FOUND;
+
+  if (ready_handles_.empty()) {
+    // Fast path: No handles are ready to notify, so we're done.
+    armed_ = true;
+    return MOJO_RESULT_OK;
+  }
+
+  if (num_ready_contexts) {
+    uint32_t max_ready_contexts = *num_ready_contexts;
+    *num_ready_contexts = 0;
+    for (auto& handles : watched_handles_) {

[yzshen1, 2017/03/11 00:44:58]

nit: handles -> handle?

[Ken Rockot(use gerrit already), 2017/03/12 22:24:13]

Done

+      if (max_ready_contexts == 0)
+        break;
+      uint32_t n = handles.second->StoreReadyContextState(
+          max_ready_contexts, ready_contexts, ready_results,
+          ready_signals_states);
+      DCHECK_GE(max_ready_contexts, n);
+
+      *num_ready_contexts += n;
+      ready_contexts += n;
+      ready_results += n;
+      ready_signals_states += n;
+      max_ready_contexts -= n;
+    }
+  }
+
+  return MOJO_RESULT_FAILED_PRECONDITION;
+}
+
+WatcherDispatcher::~WatcherDispatcher() {}
+
+}  // namespace edk
+}  // namespace mojo