ipc_sync_channel.h 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255
  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. #ifndef IPC_IPC_SYNC_CHANNEL_H_
  5. #define IPC_IPC_SYNC_CHANNEL_H_
  6. #include <memory>
  7. #include <string>
  8. #include <vector>
  9. #include "base/component_export.h"
  10. #include "base/containers/circular_deque.h"
  11. #include "base/memory/raw_ptr.h"
  12. #include "base/memory/ref_counted.h"
  13. #include "base/synchronization/lock.h"
  14. #include "base/synchronization/waitable_event_watcher.h"
  15. #include "ipc/ipc_channel_handle.h"
  16. #include "ipc/ipc_channel_proxy.h"
  17. #include "ipc/ipc_sync_message.h"
  18. #include "ipc/ipc_sync_message_filter.h"
  19. #include "mojo/public/c/system/types.h"
  20. #include "mojo/public/cpp/system/simple_watcher.h"
  21. namespace base {
  22. class RunLoop;
  23. class WaitableEvent;
  24. } // namespace base
  25. namespace mojo {
  26. class SyncHandleRegistry;
  27. }
  28. namespace IPC {
  29. class SyncMessage;
  30. // This is similar to ChannelProxy, with the added feature of supporting sending
  31. // synchronous messages.
  32. //
  33. // Overview of how the sync channel works
  34. // --------------------------------------
  35. // When the sending thread sends a synchronous message, we create a bunch
  36. // of tracking info (created in Send, stored in the PendingSyncMsg
  37. // structure) associated with the message that we identify by the unique
  38. // "MessageId" on the SyncMessage. Among the things we save is the
  39. // "Deserializer" which is provided by the sync message. This object is in
  40. // charge of reading the parameters from the reply message and putting them in
  41. // the output variables provided by its caller.
  42. //
  43. // The info gets stashed in a queue since we could have a nested stack of sync
  44. // messages (each side could send sync messages in response to sync messages,
  45. // so it works like calling a function). The message is sent to the I/O thread
  46. // for dispatch and the original thread blocks waiting for the reply.
  47. //
  48. // SyncContext maintains the queue in a threadsafe way and listens for replies
  49. // on the I/O thread. When a reply comes in that matches one of the messages
  50. // it's looking for (using the unique message ID), it will execute the
  51. // deserializer stashed from before, and unblock the original thread.
  52. //
  53. //
  54. // Significant complexity results from the fact that messages are still coming
  55. // in while the original thread is blocked. Normal async messages are queued
  56. // and dispatched after the blocking call is complete. Sync messages must
  57. // be dispatched in a reentrant manner to avoid deadlock.
  58. //
  59. //
  60. // Note that care must be taken that the lifetime of the ipc_thread argument
  61. // is more than this object. If the message loop goes away while this object
  62. // is running and it's used to send a message, then it will use the invalid
  63. // message loop pointer to proxy it to the ipc thread.
  64. class COMPONENT_EXPORT(IPC) SyncChannel : public ChannelProxy {
  65. public:
  66. enum RestrictDispatchGroup {
  67. kRestrictDispatchGroup_None = 0,
  68. };
  69. // Creates and initializes a sync channel. If create_pipe_now is specified,
  70. // the channel will be initialized synchronously.
  71. // The naming pattern follows IPC::Channel.
  72. static std::unique_ptr<SyncChannel> Create(
  73. const IPC::ChannelHandle& channel_handle,
  74. IPC::Channel::Mode mode,
  75. Listener* listener,
  76. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  77. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  78. bool create_pipe_now,
  79. base::WaitableEvent* shutdown_event);
  80. // Creates an uninitialized sync channel. Call ChannelProxy::Init to
  81. // initialize the channel. This two-step setup allows message filters to be
  82. // added before any messages are sent or received.
  83. static std::unique_ptr<SyncChannel> Create(
  84. Listener* listener,
  85. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  86. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  87. base::WaitableEvent* shutdown_event);
  88. void AddListenerTaskRunner(
  89. int32_t routing_id,
  90. scoped_refptr<base::SingleThreadTaskRunner> task_runner);
  91. void RemoveListenerTaskRunner(int32_t routing_id);
  92. SyncChannel(const SyncChannel&) = delete;
  93. SyncChannel& operator=(const SyncChannel&) = delete;
  94. ~SyncChannel() override;
  95. bool Send(Message* message) override;
  96. // Sets the dispatch group for this channel, to only allow re-entrant dispatch
  97. // of messages to other channels in the same group.
  98. //
  99. // Normally, any unblocking message coming from any channel can be dispatched
  100. // when any (possibly other) channel is blocked on sending a message. This is
  101. // needed in some cases to unblock certain loops (e.g. necessary when some
  102. // processes share a window hierarchy), but may cause re-entrancy issues in
  103. // some cases where such loops are not possible. This flags allows the tagging
  104. // of some particular channels to only re-enter in known correct cases.
  105. //
  106. // Incoming messages on channels belonging to a group that is not
  107. // kRestrictDispatchGroup_None will only be dispatched while a sync message is
  108. // being sent on a channel of the *same* group.
  109. // Incoming messages belonging to the kRestrictDispatchGroup_None group (the
  110. // default) will be dispatched in any case.
  111. void SetRestrictDispatchChannelGroup(int group);
  112. // Creates a new IPC::SyncMessageFilter and adds it to this SyncChannel.
  113. // This should be used instead of directly constructing a new
  114. // SyncMessageFilter.
  115. scoped_refptr<IPC::SyncMessageFilter> CreateSyncMessageFilter();
  116. protected:
  117. class ReceivedSyncMsgQueue;
  118. friend class ReceivedSyncMsgQueue;
  119. // SyncContext holds the per object data for SyncChannel, so that SyncChannel
  120. // can be deleted while it's being used in a different thread. See
  121. // ChannelProxy::Context for more information.
  122. class SyncContext : public Context {
  123. public:
  124. SyncContext(
  125. Listener* listener,
  126. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  127. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  128. base::WaitableEvent* shutdown_event);
  129. // Adds information about an outgoing sync message to the context so that
  130. // we know how to deserialize the reply.
  131. bool Push(SyncMessage* sync_msg);
  132. // Cleanly remove the top deserializer (and throw it away). Returns the
  133. // result of the Send call for that message.
  134. bool Pop();
  135. // Returns a Mojo Event that signals when a sync send is complete or timed
  136. // out or the process shut down.
  137. base::WaitableEvent* GetSendDoneEvent();
  138. // Returns a Mojo Event that signals when an incoming message that's not the
  139. // pending reply needs to get dispatched (by calling DispatchMessages.)
  140. base::WaitableEvent* GetDispatchEvent();
  141. void DispatchMessages();
  142. // Checks if the given message is blocking the listener thread because of a
  143. // synchronous send. If it is, the thread is unblocked and true is
  144. // returned. Otherwise the function returns false.
  145. bool TryToUnblockListener(const Message* msg);
  146. base::WaitableEvent* shutdown_event() { return shutdown_event_; }
  147. ReceivedSyncMsgQueue* received_sync_msgs() {
  148. return received_sync_msgs_.get();
  149. }
  150. void set_restrict_dispatch_group(int group) {
  151. restrict_dispatch_group_ = group;
  152. }
  153. int restrict_dispatch_group() const {
  154. return restrict_dispatch_group_;
  155. }
  156. void OnSendDoneEventSignaled(base::RunLoop* nested_loop,
  157. base::WaitableEvent* event);
  158. private:
  159. ~SyncContext() override;
  160. // ChannelProxy methods that we override.
  161. // Called on the listener thread.
  162. void Clear() override;
  163. // Called on the IPC thread.
  164. bool OnMessageReceived(const Message& msg) override;
  165. void OnChannelError() override;
  166. void OnChannelOpened() override;
  167. void OnChannelClosed() override;
  168. // Cancels all pending Send calls.
  169. void CancelPendingSends();
  170. void OnShutdownEventSignaled(base::WaitableEvent* event);
  171. using PendingSyncMessageQueue = base::circular_deque<PendingSyncMsg>;
  172. PendingSyncMessageQueue deserializers_;
  173. bool reject_new_deserializers_ = false;
  174. base::Lock deserializers_lock_;
  175. scoped_refptr<ReceivedSyncMsgQueue> received_sync_msgs_;
  176. raw_ptr<base::WaitableEvent> shutdown_event_;
  177. base::WaitableEventWatcher shutdown_watcher_;
  178. base::WaitableEventWatcher::EventCallback shutdown_watcher_callback_;
  179. int restrict_dispatch_group_;
  180. };
  181. private:
  182. SyncChannel(
  183. Listener* listener,
  184. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  185. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
  186. base::WaitableEvent* shutdown_event);
  187. void OnDispatchEventSignaled(base::WaitableEvent* event);
  188. SyncContext* sync_context() {
  189. return reinterpret_cast<SyncContext*>(context());
  190. }
  191. // Waits for a reply, timeout or process shutdown.
  192. static void WaitForReply(mojo::SyncHandleRegistry* registry,
  193. SyncContext* context);
  194. // Starts the dispatch watcher.
  195. void StartWatching();
  196. // ChannelProxy overrides:
  197. void OnChannelInit() override;
  198. scoped_refptr<mojo::SyncHandleRegistry> sync_handle_registry_;
  199. // Used to signal events between the IPC and listener threads.
  200. base::WaitableEventWatcher dispatch_watcher_;
  201. base::WaitableEventWatcher::EventCallback dispatch_watcher_callback_;
  202. // Tracks SyncMessageFilters created before complete channel initialization.
  203. std::vector<scoped_refptr<SyncMessageFilter>> pre_init_sync_message_filters_;
  204. };
  205. } // namespace IPC
  206. #endif // IPC_IPC_SYNC_CHANNEL_H_