ipc_sync_channel.cc 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628
  1. // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "ipc/ipc_sync_channel.h"
  5. #include <stddef.h>
  6. #include <stdint.h>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/lazy_instance.h"
  10. #include "base/location.h"
  11. #include "base/logging.h"
  12. #include "base/memory/ptr_util.h"
  13. #include "base/memory/raw_ptr.h"
  14. #include "base/run_loop.h"
  15. #include "base/synchronization/waitable_event.h"
  16. #include "base/task/sequenced_task_runner.h"
  17. #include "base/threading/thread_local.h"
  18. #include "base/threading/thread_task_runner_handle.h"
  19. #include "base/trace_event/trace_event.h"
  20. #include "build/build_config.h"
  21. #include "ipc/ipc_channel_factory.h"
  22. #include "ipc/ipc_logging.h"
  23. #include "ipc/ipc_message_macros.h"
  24. #include "ipc/ipc_sync_message.h"
  25. #include "mojo/public/cpp/bindings/sync_event_watcher.h"
  26. #if !BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
  27. #include "ipc/trace_ipc_message.h"
  28. #endif
  29. using base::WaitableEvent;
  30. namespace IPC {
  31. namespace {
  32. // A generic callback used when watching handles synchronously. Sets |*signal|
  33. // to true.
  34. void OnEventReady(bool* signal) {
  35. *signal = true;
  36. }
  37. } // namespace
  38. // When we're blocked in a Send(), we need to process incoming synchronous
  39. // messages right away because it could be blocking our reply (either
  40. // directly from the same object we're calling, or indirectly through one or
  41. // more other channels). That means that in SyncContext's OnMessageReceived,
  42. // we need to process sync message right away if we're blocked. However a
  43. // simple check isn't sufficient, because the listener thread can be in the
  44. // process of calling Send.
  45. // To work around this, when SyncChannel filters a sync message, it sets
  46. // an event that the listener thread waits on during its Send() call. This
  47. // allows us to dispatch incoming sync messages when blocked. The race
  48. // condition is handled because if Send is in the process of being called, it
  49. // will check the event. In case the listener thread isn't sending a message,
  50. // we queue a task on the listener thread to dispatch the received messages.
  51. // The messages are stored in this queue object that's shared among all
  52. // SyncChannel objects on the same thread (since one object can receive a
  53. // sync message while another one is blocked).
  54. class SyncChannel::ReceivedSyncMsgQueue :
  55. public base::RefCountedThreadSafe<ReceivedSyncMsgQueue> {
  56. public:
  57. // Returns the ReceivedSyncMsgQueue instance for this thread, creating one
  58. // if necessary. Call RemoveContext on the same thread when done.
  59. static ReceivedSyncMsgQueue* AddContext() {
  60. // We want one ReceivedSyncMsgQueue per listener thread (i.e. since multiple
  61. // SyncChannel objects can block the same thread).
  62. ReceivedSyncMsgQueue* rv = lazy_tls_ptr_.Pointer()->Get();
  63. if (!rv) {
  64. rv = new ReceivedSyncMsgQueue();
  65. ReceivedSyncMsgQueue::lazy_tls_ptr_.Pointer()->Set(rv);
  66. }
  67. rv->listener_count_++;
  68. return rv;
  69. }
  70. // Prevents messages from being dispatched immediately when the dispatch event
  71. // is signaled. Instead, |*dispatch_flag| will be set.
  72. void BlockDispatch(bool* dispatch_flag) { dispatch_flag_ = dispatch_flag; }
  73. // Allows messages to be dispatched immediately when the dispatch event is
  74. // signaled.
  75. void UnblockDispatch() { dispatch_flag_ = nullptr; }
  76. // Called on IPC thread when a synchronous message or reply arrives.
  77. void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
  78. bool was_task_pending;
  79. {
  80. base::AutoLock auto_lock(message_lock_);
  81. was_task_pending = task_pending_;
  82. task_pending_ = true;
  83. // We set the event in case the listener thread is blocked (or is about
  84. // to). In case it's not, the PostTask dispatches the messages.
  85. message_queue_.push_back(QueuedMessage(new Message(msg), context));
  86. message_queue_version_++;
  87. }
  88. dispatch_event_.Signal();
  89. if (!was_task_pending) {
  90. listener_task_runner_->PostTask(
  91. FROM_HERE, base::BindOnce(&ReceivedSyncMsgQueue::DispatchMessagesTask,
  92. this, base::RetainedRef(context)));
  93. }
  94. }
  95. void QueueReply(const Message &msg, SyncChannel::SyncContext* context) {
  96. received_replies_.push_back(QueuedMessage(new Message(msg), context));
  97. }
  98. // Called on the listener's thread to process any queues synchronous
  99. // messages.
  100. void DispatchMessagesTask(SyncContext* context) {
  101. {
  102. base::AutoLock auto_lock(message_lock_);
  103. task_pending_ = false;
  104. }
  105. context->DispatchMessages();
  106. }
  107. // Dispatches any queued incoming sync messages. If |dispatching_context| is
  108. // not null, messages which target a restricted dispatch channel will only be
  109. // dispatched if |dispatching_context| belongs to the same restricted dispatch
  110. // group as that channel. If |dispatching_context| is null, all queued
  111. // messages are dispatched.
  112. void DispatchMessages(SyncContext* dispatching_context) {
  113. bool first_time = true;
  114. uint32_t expected_version = 0;
  115. SyncMessageQueue::iterator it;
  116. while (true) {
  117. Message* message = nullptr;
  118. scoped_refptr<SyncChannel::SyncContext> context;
  119. {
  120. base::AutoLock auto_lock(message_lock_);
  121. if (first_time || message_queue_version_ != expected_version) {
  122. it = message_queue_.begin();
  123. first_time = false;
  124. }
  125. for (; it != message_queue_.end(); it++) {
  126. int message_group = it->context->restrict_dispatch_group();
  127. if (message_group == kRestrictDispatchGroup_None ||
  128. (dispatching_context &&
  129. message_group ==
  130. dispatching_context->restrict_dispatch_group())) {
  131. message = it->message;
  132. context = it->context;
  133. it = message_queue_.erase(it);
  134. message_queue_version_++;
  135. expected_version = message_queue_version_;
  136. break;
  137. }
  138. }
  139. }
  140. if (message == nullptr)
  141. break;
  142. context->OnDispatchMessage(*message);
  143. delete message;
  144. }
  145. }
  146. // SyncChannel calls this in its destructor.
  147. void RemoveContext(SyncContext* context) {
  148. base::AutoLock auto_lock(message_lock_);
  149. SyncMessageQueue::iterator iter = message_queue_.begin();
  150. while (iter != message_queue_.end()) {
  151. if (iter->context.get() == context) {
  152. delete iter->message;
  153. iter = message_queue_.erase(iter);
  154. message_queue_version_++;
  155. } else {
  156. iter++;
  157. }
  158. }
  159. if (--listener_count_ == 0) {
  160. DCHECK(lazy_tls_ptr_.Pointer()->Get());
  161. lazy_tls_ptr_.Pointer()->Set(nullptr);
  162. sync_dispatch_watcher_.reset();
  163. }
  164. }
  165. base::WaitableEvent* dispatch_event() { return &dispatch_event_; }
  166. base::SingleThreadTaskRunner* listener_task_runner() {
  167. return listener_task_runner_.get();
  168. }
  169. // Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
  170. static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue>>::
  171. DestructorAtExit lazy_tls_ptr_;
  172. // Called on the ipc thread to check if we can unblock any current Send()
  173. // calls based on a queued reply.
  174. void DispatchReplies() {
  175. for (size_t i = 0; i < received_replies_.size(); ++i) {
  176. Message* message = received_replies_[i].message;
  177. if (received_replies_[i].context->TryToUnblockListener(message)) {
  178. delete message;
  179. received_replies_.erase(received_replies_.begin() + i);
  180. return;
  181. }
  182. }
  183. }
  184. private:
  185. friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
  186. // See the comment in SyncChannel::SyncChannel for why this event is created
  187. // as manual reset.
  188. ReceivedSyncMsgQueue()
  189. : message_queue_version_(0),
  190. dispatch_event_(base::WaitableEvent::ResetPolicy::MANUAL,
  191. base::WaitableEvent::InitialState::NOT_SIGNALED),
  192. listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
  193. sync_dispatch_watcher_(std::make_unique<mojo::SyncEventWatcher>(
  194. &dispatch_event_,
  195. base::BindRepeating(&ReceivedSyncMsgQueue::OnDispatchEventReady,
  196. base::Unretained(this)))) {
  197. sync_dispatch_watcher_->AllowWokenUpBySyncWatchOnSameThread();
  198. }
  199. ~ReceivedSyncMsgQueue() = default;
  200. void OnDispatchEventReady() {
  201. if (dispatch_flag_) {
  202. *dispatch_flag_ = true;
  203. return;
  204. }
  205. // We were woken up during a sync wait, but no specific SyncChannel is
  206. // currently waiting. i.e., some other Mojo interface on this thread is
  207. // waiting for a response. Since we don't support anything analogous to
  208. // restricted dispatch on Mojo interfaces, in this case it's safe to
  209. // dispatch sync messages for any context.
  210. DispatchMessages(nullptr);
  211. }
  212. // Holds information about a queued synchronous message or reply.
  213. struct QueuedMessage {
  214. QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
  215. raw_ptr<Message> message;
  216. scoped_refptr<SyncChannel::SyncContext> context;
  217. };
  218. typedef std::list<QueuedMessage> SyncMessageQueue;
  219. SyncMessageQueue message_queue_;
  220. // Used to signal DispatchMessages to rescan
  221. uint32_t message_queue_version_ = 0;
  222. std::vector<QueuedMessage> received_replies_;
  223. // Signaled when we get a synchronous message that we must respond to, as the
  224. // sender needs its reply before it can reply to our original synchronous
  225. // message.
  226. base::WaitableEvent dispatch_event_;
  227. scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
  228. base::Lock message_lock_;
  229. bool task_pending_ = false;
  230. int listener_count_ = 0;
  231. // If not null, the address of a flag to set when the dispatch event signals,
  232. // in lieu of actually dispatching messages. This is used by
  233. // SyncChannel::WaitForReply to restrict the scope of queued messages we're
  234. // allowed to process while it's waiting.
  235. raw_ptr<bool> dispatch_flag_ = nullptr;
  236. // Watches |dispatch_event_| during all sync handle watches on this thread.
  237. std::unique_ptr<mojo::SyncEventWatcher> sync_dispatch_watcher_;
  238. };
  239. base::LazyInstance<base::ThreadLocalPointer<
  240. SyncChannel::ReceivedSyncMsgQueue>>::DestructorAtExit
  241. SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
  242. LAZY_INSTANCE_INITIALIZER;
  243. SyncChannel::SyncContext::SyncContext(
  244. Listener* listener,
  245. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  246. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  247. WaitableEvent* shutdown_event)
  248. : ChannelProxy::Context(listener, ipc_task_runner, listener_task_runner),
  249. received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
  250. shutdown_event_(shutdown_event),
  251. restrict_dispatch_group_(kRestrictDispatchGroup_None) {}
  252. void SyncChannel::SyncContext::OnSendDoneEventSignaled(
  253. base::RunLoop* nested_loop,
  254. base::WaitableEvent* event) {
  255. DCHECK_EQ(GetSendDoneEvent(), event);
  256. nested_loop->Quit();
  257. }
  258. SyncChannel::SyncContext::~SyncContext() {
  259. while (!deserializers_.empty())
  260. Pop();
  261. }
  262. // Adds information about an outgoing sync message to the context so that
  263. // we know how to deserialize the reply. Returns |true| if the message was added
  264. // to the context or |false| if it was rejected (e.g. due to shutdown.)
  265. bool SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
  266. // Create the tracking information for this message. This object is stored
  267. // by value since all members are pointers that are cheap to copy. These
  268. // pointers are cleaned up in the Pop() function.
  269. //
  270. // The event is created as manual reset because in between Signal and
  271. // OnObjectSignalled, another Send can happen which would stop the watcher
  272. // from being called. The event would get watched later, when the nested
  273. // Send completes, so the event will need to remain set.
  274. base::AutoLock auto_lock(deserializers_lock_);
  275. if (reject_new_deserializers_)
  276. return false;
  277. PendingSyncMsg pending(
  278. SyncMessage::GetMessageId(*sync_msg), sync_msg->GetReplyDeserializer(),
  279. new base::WaitableEvent(base::WaitableEvent::ResetPolicy::MANUAL,
  280. base::WaitableEvent::InitialState::NOT_SIGNALED));
  281. deserializers_.push_back(pending);
  282. return true;
  283. }
  284. bool SyncChannel::SyncContext::Pop() {
  285. bool result;
  286. {
  287. base::AutoLock auto_lock(deserializers_lock_);
  288. PendingSyncMsg msg = deserializers_.back();
  289. delete msg.deserializer;
  290. delete msg.done_event;
  291. msg.done_event = nullptr;
  292. deserializers_.pop_back();
  293. result = msg.send_result;
  294. }
  295. // We got a reply to a synchronous Send() call that's blocking the listener
  296. // thread. However, further down the call stack there could be another
  297. // blocking Send() call, whose reply we received after we made this last
  298. // Send() call. So check if we have any queued replies available that
  299. // can now unblock the listener thread.
  300. ipc_task_runner()->PostTask(
  301. FROM_HERE, base::BindOnce(&ReceivedSyncMsgQueue::DispatchReplies,
  302. received_sync_msgs_));
  303. return result;
  304. }
  305. base::WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
  306. base::AutoLock auto_lock(deserializers_lock_);
  307. return deserializers_.back().done_event;
  308. }
  309. base::WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
  310. return received_sync_msgs_->dispatch_event();
  311. }
  312. void SyncChannel::SyncContext::DispatchMessages() {
  313. received_sync_msgs_->DispatchMessages(this);
  314. }
  315. bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
  316. base::AutoLock auto_lock(deserializers_lock_);
  317. if (deserializers_.empty() ||
  318. !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
  319. return false;
  320. }
  321. if (!msg->is_reply_error()) {
  322. bool send_result = deserializers_.back().deserializer->
  323. SerializeOutputParameters(*msg);
  324. deserializers_.back().send_result = send_result;
  325. DVLOG_IF(1, !send_result) << "Couldn't deserialize reply message";
  326. } else {
  327. DVLOG(1) << "Received error reply";
  328. }
  329. base::WaitableEvent* done_event = deserializers_.back().done_event;
  330. TRACE_EVENT_WITH_FLOW0("toplevel.flow",
  331. "SyncChannel::SyncContext::TryToUnblockListener",
  332. done_event, TRACE_EVENT_FLAG_FLOW_OUT);
  333. done_event->Signal();
  334. return true;
  335. }
  336. void SyncChannel::SyncContext::Clear() {
  337. CancelPendingSends();
  338. received_sync_msgs_->RemoveContext(this);
  339. Context::Clear();
  340. }
  341. bool SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
  342. // Give the filters a chance at processing this message.
  343. if (TryFilters(msg))
  344. return true;
  345. if (TryToUnblockListener(&msg))
  346. return true;
  347. if (msg.is_reply()) {
  348. received_sync_msgs_->QueueReply(msg, this);
  349. return true;
  350. }
  351. if (msg.should_unblock()) {
  352. received_sync_msgs_->QueueMessage(msg, this);
  353. return true;
  354. }
  355. return Context::OnMessageReceivedNoFilter(msg);
  356. }
  357. void SyncChannel::SyncContext::OnChannelError() {
  358. CancelPendingSends();
  359. shutdown_watcher_.StopWatching();
  360. Context::OnChannelError();
  361. }
  362. void SyncChannel::SyncContext::OnChannelOpened() {
  363. if (shutdown_event_) {
  364. shutdown_watcher_.StartWatching(
  365. shutdown_event_,
  366. base::BindOnce(&SyncChannel::SyncContext::OnShutdownEventSignaled,
  367. base::Unretained(this)),
  368. base::SequencedTaskRunnerHandle::Get());
  369. }
  370. Context::OnChannelOpened();
  371. }
  372. void SyncChannel::SyncContext::OnChannelClosed() {
  373. CancelPendingSends();
  374. shutdown_watcher_.StopWatching();
  375. Context::OnChannelClosed();
  376. }
  377. void SyncChannel::SyncContext::CancelPendingSends() {
  378. base::AutoLock auto_lock(deserializers_lock_);
  379. reject_new_deserializers_ = true;
  380. PendingSyncMessageQueue::iterator iter;
  381. DVLOG(1) << "Canceling pending sends";
  382. for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
  383. TRACE_EVENT_WITH_FLOW0("toplevel.flow",
  384. "SyncChannel::SyncContext::CancelPendingSends",
  385. iter->done_event, TRACE_EVENT_FLAG_FLOW_OUT);
  386. iter->done_event->Signal();
  387. }
  388. }
  389. void SyncChannel::SyncContext::OnShutdownEventSignaled(WaitableEvent* event) {
  390. DCHECK_EQ(event, shutdown_event_);
  391. // Process shut down before we can get a reply to a synchronous message.
  392. // Cancel pending Send calls, which will end up setting the send done event.
  393. CancelPendingSends();
  394. }
  395. // static
  396. std::unique_ptr<SyncChannel> SyncChannel::Create(
  397. const IPC::ChannelHandle& channel_handle,
  398. Channel::Mode mode,
  399. Listener* listener,
  400. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  401. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  402. bool create_pipe_now,
  403. base::WaitableEvent* shutdown_event) {
  404. // TODO(tobiasjs): The shutdown_event object is passed to a refcounted
  405. // Context object, and as a result it is not easy to ensure that it
  406. // outlives the Context. There should be some way to either reset
  407. // the shutdown_event when it is destroyed, or allow the Context to
  408. // control the lifetime of shutdown_event.
  409. std::unique_ptr<SyncChannel> channel =
  410. Create(listener, ipc_task_runner, listener_task_runner, shutdown_event);
  411. channel->Init(channel_handle, mode, create_pipe_now);
  412. return channel;
  413. }
  414. // static
  415. std::unique_ptr<SyncChannel> SyncChannel::Create(
  416. Listener* listener,
  417. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  418. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  419. WaitableEvent* shutdown_event) {
  420. return base::WrapUnique(new SyncChannel(
  421. listener, ipc_task_runner, listener_task_runner, shutdown_event));
  422. }
  423. SyncChannel::SyncChannel(
  424. Listener* listener,
  425. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  426. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  427. WaitableEvent* shutdown_event)
  428. : ChannelProxy(new SyncContext(listener,
  429. ipc_task_runner,
  430. listener_task_runner,
  431. shutdown_event)),
  432. sync_handle_registry_(mojo::SyncHandleRegistry::current()) {
  433. // The current (listener) thread must be distinct from the IPC thread, or else
  434. // sending synchronous messages will deadlock.
  435. DCHECK_NE(ipc_task_runner.get(), base::ThreadTaskRunnerHandle::Get().get());
  436. StartWatching();
  437. }
  438. void SyncChannel::AddListenerTaskRunner(
  439. int32_t routing_id,
  440. scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
  441. context()->AddListenerTaskRunner(routing_id, std::move(task_runner));
  442. }
  443. void SyncChannel::RemoveListenerTaskRunner(int32_t routing_id) {
  444. context()->RemoveListenerTaskRunner(routing_id);
  445. }
  446. SyncChannel::~SyncChannel() = default;
  447. void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
  448. sync_context()->set_restrict_dispatch_group(group);
  449. }
  450. scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() {
  451. scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter(
  452. sync_context()->shutdown_event());
  453. AddFilter(filter.get());
  454. if (!did_init())
  455. pre_init_sync_message_filters_.push_back(filter);
  456. return filter;
  457. }
  458. bool SyncChannel::Send(Message* message) {
  459. #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
  460. std::string name;
  461. Logging::GetInstance()->GetMessageText(
  462. message->type(), &name, message, nullptr);
  463. TRACE_EVENT1("ipc", "SyncChannel::Send", "name", name);
  464. #else
  465. TRACE_IPC_MESSAGE_SEND("ipc", "SyncChannel::Send", message);
  466. #endif
  467. if (!message->is_sync()) {
  468. ChannelProxy::SendInternal(message);
  469. return true;
  470. }
  471. SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
  472. // *this* might get deleted in WaitForReply.
  473. scoped_refptr<SyncContext> context(sync_context());
  474. if (!context->Push(sync_msg)) {
  475. DVLOG(1) << "Channel is shutting down. Dropping sync message.";
  476. delete message;
  477. return false;
  478. }
  479. ChannelProxy::SendInternal(message);
  480. // Wait for reply, or for any other incoming synchronous messages.
  481. // |this| might get deleted, so only call static functions at this point.
  482. scoped_refptr<mojo::SyncHandleRegistry> registry = sync_handle_registry_;
  483. WaitForReply(registry.get(), context.get());
  484. TRACE_EVENT_WITH_FLOW0("toplevel.flow", "SyncChannel::Send",
  485. context->GetSendDoneEvent(), TRACE_EVENT_FLAG_FLOW_IN);
  486. return context->Pop();
  487. }
  488. void SyncChannel::WaitForReply(mojo::SyncHandleRegistry* registry,
  489. SyncContext* context) {
  490. context->DispatchMessages();
  491. while (true) {
  492. bool dispatch = false;
  493. {
  494. bool send_done = false;
  495. mojo::SyncHandleRegistry::EventCallbackSubscription
  496. send_done_subscription = registry->RegisterEvent(
  497. context->GetSendDoneEvent(),
  498. base::BindRepeating(&OnEventReady, &send_done));
  499. const bool* stop_flags[] = {&dispatch, &send_done};
  500. context->received_sync_msgs()->BlockDispatch(&dispatch);
  501. registry->Wait(stop_flags, 2);
  502. context->received_sync_msgs()->UnblockDispatch();
  503. }
  504. if (dispatch) {
  505. // We're waiting for a reply, but we received a blocking synchronous call.
  506. // We must process it to avoid potential deadlocks.
  507. context->GetDispatchEvent()->Reset();
  508. context->DispatchMessages();
  509. continue;
  510. }
  511. break;
  512. }
  513. }
  514. void SyncChannel::OnDispatchEventSignaled(base::WaitableEvent* event) {
  515. DCHECK_EQ(sync_context()->GetDispatchEvent(), event);
  516. sync_context()->GetDispatchEvent()->Reset();
  517. StartWatching();
  518. // NOTE: May delete |this|.
  519. sync_context()->DispatchMessages();
  520. }
  521. void SyncChannel::StartWatching() {
  522. // |dispatch_watcher_| watches the event asynchronously, only dispatching
  523. // messages once the listener thread is unblocked and pumping its task queue.
  524. // The ReceivedSyncMsgQueue also watches this event and may dispatch
  525. // immediately if woken up by a message which it's allowed to dispatch.
  526. dispatch_watcher_.StartWatching(
  527. sync_context()->GetDispatchEvent(),
  528. base::BindOnce(&SyncChannel::OnDispatchEventSignaled,
  529. base::Unretained(this)),
  530. sync_context()->listener_task_runner());
  531. }
  532. void SyncChannel::OnChannelInit() {
  533. pre_init_sync_message_filters_.clear();
  534. }
  535. } // namespace IPC