ipc/mojo/async_handle_waiter.cc

   1 // Copyright 2015 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "ipc/mojo/async_handle_waiter.h"
   6
   7 #include "base/atomic_ref_count.h"
   8 #include "base/bind.h"
   9 #include "base/bind_helpers.h"
  10 #include "base/location.h"
  11 #include "base/logging.h"
  12 #include "third_party/mojo/src/mojo/edk/embedder/embedder.h"
  13
  14 namespace IPC {
  15 namespace internal {
  16
  17 class AsyncHandleWaiterContextTraits {
  18  public:
  19   static void Destruct(const AsyncHandleWaiter::Context* context);
  20 };
  21
  22 // The thread-safe part of |AsyncHandleWaiter|.
  23 // As |AsyncWait()| invokes the given callback from an arbitrary thread,
  24 // |HandleIsReady()| and the bound |this| have to be thread-safe.
  25 class AsyncHandleWaiter::Context
  26     : public base::RefCountedThreadSafe<AsyncHandleWaiter::Context,
  27                                         AsyncHandleWaiterContextTraits>,
  28       public base::MessageLoopForIO::IOObserver {
  29  public:
  30   Context(base::WeakPtr<AsyncHandleWaiter> waiter)
  31       : io_runner_(base::MessageLoopForIO::current()->task_runner()),
  32         waiter_(waiter),
  33         last_result_(MOJO_RESULT_INTERNAL),
  34         io_loop_level_(0),
  35         should_invoke_callback_(false) {
  36     base::MessageLoopForIO::current()->AddIOObserver(this);
  37   }
  38
  39   void HandleIsReady(MojoResult result) {
  40     last_result_ = result;
  41
  42     // If the signaling happens in the IO handler, use |IOObserver| callback
  43     // to invoke the callback.
  44     if (IsCalledFromIOHandler()) {
  45       should_invoke_callback_ = true;
  46       return;
  47     }
  48
  49     io_runner_->PostTask(FROM_HERE,
  50                          base::Bind(&Context::InvokeWaiterCallback, this));
  51   }
  52
  53  private:
  54   friend void base::DeletePointer<const Context>(const Context* self);
  55   friend class AsyncHandleWaiterContextTraits;
  56   friend class base::RefCountedThreadSafe<Context>;
  57
  58   ~Context() override {
  59     DCHECK(base::MessageLoopForIO::current()->task_runner() == io_runner_);
  60     base::MessageLoopForIO::current()->RemoveIOObserver(this);
  61   }
  62
  63   bool IsCalledFromIOHandler() const {
  64     base::MessageLoop* loop = base::MessageLoop::current();
  65     if (!loop)
  66       return false;
  67     if (loop->task_runner() != io_runner_)
  68       return false;
  69     return io_loop_level_ > 0;
  70   }
  71
  72   // Called from |io_runner_| thus safe to touch |waiter_|.
  73   void InvokeWaiterCallback() {
  74     MojoResult result = last_result_;
  75     last_result_ = MOJO_RESULT_INTERNAL;
  76     if (waiter_)
  77       waiter_->InvokeCallback(result);
  78   }
  79
  80   // IOObserver implementation:
  81
  82   void WillProcessIOEvent() override {
  83     DCHECK(io_loop_level_ != 0 || !should_invoke_callback_);
  84     DCHECK_GE(io_loop_level_, 0);
  85     io_loop_level_++;
  86   }
  87
  88   void DidProcessIOEvent() override {
  89     DCHECK_GE(io_loop_level_, 1);
  90
  91     // Leaving a nested loop.
  92     if (io_loop_level_ > 1) {
  93       io_loop_level_--;
  94       return;
  95     }
  96
  97     // The zero |waiter_| indicates that |this| have lost the owner and can be
  98     // under destruction. So we cannot wrap it with a |scoped_refptr| anymore.
  99     if (!waiter_) {
 100       should_invoke_callback_ = false;
 101       io_loop_level_--;
 102       return;
 103     }
 104
 105     // We have to protect |this| because |AsyncHandleWaiter| can be
 106     // deleted during the callback.
 107     scoped_refptr<Context> protect(this);
 108     while (should_invoke_callback_) {
 109       should_invoke_callback_ = false;
 110       InvokeWaiterCallback();
 111     }
 112
 113     io_loop_level_--;
 114   }
 115
 116   // Only |io_runner_| is accessed from arbitrary threads.  Others are touched
 117   // only from the IO thread.
 118   const scoped_refptr<base::TaskRunner> io_runner_;
 119
 120   const base::WeakPtr<AsyncHandleWaiter> waiter_;
 121   MojoResult last_result_;
 122   int io_loop_level_;
 123   bool should_invoke_callback_;
 124
 125   DISALLOW_COPY_AND_ASSIGN(Context);
 126 };
 127
 128 AsyncHandleWaiter::AsyncHandleWaiter(base::Callback<void(MojoResult)> callback)
 129     : callback_(callback),
 130       weak_factory_(this) {
 131   context_ = new Context(weak_factory_.GetWeakPtr());
 132 }
 133
 134 AsyncHandleWaiter::~AsyncHandleWaiter() {
 135 }
 136
 137 MojoResult AsyncHandleWaiter::Wait(MojoHandle handle,
 138                                    MojoHandleSignals signals) {
 139   return mojo::embedder::AsyncWait(
 140       handle, signals, base::Bind(&Context::HandleIsReady, context_));
 141 }
 142
 143 void AsyncHandleWaiter::InvokeCallback(MojoResult result) {
 144   callback_.Run(result);
 145 }
 146
 147 base::MessageLoopForIO::IOObserver* AsyncHandleWaiter::GetIOObserverForTest() {
 148   return context_.get();
 149 }
 150
 151 base::Callback<void(MojoResult)> AsyncHandleWaiter::GetWaitCallbackForTest() {
 152   return base::Bind(&Context::HandleIsReady, context_);
 153 }
 154
 155 // static
 156 void AsyncHandleWaiterContextTraits::Destruct(
 157     const AsyncHandleWaiter::Context* context) {
 158   context->io_runner_->PostTask(
 159       FROM_HERE,
 160       base::Bind(&base::DeletePointer<const AsyncHandleWaiter::Context>,
 161                  base::Unretained(context)));
 162 }
 163
 164 }  // namespace internal
 165 }  // namespace IPC