Add a poll(2)-based event loop implementation.
Currently, all of our code is hardcoded to assume EpollServer is used for the events I/O. This is not cross-platform, since epoll is only supported on Linux. This adds an API that abstracts the event loop away, and a default implementation that uses poll(2), which is the most cross-platform an API can get for this purpose.
PiperOrigin-RevId: 454932600
diff --git a/build/source_list.bzl b/build/source_list.bzl
index 8b92c14..551b30a 100644
--- a/build/source_list.bzl
+++ b/build/source_list.bzl
@@ -234,6 +234,8 @@
"quic/core/http/spdy_utils.h",
"quic/core/http/web_transport_http3.h",
"quic/core/http/web_transport_stream_adapter.h",
+ "quic/core/io/quic_default_event_loop.h",
+ "quic/core/io/quic_event_loop.h",
"quic/core/legacy_quic_stream_id_manager.h",
"quic/core/packet_number_indexed_queue.h",
"quic/core/proto/cached_network_parameters_proto.h",
@@ -569,6 +571,7 @@
"quic/core/http/spdy_utils.cc",
"quic/core/http/web_transport_http3.cc",
"quic/core/http/web_transport_stream_adapter.cc",
+ "quic/core/io/quic_default_event_loop.cc",
"quic/core/legacy_quic_stream_id_manager.cc",
"quic/core/qpack/qpack_blocking_manager.cc",
"quic/core/qpack/qpack_decoded_headers_accumulator.cc",
@@ -1150,6 +1153,7 @@
"quic/core/http/spdy_server_push_utils_test.cc",
"quic/core/http/spdy_utils_test.cc",
"quic/core/http/web_transport_http3_test.cc",
+ "quic/core/io/quic_default_event_loop_test.cc",
"quic/core/legacy_quic_stream_id_manager_test.cc",
"quic/core/packet_number_indexed_queue_test.cc",
"quic/core/qpack/qpack_blocking_manager_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni
index 4b2f8f8..934a09d 100644
--- a/build/source_list.gni
+++ b/build/source_list.gni
@@ -234,6 +234,8 @@
"src/quiche/quic/core/http/spdy_utils.h",
"src/quiche/quic/core/http/web_transport_http3.h",
"src/quiche/quic/core/http/web_transport_stream_adapter.h",
+ "src/quiche/quic/core/io/quic_default_event_loop.h",
+ "src/quiche/quic/core/io/quic_event_loop.h",
"src/quiche/quic/core/legacy_quic_stream_id_manager.h",
"src/quiche/quic/core/packet_number_indexed_queue.h",
"src/quiche/quic/core/proto/cached_network_parameters_proto.h",
@@ -569,6 +571,7 @@
"src/quiche/quic/core/http/spdy_utils.cc",
"src/quiche/quic/core/http/web_transport_http3.cc",
"src/quiche/quic/core/http/web_transport_stream_adapter.cc",
+ "src/quiche/quic/core/io/quic_default_event_loop.cc",
"src/quiche/quic/core/legacy_quic_stream_id_manager.cc",
"src/quiche/quic/core/qpack/qpack_blocking_manager.cc",
"src/quiche/quic/core/qpack/qpack_decoded_headers_accumulator.cc",
@@ -1150,6 +1153,7 @@
"src/quiche/quic/core/http/spdy_server_push_utils_test.cc",
"src/quiche/quic/core/http/spdy_utils_test.cc",
"src/quiche/quic/core/http/web_transport_http3_test.cc",
+ "src/quiche/quic/core/io/quic_default_event_loop_test.cc",
"src/quiche/quic/core/legacy_quic_stream_id_manager_test.cc",
"src/quiche/quic/core/packet_number_indexed_queue_test.cc",
"src/quiche/quic/core/qpack/qpack_blocking_manager_test.cc",
diff --git a/build/source_list.json b/build/source_list.json
index 37a1d63..a93bfae 100644
--- a/build/source_list.json
+++ b/build/source_list.json
@@ -233,6 +233,8 @@
"quiche/quic/core/http/spdy_utils.h",
"quiche/quic/core/http/web_transport_http3.h",
"quiche/quic/core/http/web_transport_stream_adapter.h",
+ "quiche/quic/core/io/quic_default_event_loop.h",
+ "quiche/quic/core/io/quic_event_loop.h",
"quiche/quic/core/legacy_quic_stream_id_manager.h",
"quiche/quic/core/packet_number_indexed_queue.h",
"quiche/quic/core/proto/cached_network_parameters_proto.h",
@@ -568,6 +570,7 @@
"quiche/quic/core/http/spdy_utils.cc",
"quiche/quic/core/http/web_transport_http3.cc",
"quiche/quic/core/http/web_transport_stream_adapter.cc",
+ "quiche/quic/core/io/quic_default_event_loop.cc",
"quiche/quic/core/legacy_quic_stream_id_manager.cc",
"quiche/quic/core/qpack/qpack_blocking_manager.cc",
"quiche/quic/core/qpack/qpack_decoded_headers_accumulator.cc",
@@ -1149,6 +1152,7 @@
"quiche/quic/core/http/spdy_server_push_utils_test.cc",
"quiche/quic/core/http/spdy_utils_test.cc",
"quiche/quic/core/http/web_transport_http3_test.cc",
+ "quiche/quic/core/io/quic_default_event_loop_test.cc",
"quiche/quic/core/legacy_quic_stream_id_manager_test.cc",
"quiche/quic/core/packet_number_indexed_queue_test.cc",
"quiche/quic/core/qpack/qpack_blocking_manager_test.cc",
diff --git a/quiche/quic/core/io/quic_default_event_loop.cc b/quiche/quic/core/io/quic_default_event_loop.cc
new file mode 100644
index 0000000..bd7be5a
--- /dev/null
+++ b/quiche/quic/core/io/quic_default_event_loop.cc
@@ -0,0 +1,267 @@
+// 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 "quiche/quic/core/io/quic_default_event_loop.h"
+
+#include <poll.h>
+
+#include <algorithm>
+#include <cerrno>
+#include <memory>
+
+#include "absl/types/span.h"
+#include "quiche/quic/core/io/quic_event_loop.h"
+#include "quiche/quic/core/quic_alarm.h"
+#include "quiche/quic/core/quic_time.h"
+#include "quiche/quic/platform/api/quic_bug_tracker.h"
+
+namespace quic {
+
+namespace {
+
+using PollMask = decltype(::pollfd().events);
+
+PollMask GetPollMask(QuicSocketEventMask event_mask) {
+ return ((event_mask & kSocketEventReadable) ? POLLIN : 0) |
+ ((event_mask & kSocketEventWritable) ? POLLOUT : 0) |
+ ((event_mask & kSocketEventError) ? POLLERR : 0);
+}
+
+QuicSocketEventMask GetEventMask(PollMask poll_mask) {
+ return ((poll_mask & POLLIN) ? kSocketEventReadable : 0) |
+ ((poll_mask & POLLOUT) ? kSocketEventWritable : 0) |
+ ((poll_mask & POLLERR) ? kSocketEventError : 0);
+}
+
+} // namespace
+
+QuicDefaultEventLoop::QuicDefaultEventLoop(QuicClock* clock)
+ : clock_(clock), alarm_factory_(this) {}
+
+bool QuicDefaultEventLoop::RegisterSocket(QuicUdpSocketFd fd,
+ QuicSocketEventMask events,
+ QuicSocketEventListener* listener) {
+ auto [it, success] =
+ registrations_.insert({fd, std::make_shared<Registration>()});
+ if (!success) {
+ return false;
+ }
+ Registration& registration = *it->second;
+ registration.events = events;
+ registration.listener = listener;
+ return true;
+}
+
+bool QuicDefaultEventLoop::UnregisterSocket(QuicUdpSocketFd fd) {
+ return registrations_.erase(fd);
+}
+
+bool QuicDefaultEventLoop::RearmSocket(QuicUdpSocketFd fd,
+ QuicSocketEventMask events) {
+ auto it = registrations_.find(fd);
+ if (it == registrations_.end()) {
+ return false;
+ }
+ it->second->events |= events;
+ return true;
+}
+
+bool QuicDefaultEventLoop::ArtificiallyNotifyEvent(QuicUdpSocketFd fd,
+ QuicSocketEventMask events) {
+ auto it = registrations_.find(fd);
+ if (it == registrations_.end()) {
+ return false;
+ }
+ has_artificial_events_pending_ = true;
+ it->second->artificially_notify_at_next_iteration |= events;
+ return true;
+}
+
+void QuicDefaultEventLoop::RunEventLoopOnce(QuicTime::Delta default_timeout) {
+ const QuicTime start_time = clock_->Now();
+ ProcessAlarmsUpTo(start_time);
+
+ QuicTime::Delta timeout = ComputePollTimeout(start_time, default_timeout);
+ ProcessIoEvents(start_time, timeout);
+
+ const QuicTime end_time = clock_->Now();
+ ProcessAlarmsUpTo(end_time);
+}
+
+QuicTime::Delta QuicDefaultEventLoop::ComputePollTimeout(
+ QuicTime now, QuicTime::Delta default_timeout) const {
+ if (has_artificial_events_pending_) {
+ return QuicTime::Delta::Zero();
+ }
+ if (alarms_.empty()) {
+ return default_timeout;
+ }
+ QuicTime end_time = std::min(now + default_timeout, alarms_.begin()->first);
+ if (end_time <= now) {
+ // Since we call ProcessAlarmsUpTo() right before this, this should never
+ // happen.
+ QUIC_BUG(Newest alarm is in the past)
+ << "now " << now.ToDebuggingValue()
+ << ", end_time: " << end_time.ToDebuggingValue();
+ return QuicTime::Delta::Zero();
+ }
+ return end_time - now;
+}
+
+int QuicDefaultEventLoop::PollWithRetries(absl::Span<pollfd> fds,
+ QuicTime start_time,
+ QuicTime::Delta timeout) {
+ const QuicTime timeout_at = start_time + timeout;
+ int poll_result;
+ for (;;) {
+ float timeout_ms = std::ceil(timeout.ToMicroseconds() / 1000.f);
+ poll_result =
+ PollSyscall(fds.data(), fds.size(), static_cast<int>(timeout_ms));
+
+ // Retry if EINTR happens.
+ bool is_eintr = poll_result < 0 && errno == EINTR;
+ if (!is_eintr) {
+ break;
+ }
+ QuicTime now = clock_->Now();
+ if (now >= timeout_at) {
+ break;
+ }
+ timeout = timeout_at - now;
+ }
+ return poll_result;
+}
+
+void QuicDefaultEventLoop::ProcessIoEvents(QuicTime start_time,
+ QuicTime::Delta timeout) {
+ // Set up the pollfd[] array.
+ const size_t registration_count = registrations_.size();
+ auto pollfds = std::make_unique<pollfd[]>(registration_count);
+ size_t i = 0;
+ for (auto& [fd, registration] : registrations_) {
+ QUICHE_CHECK_LT(
+ i, registration_count); // Crash instead of out-of-bounds access.
+ pollfds[i].fd = fd;
+ pollfds[i].events = GetPollMask(registration->events);
+ pollfds[i].revents = 0;
+ ++i;
+ }
+
+ // Actually run poll(2).
+ int poll_result =
+ PollWithRetries(absl::Span<pollfd>(pollfds.get(), registration_count),
+ start_time, timeout);
+ if (poll_result == 0) {
+ return;
+ }
+
+ // Prepare the list of all callbacks to be called, while resetting all events,
+ // since we're operating in the level-triggered mode.
+ std::vector<ReadyListEntry> ready_list;
+ ready_list.reserve(registration_count);
+ for (i = 0; i < registration_count; i++) {
+ DispatchIoEvent(ready_list, pollfds[i].fd, pollfds[i].revents);
+ }
+ has_artificial_events_pending_ = false;
+
+ // Actually call all of the callbacks.
+ RunReadyCallbacks(ready_list);
+}
+
+void QuicDefaultEventLoop::DispatchIoEvent(
+ std::vector<ReadyListEntry>& ready_list, QuicUdpSocketFd fd,
+ PollMask mask) {
+ auto it = registrations_.find(fd);
+ if (it == registrations_.end()) {
+ QUIC_BUG(poll returned an unregistered fd) << fd;
+ return;
+ }
+ Registration& registration = *it->second;
+
+ mask |= GetPollMask(registration.artificially_notify_at_next_iteration);
+ registration.artificially_notify_at_next_iteration = QuicSocketEventMask();
+
+ // poll() always returns certain classes of events even if not requested.
+ mask &= GetPollMask(registration.events);
+ if (!mask) {
+ return;
+ }
+
+ ready_list.push_back(ReadyListEntry{fd, it->second, GetEventMask(mask)});
+ registration.events &= ~GetEventMask(mask);
+}
+
+void QuicDefaultEventLoop::RunReadyCallbacks(
+ std::vector<ReadyListEntry>& ready_list) {
+ for (ReadyListEntry& entry : ready_list) {
+ std::shared_ptr<Registration> registration = entry.registration.lock();
+ if (!registration) {
+ // The socket has been unregistered from within one of the callbacks.
+ continue;
+ }
+ registration->listener->OnSocketEvent(this, entry.fd, entry.events);
+ }
+ ready_list.clear();
+}
+
+void QuicDefaultEventLoop::ProcessAlarmsUpTo(QuicTime time) {
+ // Determine which alarm callbacks needs to be run.
+ std::vector<std::weak_ptr<Alarm*>> alarms_to_call;
+ while (!alarms_.empty() && alarms_.begin()->first <= time) {
+ auto& [deadline, schedule_handle_weak] = *alarms_.begin();
+ alarms_to_call.push_back(std::move(schedule_handle_weak));
+ alarms_.erase(alarms_.begin());
+ }
+ // Actually run those callbacks.
+ for (std::weak_ptr<Alarm*>& schedule_handle_weak : alarms_to_call) {
+ std::shared_ptr<Alarm*> schedule_handle = schedule_handle_weak.lock();
+ if (!schedule_handle) {
+ // The alarm has been cancelled and might not even exist anymore.
+ continue;
+ }
+ (*schedule_handle)->DoFire();
+ }
+ // Clean up all of the alarms in the front that have been cancelled.
+ while (!alarms_.empty()) {
+ if (alarms_.begin()->second.expired()) {
+ alarms_.erase(alarms_.begin());
+ } else {
+ break;
+ }
+ }
+}
+
+QuicAlarm* QuicDefaultEventLoop::AlarmFactory::CreateAlarm(
+ QuicAlarm::Delegate* delegate) {
+ return new Alarm(loop_, QuicArenaScopedPtr<QuicAlarm::Delegate>(delegate));
+}
+
+QuicArenaScopedPtr<QuicAlarm> QuicDefaultEventLoop::AlarmFactory::CreateAlarm(
+ QuicArenaScopedPtr<QuicAlarm::Delegate> delegate,
+ QuicConnectionArena* arena) {
+ if (arena != nullptr) {
+ return arena->New<Alarm>(loop_, std::move(delegate));
+ }
+ return QuicArenaScopedPtr<QuicAlarm>(new Alarm(loop_, std::move(delegate)));
+}
+
+QuicDefaultEventLoop::Alarm::Alarm(
+ QuicDefaultEventLoop* loop,
+ QuicArenaScopedPtr<QuicAlarm::Delegate> delegate)
+ : QuicAlarm(std::move(delegate)), loop_(loop) {}
+
+void QuicDefaultEventLoop::Alarm::SetImpl() {
+ current_schedule_handle_ = std::make_shared<Alarm*>(this);
+ loop_->alarms_.insert({deadline(), current_schedule_handle_});
+}
+
+void QuicDefaultEventLoop::Alarm::CancelImpl() {
+ current_schedule_handle_.reset();
+}
+
+QuicAlarmFactory* QuicDefaultEventLoop::GetAlarmFactory() {
+ return &alarm_factory_;
+}
+
+} // namespace quic
diff --git a/quiche/quic/core/io/quic_default_event_loop.h b/quiche/quic/core/io/quic_default_event_loop.h
new file mode 100644
index 0000000..ca2a958
--- /dev/null
+++ b/quiche/quic/core/io/quic_default_event_loop.h
@@ -0,0 +1,152 @@
+// 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.
+
+#ifndef QUICHE_QUIC_CORE_IO_QUIC_DEFAULT_EVENT_LOOP_H_
+#define QUICHE_QUIC_CORE_IO_QUIC_DEFAULT_EVENT_LOOP_H_
+
+#include <poll.h>
+
+#include <memory>
+
+#include "absl/container/btree_map.h"
+#include "absl/types/span.h"
+#include "quiche/quic/core/io/quic_event_loop.h"
+#include "quiche/quic/core/quic_alarm.h"
+#include "quiche/quic/core/quic_alarm_factory.h"
+#include "quiche/quic/core/quic_clock.h"
+#include "quiche/quic/core/quic_udp_socket.h"
+#include "quiche/common/quiche_linked_hash_map.h"
+
+namespace quic {
+
+// A simple and portable implementation of QuicEventLoop using poll(2). Works
+// on all POSIX platforms (and can be potentially made to support Windows using
+// WSAPoll).
+//
+// For most operations, this implementation has a typical runtime of
+// O(N + log M), where N is the number of file descriptors, and M is the number
+// of pending alarms.
+//
+// This API has to deal with the situations where callbacks are modified from
+// the callbacks themselves. To address this, we use the following two
+// approaches:
+// 1. The code does not execute any callbacks until the very end of the
+// processing, when all of the state for the event loop is consistent.
+// 2. The callbacks are stored as weak pointers, since other callbacks can
+// cause them to be unregistered.
+class QUICHE_NO_EXPORT QuicDefaultEventLoop : public QuicEventLoop {
+ public:
+ QuicDefaultEventLoop(QuicClock* clock);
+
+ // QuicEventLoop implementation.
+ bool SupportsEdgeTriggered() const override { return false; }
+ ABSL_MUST_USE_RESULT bool RegisterSocket(
+ QuicUdpSocketFd fd, QuicSocketEventMask events,
+ QuicSocketEventListener* listener) override;
+ ABSL_MUST_USE_RESULT bool UnregisterSocket(QuicUdpSocketFd fd) override;
+ ABSL_MUST_USE_RESULT bool RearmSocket(QuicUdpSocketFd fd,
+ QuicSocketEventMask events) override;
+ ABSL_MUST_USE_RESULT bool ArtificiallyNotifyEvent(
+ QuicUdpSocketFd fd, QuicSocketEventMask events) override;
+ void RunEventLoopOnce(QuicTime::Delta default_timeout) override;
+ QuicAlarmFactory* GetAlarmFactory() override;
+
+ protected:
+ // Allows poll(2) calls to be mocked out in unit tests.
+ virtual int PollSyscall(pollfd* fds, nfds_t nfds, int timeout) {
+ return ::poll(fds, nfds, timeout);
+ }
+
+ private:
+ friend class QuicDefaultEventLoopPeer;
+
+ struct Registration {
+ QuicSocketEventMask events = 0;
+ QuicSocketEventListener* listener;
+
+ QuicSocketEventMask artificially_notify_at_next_iteration = 0;
+ };
+
+ class Alarm : public QuicAlarm {
+ public:
+ Alarm(QuicDefaultEventLoop* loop,
+ QuicArenaScopedPtr<QuicAlarm::Delegate> delegate);
+
+ void SetImpl() override;
+ void CancelImpl() override;
+
+ void DoFire() {
+ current_schedule_handle_.reset();
+ Fire();
+ }
+
+ private:
+ QuicDefaultEventLoop* loop_;
+ // Deleted when the alarm is cancelled, causing the corresponding weak_ptr
+ // in the alarm list to not be executed.
+ std::shared_ptr<Alarm*> current_schedule_handle_;
+ };
+
+ class AlarmFactory : public QuicAlarmFactory {
+ public:
+ AlarmFactory(QuicDefaultEventLoop* loop) : loop_(loop) {}
+
+ // QuicAlarmFactory implementation.
+ QuicAlarm* CreateAlarm(QuicAlarm::Delegate* delegate) override;
+ QuicArenaScopedPtr<QuicAlarm> CreateAlarm(
+ QuicArenaScopedPtr<QuicAlarm::Delegate> delegate,
+ QuicConnectionArena* arena) override;
+
+ private:
+ QuicDefaultEventLoop* loop_;
+ };
+
+ // Used for deferred execution of I/O callbacks.
+ struct ReadyListEntry {
+ QuicUdpSocketFd fd;
+ std::weak_ptr<Registration> registration;
+ QuicSocketEventMask events;
+ };
+
+ // We're using a linked hash map here to ensure the events are called in the
+ // registration order. This isn't strictly speaking necessary, but makes
+ // testing things easier.
+ using RegistrationMap =
+ quiche::QuicheLinkedHashMap<QuicUdpSocketFd,
+ std::shared_ptr<Registration>>;
+ // Alarms are stored as weak pointers, since the alarm can be cancelled and
+ // disappear while in the queue.
+ using AlarmList = absl::btree_multimap<QuicTime, std::weak_ptr<Alarm*>>;
+
+ // Returns the timeout for the next poll(2) call. It is typically the time at
+ // which the next alarm is supposed to activate.
+ QuicTime::Delta ComputePollTimeout(QuicTime now,
+ QuicTime::Delta default_timeout) const;
+ // Calls poll(2) with the provided timeout and dispatches the callbacks
+ // accordingly.
+ void ProcessIoEvents(QuicTime start_time, QuicTime::Delta timeout);
+ // Calls all of the alarm callbacks that are scheduled before or at |time|.
+ void ProcessAlarmsUpTo(QuicTime time);
+
+ // Adds the I/O callbacks for |fd| to the |ready_lits| as appopriate.
+ void DispatchIoEvent(std::vector<ReadyListEntry>& ready_list,
+ QuicUdpSocketFd fd, short mask); // NOLINT(runtime/int)
+ // Runs all of the callbacks on the ready list.
+ void RunReadyCallbacks(std::vector<ReadyListEntry>& ready_list);
+
+ // Calls poll() while handling EINTR. Returns the return value of poll(2)
+ // system call.
+ int PollWithRetries(absl::Span<pollfd> fds, QuicTime start_time,
+ QuicTime::Delta timeout);
+
+ const QuicClock* clock_;
+ RegistrationMap registrations_;
+ AlarmList alarms_;
+ AlarmFactory alarm_factory_;
+ bool has_artificial_events_pending_ = false;
+};
+
+} // namespace quic
+
+#endif // QUICHE_QUIC_CORE_IO_QUIC_DEFAULT_EVENT_LOOP_H_
diff --git a/quiche/quic/core/io/quic_default_event_loop_test.cc b/quiche/quic/core/io/quic_default_event_loop_test.cc
new file mode 100644
index 0000000..a530811
--- /dev/null
+++ b/quiche/quic/core/io/quic_default_event_loop_test.cc
@@ -0,0 +1,342 @@
+// 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 "quiche/quic/core/io/quic_default_event_loop.h"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <memory>
+#include <vector>
+
+#include "absl/memory/memory.h"
+#include "quiche/quic/core/io/quic_event_loop.h"
+#include "quiche/quic/core/quic_alarm.h"
+#include "quiche/quic/core/quic_alarm_factory.h"
+#include "quiche/quic/core/quic_time.h"
+#include "quiche/quic/platform/api/quic_test.h"
+#include "quiche/quic/test_tools/mock_clock.h"
+
+namespace quic {
+
+class QuicDefaultEventLoopPeer {
+ public:
+ static QuicTime::Delta ComputePollTimeout(const QuicDefaultEventLoop& loop,
+ QuicTime now,
+ QuicTime::Delta default_timeout) {
+ return loop.ComputePollTimeout(now, default_timeout);
+ }
+};
+
+} // namespace quic
+
+namespace quic::test {
+namespace {
+
+using testing::_;
+using testing::AtMost;
+using testing::ElementsAre;
+
+constexpr QuicSocketEventMask kAllEvents =
+ kSocketEventReadable | kSocketEventWritable | kSocketEventError;
+constexpr QuicTime::Delta kDefaultTimeout = QuicTime::Delta::FromSeconds(100);
+
+class MockQuicSocketEventListener : public QuicSocketEventListener {
+ public:
+ MOCK_METHOD(void, OnSocketEvent,
+ (QuicEventLoop* /*event_loop*/, QuicUdpSocketFd /*fd*/,
+ QuicSocketEventMask /*events*/),
+ (override));
+};
+
+class MockDelegate : public QuicAlarm::Delegate {
+ public:
+ QuicConnectionContext* GetConnectionContext() override { return nullptr; }
+ MOCK_METHOD(void, OnAlarm, (), (override));
+};
+
+class QuicDefaultEventLoopForTest : public QuicDefaultEventLoop {
+ public:
+ QuicDefaultEventLoopForTest(MockClock* clock)
+ : QuicDefaultEventLoop(clock), clock_(clock) {}
+
+ int PollSyscall(pollfd* fds, nfds_t nfds, int timeout) override {
+ timeouts_.push_back(timeout);
+ if (eintr_after_ != QuicTime::Delta::Infinite()) {
+ errno = EINTR;
+ clock_->AdvanceTime(eintr_after_);
+ eintr_after_ = QuicTime::Delta::Infinite();
+ return -1;
+ }
+ clock_->AdvanceTime(QuicTime::Delta::FromMilliseconds(timeout));
+ return QuicDefaultEventLoop::PollSyscall(fds, nfds, timeout);
+ }
+
+ void TriggerEintrAfter(QuicTime::Delta time) { eintr_after_ = time; }
+
+ const std::vector<int>& timeouts() const { return timeouts_; }
+
+ private:
+ MockClock* clock_;
+ QuicTime::Delta eintr_after_ = QuicTime::Delta::Infinite();
+ std::vector<int> timeouts_;
+};
+
+class QuicDefaultEventLoopTest : public QuicTest {
+ public:
+ QuicDefaultEventLoopTest()
+ : loop_(&clock_), factory_(loop_.GetAlarmFactory()) {
+ int fds[2];
+ int result = ::pipe(fds);
+ QUICHE_CHECK(result >= 0) << "Failed to create a pipe, errno: " << errno;
+ read_fd_ = fds[0];
+ write_fd_ = fds[1];
+
+ QUICHE_CHECK(::fcntl(read_fd_, F_SETFL,
+ ::fcntl(read_fd_, F_GETFL) | O_NONBLOCK) == 0)
+ << "Failed to mark pipe FD non-blocking, errno: " << errno;
+ QUICHE_CHECK(::fcntl(write_fd_, F_SETFL,
+ ::fcntl(write_fd_, F_GETFL) | O_NONBLOCK) == 0)
+ << "Failed to mark pipe FD non-blocking, errno: " << errno;
+
+ clock_.AdvanceTime(10 * kDefaultTimeout);
+ }
+
+ ~QuicDefaultEventLoopTest() {
+ close(read_fd_);
+ close(write_fd_);
+ }
+
+ QuicTime::Delta ComputePollTimeout() {
+ return QuicDefaultEventLoopPeer::ComputePollTimeout(loop_, clock_.Now(),
+ kDefaultTimeout);
+ }
+
+ std::pair<std::unique_ptr<QuicAlarm>, MockDelegate*> CreateAlarm() {
+ auto delegate = std::make_unique<testing::StrictMock<MockDelegate>>();
+ MockDelegate* delegate_unowned = delegate.get();
+ auto alarm = absl::WrapUnique(factory_->CreateAlarm(delegate.release()));
+ return std::make_pair(std::move(alarm), delegate_unowned);
+ }
+
+ protected:
+ MockClock clock_;
+ QuicDefaultEventLoopForTest loop_;
+ QuicAlarmFactory* factory_;
+ int read_fd_;
+ int write_fd_;
+};
+
+TEST_F(QuicDefaultEventLoopTest, NothingHappens) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ // Attempt double-registration.
+ EXPECT_FALSE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(4));
+ // Expect no further calls.
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(5));
+ EXPECT_THAT(loop_.timeouts(), ElementsAre(4, 5));
+}
+
+TEST_F(QuicDefaultEventLoopTest, RearmWriter) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable))
+ .Times(2);
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+ ASSERT_TRUE(loop_.RearmSocket(write_fd_, kSocketEventWritable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+}
+
+TEST_F(QuicDefaultEventLoopTest, Readable) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+
+ ASSERT_EQ(4, write(write_fd_, "test", 4));
+ EXPECT_CALL(listener, OnSocketEvent(_, read_fd_, kSocketEventReadable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+ // Expect no further calls.
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+}
+
+TEST_F(QuicDefaultEventLoopTest, RearmReader) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+
+ ASSERT_EQ(4, write(write_fd_, "test", 4));
+ EXPECT_CALL(listener, OnSocketEvent(_, read_fd_, kSocketEventReadable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+ // Expect no further calls.
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+}
+
+TEST_F(QuicDefaultEventLoopTest, WriterUnblocked) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+
+ int io_result;
+ std::string data(2048, 'a');
+ do {
+ io_result = write(write_fd_, data.data(), data.size());
+ } while (io_result > 0);
+ ASSERT_EQ(errno, EAGAIN);
+
+ // Rearm and expect no immediate calls.
+ ASSERT_TRUE(loop_.RearmSocket(write_fd_, kSocketEventWritable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable));
+ do {
+ io_result = read(read_fd_, data.data(), data.size());
+ } while (io_result > 0);
+ ASSERT_EQ(errno, EAGAIN);
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+}
+
+TEST_F(QuicDefaultEventLoopTest, ArtificialEvent) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+ ASSERT_TRUE(loop_.ArtificiallyNotifyEvent(read_fd_, kSocketEventReadable));
+ EXPECT_EQ(ComputePollTimeout(), QuicTime::Delta::Zero());
+
+ {
+ testing::InSequence s;
+ EXPECT_CALL(listener, OnSocketEvent(_, read_fd_, kSocketEventReadable));
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable));
+ }
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+}
+
+TEST_F(QuicDefaultEventLoopTest, Unregister) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+ ASSERT_TRUE(loop_.UnregisterSocket(write_fd_));
+
+ // Expect nothing to happen.
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+
+ EXPECT_FALSE(loop_.UnregisterSocket(write_fd_));
+ EXPECT_FALSE(loop_.RearmSocket(write_fd_, kSocketEventWritable));
+ EXPECT_FALSE(loop_.ArtificiallyNotifyEvent(write_fd_, kSocketEventWritable));
+}
+
+TEST_F(QuicDefaultEventLoopTest, UnregisterInsideEventHandler) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+ ASSERT_TRUE(loop_.RegisterSocket(write_fd_, kAllEvents, &listener));
+
+ EXPECT_CALL(listener, OnSocketEvent(_, read_fd_, kSocketEventReadable))
+ .WillOnce([this]() { ASSERT_TRUE(loop_.UnregisterSocket(write_fd_)); });
+ EXPECT_CALL(listener, OnSocketEvent(_, write_fd_, kSocketEventWritable))
+ .Times(0);
+ ASSERT_TRUE(loop_.ArtificiallyNotifyEvent(read_fd_, kSocketEventReadable));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(1));
+}
+
+TEST_F(QuicDefaultEventLoopTest, EintrHandler) {
+ testing::StrictMock<MockQuicSocketEventListener> listener;
+ ASSERT_TRUE(loop_.RegisterSocket(read_fd_, kAllEvents, &listener));
+
+ loop_.TriggerEintrAfter(QuicTime::Delta::FromMilliseconds(25));
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(100));
+ EXPECT_THAT(loop_.timeouts(), ElementsAre(100, 75));
+}
+
+TEST_F(QuicDefaultEventLoopTest, AlarmInFuture) {
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ constexpr auto kAlarmTimeout = QuicTime::Delta::FromMilliseconds(5);
+ auto [alarm, delegate] = CreateAlarm();
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ alarm->Set(clock_.Now() + kAlarmTimeout);
+ EXPECT_EQ(ComputePollTimeout(), kAlarmTimeout);
+
+ EXPECT_CALL(*delegate, OnAlarm());
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(100));
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+}
+
+TEST_F(QuicDefaultEventLoopTest, AlarmsInPast) {
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ constexpr auto kAlarmTimeout = QuicTime::Delta::FromMilliseconds(5);
+ auto [alarm1, delegate1] = CreateAlarm();
+ auto [alarm2, delegate2] = CreateAlarm();
+
+ alarm1->Set(clock_.Now() - 2 * kAlarmTimeout);
+ alarm2->Set(clock_.Now() - kAlarmTimeout);
+
+ {
+ testing::InSequence s;
+ EXPECT_CALL(*delegate1, OnAlarm());
+ EXPECT_CALL(*delegate2, OnAlarm());
+ }
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(100));
+}
+
+TEST_F(QuicDefaultEventLoopTest, AlarmCancelled) {
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ constexpr auto kAlarmTimeout = QuicTime::Delta::FromMilliseconds(5);
+ auto [alarm, delegate] = CreateAlarm();
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ alarm->Set(clock_.Now() + kAlarmTimeout);
+ alarm->Cancel();
+ alarm->Set(clock_.Now() + 2 * kAlarmTimeout);
+ EXPECT_EQ(ComputePollTimeout(), kAlarmTimeout);
+
+ EXPECT_CALL(*delegate, OnAlarm());
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(100));
+ EXPECT_THAT(loop_.timeouts(), ElementsAre(10));
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+}
+
+TEST_F(QuicDefaultEventLoopTest, AlarmCancelsAnotherAlarm) {
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+
+ constexpr auto kAlarmTimeout = QuicTime::Delta::FromMilliseconds(5);
+ auto [alarm1_ptr, delegate1] = CreateAlarm();
+ auto [alarm2_ptr, delegate2] = CreateAlarm();
+
+ QuicAlarm& alarm1 = *alarm1_ptr;
+ QuicAlarm& alarm2 = *alarm2_ptr;
+ alarm1.Set(clock_.Now() - kAlarmTimeout);
+ alarm2.Set(clock_.Now() - kAlarmTimeout);
+
+ int alarms_called = 0;
+ // Since the order in which alarms are cancelled is not well-determined, make
+ // each one cancel another.
+ EXPECT_CALL(*delegate1, OnAlarm()).Times(AtMost(1)).WillOnce([&]() {
+ alarm2.Cancel();
+ ++alarms_called;
+ });
+ EXPECT_CALL(*delegate2, OnAlarm()).Times(AtMost(1)).WillOnce([&]() {
+ alarm1.Cancel();
+ ++alarms_called;
+ });
+ loop_.RunEventLoopOnce(QuicTime::Delta::FromMilliseconds(100));
+ EXPECT_EQ(alarms_called, 1);
+ EXPECT_EQ(ComputePollTimeout(), kDefaultTimeout);
+}
+
+} // namespace
+} // namespace quic::test
diff --git a/quiche/quic/core/io/quic_event_loop.h b/quiche/quic/core/io/quic_event_loop.h
new file mode 100644
index 0000000..cc41392
--- /dev/null
+++ b/quiche/quic/core/io/quic_event_loop.h
@@ -0,0 +1,81 @@
+// 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.
+
+#ifndef QUICHE_QUIC_IO_QUIC_EVENT_LOOP_H_
+#define QUICHE_QUIC_IO_QUIC_EVENT_LOOP_H_
+
+#include <cstdint>
+#include <memory>
+
+#include "absl/base/attributes.h"
+#include "quiche/quic/core/quic_alarm_factory.h"
+#include "quiche/quic/core/quic_udp_socket.h"
+
+namespace quic {
+
+// A bitmask indicating a set of I/O events.
+using QuicSocketEventMask = uint8_t;
+inline constexpr QuicSocketEventMask kSocketEventReadable = 0x01;
+inline constexpr QuicSocketEventMask kSocketEventWritable = 0x02;
+inline constexpr QuicSocketEventMask kSocketEventError = 0x04;
+
+class QuicEventLoop;
+
+// A listener associated with a file descriptor.
+class QUICHE_NO_EXPORT QuicSocketEventListener {
+ public:
+ virtual ~QuicSocketEventListener() = default;
+
+ virtual void OnSocketEvent(QuicEventLoop* event_loop, QuicUdpSocketFd fd,
+ QuicSocketEventMask events) = 0;
+};
+
+// An abstraction for an event loop that can handle alarms and notify the
+// listener about I/O events occuring to the registered UDP sockets.
+//
+// Note on error handling: while most of the methods below return a boolean to
+// indicate whether the operation has succeeded or not, some will QUIC_BUG
+// instead.
+class QUICHE_NO_EXPORT QuicEventLoop {
+ public:
+ virtual ~QuicEventLoop() = default;
+
+ // Indicates whether the event loop implementation supports edge-triggered
+ // notifications. If true, all of the events are permanent and are notified
+ // as long as they are registered. If false, whenever an event is triggered,
+ // the event registration is unset and has to be re-armed using RearmSocket().
+ virtual bool SupportsEdgeTriggered() const = 0;
+
+ // Watches for all of the requested |events| that occur on the |fd| and
+ // notifies the |listener| about them. |fd| must not be already registered;
+ // if it is, the function returns false. The |listener| must be alive for as
+ // long as it is registered.
+ virtual ABSL_MUST_USE_RESULT bool RegisterSocket(
+ QuicUdpSocketFd fd, QuicSocketEventMask events,
+ QuicSocketEventListener* listener) = 0;
+ // Removes the listener associated with |fd|. Returns false if the listener
+ // is not found.
+ virtual ABSL_MUST_USE_RESULT bool UnregisterSocket(QuicUdpSocketFd fd) = 0;
+ // Adds |events| to the list of the listened events for |fd|, given that |fd|
+ // is already registered. Must be only called if SupportsEdgeTriggered() is
+ // false.
+ virtual ABSL_MUST_USE_RESULT bool RearmSocket(QuicUdpSocketFd fd,
+ QuicSocketEventMask events) = 0;
+ // Causes the |fd| to be notified of |events| on the next event loop iteration
+ // even if none of the specified events has happened.
+ virtual ABSL_MUST_USE_RESULT bool ArtificiallyNotifyEvent(
+ QuicUdpSocketFd fd, QuicSocketEventMask events) = 0;
+
+ // Runs a single iteration of the event loop. The iteration will run for at
+ // most |default_timeout|.
+ virtual void RunEventLoopOnce(QuicTime::Delta default_timeout) = 0;
+
+ // Returns an alarm factory that allows alarms to be scheduled on this event
+ // loop. The factory is owned by the event loop.
+ virtual QuicAlarmFactory* GetAlarmFactory() = 0;
+};
+
+} // namespace quic
+
+#endif // QUICHE_QUIC_IO_QUIC_EVENT_LOOP_H_
