ipc_channel_mojo.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405
  1. // Copyright 2014 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_channel_mojo.h"
  5. #include <stddef.h>
  6. #include <stdint.h>
  7. #include <memory>
  8. #include <utility>
  9. #include "base/bind.h"
  10. #include "base/callback_helpers.h"
  11. #include "base/command_line.h"
  12. #include "base/lazy_instance.h"
  13. #include "base/memory/ptr_util.h"
  14. #include "base/process/process_handle.h"
  15. #include "base/threading/thread_task_runner_handle.h"
  16. #include "build/build_config.h"
  17. #include "ipc/ipc_listener.h"
  18. #include "ipc/ipc_logging.h"
  19. #include "ipc/ipc_message_attachment_set.h"
  20. #include "ipc/ipc_message_macros.h"
  21. #include "ipc/ipc_mojo_bootstrap.h"
  22. #include "ipc/ipc_mojo_handle_attachment.h"
  23. #include "ipc/native_handle_type_converters.h"
  24. #include "ipc/trace_ipc_message.h"
  25. #include "mojo/public/cpp/bindings/associated_receiver.h"
  26. #include "mojo/public/cpp/bindings/associated_remote.h"
  27. #include "mojo/public/cpp/bindings/generic_pending_associated_receiver.h"
  28. #include "mojo/public/cpp/bindings/lib/message_quota_checker.h"
  29. #include "mojo/public/cpp/bindings/pending_associated_receiver.h"
  30. #include "mojo/public/cpp/bindings/thread_safe_proxy.h"
  31. #include "mojo/public/cpp/system/platform_handle.h"
  32. namespace IPC {
  33. namespace {
  34. class MojoChannelFactory : public ChannelFactory {
  35. public:
  36. MojoChannelFactory(
  37. mojo::ScopedMessagePipeHandle handle,
  38. Channel::Mode mode,
  39. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  40. const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner)
  41. : handle_(std::move(handle)),
  42. mode_(mode),
  43. ipc_task_runner_(ipc_task_runner),
  44. proxy_task_runner_(proxy_task_runner),
  45. quota_checker_(mojo::internal::MessageQuotaChecker::MaybeCreate()) {}
  46. MojoChannelFactory(const MojoChannelFactory&) = delete;
  47. MojoChannelFactory& operator=(const MojoChannelFactory&) = delete;
  48. std::unique_ptr<Channel> BuildChannel(Listener* listener) override {
  49. return ChannelMojo::Create(std::move(handle_), mode_, listener,
  50. ipc_task_runner_, proxy_task_runner_,
  51. quota_checker_);
  52. }
  53. scoped_refptr<base::SingleThreadTaskRunner> GetIPCTaskRunner() override {
  54. return ipc_task_runner_;
  55. }
  56. scoped_refptr<mojo::internal::MessageQuotaChecker> GetQuotaChecker()
  57. override {
  58. return quota_checker_;
  59. }
  60. private:
  61. mojo::ScopedMessagePipeHandle handle_;
  62. const Channel::Mode mode_;
  63. scoped_refptr<base::SingleThreadTaskRunner> ipc_task_runner_;
  64. scoped_refptr<base::SingleThreadTaskRunner> proxy_task_runner_;
  65. scoped_refptr<mojo::internal::MessageQuotaChecker> quota_checker_;
  66. };
  67. class ThreadSafeChannelProxy : public mojo::ThreadSafeProxy {
  68. public:
  69. using Forwarder = base::RepeatingCallback<void(mojo::Message)>;
  70. ThreadSafeChannelProxy(
  71. scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  72. Forwarder forwarder,
  73. mojo::AssociatedGroupController& group_controller)
  74. : task_runner_(std::move(task_runner)),
  75. forwarder_(std::move(forwarder)),
  76. group_controller_(group_controller) {}
  77. // mojo::ThreadSafeProxy:
  78. void SendMessage(mojo::Message& message) override {
  79. message.SerializeHandles(&group_controller_);
  80. task_runner_->PostTask(FROM_HERE,
  81. base::BindOnce(forwarder_, std::move(message)));
  82. }
  83. void SendMessageWithResponder(
  84. mojo::Message& message,
  85. std::unique_ptr<mojo::MessageReceiver> responder) override {
  86. // We don't bother supporting this because it's not used in practice.
  87. NOTREACHED();
  88. }
  89. private:
  90. ~ThreadSafeChannelProxy() override = default;
  91. const scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
  92. const Forwarder forwarder_;
  93. mojo::AssociatedGroupController& group_controller_;
  94. };
  95. base::ProcessId GetSelfPID() {
  96. #if BUILDFLAG(IS_LINUX) || BUILDFLAG(IS_CHROMEOS)
  97. if (int global_pid = Channel::GetGlobalPid())
  98. return global_pid;
  99. #endif // BUILDFLAG(IS_LINUX) || BUILDFLAG(IS_CHROMEOS)
  100. #if BUILDFLAG(IS_NACL)
  101. return -1;
  102. #else
  103. return base::GetCurrentProcId();
  104. #endif // BUILDFLAG(IS_NACL)
  105. }
  106. } // namespace
  107. //------------------------------------------------------------------------------
  108. // static
  109. std::unique_ptr<ChannelMojo> ChannelMojo::Create(
  110. mojo::ScopedMessagePipeHandle handle,
  111. Mode mode,
  112. Listener* listener,
  113. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  114. const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner,
  115. const scoped_refptr<mojo::internal::MessageQuotaChecker>& quota_checker) {
  116. return base::WrapUnique(new ChannelMojo(std::move(handle), mode, listener,
  117. ipc_task_runner, proxy_task_runner,
  118. quota_checker));
  119. }
  120. // static
  121. std::unique_ptr<ChannelFactory> ChannelMojo::CreateServerFactory(
  122. mojo::ScopedMessagePipeHandle handle,
  123. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  124. const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner) {
  125. return std::make_unique<MojoChannelFactory>(
  126. std::move(handle), Channel::MODE_SERVER, ipc_task_runner,
  127. proxy_task_runner);
  128. }
  129. // static
  130. std::unique_ptr<ChannelFactory> ChannelMojo::CreateClientFactory(
  131. mojo::ScopedMessagePipeHandle handle,
  132. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  133. const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner) {
  134. return std::make_unique<MojoChannelFactory>(
  135. std::move(handle), Channel::MODE_CLIENT, ipc_task_runner,
  136. proxy_task_runner);
  137. }
  138. ChannelMojo::ChannelMojo(
  139. mojo::ScopedMessagePipeHandle handle,
  140. Mode mode,
  141. Listener* listener,
  142. const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  143. const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner,
  144. const scoped_refptr<mojo::internal::MessageQuotaChecker>& quota_checker)
  145. : task_runner_(ipc_task_runner), pipe_(handle.get()), listener_(listener) {
  146. weak_ptr_ = weak_factory_.GetWeakPtr();
  147. bootstrap_ = MojoBootstrap::Create(std::move(handle), mode, ipc_task_runner,
  148. proxy_task_runner, quota_checker);
  149. }
  150. void ChannelMojo::ForwardMessage(mojo::Message message) {
  151. DCHECK(task_runner_->RunsTasksInCurrentSequence());
  152. if (!message_reader_ || !message_reader_->sender().is_bound())
  153. return;
  154. message_reader_->sender().internal_state()->ForwardMessage(
  155. std::move(message));
  156. }
  157. ChannelMojo::~ChannelMojo() {
  158. DCHECK(task_runner_->RunsTasksInCurrentSequence());
  159. Close();
  160. }
  161. bool ChannelMojo::Connect() {
  162. WillConnect();
  163. mojo::PendingAssociatedRemote<mojom::Channel> sender;
  164. mojo::PendingAssociatedReceiver<mojom::Channel> receiver;
  165. bootstrap_->Connect(&sender, &receiver);
  166. DCHECK(!message_reader_);
  167. message_reader_ = std::make_unique<internal::MessagePipeReader>(
  168. pipe_, std::move(sender), std::move(receiver), task_runner_, this);
  169. if (task_runner_->RunsTasksInCurrentSequence()) {
  170. FinishConnectOnIOThread();
  171. } else {
  172. task_runner_->PostTask(
  173. FROM_HERE,
  174. base::BindOnce(&ChannelMojo::FinishConnectOnIOThread, weak_ptr_));
  175. }
  176. return true;
  177. }
  178. void ChannelMojo::FinishConnectOnIOThread() {
  179. DCHECK(message_reader_);
  180. message_reader_->FinishInitializationOnIOThread(GetSelfPID());
  181. bootstrap_->StartReceiving();
  182. }
  183. void ChannelMojo::Pause() {
  184. bootstrap_->Pause();
  185. }
  186. void ChannelMojo::Unpause(bool flush) {
  187. bootstrap_->Unpause();
  188. if (flush)
  189. Flush();
  190. }
  191. void ChannelMojo::Flush() {
  192. bootstrap_->Flush();
  193. }
  194. void ChannelMojo::Close() {
  195. // NOTE: The MessagePipeReader's destructor may re-enter this function. Use
  196. // caution when changing this method.
  197. std::unique_ptr<internal::MessagePipeReader> reader =
  198. std::move(message_reader_);
  199. reader.reset();
  200. base::AutoLock lock(associated_interface_lock_);
  201. associated_interfaces_.clear();
  202. }
  203. void ChannelMojo::OnPipeError() {
  204. DCHECK(task_runner_);
  205. if (task_runner_->RunsTasksInCurrentSequence()) {
  206. listener_->OnChannelError();
  207. } else {
  208. task_runner_->PostTask(
  209. FROM_HERE, base::BindOnce(&ChannelMojo::OnPipeError, weak_ptr_));
  210. }
  211. }
  212. void ChannelMojo::OnAssociatedInterfaceRequest(
  213. mojo::GenericPendingAssociatedReceiver receiver) {
  214. GenericAssociatedInterfaceFactory factory;
  215. {
  216. base::AutoLock locker(associated_interface_lock_);
  217. auto iter = associated_interfaces_.find(*receiver.interface_name());
  218. if (iter != associated_interfaces_.end())
  219. factory = iter->second;
  220. }
  221. if (!factory.is_null()) {
  222. factory.Run(receiver.PassHandle());
  223. } else {
  224. const std::string interface_name = *receiver.interface_name();
  225. listener_->OnAssociatedInterfaceRequest(interface_name,
  226. receiver.PassHandle());
  227. }
  228. }
  229. bool ChannelMojo::Send(Message* message) {
  230. DVLOG(2) << "sending message @" << message << " on channel @" << this
  231. << " with type " << message->type();
  232. #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
  233. Logging::GetInstance()->OnSendMessage(message);
  234. #endif
  235. std::unique_ptr<Message> scoped_message = base::WrapUnique(message);
  236. if (!message_reader_)
  237. return false;
  238. // Comment copied from ipc_channel_posix.cc:
  239. // We can't close the pipe here, because calling OnChannelError may destroy
  240. // this object, and that would be bad if we are called from Send(). Instead,
  241. // we return false and hope the caller will close the pipe. If they do not,
  242. // the pipe will still be closed next time OnFileCanReadWithoutBlocking is
  243. // called.
  244. //
  245. // With Mojo, there's no OnFileCanReadWithoutBlocking, but we expect the
  246. // pipe's connection error handler will be invoked in its place.
  247. return message_reader_->Send(std::move(scoped_message));
  248. }
  249. Channel::AssociatedInterfaceSupport*
  250. ChannelMojo::GetAssociatedInterfaceSupport() { return this; }
  251. std::unique_ptr<mojo::ThreadSafeForwarder<mojom::Channel>>
  252. ChannelMojo::CreateThreadSafeChannel() {
  253. return std::make_unique<mojo::ThreadSafeForwarder<mojom::Channel>>(
  254. base::MakeRefCounted<ThreadSafeChannelProxy>(
  255. task_runner_,
  256. base::BindRepeating(&ChannelMojo::ForwardMessage, weak_ptr_),
  257. *bootstrap_->GetAssociatedGroup()->GetController()));
  258. }
  259. void ChannelMojo::OnPeerPidReceived(int32_t peer_pid) {
  260. listener_->OnChannelConnected(peer_pid);
  261. }
  262. void ChannelMojo::OnMessageReceived(const Message& message) {
  263. const Message* message_ptr = &message;
  264. TRACE_IPC_MESSAGE_SEND("ipc,toplevel", "ChannelMojo::OnMessageReceived",
  265. message_ptr);
  266. listener_->OnMessageReceived(message);
  267. if (message.dispatch_error())
  268. listener_->OnBadMessageReceived(message);
  269. }
  270. void ChannelMojo::OnBrokenDataReceived() {
  271. listener_->OnBadMessageReceived(Message());
  272. }
  273. // static
  274. MojoResult ChannelMojo::ReadFromMessageAttachmentSet(
  275. Message* message,
  276. absl::optional<std::vector<mojo::native::SerializedHandlePtr>>* handles) {
  277. DCHECK(!*handles);
  278. MojoResult result = MOJO_RESULT_OK;
  279. if (!message->HasAttachments())
  280. return result;
  281. std::vector<mojo::native::SerializedHandlePtr> output_handles;
  282. MessageAttachmentSet* set = message->attachment_set();
  283. for (unsigned i = 0; result == MOJO_RESULT_OK && i < set->size(); ++i) {
  284. auto attachment = set->GetAttachmentAt(i);
  285. auto serialized_handle = mojo::native::SerializedHandle::New();
  286. serialized_handle->the_handle = attachment->TakeMojoHandle();
  287. serialized_handle->type =
  288. mojo::ConvertTo<mojo::native::SerializedHandleType>(
  289. attachment->GetType());
  290. output_handles.emplace_back(std::move(serialized_handle));
  291. }
  292. set->CommitAllDescriptors();
  293. if (!output_handles.empty())
  294. *handles = std::move(output_handles);
  295. return result;
  296. }
  297. // static
  298. MojoResult ChannelMojo::WriteToMessageAttachmentSet(
  299. absl::optional<std::vector<mojo::native::SerializedHandlePtr>> handles,
  300. Message* message) {
  301. if (!handles)
  302. return MOJO_RESULT_OK;
  303. for (size_t i = 0; i < handles->size(); ++i) {
  304. auto& handle = handles->at(i);
  305. scoped_refptr<MessageAttachment> unwrapped_attachment =
  306. MessageAttachment::CreateFromMojoHandle(
  307. std::move(handle->the_handle),
  308. mojo::ConvertTo<MessageAttachment::Type>(handle->type));
  309. if (!unwrapped_attachment) {
  310. DLOG(WARNING) << "Pipe failed to unwrap handles.";
  311. return MOJO_RESULT_UNKNOWN;
  312. }
  313. bool ok = message->attachment_set()->AddAttachment(
  314. std::move(unwrapped_attachment));
  315. DCHECK(ok);
  316. if (!ok) {
  317. LOG(ERROR) << "Failed to add new Mojo handle.";
  318. return MOJO_RESULT_UNKNOWN;
  319. }
  320. }
  321. return MOJO_RESULT_OK;
  322. }
  323. void ChannelMojo::AddGenericAssociatedInterface(
  324. const std::string& name,
  325. const GenericAssociatedInterfaceFactory& factory) {
  326. base::AutoLock locker(associated_interface_lock_);
  327. auto result = associated_interfaces_.insert({ name, factory });
  328. DCHECK(result.second);
  329. }
  330. void ChannelMojo::GetRemoteAssociatedInterface(
  331. mojo::GenericPendingAssociatedReceiver receiver) {
  332. if (message_reader_) {
  333. if (!task_runner_->RunsTasksInCurrentSequence()) {
  334. message_reader_->thread_safe_sender().GetAssociatedInterface(
  335. std::move(receiver));
  336. return;
  337. }
  338. message_reader_->GetRemoteInterface(std::move(receiver));
  339. } else {
  340. // Attach the associated interface to a disconnected pipe, so that the
  341. // associated interface pointer can be used to make calls (which are
  342. // dropped).
  343. mojo::AssociateWithDisconnectedPipe(receiver.PassHandle());
  344. }
  345. }
  346. } // namespace IPC