mojo: Support sync calls through ThreadSafeInterfacePtr

mojo/public/cpp/bindings/thread_safe_interface_ptr.h

 // Copyright 2016 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.
 
 #ifndef MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_
 #define MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_
 
 #include <memory>
 
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.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_event_watcher.h"
 
 namespace mojo {
+namespace {

[yzshen1, 2017/03/27 22:18:25]

It is pretty uncommon and not recommended to use anonymous namespace in header
files.

How about making it a static member of ThreadSafeForwarder?

[watk, 2017/03/27 23:45:28]

Oops, forgot this was a header.

+void AssignTrue(bool* b) {
+  *b = true;
+}
+}  // namespace
 
 // Instances of this class may be used from any thread to serialize |Interface|
 // messages and forward them elsewhere. In general you should use one of the
 // ThreadSafeInterfacePtrBase helper aliases defined below, but this type may be
 // useful if you need/want to manually manage the lifetime of the underlying
 // proxy object which will be used to ultimately send messages.
 template <typename Interface>
 class ThreadSafeForwarder : public MessageReceiverWithResponder {
  public:
   using ProxyType = typename Interface::Proxy_;
   using ForwardMessageCallback = base::Callback<void(Message)>;
   using ForwardMessageWithResponderCallback =
       base::Callback<void(Message, std::unique_ptr<MessageReceiver>)>;
 
   // Constructs a ThreadSafeForwarder through which Messages are forwarded to
   // |forward| or |forward_with_responder| by posting to |task_runner|.
   //
   // Any message sent through this forwarding interface will dispatch its reply,
   // if any, back to the thread which called the corresponding interface method.
   ThreadSafeForwarder(
       const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
       const ForwardMessageCallback& forward,
       const ForwardMessageWithResponderCallback& forward_with_responder,
       const AssociatedGroup& associated_group)
       : proxy_(this),
         task_runner_(task_runner),
         forward_(forward),
         forward_with_responder_(forward_with_responder),
-        associated_group_(associated_group) {}
+        associated_group_(associated_group),
+        sync_message_event_(base::WaitableEvent::ResetPolicy::MANUAL,
+                            base::WaitableEvent::InitialState::NOT_SIGNALED),
+        weak_factory_(this) {}
 
   ~ThreadSafeForwarder() override {}
 
   ProxyType& proxy() { return proxy_; }
 
  private:
   // MessageReceiverWithResponder implementation:
   bool Accept(Message* message) override {
     if (!message->associated_endpoint_handles()->empty()) {
       // If this DCHECK fails, it is likely because:
       // - This is a non-associated interface pointer setup using
       //     PtrWrapper::BindOnTaskRunner(
       //         InterfacePtrInfo<InterfaceType> ptr_info);
       //   Please see the TODO in that method.
       // - This is an associated interface which hasn't been associated with a
       //   message pipe. In other words, the corresponding
       //   AssociatedInterfaceRequest hasn't been sent.
       DCHECK(associated_group_.GetController());
       message->SerializeAssociatedEndpointHandles(
           associated_group_.GetController());
     }
     task_runner_->PostTask(FROM_HERE,
                            base::Bind(forward_, base::Passed(message)));
     return true;
   }
 
   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));
+
+    if (!message->has_flag(Message::kFlagIsSync)) {
+      auto reply_forwarder =
+          base::MakeUnique<ForwardToCallingThread>(std::move(responder));
+      if (task_runner_->RunsTasksOnCurrentThread()) {
+        forward_with_responder_.Run(std::move(*message),

[yzshen1, 2017/03/27 22:18:24]

I think we would like to preserve order between calls, and therefore we don't
directly run |forward_with_responder_| even if on the same thread.

EXAMPLE:
Imagine thread_a uses ThreadSafeForwarder to make a call X, and then signals
(using some way other than posting a task) thread_b to use the same forwarder to
make a call Y. The forwarder's |task_runner_| is bound to thread_b.

If we pass the Y call directly to the interface pointer, it is possible that Y
happens prior to X.

[watk, 2017/03/27 23:45:29]

True. I had assumed there was a contract that TSIP is to be used on one thread
only, but that was a false assumption. Hmm, this is tricky then. I added this to
preserve the invariant that queued async resonses are dispatched before the
error notification. If we always post, that is broken. So either that's not
worth trying to preserve for TSIP, or I should find another way. I'll default to
the latter if you have no thoughts.

[yzshen1, 2017/03/28 00:06:48]

What error notification? :)

[watk, 2017/03/28 00:14:35]

Take a look at sync_method_unittest.cc line 853. That's the test that motivated
not always posting async messages.

+                                    std::move(reply_forwarder));
+      } else {
+        task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(forward_with_responder_, base::Passed(message),
+                       base::Passed(&reply_forwarder)));
+      }
+      return true;
+    }
+
+    // Sync call on this thread.
+    if (task_runner_->RunsTasksOnCurrentThread()) {
+      forward_with_responder_.Run(std::move(*message), std::move(responder));

[yzshen1, 2017/03/27 22:18:24]

+CC Ken:

At the beginning, I thought this direct call would have the same issue as the
one in the async case. But after some more thoughts, it seems fine:
First, we don't care about ordering between sync calls and async calls, because
they could be out-of-order due to re-entrancy anyway.

If the order between two sync calls made from a ThreadSafeInterfacePtr is
well-defined, in all the cases that I can think up, the underlying interface ptr
will run those two calls in order. 

WDYT? Have I missed anything obvious?

[watk, 2017/03/27 23:45:29]

I did this because I wasn't sure what the alternative would be, i.e., we can't
post the task because we have to SyncWatch for the response on the same thread.
Oh, we could run a nested message loop? 

[yzshen1, 2017/03/28 00:06:48]

What I was trying to say is "I thought this doesn't work but after some more
thoughts I think you are doing the right thing". :)

[watk, 2017/03/28 00:14:35]

Oh, ok :)

+      return true;
+    }
+
+    // Sync call on a different thread.
+    // TODO: allow the call to be async on the other thread.
+    Message response;
+    auto reply_signaler =
+        base::MakeUnique<SyncReplySignaler>(&response, &sync_message_event_);
     task_runner_->PostTask(
         FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message),
-                              base::Passed(&responder)));
+                              base::Passed(&reply_signaler)));
+    bool response_received = false;
+    // XXX: What does it mean to have reentrant sync calls that are all waiting
+    // on the same sync_message_event_? Correct because only reentrant sync

[yzshen1, 2017/03/27 22:18:25]

I am not quite sure I understand the question. Would you please explain?

[watk, 2017/03/27 23:45:29]

I think I wrote this before I had the sync_message_event_.Reset() below and I
wasn't sure if it was correct that a nested sync call completing would also wake
up the outer sync call. I don't think this is a concern any more because the
event is reset. So the outer sync call will need to be signaled separately from
the inner one.

+    // calls are allowed?
+    SyncEventWatcher watcher(&sync_message_event_,
+                             base::Bind(&AssignTrue, &response_received));
+    watcher.AllowWokenUpBySyncWatchOnSameThread();

[yzshen1, 2017/03/27 22:18:24]

I don't think you need this line.

This is for the implementation of interfaces that are listening for sync
requests on the same thread. They may need to be woken up while this interface
pointer is waiting for the sync call response.

[watk, 2017/03/27 23:45:29]

Ah, gotcha.

+    // XXX: Check that this handles the case that sync_message_event_ is already
+    // signaled.
+    auto weak_self = weak_factory_.GetWeakPtr();

[yzshen1, 2017/03/27 22:18:24]

WeakPtrFactory will force this object to be only used on a specific thread.

If all we need is to track whether |this| is destroyed. You could use something
like
https://cs.chromium.org/chromium/src/mojo/public/cpp/bindings/sync_handle_wat...

[watk, 2017/03/27 23:45:29]

SG, thanks

+    watcher.SyncWatch(&response_received);
+    if (weak_self) {
+      sync_message_event_.Reset();
+      if (response_received)
+        ignore_result(responder->Accept(&response));

[yzshen1, 2017/03/27 22:18:24]

The response may not correspond to the |responder|.

Example:

1) ptr_A make sync call X;
2) while waiting for X's response, an incoming sync request for another
interface B renters this thread
3) while interface B handles this request, it uses ptr_A to make sync call Y;
4) the X's response arrive, in this case it will be incorrectly considered as
response for Y.

That is why InterfaceEndpointClient needs to have a SyncResponseMap. You could
take a look at that.

[watk, 2017/03/27 23:45:28]

Ah, yes, thanks. I hadn't even thought about non-nested sync calls :O

+    }
+
     return true;
   }
 
+  class SyncReplySignaler : public MessageReceiver {
+   public:
+    SyncReplySignaler(Message* sync_response_destination,
+                      base::WaitableEvent* sync_response_event)
+        : sync_response_destination_(sync_response_destination),
+          sync_response_event_(sync_response_event) {}
+
+   private:
+    bool Accept(Message* message) {
+      *sync_response_destination_ = std::move(*message);
+      sync_response_event_->Signal();
+      return true;
+    }
+
+    // XXX: What guarantees do we have about the lifetime of these?

[yzshen1, 2017/03/27 22:18:25]

The sync_response_event_ may be destroyed in the middle, right?

Example:

1) forwarder_A is used to make sync call X;
2) while waiting for X's response, an incoming sync request for another
interface B renters this thread;
3) while interface B handles this request, it destroys forwarder_A;
4) if the underlying interface pointer is kept alive, it holds a
SyncReplySignaler with invalid pointer.

[watk, 2017/03/27 23:45:29]

Yes, true. I'll make this refcounted then.

+    Message* sync_response_destination_;
+    base::WaitableEvent* sync_response_event_;
+  };
+
   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,
-          base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder,
-                     base::Passed(std::move(responder_)),
-                     base::Passed(std::move(*message))));
+      if (caller_task_runner_->RunsTasksOnCurrentThread()) {
+        CallAcceptAndDeleteResponder(std::move(responder_),
+                                     std::move(*message));
+      } else {
+        caller_task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder,
+                       base::Passed(std::move(responder_)),
+                       base::Passed(std::move(*message))));
+      }
       return true;
     }
 
     static void CallAcceptAndDeleteResponder(
         std::unique_ptr<MessageReceiver> responder,
         Message message) {
       ignore_result(responder->Accept(&message));
     }
 
     std::unique_ptr<MessageReceiver> responder_;
     scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_;
   };
 
   ProxyType proxy_;
   const scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   const ForwardMessageCallback forward_;
   const ForwardMessageWithResponderCallback forward_with_responder_;
   AssociatedGroup associated_group_;
 
+  // An event used to signal that sync messages are available.
+  base::WaitableEvent sync_message_event_;
+  base::WeakPtrFactory<ThreadSafeForwarder> weak_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(ThreadSafeForwarder);
 };
 
 template <typename InterfacePtrType>
 class ThreadSafeInterfacePtrBase
     : public base::RefCountedThreadSafe<
           ThreadSafeInterfacePtrBase<InterfacePtrType>> {
  public:
   using InterfaceType = typename InterfacePtrType::InterfaceType;
   using PtrInfoType = typename InterfacePtrType::PtrInfoType;
@@ skipping 132 matching lines @@
 using ThreadSafeAssociatedInterfacePtr =
     ThreadSafeInterfacePtrBase<AssociatedInterfacePtr<Interface>>;
 
 template <typename Interface>
 using ThreadSafeInterfacePtr =
     ThreadSafeInterfacePtrBase<InterfacePtr<Interface>>;
 
 }  // namespace mojo
 
 #endif  // MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_