Index: ipc/ipc_sync_message_filter.cc |
diff --git a/ipc/ipc_sync_message_filter.cc b/ipc/ipc_sync_message_filter.cc |
index 1d17432a319ff6f12d614b037edefde701d109eb..53ead2c0b8d17d52a25894839ced76f02479c8b0 100644 |
--- a/ipc/ipc_sync_message_filter.cc |
+++ b/ipc/ipc_sync_message_filter.cc |
@@ -7,14 +7,94 @@ |
#include "base/bind.h" |
#include "base/location.h" |
#include "base/logging.h" |
+#include "base/macros.h" |
+#include "base/memory/ptr_util.h" |
+#include "base/memory/ref_counted.h" |
+#include "base/message_loop/message_loop.h" |
#include "base/single_thread_task_runner.h" |
#include "base/synchronization/waitable_event.h" |
#include "base/threading/thread_task_runner_handle.h" |
#include "ipc/ipc_channel.h" |
#include "ipc/ipc_sync_message.h" |
+#include "ipc/mojo_event.h" |
+#include "mojo/public/cpp/bindings/sync_handle_registry.h" |
namespace IPC { |
+namespace { |
+ |
+// A generic callback used when watching handles synchronously. Sets |*signal| |
+// to true. Also sets |*error| to true in case of an error. |
+void OnSyncHandleReady(bool* signal, bool* error, MojoResult result) { |
+ *signal = true; |
+ *error = result != MOJO_RESULT_OK; |
+} |
+ |
+} // namespace |
+ |
+// A helper class created by SyncMessageFilter to watch the lifetime of the IO |
+// MessageLoop. This holds a weak ref to the SyncMessageFilter and notifies it |
+// on its own thread if the SyncMessageFilter is still alive at the time of |
+// IO MessageLoop destruction. |
+class SyncMessageFilter::IOMessageLoopObserver |
+ : public base::MessageLoop::DestructionObserver, |
+ public base::RefCountedThreadSafe<IOMessageLoopObserver> { |
+ public: |
+ IOMessageLoopObserver( |
+ base::WeakPtr<SyncMessageFilter> weak_filter, |
+ scoped_refptr<base::SingleThreadTaskRunner> filter_task_runner) |
+ : weak_filter_(weak_filter), filter_task_runner_(filter_task_runner) {} |
+ |
+ void StartOnIOThread() { |
+ DCHECK(!watching_); |
+ watching_ = true; |
+ io_task_runner_ = base::ThreadTaskRunnerHandle::Get(); |
+ base::MessageLoop::current()->AddDestructionObserver(this); |
+ } |
+ |
+ void Stop() { |
+ if (!io_task_runner_) |
+ return; |
+ |
+ if (io_task_runner_->BelongsToCurrentThread()) { |
+ StopOnIOThread(); |
+ } else { |
+ io_task_runner_->PostTask( |
+ FROM_HERE, base::Bind(&IOMessageLoopObserver::StopOnIOThread, this)); |
+ } |
+ } |
+ |
+ private: |
+ void StopOnIOThread() { |
+ DCHECK(io_task_runner_->BelongsToCurrentThread()); |
+ if (!watching_) |
+ return; |
+ watching_ = false; |
+ base::MessageLoop::current()->RemoveDestructionObserver(this); |
+ } |
+ |
+ // base::MessageLoop::DestructionObserver: |
+ void WillDestroyCurrentMessageLoop() override { |
+ DCHECK(io_task_runner_ && io_task_runner_->BelongsToCurrentThread()); |
+ DCHECK(watching_); |
+ StopOnIOThread(); |
+ filter_task_runner_->PostTask( |
+ FROM_HERE, |
+ base::Bind(&SyncMessageFilter::OnIOMessageLoopDestroyed, weak_filter_)); |
+ } |
+ |
+ friend class base::RefCountedThreadSafe<IOMessageLoopObserver>; |
+ |
+ ~IOMessageLoopObserver() override {} |
+ |
+ bool watching_ = false; |
+ base::WeakPtr<SyncMessageFilter> weak_filter_; |
+ scoped_refptr<base::SingleThreadTaskRunner> filter_task_runner_; |
+ scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_; |
+ |
+ DISALLOW_COPY_AND_ASSIGN(IOMessageLoopObserver); |
+}; |
+ |
bool SyncMessageFilter::Send(Message* message) { |
if (!message->is_sync()) { |
{ |
@@ -23,7 +103,7 @@ bool SyncMessageFilter::Send(Message* message) { |
sender_->Send(message); |
return true; |
} else if (!io_task_runner_.get()) { |
- pending_messages_.push_back(message); |
+ pending_messages_.emplace_back(base::WrapUnique(message)); |
return true; |
} |
} |
@@ -33,9 +113,7 @@ bool SyncMessageFilter::Send(Message* message) { |
return true; |
} |
- base::WaitableEvent done_event( |
- base::WaitableEvent::ResetPolicy::MANUAL, |
- base::WaitableEvent::InitialState::NOT_SIGNALED); |
+ MojoEvent done_event; |
PendingSyncMsg pending_message( |
SyncMessage::GetMessageId(*message), |
static_cast<SyncMessage*>(message)->GetReplyDeserializer(), |
@@ -56,15 +134,32 @@ bool SyncMessageFilter::Send(Message* message) { |
FROM_HERE, |
base::Bind(&SyncMessageFilter::SendOnIOThread, this, message)); |
} else { |
- pending_messages_.push_back(message); |
+ pending_messages_.emplace_back(base::WrapUnique(message)); |
} |
} |
- base::WaitableEvent* events[2] = { shutdown_event_, &done_event }; |
- if (base::WaitableEvent::WaitMany(events, 2) == 1) { |
+ bool done = false; |
+ bool shutdown = false; |
+ bool error = false; |
+ scoped_refptr<mojo::SyncHandleRegistry> registry = |
+ mojo::SyncHandleRegistry::current(); |
+ registry->RegisterHandle(shutdown_mojo_event_.GetHandle(), |
+ MOJO_HANDLE_SIGNAL_READABLE, |
+ base::Bind(&OnSyncHandleReady, &shutdown, &error)); |
+ registry->RegisterHandle(done_event.GetHandle(), |
+ MOJO_HANDLE_SIGNAL_READABLE, |
+ base::Bind(&OnSyncHandleReady, &done, &error)); |
+ |
+ const bool* stop_flags[] = { &done, &shutdown }; |
+ registry->WatchAllHandles(stop_flags, 2); |
+ DCHECK(!error); |
+ |
+ if (done) { |
TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"), |
"SyncMessageFilter::Send", &done_event); |
} |
+ registry->UnregisterHandle(shutdown_mojo_event_.GetHandle()); |
+ registry->UnregisterHandle(done_event.GetHandle()); |
{ |
base::AutoLock auto_lock(lock_); |
@@ -76,26 +171,32 @@ bool SyncMessageFilter::Send(Message* message) { |
} |
void SyncMessageFilter::OnFilterAdded(Sender* sender) { |
- std::vector<Message*> pending_messages; |
+ std::vector<std::unique_ptr<Message>> pending_messages; |
{ |
base::AutoLock auto_lock(lock_); |
sender_ = sender; |
io_task_runner_ = base::ThreadTaskRunnerHandle::Get(); |
- pending_messages_.release(&pending_messages); |
+ shutdown_watcher_.StartWatching( |
+ shutdown_event_, |
+ base::Bind(&SyncMessageFilter::OnShutdownEventSignaled, this)); |
+ io_message_loop_observer_->StartOnIOThread(); |
+ std::swap(pending_messages_, pending_messages); |
} |
- for (auto* msg : pending_messages) |
- SendOnIOThread(msg); |
+ for (auto& msg : pending_messages) |
+ SendOnIOThread(msg.release()); |
} |
void SyncMessageFilter::OnChannelError() { |
base::AutoLock auto_lock(lock_); |
sender_ = NULL; |
+ shutdown_watcher_.StopWatching(); |
SignalAllEvents(); |
} |
void SyncMessageFilter::OnChannelClosing() { |
base::AutoLock auto_lock(lock_); |
sender_ = NULL; |
+ shutdown_watcher_.StopWatching(); |
SignalAllEvents(); |
} |
@@ -124,10 +225,14 @@ SyncMessageFilter::SyncMessageFilter(base::WaitableEvent* shutdown_event, |
: sender_(NULL), |
is_channel_send_thread_safe_(is_channel_send_thread_safe), |
listener_task_runner_(base::ThreadTaskRunnerHandle::Get()), |
- shutdown_event_(shutdown_event) { |
+ shutdown_event_(shutdown_event), |
+ weak_factory_(this) { |
+ io_message_loop_observer_ = new IOMessageLoopObserver( |
+ weak_factory_.GetWeakPtr(), listener_task_runner_); |
} |
SyncMessageFilter::~SyncMessageFilter() { |
+ io_message_loop_observer_->Stop(); |
} |
void SyncMessageFilter::SendOnIOThread(Message* message) { |
@@ -157,4 +262,17 @@ void SyncMessageFilter::SignalAllEvents() { |
} |
} |
+void SyncMessageFilter::OnShutdownEventSignaled(base::WaitableEvent* event) { |
+ DCHECK_EQ(event, shutdown_event_); |
+ shutdown_mojo_event_.Signal(); |
+} |
+ |
+void SyncMessageFilter::OnIOMessageLoopDestroyed() { |
+ // Since we use an async WaitableEventWatcher to watch the shutdown event |
+ // from the IO thread, we can't forward the shutdown signal after the IO |
+ // message loop is destroyed. Since that destruction indicates shutdown |
+ // anyway, we manually signal the shutdown event in this case. |
+ shutdown_mojo_event_.Signal(); |
+} |
+ |
} // namespace IPC |