ipc_channel_proxy.h 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449
  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_CHANNEL_PROXY_H_
  5. #define IPC_IPC_CHANNEL_PROXY_H_
  6. #include <stdint.h>
  7. #include <map>
  8. #include <memory>
  9. #include <string>
  10. #include <vector>
  11. #include "base/bind.h"
  12. #include "base/callback.h"
  13. #include "base/component_export.h"
  14. #include "base/memory/raw_ptr.h"
  15. #include "base/memory/ref_counted.h"
  16. #include "base/sequence_checker.h"
  17. #include "base/synchronization/lock.h"
  18. #include "build/build_config.h"
  19. #include "ipc/ipc.mojom.h"
  20. #include "ipc/ipc_channel.h"
  21. #include "ipc/ipc_channel_handle.h"
  22. #include "ipc/ipc_listener.h"
  23. #include "ipc/ipc_sender.h"
  24. #include "mojo/public/cpp/bindings/associated_remote.h"
  25. #include "mojo/public/cpp/bindings/generic_pending_associated_receiver.h"
  26. #include "mojo/public/cpp/bindings/lib/message_quota_checker.h"
  27. #include "mojo/public/cpp/bindings/pending_associated_receiver.h"
  28. #include "mojo/public/cpp/bindings/pending_associated_remote.h"
  29. #include "mojo/public/cpp/bindings/scoped_interface_endpoint_handle.h"
  30. #include "mojo/public/cpp/bindings/shared_associated_remote.h"
  31. namespace base {
  32. class SingleThreadTaskRunner;
  33. }
  34. namespace IPC {
  35. class ChannelFactory;
  36. class MessageFilter;
  37. class MessageFilterRouter;
  38. //-----------------------------------------------------------------------------
  39. // IPC::ChannelProxy
  40. //
  41. // This class is a helper class that is useful when you wish to run an IPC
  42. // channel on a background thread. It provides you with the option of either
  43. // handling IPC messages on that background thread or having them dispatched to
  44. // your main thread (the thread on which the IPC::ChannelProxy is created).
  45. //
  46. // The API for an IPC::ChannelProxy is very similar to that of an IPC::Channel.
  47. // When you send a message to an IPC::ChannelProxy, the message is routed to
  48. // the background thread, where it is then passed to the IPC::Channel's Send
  49. // method. This means that you can send a message from your thread and your
  50. // message will be sent over the IPC channel when possible instead of being
  51. // delayed until your thread returns to its message loop. (Often IPC messages
  52. // will queue up on the IPC::Channel when there is a lot of traffic, and the
  53. // channel will not get cycles to flush its message queue until the thread, on
  54. // which it is running, returns to its message loop.)
  55. //
  56. // An IPC::ChannelProxy can have a MessageFilter associated with it, which will
  57. // be notified of incoming messages on the IPC::Channel's thread. This gives
  58. // the consumer of IPC::ChannelProxy the ability to respond to incoming
  59. // messages on this background thread instead of on their own thread, which may
  60. // be bogged down with other processing. The result can be greatly improved
  61. // latency for messages that can be handled on a background thread.
  62. //
  63. // The consumer of IPC::ChannelProxy is responsible for allocating the Thread
  64. // instance where the IPC::Channel will be created and operated.
  65. //
  66. // Thread-safe send
  67. //
  68. // If a particular |Channel| implementation has a thread-safe |Send()| operation
  69. // then ChannelProxy skips the inter-thread hop and calls |Send()| directly. In
  70. // this case the |channel_| variable is touched by multiple threads so
  71. // |channel_lifetime_lock_| is used to protect it. The locking overhead is only
  72. // paid if the underlying channel supports thread-safe |Send|.
  73. //
  74. class COMPONENT_EXPORT(IPC) ChannelProxy : public Sender {
  75. public:
  76. #if defined(ENABLE_IPC_FUZZER)
  77. // Interface for a filter to be imposed on outgoing messages which can
  78. // re-write the message. Used for testing.
  79. class OutgoingMessageFilter {
  80. public:
  81. virtual Message* Rewrite(Message* message) = 0;
  82. };
  83. #endif
  84. // Initializes a channel proxy. The channel_handle and mode parameters are
  85. // passed directly to the underlying IPC::Channel. The listener is called on
  86. // the thread that creates the ChannelProxy. The filter's OnMessageReceived
  87. // method is called on the thread where the IPC::Channel is running. The
  88. // filter may be null if the consumer is not interested in handling messages
  89. // on the background thread. Any message not handled by the filter will be
  90. // dispatched to the listener. The given task runner correspond to a thread
  91. // on which IPC::Channel is created and used (e.g. IO thread).
  92. static std::unique_ptr<ChannelProxy> Create(
  93. const IPC::ChannelHandle& channel_handle,
  94. Channel::Mode mode,
  95. Listener* listener,
  96. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  97. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
  98. static std::unique_ptr<ChannelProxy> Create(
  99. std::unique_ptr<ChannelFactory> factory,
  100. Listener* listener,
  101. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  102. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
  103. // Constructs a ChannelProxy without initializing it.
  104. ChannelProxy(
  105. Listener* listener,
  106. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  107. const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
  108. ~ChannelProxy() override;
  109. // Initializes the channel proxy. Only call this once to initialize a channel
  110. // proxy that was not initialized in its constructor. If |create_pipe_now| is
  111. // true, the pipe is created synchronously. Otherwise it's created on the IO
  112. // thread.
  113. void Init(const IPC::ChannelHandle& channel_handle,
  114. Channel::Mode mode,
  115. bool create_pipe_now);
  116. void Init(std::unique_ptr<ChannelFactory> factory,
  117. bool create_pipe_now);
  118. // Pause the channel. Subsequent calls to Send() will be internally queued
  119. // until Unpause() is called. Queued messages will not be sent until the
  120. // channel is flushed.
  121. void Pause();
  122. // Unpause the channel. If |flush| is true the channel will be flushed as soon
  123. // as it's unpaused (see Flush() below.) Otherwise you must explicitly call
  124. // Flush() to flush messages which were queued while the channel was paused.
  125. void Unpause(bool flush);
  126. // Flush the channel. This sends any messages which were queued before calling
  127. // Connect. Only useful if Unpause(false) was called previously.
  128. void Flush();
  129. // Close the IPC::Channel. This operation completes asynchronously, once the
  130. // background thread processes the command to close the channel. It is ok to
  131. // call this method multiple times. Redundant calls are ignored.
  132. //
  133. // WARNING: MessageFilter objects held by the ChannelProxy is also
  134. // released asynchronously, and it may in fact have its final reference
  135. // released on the background thread. The caller should be careful to deal
  136. // with / allow for this possibility.
  137. void Close();
  138. // Send a message asynchronously. The message is routed to the background
  139. // thread where it is passed to the IPC::Channel's Send method.
  140. bool Send(Message* message) override;
  141. // Used to intercept messages as they are received on the background thread.
  142. //
  143. // Ordinarily, messages sent to the ChannelProxy are routed to the matching
  144. // listener on the worker thread. This API allows code to intercept messages
  145. // before they are sent to the worker thread.
  146. // If you call this before the target process is launched, then you're
  147. // guaranteed to not miss any messages. But if you call this anytime after,
  148. // then some messages might be missed since the filter is added internally on
  149. // the IO thread.
  150. void AddFilter(MessageFilter* filter);
  151. void RemoveFilter(MessageFilter* filter);
  152. using GenericAssociatedInterfaceFactory =
  153. base::RepeatingCallback<void(mojo::ScopedInterfaceEndpointHandle)>;
  154. // Adds a generic associated interface factory to bind incoming interface
  155. // requests directly on the IO thread. MUST be called either before Init() or
  156. // before the remote end of the Channel is able to send messages (e.g. before
  157. // its process is launched.)
  158. void AddGenericAssociatedInterfaceForIOThread(
  159. const std::string& name,
  160. const GenericAssociatedInterfaceFactory& factory);
  161. template <typename Interface>
  162. using AssociatedInterfaceFactory =
  163. base::RepeatingCallback<void(mojo::PendingAssociatedReceiver<Interface>)>;
  164. // Helper to bind an IO-thread associated interface factory, inferring the
  165. // interface name from the callback argument's type. MUST be called before
  166. // Init().
  167. template <typename Interface>
  168. void AddAssociatedInterfaceForIOThread(
  169. const AssociatedInterfaceFactory<Interface>& factory) {
  170. AddGenericAssociatedInterfaceForIOThread(
  171. Interface::Name_,
  172. base::BindRepeating(
  173. &ChannelProxy::BindPendingAssociatedReceiver<Interface>, factory));
  174. }
  175. // Requests an associated interface from the remote endpoint.
  176. void GetRemoteAssociatedInterface(
  177. mojo::GenericPendingAssociatedReceiver receiver);
  178. // Template helper to receive associated interfaces from the remote endpoint.
  179. template <typename Interface>
  180. void GetRemoteAssociatedInterface(mojo::AssociatedRemote<Interface>* proxy) {
  181. GetRemoteAssociatedInterface(proxy->BindNewEndpointAndPassReceiver());
  182. }
  183. #if defined(ENABLE_IPC_FUZZER)
  184. void set_outgoing_message_filter(OutgoingMessageFilter* filter) {
  185. outgoing_message_filter_ = filter;
  186. }
  187. #endif
  188. // Creates a SharedAssociatedRemote for |Interface|. This object may be used
  189. // to send messages on the interface from any thread and those messages will
  190. // remain ordered with respect to other messages sent on the same thread over
  191. // other SharedAssociatedRemotes associated with the same Channel.
  192. template <typename Interface>
  193. void GetThreadSafeRemoteAssociatedInterface(
  194. scoped_refptr<mojo::SharedAssociatedRemote<Interface>>* out_remote) {
  195. mojo::PendingAssociatedRemote<Interface> pending_remote;
  196. auto receiver = pending_remote.InitWithNewEndpointAndPassReceiver();
  197. GetGenericRemoteAssociatedInterface(Interface::Name_,
  198. receiver.PassHandle());
  199. *out_remote = mojo::SharedAssociatedRemote<Interface>::Create(
  200. std::move(pending_remote), ipc_task_runner());
  201. }
  202. base::SingleThreadTaskRunner* ipc_task_runner() const {
  203. return context_->ipc_task_runner();
  204. }
  205. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner_refptr()
  206. const {
  207. return context_->ipc_task_runner_refptr();
  208. }
  209. // Called to clear the pointer to the IPC task runner when it's going away.
  210. void ClearIPCTaskRunner();
  211. protected:
  212. class Context;
  213. // A subclass uses this constructor if it needs to add more information
  214. // to the internal state.
  215. explicit ChannelProxy(Context* context);
  216. // Used internally to hold state that is referenced on the IPC thread.
  217. class Context : public base::RefCountedThreadSafe<Context>,
  218. public Listener {
  219. public:
  220. Context(Listener* listener,
  221. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  222. const scoped_refptr<base::SingleThreadTaskRunner>&
  223. listener_task_runner);
  224. void ClearIPCTaskRunner();
  225. base::SingleThreadTaskRunner* ipc_task_runner() const {
  226. return ipc_task_runner_.get();
  227. }
  228. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner_refptr()
  229. const {
  230. return ipc_task_runner_;
  231. }
  232. scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner() {
  233. return default_listener_task_runner_;
  234. }
  235. // Dispatches a message on the listener thread.
  236. void OnDispatchMessage(const Message& message);
  237. // Sends |message| from appropriate thread.
  238. void Send(Message* message);
  239. // Adds |task_runner| for the task to be executed later.
  240. void AddListenerTaskRunner(
  241. int32_t routing_id,
  242. scoped_refptr<base::SingleThreadTaskRunner> task_runner);
  243. // Removes task runner for |routing_id|.
  244. void RemoveListenerTaskRunner(int32_t routing_id);
  245. // Called on the IPC::Channel thread.
  246. // Returns the task runner associated with |routing_id|.
  247. scoped_refptr<base::SingleThreadTaskRunner> GetTaskRunner(
  248. int32_t routing_id);
  249. protected:
  250. friend class base::RefCountedThreadSafe<Context>;
  251. ~Context() override;
  252. // IPC::Listener methods:
  253. bool OnMessageReceived(const Message& message) override;
  254. void OnChannelConnected(int32_t peer_pid) override;
  255. void OnChannelError() override;
  256. void OnAssociatedInterfaceRequest(
  257. const std::string& interface_name,
  258. mojo::ScopedInterfaceEndpointHandle handle) override;
  259. // Like OnMessageReceived but doesn't try the filters.
  260. bool OnMessageReceivedNoFilter(const Message& message);
  261. // Gives the filters a chance at processing |message|.
  262. // Returns true if the message was processed, false otherwise.
  263. bool TryFilters(const Message& message);
  264. void PauseChannel();
  265. void UnpauseChannel(bool flush);
  266. void FlushChannel();
  267. // Like Open and Close, but called on the IPC thread.
  268. virtual void OnChannelOpened();
  269. virtual void OnChannelClosed();
  270. // Called on the consumers thread when the ChannelProxy is closed. At that
  271. // point the consumer is telling us that they don't want to receive any
  272. // more messages, so we honor that wish by forgetting them!
  273. virtual void Clear();
  274. private:
  275. friend class ChannelProxy;
  276. friend class IpcSecurityTestUtil;
  277. // Create the Channel
  278. void CreateChannel(std::unique_ptr<ChannelFactory> factory);
  279. // Methods called on the IO thread.
  280. void OnSendMessage(std::unique_ptr<Message> message_ptr);
  281. void OnAddFilter();
  282. void OnRemoveFilter(MessageFilter* filter);
  283. // Methods called on the listener thread.
  284. void AddFilter(MessageFilter* filter);
  285. void OnDispatchConnected();
  286. void OnDispatchError();
  287. void OnDispatchBadMessage(const Message& message);
  288. void OnDispatchAssociatedInterfaceRequest(
  289. const std::string& interface_name,
  290. mojo::ScopedInterfaceEndpointHandle handle);
  291. void ClearChannel();
  292. mojom::Channel& thread_safe_channel() {
  293. return thread_safe_channel_->proxy();
  294. }
  295. void AddGenericAssociatedInterfaceForIOThread(
  296. const std::string& name,
  297. const GenericAssociatedInterfaceFactory& factory);
  298. base::Lock listener_thread_task_runners_lock_;
  299. // Map of routing_id and listener's thread task runner.
  300. std::map<int32_t, scoped_refptr<base::SingleThreadTaskRunner>>
  301. listener_thread_task_runners_
  302. GUARDED_BY(listener_thread_task_runners_lock_);
  303. scoped_refptr<base::SingleThreadTaskRunner> default_listener_task_runner_;
  304. raw_ptr<Listener> listener_;
  305. // List of filters. This is only accessed on the IPC thread.
  306. std::vector<scoped_refptr<MessageFilter> > filters_;
  307. scoped_refptr<base::SingleThreadTaskRunner> ipc_task_runner_;
  308. // Note, channel_ may be set on the Listener thread or the IPC thread.
  309. // But once it has been set, it must only be read or cleared on the IPC
  310. // thread.
  311. // One exception is the thread-safe send. See the class comment.
  312. std::unique_ptr<Channel> channel_;
  313. bool channel_connected_called_;
  314. // The quota checker associated with this channel, if any.
  315. scoped_refptr<mojo::internal::MessageQuotaChecker> quota_checker_;
  316. // Lock for |channel_| value. This is only relevant in the context of
  317. // thread-safe send.
  318. base::Lock channel_lifetime_lock_;
  319. // Routes a given message to a proper subset of |filters_|, depending
  320. // on which message classes a filter might support.
  321. std::unique_ptr<MessageFilterRouter> message_filter_router_;
  322. // Holds filters between the AddFilter call on the listerner thread and the
  323. // IPC thread when they're added to filters_.
  324. std::vector<scoped_refptr<MessageFilter> > pending_filters_;
  325. // Lock for pending_filters_.
  326. base::Lock pending_filters_lock_;
  327. // Cached copy of the peer process ID. Set on IPC but read on both IPC and
  328. // listener threads.
  329. base::ProcessId peer_pid_;
  330. base::Lock peer_pid_lock_;
  331. // A thread-safe mojom::Channel interface we use to make remote interface
  332. // requests from the proxy thread.
  333. std::unique_ptr<mojo::ThreadSafeForwarder<mojom::Channel>>
  334. thread_safe_channel_;
  335. // Holds associated interface binders added by
  336. // AddGenericAssociatedInterfaceForIOThread until the underlying channel has
  337. // been initialized.
  338. base::Lock pending_io_thread_interfaces_lock_;
  339. std::vector<std::pair<std::string, GenericAssociatedInterfaceFactory>>
  340. pending_io_thread_interfaces_;
  341. };
  342. Context* context() { return context_.get(); }
  343. #if defined(ENABLE_IPC_FUZZER)
  344. OutgoingMessageFilter* outgoing_message_filter() const {
  345. return outgoing_message_filter_;
  346. }
  347. #endif
  348. bool did_init() const { return did_init_; }
  349. // A Send() which doesn't DCHECK if the message is synchronous.
  350. void SendInternal(Message* message);
  351. private:
  352. friend class IpcSecurityTestUtil;
  353. template <typename Interface>
  354. static void BindPendingAssociatedReceiver(
  355. const AssociatedInterfaceFactory<Interface>& factory,
  356. mojo::ScopedInterfaceEndpointHandle handle) {
  357. factory.Run(mojo::PendingAssociatedReceiver<Interface>(std::move(handle)));
  358. }
  359. // Always called once immediately after Init.
  360. virtual void OnChannelInit();
  361. // By maintaining this indirection (ref-counted) to our internal state, we
  362. // can safely be destroyed while the background thread continues to do stuff
  363. // that involves this data.
  364. scoped_refptr<Context> context_;
  365. // Whether the channel has been initialized.
  366. bool did_init_;
  367. #if defined(ENABLE_IPC_FUZZER)
  368. OutgoingMessageFilter* outgoing_message_filter_;
  369. #endif
  370. SEQUENCE_CHECKER(sequence_checker_);
  371. };
  372. } // namespace IPC
  373. #endif // IPC_IPC_CHANNEL_PROXY_H_