123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332 |
- // Copyright 2022 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- #include "base/message_loop/message_pump_epoll.h"
- #include <sys/epoll.h>
- #include <sys/eventfd.h>
- #include <algorithm>
- #include <cstddef>
- #include <cstdint>
- #include <utility>
- #include "base/auto_reset.h"
- #include "base/check_op.h"
- #include "base/memory/ref_counted.h"
- #include "base/posix/eintr_wrapper.h"
- #include "base/threading/thread_checker.h"
- #include "base/trace_event/base_tracing.h"
- #include "third_party/abseil-cpp/absl/types/optional.h"
- namespace base {
- MessagePumpEpoll::MessagePumpEpoll() {
- epoll_.reset(epoll_create(/*ignored_but_must_be_positive=*/1));
- PCHECK(epoll_.is_valid());
- wake_event_.reset(eventfd(0, EFD_NONBLOCK));
- PCHECK(wake_event_.is_valid());
- epoll_event wake{.events = EPOLLIN, .data = {.ptr = &wake_event_}};
- int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_ADD, wake_event_.get(), &wake);
- PCHECK(rv == 0);
- }
- MessagePumpEpoll::~MessagePumpEpoll() = default;
- bool MessagePumpEpoll::WatchFileDescriptor(int fd,
- bool persistent,
- int mode,
- FdWatchController* controller,
- FdWatcher* watcher) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- TRACE_EVENT("base", "MessagePumpEpoll::WatchFileDescriptor", "fd", fd,
- "persistent", persistent, "watch_read", mode & WATCH_READ,
- "watch_write", mode & WATCH_WRITE);
- const InterestParams params{
- .fd = fd,
- .read = (mode == WATCH_READ || mode == WATCH_READ_WRITE),
- .write = (mode == WATCH_WRITE || mode == WATCH_READ_WRITE),
- .one_shot = !persistent,
- };
- auto [it, is_new_fd_entry] = entries_.emplace(fd, fd);
- EpollEventEntry& entry = it->second;
- scoped_refptr<Interest> existing_interest = controller->epoll_interest();
- if (existing_interest && existing_interest->params().IsEqual(params)) {
- // WatchFileDescriptor() has already been called for this controller at
- // least once before, and as in the most common cases, it is now being
- // called again with the same parameters.
- //
- // We don't need to allocate and register a new Interest in this case, but
- // we can instead reactivate the existing (presumably deactivated,
- // non-persistent) Interest.
- existing_interest->set_active(true);
- } else {
- entry.interests->push_back(controller->AssignEpollInterest(params));
- if (existing_interest) {
- UnregisterInterest(existing_interest);
- }
- }
- if (is_new_fd_entry) {
- AddEpollEvent(entry);
- } else {
- UpdateEpollEvent(entry);
- }
- controller->set_epoll_pump(weak_ptr_factory_.GetWeakPtr());
- controller->set_watcher(watcher);
- return true;
- }
- void MessagePumpEpoll::Run(Delegate* delegate) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- RunState run_state(delegate);
- AutoReset<RunState*> auto_reset_run_state(&run_state_, &run_state);
- for (;;) {
- // Do some work and see if the next task is ready right away.
- Delegate::NextWorkInfo next_work_info = delegate->DoWork();
- const bool immediate_work_available = next_work_info.is_immediate();
- if (run_state.should_quit) {
- break;
- }
- // Process any immediately ready IO event, but don't wait for more yet.
- const bool processed_events = WaitForEpollEvent(TimeDelta());
- if (run_state.should_quit) {
- break;
- }
- if (immediate_work_available || processed_events) {
- continue;
- }
- const bool did_idle_work = delegate->DoIdleWork();
- if (run_state.should_quit) {
- break;
- }
- if (did_idle_work) {
- continue;
- }
- TimeDelta timeout = TimeDelta::Max();
- DCHECK(!next_work_info.delayed_run_time.is_null());
- if (!next_work_info.delayed_run_time.is_max()) {
- timeout = next_work_info.remaining_delay();
- }
- delegate->BeforeWait();
- WaitForEpollEvent(timeout);
- if (run_state.should_quit) {
- break;
- }
- }
- }
- void MessagePumpEpoll::Quit() {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- DCHECK(run_state_) << "Quit() called outside of Run()";
- run_state_->should_quit = true;
- }
- void MessagePumpEpoll::ScheduleWork() {
- const uint64_t value = 1;
- ssize_t n = HANDLE_EINTR(write(wake_event_.get(), &value, sizeof(value)));
- // EAGAIN here implies that the write() would overflow of the event counter,
- // which is a condition we can safely ignore. It implies that the event
- // counter is non-zero and therefore readable, which is enough to ensure that
- // any pending wait eventually wakes up.
- DPCHECK(n == sizeof(value) || errno == EAGAIN);
- }
- void MessagePumpEpoll::ScheduleDelayedWork(
- const Delegate::NextWorkInfo& next_work_info) {
- // Nothing to do. This can only be called from the same thread as Run(), so
- // the pump must be in between waits. The scheduled work therefore will be
- // seen in time for the next wait.
- }
- void MessagePumpEpoll::AddEpollEvent(EpollEventEntry& entry) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- const uint32_t events = entry.ComputeActiveEvents();
- epoll_event event{.events = events, .data = {.ptr = &entry}};
- int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_ADD, entry.fd, &event);
- DPCHECK(rv == 0);
- entry.registered_events = events;
- }
- void MessagePumpEpoll::UpdateEpollEvent(EpollEventEntry& entry) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- const uint32_t events = entry.ComputeActiveEvents();
- if (events == entry.registered_events && !(events & EPOLLONESHOT)) {
- // Persistent events don't need to be modified if no bits are changing.
- return;
- }
- epoll_event event{.events = events, .data = {.ptr = &entry}};
- int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_MOD, entry.fd, &event);
- DPCHECK(rv == 0);
- entry.registered_events = events;
- }
- void MessagePumpEpoll::UnregisterInterest(
- const scoped_refptr<Interest>& interest) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- const int fd = interest->params().fd;
- auto entry_it = entries_.find(fd);
- DCHECK(entry_it != entries_.end());
- EpollEventEntry& entry = entry_it->second;
- auto& interests = entry.interests.container();
- auto it = std::find(interests.begin(), interests.end(), interest);
- DCHECK(it != interests.end());
- interests.erase(it);
- if (interests.empty()) {
- entries_.erase(entry_it);
- int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_DEL, fd, nullptr);
- DPCHECK(rv == 0);
- } else {
- UpdateEpollEvent(entry);
- }
- }
- bool MessagePumpEpoll::WaitForEpollEvent(TimeDelta timeout) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- const int epoll_timeout =
- timeout.is_max() ? -1 : saturated_cast<int>(timeout.InMilliseconds());
- epoll_event event;
- const int epoll_result =
- epoll_wait(epoll_.get(), &event, /*maxevents=*/1, epoll_timeout);
- if (epoll_result < 0) {
- DPCHECK(errno == EINTR);
- return false;
- }
- if (epoll_result == 0) {
- return false;
- }
- DPCHECK(epoll_result == 1);
- OnEpollEvent(event);
- return true;
- }
- void MessagePumpEpoll::OnEpollEvent(const epoll_event& e) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- if (e.data.ptr == &wake_event_) {
- HandleWakeUp();
- return;
- }
- const bool readable = (e.events & EPOLLIN) != 0;
- const bool writable = (e.events & EPOLLOUT) != 0;
- // Under different circumstances, peer closure may raise both/either EPOLLHUP
- // and/or EPOLLERR. Treat them as equivalent.
- const bool disconnected = (e.events & (EPOLLHUP | EPOLLERR)) != 0;
- DCHECK(e.data.ptr);
- auto& entry = *static_cast<EpollEventEntry*>(e.data.ptr);
- // Copy the set of Interests, since interests may be added to or removed from
- // `entry` during the loop below. This copy is inexpensive in practice
- // because the size of this vector is expected to be very small (<= 2).
- auto interests = entry.interests;
- for (const auto& interest : interests.container()) {
- if (!interest->active()) {
- continue;
- }
- const bool can_read = (readable || disconnected) && interest->params().read;
- const bool can_write = writable && interest->params().write;
- if (!can_read && !can_write) {
- // If this Interest is active but not watching for whichever event was
- // raised here, there's nothing to do. This can occur if a descriptor has
- // multiple active interests, since only one interest needs to be
- // satisfied in order for us to process an epoll event.
- continue;
- }
- if (interest->params().one_shot) {
- // This is a one-shot event watch which is about to be triggered. We
- // deactivate the interest and update epoll immediately. The event handler
- // may reactivate it.
- interest->set_active(false);
- UpdateEpollEvent(entry);
- }
- HandleEvent(entry.fd, can_read, can_write, interest->controller());
- }
- }
- void MessagePumpEpoll::HandleEvent(int fd,
- bool can_read,
- bool can_write,
- FdWatchController* controller) {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- // Make the MessagePumpDelegate aware of this other form of "DoWork". Skip if
- // HandleNotification() is called outside of Run() (e.g. in unit tests).
- Delegate::ScopedDoWorkItem scoped_do_work_item;
- if (run_state_) {
- scoped_do_work_item = run_state_->delegate->BeginWorkItem();
- }
- // Trace events must begin after the above BeginWorkItem() so that the
- // ensuing "ThreadController active" outscopes all the events under it.
- TRACE_EVENT("toplevel", "EpollEvent", "controller_created_from",
- controller->created_from_location(), "fd", fd, "can_read",
- can_read, "can_write", can_write, "context",
- static_cast<void*>(controller));
- TRACE_HEAP_PROFILER_API_SCOPED_TASK_EXECUTION heap_profiler_scope(
- controller->created_from_location().file_name());
- if (can_read && can_write) {
- bool controller_was_destroyed = false;
- controller->was_destroyed_ = &controller_was_destroyed;
- controller->OnFdWritable();
- if (!controller_was_destroyed) {
- controller->OnFdReadable();
- }
- if (!controller_was_destroyed) {
- controller->was_destroyed_ = nullptr;
- }
- } else if (can_write) {
- controller->OnFdWritable();
- } else if (can_read) {
- controller->OnFdReadable();
- }
- }
- void MessagePumpEpoll::HandleWakeUp() {
- DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
- uint64_t value;
- ssize_t n = HANDLE_EINTR(read(wake_event_.get(), &value, sizeof(value)));
- DPCHECK(n == sizeof(value));
- }
- MessagePumpEpoll::EpollEventEntry::EpollEventEntry(int fd) : fd(fd) {}
- MessagePumpEpoll::EpollEventEntry::~EpollEventEntry() = default;
- uint32_t MessagePumpEpoll::EpollEventEntry::ComputeActiveEvents() {
- uint32_t events = 0;
- bool one_shot = true;
- for (const auto& interest : interests.container()) {
- if (!interest->active()) {
- continue;
- }
- const InterestParams& params = interest->params();
- events |= (params.read ? EPOLLIN : 0) | (params.write ? EPOLLOUT : 0);
- one_shot &= params.one_shot;
- }
- if (events != 0 && one_shot) {
- return events | EPOLLONESHOT;
- }
- return events;
- }
- } // namespace base
|