message_pump_kqueue_unittest.cc 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214
  1. // Copyright 2019 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 "base/message_loop/message_pump_kqueue.h"
  5. #include <mach/mach.h>
  6. #include <mach/message.h>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/memory/ptr_util.h"
  10. #include "base/memory/raw_ptr.h"
  11. #include "base/memory/scoped_refptr.h"
  12. #include "base/run_loop.h"
  13. #include "base/task/single_thread_task_executor.h"
  14. #include "base/task/single_thread_task_runner.h"
  15. #include "base/threading/thread_task_runner_handle.h"
  16. #include "testing/gtest/include/gtest/gtest.h"
  17. namespace base {
  18. namespace {
  19. class MessagePumpKqueueTest : public testing::Test {
  20. public:
  21. MessagePumpKqueueTest()
  22. : pump_(new MessagePumpKqueue()), executor_(WrapUnique(pump_.get())) {}
  23. MessagePumpKqueue* pump() { return pump_; }
  24. static void CreatePortPair(mac::ScopedMachReceiveRight* receive,
  25. mac::ScopedMachSendRight* send) {
  26. mach_port_options_t options{};
  27. options.flags = MPO_INSERT_SEND_RIGHT;
  28. mac::ScopedMachReceiveRight port;
  29. kern_return_t kr = mach_port_construct(
  30. mach_task_self(), &options, 0,
  31. mac::ScopedMachReceiveRight::Receiver(*receive).get());
  32. ASSERT_EQ(kr, KERN_SUCCESS);
  33. *send = mac::ScopedMachSendRight(receive->get());
  34. }
  35. static mach_msg_return_t SendEmptyMessage(mach_port_t remote_port,
  36. mach_msg_id_t msgid) {
  37. mach_msg_empty_send_t message{};
  38. message.header.msgh_bits = MACH_MSGH_BITS_REMOTE(MACH_MSG_TYPE_COPY_SEND);
  39. message.header.msgh_size = sizeof(message);
  40. message.header.msgh_remote_port = remote_port;
  41. message.header.msgh_id = msgid;
  42. return mach_msg_send(&message.header);
  43. }
  44. private:
  45. raw_ptr<MessagePumpKqueue> pump_; // Weak, owned by |executor_|.
  46. SingleThreadTaskExecutor executor_;
  47. };
  48. class PortWatcher : public MessagePumpKqueue::MachPortWatcher {
  49. public:
  50. PortWatcher(RepeatingClosure callback) : callback_(std::move(callback)) {}
  51. ~PortWatcher() override {}
  52. void OnMachMessageReceived(mach_port_t port) override {
  53. mach_msg_empty_rcv_t message{};
  54. kern_return_t kr = mach_msg(&message.header, MACH_RCV_MSG, 0,
  55. sizeof(message), port, 0, MACH_PORT_NULL);
  56. ASSERT_EQ(kr, KERN_SUCCESS);
  57. messages_.push_back(message.header);
  58. callback_.Run();
  59. }
  60. std::vector<mach_msg_header_t> messages_;
  61. private:
  62. RepeatingClosure callback_;
  63. };
  64. TEST_F(MessagePumpKqueueTest, MachPortBasicWatch) {
  65. mac::ScopedMachReceiveRight port;
  66. mac::ScopedMachSendRight send_right;
  67. CreatePortPair(&port, &send_right);
  68. mach_msg_id_t msgid = 'helo';
  69. RunLoop run_loop;
  70. PortWatcher watcher(run_loop.QuitClosure());
  71. MessagePumpKqueue::MachPortWatchController controller(FROM_HERE);
  72. ThreadTaskRunnerHandle::Get()->PostTask(
  73. FROM_HERE, BindOnce(
  74. [](mach_port_t port, mach_msg_id_t msgid, RunLoop* loop) {
  75. mach_msg_return_t kr = SendEmptyMessage(port, msgid);
  76. EXPECT_EQ(kr, KERN_SUCCESS);
  77. if (kr != KERN_SUCCESS) {
  78. loop->Quit();
  79. }
  80. },
  81. port.get(), msgid, Unretained(&run_loop)));
  82. pump()->WatchMachReceivePort(port.get(), &controller, &watcher);
  83. run_loop.Run();
  84. ASSERT_EQ(1u, watcher.messages_.size());
  85. EXPECT_EQ(port.get(), watcher.messages_[0].msgh_local_port);
  86. EXPECT_EQ(msgid, watcher.messages_[0].msgh_id);
  87. }
  88. TEST_F(MessagePumpKqueueTest, MachPortStopWatching) {
  89. mac::ScopedMachReceiveRight port;
  90. mac::ScopedMachSendRight send_right;
  91. CreatePortPair(&port, &send_right);
  92. RunLoop run_loop;
  93. PortWatcher watcher(run_loop.QuitClosure());
  94. MessagePumpKqueue::MachPortWatchController controller(FROM_HERE);
  95. pump()->WatchMachReceivePort(port.get(), &controller, &watcher);
  96. ThreadTaskRunnerHandle::Get()->PostTask(
  97. FROM_HERE,
  98. BindOnce(
  99. [](MessagePumpKqueue::MachPortWatchController* controller) {
  100. controller->StopWatchingMachPort();
  101. },
  102. Unretained(&controller)));
  103. ThreadTaskRunnerHandle::Get()->PostTask(
  104. FROM_HERE, BindOnce(
  105. [](mach_port_t port) {
  106. EXPECT_EQ(KERN_SUCCESS, SendEmptyMessage(port, 100));
  107. },
  108. port.get()));
  109. run_loop.RunUntilIdle();
  110. EXPECT_EQ(0u, watcher.messages_.size());
  111. }
  112. TEST_F(MessagePumpKqueueTest, MultipleMachWatchers) {
  113. mac::ScopedMachReceiveRight port1, port2;
  114. mac::ScopedMachSendRight send_right1, send_right2;
  115. CreatePortPair(&port1, &send_right1);
  116. CreatePortPair(&port2, &send_right2);
  117. RunLoop run_loop;
  118. int port1_count = 0, port2_count = 0;
  119. // Whenever port1 receives a message, it will send to port2.
  120. // Whenever port2 receives a message, it will send to port1.
  121. // When port2 has sent 3 messages to port1, it will stop.
  122. PortWatcher watcher1(BindRepeating(
  123. [](mach_port_t port2, int* port2_count, RunLoop* loop) {
  124. mach_msg_id_t id = (0x2 << 16) | ++(*port2_count);
  125. mach_msg_return_t kr = SendEmptyMessage(port2, id);
  126. EXPECT_EQ(kr, KERN_SUCCESS);
  127. if (kr != KERN_SUCCESS) {
  128. loop->Quit();
  129. }
  130. },
  131. port2.get(), &port2_count, &run_loop));
  132. MessagePumpKqueue::MachPortWatchController controller1(FROM_HERE);
  133. PortWatcher watcher2(BindRepeating(
  134. [](mach_port_t port1, int* port1_count, RunLoop* loop) {
  135. if (*port1_count == 3) {
  136. loop->Quit();
  137. return;
  138. }
  139. mach_msg_id_t id = (0x1 << 16) | ++(*port1_count);
  140. mach_msg_return_t kr = SendEmptyMessage(port1, id);
  141. EXPECT_EQ(kr, KERN_SUCCESS);
  142. if (kr != KERN_SUCCESS) {
  143. loop->Quit();
  144. }
  145. },
  146. port1.get(), &port1_count, &run_loop));
  147. MessagePumpKqueue::MachPortWatchController controller2(FROM_HERE);
  148. pump()->WatchMachReceivePort(port1.get(), &controller1, &watcher1);
  149. pump()->WatchMachReceivePort(port2.get(), &controller2, &watcher2);
  150. // Start ping-ponging with by sending the first message to port1.
  151. ThreadTaskRunnerHandle::Get()->PostTask(
  152. FROM_HERE, BindOnce(
  153. [](mach_port_t port1) {
  154. ASSERT_EQ(KERN_SUCCESS,
  155. SendEmptyMessage(port1, 0xf000f));
  156. },
  157. port1.get()));
  158. run_loop.Run();
  159. ASSERT_EQ(4u, watcher1.messages_.size());
  160. ASSERT_EQ(4u, watcher2.messages_.size());
  161. EXPECT_EQ(0xf000f, watcher1.messages_[0].msgh_id);
  162. EXPECT_EQ(0x10001, watcher1.messages_[1].msgh_id);
  163. EXPECT_EQ(0x10002, watcher1.messages_[2].msgh_id);
  164. EXPECT_EQ(0x10003, watcher1.messages_[3].msgh_id);
  165. EXPECT_EQ(0x20001, watcher2.messages_[0].msgh_id);
  166. EXPECT_EQ(0x20002, watcher2.messages_[1].msgh_id);
  167. EXPECT_EQ(0x20003, watcher2.messages_[2].msgh_id);
  168. EXPECT_EQ(0x20004, watcher2.messages_[3].msgh_id);
  169. }
  170. } // namespace
  171. } // namespace base