Add an opt-in UDP GRO mode to QuicPacketReader. With GRO, the kernel can hand a socket back-to-back datagrams of the same size from the same peer as one GRO train, which a single read returns. Supporting UDP GRO on the QUIC socket requires to configure the reader and the socket at QUIC client creation time. For the reader case, `QuicPacketReader` cannot read GRO trains today because its packet buffers can only hold one datagram, and it would treat a train as one packet. This change adds `QuicPacketReader::Options::enable_gro`. When it is set, the reader uses packet buffers large enough for a whole train (64 KB per slot instead of 1,472 bytes, about 1 MB per reader), asks for the GRO segment size on each read, and dispatches each packet of a train separately, with the train's addresses, receipt time, and other per-datagram metadata. For the socket case, this adds `QuicUdpSocketApi::EnableUdpGro` to set the UDP_GRO socket option. Readers will only allocate the GRO-sized buffers when explicitly passing the option at creation time, so behavior is unchanged for existing users. PiperOrigin-RevId: 1019227871
diff --git a/build/source_list.bzl b/build/source_list.bzl index 1d4515c..808b5cd 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -1441,6 +1441,7 @@ "quic/core/io/quic_poll_event_loop_test.cc", "quic/core/io/quic_server_io_harness_test.cc", "quic/core/io/socket_test.cc", + "quic/core/quic_packet_reader_test.cc", "quic/core/quic_udp_socket_test.cc", "quic/tools/quic_default_client_test.cc", "quic/tools/quic_server_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index ecbc2d3..d87c967 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -1445,6 +1445,7 @@ "src/quiche/quic/core/io/quic_poll_event_loop_test.cc", "src/quiche/quic/core/io/quic_server_io_harness_test.cc", "src/quiche/quic/core/io/socket_test.cc", + "src/quiche/quic/core/quic_packet_reader_test.cc", "src/quiche/quic/core/quic_udp_socket_test.cc", "src/quiche/quic/tools/quic_default_client_test.cc", "src/quiche/quic/tools/quic_server_test.cc",
diff --git a/build/source_list.json b/build/source_list.json index 1945c17..9be762b 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -1444,6 +1444,7 @@ "quiche/quic/core/io/quic_poll_event_loop_test.cc", "quiche/quic/core/io/quic_server_io_harness_test.cc", "quiche/quic/core/io/socket_test.cc", + "quiche/quic/core/quic_packet_reader_test.cc", "quiche/quic/core/quic_udp_socket_test.cc", "quiche/quic/tools/quic_default_client_test.cc", "quiche/quic/tools/quic_server_test.cc",
diff --git a/quiche/quic/core/quic_packet_reader.cc b/quiche/quic/core/quic_packet_reader.cc index 96b9a66..3dea5d6 100644 --- a/quiche/quic/core/quic_packet_reader.cc +++ b/quiche/quic/core/quic_packet_reader.cc
@@ -4,6 +4,9 @@ #include "quiche/quic/core/quic_packet_reader.h" +#include <cstddef> +#include <vector> + #include "absl/base/macros.h" #include "quiche/quic/core/quic_packets.h" #include "quiche/quic/core/quic_process_packet_interface.h" @@ -20,18 +23,37 @@ namespace quic { -QuicPacketReader::QuicPacketReader() - : read_buffers_(kNumPacketsPerReadMmsgCall), - read_results_(kNumPacketsPerReadMmsgCall) { - QUICHE_DCHECK_EQ(read_buffers_.size(), read_results_.size()); - for (size_t i = 0; i < read_results_.size(); ++i) { - read_results_[i].packet_buffer.buffer = read_buffers_[i].packet_buffer; - read_results_[i].packet_buffer.buffer_len = - sizeof(read_buffers_[i].packet_buffer); +namespace { - read_results_[i].control_buffer.buffer = read_buffers_[i].control_buffer; - read_results_[i].control_buffer.buffer_len = - sizeof(read_buffers_[i].control_buffer); +// Points each of |read_results| at the matching buffer in |read_buffers|. +template <typename ReadBufferType> +void AttachBuffers(std::vector<ReadBufferType>& read_buffers, + QuicUdpSocketApi::ReadPacketResults& read_results) { + QUICHE_DCHECK_EQ(read_buffers.size(), read_results.size()); + for (size_t i = 0; i < read_results.size(); ++i) { + read_results[i].packet_buffer.buffer = read_buffers[i].packet_buffer; + read_results[i].packet_buffer.buffer_len = + sizeof(read_buffers[i].packet_buffer); + + read_results[i].control_buffer.buffer = read_buffers[i].control_buffer; + read_results[i].control_buffer.buffer_len = + sizeof(read_buffers[i].control_buffer); + } +} + +} // namespace + +QuicPacketReader::QuicPacketReader() : QuicPacketReader(Options()) {} + +QuicPacketReader::QuicPacketReader(Options options) + : gro_enabled_(options.enable_gro), + read_results_(kNumPacketsPerReadMmsgCall) { + if (gro_enabled_) { + gro_read_buffers_.resize(kNumPacketsPerReadMmsgCall); + AttachBuffers(gro_read_buffers_, read_results_); + } else { + read_buffers_.resize(kNumPacketsPerReadMmsgCall); + AttachBuffers(read_buffers_, read_results_); } } @@ -41,9 +63,11 @@ int fd, int port, const QuicClock& clock, ProcessPacketInterface* processor, QuicPacketCount* packets_dropped) { // Reset all read_results for reuse. + const size_t packet_buffer_length = gro_enabled_ + ? sizeof(GroReadBuffer::packet_buffer) + : sizeof(ReadBuffer::packet_buffer); for (size_t i = 0; i < read_results_.size(); ++i) { - read_results_[i].Reset( - /*packet_buffer_length=*/sizeof(read_buffers_[i].packet_buffer)); + read_results_[i].Reset(packet_buffer_length); } // Use clock.Now() as the packet receipt time, the time between packet @@ -62,6 +86,9 @@ QUIC_CODE_COUNT(quic_record_tos_byte); // Note ToS bit will also populate ECN codepoint. info_bits.Set(QuicUdpPacketInfoBit::TOS); + if (gro_enabled_) { + info_bits.Set(QuicUdpPacketInfoBit::IS_GRO); + } size_t packets_read = socket_api_.ReadMultiplePackets(fd, info_bits, &read_results_); if (GetQuicReloadableFlag(quic_move_clock_now)) { @@ -80,7 +107,15 @@ continue; } - DispatchPacket(result, now, port, processor); + if (gro_enabled_) { + // The result may be a GRO train of several packets. + socket_api_.ForEachGroSegment( + result, [&](const QuicUdpSocketApi::ReadPacketResult& segment) { + DispatchPacket(segment, now, port, processor); + }); + } else { + DispatchPacket(result, now, port, processor); + } } // We may not have read all of the packets available on the socket.
diff --git a/quiche/quic/core/quic_packet_reader.h b/quiche/quic/core/quic_packet_reader.h index 34c171b..94a5d17 100644 --- a/quiche/quic/core/quic_packet_reader.h +++ b/quiche/quic/core/quic_packet_reader.h
@@ -7,6 +7,9 @@ #ifndef QUICHE_QUIC_CORE_QUIC_PACKET_READER_H_ #define QUICHE_QUIC_CORE_QUIC_PACKET_READER_H_ +#include <cstddef> +#include <vector> + #include "absl/base/optimization.h" #include "quiche/quic/core/quic_clock.h" #include "quiche/quic/core/quic_packets.h" @@ -21,9 +24,22 @@ // Read in larger batches to minimize recvmmsg overhead. inline constexpr int kNumPacketsPerReadMmsgCall = 16; +// The size of each packet buffer when reading UDP GRO trains: room for a whole +// train, which the kernel caps at 64 KB. +inline constexpr size_t kGroPacketBufferSize = 64 * 1024; + class QUICHE_EXPORT QuicPacketReader { public: + struct QUICHE_EXPORT Options { + // If true, each packet buffer has room for a whole UDP GRO train, and each + // train read is dispatched as the individual packets in it. This takes + // about 1 MB of buffers instead of about 32 KB. Only enable UDP GRO on a + // socket (QuicUdpSocketApi::EnableUdpGro) if its reader has this set. + bool enable_gro = false; + }; + QuicPacketReader(); + explicit QuicPacketReader(Options options); QuicPacketReader(const QuicPacketReader&) = delete; QuicPacketReader& operator=(const QuicPacketReader&) = delete; @@ -40,6 +56,9 @@ ProcessPacketInterface* processor, QuicPacketCount* packets_dropped); + // Whether this reader reads UDP GRO trains. See Options::enable_gro. + bool gro_enabled() const { return gro_enabled_; } + private: // Return the self ip from |packet_info|. // For dual stack sockets, |packet_info| may contain both a v4 and a v6 ip, in @@ -60,8 +79,20 @@ ABSL_CACHELINE_ALIGNED char packet_buffer[kMaxIncomingPacketSize]; }; + // A ReadBuffer with room for a whole UDP GRO train. + struct QUICHE_EXPORT GroReadBuffer { + ABSL_CACHELINE_ALIGNED char + control_buffer[kDefaultUdpPacketControlBufferSize]; + ABSL_CACHELINE_ALIGNED char packet_buffer[kGroPacketBufferSize]; + }; + + const bool gro_enabled_; QuicUdpSocketApi socket_api_; + + // Only one of these is allocated and used, depending on |gro_enabled_|. std::vector<ReadBuffer> read_buffers_; + std::vector<GroReadBuffer> gro_read_buffers_; + QuicUdpSocketApi::ReadPacketResults read_results_; };
diff --git a/quiche/quic/core/quic_packet_reader_test.cc b/quiche/quic/core/quic_packet_reader_test.cc new file mode 100644 index 0000000..494dffa --- /dev/null +++ b/quiche/quic/core/quic_packet_reader_test.cc
@@ -0,0 +1,238 @@ +// Copyright 2026 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/quic_packet_reader.h" + +#include <cstddef> +#include <string> +#include <vector> + +#include "absl/base/optimization.h" +#include "absl/strings/string_view.h" +#include "quiche/quic/core/quic_packets.h" +#include "quiche/quic/core/quic_process_packet_interface.h" +#include "quiche/quic/core/quic_time.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/core/quic_udp_socket.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/quic/platform/api/quic_test.h" +#include "quiche/quic/platform/api/quic_test_loopback.h" +#include "quiche/quic/test_tools/mock_clock.h" + +#if defined(__linux__) +#include <netinet/udp.h> +#include <sys/socket.h> + +#ifndef UDP_SEGMENT +#define UDP_SEGMENT 103 +#endif +#endif // defined(__linux__) + +namespace quic { +namespace test { +namespace { + +using ::testing::ElementsAre; + +constexpr int kSocketBufferSize = 1024 * 1024; +// How long to wait for a socket to become readable, and how many times. +constexpr QuicTime::Delta kReadTimeout = QuicTime::Delta::FromMilliseconds(100); +constexpr int kMaxReadAttempts = 50; + +// Records the packets that a QuicPacketReader dispatches. +class RecordingProcessor : public ProcessPacketInterface { + public: + struct Packet { + QuicSocketAddress self_address; + QuicSocketAddress peer_address; + std::string payload; + }; + + void ProcessPacket(const QuicSocketAddress& self_address, + const QuicSocketAddress& peer_address, + const QuicReceivedPacket& packet) override { + packets_.push_back({self_address, peer_address, + std::string(packet.data(), packet.length())}); + } + + const std::vector<Packet>& packets() const { return packets_; } + + // Returns the payloads of the packets, in the order they were dispatched. + std::vector<std::string> payloads() const { + std::vector<std::string> payloads; + for (const Packet& packet : packets_) { + payloads.push_back(packet.payload); + } + return payloads; + } + + private: + std::vector<Packet> packets_; +}; + +// Sends datagrams from a client socket to a server socket over loopback. +class QuicPacketReaderTest : public QuicTest { + protected: + void SetUp() override { + ASSERT_NO_FATAL_FAILURE( + CreateLoopbackSocket(&reader_fd_, &server_address_)); + ASSERT_NO_FATAL_FAILURE( + CreateLoopbackSocket(&writer_fd_, &client_address_)); + } + + void TearDown() override { + socket_api_.Destroy(writer_fd_); + socket_api_.Destroy(reader_fd_); + } + + // Creates a UDP socket bound to an ephemeral loopback port. + void CreateLoopbackSocket(QuicUdpSocketFd* fd, QuicSocketAddress* address) { + const QuicSocketAddress loopback(TestLoopback(), /*port=*/0); + *fd = socket_api_.Create(loopback.host().AddressFamilyToInt(), + kSocketBufferSize, kSocketBufferSize); + ASSERT_NE(*fd, kQuicInvalidSocketFd); + ASSERT_TRUE(socket_api_.Bind(*fd, loopback)); + ASSERT_EQ(address->FromSocket(*fd), 0); + } + + // Sends |data| from the client socket to the server socket in one write. + bool Send(absl::string_view data) { + QuicUdpPacketInfo packet_info; + packet_info.SetPeerAddress(server_address_); + return socket_api_ + .WritePacket(writer_fd_, data.data(), data.length(), packet_info) + .status == WRITE_STATUS_OK; + } + + // Lets |reader| read from the server socket until |processor_| has received + // at least |num_packets| packets, or the reads time out. + void ReadPackets(QuicPacketReader& reader, size_t num_packets) { + for (int attempt = 0; attempt < kMaxReadAttempts && + processor_.packets().size() < num_packets; + ++attempt) { + if (socket_api_.WaitUntilReadable(reader_fd_, kReadTimeout)) { + reader.ReadAndDispatchPackets(reader_fd_, server_address_.port(), + clock_, &processor_, + /*packets_dropped=*/nullptr); + } + } + } + + // Reads one datagram, or one GRO train, from the server socket without a + // QuicPacketReader. The result points into |read_packet_buffer_|. + QuicUdpSocketApi::ReadPacketResult ReadDatagram() { + QuicUdpSocketApi::ReadPacketResult result; + result.packet_buffer = {read_packet_buffer_.data(), + read_packet_buffer_.size()}; + result.control_buffer = {read_control_buffer_, + sizeof(read_control_buffer_)}; + for (int attempt = 0; attempt < kMaxReadAttempts && !result.ok; ++attempt) { + if (socket_api_.WaitUntilReadable(reader_fd_, kReadTimeout)) { + result.Reset(read_packet_buffer_.size()); + socket_api_.ReadPacket( + reader_fd_, + QuicUdpPacketInfoBitMask({QuicUdpPacketInfoBit::IS_GRO}), &result); + } + } + return result; + } + + // Expects every packet to have been sent from the client to the server. + void ExpectPacketAddresses() { + for (const RecordingProcessor::Packet& packet : processor_.packets()) { + EXPECT_EQ(packet.self_address, server_address_); + EXPECT_EQ(packet.peer_address, client_address_); + } + } + + QuicUdpSocketApi socket_api_; + // The socket which the QuicPacketReader reads from. + QuicUdpSocketFd reader_fd_ = kQuicInvalidSocketFd; + // The socket which sends datagrams to the receiver socket. + QuicUdpSocketFd writer_fd_ = kQuicInvalidSocketFd; + QuicSocketAddress server_address_; + QuicSocketAddress client_address_; + MockClock clock_; + RecordingProcessor processor_; + std::string read_packet_buffer_ = std::string(kGroPacketBufferSize, '\0'); + ABSL_CACHELINE_ALIGNED char + read_control_buffer_[kDefaultUdpPacketControlBufferSize]; +}; + +TEST_F(QuicPacketReaderTest, DispatchesEachDatagram) { + QuicPacketReader reader; + EXPECT_FALSE(reader.gro_enabled()); + + ASSERT_TRUE(Send("a")); + ASSERT_TRUE(Send("bb")); + ASSERT_TRUE(Send("ccc")); + ReadPackets(reader, 3); + + EXPECT_THAT(processor_.payloads(), ElementsAre("a", "bb", "ccc")); + ExpectPacketAddresses(); +} + +// A GRO reader dispatches datagrams that are not GRO trains as they are, for +// example on a socket without UDP GRO enabled. +TEST_F(QuicPacketReaderTest, GroReaderDispatchesEachDatagram) { + QuicPacketReader reader({.enable_gro = true}); + EXPECT_TRUE(reader.gro_enabled()); + + ASSERT_TRUE(Send("a")); + ASSERT_TRUE(Send("bb")); + ASSERT_TRUE(Send("ccc")); + ReadPackets(reader, 3); + + EXPECT_THAT(processor_.payloads(), ElementsAre("a", "bb", "ccc")); + ExpectPacketAddresses(); +} + +#if defined(__linux__) +TEST_F(QuicPacketReaderTest, GroReaderDispatchesEachPacketInGroTrain) { + if (!socket_api_.EnableUdpGro(reader_fd_)) { + GTEST_SKIP() << "UDP GRO is not supported."; + } + + // Have the kernel split each write into datagrams of up to |segment_size| + // bytes (UDP GSO). Over loopback, a socket with UDP GRO enabled receives the + // datagrams of a write as one GRO train. + const int segment_size = 1000; + if (setsockopt(writer_fd_, SOL_UDP, UDP_SEGMENT, &segment_size, + sizeof(segment_size)) != 0) { + GTEST_SKIP() << "UDP GSO is not supported."; + } + + // Three full size packets and a smaller trailing packet. + const std::vector<std::string> packets = { + std::string(segment_size, 'a'), std::string(segment_size, 'b'), + std::string(segment_size, 'c'), std::string(segment_size / 2, 'd')}; + std::string train; + for (const std::string& packet : packets) { + train += packet; + } + + // Check that the kernel delivers a write as one GRO train, so that the reader + // gets a train to split below. + ASSERT_TRUE(Send(train)); + QuicUdpSocketApi::ReadPacketResult result = ReadDatagram(); + ASSERT_TRUE(result.ok); + if (!result.packet_info.HasValue(QuicUdpPacketInfoBit::IS_GRO)) { + GTEST_SKIP() << "The kernel did not coalesce the datagrams."; + } + EXPECT_EQ(result.packet_info.gso_size(), static_cast<size_t>(segment_size)); + EXPECT_EQ(result.packet_buffer.buffer_len, train.size()); + + // Now read the train with the GRO-enabled reader. + QuicPacketReader reader({.enable_gro = true}); + ASSERT_TRUE(Send(train)); + ReadPackets(reader, packets.size()); + + EXPECT_EQ(processor_.payloads(), packets); + ExpectPacketAddresses(); +} +#endif // defined(__linux__) + +} // namespace +} // namespace test +} // namespace quic
diff --git a/quiche/quic/core/quic_udp_socket.cc b/quiche/quic/core/quic_udp_socket.cc index 145c24a..280cbf7 100644 --- a/quiche/quic/core/quic_udp_socket.cc +++ b/quiche/quic/core/quic_udp_socket.cc
@@ -208,6 +208,16 @@ #endif } +bool QuicUdpSocketApi::EnableUdpGro(QuicUdpSocketFd fd) { +#if defined(__linux__) && defined(SOL_UDP) + int enable_gro = 1; + return 0 == setsockopt(fd, SOL_UDP, UDP_GRO, &enable_gro, sizeof(enable_gro)); +#else + (void)fd; + return false; +#endif +} + size_t QuicUdpSocketApi::ForEachGroSegment( const ReadPacketResult& result, quiche::UnretainedCallback<void(const ReadPacketResult&)> packet_sink) {
diff --git a/quiche/quic/core/quic_udp_socket.h b/quiche/quic/core/quic_udp_socket.h index 750d9ca..3fa3d8f 100644 --- a/quiche/quic/core/quic_udp_socket.h +++ b/quiche/quic/core/quic_udp_socket.h
@@ -249,6 +249,14 @@ bool EnableReceiveTtlForV4(QuicUdpSocketFd fd); bool EnableReceiveTtlForV6(QuicUdpSocketFd fd); + // Enables UDP generic receive offload (GRO) on |fd|. The kernel may then + // coalesce back-to-back datagrams of the same size from the same peer into a + // GRO train, which a read returns as a single result. Only enable this if + // every reader of |fd| requests QuicUdpPacketInfoBit::IS_GRO and has packet + // buffers large enough for a whole train (see ForEachGroSegment); otherwise + // trains are lost. Returns false if UDP GRO is not supported. + bool EnableUdpGro(QuicUdpSocketFd fd); + // Wait for |fd| to become readable, up to |timeout|. // Return true if |fd| is readable upon return. bool WaitUntilReadable(QuicUdpSocketFd fd, QuicTime::Delta timeout);
diff --git a/quiche/quic/core/quic_udp_socket_posix.inc b/quiche/quic/core/quic_udp_socket_posix.inc index 0635c3f..2dbe5dd 100644 --- a/quiche/quic/core/quic_udp_socket_posix.inc +++ b/quiche/quic/core/quic_udp_socket_posix.inc
@@ -27,6 +27,8 @@ #if defined(__linux__) #include <alloca.h> +// For SOL_UDP, which UDP GRO uses. +#include <netinet/udp.h> // For SO_TIMESTAMPING. #include <linux/net_tstamp.h> #endif