1 // Copyright 2017 The Chromium Authors
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "base/synchronization/waitable_event.h"
6
7 #include <mach/mach.h>
8 #include <sys/event.h>
9
10 #include <limits>
11 #include <memory>
12
13 #include "base/apple/mach_logging.h"
14 #include "base/files/scoped_file.h"
15 #include "base/notreached.h"
16 #include "base/posix/eintr_wrapper.h"
17 #include "base/threading/scoped_blocking_call.h"
18 #include "base/time/time.h"
19 #include "base/time/time_override.h"
20 #include "build/build_config.h"
21
22 namespace base {
23
WaitableEvent(ResetPolicy reset_policy,InitialState initial_state)24 WaitableEvent::WaitableEvent(ResetPolicy reset_policy,
25 InitialState initial_state)
26 : policy_(reset_policy) {
27 mach_port_options_t options{};
28 options.flags = MPO_INSERT_SEND_RIGHT;
29 options.mpl.mpl_qlimit = 1;
30
31 mach_port_t name;
32 kern_return_t kr =
33 mach_port_construct(mach_task_self(), &options, /*context=*/0, &name);
34 MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_construct";
35
36 receive_right_ = new ReceiveRight(name);
37 send_right_.reset(name);
38
39 if (initial_state == InitialState::SIGNALED) {
40 Signal();
41 }
42 }
43
Reset()44 void WaitableEvent::Reset() {
45 PeekPort(receive_right_->Name(), true);
46 }
47
SignalImpl()48 void WaitableEvent::SignalImpl() {
49 mach_msg_empty_send_t msg{};
50 msg.header.msgh_bits = MACH_MSGH_BITS_REMOTE(MACH_MSG_TYPE_COPY_SEND);
51 msg.header.msgh_size = sizeof(&msg);
52 msg.header.msgh_remote_port = send_right_.get();
53 // If the event is already signaled, this will time out because the queue
54 // has a length of one.
55 kern_return_t kr =
56 mach_msg(&msg.header, MACH_SEND_MSG | MACH_SEND_TIMEOUT, sizeof(msg),
57 /*rcv_size=*/0, /*rcv_name=*/MACH_PORT_NULL, /*timeout=*/0,
58 /*notify=*/MACH_PORT_NULL);
59 MACH_CHECK(kr == KERN_SUCCESS || kr == MACH_SEND_TIMED_OUT, kr) << "mach_msg";
60 }
61
IsSignaled()62 bool WaitableEvent::IsSignaled() {
63 return PeekPort(receive_right_->Name(), policy_ == ResetPolicy::AUTOMATIC);
64 }
65
TimedWaitImpl(TimeDelta wait_delta)66 bool WaitableEvent::TimedWaitImpl(TimeDelta wait_delta) {
67 mach_msg_empty_rcv_t msg{};
68 msg.header.msgh_local_port = receive_right_->Name();
69
70 mach_msg_option_t options = MACH_RCV_MSG;
71
72 if (!wait_delta.is_max()) {
73 options |= MACH_RCV_TIMEOUT | MACH_RCV_INTERRUPT;
74 }
75
76 mach_msg_size_t rcv_size = sizeof(msg);
77 if (policy_ == ResetPolicy::MANUAL) {
78 // To avoid dequeuing the message, receive with a size of 0 and set
79 // MACH_RCV_LARGE to keep the message in the queue.
80 options |= MACH_RCV_LARGE;
81 rcv_size = 0;
82 }
83
84 // TimeTicks takes care of overflow but we special case is_max() nonetheless
85 // to avoid invoking TimeTicksNowIgnoringOverride() unnecessarily (same for
86 // the increment step of the for loop if the condition variable returns
87 // early). Ref: https://crbug.com/910524#c7
88 const TimeTicks end_time =
89 wait_delta.is_max() ? TimeTicks::Max()
90 : subtle::TimeTicksNowIgnoringOverride() + wait_delta;
91 // Fake |kr| value to bootstrap the for loop.
92 kern_return_t kr = MACH_RCV_INTERRUPTED;
93 for (mach_msg_timeout_t timeout =
94 wait_delta.is_max() ? MACH_MSG_TIMEOUT_NONE
95 : saturated_cast<mach_msg_timeout_t>(
96 wait_delta.InMillisecondsRoundedUp());
97 // If the thread is interrupted during mach_msg(), the system call will
98 // be restarted. However, the libsyscall wrapper does not adjust the
99 // timeout by the amount of time already waited. Using MACH_RCV_INTERRUPT
100 // will instead return from mach_msg(), so that the call can be retried
101 // with an adjusted timeout.
102 kr == MACH_RCV_INTERRUPTED;
103 timeout = end_time.is_max()
104 ? MACH_MSG_TIMEOUT_NONE
105 : std::max(mach_msg_timeout_t{0},
106 saturated_cast<mach_msg_timeout_t>(
107 (end_time -
108 subtle::TimeTicksNowIgnoringOverride())
109 .InMillisecondsRoundedUp()))) {
110 kr = mach_msg(&msg.header, options, /*send_size=*/0, rcv_size,
111 receive_right_->Name(), timeout, /*notify=*/MACH_PORT_NULL);
112 }
113
114 if (kr == KERN_SUCCESS) {
115 return true;
116 } else if (rcv_size == 0 && kr == MACH_RCV_TOO_LARGE) {
117 return true;
118 } else {
119 MACH_CHECK(kr == MACH_RCV_TIMED_OUT, kr) << "mach_msg";
120 return false;
121 }
122 }
123
124 // static
WaitManyImpl(WaitableEvent ** raw_waitables,size_t count)125 size_t WaitableEvent::WaitManyImpl(WaitableEvent** raw_waitables,
126 size_t count) {
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
146 std::vector<kevent64_s> out_events(count);
147
148 ScopedFD wait_many(kqueue());
149 PCHECK(wait_many.is_valid()) << "kqueue";
150
151 const int count_int = checked_cast<int>(count);
152 int rv = HANDLE_EINTR(kevent64(wait_many.get(), events.data(), count_int,
153 out_events.data(), count_int, /*flags=*/0,
154 /*timeout=*/nullptr));
155 PCHECK(rv > 0) << "kevent64";
156
157 size_t triggered = std::numeric_limits<size_t>::max();
158 for (size_t i = 0; i < static_cast<size_t>(rv); ++i) {
159 // WaitMany should return the lowest index in |raw_waitables| that was
160 // triggered.
161 size_t index = static_cast<size_t>(out_events[i].udata);
162 triggered = std::min(triggered, index);
163 }
164
165 if (raw_waitables[triggered]->policy_ == ResetPolicy::AUTOMATIC) {
166 // The message needs to be dequeued to reset the event.
167 PeekPort(raw_waitables[triggered]->receive_right_->Name(),
168 /*dequeue=*/true);
169 }
170
171 return triggered;
172 } else {
173 DCHECK_EQ(kPrimitive, PORT_SET);
174
175 kern_return_t kr;
176
177 apple::ScopedMachPortSet port_set;
178 {
179 mach_port_t name;
180 kr =
181 mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET, &name);
182 MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_allocate";
183 port_set.reset(name);
184 }
185
186 for (size_t i = 0; i < count; ++i) {
187 kr = mach_port_insert_member(mach_task_self(),
188 raw_waitables[i]->receive_right_->Name(),
189 port_set.get());
190 MACH_CHECK(kr == KERN_SUCCESS, kr) << "index " << i;
191 }
192
193 mach_msg_empty_rcv_t msg{};
194 // Wait on the port set. Only specify space enough for the header, to
195 // identify which port in the set is signaled. Otherwise, receiving from the
196 // port set may dequeue a message for a manual-reset event object, which
197 // would cause it to be reset.
198 kr = mach_msg(&msg.header,
199 MACH_RCV_MSG | MACH_RCV_LARGE | MACH_RCV_LARGE_IDENTITY,
200 /*send_size=*/0, sizeof(msg.header), port_set.get(),
201 /*timeout=*/0, /*notify=*/MACH_PORT_NULL);
202 MACH_CHECK(kr == MACH_RCV_TOO_LARGE, kr) << "mach_msg";
203
204 for (size_t i = 0; i < count; ++i) {
205 WaitableEvent* event = raw_waitables[i];
206 if (msg.header.msgh_local_port == event->receive_right_->Name()) {
207 if (event->policy_ == ResetPolicy::AUTOMATIC) {
208 // The message needs to be dequeued to reset the event.
209 PeekPort(msg.header.msgh_local_port, true);
210 }
211 return i;
212 }
213 }
214
215 NOTREACHED();
216 return 0;
217 }
218 }
219
220 // static
PeekPort(mach_port_t port,bool dequeue)221 bool WaitableEvent::PeekPort(mach_port_t port, bool dequeue) {
222 if (dequeue) {
223 mach_msg_empty_rcv_t msg{};
224 msg.header.msgh_local_port = port;
225 kern_return_t kr =
226 mach_msg(&msg.header, MACH_RCV_MSG | MACH_RCV_TIMEOUT, /*send_size=*/0,
227 sizeof(msg), port, /*timeout=*/0, /*notify=*/MACH_PORT_NULL);
228 if (kr == KERN_SUCCESS) {
229 return true;
230 } else {
231 MACH_CHECK(kr == MACH_RCV_TIMED_OUT, kr) << "mach_msg";
232 return false;
233 }
234 } else {
235 mach_port_seqno_t seqno = 0;
236 mach_msg_size_t size;
237 mach_msg_id_t id;
238 mach_msg_trailer_t trailer;
239 mach_msg_type_number_t trailer_size = sizeof(trailer);
240 kern_return_t kr = mach_port_peek(
241 mach_task_self(), port, MACH_RCV_TRAILER_TYPE(MACH_RCV_TRAILER_NULL),
242 &seqno, &size, &id, reinterpret_cast<mach_msg_trailer_info_t>(&trailer),
243 &trailer_size);
244 if (kr == KERN_SUCCESS) {
245 return true;
246 } else {
247 MACH_CHECK(kr == KERN_FAILURE, kr) << "mach_port_peek";
248 return false;
249 }
250 }
251 }
252
ReceiveRight(mach_port_t name)253 WaitableEvent::ReceiveRight::ReceiveRight(mach_port_t name) : right_(name) {}
254
255 WaitableEvent::ReceiveRight::~ReceiveRight() = default;
256
257 } // namespace base
258