mojo: Support sync calls through ThreadSafeInterfacePtr

mojo/public/cpp/bindings/thread_safe_interface_ptr.h

Index: mojo/public/cpp/bindings/thread_safe_interface_ptr.h
diff --git a/mojo/public/cpp/bindings/thread_safe_interface_ptr.h b/mojo/public/cpp/bindings/thread_safe_interface_ptr.h
index 8b32b30b4fbefe1a5a07e1bb089b97e337d9be74..21ac15ab06b33952e01e0de5a7b10a36c467b1b0 100644
--- a/mojo/public/cpp/bindings/thread_safe_interface_ptr.h
+++ b/mojo/public/cpp/bindings/thread_safe_interface_ptr.h
@@ -10,12 +10,16 @@
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
+#include "base/stl_util.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/task_runner.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/associated_interface_ptr.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
 #include "mojo/public/cpp/bindings/message.h"
+#include "mojo/public/cpp/bindings/sync_call_restrictions.h"
+#include "mojo/public/cpp/bindings/sync_event_watcher.h"
 
 namespace mojo {
 
@@ -48,7 +52,15 @@ class ThreadSafeForwarder : public MessageReceiverWithResponder {
         forward_with_responder_(forward_with_responder),
         associated_group_(associated_group) {}
 
-  ~ThreadSafeForwarder() override {}
+  ~ThreadSafeForwarder() override {
+    // If there's an ongoing sync call (which deleted |this|), signal its
+    // completion now.
+    {
+      base::AutoLock l(lock_);
+      if (!pending_sync_responses_.empty())
+        pending_sync_responses_.front()->event.Signal();

[yzshen1, 2017/03/29 23:16:47]

Shouldn't we signal all the events?

[watk, 2017/03/30 02:26:05]

Done.

+    }
+  }
 
   ProxyType& proxy() { return proxy_; }
 
@@ -75,34 +87,116 @@ class ThreadSafeForwarder : public MessageReceiverWithResponder {
 
   bool AcceptWithResponder(
       Message* message,
-      std::unique_ptr<MessageReceiver> response_receiver) override {
+      std::unique_ptr<MessageReceiver> responder) override {
     if (!message->associated_endpoint_handles()->empty()) {
       // Please see comment for the DCHECK in the previous method.
       DCHECK(associated_group_.GetController());
       message->SerializeAssociatedEndpointHandles(
           associated_group_.GetController());
     }
-    auto responder =
-        base::MakeUnique<ForwardToCallingThread>(std::move(response_receiver));
+
+    // Async messages are always posted (even if |task_runner_| runs tasks on
+    // this thread) to guarantee that two async calls can't be reordered.
+    if (!message->has_flag(Message::kFlagIsSync)) {
+      auto reply_forwarder =
+          base::MakeUnique<ForwardToCallingThread>(std::move(responder));
+      task_runner_->PostTask(
+          FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message),
+                                base::Passed(&reply_forwarder)));
+      return true;
+    }
+
+    SyncCallRestrictions::AssertSyncCallAllowed();
+
+    // If the InterfacePtr is bound to this thread, dispatch it directly.
+    if (task_runner_->RunsTasksOnCurrentThread()) {
+      forward_with_responder_.Run(std::move(*message), std::move(responder));
+      return true;
+    }
+
+    // If the InterfacePtr is bound on another thread, post the call.
+    // TODO(yzshen, watk): We block both this thread and the InterfacePtr

[yzshen1, 2017/03/29 23:16:47]

IO thread must not be blocked. This prevents users from using sync calls with
ThreadSafeInterfacePtr whose InterfacePtr lives on IO thread.

That being said, I am fine leaving it as a TODO.

[watk, 2017/03/30 02:26:05]

Yeah this is a shame :( I just ran out of time to implement that. I suspect
there's some trickiness there.

[yzshen1, 2017/03/30 20:53:05]

Please document this restriction WRT IO thread in the comment.

[watk, 2017/03/31 00:27:27]

I added a file-level comment. Is that OK?

+    // thread. Ideally only this thread would block.
+    auto response =
+        make_scoped_refptr(new base::RefCountedData<SyncResponseInfo>());
+    auto response_signaler = base::MakeUnique<SyncResponseSignaler>(response);
     task_runner_->PostTask(
         FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message),
-                              base::Passed(&responder)));
+                              base::Passed(&response_signaler)));
+
+    // Save the pending SyncResponseInfo so that if the sync call manages to
+    // delete |this|, we can signal the completion of the call to return from
+    // SyncWatch().
+    {
+      base::AutoLock l(lock_);
+      pending_sync_responses_.push_back(&response->data);
+    }
+
+    auto assign_true = [](bool* b) { *b = true; };
+    bool event_signaled = false;
+    SyncEventWatcher watcher(&response->data.event,
+                             base::Bind(assign_true, &event_signaled));
+    watcher.SyncWatch(&event_signaled);
+
+    if (event_signaled && response->data.received)
+      ignore_result(responder->Accept(&response->data.message));
+
+    {
+      base::AutoLock l(lock_);

[yzshen1, 2017/03/29 23:16:47]

The object may be destroyed, so it is not safe to access the member variables
without checking, right?

[watk, 2017/03/30 02:26:04]

Oops! Made it refcounted.

+      base::Erase(pending_sync_responses_, &response->data);
+    }
+
     return true;
   }
 
+  // Data that we need to share between the threads involved in a sync call.
+  struct SyncResponseInfo {

[yzshen1, 2017/03/29 23:16:47]

nit: because you have full control over this type, you could directly make it to
support ref-counted, instead of using RefCountedData

[watk, 2017/03/30 02:26:04]

Done.

+    Message message;
+    bool received = false;
+    base::WaitableEvent event{base::WaitableEvent::ResetPolicy::MANUAL,
+                              base::WaitableEvent::InitialState::NOT_SIGNALED};
+  };
+
+  // A MessageReceiver that signals its SyncResponse as complete when it either
+  // accepts the response message, or is destructed.
+  class SyncResponseSignaler : public MessageReceiver {
+   public:
+    explicit SyncResponseSignaler(
+        scoped_refptr<base::RefCountedData<SyncResponseInfo>> response)
+        : response_(response) {}
+
+    ~SyncResponseSignaler() override {
+      // If Accept() was not called we must still notify the waiter that the
+      // sync call is finished.
+      if (response_)
+        response_->data.event.Signal();
+    }
+
+    bool Accept(Message* message) {
+      response_->data.message = std::move(*message);
+      response_->data.received = true;
+      response_->data.event.Signal();
+      response_ = nullptr;
+      return true;
+    }
+
+   private:
+    scoped_refptr<base::RefCountedData<SyncResponseInfo>> response_;
+  };
+
   class ForwardToCallingThread : public MessageReceiver {
    public:
     explicit ForwardToCallingThread(std::unique_ptr<MessageReceiver> responder)
         : responder_(std::move(responder)),
-          caller_task_runner_(base::ThreadTaskRunnerHandle::Get()) {
-    }
+          caller_task_runner_(base::ThreadTaskRunnerHandle::Get()) {}
 
    private:
     bool Accept(Message* message) {
       // The current instance will be deleted when this method returns, so we
       // have to relinquish the responder's ownership so it does not get
       // deleted.
-      caller_task_runner_->PostTask(FROM_HERE,
+      caller_task_runner_->PostTask(
+          FROM_HERE,
           base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder,
                      base::Passed(std::move(responder_)),
                      base::Passed(std::move(*message))));
@@ -125,6 +219,10 @@ class ThreadSafeForwarder : public MessageReceiverWithResponder {
   const ForwardMessageWithResponderCallback forward_with_responder_;
   AssociatedGroup associated_group_;
 
+  // |lock_| protects access to |pending_sync_responses_|.
+  base::Lock lock_;
+  std::vector<SyncResponseInfo*> pending_sync_responses_;
+
   DISALLOW_COPY_AND_ASSIGN(ThreadSafeForwarder);
 };