waitable_event_mac.cc 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277
  1. // Copyright 2017 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/synchronization/waitable_event.h"
  5. #include <mach/mach.h>
  6. #include <sys/event.h>
  7. #include <limits>
  8. #include <memory>
  9. #include "base/debug/activity_tracker.h"
  10. #include "base/files/scoped_file.h"
  11. #include "base/mac/mach_logging.h"
  12. #include "base/posix/eintr_wrapper.h"
  13. #include "base/threading/scoped_blocking_call.h"
  14. #include "base/threading/thread_restrictions.h"
  15. #include "base/time/time.h"
  16. #include "base/time/time_override.h"
  17. #include "build/build_config.h"
  18. #include "third_party/abseil-cpp/absl/types/optional.h"
  19. namespace base {
  20. WaitableEvent::WaitableEvent(ResetPolicy reset_policy,
  21. InitialState initial_state)
  22. : policy_(reset_policy) {
  23. mach_port_options_t options{};
  24. options.flags = MPO_INSERT_SEND_RIGHT;
  25. options.mpl.mpl_qlimit = 1;
  26. mach_port_t name;
  27. kern_return_t kr = mach_port_construct(mach_task_self(), &options, 0, &name);
  28. MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_construct";
  29. receive_right_ = new ReceiveRight(name);
  30. send_right_.reset(name);
  31. if (initial_state == InitialState::SIGNALED)
  32. Signal();
  33. }
  34. WaitableEvent::~WaitableEvent() = default;
  35. void WaitableEvent::Reset() {
  36. PeekPort(receive_right_->Name(), true);
  37. }
  38. void WaitableEvent::Signal() {
  39. mach_msg_empty_send_t msg{};
  40. msg.header.msgh_bits = MACH_MSGH_BITS_REMOTE(MACH_MSG_TYPE_COPY_SEND);
  41. msg.header.msgh_size = sizeof(&msg);
  42. msg.header.msgh_remote_port = send_right_.get();
  43. // If the event is already signaled, this will time out because the queue
  44. // has a length of one.
  45. kern_return_t kr =
  46. mach_msg(&msg.header, MACH_SEND_MSG | MACH_SEND_TIMEOUT, sizeof(msg), 0,
  47. MACH_PORT_NULL, 0, MACH_PORT_NULL);
  48. MACH_CHECK(kr == KERN_SUCCESS || kr == MACH_SEND_TIMED_OUT, kr) << "mach_msg";
  49. }
  50. bool WaitableEvent::IsSignaled() {
  51. return PeekPort(receive_right_->Name(), policy_ == ResetPolicy::AUTOMATIC);
  52. }
  53. void WaitableEvent::Wait() {
  54. bool result = TimedWait(TimeDelta::Max());
  55. DCHECK(result) << "TimedWait() should never fail with infinite timeout";
  56. }
  57. bool WaitableEvent::TimedWait(const TimeDelta& wait_delta) {
  58. if (wait_delta <= TimeDelta())
  59. return IsSignaled();
  60. // Record the event that this thread is blocking upon (for hang diagnosis) and
  61. // consider blocked for scheduling purposes. Ignore this for non-blocking
  62. // WaitableEvents.
  63. absl::optional<debug::ScopedEventWaitActivity> event_activity;
  64. absl::optional<internal::ScopedBlockingCallWithBaseSyncPrimitives>
  65. scoped_blocking_call;
  66. if (waiting_is_blocking_) {
  67. event_activity.emplace(this);
  68. scoped_blocking_call.emplace(FROM_HERE, BlockingType::MAY_BLOCK);
  69. }
  70. mach_msg_empty_rcv_t msg{};
  71. msg.header.msgh_local_port = receive_right_->Name();
  72. mach_msg_option_t options = MACH_RCV_MSG;
  73. if (!wait_delta.is_max())
  74. options |= MACH_RCV_TIMEOUT | MACH_RCV_INTERRUPT;
  75. mach_msg_size_t rcv_size = sizeof(msg);
  76. if (policy_ == ResetPolicy::MANUAL) {
  77. // To avoid dequeing the message, receive with a size of 0 and set
  78. // MACH_RCV_LARGE to keep the message in the queue.
  79. options |= MACH_RCV_LARGE;
  80. rcv_size = 0;
  81. }
  82. // TimeTicks takes care of overflow but we special case is_max() nonetheless
  83. // to avoid invoking TimeTicksNowIgnoringOverride() unnecessarily (same for
  84. // the increment step of the for loop if the condition variable returns
  85. // early). Ref: https://crbug.com/910524#c7
  86. const TimeTicks end_time =
  87. wait_delta.is_max() ? TimeTicks::Max()
  88. : subtle::TimeTicksNowIgnoringOverride() + wait_delta;
  89. // Fake |kr| value to boostrap the for loop.
  90. kern_return_t kr = MACH_RCV_INTERRUPTED;
  91. for (mach_msg_timeout_t timeout =
  92. wait_delta.is_max() ? MACH_MSG_TIMEOUT_NONE
  93. : saturated_cast<mach_msg_timeout_t>(
  94. wait_delta.InMillisecondsRoundedUp());
  95. // If the thread is interrupted during mach_msg(), the system call will
  96. // be restarted. However, the libsyscall wrapper does not adjust the
  97. // timeout by the amount of time already waited. Using MACH_RCV_INTERRUPT
  98. // will instead return from mach_msg(), so that the call can be retried
  99. // with an adjusted timeout.
  100. kr == MACH_RCV_INTERRUPTED;
  101. timeout = end_time.is_max()
  102. ? MACH_MSG_TIMEOUT_NONE
  103. : std::max(mach_msg_timeout_t{0},
  104. saturated_cast<mach_msg_timeout_t>(
  105. (end_time -
  106. subtle::TimeTicksNowIgnoringOverride())
  107. .InMillisecondsRoundedUp()))) {
  108. kr = mach_msg(&msg.header, options, 0, rcv_size, receive_right_->Name(),
  109. timeout, MACH_PORT_NULL);
  110. }
  111. if (kr == KERN_SUCCESS) {
  112. return true;
  113. } else if (rcv_size == 0 && kr == MACH_RCV_TOO_LARGE) {
  114. return true;
  115. } else {
  116. MACH_CHECK(kr == MACH_RCV_TIMED_OUT, kr) << "mach_msg";
  117. return false;
  118. }
  119. }
  120. // static
  121. size_t WaitableEvent::WaitMany(WaitableEvent** raw_waitables, size_t count) {
  122. DCHECK(count) << "Cannot wait on no events";
  123. internal::ScopedBlockingCallWithBaseSyncPrimitives scoped_blocking_call(
  124. FROM_HERE, BlockingType::MAY_BLOCK);
  125. // Record an event (the first) that this thread is blocking upon.
  126. debug::ScopedEventWaitActivity event_activity(raw_waitables[0]);
  127. // On macOS 10.11+, using Mach port sets may cause system instability, per
  128. // https://crbug.com/756102. On macOS 10.12+, a kqueue can be used
  129. // instead to work around that.
  130. enum WaitManyPrimitive {
  131. KQUEUE,
  132. PORT_SET,
  133. };
  134. #if BUILDFLAG(IS_IOS)
  135. const WaitManyPrimitive kPrimitive = PORT_SET;
  136. #else
  137. const WaitManyPrimitive kPrimitive = KQUEUE;
  138. #endif
  139. if (kPrimitive == KQUEUE) {
  140. std::vector<kevent64_s> events(count);
  141. for (size_t i = 0; i < count; ++i) {
  142. EV_SET64(&events[i], raw_waitables[i]->receive_right_->Name(),
  143. EVFILT_MACHPORT, EV_ADD, 0, 0, i, 0, 0);
  144. }
  145. std::vector<kevent64_s> out_events(count);
  146. ScopedFD wait_many(kqueue());
  147. PCHECK(wait_many.is_valid()) << "kqueue";
  148. const int count_int = checked_cast<int>(count);
  149. int rv = HANDLE_EINTR(kevent64(wait_many.get(), events.data(), count_int,
  150. out_events.data(), count_int, 0, nullptr));
  151. PCHECK(rv > 0) << "kevent64";
  152. size_t triggered = std::numeric_limits<size_t>::max();
  153. for (size_t i = 0; i < static_cast<size_t>(rv); ++i) {
  154. // WaitMany should return the lowest index in |raw_waitables| that was
  155. // triggered.
  156. size_t index = static_cast<size_t>(out_events[i].udata);
  157. triggered = std::min(triggered, index);
  158. }
  159. if (raw_waitables[triggered]->policy_ == ResetPolicy::AUTOMATIC) {
  160. // The message needs to be dequeued to reset the event.
  161. PeekPort(raw_waitables[triggered]->receive_right_->Name(), true);
  162. }
  163. return triggered;
  164. } else {
  165. DCHECK_EQ(kPrimitive, PORT_SET);
  166. kern_return_t kr;
  167. mac::ScopedMachPortSet port_set;
  168. {
  169. mach_port_t name;
  170. kr =
  171. mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET, &name);
  172. MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_allocate";
  173. port_set.reset(name);
  174. }
  175. for (size_t i = 0; i < count; ++i) {
  176. kr = mach_port_insert_member(mach_task_self(),
  177. raw_waitables[i]->receive_right_->Name(),
  178. port_set.get());
  179. MACH_CHECK(kr == KERN_SUCCESS, kr) << "index " << i;
  180. }
  181. mach_msg_empty_rcv_t msg{};
  182. // Wait on the port set. Only specify space enough for the header, to
  183. // identify which port in the set is signaled. Otherwise, receiving from the
  184. // port set may dequeue a message for a manual-reset event object, which
  185. // would cause it to be reset.
  186. kr = mach_msg(&msg.header,
  187. MACH_RCV_MSG | MACH_RCV_LARGE | MACH_RCV_LARGE_IDENTITY, 0,
  188. sizeof(msg.header), port_set.get(), 0, MACH_PORT_NULL);
  189. MACH_CHECK(kr == MACH_RCV_TOO_LARGE, kr) << "mach_msg";
  190. for (size_t i = 0; i < count; ++i) {
  191. WaitableEvent* event = raw_waitables[i];
  192. if (msg.header.msgh_local_port == event->receive_right_->Name()) {
  193. if (event->policy_ == ResetPolicy::AUTOMATIC) {
  194. // The message needs to be dequeued to reset the event.
  195. PeekPort(msg.header.msgh_local_port, true);
  196. }
  197. return i;
  198. }
  199. }
  200. NOTREACHED();
  201. return 0;
  202. }
  203. }
  204. // static
  205. bool WaitableEvent::PeekPort(mach_port_t port, bool dequeue) {
  206. if (dequeue) {
  207. mach_msg_empty_rcv_t msg{};
  208. msg.header.msgh_local_port = port;
  209. kern_return_t kr = mach_msg(&msg.header, MACH_RCV_MSG | MACH_RCV_TIMEOUT, 0,
  210. sizeof(msg), port, 0, MACH_PORT_NULL);
  211. if (kr == KERN_SUCCESS) {
  212. return true;
  213. } else {
  214. MACH_CHECK(kr == MACH_RCV_TIMED_OUT, kr) << "mach_msg";
  215. return false;
  216. }
  217. } else {
  218. mach_port_seqno_t seqno = 0;
  219. mach_msg_size_t size;
  220. mach_msg_id_t id;
  221. mach_msg_trailer_t trailer;
  222. mach_msg_type_number_t trailer_size = sizeof(trailer);
  223. kern_return_t kr = mach_port_peek(
  224. mach_task_self(), port, MACH_RCV_TRAILER_TYPE(MACH_RCV_TRAILER_NULL),
  225. &seqno, &size, &id, reinterpret_cast<mach_msg_trailer_info_t>(&trailer),
  226. &trailer_size);
  227. if (kr == KERN_SUCCESS) {
  228. return true;
  229. } else {
  230. MACH_CHECK(kr == KERN_FAILURE, kr) << "mach_port_peek";
  231. return false;
  232. }
  233. }
  234. }
  235. WaitableEvent::ReceiveRight::ReceiveRight(mach_port_t name) : right_(name) {}
  236. WaitableEvent::ReceiveRight::~ReceiveRight() = default;
  237. } // namespace base