Make mojo::Callback safe to copy across threads.

mojo/public/cpp/bindings/lib/router.cc

 // Copyright 2014 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/public/cpp/bindings/lib/router.h"
 
 #include <stdint.h>
 #include <utility>
 
 #include "base/bind.h"
+#include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
+#include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
+#include "base/thread_task_runner_handle.h"
 
 namespace mojo {
 namespace internal {
 
 // ----------------------------------------------------------------------------
 
 namespace {
 
+void DCheckIfValid(const base::WeakPtr<Router>& router,
+                   const std::string message) {

[yzshen1, 2016/03/18 23:47:04]

"const &"?

[Anand Mistry (off Chromium), 2016/03/19 01:02:13]

Done.

+  if (router && !router->encountered_error() && router->is_valid()) {
+    NOTREACHED() << message;

[yzshen1, 2016/03/18 23:47:04]

nit: maybe consider DCHECK(!the_current_condition)?

[Anand Mistry (off Chromium), 2016/03/19 01:02:13]

Done.

+  }
+}
+
 class ResponderThunk : public MessageReceiverWithStatus {
  public:
   explicit ResponderThunk(const base::WeakPtr<Router>& router)
-      : router_(router), accept_was_invoked_(false) {}
+      : router_(router), accept_was_invoked_(false),
+        task_runner_(base::ThreadTaskRunnerHandle::Get()) {}
   ~ResponderThunk() override {
     if (!accept_was_invoked_) {
       // The Mojo application handled a message that was expecting a response
       // but did not send a response.
-      if (router_) {
-        // We raise an error to signal the calling application that an error
-        // condition occurred. Without this the calling application would have
-        // no way of knowing it should stop waiting for a response.
-        router_->RaiseError();
+      if (task_runner_->RunsTasksOnCurrentThread()) {
+        if (router_) {
+          // We raise an error to signal the calling application that an error
+          // condition occurred. Without this the calling application would have
+          // no way of knowing it should stop waiting for a response.
+          router_->RaiseError();

[yzshen1, 2016/03/18 23:47:04]

In interface_endpoint_client.cc (which is used for the associated interface case
-- I really need to unify the code!), I post a task to raise error even on the
same thread. That way it won't break the case where the user abandons the
interface ptr/binding soon after he/she abandons the callback. For example, if
the callback and the interface ptr/binding are both destroyed by a class'
destructor, the user doesn't have to worry about which is destroyed first.

[Anand Mistry (off Chromium), 2016/03/19 01:02:13]

I traced the code down to Connector and noticed that Connector::RaiseError()
calls Connector::HandleError() with force_async_handler = true. This should
force any error handler to be posted instead of run synchronously. You're
concerned about error handler callbacks, right?

[yzshen1, 2016/03/19 01:29:54]

I think you are right. :)
So maybe we should change InterfaceEndpointClient to avoid posting task when it
is on the same thread.

Thanks for pointing this out.

[Anand Mistry (off Chromium), 2016/03/19 03:56:53]

Done.

+        }
+      } else {
+        task_runner_->PostTask(FROM_HERE,
+                               base::Bind(&Router::RaiseError, router_));
       }
     }
   }
 
   // MessageReceiver implementation:
   bool Accept(Message* message) override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     accept_was_invoked_ = true;
     DCHECK(message->has_flag(kMessageIsResponse));
 
     bool result = false;
 
     if (router_)
       result = router_->Accept(message);
 
     return result;
   }
 
   // MessageReceiverWithStatus implementation:
   bool IsValid() override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     return router_ && !router_->encountered_error() && router_->is_valid();
   }
 
+  void DCheckValid(const std::string& message) override {

[yzshen1, 2016/03/18 23:47:04]

Please do the same thing to the ResponderThunk in interface_endpoint_client.cc

[Anand Mistry (off Chromium), 2016/03/19 01:02:13]

Done.

+    if (task_runner_->RunsTasksOnCurrentThread()) {
+      DCheckIfValid(router_, message);
+    } else {
+      task_runner_->PostTask(FROM_HERE,
+                             base::Bind(&DCheckIfValid, router_, message));
+    }
+  }
+
  private:
   base::WeakPtr<Router> router_;
   bool accept_was_invoked_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 };
 
 }  // namespace
 
 // ----------------------------------------------------------------------------
 
 Router::SyncResponseInfo::SyncResponseInfo(bool* in_response_received)
     : response_received(in_response_received) {}
 
 Router::SyncResponseInfo::~SyncResponseInfo() {}
@@ skipping 200 matching lines @@
   }
 
   encountered_error_ = true;
   error_handler_.Run();
 }
 
 // ----------------------------------------------------------------------------
 
 }  // namespace internal
 }  // namespace mojo