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