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_