mojo/core/watcher_dispatcher.cc

// 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/core/watcher_dispatcher.h"

#include <algorithm>
#include <limits>

#include "base/debug/alias.h"
#include "base/macros.h"
#include "base/memory/ptr_util.h"
#include "mojo/core/watch.h"

namespace mojo {
namespace core {

WatcherDispatcher::WatcherDispatcher(MojoTrapEventHandler handler)
    : handler_(handler) {}

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 the watch associated with this dispatcher,
  // provided we're armed and it cares about the new state.
  if (it->second->NotifyState(state, armed_)) {
    ready_watches_.insert(it->second.get());

    // If we were armed and got here, we notified the watch. Disarm.
    armed_ = false;
  } else {
    ready_watches_.erase(it->second.get());
  }
}

void WatcherDispatcher::NotifyHandleClosed(Dispatcher* dispatcher) {
  scoped_refptr<Watch> watch;
  {
    base::AutoLock lock(lock_);
    auto it = watched_handles_.find(dispatcher);
    if (it == watched_handles_.end())
      return;

    watch = std::move(it->second);

    // Wipe out all state associated with the closed dispatcher.
    watches_.erase(watch->context());
    ready_watches_.erase(watch.get());
    watched_handles_.erase(it);
  }

  // NOTE: It's important that this is called outside of |lock_| since it
  // acquires internal Watch locks.
  watch->Cancel();
}

void WatcherDispatcher::InvokeWatchCallback(uintptr_t context,
                                            MojoResult result,
                                            const HandleSignalsState& state,
                                            MojoTrapEventFlags flags) {
  MojoTrapEvent event;
  event.struct_size = sizeof(event);
  event.trigger_context = context;
  event.result = result;
  event.signals_state = static_cast<MojoHandleSignalsState>(state);
  event.flags = 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;
  }

  handler_(&event);
}

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.
  base::flat_map<uintptr_t, scoped_refptr<Watch>> watches;
  {
    base::AutoLock lock(lock_);
    if (closed_)
      return MOJO_RESULT_INVALID_ARGUMENT;
    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,
    MojoTriggerCondition condition,
    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 (closed_)
      return MOJO_RESULT_INVALID_ARGUMENT;

    if (watches_.count(context) || watched_handles_.count(dispatcher.get()))
      return MOJO_RESULT_ALREADY_EXISTS;

    scoped_refptr<Watch> watch =
        new Watch(this, dispatcher, context, signals, condition);
    watches_.insert({context, watch});
    auto result =
        watched_handles_.insert(std::make_pair(dispatcher.get(), watch));
    DCHECK(result.second);
  }

  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.
    base::AutoLock lock(lock_);
    watches_.erase(context);
    watched_handles_.erase(dispatcher.get());
    return rv;
  }

  bool remove_now;
  {
    // If we've been closed already, there's a chance our closure raced with
    // the call to AddWatcherRef() above. In that case we want to ensure we've
    // removed our ref from |dispatcher|. Note that this may in turn race
    // with normal removal, but that's fine.
    base::AutoLock lock(lock_);
    remove_now = closed_;
  }
  if (remove_now)
    dispatcher->RemoveWatcherRef(this, context);

  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_);
    if (closed_)
      return MOJO_RESULT_INVALID_ARGUMENT;
    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());

    // If another thread races to close this watcher handler, |watched_handles_|
    // may have been cleared by the time we reach this section.
    if (handle_it == watched_handles_.end())
      return MOJO_RESULT_OK;

    ready_watches_.erase(handle_it->second.get());
    watched_handles_.erase(handle_it);
  }

  return MOJO_RESULT_OK;
}

MojoResult WatcherDispatcher::Arm(uint32_t* num_blocking_events,
                                  MojoTrapEvent* blocking_events) {
  base::AutoLock lock(lock_);
  if (num_blocking_events && !blocking_events)
    return MOJO_RESULT_INVALID_ARGUMENT;
  if (closed_)
    return MOJO_RESULT_INVALID_ARGUMENT;

  if (watched_handles_.empty())
    return MOJO_RESULT_NOT_FOUND;

  if (ready_watches_.empty()) {
    // Fast path: No watches are ready to notify, so we're done.
    armed_ = true;
    return MOJO_RESULT_OK;
  }

  if (num_blocking_events) {
    DCHECK_LE(ready_watches_.size(), std::numeric_limits<uint32_t>::max());
    *num_blocking_events = std::min(
        *num_blocking_events, static_cast<uint32_t>(ready_watches_.size()));

    WatchSet::const_iterator next_ready_iter = ready_watches_.begin();
    if (last_watch_to_block_arming_) {
      // Find the next watch to notify in simple round-robin order on the
      // |ready_watches_| map, wrapping around to the beginning if necessary.
      next_ready_iter = ready_watches_.find(last_watch_to_block_arming_);
      if (next_ready_iter != ready_watches_.end())
        ++next_ready_iter;
      if (next_ready_iter == ready_watches_.end())
        next_ready_iter = ready_watches_.begin();
    }

    for (size_t i = 0; i < *num_blocking_events; ++i) {
      const Watch* const watch = *next_ready_iter;
      if (blocking_events[i].struct_size < sizeof(*blocking_events))
        return MOJO_RESULT_INVALID_ARGUMENT;
      blocking_events[i].flags = MOJO_TRAP_EVENT_FLAG_WITHIN_API_CALL;
      blocking_events[i].trigger_context = watch->context();
      blocking_events[i].result = watch->last_known_result();
      blocking_events[i].signals_state = watch->last_known_signals_state();

      // Iterate and wrap around.
      last_watch_to_block_arming_ = watch;
      ++next_ready_iter;
      if (next_ready_iter == ready_watches_.end())
        next_ready_iter = ready_watches_.begin();
    }
  }

  return MOJO_RESULT_FAILED_PRECONDITION;
}

WatcherDispatcher::~WatcherDispatcher() = default;

}  // namespace core
}  // namespace mojo