message_pump_kqueue.cc 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502
  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 <sys/errno.h>
  6. #include "base/auto_reset.h"
  7. #include "base/logging.h"
  8. #include "base/mac/mac_util.h"
  9. #include "base/mac/mach_logging.h"
  10. #include "base/mac/scoped_nsautorelease_pool.h"
  11. #include "base/posix/eintr_wrapper.h"
  12. #include "base/time/time_override.h"
  13. namespace base {
  14. namespace {
  15. #if DCHECK_IS_ON()
  16. // Prior to macOS 10.14, kqueue timers may spuriously wake up, because earlier
  17. // wake ups race with timer resets in the kernel. As of macOS 10.14, updating a
  18. // timer from the thread that reads the kqueue does not cause spurious wakeups.
  19. // Note that updating a kqueue timer from one thread while another thread is
  20. // waiting in a kevent64 invocation is still (inherently) racy.
  21. bool KqueueTimersSpuriouslyWakeUp() {
  22. #if BUILDFLAG(IS_MAC)
  23. static const bool kqueue_timers_spuriously_wakeup = mac::IsAtMostOS10_13();
  24. return kqueue_timers_spuriously_wakeup;
  25. #else
  26. // This still happens on iOS15.
  27. return true;
  28. #endif
  29. }
  30. #endif
  31. int ChangeOneEvent(const ScopedFD& kqueue, kevent64_s* event) {
  32. return HANDLE_EINTR(kevent64(kqueue.get(), event, 1, nullptr, 0, 0, nullptr));
  33. }
  34. } // namespace
  35. MessagePumpKqueue::FdWatchController::FdWatchController(
  36. const Location& from_here)
  37. : FdWatchControllerInterface(from_here) {}
  38. MessagePumpKqueue::FdWatchController::~FdWatchController() {
  39. StopWatchingFileDescriptor();
  40. }
  41. bool MessagePumpKqueue::FdWatchController::StopWatchingFileDescriptor() {
  42. if (!pump_)
  43. return true;
  44. return pump_->StopWatchingFileDescriptor(this);
  45. }
  46. void MessagePumpKqueue::FdWatchController::Init(WeakPtr<MessagePumpKqueue> pump,
  47. int fd,
  48. int mode,
  49. FdWatcher* watcher) {
  50. DCHECK_NE(fd, -1);
  51. DCHECK(!watcher_);
  52. DCHECK(watcher);
  53. DCHECK(pump);
  54. fd_ = fd;
  55. mode_ = mode;
  56. watcher_ = watcher;
  57. pump_ = pump;
  58. }
  59. void MessagePumpKqueue::FdWatchController::Reset() {
  60. fd_ = -1;
  61. mode_ = 0;
  62. watcher_ = nullptr;
  63. pump_ = nullptr;
  64. }
  65. MessagePumpKqueue::MachPortWatchController::MachPortWatchController(
  66. const Location& from_here)
  67. : from_here_(from_here) {}
  68. MessagePumpKqueue::MachPortWatchController::~MachPortWatchController() {
  69. StopWatchingMachPort();
  70. }
  71. bool MessagePumpKqueue::MachPortWatchController::StopWatchingMachPort() {
  72. if (!pump_)
  73. return true;
  74. return pump_->StopWatchingMachPort(this);
  75. }
  76. void MessagePumpKqueue::MachPortWatchController::Init(
  77. WeakPtr<MessagePumpKqueue> pump,
  78. mach_port_t port,
  79. MachPortWatcher* watcher) {
  80. DCHECK(!watcher_);
  81. DCHECK(watcher);
  82. DCHECK(pump);
  83. port_ = port;
  84. watcher_ = watcher;
  85. pump_ = pump;
  86. }
  87. void MessagePumpKqueue::MachPortWatchController::Reset() {
  88. port_ = MACH_PORT_NULL;
  89. watcher_ = nullptr;
  90. pump_ = nullptr;
  91. }
  92. MessagePumpKqueue::MessagePumpKqueue()
  93. : kqueue_(kqueue()), weak_factory_(this) {
  94. PCHECK(kqueue_.is_valid()) << "kqueue";
  95. // Create a Mach port that will be used to wake up the pump by sending
  96. // a message in response to ScheduleWork(). This is significantly faster than
  97. // using an EVFILT_USER event, especially when triggered across threads.
  98. kern_return_t kr = mach_port_allocate(
  99. mach_task_self(), MACH_PORT_RIGHT_RECEIVE,
  100. base::mac::ScopedMachReceiveRight::Receiver(wakeup_).get());
  101. MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_allocate";
  102. // Configure the event to directly receive the Mach message as part of the
  103. // kevent64() call.
  104. kevent64_s event{};
  105. event.ident = wakeup_.get();
  106. event.filter = EVFILT_MACHPORT;
  107. event.flags = EV_ADD;
  108. event.fflags = MACH_RCV_MSG;
  109. event.ext[0] = reinterpret_cast<uint64_t>(&wakeup_buffer_);
  110. event.ext[1] = sizeof(wakeup_buffer_);
  111. int rv = ChangeOneEvent(kqueue_, &event);
  112. PCHECK(rv == 0) << "kevent64";
  113. }
  114. MessagePumpKqueue::~MessagePumpKqueue() {}
  115. void MessagePumpKqueue::Run(Delegate* delegate) {
  116. AutoReset<bool> reset_keep_running(&keep_running_, true);
  117. while (keep_running_) {
  118. mac::ScopedNSAutoreleasePool pool;
  119. bool do_more_work = DoInternalWork(delegate, nullptr);
  120. if (!keep_running_)
  121. break;
  122. Delegate::NextWorkInfo next_work_info = delegate->DoWork();
  123. do_more_work |= next_work_info.is_immediate();
  124. if (!keep_running_)
  125. break;
  126. if (do_more_work)
  127. continue;
  128. do_more_work |= delegate->DoIdleWork();
  129. if (!keep_running_)
  130. break;
  131. if (do_more_work)
  132. continue;
  133. DoInternalWork(delegate, &next_work_info);
  134. }
  135. }
  136. void MessagePumpKqueue::Quit() {
  137. keep_running_ = false;
  138. ScheduleWork();
  139. }
  140. void MessagePumpKqueue::ScheduleWork() {
  141. mach_msg_empty_send_t message{};
  142. message.header.msgh_size = sizeof(message);
  143. message.header.msgh_bits =
  144. MACH_MSGH_BITS_REMOTE(MACH_MSG_TYPE_MAKE_SEND_ONCE);
  145. message.header.msgh_remote_port = wakeup_.get();
  146. kern_return_t kr = mach_msg_send(&message.header);
  147. if (kr != KERN_SUCCESS) {
  148. // If ScheduleWork() is being called by other threads faster than the pump
  149. // can dispatch work, the kernel message queue for the wakeup port can fill
  150. // up (this happens under base_perftests, for example). The kernel does
  151. // return a SEND_ONCE right in the case of failure, which must be destroyed
  152. // to avoid leaking.
  153. MACH_DLOG_IF(ERROR, (kr & ~MACH_MSG_IPC_SPACE) != MACH_SEND_NO_BUFFER, kr)
  154. << "mach_msg_send";
  155. mach_msg_destroy(&message.header);
  156. }
  157. }
  158. void MessagePumpKqueue::ScheduleDelayedWork(
  159. const Delegate::NextWorkInfo& next_work_info) {
  160. // Nothing to do. This MessagePump uses DoWork().
  161. }
  162. bool MessagePumpKqueue::WatchMachReceivePort(
  163. mach_port_t port,
  164. MachPortWatchController* controller,
  165. MachPortWatcher* delegate) {
  166. DCHECK(port != MACH_PORT_NULL);
  167. DCHECK(controller);
  168. DCHECK(delegate);
  169. if (controller->port() != MACH_PORT_NULL) {
  170. DLOG(ERROR)
  171. << "Cannot use the same MachPortWatchController while it is active";
  172. return false;
  173. }
  174. kevent64_s event{};
  175. event.ident = port;
  176. event.filter = EVFILT_MACHPORT;
  177. event.flags = EV_ADD;
  178. int rv = ChangeOneEvent(kqueue_, &event);
  179. if (rv < 0) {
  180. DPLOG(ERROR) << "kevent64";
  181. return false;
  182. }
  183. ++event_count_;
  184. controller->Init(weak_factory_.GetWeakPtr(), port, delegate);
  185. port_controllers_.AddWithID(controller, port);
  186. return true;
  187. }
  188. bool MessagePumpKqueue::WatchFileDescriptor(int fd,
  189. bool persistent,
  190. int mode,
  191. FdWatchController* controller,
  192. FdWatcher* delegate) {
  193. DCHECK_GE(fd, 0);
  194. DCHECK(controller);
  195. DCHECK(delegate);
  196. DCHECK_NE(mode & Mode::WATCH_READ_WRITE, 0);
  197. if (controller->fd() != -1 && controller->fd() != fd) {
  198. DLOG(ERROR) << "Cannot use the same FdWatchController on two different FDs";
  199. return false;
  200. }
  201. StopWatchingFileDescriptor(controller);
  202. std::vector<kevent64_s> events;
  203. kevent64_s base_event{};
  204. base_event.ident = static_cast<uint64_t>(fd);
  205. base_event.flags = EV_ADD | (!persistent ? EV_ONESHOT : 0);
  206. if (mode & Mode::WATCH_READ) {
  207. base_event.filter = EVFILT_READ;
  208. base_event.udata = fd_controllers_.Add(controller);
  209. events.push_back(base_event);
  210. }
  211. if (mode & Mode::WATCH_WRITE) {
  212. base_event.filter = EVFILT_WRITE;
  213. base_event.udata = fd_controllers_.Add(controller);
  214. events.push_back(base_event);
  215. }
  216. int rv = HANDLE_EINTR(kevent64(kqueue_.get(), events.data(),
  217. checked_cast<int>(events.size()), nullptr, 0,
  218. 0, nullptr));
  219. if (rv < 0) {
  220. DPLOG(ERROR) << "WatchFileDescriptor kevent64";
  221. return false;
  222. }
  223. event_count_ += events.size();
  224. controller->Init(weak_factory_.GetWeakPtr(), fd, mode, delegate);
  225. return true;
  226. }
  227. void MessagePumpKqueue::SetWakeupTimerEvent(const base::TimeTicks& wakeup_time,
  228. kevent64_s* timer_event) {
  229. // The ident of the wakeup timer. There's only the one timer as the pair
  230. // (ident, filter) is the identity of the event.
  231. constexpr uint64_t kWakeupTimerIdent = 0x0;
  232. timer_event->ident = kWakeupTimerIdent;
  233. timer_event->filter = EVFILT_TIMER;
  234. if (wakeup_time == base::TimeTicks::Max()) {
  235. timer_event->flags = EV_DELETE;
  236. } else {
  237. timer_event->filter = EVFILT_TIMER;
  238. // This updates the timer if it already exists in |kqueue_|.
  239. timer_event->flags = EV_ADD | EV_ONESHOT;
  240. // Specify the sleep in microseconds to avoid undersleeping due to
  241. // numeric problems. The sleep is computed from TimeTicks::Now rather than
  242. // NextWorkInfo::recent_now because recent_now is strictly earlier than
  243. // current wall-clock. Using an earlier wall clock time to compute the
  244. // delta to the next wakeup wall-clock time would guarantee oversleep.
  245. // If wakeup_time is in the past, the delta below will be negative and the
  246. // timer is set immediately.
  247. timer_event->fflags = NOTE_USECONDS;
  248. timer_event->data = (wakeup_time - base::TimeTicks::Now()).InMicroseconds();
  249. }
  250. }
  251. bool MessagePumpKqueue::StopWatchingMachPort(
  252. MachPortWatchController* controller) {
  253. mach_port_t port = controller->port();
  254. controller->Reset();
  255. port_controllers_.Remove(port);
  256. kevent64_s event{};
  257. event.ident = port;
  258. event.filter = EVFILT_MACHPORT;
  259. event.flags = EV_DELETE;
  260. --event_count_;
  261. int rv = ChangeOneEvent(kqueue_, &event);
  262. if (rv < 0) {
  263. DPLOG(ERROR) << "kevent64";
  264. return false;
  265. }
  266. return true;
  267. }
  268. bool MessagePumpKqueue::StopWatchingFileDescriptor(
  269. FdWatchController* controller) {
  270. int fd = controller->fd();
  271. int mode = controller->mode();
  272. controller->Reset();
  273. if (fd < 0)
  274. return true;
  275. std::vector<kevent64_s> events;
  276. kevent64_s base_event{};
  277. base_event.ident = static_cast<uint64_t>(fd);
  278. base_event.flags = EV_DELETE;
  279. if (mode & Mode::WATCH_READ) {
  280. base_event.filter = EVFILT_READ;
  281. events.push_back(base_event);
  282. }
  283. if (mode & Mode::WATCH_WRITE) {
  284. base_event.filter = EVFILT_WRITE;
  285. events.push_back(base_event);
  286. }
  287. int rv = HANDLE_EINTR(kevent64(kqueue_.get(), events.data(),
  288. checked_cast<int>(events.size()), nullptr, 0,
  289. 0, nullptr));
  290. DPLOG_IF(ERROR, rv < 0) << "StopWatchingFileDescriptor kevent64";
  291. // The keys for the IDMap aren't recorded anywhere (they're attached to the
  292. // kevent object in the kernel), so locate the entries by controller pointer.
  293. for (IDMap<FdWatchController*, uint64_t>::iterator it(&fd_controllers_);
  294. !it.IsAtEnd(); it.Advance()) {
  295. if (it.GetCurrentValue() == controller) {
  296. fd_controllers_.Remove(it.GetCurrentKey());
  297. }
  298. }
  299. event_count_ -= events.size();
  300. return rv >= 0;
  301. }
  302. bool MessagePumpKqueue::DoInternalWork(Delegate* delegate,
  303. Delegate::NextWorkInfo* next_work_info) {
  304. if (events_.size() < event_count_) {
  305. events_.resize(event_count_);
  306. }
  307. bool immediate = next_work_info == nullptr;
  308. unsigned int flags = immediate ? KEVENT_FLAG_IMMEDIATE : 0;
  309. if (!immediate) {
  310. MaybeUpdateWakeupTimer(next_work_info->delayed_run_time);
  311. DCHECK_EQ(scheduled_wakeup_time_, next_work_info->delayed_run_time);
  312. delegate->BeforeWait();
  313. }
  314. int rv =
  315. HANDLE_EINTR(kevent64(kqueue_.get(), nullptr, 0, events_.data(),
  316. checked_cast<int>(events_.size()), flags, nullptr));
  317. if (rv == 0) {
  318. // No events to dispatch so no need to call ProcessEvents().
  319. return false;
  320. }
  321. PCHECK(rv > 0) << "kevent64";
  322. return ProcessEvents(delegate, static_cast<size_t>(rv));
  323. }
  324. bool MessagePumpKqueue::ProcessEvents(Delegate* delegate, size_t count) {
  325. bool did_work = false;
  326. for (size_t i = 0; i < count; ++i) {
  327. auto* event = &events_[i];
  328. if (event->filter == EVFILT_READ || event->filter == EVFILT_WRITE) {
  329. did_work = true;
  330. FdWatchController* controller = fd_controllers_.Lookup(event->udata);
  331. if (!controller) {
  332. // The controller was removed by some other work callout before
  333. // this event could be processed.
  334. continue;
  335. }
  336. FdWatcher* fd_watcher = controller->watcher();
  337. if (event->flags & EV_ONESHOT) {
  338. // If this was a one-shot event, the Controller needs to stop tracking
  339. // the descriptor, so it is not double-removed when it is told to stop
  340. // watching.
  341. controller->Reset();
  342. fd_controllers_.Remove(event->udata);
  343. --event_count_;
  344. }
  345. auto scoped_do_work_item = delegate->BeginWorkItem();
  346. // WatchFileDescriptor() originally upcasts event->ident from an int.
  347. if (event->filter == EVFILT_READ) {
  348. fd_watcher->OnFileCanReadWithoutBlocking(
  349. static_cast<int>(event->ident));
  350. } else if (event->filter == EVFILT_WRITE) {
  351. fd_watcher->OnFileCanWriteWithoutBlocking(
  352. static_cast<int>(event->ident));
  353. }
  354. } else if (event->filter == EVFILT_MACHPORT) {
  355. // WatchMachReceivePort() originally sets event->ident from a mach_port_t.
  356. mach_port_t port = static_cast<mach_port_t>(event->ident);
  357. if (port == wakeup_.get()) {
  358. // The wakeup event has been received, do not treat this as "doing
  359. // work", this just wakes up the pump.
  360. continue;
  361. }
  362. did_work = true;
  363. MachPortWatchController* controller = port_controllers_.Lookup(port);
  364. // The controller could have been removed by some other work callout
  365. // before this event could be processed.
  366. if (controller) {
  367. auto scoped_do_work_item = delegate->BeginWorkItem();
  368. controller->watcher()->OnMachMessageReceived(port);
  369. }
  370. } else if (event->filter == EVFILT_TIMER) {
  371. // The wakeup timer fired.
  372. #if DCHECK_IS_ON()
  373. // On macOS 10.13 and earlier, kqueue timers may spuriously wake up.
  374. // When this happens, the timer will be re-scheduled the next time
  375. // DoInternalWork is entered, which means this doesn't lead to a
  376. // spinning wait.
  377. // When clock overrides are active, TimeTicks::Now may be decoupled from
  378. // wall-clock time, and can therefore not be used to validate whether the
  379. // expected wall-clock time has passed.
  380. if (!KqueueTimersSpuriouslyWakeUp() &&
  381. !subtle::ScopedTimeClockOverrides::overrides_active()) {
  382. // Given the caveats above, assert that the timer didn't fire early.
  383. DCHECK_LE(scheduled_wakeup_time_, base::TimeTicks::Now());
  384. }
  385. #endif
  386. DCHECK_NE(scheduled_wakeup_time_, base::TimeTicks::Max());
  387. scheduled_wakeup_time_ = base::TimeTicks::Max();
  388. --event_count_;
  389. } else {
  390. NOTREACHED() << "Unexpected event for filter " << event->filter;
  391. }
  392. }
  393. return did_work;
  394. }
  395. void MessagePumpKqueue::MaybeUpdateWakeupTimer(
  396. const base::TimeTicks& wakeup_time) {
  397. if (wakeup_time == scheduled_wakeup_time_) {
  398. // No change in the timer setting necessary.
  399. return;
  400. }
  401. if (wakeup_time == base::TimeTicks::Max()) {
  402. // If the timer was already reset, don't re-reset it on a suspend toggle.
  403. if (scheduled_wakeup_time_ != base::TimeTicks::Max()) {
  404. // Clear the timer.
  405. kevent64_s timer{};
  406. SetWakeupTimerEvent(wakeup_time, &timer);
  407. int rv = ChangeOneEvent(kqueue_, &timer);
  408. PCHECK(rv == 0) << "kevent64, delete timer";
  409. --event_count_;
  410. }
  411. } else {
  412. // Set/reset the timer.
  413. kevent64_s timer{};
  414. SetWakeupTimerEvent(wakeup_time, &timer);
  415. int rv = ChangeOneEvent(kqueue_, &timer);
  416. PCHECK(rv == 0) << "kevent64, set timer";
  417. // Bump the event count if we just added the timer.
  418. if (scheduled_wakeup_time_ == base::TimeTicks::Max())
  419. ++event_count_;
  420. }
  421. scheduled_wakeup_time_ = wakeup_time;
  422. }
  423. } // namespace base