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
watk
2017/03/30 02:26:05
Yeah this is a shame :( I just ran out of time to
yzshen1
2017/03/30 20:53:05
Please document this restriction WRT IO thread in
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
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,
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); |
}; |