| // Copyright 2019 The Chromium Authors |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #ifdef UNSAFE_BUFFERS_BUILD |
| // TODO(crbug.com/40284755): Remove this and spanify to fix the errors. |
| #pragma allow_unsafe_buffers |
| #endif |
| |
| #include "base/message_loop/message_pump_kqueue.h" |
| |
| #include <sys/errno.h> |
| |
| #include <atomic> |
| |
| #include "base/apple/mach_logging.h" |
| #include "base/apple/scoped_nsautorelease_pool.h" |
| #include "base/auto_reset.h" |
| #include "base/feature_list.h" |
| #include "base/logging.h" |
| #include "base/mac/mac_util.h" |
| #include "base/notreached.h" |
| #include "base/posix/eintr_wrapper.h" |
| #include "base/task/task_features.h" |
| #include "base/time/time_override.h" |
| |
| namespace base { |
| |
| namespace { |
| |
| // Under this feature native work is batched. Remove it once crbug.com/1200141 |
| // is resolved. |
| BASE_FEATURE(kBatchNativeEventsInMessagePumpKqueue, |
| "BatchNativeEventsInMessagePumpKqueue", |
| base::FEATURE_DISABLED_BY_DEFAULT); |
| |
| // Caches the state of the "BatchNativeEventsInMessagePumpKqueue". |
| std::atomic_bool g_use_batched_version = false; |
| |
| // Caches the state of the "TimerSlackMac" feature for efficiency. |
| std::atomic_bool g_timer_slack = false; |
| |
| #if DCHECK_IS_ON() |
| // Prior to macOS 10.14, kqueue timers may spuriously wake up, because earlier |
| // wake ups race with timer resets in the kernel. As of macOS 10.14, updating a |
| // timer from the thread that reads the kqueue does not cause spurious wakeups. |
| // Note that updating a kqueue timer from one thread while another thread is |
| // waiting in a kevent64 invocation is still (inherently) racy. |
| bool KqueueTimersSpuriouslyWakeUp() { |
| #if BUILDFLAG(IS_MAC) |
| return false; |
| #else |
| // This still happens on iOS15. |
| return true; |
| #endif |
| } |
| #endif |
| |
| int ChangeOneEvent(const ScopedFD& kqueue, kevent64_s* event) { |
| return HANDLE_EINTR(kevent64(kqueue.get(), event, 1, nullptr, 0, 0, nullptr)); |
| } |
| |
| } // namespace |
| |
| MessagePumpKqueue::FdWatchController::FdWatchController( |
| const Location& from_here) |
| : FdWatchControllerInterface(from_here) {} |
| |
| MessagePumpKqueue::FdWatchController::~FdWatchController() { |
| StopWatchingFileDescriptor(); |
| } |
| |
| bool MessagePumpKqueue::FdWatchController::StopWatchingFileDescriptor() { |
| if (!pump_) |
| return true; |
| return pump_->StopWatchingFileDescriptor(this); |
| } |
| |
| void MessagePumpKqueue::FdWatchController::Init(WeakPtr<MessagePumpKqueue> pump, |
| int fd, |
| int mode, |
| FdWatcher* watcher) { |
| DCHECK_NE(fd, -1); |
| DCHECK(!watcher_); |
| DCHECK(watcher); |
| DCHECK(pump); |
| fd_ = fd; |
| mode_ = mode; |
| watcher_ = watcher; |
| pump_ = pump; |
| } |
| |
| void MessagePumpKqueue::FdWatchController::Reset() { |
| fd_ = -1; |
| mode_ = 0; |
| watcher_ = nullptr; |
| pump_ = nullptr; |
| } |
| |
| MessagePumpKqueue::MachPortWatchController::MachPortWatchController( |
| const Location& from_here) |
| : from_here_(from_here) {} |
| |
| MessagePumpKqueue::MachPortWatchController::~MachPortWatchController() { |
| StopWatchingMachPort(); |
| } |
| |
| bool MessagePumpKqueue::MachPortWatchController::StopWatchingMachPort() { |
| if (!pump_) |
| return true; |
| return pump_->StopWatchingMachPort(this); |
| } |
| |
| void MessagePumpKqueue::MachPortWatchController::Init( |
| WeakPtr<MessagePumpKqueue> pump, |
| mach_port_t port, |
| MachPortWatcher* watcher) { |
| DCHECK(!watcher_); |
| DCHECK(watcher); |
| DCHECK(pump); |
| port_ = port; |
| watcher_ = watcher; |
| pump_ = pump; |
| } |
| |
| void MessagePumpKqueue::MachPortWatchController::Reset() { |
| port_ = MACH_PORT_NULL; |
| watcher_ = nullptr; |
| pump_ = nullptr; |
| } |
| |
| MessagePumpKqueue::MessagePumpKqueue() |
| : kqueue_(kqueue()), weak_factory_(this) { |
| PCHECK(kqueue_.is_valid()) << "kqueue"; |
| |
| // Create a Mach port that will be used to wake up the pump by sending |
| // a message in response to ScheduleWork(). This is significantly faster than |
| // using an EVFILT_USER event, especially when triggered across threads. |
| kern_return_t kr = mach_port_allocate( |
| mach_task_self(), MACH_PORT_RIGHT_RECEIVE, |
| base::apple::ScopedMachReceiveRight::Receiver(wakeup_).get()); |
| MACH_CHECK(kr == KERN_SUCCESS, kr) << "mach_port_allocate"; |
| |
| // Configure the event to directly receive the Mach message as part of the |
| // kevent64() call. |
| kevent64_s event{}; |
| event.ident = wakeup_.get(); |
| event.filter = EVFILT_MACHPORT; |
| event.flags = EV_ADD; |
| event.fflags = MACH_RCV_MSG; |
| event.ext[0] = reinterpret_cast<uint64_t>(&wakeup_buffer_); |
| event.ext[1] = sizeof(wakeup_buffer_); |
| |
| int rv = ChangeOneEvent(kqueue_, &event); |
| PCHECK(rv == 0) << "kevent64"; |
| } |
| |
| MessagePumpKqueue::~MessagePumpKqueue() {} |
| |
| void MessagePumpKqueue::InitializeFeatures() { |
| g_use_batched_version.store( |
| base::FeatureList::IsEnabled(kBatchNativeEventsInMessagePumpKqueue), |
| std::memory_order_relaxed); |
| g_timer_slack.store(FeatureList::IsEnabled(kTimerSlackMac), |
| std::memory_order_relaxed); |
| } |
| |
| void MessagePumpKqueue::Run(Delegate* delegate) { |
| AutoReset<bool> reset_keep_running(&keep_running_, true); |
| |
| if (g_use_batched_version.load(std::memory_order_relaxed)) { |
| RunBatched(delegate); |
| } else { |
| while (keep_running_) { |
| apple::ScopedNSAutoreleasePool pool; |
| |
| bool do_more_work = DoInternalWork(delegate, nullptr); |
| if (!keep_running_) |
| break; |
| |
| Delegate::NextWorkInfo next_work_info = delegate->DoWork(); |
| do_more_work |= next_work_info.is_immediate(); |
| if (!keep_running_) |
| break; |
| |
| if (do_more_work) |
| continue; |
| |
| do_more_work |= delegate->DoIdleWork(); |
| if (!keep_running_) |
| break; |
| |
| if (do_more_work) |
| continue; |
| |
| DoInternalWork(delegate, &next_work_info); |
| } |
| } |
| } |
| |
| void MessagePumpKqueue::RunBatched(Delegate* delegate) { |
| // Look for native work once before the loop starts. Without this call the |
| // loop would break without checking native work even once in cases where |
| // QuitWhenIdle was used. This is sometimes the case in tests. |
| DoInternalWork(delegate, nullptr); |
| |
| while (keep_running_) { |
| apple::ScopedNSAutoreleasePool pool; |
| |
| Delegate::NextWorkInfo next_work_info = delegate->DoWork(); |
| if (!keep_running_) |
| break; |
| |
| if (!next_work_info.is_immediate()) { |
| delegate->DoIdleWork(); |
| } |
| if (!keep_running_) |
| break; |
| |
| int batch_size = 0; |
| if (DoInternalWork(delegate, &next_work_info)) { |
| // More than one call can be necessary to fully dispatch all available |
| // internal work. Making an effort to dispatch more than the minimum |
| // before moving on to application tasks reduces the overhead of going |
| // through the whole loop. It also more closely mirrors the behavior of |
| // application task execution where tasks are batched. A value of 16 was |
| // chosen via local experimentation showing that is was sufficient to |
| // dispatch all work in roughly 95% of cases. |
| constexpr int kMaxAttempts = 16; |
| while (DoInternalWork(delegate, nullptr) && batch_size < kMaxAttempts) { |
| ++batch_size; |
| } |
| } |
| } |
| } |
| |
| void MessagePumpKqueue::Quit() { |
| keep_running_ = false; |
| ScheduleWork(); |
| } |
| |
| void MessagePumpKqueue::ScheduleWork() { |
| mach_msg_empty_send_t message{}; |
| message.header.msgh_size = sizeof(message); |
| message.header.msgh_bits = |
| MACH_MSGH_BITS_REMOTE(MACH_MSG_TYPE_MAKE_SEND_ONCE); |
| message.header.msgh_remote_port = wakeup_.get(); |
| kern_return_t kr = mach_msg_send(&message.header); |
| if (kr != KERN_SUCCESS) { |
| // If ScheduleWork() is being called by other threads faster than the pump |
| // can dispatch work, the kernel message queue for the wakeup port can fill |
| // up (this happens under base_perftests, for example). The kernel does |
| // return a SEND_ONCE right in the case of failure, which must be destroyed |
| // to avoid leaking. |
| MACH_DLOG_IF(ERROR, (kr & ~MACH_MSG_IPC_SPACE) != MACH_SEND_NO_BUFFER, kr) |
| << "mach_msg_send"; |
| mach_msg_destroy(&message.header); |
| } |
| } |
| |
| void MessagePumpKqueue::ScheduleDelayedWork( |
| const Delegate::NextWorkInfo& next_work_info) { |
| // Nothing to do. This MessagePump uses DoWork(). |
| } |
| |
| bool MessagePumpKqueue::WatchMachReceivePort( |
| mach_port_t port, |
| MachPortWatchController* controller, |
| MachPortWatcher* delegate) { |
| DCHECK(port != MACH_PORT_NULL); |
| DCHECK(controller); |
| DCHECK(delegate); |
| |
| if (controller->port() != MACH_PORT_NULL) { |
| DLOG(ERROR) |
| << "Cannot use the same MachPortWatchController while it is active"; |
| return false; |
| } |
| |
| kevent64_s event{}; |
| event.ident = port; |
| event.filter = EVFILT_MACHPORT; |
| event.flags = EV_ADD; |
| int rv = ChangeOneEvent(kqueue_, &event); |
| if (rv < 0) { |
| DPLOG(ERROR) << "kevent64"; |
| return false; |
| } |
| ++event_count_; |
| |
| controller->Init(weak_factory_.GetWeakPtr(), port, delegate); |
| port_controllers_.AddWithID(controller, port); |
| |
| return true; |
| } |
| |
| TimeTicks MessagePumpKqueue::AdjustDelayedRunTime(TimeTicks earliest_time, |
| TimeTicks run_time, |
| TimeTicks latest_time) { |
| if (g_timer_slack.load(std::memory_order_relaxed)) { |
| return earliest_time; |
| } |
| return MessagePump::AdjustDelayedRunTime(earliest_time, run_time, |
| latest_time); |
| } |
| |
| bool MessagePumpKqueue::WatchFileDescriptor(int fd, |
| bool persistent, |
| int mode, |
| FdWatchController* controller, |
| FdWatcher* delegate) { |
| DCHECK_GE(fd, 0); |
| DCHECK(controller); |
| DCHECK(delegate); |
| DCHECK_NE(mode & Mode::WATCH_READ_WRITE, 0); |
| |
| if (controller->fd() != -1 && controller->fd() != fd) { |
| DLOG(ERROR) << "Cannot use the same FdWatchController on two different FDs"; |
| return false; |
| } |
| StopWatchingFileDescriptor(controller); |
| |
| std::vector<kevent64_s> events; |
| |
| kevent64_s base_event{}; |
| base_event.ident = static_cast<uint64_t>(fd); |
| base_event.flags = EV_ADD | (!persistent ? EV_ONESHOT : 0); |
| |
| if (mode & Mode::WATCH_READ) { |
| base_event.filter = EVFILT_READ; |
| base_event.udata = fd_controllers_.Add(controller); |
| events.push_back(base_event); |
| } |
| if (mode & Mode::WATCH_WRITE) { |
| base_event.filter = EVFILT_WRITE; |
| base_event.udata = fd_controllers_.Add(controller); |
| events.push_back(base_event); |
| } |
| |
| int rv = HANDLE_EINTR(kevent64(kqueue_.get(), events.data(), |
| checked_cast<int>(events.size()), nullptr, 0, |
| 0, nullptr)); |
| if (rv < 0) { |
| DPLOG(ERROR) << "WatchFileDescriptor kevent64"; |
| return false; |
| } |
| |
| event_count_ += events.size(); |
| controller->Init(weak_factory_.GetWeakPtr(), fd, mode, delegate); |
| |
| return true; |
| } |
| |
| void MessagePumpKqueue::SetWakeupTimerEvent(const base::TimeTicks& wakeup_time, |
| base::TimeDelta leeway, |
| kevent64_s* timer_event) { |
| // The ident of the wakeup timer. There's only the one timer as the pair |
| // (ident, filter) is the identity of the event. |
| constexpr uint64_t kWakeupTimerIdent = 0x0; |
| timer_event->ident = kWakeupTimerIdent; |
| timer_event->filter = EVFILT_TIMER; |
| if (wakeup_time == base::TimeTicks::Max()) { |
| timer_event->flags = EV_DELETE; |
| } else { |
| timer_event->filter = EVFILT_TIMER; |
| // This updates the timer if it already exists in |kqueue_|. |
| timer_event->flags = EV_ADD | EV_ONESHOT; |
| |
| // Specify the sleep in microseconds to avoid undersleeping due to |
| // numeric problems. The sleep is computed from TimeTicks::Now rather than |
| // NextWorkInfo::recent_now because recent_now is strictly earlier than |
| // current wall-clock. Using an earlier wall clock time to compute the |
| // delta to the next wakeup wall-clock time would guarantee oversleep. |
| // If wakeup_time is in the past, the delta below will be negative and the |
| // timer is set immediately. |
| timer_event->fflags = NOTE_USECONDS; |
| timer_event->data = (wakeup_time - base::TimeTicks::Now()).InMicroseconds(); |
| |
| if (!leeway.is_zero() && g_timer_slack.load(std::memory_order_relaxed)) { |
| // Specify slack based on |leeway|. |
| // See "man kqueue" in recent macOSen for documentation. |
| timer_event->fflags |= NOTE_LEEWAY; |
| timer_event->ext[1] = static_cast<uint64_t>(leeway.InMicroseconds()); |
| } |
| } |
| } |
| |
| bool MessagePumpKqueue::StopWatchingMachPort( |
| MachPortWatchController* controller) { |
| mach_port_t port = controller->port(); |
| controller->Reset(); |
| port_controllers_.Remove(port); |
| |
| kevent64_s event{}; |
| event.ident = port; |
| event.filter = EVFILT_MACHPORT; |
| event.flags = EV_DELETE; |
| --event_count_; |
| int rv = ChangeOneEvent(kqueue_, &event); |
| if (rv < 0) { |
| DPLOG(ERROR) << "kevent64"; |
| return false; |
| } |
| |
| return true; |
| } |
| |
| bool MessagePumpKqueue::StopWatchingFileDescriptor( |
| FdWatchController* controller) { |
| int fd = controller->fd(); |
| int mode = controller->mode(); |
| controller->Reset(); |
| |
| if (fd < 0) |
| return true; |
| |
| std::vector<kevent64_s> events; |
| |
| kevent64_s base_event{}; |
| base_event.ident = static_cast<uint64_t>(fd); |
| base_event.flags = EV_DELETE; |
| |
| if (mode & Mode::WATCH_READ) { |
| base_event.filter = EVFILT_READ; |
| events.push_back(base_event); |
| } |
| if (mode & Mode::WATCH_WRITE) { |
| base_event.filter = EVFILT_WRITE; |
| events.push_back(base_event); |
| } |
| |
| int rv = HANDLE_EINTR(kevent64(kqueue_.get(), events.data(), |
| checked_cast<int>(events.size()), nullptr, 0, |
| 0, nullptr)); |
| DPLOG_IF(ERROR, rv < 0) << "StopWatchingFileDescriptor kevent64"; |
| |
| // The keys for the IDMap aren't recorded anywhere (they're attached to the |
| // kevent object in the kernel), so locate the entries by controller pointer. |
| for (IDMap<FdWatchController*, uint64_t>::iterator it(&fd_controllers_); |
| !it.IsAtEnd(); it.Advance()) { |
| if (it.GetCurrentValue() == controller) { |
| fd_controllers_.Remove(it.GetCurrentKey()); |
| } |
| } |
| |
| event_count_ -= events.size(); |
| |
| return rv >= 0; |
| } |
| |
| bool MessagePumpKqueue::DoInternalWork(Delegate* delegate, |
| Delegate::NextWorkInfo* next_work_info) { |
| if (events_.size() < event_count_) { |
| events_.resize(event_count_); |
| } |
| |
| bool immediate = next_work_info == nullptr; |
| unsigned int flags = immediate ? KEVENT_FLAG_IMMEDIATE : 0; |
| |
| if (!immediate) { |
| MaybeUpdateWakeupTimer(next_work_info->delayed_run_time, |
| next_work_info->leeway); |
| DCHECK_EQ(scheduled_wakeup_time_, next_work_info->delayed_run_time); |
| delegate->BeforeWait(); |
| } |
| |
| int rv = |
| HANDLE_EINTR(kevent64(kqueue_.get(), nullptr, 0, events_.data(), |
| checked_cast<int>(events_.size()), flags, nullptr)); |
| if (rv == 0) { |
| // No events to dispatch so no need to call ProcessEvents(). |
| return false; |
| } |
| |
| PCHECK(rv > 0) << "kevent64"; |
| return ProcessEvents(delegate, static_cast<size_t>(rv)); |
| } |
| |
| bool MessagePumpKqueue::ProcessEvents(Delegate* delegate, size_t count) { |
| bool did_work = false; |
| |
| delegate->BeginNativeWorkBeforeDoWork(); |
| for (size_t i = 0; i < count; ++i) { |
| auto* event = &events_[i]; |
| if (event->filter == EVFILT_READ || event->filter == EVFILT_WRITE) { |
| did_work = true; |
| |
| FdWatchController* controller = fd_controllers_.Lookup(event->udata); |
| if (!controller) { |
| // The controller was removed by some other work callout before |
| // this event could be processed. |
| continue; |
| } |
| FdWatcher* fd_watcher = controller->watcher(); |
| |
| if (event->flags & EV_ONESHOT) { |
| // If this was a one-shot event, the Controller needs to stop tracking |
| // the descriptor, so it is not double-removed when it is told to stop |
| // watching. |
| controller->Reset(); |
| fd_controllers_.Remove(event->udata); |
| --event_count_; |
| } |
| |
| auto scoped_do_work_item = delegate->BeginWorkItem(); |
| // WatchFileDescriptor() originally upcasts event->ident from an int. |
| if (event->filter == EVFILT_READ) { |
| fd_watcher->OnFileCanReadWithoutBlocking( |
| static_cast<int>(event->ident)); |
| } else if (event->filter == EVFILT_WRITE) { |
| fd_watcher->OnFileCanWriteWithoutBlocking( |
| static_cast<int>(event->ident)); |
| } |
| } else if (event->filter == EVFILT_MACHPORT) { |
| // WatchMachReceivePort() originally sets event->ident from a mach_port_t. |
| mach_port_t port = static_cast<mach_port_t>(event->ident); |
| if (port == wakeup_.get()) { |
| // The wakeup event has been received, do not treat this as "doing |
| // work", this just wakes up the pump. |
| continue; |
| } |
| |
| did_work = true; |
| |
| MachPortWatchController* controller = port_controllers_.Lookup(port); |
| // The controller could have been removed by some other work callout |
| // before this event could be processed. |
| if (controller) { |
| auto scoped_do_work_item = delegate->BeginWorkItem(); |
| controller->watcher()->OnMachMessageReceived(port); |
| } |
| } else if (event->filter == EVFILT_TIMER) { |
| // The wakeup timer fired. |
| #if DCHECK_IS_ON() |
| // On macOS 10.13 and earlier, kqueue timers may spuriously wake up. |
| // When this happens, the timer will be re-scheduled the next time |
| // DoInternalWork is entered, which means this doesn't lead to a |
| // spinning wait. |
| // When clock overrides are active, TimeTicks::Now may be decoupled from |
| // wall-clock time, and can therefore not be used to validate whether the |
| // expected wall-clock time has passed. |
| if (!KqueueTimersSpuriouslyWakeUp() && |
| !subtle::ScopedTimeClockOverrides::overrides_active()) { |
| // Given the caveats above, assert that the timer didn't fire early. |
| DCHECK_LE(scheduled_wakeup_time_, base::TimeTicks::Now()); |
| } |
| #endif |
| DCHECK_NE(scheduled_wakeup_time_, base::TimeTicks::Max()); |
| scheduled_wakeup_time_ = base::TimeTicks::Max(); |
| --event_count_; |
| } else { |
| NOTREACHED_IN_MIGRATION() |
| << "Unexpected event for filter " << event->filter; |
| } |
| } |
| |
| return did_work; |
| } |
| |
| void MessagePumpKqueue::MaybeUpdateWakeupTimer( |
| const base::TimeTicks& wakeup_time, |
| base::TimeDelta leeway) { |
| if (wakeup_time == scheduled_wakeup_time_) { |
| // No change in the timer setting necessary. |
| return; |
| } |
| |
| if (wakeup_time == base::TimeTicks::Max()) { |
| // If the timer was already reset, don't re-reset it on a suspend toggle. |
| if (scheduled_wakeup_time_ != base::TimeTicks::Max()) { |
| // Clear the timer. |
| kevent64_s timer{}; |
| SetWakeupTimerEvent(wakeup_time, leeway, &timer); |
| int rv = ChangeOneEvent(kqueue_, &timer); |
| PCHECK(rv == 0) << "kevent64, delete timer"; |
| --event_count_; |
| } |
| } else { |
| // Set/reset the timer. |
| kevent64_s timer{}; |
| SetWakeupTimerEvent(wakeup_time, leeway, &timer); |
| int rv = ChangeOneEvent(kqueue_, &timer); |
| PCHECK(rv == 0) << "kevent64, set timer"; |
| |
| // Bump the event count if we just added the timer. |
| if (scheduled_wakeup_time_ == base::TimeTicks::Max()) |
| ++event_count_; |
| } |
| |
| scheduled_wakeup_time_ = wakeup_time; |
| } |
| |
| } // namespace base |