QBONE TUN exchanger async refactor: Move multi-packet read support into exchanger impl Even with our plan to leave reads in-thread for now, I couldn't come up with a not-too-much-work way to do the multiple read calls via an async-supporting API. Easier to just move the logic to do it via a single call. Bonus advantage that this moves some more logic into open-source that might as well be open sourced. While at it, made some fixes to the relevant logic. Now avoids unnecessary logging of blocked errors (after first packet), and can continue reading after errors that shouldn't signal no-more-packets. PiperOrigin-RevId: 962864740
diff --git a/quiche/quic/qbone/bonnet/mock_qbone_client_packet_exchanger.h b/quiche/quic/qbone/bonnet/mock_qbone_client_packet_exchanger.h index d42ab2c..7168d42 100644 --- a/quiche/quic/qbone/bonnet/mock_qbone_client_packet_exchanger.h +++ b/quiche/quic/qbone/bonnet/mock_qbone_client_packet_exchanger.h
@@ -31,7 +31,8 @@ MOCK_METHOD(void, Start, (int read_fd, int write_fd), (override)); MOCK_METHOD(void, Stop, (), (override)); - MOCK_METHOD(bool, ReadAndDeliverPacket, (QboneClientInterface * qbone_client), + MOCK_METHOD(int, OnReadFromNetworkReady, + (int max_packets_to_read, QboneClientInterface* qbone_client), (override)); MOCK_METHOD(void, WritePacketToNetwork, (const char* packet, size_t size), (override));
diff --git a/quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h b/quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h index c88b48d..2f15da9 100644 --- a/quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h +++ b/quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h
@@ -48,10 +48,13 @@ // made after this completes. virtual void Stop() = 0; - // Reads a packet from the local network and delivers the packet to - // qbone_client. Returns true if there may be more packets to read. Must not - // be called before Start() or after Stop(). - virtual bool ReadAndDeliverPacket(QboneClientInterface* qbone_client) = 0; + // Notifies the exchanger that at least one packet is ready to be read from + // the network, and reads up to `max_packets_to_read`. Returns the number of + // packets synchronously read from the socket (not number of valid packets + // processed to client and visitor, and not useful if implementation handles + // reads asynchronously). Must not be called before Start() or after Stop(). + virtual int OnReadFromNetworkReady(int max_packets_to_read, + QboneClientInterface* qbone_client) = 0; // Writes a packet to the local network. If the write would be blocked, the // packet is dropped. Must not be called before Start() or after Stop().
diff --git a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.cc b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.cc index 8489b3e..01a99db 100644 --- a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.cc +++ b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.cc
@@ -34,6 +34,7 @@ #include "quiche/quic/qbone/platform/netlink_interface.h" #include "quiche/quic/qbone/qbone_client_interface.h" #include "quiche/quic/qbone/qbone_constants.h" +#include "quiche/common/platform/api/quiche_logging.h" #include "quiche/common/quiche_endian.h" namespace quic { @@ -78,77 +79,29 @@ write_fd_ = -1; } -bool TunDevicePacketExchanger::ReadAndDeliverPacket( - QboneClientInterface* qbone_client) { +int TunDevicePacketExchanger::OnReadFromNetworkReady( + int max_packets_to_read, QboneClientInterface* qbone_client) { if (read_fd_ < 0) { QUIC_BUG(qbone_tun_device_packet_exchanger_read_with_invalid_fd) << "Invalid file descriptor of the TUN device: " << read_fd_; - return false; + return 0; } - ethhdr eth_header; - struct iovec iov[2]; + int packets_read = 0; + for (int i = 0; i < max_packets_to_read; ++i) { + // Should be at least one packet available to read if this is called, so + // fully process any blocked errors on the first read. After that, blocked + // errors are just the signal that there are no more packets to read. + bool exchange_blocked_error = packets_read == 0; - iov[0].iov_base = is_tap_ ? ð_header : nullptr; - iov[0].iov_len = is_tap_ ? ETH_HLEN : 0; - iov[1].iov_base = read_buffer_.data(); - iov[1].iov_len = read_buffer_.size(); - - absl::Status status = absl::OkStatus(); - absl::Time start = absl::Now(); - int result = kernel_->readv(read_fd_, iov, ABSL_ARRAYSIZE(iov)); - if (result < 0) { - status = absl::ErrnoToStatus(errno, "Read from the TUN device failed."); - } else if (result == 0) { - // Note that 0 means end of file, but we're talking about a TUN device - - // there is no end of file. Therefore 0 also indicates error. - status = absl::InternalError( - "Read from the TUN device returned unexpected 0 (EOF)."); - } - absl::Duration latency = std::max(absl::Now() - start, absl::ZeroDuration()); - - if (!status.ok()) { - QUIC_LOG_EVERY_N_SEC(ERROR, 60) << "Packet read failed: " << status; - visitor_.OnRead(std::move(status)); - return false; - } - - int l3_packet_size = is_tap_ ? result - ETH_HLEN : result; - if (l3_packet_size <= 0 || l3_packet_size > read_buffer_.size()) { - absl::Status error = - absl::InternalError(absl::StrCat("Invalid packet size.")); - QUIC_LOG_EVERY_N_SEC(ERROR, 60) << "Packet read failed: " << error; - visitor_.OnRead(std::move(error)); - return false; - } - absl::Span<const std::byte> l3_packet = - absl::MakeSpan(read_buffer_.data(), l3_packet_size); - - if (is_tap_) { - switch (ValidateL2Headers(eth_header, l3_packet)) { - case L2ValidationResult::kInvalid: { - absl::Status error = absl::InvalidArgumentError("Invalid L2 headers."); - visitor_.OnRead(std::move(error)); - return false; - } - case L2ValidationResult::kValidLinkLocal: - // TODO(b/535980431): This returns false to match the behavior of a - // previous implementation because no packet is forwarded to the tunnel, - // but consider changing this to true. A link-local packet does not mean - // there are no more packets to read from the TUN device. - return false; - case L2ValidationResult::kValidNormal: - // Packet is valid and should be forwarded to the tunnel. Fall through - // to normal processing. - break; + if (ReadAndExchangeSinglePacket(qbone_client, exchange_blocked_error)) { + packets_read++; + } else { + break; } } - visitor_.OnRead(std::vector<ReadResult>{ - ReadResult{.packet = l3_packet, .latency = latency}}); - qbone_client->ProcessPacketFromNetwork(absl::string_view( - reinterpret_cast<const char*>(l3_packet.data()), l3_packet.size())); - return true; + return packets_read; } void TunDevicePacketExchanger::WritePacketToNetwork(const char* packet, @@ -188,6 +141,82 @@ .latency = latency}}); } +bool TunDevicePacketExchanger::ReadAndExchangeSinglePacket( + QboneClientInterface* qbone_client, bool exchange_blocked_error) { + QUICHE_DCHECK_GE(read_fd_, 0); + + // TODO(ericorth): Consider allocating these buffers once and reusing rather + // than repeating for each packet. + ethhdr eth_header; + struct iovec iov[2]; + + iov[0].iov_base = is_tap_ ? ð_header : nullptr; + iov[0].iov_len = is_tap_ ? ETH_HLEN : 0; + iov[1].iov_base = read_buffer_.data(); + iov[1].iov_len = read_buffer_.size(); + + absl::Status status = absl::OkStatus(); + absl::Time start = absl::Now(); + int result = kernel_->readv(read_fd_, iov, ABSL_ARRAYSIZE(iov)); + int saved_errno = errno; + absl::Duration latency = std::max(absl::Now() - start, absl::ZeroDuration()); + + if (result < 0) { + if ((saved_errno == EAGAIN || saved_errno == EWOULDBLOCK) && + !exchange_blocked_error) { + // No more packets available to read. + return false; + } else { + status = + absl::ErrnoToStatus(saved_errno, "Read from the TUN device failed."); + } + } else if (result == 0) { + // Note that 0 means end of file, but we're talking about a TUN device - + // there is no end of file. Therefore 0 also indicates error. + status = absl::InternalError( + "Read from the TUN device returned unexpected 0 (EOF)."); + } + + if (!status.ok()) { + QUIC_LOG_EVERY_N_SEC(ERROR, 60) << "Packet read failed: " << status; + visitor_.OnRead(std::move(status)); + return false; // Assume no more packets after any socket read error. + } + + int l3_packet_size = is_tap_ ? result - ETH_HLEN : result; + if (l3_packet_size <= 0 || l3_packet_size > read_buffer_.size()) { + absl::Status error = + absl::InternalError(absl::StrCat("Invalid packet size.")); + QUIC_LOG_EVERY_N_SEC(ERROR, 60) << "Packet read failed: " << error; + visitor_.OnRead(std::move(error)); + return true; // Invalid packet does not mean there are no more packets. + } + absl::Span<const std::byte> l3_packet = + absl::MakeSpan(read_buffer_.data(), l3_packet_size); + + if (is_tap_) { + switch (ValidateL2Headers(eth_header, l3_packet)) { + case L2ValidationResult::kInvalid: { + absl::Status error = absl::InvalidArgumentError("Invalid L2 headers."); + visitor_.OnRead(std::move(error)); + return true; // Invalid packet does not mean there are no more packets. + } + case L2ValidationResult::kValidLinkLocal: + return true; + case L2ValidationResult::kValidNormal: + // Packet is valid and should be forwarded to the tunnel. Fall through + // to normal processing. + break; + } + } + + visitor_.OnRead(std::vector<ReadResult>{ + ReadResult{.packet = l3_packet, .latency = latency}}); + qbone_client->ProcessPacketFromNetwork(absl::string_view( + reinterpret_cast<const char*>(l3_packet.data()), l3_packet.size())); + return true; +} + void TunDevicePacketExchanger::InitializeEthHdr() { if (!eth_hdr_initialized_) { NetlinkInterface::LinkInfo link_info{};
diff --git a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.h b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.h index 19b03d5..7ba8748 100644 --- a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.h +++ b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger.h
@@ -38,7 +38,8 @@ // QboneClientPacketExchanger: void Start(int read_fd, int write_fd) override; void Stop() override; - bool ReadAndDeliverPacket(QboneClientInterface* qbone_client) override; + int OnReadFromNetworkReady(int max_packets_to_read, + QboneClientInterface* qbone_client) override; void WritePacketToNetwork(const char* packet, size_t size) override; private: @@ -55,8 +56,11 @@ kValidLinkLocal }; - void InitializeEthHdr(); + // Returns true if more packets may be available to read. + bool ReadAndExchangeSinglePacket(QboneClientInterface* qbone_client, + bool exchange_blocked_error); + void InitializeEthHdr(); L2ValidationResult ValidateL2Headers(const ethhdr& eth_header, absl::Span<const std::byte> packet);
diff --git a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger_test.cc b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger_test.cc index cecca0a..442af30 100644 --- a/quiche/quic/qbone/bonnet/tun_device_packet_exchanger_test.cc +++ b/quiche/quic/qbone/bonnet/tun_device_packet_exchanger_test.cc
@@ -224,7 +224,9 @@ return -1; }); EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); - EXPECT_FALSE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 0); exchanger_.Stop(); } @@ -238,7 +240,9 @@ return -1; }); EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); - EXPECT_FALSE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 0); exchanger_.Stop(); } @@ -261,7 +265,133 @@ &QboneClientPacketExchanger::ReadResult::packet, ElementsAreArray(reinterpret_cast<const std::byte*>(packet.data()), packet.size())))))); - EXPECT_TRUE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 1); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTest, MultipleReadsMoreAvailableThanMax) { + exchanger_.Start(kReadFd, kWriteFd); + + std::string packet1 = "fake_packet_1"; + std::string packet2 = "fake_packet_2"; + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([packet1](int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[1].iov_base, packet1.data(), packet1.size()); + return packet1.size(); + }) + .WillOnce([packet2](int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[1].iov_base, packet2.data(), packet2.size()); + return packet2.size(); + }); + + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(packet1))); + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(packet2))); + + EXPECT_CALL( + mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet1.data()), + packet1.size())))))); + EXPECT_CALL( + mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet2.data()), + packet2.size())))))); + + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/2, + &mock_client_), + 2); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTest, MultipleReadsBlockedBeforeMax) { + exchanger_.Start(kReadFd, kWriteFd); + + std::string packet1 = "fake_packet_1"; + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([packet1](int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[1].iov_base, packet1.data(), packet1.size()); + return packet1.size(); + }) + .WillOnce([](int fd, const struct iovec* iov, int iovcnt) { + errno = EAGAIN; + return -1; + }); + + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(packet1))); + + // Expect no error callbacks from the blocked read. In this scenario, the + // blocked socket is just a signal that there are no more packets to be read, + // rather than an actual error. + + EXPECT_CALL( + mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet1.data()), + packet1.size())))))); + + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/5, + &mock_client_), + 1); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTest, MultiReadInvalidSizeHuge) { + exchanger_.Start(kReadFd, kWriteFd); + + std::string valid_packet = "valid_packet"; + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([](int fd, const struct iovec* iov, int iovcnt) { + return kMtu + 1; // Invalid size + }) + .WillOnce([valid_packet](int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[1].iov_base, valid_packet.data(), valid_packet.size()); + return valid_packet.size(); + }); + + EXPECT_CALL(mock_visitor_, OnRead(StatusIs(absl::StatusCode::kInternal))); + + // Expect subsequent packet to still be read and processed after the invalid + // packet. + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(valid_packet))); + EXPECT_CALL(mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray( + reinterpret_cast<const std::byte*>(valid_packet.data()), + valid_packet.size())))))); + + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/2, + &mock_client_), + 2); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTest, + ReadPacketBlockedOnFirstWithMaxMoreThanOne) { + exchanger_.Start(kReadFd, kWriteFd); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([](int fd, const struct iovec* iov, int iovcnt) { + errno = EAGAIN; + return -1; + }); + EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/2, + &mock_client_), + 0); exchanger_.Stop(); } @@ -313,7 +443,9 @@ &QboneClientPacketExchanger::ReadResult::packet, ElementsAreArray(reinterpret_cast<const std::byte*>(l3_packet.data()), l3_packet.size())))))); - EXPECT_TRUE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 1); exchanger_.Stop(); } @@ -331,7 +463,9 @@ }); EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); - EXPECT_FALSE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 1); exchanger_.Stop(); } @@ -382,9 +516,106 @@ }); EXPECT_CALL(mock_visitor_, OnWrite(IsOkAndHolds(SizeIs(1)))); - // ReadAndDeliverPacket should return false because packet was handled - // internally (Neighbor Discovery). - EXPECT_FALSE(exchanger_.ReadAndDeliverPacket(&mock_client_)); + // OnReadFromNetworkReady should return 1 because packet was handled + // internally (Neighbor Discovery) but still read from network. + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/1, + &mock_client_), + 1); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTapTest, MultiReadInvalidSizeShort) { + exchanger_.Start(kReadFd, kWriteFd); + + ip6_hdr ip_hdr{}; + ip_hdr.ip6_vfc = 0x60; // Version 6 + ip_hdr.ip6_nxt = 59; // No next header + + std::string l3_payload = "hello"; + std::string valid_l3_packet = + std::string(reinterpret_cast<char*>(&ip_hdr), sizeof(ip_hdr)) + + l3_payload; + + ethhdr valid_eth_hdr{}; + valid_eth_hdr.h_proto = QuicheEndian::HostToNet16(ETH_P_IPV6); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([](int fd, const struct iovec* iov, int iovcnt) { + return ETH_HLEN - 1; // Invalid size (too short to contain L3 packet) + }) + .WillOnce([valid_eth_hdr, valid_l3_packet]( + int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[0].iov_base, &valid_eth_hdr, ETH_HLEN); + memcpy(iov[1].iov_base, valid_l3_packet.data(), valid_l3_packet.size()); + return ETH_HLEN + valid_l3_packet.size(); + }); + + EXPECT_CALL(mock_visitor_, OnRead(StatusIs(absl::StatusCode::kInternal))); + + // Expect subsequent packet to still be read and processed after the invalid + // packet. + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(valid_l3_packet))); + EXPECT_CALL(mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre( + Field(&QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>( + valid_l3_packet.data()), + valid_l3_packet.size())))))); + + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/2, + &mock_client_), + 2); + + exchanger_.Stop(); +} + +TEST_F(TunDevicePacketExchangerTapTest, MultiReadInvalidL2) { + exchanger_.Start(kReadFd, kWriteFd); + + ethhdr invalid_eth_hdr{}; + invalid_eth_hdr.h_proto = QuicheEndian::HostToNet16(ETH_P_ARP); // Non-IPv6 + + ip6_hdr ip_hdr{}; + ip_hdr.ip6_vfc = 0x60; // Version 6 + ip_hdr.ip6_nxt = 59; // No next header + + std::string l3_payload = "hello"; + std::string valid_l3_packet = + std::string(reinterpret_cast<char*>(&ip_hdr), sizeof(ip_hdr)) + + l3_payload; + + ethhdr valid_eth_hdr{}; + valid_eth_hdr.h_proto = QuicheEndian::HostToNet16(ETH_P_IPV6); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([invalid_eth_hdr](int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[0].iov_base, &invalid_eth_hdr, ETH_HLEN); + return ETH_HLEN + 10; + }) + .WillOnce([valid_eth_hdr, valid_l3_packet]( + int fd, const struct iovec* iov, int iovcnt) { + memcpy(iov[0].iov_base, &valid_eth_hdr, ETH_HLEN); + memcpy(iov[1].iov_base, valid_l3_packet.data(), valid_l3_packet.size()); + return ETH_HLEN + valid_l3_packet.size(); + }); + + EXPECT_CALL(mock_visitor_, + OnRead(StatusIs(absl::StatusCode::kInvalidArgument))); + + // Expect subsequent packet to still be read and processed after the invalid + // packet. + EXPECT_CALL(mock_client_, ProcessPacketFromNetwork(StrEq(valid_l3_packet))); + EXPECT_CALL(mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre( + Field(&QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>( + valid_l3_packet.data()), + valid_l3_packet.size())))))); + + EXPECT_EQ(exchanger_.OnReadFromNetworkReady(/*max_packets_to_read=*/2, + &mock_client_), + 2); exchanger_.Stop(); }