Create AsyncWritePacketExchanger Offloads TUN writes to a worker thread. Doing all the thread marshaling and message queuing manually to keep the logic open-source and to maximize environment genericness. PiperOrigin-RevId: 972633784
diff --git a/build/source_list.bzl b/build/source_list.bzl index f85da00..da09290 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -1752,6 +1752,7 @@ "oblivious_http/oblivious_http_gateway.cc", ] qbone_hdrs = [ + "quic/qbone/bonnet/async_write_packet_exchanger.h", "quic/qbone/bonnet/icmp_reachable.h", "quic/qbone/bonnet/icmp_reachable_interface.h", "quic/qbone/bonnet/mock_icmp_reachable.h", @@ -1792,6 +1793,8 @@ "quic/qbone/test_tools/qbone_basic_quic_server_handler.h", ] qbone_srcs = [ + "quic/qbone/bonnet/async_write_packet_exchanger.cc", + "quic/qbone/bonnet/async_write_packet_exchanger_test.cc", "quic/qbone/bonnet/icmp_reachable.cc", "quic/qbone/bonnet/icmp_reachable_test.cc", "quic/qbone/bonnet/qbone_tunnel_info.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index 84571da..fd66f95 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -1758,6 +1758,7 @@ "src/quiche/oblivious_http/oblivious_http_gateway.cc", ] qbone_hdrs = [ + "src/quiche/quic/qbone/bonnet/async_write_packet_exchanger.h", "src/quiche/quic/qbone/bonnet/icmp_reachable.h", "src/quiche/quic/qbone/bonnet/icmp_reachable_interface.h", "src/quiche/quic/qbone/bonnet/mock_icmp_reachable.h", @@ -1798,6 +1799,8 @@ "src/quiche/quic/qbone/test_tools/qbone_basic_quic_server_handler.h", ] qbone_srcs = [ + "src/quiche/quic/qbone/bonnet/async_write_packet_exchanger.cc", + "src/quiche/quic/qbone/bonnet/async_write_packet_exchanger_test.cc", "src/quiche/quic/qbone/bonnet/icmp_reachable.cc", "src/quiche/quic/qbone/bonnet/icmp_reachable_test.cc", "src/quiche/quic/qbone/bonnet/qbone_tunnel_info.cc",
diff --git a/build/source_list.json b/build/source_list.json index f504793..62889c0 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -1757,6 +1757,7 @@ "quiche/oblivious_http/oblivious_http_gateway.cc" ], "qbone_hdrs": [ + "quiche/quic/qbone/bonnet/async_write_packet_exchanger.h", "quiche/quic/qbone/bonnet/icmp_reachable.h", "quiche/quic/qbone/bonnet/icmp_reachable_interface.h", "quiche/quic/qbone/bonnet/mock_icmp_reachable.h", @@ -1797,6 +1798,8 @@ "quiche/quic/qbone/test_tools/qbone_basic_quic_server_handler.h" ], "qbone_srcs": [ + "quiche/quic/qbone/bonnet/async_write_packet_exchanger.cc", + "quiche/quic/qbone/bonnet/async_write_packet_exchanger_test.cc", "quiche/quic/qbone/bonnet/icmp_reachable.cc", "quiche/quic/qbone/bonnet/icmp_reachable_test.cc", "quiche/quic/qbone/bonnet/qbone_tunnel_info.cc",
diff --git a/quiche/quic/qbone/bonnet/async_write_packet_exchanger.cc b/quiche/quic/qbone/bonnet/async_write_packet_exchanger.cc new file mode 100644 index 0000000..5fdabd0 --- /dev/null +++ b/quiche/quic/qbone/bonnet/async_write_packet_exchanger.cc
@@ -0,0 +1,373 @@ +// Copyright 2026 The Chromium Authors +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/qbone/bonnet/async_write_packet_exchanger.h" + +#include <cstddef> +#include <cstdint> +#include <memory> +#include <queue> +#include <thread> // NOLINT (for open-sourceable thread ID) +#include <utility> +#include <vector> + +#include "absl/base/nullability.h" +#include "absl/base/thread_annotations.h" +#include "absl/container/flat_hash_map.h" +#include "absl/status/status.h" +#include "absl/strings/string_view.h" +#include "absl/synchronization/mutex.h" +#include "absl/synchronization/notification.h" +#include "absl/types/span.h" +#include "quiche/quic/platform/api/quic_thread.h" +#include "quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h" +#include "quiche/quic/qbone/platform/kernel_interface.h" +#include "quiche/quic/qbone/platform/netlink_interface.h" +#include "quiche/common/platform/api/quiche_bug_tracker.h" +#include "quiche/common/platform/api/quiche_logging.h" + +namespace quic { + +class AsyncWritePacketExchanger::WriteThread : public QuicThread { + public: + WriteThread(QboneClientPacketExchanger* absl_nonnull thread_local_exchanger, + AsyncWritePacketExchanger* absl_nonnull async_exchanger) + : QuicThread("AsyncWritePacketExchanger::WriteThread"), + thread_local_exchanger_(*thread_local_exchanger), + async_exchanger_(*async_exchanger) {} + + void CompleteWritesAndStop() { + QUICHE_DCHECK(async_exchanger_.silo_executor_->IsOnSiloThread()); + { + absl::MutexLock lock(async_exchanger_.write_queue_mutex_); + stop_requested_ = true; + } + Join(); + } + + bool IsRunningOnThread() const { + return thread_started_.HasBeenNotified() && + std::this_thread::get_id() == thread_id_; + } + + void ProcessWriteResults(absl::StatusOr<std::vector<WriteResult>> results) { + QUICHE_DCHECK(IsRunningOnThread()); + + if (results.ok()) { + for (const WriteResult& result : *results) { + auto packet_node = in_flight_writes_.extract(result.packet.data()); + if (packet_node.empty()) { + // Unexpected. Assuming `thread_local_exchanger_` should always pass + // results to the visitor during the WritePacketToNetwork() call and + // that the result packet should point to the original packet buffer + // stored in `in_flight_writes_`. + QUICHE_BUG(qbone_async_write_packet_exchanger_write_result_not_found); + continue; + } + + ready_results_.push( + InternalWriteResult{.packet = std::move(packet_node.mapped()), + .latency = result.latency}); + } + } else { + in_flight_error_count_++; + ready_results_.push(results.status()); + } + } + + protected: + void Run() override { + QUICHE_DCHECK_EQ(thread_id_, std::thread::id()); + + thread_id_ = std::this_thread::get_id(); + thread_started_.Notify(); + + while (true) { + std::queue<std::vector<std::byte>> packets_to_write = + GetPacketsToWriteCriticalSection(); + + if (packets_to_write.empty()) { + break; + } + + while (!packets_to_write.empty()) { + // Store packet buffers in `in_flight_writes_` during the write. Should + // be unnecessary, assuming `thread_local_exchanger_` always completes + // the write and calls the visitor synchronously, but this allows for + // validating that the received write results are for the expected + // packets. + absl::Span<const std::byte> packet_span = packets_to_write.front(); + QUICHE_DCHECK(packet_span.data() != nullptr); + in_flight_writes_[packet_span.data()] = + std::move(packets_to_write.front()); + QUICHE_DCHECK_EQ(in_flight_writes_[packet_span.data()].data(), + packet_span.data()); + thread_local_exchanger_.WritePacketToNetwork(packet_span); + packets_to_write.pop(); + } + + // Abandon any packets that failed to write. + QUICHE_BUG_IF(qbone_async_write_packet_exchanger_missing_write_results, + in_flight_writes_.size() != in_flight_error_count_); + in_flight_writes_.clear(); + in_flight_error_count_ = 0; + + if (!ready_results_.empty()) { + PassResultsCriticalSection(); + + // Trigger the silo thread to process the results and pass out to the + // visitor. This may trigger unnecessary extra calls if this happens + // multiple times before the silo thread executes the task, but it + // should be safe and have negligible impact on performance, so we won't + // worry about tracking when the call is or isn't necessary. + // + // Use a direct reference to the exchanger because `this` may be stopped + // and destroyed before the lambda is executed. But we assume the + // exchanger itself is safe because it owns the executor and stops its + // execution on shutdown. + AsyncWritePacketExchanger* exchanger = &async_exchanger_; + async_exchanger_.silo_executor_->ExecuteInSilo( + [exchanger]() { exchanger->OnWriteResultsReadyForSiloThread(); }); + } + } + + { + absl::MutexLock lock(async_exchanger_.write_queue_mutex_); + QUICHE_DCHECK(stop_requested_); + } + } + + private: + bool IsQueueNonEmptyOrStopRequested() const + ABSL_SHARED_LOCKS_REQUIRED(async_exchanger_.write_queue_mutex_) { + return !async_exchanger_.write_queue_.empty() || stop_requested_; + } + + // May return an empty queue if stop is requested. Otherwise, blocks until the + // queue is non-empty. + std::queue<std::vector<std::byte>> GetPacketsToWriteCriticalSection() { + std::queue<std::vector<std::byte>> packets_to_write; + + absl::MutexLock lock( + async_exchanger_.write_queue_mutex_, + absl::Condition(this, &WriteThread::IsQueueNonEmptyOrStopRequested)); + + if (async_exchanger_.write_queue_.empty()) { + return {}; + } else { + packets_to_write.swap(async_exchanger_.write_queue_); + async_exchanger_.write_queue_size_ = 0; + return packets_to_write; + } + } + + void PassResultsCriticalSection() { + absl::MutexLock lock(async_exchanger_.result_queue_mutex_); + + while (!ready_results_.empty()) { + // For simplicity, only track packet size in the result queue, rather than + // complete size of the result struct. Expect that to be the only part + // with significant memory usage. + int64_t packet_size = ready_results_.front().ok() + ? ready_results_.front()->packet.size() + : 0; + int64_t new_queue_size = + async_exchanger_.result_queue_size_ + packet_size; + if (new_queue_size <= async_exchanger_.max_result_buffer_size_bytes_) { + async_exchanger_.result_queue_size_ = new_queue_size; + async_exchanger_.result_queue_.push(std::move(ready_results_.front())); + } else { + // Result passing is even more best-effort than packet writing. If the + // results queue is full, drop the result. + QUICHE_LOG_EVERY_N_SEC(WARNING, 5) + << "Result queue is full, dropping result."; + } + + ready_results_.pop(); + } + } + + QboneClientPacketExchanger& thread_local_exchanger_; + AsyncWritePacketExchanger& async_exchanger_; + + absl::flat_hash_map<const std::byte*, std::vector<std::byte>> + in_flight_writes_; + int in_flight_error_count_ = 0; + std::queue<absl::StatusOr<AsyncWritePacketExchanger::InternalWriteResult>> + ready_results_; + + absl::Notification thread_started_; + bool stop_requested_ ABSL_GUARDED_BY(async_exchanger_.write_queue_mutex_) = + false; + + // Thread-safe to read after notification of `thread_started_`. + std::thread::id thread_id_; +}; + +class AsyncWritePacketExchanger::IntermediateVisitor : public Visitor { + public: + IntermediateVisitor(AsyncWritePacketExchanger* absl_nonnull async_exchanger) + : async_exchanger_(*async_exchanger) {} + + void OnRead(absl::StatusOr<std::vector<ReadResult>> results) override { + // Expect read to occur on the silo thread, so results can be passed through + // directly. + QUICHE_BUG_IF(qbone_async_write_packet_exchanger_read_result_wrong_thread, + !async_exchanger_.silo_executor_->IsOnSiloThread()); + async_exchanger_.visitor_.OnRead(std::move(results)); + } + + void OnWrite(absl::StatusOr<std::vector<WriteResult>> results) override { + // Expect write to occur on the separate write thread. + QUICHE_BUG_IF(qbone_async_write_packet_exchanger_write_result_wrong_thread, + !async_exchanger_.write_thread_ || + !async_exchanger_.write_thread_->IsRunningOnThread()); + async_exchanger_.write_thread_->ProcessWriteResults(std::move(results)); + } + + private: + AsyncWritePacketExchanger& async_exchanger_; +}; + +AsyncWritePacketExchanger::AsyncWritePacketExchanger( + size_t mtu, KernelInterface* absl_nonnull kernel, + NetlinkInterface* absl_nonnull netlink, Visitor* absl_nonnull visitor, + bool is_tap, absl::string_view ifname, int64_t max_buffer_size_bytes, + int64_t max_result_buffer_size_bytes, + absl_nonnull std::unique_ptr<SiloExecutor> silo_executor) + : max_buffer_size_bytes_(max_buffer_size_bytes), + max_result_buffer_size_bytes_(max_result_buffer_size_bytes), + silo_executor_(std::move(silo_executor)), + visitor_(*visitor), + intermediate_visitor_(std::make_unique<IntermediateVisitor>(this)), + thread_local_exchanger_(mtu, kernel, netlink, intermediate_visitor_.get(), + is_tap, ifname) {} + +AsyncWritePacketExchanger::~AsyncWritePacketExchanger() { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + QUICHE_BUG_IF(qbone_async_write_packet_exchanger_not_stopped, + write_thread_ != nullptr); + + Stop(); + silo_executor_->Stop(); +} + +void AsyncWritePacketExchanger::Start( + int read_fd, int write_fd, + QboneClientPacketExchanger* absl_nullable exchanger) { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + + if (exchanger == nullptr) { + exchanger = this; + } + + thread_local_exchanger_.Start(read_fd, write_fd, exchanger); + + // Allow idempotent Start() calls (assuming `thread_local_exchanger_` + // validates same inputs). + if (!write_thread_) { + write_thread_ = + std::make_unique<WriteThread>(&thread_local_exchanger_, this); + write_thread_->Start(); + } +} + +void AsyncWritePacketExchanger::Stop() { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + + if (write_thread_) { + write_thread_->CompleteWritesAndStop(); + } + thread_local_exchanger_.Stop(); + + // Flush any pending write results to visitor. Along with stopping the write + // thread to ensure no more results are added, this will ensure that if + // `silo_executor_` has any pending calls to run + // OnWriteResultsReadyForSiloThread(), it will safely find an empty results + // queue and not call `visitor_` callbacks. + OnWriteResultsReadyForSiloThread(); + + write_thread_.reset(); +} + +int AsyncWritePacketExchanger::OnReadFromNetworkReady(int max_packets_to_read) { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + return thread_local_exchanger_.OnReadFromNetworkReady(max_packets_to_read); +} + +void AsyncWritePacketExchanger::WritePacketToNetwork( + absl::Span<const std::byte> packet) { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + + if (packet.empty()) { + QUICHE_BUG(qbone_async_write_packet_exchanger_write_empty_packet); + return; + } + + if (!write_thread_) { + QUICHE_BUG(qbone_async_write_packet_exchanger_write_not_started); + return; + } + + // This will waste time copying if the write queue is full, but that should be + // an exceptional case. Better to not hold the lock while copying. + std::vector<std::byte> packet_copy(packet.begin(), packet.end()); + + bool buffer_full = false; + { + absl::MutexLock lock(write_queue_mutex_); + if (write_queue_size_ + packet_copy.size() <= max_buffer_size_bytes_) { + write_queue_size_ += packet_copy.size(); + write_queue_.push(std::move(packet_copy)); + } else { + buffer_full = true; + } + } + + // QBONE packet delivery is best-effort, so if the buffer is full, drop the + // packet. No attempt to wait/retry. + if (buffer_full) { + visitor_.OnWrite(absl::ResourceExhaustedError( + "AsyncWritePacketExchanger::WritePacketToNetwork: buffer full")); + } +} + +void AsyncWritePacketExchanger::OnWriteResultsReadyForSiloThread() { + QUICHE_DCHECK(silo_executor_->IsOnSiloThread()); + + std::vector<absl::StatusOr<InternalWriteResult>> results; + { + absl::MutexLock lock(result_queue_mutex_); + results.reserve(result_queue_.size()); + while (!result_queue_.empty()) { + results.push_back(std::move(result_queue_.front())); + result_queue_.pop(); + } + result_queue_size_ = 0; + } + + std::vector<WriteResult> write_results; + std::vector<absl::Status> errors; + for (const auto& result : results) { + if (result.ok()) { + write_results.push_back( + WriteResult{.packet = result->packet, .latency = result->latency}); + } else { + errors.push_back(result.status()); + } + } + + // Report all successful writes in one batch for efficiency. Errors are less + // common, and we don't optimize for them. + if (!write_results.empty()) { + QUICHE_DCHECK(write_thread_); // Expect no results after stop. + visitor_.OnWrite(std::move(write_results)); + } + for (const auto& error : errors) { + QUICHE_DCHECK(write_thread_); // Expect no errors after stop. + visitor_.OnWrite(error); + } +} + +} // namespace quic
diff --git a/quiche/quic/qbone/bonnet/async_write_packet_exchanger.h b/quiche/quic/qbone/bonnet/async_write_packet_exchanger.h new file mode 100644 index 0000000..bb69645 --- /dev/null +++ b/quiche/quic/qbone/bonnet/async_write_packet_exchanger.h
@@ -0,0 +1,106 @@ +// Copyright 2026 The Chromium Authors +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_QBONE_BONNET_ASYNC_WRITE_PACKET_EXCHANGER_H_ +#define QUICHE_QUIC_QBONE_BONNET_ASYNC_WRITE_PACKET_EXCHANGER_H_ + +#include <cstddef> +#include <cstdint> +#include <memory> +#include <queue> +#include <vector> + +#include "absl/base/attributes.h" +#include "absl/base/nullability.h" +#include "absl/base/thread_annotations.h" +#include "absl/status/statusor.h" +#include "absl/strings/string_view.h" +#include "absl/synchronization/mutex.h" +#include "absl/time/time.h" +#include "absl/types/span.h" +#include "quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h" +#include "quiche/quic/qbone/bonnet/tun_device_packet_exchanger.h" +#include "quiche/quic/qbone/platform/kernel_interface.h" +#include "quiche/quic/qbone/platform/netlink_interface.h" +#include "quiche/common/quiche_callbacks.h" + +namespace quic { + +// Packet exchanger that handles write operations asynchronously on a separate +// worker thread. +class AsyncWritePacketExchanger : public QboneClientPacketExchanger { + public: + // Executor that runs given callbacks on the main silo thread (the thread that + // owns and interacts with the exchanger). + class SiloExecutor { + public: + virtual ~SiloExecutor() = default; + + // Stops the executor and either blocks until all pending callbacks are + // completed or cancels them. No callbacks will be executed after this + // returns. Must be called from the silo thread. + virtual void Stop() = 0; + + // Return true iff called from the silo thread. Thread-safe. + virtual bool IsOnSiloThread() const = 0; + + // Schedules the given callback to be executed on the silo thread. No + // guarantees about when the callback will be executed or the order in + // which it will be executed relative to other callbacks, but all callbacks + // will eventually be executed if Stop() is not called. Thread-safe. + virtual void ExecuteInSilo(quiche::SingleUseCallback<void()> callback) = 0; + }; + + AsyncWritePacketExchanger( + size_t mtu, + KernelInterface* absl_nonnull kernel ABSL_ATTRIBUTE_LIFETIME_BOUND, + NetlinkInterface* absl_nonnull netlink ABSL_ATTRIBUTE_LIFETIME_BOUND, + Visitor* absl_nonnull visitor ABSL_ATTRIBUTE_LIFETIME_BOUND, bool is_tap, + absl::string_view ifname, int64_t max_buffer_size_bytes, + int64_t max_result_buffer_size_bytes, + absl_nonnull std::unique_ptr<SiloExecutor> silo_executor); + ~AsyncWritePacketExchanger() override; + + // QboneClientPacketExchanger: + void Start(int read_fd, int write_fd, + QboneClientPacketExchanger* absl_nullable exchanger) override; + void Stop() override; + int OnReadFromNetworkReady(int max_packets_to_read) override; + void WritePacketToNetwork(absl::Span<const std::byte> packet) override; + + private: + class WriteThread; + class IntermediateVisitor; + + struct InternalWriteResult { + std::vector<std::byte> packet; + absl::Duration latency; + }; + + void OnWriteResultsReadyForSiloThread(); + + const int64_t max_buffer_size_bytes_; + const int64_t max_result_buffer_size_bytes_; + absl_nonnull std::unique_ptr<SiloExecutor> silo_executor_; + Visitor& visitor_; + + std::unique_ptr<IntermediateVisitor> intermediate_visitor_; + TunDevicePacketExchanger thread_local_exchanger_; + + absl::Mutex write_queue_mutex_; + int64_t write_queue_size_ ABSL_GUARDED_BY(write_queue_mutex_) = 0; + std::queue<std::vector<std::byte>> write_queue_ + ABSL_GUARDED_BY(write_queue_mutex_); + + absl::Mutex result_queue_mutex_; + int64_t result_queue_size_ ABSL_GUARDED_BY(result_queue_mutex_) = 0; + std::queue<absl::StatusOr<InternalWriteResult>> result_queue_ + ABSL_GUARDED_BY(result_queue_mutex_); + + std::unique_ptr<WriteThread> write_thread_; +}; + +} // namespace quic + +#endif // QUICHE_QUIC_QBONE_BONNET_ASYNC_WRITE_PACKET_EXCHANGER_H_
diff --git a/quiche/quic/qbone/bonnet/async_write_packet_exchanger_test.cc b/quiche/quic/qbone/bonnet/async_write_packet_exchanger_test.cc new file mode 100644 index 0000000..d1cb113 --- /dev/null +++ b/quiche/quic/qbone/bonnet/async_write_packet_exchanger_test.cc
@@ -0,0 +1,549 @@ +// Copyright 2026 The Chromium Authors +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/qbone/bonnet/async_write_packet_exchanger.h" + +#include <netinet/icmp6.h> +#include <netinet/ip6.h> + +#include <cerrno> +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <memory> +#include <string> +#include <thread> // NOLINT (for open-sourceable thread ID) +#include <utility> +#include <vector> + +#include "absl/base/thread_annotations.h" +#include "absl/status/status.h" +#include "absl/strings/string_view.h" +#include "absl/synchronization/mutex.h" +#include "absl/time/time.h" +#include "absl/types/span.h" +#include "quiche/quic/platform/api/quic_test.h" +#include "quiche/quic/qbone/bonnet/mock_qbone_client_packet_exchanger.h" +#include "quiche/quic/qbone/bonnet/qbone_client_packet_exchanger.h" +#include "quiche/quic/qbone/platform/mock_kernel.h" +#include "quiche/quic/qbone/platform/mock_netlink.h" +#include "quiche/quic/qbone/qbone_constants.h" +#include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/quiche_callbacks.h" +#include "quiche/common/quiche_endian.h" + +namespace quic::test { +namespace { + +constexpr size_t kMtu = 1000; +constexpr int kReadFd = 15; +constexpr int kWriteFd = 16; +constexpr int64_t kMaxBufferSizeBytes = 1024 * 1024; +constexpr int64_t kMaxResultsBufferSizeBytes = kMaxBufferSizeBytes; + +using ::absl_testing::IsOkAndHolds; +using ::absl_testing::StatusIs; +using ::quiche::QuicheEndian; +using ::testing::_; +using ::testing::ElementsAre; +using ::testing::ElementsAreArray; +using ::testing::Field; +using ::testing::Mock; +using ::testing::Ne; +using ::testing::NiceMock; +using ::testing::SizeIs; +using ::testing::StrictMock; + +// Executor that collects callbacks and only runs them when released via +// RunCallbacks(). +class TestExecutor : public AsyncWritePacketExchanger::SiloExecutor { + public: + TestExecutor() = default; + ~TestExecutor() override = default; + + void Stop() override { + QUICHE_CHECK(IsOnSiloThread()); + + absl::MutexLock lock(mutex_); + stopped_ = true; + + // While a real executor is expected to be able to handle the case of + // stopping with pending callbacks, it's an error here because it means the + // executor received unexpected callbacks. + QUICHE_CHECK(callbacks_.empty()); + } + + void ExecuteInSilo(quiche::SingleUseCallback<void()> callback) override { + absl::MutexLock lock(mutex_); + QUICHE_CHECK(!stopped_); + callbacks_.push_back(std::move(callback)); + } + + void WaitForCallback() { WaitForNCallbacks(1); } + + void WaitForNCallbacks(int n) { + absl::MutexLock lock(mutex_); + QUICHE_CHECK_LE(num_expected_callbacks_, 0); + num_expected_callbacks_ = n; + + QUICHE_CHECK(mutex_.AwaitWithTimeout( + absl::Condition(this, &TestExecutor::HasExpectedCallbacks), + absl::Seconds(5))); + QUICHE_CHECK(!stopped_); + + num_expected_callbacks_ = 0; + } + + void RunCallbacks() { + QUICHE_CHECK(IsOnSiloThread()); + + std::vector<quiche::SingleUseCallback<void()>> to_run; + { + absl::MutexLock lock(mutex_); + QUICHE_CHECK(!stopped_); + to_run.swap(callbacks_); + } + for (auto& cb : to_run) { + std::move(cb)(); + } + } + + private: + bool HasExpectedCallbacks() const ABSL_SHARED_LOCKS_REQUIRED(mutex_) { + return stopped_ || callbacks_.size() >= num_expected_callbacks_; + } + + bool IsOnSiloThread() const override { + return std::this_thread::get_id() == thread_id_; + } + + const std::thread::id thread_id_ = std::this_thread::get_id(); + + mutable absl::Mutex mutex_; + bool stopped_ ABSL_GUARDED_BY(mutex_) = false; + std::vector<quiche::SingleUseCallback<void()>> callbacks_ + ABSL_GUARDED_BY(mutex_); + int num_expected_callbacks_ ABSL_GUARDED_BY(mutex_) = 0; +}; + +class AsyncWritePacketExchangerTest : public QuicTest { + protected: + AsyncWritePacketExchangerTest() { + auto executor = std::make_unique<TestExecutor>(); + executor_ = executor.get(); + exchanger_ = std::make_unique<AsyncWritePacketExchanger>( + kMtu, &mock_kernel_, &mock_netlink_, &mock_visitor_, /*is_tap=*/false, + "ifname", kMaxBufferSizeBytes, kMaxResultsBufferSizeBytes, + std::move(executor)); + } + + ~AsyncWritePacketExchangerTest() override = default; + + void ValidateOnMainThread() { + QUICHE_CHECK_EQ(std::this_thread::get_id(), thread_id_); + } + + void ValidateOffThread() { + QUICHE_CHECK_NE(std::this_thread::get_id(), thread_id_); + } + + const std::thread::id thread_id_ = std::this_thread::get_id(); + + StrictMock<MockKernel> mock_kernel_; + NiceMock<MockNetlink> mock_netlink_; + StrictMock<MockQboneClientPacketExchanger::MockVisitor> mock_visitor_; + TestExecutor* executor_; + std::unique_ptr<AsyncWritePacketExchanger> exchanger_; +}; + +TEST_F(AsyncWritePacketExchangerTest, WritePacket) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; + + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this, &packet](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) -> ssize_t { + ValidateOffThread(); + EXPECT_EQ(iov[0].iov_base, nullptr); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(absl::string_view(static_cast<const char*>(iov[1].iov_base), + iov[1].iov_len), + packet); + EXPECT_EQ(iov[1].iov_len, packet.size()); + return packet.size(); + }); + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + executor_->WaitForCallback(); + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + + EXPECT_CALL( + mock_visitor_, + OnWrite(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::WriteResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet.data()), + packet.size())))))); + executor_->RunCallbacks(); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, WritePacketError) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; + + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this, &packet](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) -> ssize_t { + ValidateOffThread(); + EXPECT_EQ(iov[0].iov_base, nullptr); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(absl::string_view(static_cast<const char*>(iov[1].iov_base), + iov[1].iov_len), + packet); + EXPECT_EQ(iov[1].iov_len, packet.size()); + errno = EAGAIN; + return -1; + }); + + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + executor_->WaitForCallback(); + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + + EXPECT_CALL(mock_visitor_, OnWrite(StatusIs(Ne(absl::StatusCode::kOk)))); + + executor_->RunCallbacks(); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, RestartExchanger) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + exchanger_->Stop(); + + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; + + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this, &packet](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) -> ssize_t { + ValidateOffThread(); + EXPECT_EQ(iov[0].iov_base, nullptr); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(absl::string_view(static_cast<const char*>(iov[1].iov_base), + iov[1].iov_len), + packet); + EXPECT_EQ(iov[1].iov_len, packet.size()); + return packet.size(); + }); + + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + executor_->WaitForCallback(); + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + + EXPECT_CALL( + mock_visitor_, + OnWrite(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::WriteResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet.data()), + packet.size())))))); + + executor_->RunCallbacks(); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, ReadPacket) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake_packet"; + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce( + [this, packet](int /*fd*/, const struct iovec* iov, int /*iovcnt*/) { + ValidateOnMainThread(); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(iov[1].iov_len, kMtu); + ::memcpy(iov[1].iov_base, packet.data(), packet.size()); + return packet.size(); + }); + EXPECT_CALL( + mock_visitor_, + OnRead(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::ReadResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet.data()), + packet.size())))))); + EXPECT_EQ(exchanger_->OnReadFromNetworkReady(/*max_packets_to_read=*/1), 1); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, ReadPacketError) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce( + [this](int /*fd*/, const struct iovec* /*iov*/, int /*iovcnt*/) { + ValidateOnMainThread(); + errno = ECOMM; + return -1; + }); + EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); + EXPECT_EQ(exchanger_->OnReadFromNetworkReady(/*max_packets_to_read=*/1), 0); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, ReadPacketBlocked) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce( + [this](int /*fd*/, const struct iovec* /*iov*/, int /*iovcnt*/) { + ValidateOnMainThread(); + errno = EAGAIN; + return -1; + }); + EXPECT_CALL(mock_visitor_, OnRead(StatusIs(Ne(absl::StatusCode::kOk)))); + EXPECT_EQ(exchanger_->OnReadFromNetworkReady(/*max_packets_to_read=*/1), 0); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, WriteBufferFull) { + auto executor = std::make_unique<TestExecutor>(); + executor_ = executor.get(); + + // Create an exchanger with a very small buffer. + exchanger_ = std::make_unique<AsyncWritePacketExchanger>( + kMtu, &mock_kernel_, &mock_netlink_, &mock_visitor_, false, + absl::string_view(), /*max_buffer_size_bytes=*/5, + kMaxResultsBufferSizeBytes, std::move(executor)); + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; // size 11 + + // Write should immediately fail and call OnWrite because buffer is full (max + // size 5, packet is 11) + EXPECT_CALL(mock_visitor_, + OnWrite(StatusIs(absl::StatusCode::kResourceExhausted))); + + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, WriteResultBufferFull) { + auto executor = std::make_unique<TestExecutor>(); + executor_ = executor.get(); + + // Create an exchanger with a very small response buffer. + exchanger_ = std::make_unique<AsyncWritePacketExchanger>( + kMtu, &mock_kernel_, &mock_netlink_, &mock_visitor_, false, + absl::string_view(), kMaxBufferSizeBytes, + /*max_result_buffer_size_bytes=*/5, std::move(executor)); + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; // size 11 + + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this, &packet](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) -> ssize_t { + ValidateOffThread(); + EXPECT_EQ(iov[0].iov_base, nullptr); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(absl::string_view(static_cast<const char*>(iov[1].iov_base), + iov[1].iov_len), + packet); + EXPECT_EQ(iov[1].iov_len, packet.size()); + return packet.size(); + }); + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + executor_->WaitForCallback(); + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + + // Expect no interaction with the visitor as no results were added to the + // result buffer. + + executor_->RunCallbacks(); + + exchanger_->Stop(); +} + +TEST_F(AsyncWritePacketExchangerTest, StopWithPendingWrites) { + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + std::string packet = "fake packet"; + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this, &packet](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) -> ssize_t { + ValidateOffThread(); + EXPECT_EQ(iov[0].iov_base, nullptr); + EXPECT_EQ(iov[0].iov_len, 0); + EXPECT_EQ(absl::string_view(static_cast<const char*>(iov[1].iov_base), + iov[1].iov_len), + packet); + EXPECT_EQ(iov[1].iov_len, packet.size()); + return packet.size(); + }); + + EXPECT_CALL( + mock_visitor_, + OnWrite(IsOkAndHolds(ElementsAre(Field( + &QboneClientPacketExchanger::WriteResult::packet, + ElementsAreArray(reinterpret_cast<const std::byte*>(packet.data()), + packet.size())))))); + + exchanger_->WritePacketToNetwork(absl::MakeConstSpan( + reinterpret_cast<const std::byte*>(packet.data()), packet.size())); + + // Immediately call Stop() without waiting for completion/callbacks. Expect to + // cleanly block on completion and run visitor callbacks. + exchanger_->Stop(); + + // Executor may have stale callbacks queued up, but visitor callbacks should + // already have been made. + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + executor_->RunCallbacks(); +} + +// Neighbor solicitation packets over a TAP interface are expected to be +// immediately responded to. Ensure that response is correctly handled +// off-thread. +TEST_F(AsyncWritePacketExchangerTest, ReadNeighborSolicitationPacket) { + auto executor = std::make_unique<TestExecutor>(); + executor_ = executor.get(); + + // Create an exchanger with TAP enabled. + exchanger_ = std::make_unique<AsyncWritePacketExchanger>( + kMtu, &mock_kernel_, &mock_netlink_, &mock_visitor_, /*is_tap=*/true, + absl::string_view(), kMaxBufferSizeBytes, kMaxResultsBufferSizeBytes, + std::move(executor)); + + exchanger_->Start(kReadFd, kWriteFd, /*exchanger=*/nullptr); + + ip6_hdr ip_hdr{}; + ip_hdr.ip6_vfc = 0x60; // Version 6 + ip_hdr.ip6_nxt = IPPROTO_ICMPV6; + inet_pton(AF_INET6, "fe80::2", &ip_hdr.ip6_src); + inet_pton(AF_INET6, "fe80::1", &ip_hdr.ip6_dst); + + icmp6_hdr icmp_hdr{}; + icmp_hdr.icmp6_type = ND_NEIGHBOR_SOLICIT; + + in6_addr target_address = QboneConstants::GatewayAddress()->GetIPv6(); + + std::vector<std::byte> l3_packet(sizeof(ip_hdr) + sizeof(icmp_hdr) + + sizeof(target_address)); + ::memcpy(l3_packet.data(), &ip_hdr, sizeof(ip_hdr)); + ::memcpy(l3_packet.data() + sizeof(ip_hdr), &icmp_hdr, sizeof(icmp_hdr)); + ::memcpy(l3_packet.data() + sizeof(ip_hdr) + sizeof(icmp_hdr), + &target_address, sizeof(target_address)); + + ethhdr eth_hdr{}; + eth_hdr.h_proto = QuicheEndian::HostToNet16(ETH_P_IPV6); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([this, l3_packet, eth_hdr](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) { + ValidateOnMainThread(); + EXPECT_EQ(iov[0].iov_len, ETH_HLEN); + ::memcpy(iov[0].iov_base, ð_hdr, ETH_HLEN); + EXPECT_EQ(iov[1].iov_len, kMtu); + ::memcpy(iov[1].iov_base, l3_packet.data(), l3_packet.size()); + return ETH_HLEN + l3_packet.size(); + }); + + // Expect neighbor solicitation response to be written out asynchronously. + EXPECT_CALL(mock_kernel_, writev(kWriteFd, _, 2)) + .WillOnce([this](int fd, const struct iovec* iov, int iovcnt) -> ssize_t { + ValidateOffThread(); + return iov[0].iov_len + iov[1].iov_len; + }); + + EXPECT_EQ(exchanger_->OnReadFromNetworkReady(/*max_packets_to_read=*/1), 1); + + executor_->WaitForCallback(); + EXPECT_TRUE(Mock::VerifyAndClear(&mock_visitor_)); + + // Because the read packet is link-local and immediately responded to, expect + // the visitor to be called with a *write* result. + EXPECT_CALL(mock_visitor_, OnWrite(IsOkAndHolds(SizeIs(1)))); + executor_->RunCallbacks(); + + exchanger_->Stop(); +} + +// Ensure that if an outer exchanger is provided, it is used for internal +// writes, e.g. neighbor solicitation responses. +TEST_F(AsyncWritePacketExchangerTest, + ReadNeighborSolicitationPacketWithOuterExchanger) { + auto executor = std::make_unique<TestExecutor>(); + executor_ = executor.get(); + + // Create an exchanger with TAP enabled. + exchanger_ = std::make_unique<AsyncWritePacketExchanger>( + kMtu, &mock_kernel_, &mock_netlink_, &mock_visitor_, /*is_tap=*/true, + absl::string_view(), kMaxBufferSizeBytes, kMaxResultsBufferSizeBytes, + std::move(executor)); + + StrictMock<MockQboneClientPacketExchanger> mock_outer_exchanger; + exchanger_->Start(kReadFd, kWriteFd, &mock_outer_exchanger); + + ip6_hdr ip_hdr{}; + ip_hdr.ip6_vfc = 0x60; // Version 6 + ip_hdr.ip6_nxt = IPPROTO_ICMPV6; + inet_pton(AF_INET6, "fe80::2", &ip_hdr.ip6_src); + inet_pton(AF_INET6, "fe80::1", &ip_hdr.ip6_dst); + + icmp6_hdr icmp_hdr{}; + icmp_hdr.icmp6_type = ND_NEIGHBOR_SOLICIT; + + in6_addr target_address = QboneConstants::GatewayAddress()->GetIPv6(); + + std::vector<std::byte> l3_packet(sizeof(ip_hdr) + sizeof(icmp_hdr) + + sizeof(target_address)); + ::memcpy(l3_packet.data(), &ip_hdr, sizeof(ip_hdr)); + ::memcpy(l3_packet.data() + sizeof(ip_hdr), &icmp_hdr, sizeof(icmp_hdr)); + ::memcpy(l3_packet.data() + sizeof(ip_hdr) + sizeof(icmp_hdr), + &target_address, sizeof(target_address)); + + ethhdr eth_hdr{}; + eth_hdr.h_proto = QuicheEndian::HostToNet16(ETH_P_IPV6); + + EXPECT_CALL(mock_kernel_, readv(kReadFd, _, 2)) + .WillOnce([this, l3_packet, eth_hdr](int /*fd*/, const struct iovec* iov, + int /*iovcnt*/) { + ValidateOnMainThread(); + EXPECT_EQ(iov[0].iov_len, ETH_HLEN); + ::memcpy(iov[0].iov_base, ð_hdr, ETH_HLEN); + EXPECT_EQ(iov[1].iov_len, kMtu); + ::memcpy(iov[1].iov_base, l3_packet.data(), l3_packet.size()); + return ETH_HLEN + l3_packet.size(); + }); + + // Expect neighbor solicitation response to be written via the outer + // exchanger. + EXPECT_CALL(mock_outer_exchanger, WritePacketToNetwork(_)); + + // No visitor callback expected because that is handled by the outer + // exchanger, here a mock that doesn't actually do it. + + EXPECT_EQ(exchanger_->OnReadFromNetworkReady(/*max_packets_to_read=*/1), 1); + + exchanger_->Stop(); +} + +} // namespace +} // namespace quic::test