From 7b4892b0720fe90351f36b735f38c7f60b8380c6 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Mon, 19 Oct 2026 15:58:41 -0700 Subject: [PATCH 01/15] fix(msvc): construct ApiContext in start() without ternary move. Co-authored-by: Cursor --- aether/api_protocol/make_api_call_sender.h | 39 +++++++++++++--------- 1 file changed, 23 insertions(+), 16 deletions(-) diff --git a/aether/api_protocol/make_api_call_sender.h b/aether/api_protocol/make_api_call_sender.h index b7058a14..94e5ab1a 100644 --- a/aether/api_protocol/make_api_call_sender.h +++ b/aether/api_protocol/make_api_call_sender.h @@ -105,23 +105,30 @@ class Operation final { * no operation member may be accessed afterward. */ void start() & noexcept { - auto api_context = protocol_context_ == nullptr - ? ApiContext{*api_} - : ApiContext{*api_, *protocol_context_}; - auto promise = std::invoke(fn_, api_context); - if (!promise) { - ex::set_stopped(std::move(receiver_)); - return; - } + auto run = [this](ApiContext& api_context) { + auto promise = std::invoke(fn_, api_context); + if (!promise) { + ex::set_stopped(std::move(receiver_)); + return; + } - result_subscription_ = promise.Subscribe([this](auto&& result) noexcept { - this->OnPromiseResult(std::forward(result)); - }); - auto& write_action = stream_->Write(std::move(api_context).Pack()); - write_subscription_ = write_action.status_event().Subscribe( - [this](WriteAction::Status status) noexcept { - this->OnWriteStatus(status); - }); + result_subscription_ = promise.Subscribe([this](auto&& result) noexcept { + this->OnPromiseResult(std::forward(result)); + }); + auto& write_action = stream_->Write(std::move(api_context).Pack()); + write_subscription_ = write_action.status_event().Subscribe( + [this](WriteAction::Status status) noexcept { + this->OnWriteStatus(status); + }); + }; + + if (protocol_context_ == nullptr) { + ApiContext api_context{*api_}; + run(api_context); + } else { + ApiContext api_context{*api_, *protocol_context_}; + run(api_context); + } } private: From 03d29534730938d387d04ebdeda718cc3ea7571b Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Tue, 20 Oct 2026 08:31:42 -0700 Subject: [PATCH 02/15] fix(win): treat synchronous WSARecv completion as pending recv. Set is_recv_pending_ when WSARecv returns success so IOCP OnRead can deliver bytes through HandleRecv without dropping sync completions. --- aether/transport/system_sockets/sockets/win_socket.cpp | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/aether/transport/system_sockets/sockets/win_socket.cpp b/aether/transport/system_sockets/sockets/win_socket.cpp index fcc6988a..3021e847 100644 --- a/aether/transport/system_sockets/sockets/win_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_socket.cpp @@ -168,6 +168,10 @@ bool WinSocket::RequestRecv() { AE_TELED_ERROR("Socket receive error {}", err_code); return false; } + // WSARecv returned 0: completion is available through the overlapped result + // (including synchronous success). Mark before releasing socket_lock_ so an + // IOCP thread cannot observe a completion with is_recv_pending_ still false. + is_recv_pending_ = true; return true; } From 8efd1cb0af21e52837c9d2dd1556951a752f18de Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Tue, 20 Oct 2026 08:56:28 -0700 Subject: [PATCH 03/15] docs(win): clarify recv pending flag vs socket_lock_ scope. --- aether/transport/system_sockets/sockets/win_socket.cpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/aether/transport/system_sockets/sockets/win_socket.cpp b/aether/transport/system_sockets/sockets/win_socket.cpp index 3021e847..dc56da2b 100644 --- a/aether/transport/system_sockets/sockets/win_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_socket.cpp @@ -169,8 +169,9 @@ bool WinSocket::RequestRecv() { return false; } // WSARecv returned 0: completion is available through the overlapped result - // (including synchronous success). Mark before releasing socket_lock_ so an - // IOCP thread cannot observe a completion with is_recv_pending_ still false. + // (including synchronous success). Publish is_recv_pending_ (atomic) before + // releasing socket_lock_. HandleRecv reads that flag before taking + // socket_lock_, so the mutex does not synchronize the early pending check. is_recv_pending_ = true; return true; } From f8cbe4a851ea71fb1fb735553135913e66def18a Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 13:41:00 -0700 Subject: [PATCH 04/15] Stop re-enqueueing TCP SendAction while overlapped send is pending. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 1 - 1 file changed, 1 deletion(-) diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 546847e9..195426bc 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -68,7 +68,6 @@ class SendAction final : public PacketSendAction { return; } if (*res == 0) { - reenqueue_ = true; return; } From 5187c1c8b42f24150f5f8c7176ad17949654d22e Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 14:15:48 -0700 Subject: [PATCH 05/15] Treat synchronous WSASend as pending until IOCP completion. Co-authored-by: Cursor --- aether/transport/system_sockets/sockets/win_socket.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/aether/transport/system_sockets/sockets/win_socket.cpp b/aether/transport/system_sockets/sockets/win_socket.cpp index dc56da2b..dd1f69ae 100644 --- a/aether/transport/system_sockets/sockets/win_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_socket.cpp @@ -73,6 +73,8 @@ std::optional WinSocket::Send(Span data) { AE_TELED_ERROR("Socket send error {}", error_code); return std::nullopt; } + is_send_pending_ = true; + return 0; } else { auto res = WSAGetOverlappedResult(socket_, &send_overlapped_, &bytes_transferred, false, &flags); From 176c92943d02713549e1d977caa0cd33c1a88f1a Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 19:36:45 -0700 Subject: [PATCH 06/15] Experiment: WinTcpSocket WSARecv user buffer 1500 -> 65536 (A/B only). Co-authored-by: Cursor --- aether/transport/system_sockets/sockets/win_tcp_socket.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp index f63aa793..a7a72ce2 100644 --- a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp @@ -47,7 +47,7 @@ LPFN_CONNECTEX GetConnectEx(SOCKET socket) { } // namespace win_socket_internal WinTcpSocket::WinTcpSocket(Ptr const& poller) - : WinSocket{*poller, 1500}, conn_overlapped_{} { + : WinSocket{*poller, 65536}, conn_overlapped_{} { bool created = false; // ::socket() sets WSA_FLAG_OVERLAPPED by default that allows us to use iocp From 50c80ba313a7c619c2b6267cbe947b88a9db46cb Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 20:26:04 -0700 Subject: [PATCH 07/15] tcp: allow SendAction to absorb unsent frames for batching prep. Keep per-frame WriteAction completion as sent_offset crosses segment ends; host stays inflight until the merged buffer is fully sent. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 134 ++++++++++++++++++++-- 1 file changed, 123 insertions(+), 11 deletions(-) diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 195426bc..4639e5bd 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -27,6 +27,7 @@ # include # include +# include # include "aether/ae_context.h" # include "aether/common.h" @@ -57,6 +58,37 @@ class SendAction final : public PacketSendAction { AE_CLASS_MOVE_ONLY(SendAction) + /** + * \brief Merge another fully formed, not-yet-sent frame into this action. + * \return false if either action already started sending or other is not a + * single-frame action. + */ + bool Absorb(SendAction&& other) noexcept { + if (sent_offset_ != 0 || other.sent_offset_ != 0) { + return false; + } + if (is_done_ || other.is_done_) { + return false; + } + if (socket_ != other.socket_) { + return false; + } + if (!other.segments_.empty() || !other.absorbed_.empty()) { + return false; + } + if (other.data_.empty()) { + return false; + } + + EnsurePrimarySegment(); + data_.insert(data_.end(), other.data_.begin(), other.data_.end()); + other.data_.clear(); + auto const absorbed_index = absorbed_.size(); + absorbed_.push_back(std::move(other)); + segments_.push_back({data_.size(), false, absorbed_index}); + return true; + } + void Send() override { reenqueue_ = false; @@ -64,7 +96,7 @@ class SendAction final : public PacketSendAction { auto res = socket_->Send(Span{data_.data() + sent_offset_, size_to_send}); if (!res) { AE_TELED_ERROR("Data has not been written"); - SetStatus(WriteAction::Status::kFail); + AbortPending(WriteAction::Status::kFail); return; } if (*res == 0) { @@ -73,17 +105,18 @@ class SendAction final : public PacketSendAction { AE_TELED_DEBUG("Data has been written size {} data {}", *res, data_); sent_offset_ += *res; - - if (sent_offset_ >= data_.size()) { - SetStatus(WriteAction::Status::kSuccess); - } + OnSentProgress(); // reenque must happen on event from socket } void Stop() noexcept override { - is_done_ = true; - set_status_.Reset(); - WriteAction::SetStatus(WriteAction::Status::kStop); + if (segments_.empty()) { + is_done_ = true; + set_status_.Reset(); + WriteAction::SetStatus(WriteAction::Status::kStop); + return; + } + AbortPending(WriteAction::Status::kStop); } bool is_done() const override { return is_done_; } @@ -91,16 +124,95 @@ class SendAction final : public PacketSendAction { protected: void SetStatus(WriteAction::Status status) noexcept override { - is_done_ = true; - set_status_ = ae_context_.scheduler().Task( - [&, status]() { WriteAction::SetStatus(status); }); + FinalizeHost(status); } private: + struct Segment { + std::size_t end; + bool is_host; + std::size_t absorbed_index; + }; + + SendAction& SegmentAction(Segment const& segment) noexcept { + return segment.is_host ? *this : absorbed_[segment.absorbed_index]; + } + + void EnsurePrimarySegment() { + if (!segments_.empty()) { + return; + } + segments_.push_back({data_.size(), true, 0}); + } + + bool HostHasMultipleSegments() const noexcept { + return segments_.size() > 1; + } + + void EmitWriteStatus(SendAction& target, + WriteAction::Status status) noexcept { + target.set_status_.Reset(); + target.set_status_ = target.ae_context_.scheduler().Task( + [&target, status]() { target.WriteAction::SetStatus(status); }); + } + + void CompleteSegment(SendAction& target, + WriteAction::Status status) noexcept { + if (&target != this) { + target.is_done_ = true; + EmitWriteStatus(target, status); + return; + } + if (HostHasMultipleSegments()) { + EmitWriteStatus(target, status); + return; + } + FinalizeHost(status); + } + + void FinalizeHost(WriteAction::Status status) noexcept { + is_done_ = true; + EmitWriteStatus(*this, status); + } + + void OnSentProgress() noexcept { + if (segments_.empty()) { + if (sent_offset_ >= data_.size()) { + FinalizeHost(WriteAction::Status::kSuccess); + } + return; + } + + while (completed_segment_idx_ < segments_.size() && + sent_offset_ >= segments_[completed_segment_idx_].end) { + CompleteSegment(SegmentAction(segments_[completed_segment_idx_]), + WriteAction::Status::kSuccess); + ++completed_segment_idx_; + } + if (sent_offset_ >= data_.size()) { + is_done_ = true; + } + } + + void AbortPending(WriteAction::Status status) noexcept { + if (segments_.empty()) { + FinalizeHost(status); + return; + } + is_done_ = true; + for (; completed_segment_idx_ < segments_.size(); + ++completed_segment_idx_) { + CompleteSegment(SegmentAction(segments_[completed_segment_idx_]), status); + } + } + AeContext ae_context_; Sock* socket_; DataBuffer data_; + std::vector segments_; + std::vector absorbed_; std::size_t sent_offset_ = 0; + std::size_t completed_segment_idx_ = 0; bool is_done_ = false; bool reenqueue_ = false; TaskSubscription set_status_; From f81f3c5b69c75bec567a153225afbb68346b1a35 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 20:45:46 -0700 Subject: [PATCH 08/15] Fix batch host Finish so PacketQueueManager keeps inflight send. Emit host frame status without Action::Finish until all segment tasks complete; add queue regression tests in test-transport. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 49 +++- tests/test-transport/CMakeLists.txt | 1 + tests/test-transport/main.cpp | 5 + tests/test-transport/test-tcp-send-action.cpp | 230 ++++++++++++++++++ 4 files changed, 278 insertions(+), 7 deletions(-) create mode 100644 tests/test-transport/test-tcp-send-action.cpp diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 4639e5bd..d0594ba7 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -149,22 +149,39 @@ class SendAction final : public PacketSendAction { return segments_.size() > 1; } - void EmitWriteStatus(SendAction& target, - WriteAction::Status status) noexcept { + void OnTerminalTaskDone() noexcept { + if (pending_status_tasks_ == 0) { + return; + } + --pending_status_tasks_; + TryFinishHostAction(); + } + + void ScheduleSegmentTerminal(SendAction& target, + WriteAction::Status status, + bool finish_action) noexcept { + ++pending_status_tasks_; target.set_status_.Reset(); - target.set_status_ = target.ae_context_.scheduler().Task( - [&target, status]() { target.WriteAction::SetStatus(status); }); + target.set_status_ = ae_context_.scheduler().Task( + [this, &target, status, finish_action]() { + if (finish_action) { + target.WriteAction::SetStatus(status); + } else { + target.status_event_.Emit(status); + } + OnTerminalTaskDone(); + }); } void CompleteSegment(SendAction& target, WriteAction::Status status) noexcept { if (&target != this) { target.is_done_ = true; - EmitWriteStatus(target, status); + ScheduleSegmentTerminal(target, status, true); return; } if (HostHasMultipleSegments()) { - EmitWriteStatus(target, status); + ScheduleSegmentTerminal(target, status, false); return; } FinalizeHost(status); @@ -172,7 +189,22 @@ class SendAction final : public PacketSendAction { void FinalizeHost(WriteAction::Status status) noexcept { is_done_ = true; - EmitWriteStatus(*this, status); + ++pending_status_tasks_; + set_status_.Reset(); + set_status_ = ae_context_.scheduler().Task([this, status]() { + WriteAction::SetStatus(status); + OnTerminalTaskDone(); + }); + } + + void TryFinishHostAction() noexcept { + if (!HostHasMultipleSegments()) { + return; + } + if (is_finished() || !is_done_ || pending_status_tasks_ != 0) { + return; + } + Finish(); } void OnSentProgress() noexcept { @@ -191,6 +223,7 @@ class SendAction final : public PacketSendAction { } if (sent_offset_ >= data_.size()) { is_done_ = true; + TryFinishHostAction(); } } @@ -204,6 +237,7 @@ class SendAction final : public PacketSendAction { ++completed_segment_idx_) { CompleteSegment(SegmentAction(segments_[completed_segment_idx_]), status); } + TryFinishHostAction(); } AeContext ae_context_; @@ -213,6 +247,7 @@ class SendAction final : public PacketSendAction { std::vector absorbed_; std::size_t sent_offset_ = 0; std::size_t completed_segment_idx_ = 0; + std::size_t pending_status_tasks_ = 0; bool is_done_ = false; bool reenqueue_ = false; TaskSubscription set_status_; diff --git a/tests/test-transport/CMakeLists.txt b/tests/test-transport/CMakeLists.txt index 565df837..cf3d1411 100644 --- a/tests/test-transport/CMakeLists.txt +++ b/tests/test-transport/CMakeLists.txt @@ -17,6 +17,7 @@ cmake_minimum_required( VERSION 3.18 ) list(APPEND test_srcs main.cpp test-data-packet-collector.cpp + test-tcp-send-action.cpp ) if(NOT CM_PLATFORM) diff --git a/tests/test-transport/main.cpp b/tests/test-transport/main.cpp index 2e3443d9..ff98af4a 100644 --- a/tests/test-transport/main.cpp +++ b/tests/test-transport/main.cpp @@ -21,8 +21,13 @@ void tearDown() {} extern int test_data_packet_collector(); +namespace ae::test_tcp_send_action { +int test_tcp_send_action(); +} + int main() { int res = 0; res += test_data_packet_collector(); + res += ae::test_tcp_send_action::test_tcp_send_action(); return res; } diff --git a/tests/test-transport/test-tcp-send-action.cpp b/tests/test-transport/test-tcp-send-action.cpp new file mode 100644 index 00000000..82167528 --- /dev/null +++ b/tests/test-transport/test-tcp-send-action.cpp @@ -0,0 +1,230 @@ +/* + * Regression tests for tcp_internal::SendAction batch lifecycle. + */ + +#include + +#include +#include +#include + +#include "aether/ae_context.h" +#include "aether/events/events.h" +#include "aether/tasks/manual_task_scheduler.h" +#include "aether/transport/packet_queue_manager.h" +#include "aether/transport/system_sockets/sockets/isocket.h" +#include "aether/transport/system_sockets/tcp/tcp.h" + +namespace ae::test_tcp_send_action { + +struct TestContext : public Env { + template + decltype(auto) Update(A&&... a) { + return sched.Update(std::forward(a)...); + } + + void* find_component(EnvId id) noexcept override { + if (id == EnvTypeId::value) { + return &sched; + } + if (id == EnvTypeId::value) { + return &event_system_; + } + return nullptr; + } + + TaskScheduler& scheduler() const { return sched; } + EventSystem& event_system() const { return event_system_; } + + mutable TaskScheduler sched; + mutable EventSystem event_system_; +}; + +struct MockSock final : ISocket { + std::vector sent; + std::optional limit_next{}; + bool fail_next = false; + + ISocket& ReadyToWrite(ReadyToWriteCb) override { return *this; } + ISocket& RecvData(RecvDataCb) override { return *this; } + ISocket& Error(ErrorCb) override { return *this; } + ISocket& Connect(AddressPort const&, ConnectedCb) override { return *this; } + void Disconnect() override {} + + std::optional Send(Span data) override { + if (fail_next) { + fail_next = false; + return std::nullopt; + } + auto n = data.size(); + if (limit_next) { + n = std::min(n, *limit_next); + limit_next.reset(); + } + sent.insert(sent.end(), data.begin(), + data.begin() + static_cast(n)); + return n; + } +}; + +using SendAction = tcp_internal::SendAction; +static constexpr std::size_t kQueueSize = 4; + +static DataBuffer Frame(std::uint8_t tag) { + return DataBuffer{tag, static_cast(tag + 1), tag}; +} + +static void Pump(TestContext& ctx) { static_cast(ctx.Update()); } + +void test_QueueBatchHostNotFinishedAfterFirstSegment() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + sock.limit_next = Frame(1).size(); + PacketQueueManager queue{ctx}; + + auto* host = queue.AddPacket(ctx, sock, Frame(1)); + TEST_ASSERT_NOT_NULL(host); + + SendAction extra_a{ctx, sock, Frame(2)}; + SendAction extra_b{ctx, sock, Frame(3)}; + + int ok_host = 0; + int ok_a = 0; + int ok_b = 0; + int host_finished = 0; + auto sub_host = host->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_host; + } + }); + auto sub_a = extra_a.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_a; + } + }); + auto sub_b = extra_b.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_b; + } + }); + auto sub_fin = host->finished_event().Subscribe([&]() { ++host_finished; }); + + TEST_ASSERT_TRUE(host->Absorb(std::move(extra_a))); + TEST_ASSERT_TRUE(host->Absorb(std::move(extra_b))); + + Pump(tctx); + Pump(tctx); + + TEST_ASSERT_EQUAL(1, ok_host); + TEST_ASSERT_EQUAL(0, ok_a); + TEST_ASSERT_EQUAL(0, ok_b); + TEST_ASSERT_EQUAL(0, host_finished); + TEST_ASSERT_FALSE(host->is_finished()); + TEST_ASSERT_EQUAL(Frame(1).size(), sock.sent.size()); + + queue.Send(); + TEST_ASSERT_FALSE(host->is_finished()); + Pump(tctx); + Pump(tctx); + Pump(tctx); + + TEST_ASSERT_EQUAL(9U, sock.sent.size()); + TEST_ASSERT_EQUAL(1, ok_host); + TEST_ASSERT_EQUAL(1, ok_a); + TEST_ASSERT_EQUAL(1, ok_b); + TEST_ASSERT_EQUAL(1, host_finished); + TEST_ASSERT_TRUE(host->is_finished()); + + auto* next = queue.AddPacket(ctx, sock, Frame(4)); + TEST_ASSERT_NOT_NULL(next); + queue.Send(); + Pump(tctx); + TEST_ASSERT_TRUE(next->is_finished()); +} + +void test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + sock.limit_next = Frame(10).size(); + PacketQueueManager queue{ctx}; + + auto* host = queue.AddPacket(ctx, sock, Frame(10)); + SendAction extra_a{ctx, sock, Frame(11)}; + SendAction extra_b{ctx, sock, Frame(12)}; + + int ok_host = 0; + int fail_a = 0; + int fail_b = 0; + int host_finished = 0; + host->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_host; + } + }); + extra_a.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kFail) { + ++fail_a; + } + }); + extra_b.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kFail) { + ++fail_b; + } + }); + host->finished_event().Subscribe([&]() { ++host_finished; }); + + TEST_ASSERT_TRUE(host->Absorb(std::move(extra_a))); + TEST_ASSERT_TRUE(host->Absorb(std::move(extra_b))); + + Pump(tctx); + Pump(tctx); + TEST_ASSERT_EQUAL(1, ok_host); + TEST_ASSERT_FALSE(host->is_finished()); + + sock.fail_next = true; + queue.Send(); + Pump(tctx); + Pump(tctx); + Pump(tctx); + + TEST_ASSERT_EQUAL(1, ok_host); + TEST_ASSERT_EQUAL(1, fail_a); + TEST_ASSERT_EQUAL(1, fail_b); + TEST_ASSERT_EQUAL(1, host_finished); + TEST_ASSERT_TRUE(host->is_finished()); +} + +void test_QueueSingleFrameUnchanged() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + PacketQueueManager queue{ctx}; + + auto* action = queue.AddPacket(ctx, sock, Frame(7)); + int ok = 0; + int finished = 0; + action->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok; + } + }); + action->finished_event().Subscribe([&]() { ++finished; }); + + queue.Send(); + Pump(tctx); + + TEST_ASSERT_EQUAL(1, ok); + TEST_ASSERT_EQUAL(1, finished); + TEST_ASSERT_TRUE(action->is_finished()); +} + +int test_tcp_send_action() { + UNITY_BEGIN(); + RUN_TEST(test_QueueBatchHostNotFinishedAfterFirstSegment); + RUN_TEST(test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess); + RUN_TEST(test_QueueSingleFrameUnchanged); + return UNITY_END(); +} +} // namespace ae::test_tcp_send_action From caafc45f3b93a5b3c04f701d87606c4f884f8552 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 21:22:08 -0700 Subject: [PATCH 09/15] Reject SendAction Absorb after socket send has started. Track send_started_ before Send even when zero bytes are written; add pending-socket regression tests. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 5 + tests/test-transport/test-tcp-send-action.cpp | 114 ++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index d0594ba7..928863e1 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -64,6 +64,9 @@ class SendAction final : public PacketSendAction { * single-frame action. */ bool Absorb(SendAction&& other) noexcept { + if (send_started_ || other.send_started_) { + return false; + } if (sent_offset_ != 0 || other.sent_offset_ != 0) { return false; } @@ -93,6 +96,7 @@ class SendAction final : public PacketSendAction { reenqueue_ = false; auto size_to_send = data_.size() - sent_offset_; + send_started_ = true; auto res = socket_->Send(Span{data_.data() + sent_offset_, size_to_send}); if (!res) { AE_TELED_ERROR("Data has not been written"); @@ -249,6 +253,7 @@ class SendAction final : public PacketSendAction { std::size_t completed_segment_idx_ = 0; std::size_t pending_status_tasks_ = 0; bool is_done_ = false; + bool send_started_ = false; bool reenqueue_ = false; TaskSubscription set_status_; }; diff --git a/tests/test-transport/test-tcp-send-action.cpp b/tests/test-transport/test-tcp-send-action.cpp index 82167528..97181b62 100644 --- a/tests/test-transport/test-tcp-send-action.cpp +++ b/tests/test-transport/test-tcp-send-action.cpp @@ -44,6 +44,8 @@ struct MockSock final : ISocket { std::vector sent; std::optional limit_next{}; bool fail_next = false; + bool defer_next = false; + DataBuffer deferred{}; ISocket& ReadyToWrite(ReadyToWriteCb) override { return *this; } ISocket& RecvData(RecvDataCb) override { return *this; } @@ -56,6 +58,17 @@ struct MockSock final : ISocket { fail_next = false; return std::nullopt; } + if (!deferred.empty()) { + sent.insert(sent.end(), deferred.begin(), deferred.end()); + auto n = deferred.size(); + deferred.clear(); + return n; + } + if (defer_next) { + defer_next = false; + deferred.assign(data.begin(), data.end()); + return 0; + } auto n = data.size(); if (limit_next) { n = std::min(n, *limit_next); @@ -76,6 +89,104 @@ static DataBuffer Frame(std::uint8_t tag) { static void Pump(TestContext& ctx) { static_cast(ctx.Update()); } +static void AssertBuffersEqual(DataBuffer const& a, DataBuffer const& b) { + TEST_ASSERT_EQUAL(a.size(), b.size()); + if (a.size() == b.size()) { + TEST_ASSERT_EQUAL_UINT8_ARRAY(a.data(), b.data(), a.size()); + } +} + +void test_AbsorbAllowedBeforeSend() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + + SendAction host{ctx, sock, Frame(1)}; + SendAction other{ctx, sock, Frame(2)}; + TEST_ASSERT_TRUE(host.Absorb(std::move(other))); + TEST_ASSERT_TRUE(sock.sent.empty()); +} + +void test_AbsorbRejectedWhenHostSendPending() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + sock.defer_next = true; + + auto const host_frame = Frame(1); + auto const other_frame = Frame(2); + SendAction host{ctx, sock, Frame(1)}; + SendAction other{ctx, sock, Frame(2)}; + + int host_status = 0; + int other_status = 0; + auto sub_host = host.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++host_status; + } + }); + auto sub_other = other.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++other_status; + } + }); + + host.Send(); + TEST_ASSERT_FALSE(host.Absorb(std::move(other))); + + host.Send(); + Pump(tctx); + AssertBuffersEqual(host_frame, sock.sent); + + other.Send(); + Pump(tctx); + auto expect = host_frame; + expect.insert(expect.end(), other_frame.begin(), other_frame.end()); + AssertBuffersEqual(expect, sock.sent); + TEST_ASSERT_EQUAL(1, host_status); + TEST_ASSERT_EQUAL(1, other_status); +} + +void test_AbsorbRejectedWhenOtherSendPending() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + sock.defer_next = true; + + auto const host_frame = Frame(10); + auto const other_frame = Frame(11); + SendAction host{ctx, sock, Frame(10)}; + SendAction other{ctx, sock, Frame(11)}; + + int host_status = 0; + int other_status = 0; + host.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++host_status; + } + }); + other.status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++other_status; + } + }); + + other.Send(); + TEST_ASSERT_FALSE(host.Absorb(std::move(other))); + + other.Send(); + Pump(tctx); + AssertBuffersEqual(other_frame, sock.sent); + + host.Send(); + Pump(tctx); + auto expect = other_frame; + expect.insert(expect.end(), host_frame.begin(), host_frame.end()); + AssertBuffersEqual(expect, sock.sent); + TEST_ASSERT_EQUAL(1, host_status); + TEST_ASSERT_EQUAL(1, other_status); +} + void test_QueueBatchHostNotFinishedAfterFirstSegment() { TestContext tctx; AeContext ctx{tctx}; @@ -222,6 +333,9 @@ void test_QueueSingleFrameUnchanged() { int test_tcp_send_action() { UNITY_BEGIN(); + RUN_TEST(test_AbsorbAllowedBeforeSend); + RUN_TEST(test_AbsorbRejectedWhenHostSendPending); + RUN_TEST(test_AbsorbRejectedWhenOtherSendPending); RUN_TEST(test_QueueBatchHostNotFinishedAfterFirstSegment); RUN_TEST(test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess); RUN_TEST(test_QueueSingleFrameUnchanged); From 0a74ca1162c69a99d03ea9c65f5f7ce895c6b9bc Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 21:52:28 -0700 Subject: [PATCH 10/15] Keep absorbed SendAction objects in queue for stable WriteAction identity. Absorb by reference with batch_host_ delegation; merge only data buffers and clear host links when the batch Action finishes. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 49 +++++-- tests/test-transport/test-tcp-send-action.cpp | 125 +++++++++++++----- 2 files changed, 124 insertions(+), 50 deletions(-) diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 928863e1..6298cf63 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -63,7 +63,10 @@ class SendAction final : public PacketSendAction { * \return false if either action already started sending or other is not a * single-frame action. */ - bool Absorb(SendAction&& other) noexcept { + bool Absorb(SendAction& other) noexcept { + if (this == &other) { + return false; + } if (send_started_ || other.send_started_) { return false; } @@ -73,10 +76,13 @@ class SendAction final : public PacketSendAction { if (is_done_ || other.is_done_) { return false; } + if (other.batch_host_ != nullptr) { + return false; + } if (socket_ != other.socket_) { return false; } - if (!other.segments_.empty() || !other.absorbed_.empty()) { + if (!other.segments_.empty()) { return false; } if (other.data_.empty()) { @@ -86,13 +92,15 @@ class SendAction final : public PacketSendAction { EnsurePrimarySegment(); data_.insert(data_.end(), other.data_.begin(), other.data_.end()); other.data_.clear(); - auto const absorbed_index = absorbed_.size(); - absorbed_.push_back(std::move(other)); - segments_.push_back({data_.size(), false, absorbed_index}); + other.batch_host_ = this; + segments_.push_back({data_.size(), &other}); return true; } void Send() override { + if (batch_host_ != nullptr) { + return; + } reenqueue_ = false; auto size_to_send = data_.size() - sent_offset_; @@ -114,6 +122,10 @@ class SendAction final : public PacketSendAction { } void Stop() noexcept override { + if (batch_host_ != nullptr) { + batch_host_->Stop(); + return; + } if (segments_.empty()) { is_done_ = true; set_status_.Reset(); @@ -123,7 +135,12 @@ class SendAction final : public PacketSendAction { AbortPending(WriteAction::Status::kStop); } - bool is_done() const override { return is_done_; } + bool is_done() const override { + if (batch_host_ != nullptr && !is_done_) { + return true; + } + return is_done_; + } bool re_enqueue() const override { return reenqueue_; } protected: @@ -134,19 +151,22 @@ class SendAction final : public PacketSendAction { private: struct Segment { std::size_t end; - bool is_host; - std::size_t absorbed_index; + SendAction* action; }; - SendAction& SegmentAction(Segment const& segment) noexcept { - return segment.is_host ? *this : absorbed_[segment.absorbed_index]; + void ClearBatchHostLinks() noexcept { + for (auto const& segment : segments_) { + if (segment.action != this) { + segment.action->batch_host_ = nullptr; + } + } } void EnsurePrimarySegment() { if (!segments_.empty()) { return; } - segments_.push_back({data_.size(), true, 0}); + segments_.push_back({data_.size(), this}); } bool HostHasMultipleSegments() const noexcept { @@ -208,6 +228,7 @@ class SendAction final : public PacketSendAction { if (is_finished() || !is_done_ || pending_status_tasks_ != 0) { return; } + ClearBatchHostLinks(); Finish(); } @@ -221,7 +242,7 @@ class SendAction final : public PacketSendAction { while (completed_segment_idx_ < segments_.size() && sent_offset_ >= segments_[completed_segment_idx_].end) { - CompleteSegment(SegmentAction(segments_[completed_segment_idx_]), + CompleteSegment(*segments_[completed_segment_idx_].action, WriteAction::Status::kSuccess); ++completed_segment_idx_; } @@ -239,7 +260,7 @@ class SendAction final : public PacketSendAction { is_done_ = true; for (; completed_segment_idx_ < segments_.size(); ++completed_segment_idx_) { - CompleteSegment(SegmentAction(segments_[completed_segment_idx_]), status); + CompleteSegment(*segments_[completed_segment_idx_].action, status); } TryFinishHostAction(); } @@ -248,7 +269,7 @@ class SendAction final : public PacketSendAction { Sock* socket_; DataBuffer data_; std::vector segments_; - std::vector absorbed_; + SendAction* batch_host_ = nullptr; std::size_t sent_offset_ = 0; std::size_t completed_segment_idx_ = 0; std::size_t pending_status_tasks_ = 0; diff --git a/tests/test-transport/test-tcp-send-action.cpp b/tests/test-transport/test-tcp-send-action.cpp index 97181b62..c25f0bd0 100644 --- a/tests/test-transport/test-tcp-send-action.cpp +++ b/tests/test-transport/test-tcp-send-action.cpp @@ -81,7 +81,7 @@ struct MockSock final : ISocket { }; using SendAction = tcp_internal::SendAction; -static constexpr std::size_t kQueueSize = 4; +static constexpr std::size_t kQueueSize = 8; static DataBuffer Frame(std::uint8_t tag) { return DataBuffer{tag, static_cast(tag + 1), tag}; @@ -100,10 +100,13 @@ void test_AbsorbAllowedBeforeSend() { TestContext tctx; AeContext ctx{tctx}; MockSock sock; + PacketQueueManager queue{ctx}; - SendAction host{ctx, sock, Frame(1)}; - SendAction other{ctx, sock, Frame(2)}; - TEST_ASSERT_TRUE(host.Absorb(std::move(other))); + auto* host = queue.AddPacket(ctx, sock, Frame(1)); + auto* other = queue.AddPacket(ctx, sock, Frame(2)); + TEST_ASSERT_NOT_NULL(host); + TEST_ASSERT_NOT_NULL(other); + TEST_ASSERT_TRUE(host->Absorb(*other)); TEST_ASSERT_TRUE(sock.sent.empty()); } @@ -112,33 +115,33 @@ void test_AbsorbRejectedWhenHostSendPending() { AeContext ctx{tctx}; MockSock sock; sock.defer_next = true; + PacketQueueManager queue{ctx}; auto const host_frame = Frame(1); auto const other_frame = Frame(2); - SendAction host{ctx, sock, Frame(1)}; - SendAction other{ctx, sock, Frame(2)}; + auto* host = queue.AddPacket(ctx, sock, Frame(1)); + auto* other = queue.AddPacket(ctx, sock, Frame(2)); int host_status = 0; int other_status = 0; - auto sub_host = host.status_event().Subscribe([&](WriteAction::Status s) { + host->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++host_status; } }); - auto sub_other = other.status_event().Subscribe([&](WriteAction::Status s) { + other->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++other_status; } }); - host.Send(); - TEST_ASSERT_FALSE(host.Absorb(std::move(other))); + host->Send(); + TEST_ASSERT_FALSE(host->Absorb(*other)); - host.Send(); - Pump(tctx); + host->Send(); AssertBuffersEqual(host_frame, sock.sent); - other.Send(); + queue.Send(); Pump(tctx); auto expect = host_frame; expect.insert(expect.end(), other_frame.begin(), other_frame.end()); @@ -152,33 +155,33 @@ void test_AbsorbRejectedWhenOtherSendPending() { AeContext ctx{tctx}; MockSock sock; sock.defer_next = true; + PacketQueueManager queue{ctx}; auto const host_frame = Frame(10); auto const other_frame = Frame(11); - SendAction host{ctx, sock, Frame(10)}; - SendAction other{ctx, sock, Frame(11)}; + auto* host = queue.AddPacket(ctx, sock, Frame(10)); + auto* other = queue.AddPacket(ctx, sock, Frame(11)); int host_status = 0; int other_status = 0; - host.status_event().Subscribe([&](WriteAction::Status s) { + host->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++host_status; } }); - other.status_event().Subscribe([&](WriteAction::Status s) { + other->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++other_status; } }); - other.Send(); - TEST_ASSERT_FALSE(host.Absorb(std::move(other))); + other->Send(); + TEST_ASSERT_FALSE(host->Absorb(*other)); - other.Send(); - Pump(tctx); + other->Send(); AssertBuffersEqual(other_frame, sock.sent); - host.Send(); + queue.Send(); Pump(tctx); auto expect = other_frame; expect.insert(expect.end(), host_frame.begin(), host_frame.end()); @@ -195,13 +198,14 @@ void test_QueueBatchHostNotFinishedAfterFirstSegment() { PacketQueueManager queue{ctx}; auto* host = queue.AddPacket(ctx, sock, Frame(1)); - TEST_ASSERT_NOT_NULL(host); - - SendAction extra_a{ctx, sock, Frame(2)}; - SendAction extra_b{ctx, sock, Frame(3)}; + auto* extra_a = queue.AddPacket(ctx, sock, Frame(2)); + auto* extra_b = queue.AddPacket(ctx, sock, Frame(3)); + WriteAction* wa_a = extra_a; + WriteAction* wa_b = extra_b; int ok_host = 0; int ok_a = 0; + int ok_a_late = 0; int ok_b = 0; int host_finished = 0; auto sub_host = host->status_event().Subscribe([&](WriteAction::Status s) { @@ -209,20 +213,26 @@ void test_QueueBatchHostNotFinishedAfterFirstSegment() { ++ok_host; } }); - auto sub_a = extra_a.status_event().Subscribe([&](WriteAction::Status s) { + auto sub_a = wa_a->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++ok_a; } }); - auto sub_b = extra_b.status_event().Subscribe([&](WriteAction::Status s) { + auto sub_b = wa_b->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kSuccess) { ++ok_b; } }); auto sub_fin = host->finished_event().Subscribe([&]() { ++host_finished; }); - TEST_ASSERT_TRUE(host->Absorb(std::move(extra_a))); - TEST_ASSERT_TRUE(host->Absorb(std::move(extra_b))); + TEST_ASSERT_TRUE(host->Absorb(*extra_a)); + TEST_ASSERT_TRUE(host->Absorb(*extra_b)); + + wa_a->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_a_late; + } + }); Pump(tctx); Pump(tctx); @@ -232,6 +242,7 @@ void test_QueueBatchHostNotFinishedAfterFirstSegment() { TEST_ASSERT_EQUAL(0, ok_b); TEST_ASSERT_EQUAL(0, host_finished); TEST_ASSERT_FALSE(host->is_finished()); + TEST_ASSERT_FALSE(wa_a->is_finished()); TEST_ASSERT_EQUAL(Frame(1).size(), sock.sent.size()); queue.Send(); @@ -243,9 +254,12 @@ void test_QueueBatchHostNotFinishedAfterFirstSegment() { TEST_ASSERT_EQUAL(9U, sock.sent.size()); TEST_ASSERT_EQUAL(1, ok_host); TEST_ASSERT_EQUAL(1, ok_a); + TEST_ASSERT_EQUAL(1, ok_a_late); TEST_ASSERT_EQUAL(1, ok_b); TEST_ASSERT_EQUAL(1, host_finished); TEST_ASSERT_TRUE(host->is_finished()); + TEST_ASSERT_TRUE(wa_a->is_finished()); + TEST_ASSERT_TRUE(wa_b->is_finished()); auto* next = queue.AddPacket(ctx, sock, Frame(4)); TEST_ASSERT_NOT_NULL(next); @@ -262,8 +276,10 @@ void test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess() { PacketQueueManager queue{ctx}; auto* host = queue.AddPacket(ctx, sock, Frame(10)); - SendAction extra_a{ctx, sock, Frame(11)}; - SendAction extra_b{ctx, sock, Frame(12)}; + auto* extra_a = queue.AddPacket(ctx, sock, Frame(11)); + auto* extra_b = queue.AddPacket(ctx, sock, Frame(12)); + WriteAction* wa_a = extra_a; + WriteAction* wa_b = extra_b; int ok_host = 0; int fail_a = 0; @@ -274,20 +290,20 @@ void test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess() { ++ok_host; } }); - extra_a.status_event().Subscribe([&](WriteAction::Status s) { + wa_a->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kFail) { ++fail_a; } }); - extra_b.status_event().Subscribe([&](WriteAction::Status s) { + wa_b->status_event().Subscribe([&](WriteAction::Status s) { if (s == WriteAction::Status::kFail) { ++fail_b; } }); host->finished_event().Subscribe([&]() { ++host_finished; }); - TEST_ASSERT_TRUE(host->Absorb(std::move(extra_a))); - TEST_ASSERT_TRUE(host->Absorb(std::move(extra_b))); + TEST_ASSERT_TRUE(host->Absorb(*extra_a)); + TEST_ASSERT_TRUE(host->Absorb(*extra_b)); Pump(tctx); Pump(tctx); @@ -305,6 +321,42 @@ void test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess() { TEST_ASSERT_EQUAL(1, fail_b); TEST_ASSERT_EQUAL(1, host_finished); TEST_ASSERT_TRUE(host->is_finished()); + TEST_ASSERT_TRUE(wa_a->is_finished()); + TEST_ASSERT_TRUE(wa_b->is_finished()); +} + +void test_QueueBatchStopChildBeforeSend() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + PacketQueueManager queue{ctx}; + + auto* host = queue.AddPacket(ctx, sock, Frame(20)); + auto* child = queue.AddPacket(ctx, sock, Frame(21)); + WriteAction* wa_child = child; + + int stop_host = 0; + int stop_child = 0; + host->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kStop) { + ++stop_host; + } + }); + wa_child->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kStop) { + ++stop_child; + } + }); + + TEST_ASSERT_TRUE(host->Absorb(*child)); + wa_child->Stop(); + Pump(tctx); + + TEST_ASSERT_TRUE(sock.sent.empty()); + TEST_ASSERT_EQUAL(1, stop_host); + TEST_ASSERT_EQUAL(1, stop_child); + TEST_ASSERT_TRUE(host->is_finished()); + TEST_ASSERT_TRUE(wa_child->is_finished()); } void test_QueueSingleFrameUnchanged() { @@ -338,6 +390,7 @@ int test_tcp_send_action() { RUN_TEST(test_AbsorbRejectedWhenOtherSendPending); RUN_TEST(test_QueueBatchHostNotFinishedAfterFirstSegment); RUN_TEST(test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess); + RUN_TEST(test_QueueBatchStopChildBeforeSend); RUN_TEST(test_QueueSingleFrameUnchanged); return UNITY_END(); } From 4b240ac6604388a38e1fc4637dd4441315975ac0 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 22:02:56 -0700 Subject: [PATCH 11/15] Fix PacketQueueManager to pop only contiguous finished prefix. Replace index walk with repeated front pop; add queue regression tests for finished/pending/finished ordering. Co-authored-by: Cursor --- aether/transport/packet_queue_manager.h | 11 +- tests/test-transport/test-tcp-send-action.cpp | 103 ++++++++++++++++++ 2 files changed, 107 insertions(+), 7 deletions(-) diff --git a/aether/transport/packet_queue_manager.h b/aether/transport/packet_queue_manager.h index 0921abf5..59293796 100644 --- a/aether/transport/packet_queue_manager.h +++ b/aether/transport/packet_queue_manager.h @@ -59,15 +59,12 @@ class PacketQueueManager { */ void Send() { auto lock = std::scoped_lock{queue_lock_}; - // remove finished - std::size_t i = 0; - for (; i < queue_.size(); i++) { - if (!queue_[i].is_finished()) { - break; - } + std::size_t removed = 0; + while (!queue_.empty() && queue_.front().is_finished()) { queue_.pop(); + ++removed; } - index_ = (index_ <= i) ? 0 : index_ - i; + index_ = (index_ <= removed) ? 0 : index_ - removed; // send queued packets for (; index_ < queue_.size(); index_++) { diff --git a/tests/test-transport/test-tcp-send-action.cpp b/tests/test-transport/test-tcp-send-action.cpp index c25f0bd0..c3e98739 100644 --- a/tests/test-transport/test-tcp-send-action.cpp +++ b/tests/test-transport/test-tcp-send-action.cpp @@ -359,6 +359,106 @@ void test_QueueBatchStopChildBeforeSend() { TEST_ASSERT_TRUE(wa_child->is_finished()); } +void test_QueuePopOnlyFinishedPrefix() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + static constexpr std::size_t kSmallQueue = 3; + PacketQueueManager queue{ctx}; + + auto const h_frame = Frame(31); + queue.AddPacket(ctx, sock, Frame(30)); + queue.Send(); + Pump(tctx); + + int ok_h = 0; + auto* h = queue.AddPacket(ctx, sock, Frame(31)); + h->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++ok_h; + } + }); + + sock.defer_next = true; + queue.Send(); + AssertBuffersEqual(Frame(30), sock.sent); + + int stop_z = 0; + auto* z = queue.AddPacket(ctx, sock, Frame(32)); + z->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kStop) { + ++stop_z; + } + }); + z->Stop(); + Pump(tctx); + + queue.Send(); + h->Send(); + Pump(tctx); + + auto expect = Frame(30); + expect.insert(expect.end(), h_frame.begin(), h_frame.end()); + AssertBuffersEqual(expect, sock.sent); + TEST_ASSERT_EQUAL(1, ok_h); + TEST_ASSERT_EQUAL(1, stop_z); +} + +void test_QueuePopStopsAtFirstUnfinished() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + static constexpr std::size_t kSmallQueue = 2; + PacketQueueManager queue{ctx}; + + sock.defer_next = true; + queue.AddPacket(ctx, sock, Frame(40)); + queue.Send(); + TEST_ASSERT_TRUE(sock.sent.empty()); + + int stop_tail = 0; + auto* tail = queue.AddPacket(ctx, sock, Frame(41)); + tail->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kStop) { + ++stop_tail; + } + }); + tail->Stop(); + TEST_ASSERT_EQUAL(1, stop_tail); + + queue.Send(); + AssertBuffersEqual(Frame(40), sock.sent); + TEST_ASSERT_NULL(queue.AddPacket(ctx, sock, Frame(42))); +} + +void test_QueuePopAllFinishedFreesCapacity() { + TestContext tctx; + AeContext ctx{tctx}; + MockSock sock; + static constexpr std::size_t kSmallQueue = 3; + PacketQueueManager queue{ctx}; + + auto* a = queue.AddPacket(ctx, sock, Frame(50)); + auto* b = queue.AddPacket(ctx, sock, Frame(51)); + auto* c = queue.AddPacket(ctx, sock, Frame(52)); + static_cast(a); + static_cast(b); + c->Stop(); + b->Stop(); + a->Stop(); + Pump(tctx); + + queue.Send(); + Pump(tctx); + queue.Send(); + Pump(tctx); + + TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(53))); + TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(54))); + TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(55))); + TEST_ASSERT_NULL(queue.AddPacket(ctx, sock, Frame(56))); +} + void test_QueueSingleFrameUnchanged() { TestContext tctx; AeContext ctx{tctx}; @@ -391,6 +491,9 @@ int test_tcp_send_action() { RUN_TEST(test_QueueBatchHostNotFinishedAfterFirstSegment); RUN_TEST(test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess); RUN_TEST(test_QueueBatchStopChildBeforeSend); + RUN_TEST(test_QueuePopOnlyFinishedPrefix); + RUN_TEST(test_QueuePopStopsAtFirstUnfinished); + RUN_TEST(test_QueuePopAllFinishedFreesCapacity); RUN_TEST(test_QueueSingleFrameUnchanged); return UNITY_END(); } From 9dbde7aec4437762f53c8df0b7dbdfe39e961ff0 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 22:18:32 -0700 Subject: [PATCH 12/15] Remove experimental TCP SendAction batching from the SDK. Restore single-frame SendAction; keep PacketQueueManager finished-prefix pop and WinTcpSocket 65536 recv buffer. Co-authored-by: Cursor --- aether/transport/system_sockets/tcp/tcp.h | 197 +------ tests/test-transport/CMakeLists.txt | 1 - tests/test-transport/main.cpp | 5 - tests/test-transport/test-tcp-send-action.cpp | 500 ------------------ 4 files changed, 12 insertions(+), 691 deletions(-) delete mode 100644 tests/test-transport/test-tcp-send-action.cpp diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 6298cf63..195426bc 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -27,7 +27,6 @@ # include # include -# include # include "aether/ae_context.h" # include "aether/common.h" @@ -58,57 +57,14 @@ class SendAction final : public PacketSendAction { AE_CLASS_MOVE_ONLY(SendAction) - /** - * \brief Merge another fully formed, not-yet-sent frame into this action. - * \return false if either action already started sending or other is not a - * single-frame action. - */ - bool Absorb(SendAction& other) noexcept { - if (this == &other) { - return false; - } - if (send_started_ || other.send_started_) { - return false; - } - if (sent_offset_ != 0 || other.sent_offset_ != 0) { - return false; - } - if (is_done_ || other.is_done_) { - return false; - } - if (other.batch_host_ != nullptr) { - return false; - } - if (socket_ != other.socket_) { - return false; - } - if (!other.segments_.empty()) { - return false; - } - if (other.data_.empty()) { - return false; - } - - EnsurePrimarySegment(); - data_.insert(data_.end(), other.data_.begin(), other.data_.end()); - other.data_.clear(); - other.batch_host_ = this; - segments_.push_back({data_.size(), &other}); - return true; - } - void Send() override { - if (batch_host_ != nullptr) { - return; - } reenqueue_ = false; auto size_to_send = data_.size() - sent_offset_; - send_started_ = true; auto res = socket_->Send(Span{data_.data() + sent_offset_, size_to_send}); if (!res) { AE_TELED_ERROR("Data has not been written"); - AbortPending(WriteAction::Status::kFail); + SetStatus(WriteAction::Status::kFail); return; } if (*res == 0) { @@ -117,164 +73,35 @@ class SendAction final : public PacketSendAction { AE_TELED_DEBUG("Data has been written size {} data {}", *res, data_); sent_offset_ += *res; - OnSentProgress(); + + if (sent_offset_ >= data_.size()) { + SetStatus(WriteAction::Status::kSuccess); + } // reenque must happen on event from socket } void Stop() noexcept override { - if (batch_host_ != nullptr) { - batch_host_->Stop(); - return; - } - if (segments_.empty()) { - is_done_ = true; - set_status_.Reset(); - WriteAction::SetStatus(WriteAction::Status::kStop); - return; - } - AbortPending(WriteAction::Status::kStop); + is_done_ = true; + set_status_.Reset(); + WriteAction::SetStatus(WriteAction::Status::kStop); } - bool is_done() const override { - if (batch_host_ != nullptr && !is_done_) { - return true; - } - return is_done_; - } + bool is_done() const override { return is_done_; } bool re_enqueue() const override { return reenqueue_; } protected: void SetStatus(WriteAction::Status status) noexcept override { - FinalizeHost(status); - } - - private: - struct Segment { - std::size_t end; - SendAction* action; - }; - - void ClearBatchHostLinks() noexcept { - for (auto const& segment : segments_) { - if (segment.action != this) { - segment.action->batch_host_ = nullptr; - } - } - } - - void EnsurePrimarySegment() { - if (!segments_.empty()) { - return; - } - segments_.push_back({data_.size(), this}); - } - - bool HostHasMultipleSegments() const noexcept { - return segments_.size() > 1; - } - - void OnTerminalTaskDone() noexcept { - if (pending_status_tasks_ == 0) { - return; - } - --pending_status_tasks_; - TryFinishHostAction(); - } - - void ScheduleSegmentTerminal(SendAction& target, - WriteAction::Status status, - bool finish_action) noexcept { - ++pending_status_tasks_; - target.set_status_.Reset(); - target.set_status_ = ae_context_.scheduler().Task( - [this, &target, status, finish_action]() { - if (finish_action) { - target.WriteAction::SetStatus(status); - } else { - target.status_event_.Emit(status); - } - OnTerminalTaskDone(); - }); - } - - void CompleteSegment(SendAction& target, - WriteAction::Status status) noexcept { - if (&target != this) { - target.is_done_ = true; - ScheduleSegmentTerminal(target, status, true); - return; - } - if (HostHasMultipleSegments()) { - ScheduleSegmentTerminal(target, status, false); - return; - } - FinalizeHost(status); - } - - void FinalizeHost(WriteAction::Status status) noexcept { - is_done_ = true; - ++pending_status_tasks_; - set_status_.Reset(); - set_status_ = ae_context_.scheduler().Task([this, status]() { - WriteAction::SetStatus(status); - OnTerminalTaskDone(); - }); - } - - void TryFinishHostAction() noexcept { - if (!HostHasMultipleSegments()) { - return; - } - if (is_finished() || !is_done_ || pending_status_tasks_ != 0) { - return; - } - ClearBatchHostLinks(); - Finish(); - } - - void OnSentProgress() noexcept { - if (segments_.empty()) { - if (sent_offset_ >= data_.size()) { - FinalizeHost(WriteAction::Status::kSuccess); - } - return; - } - - while (completed_segment_idx_ < segments_.size() && - sent_offset_ >= segments_[completed_segment_idx_].end) { - CompleteSegment(*segments_[completed_segment_idx_].action, - WriteAction::Status::kSuccess); - ++completed_segment_idx_; - } - if (sent_offset_ >= data_.size()) { - is_done_ = true; - TryFinishHostAction(); - } - } - - void AbortPending(WriteAction::Status status) noexcept { - if (segments_.empty()) { - FinalizeHost(status); - return; - } is_done_ = true; - for (; completed_segment_idx_ < segments_.size(); - ++completed_segment_idx_) { - CompleteSegment(*segments_[completed_segment_idx_].action, status); - } - TryFinishHostAction(); + set_status_ = ae_context_.scheduler().Task( + [&, status]() { WriteAction::SetStatus(status); }); } + private: AeContext ae_context_; Sock* socket_; DataBuffer data_; - std::vector segments_; - SendAction* batch_host_ = nullptr; std::size_t sent_offset_ = 0; - std::size_t completed_segment_idx_ = 0; - std::size_t pending_status_tasks_ = 0; bool is_done_ = false; - bool send_started_ = false; bool reenqueue_ = false; TaskSubscription set_status_; }; diff --git a/tests/test-transport/CMakeLists.txt b/tests/test-transport/CMakeLists.txt index cf3d1411..565df837 100644 --- a/tests/test-transport/CMakeLists.txt +++ b/tests/test-transport/CMakeLists.txt @@ -17,7 +17,6 @@ cmake_minimum_required( VERSION 3.18 ) list(APPEND test_srcs main.cpp test-data-packet-collector.cpp - test-tcp-send-action.cpp ) if(NOT CM_PLATFORM) diff --git a/tests/test-transport/main.cpp b/tests/test-transport/main.cpp index ff98af4a..2e3443d9 100644 --- a/tests/test-transport/main.cpp +++ b/tests/test-transport/main.cpp @@ -21,13 +21,8 @@ void tearDown() {} extern int test_data_packet_collector(); -namespace ae::test_tcp_send_action { -int test_tcp_send_action(); -} - int main() { int res = 0; res += test_data_packet_collector(); - res += ae::test_tcp_send_action::test_tcp_send_action(); return res; } diff --git a/tests/test-transport/test-tcp-send-action.cpp b/tests/test-transport/test-tcp-send-action.cpp deleted file mode 100644 index c3e98739..00000000 --- a/tests/test-transport/test-tcp-send-action.cpp +++ /dev/null @@ -1,500 +0,0 @@ -/* - * Regression tests for tcp_internal::SendAction batch lifecycle. - */ - -#include - -#include -#include -#include - -#include "aether/ae_context.h" -#include "aether/events/events.h" -#include "aether/tasks/manual_task_scheduler.h" -#include "aether/transport/packet_queue_manager.h" -#include "aether/transport/system_sockets/sockets/isocket.h" -#include "aether/transport/system_sockets/tcp/tcp.h" - -namespace ae::test_tcp_send_action { - -struct TestContext : public Env { - template - decltype(auto) Update(A&&... a) { - return sched.Update(std::forward(a)...); - } - - void* find_component(EnvId id) noexcept override { - if (id == EnvTypeId::value) { - return &sched; - } - if (id == EnvTypeId::value) { - return &event_system_; - } - return nullptr; - } - - TaskScheduler& scheduler() const { return sched; } - EventSystem& event_system() const { return event_system_; } - - mutable TaskScheduler sched; - mutable EventSystem event_system_; -}; - -struct MockSock final : ISocket { - std::vector sent; - std::optional limit_next{}; - bool fail_next = false; - bool defer_next = false; - DataBuffer deferred{}; - - ISocket& ReadyToWrite(ReadyToWriteCb) override { return *this; } - ISocket& RecvData(RecvDataCb) override { return *this; } - ISocket& Error(ErrorCb) override { return *this; } - ISocket& Connect(AddressPort const&, ConnectedCb) override { return *this; } - void Disconnect() override {} - - std::optional Send(Span data) override { - if (fail_next) { - fail_next = false; - return std::nullopt; - } - if (!deferred.empty()) { - sent.insert(sent.end(), deferred.begin(), deferred.end()); - auto n = deferred.size(); - deferred.clear(); - return n; - } - if (defer_next) { - defer_next = false; - deferred.assign(data.begin(), data.end()); - return 0; - } - auto n = data.size(); - if (limit_next) { - n = std::min(n, *limit_next); - limit_next.reset(); - } - sent.insert(sent.end(), data.begin(), - data.begin() + static_cast(n)); - return n; - } -}; - -using SendAction = tcp_internal::SendAction; -static constexpr std::size_t kQueueSize = 8; - -static DataBuffer Frame(std::uint8_t tag) { - return DataBuffer{tag, static_cast(tag + 1), tag}; -} - -static void Pump(TestContext& ctx) { static_cast(ctx.Update()); } - -static void AssertBuffersEqual(DataBuffer const& a, DataBuffer const& b) { - TEST_ASSERT_EQUAL(a.size(), b.size()); - if (a.size() == b.size()) { - TEST_ASSERT_EQUAL_UINT8_ARRAY(a.data(), b.data(), a.size()); - } -} - -void test_AbsorbAllowedBeforeSend() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - PacketQueueManager queue{ctx}; - - auto* host = queue.AddPacket(ctx, sock, Frame(1)); - auto* other = queue.AddPacket(ctx, sock, Frame(2)); - TEST_ASSERT_NOT_NULL(host); - TEST_ASSERT_NOT_NULL(other); - TEST_ASSERT_TRUE(host->Absorb(*other)); - TEST_ASSERT_TRUE(sock.sent.empty()); -} - -void test_AbsorbRejectedWhenHostSendPending() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - sock.defer_next = true; - PacketQueueManager queue{ctx}; - - auto const host_frame = Frame(1); - auto const other_frame = Frame(2); - auto* host = queue.AddPacket(ctx, sock, Frame(1)); - auto* other = queue.AddPacket(ctx, sock, Frame(2)); - - int host_status = 0; - int other_status = 0; - host->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++host_status; - } - }); - other->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++other_status; - } - }); - - host->Send(); - TEST_ASSERT_FALSE(host->Absorb(*other)); - - host->Send(); - AssertBuffersEqual(host_frame, sock.sent); - - queue.Send(); - Pump(tctx); - auto expect = host_frame; - expect.insert(expect.end(), other_frame.begin(), other_frame.end()); - AssertBuffersEqual(expect, sock.sent); - TEST_ASSERT_EQUAL(1, host_status); - TEST_ASSERT_EQUAL(1, other_status); -} - -void test_AbsorbRejectedWhenOtherSendPending() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - sock.defer_next = true; - PacketQueueManager queue{ctx}; - - auto const host_frame = Frame(10); - auto const other_frame = Frame(11); - auto* host = queue.AddPacket(ctx, sock, Frame(10)); - auto* other = queue.AddPacket(ctx, sock, Frame(11)); - - int host_status = 0; - int other_status = 0; - host->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++host_status; - } - }); - other->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++other_status; - } - }); - - other->Send(); - TEST_ASSERT_FALSE(host->Absorb(*other)); - - other->Send(); - AssertBuffersEqual(other_frame, sock.sent); - - queue.Send(); - Pump(tctx); - auto expect = other_frame; - expect.insert(expect.end(), host_frame.begin(), host_frame.end()); - AssertBuffersEqual(expect, sock.sent); - TEST_ASSERT_EQUAL(1, host_status); - TEST_ASSERT_EQUAL(1, other_status); -} - -void test_QueueBatchHostNotFinishedAfterFirstSegment() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - sock.limit_next = Frame(1).size(); - PacketQueueManager queue{ctx}; - - auto* host = queue.AddPacket(ctx, sock, Frame(1)); - auto* extra_a = queue.AddPacket(ctx, sock, Frame(2)); - auto* extra_b = queue.AddPacket(ctx, sock, Frame(3)); - WriteAction* wa_a = extra_a; - WriteAction* wa_b = extra_b; - - int ok_host = 0; - int ok_a = 0; - int ok_a_late = 0; - int ok_b = 0; - int host_finished = 0; - auto sub_host = host->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_host; - } - }); - auto sub_a = wa_a->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_a; - } - }); - auto sub_b = wa_b->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_b; - } - }); - auto sub_fin = host->finished_event().Subscribe([&]() { ++host_finished; }); - - TEST_ASSERT_TRUE(host->Absorb(*extra_a)); - TEST_ASSERT_TRUE(host->Absorb(*extra_b)); - - wa_a->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_a_late; - } - }); - - Pump(tctx); - Pump(tctx); - - TEST_ASSERT_EQUAL(1, ok_host); - TEST_ASSERT_EQUAL(0, ok_a); - TEST_ASSERT_EQUAL(0, ok_b); - TEST_ASSERT_EQUAL(0, host_finished); - TEST_ASSERT_FALSE(host->is_finished()); - TEST_ASSERT_FALSE(wa_a->is_finished()); - TEST_ASSERT_EQUAL(Frame(1).size(), sock.sent.size()); - - queue.Send(); - TEST_ASSERT_FALSE(host->is_finished()); - Pump(tctx); - Pump(tctx); - Pump(tctx); - - TEST_ASSERT_EQUAL(9U, sock.sent.size()); - TEST_ASSERT_EQUAL(1, ok_host); - TEST_ASSERT_EQUAL(1, ok_a); - TEST_ASSERT_EQUAL(1, ok_a_late); - TEST_ASSERT_EQUAL(1, ok_b); - TEST_ASSERT_EQUAL(1, host_finished); - TEST_ASSERT_TRUE(host->is_finished()); - TEST_ASSERT_TRUE(wa_a->is_finished()); - TEST_ASSERT_TRUE(wa_b->is_finished()); - - auto* next = queue.AddPacket(ctx, sock, Frame(4)); - TEST_ASSERT_NOT_NULL(next); - queue.Send(); - Pump(tctx); - TEST_ASSERT_TRUE(next->is_finished()); -} - -void test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - sock.limit_next = Frame(10).size(); - PacketQueueManager queue{ctx}; - - auto* host = queue.AddPacket(ctx, sock, Frame(10)); - auto* extra_a = queue.AddPacket(ctx, sock, Frame(11)); - auto* extra_b = queue.AddPacket(ctx, sock, Frame(12)); - WriteAction* wa_a = extra_a; - WriteAction* wa_b = extra_b; - - int ok_host = 0; - int fail_a = 0; - int fail_b = 0; - int host_finished = 0; - host->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_host; - } - }); - wa_a->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kFail) { - ++fail_a; - } - }); - wa_b->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kFail) { - ++fail_b; - } - }); - host->finished_event().Subscribe([&]() { ++host_finished; }); - - TEST_ASSERT_TRUE(host->Absorb(*extra_a)); - TEST_ASSERT_TRUE(host->Absorb(*extra_b)); - - Pump(tctx); - Pump(tctx); - TEST_ASSERT_EQUAL(1, ok_host); - TEST_ASSERT_FALSE(host->is_finished()); - - sock.fail_next = true; - queue.Send(); - Pump(tctx); - Pump(tctx); - Pump(tctx); - - TEST_ASSERT_EQUAL(1, ok_host); - TEST_ASSERT_EQUAL(1, fail_a); - TEST_ASSERT_EQUAL(1, fail_b); - TEST_ASSERT_EQUAL(1, host_finished); - TEST_ASSERT_TRUE(host->is_finished()); - TEST_ASSERT_TRUE(wa_a->is_finished()); - TEST_ASSERT_TRUE(wa_b->is_finished()); -} - -void test_QueueBatchStopChildBeforeSend() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - PacketQueueManager queue{ctx}; - - auto* host = queue.AddPacket(ctx, sock, Frame(20)); - auto* child = queue.AddPacket(ctx, sock, Frame(21)); - WriteAction* wa_child = child; - - int stop_host = 0; - int stop_child = 0; - host->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kStop) { - ++stop_host; - } - }); - wa_child->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kStop) { - ++stop_child; - } - }); - - TEST_ASSERT_TRUE(host->Absorb(*child)); - wa_child->Stop(); - Pump(tctx); - - TEST_ASSERT_TRUE(sock.sent.empty()); - TEST_ASSERT_EQUAL(1, stop_host); - TEST_ASSERT_EQUAL(1, stop_child); - TEST_ASSERT_TRUE(host->is_finished()); - TEST_ASSERT_TRUE(wa_child->is_finished()); -} - -void test_QueuePopOnlyFinishedPrefix() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - static constexpr std::size_t kSmallQueue = 3; - PacketQueueManager queue{ctx}; - - auto const h_frame = Frame(31); - queue.AddPacket(ctx, sock, Frame(30)); - queue.Send(); - Pump(tctx); - - int ok_h = 0; - auto* h = queue.AddPacket(ctx, sock, Frame(31)); - h->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok_h; - } - }); - - sock.defer_next = true; - queue.Send(); - AssertBuffersEqual(Frame(30), sock.sent); - - int stop_z = 0; - auto* z = queue.AddPacket(ctx, sock, Frame(32)); - z->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kStop) { - ++stop_z; - } - }); - z->Stop(); - Pump(tctx); - - queue.Send(); - h->Send(); - Pump(tctx); - - auto expect = Frame(30); - expect.insert(expect.end(), h_frame.begin(), h_frame.end()); - AssertBuffersEqual(expect, sock.sent); - TEST_ASSERT_EQUAL(1, ok_h); - TEST_ASSERT_EQUAL(1, stop_z); -} - -void test_QueuePopStopsAtFirstUnfinished() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - static constexpr std::size_t kSmallQueue = 2; - PacketQueueManager queue{ctx}; - - sock.defer_next = true; - queue.AddPacket(ctx, sock, Frame(40)); - queue.Send(); - TEST_ASSERT_TRUE(sock.sent.empty()); - - int stop_tail = 0; - auto* tail = queue.AddPacket(ctx, sock, Frame(41)); - tail->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kStop) { - ++stop_tail; - } - }); - tail->Stop(); - TEST_ASSERT_EQUAL(1, stop_tail); - - queue.Send(); - AssertBuffersEqual(Frame(40), sock.sent); - TEST_ASSERT_NULL(queue.AddPacket(ctx, sock, Frame(42))); -} - -void test_QueuePopAllFinishedFreesCapacity() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - static constexpr std::size_t kSmallQueue = 3; - PacketQueueManager queue{ctx}; - - auto* a = queue.AddPacket(ctx, sock, Frame(50)); - auto* b = queue.AddPacket(ctx, sock, Frame(51)); - auto* c = queue.AddPacket(ctx, sock, Frame(52)); - static_cast(a); - static_cast(b); - c->Stop(); - b->Stop(); - a->Stop(); - Pump(tctx); - - queue.Send(); - Pump(tctx); - queue.Send(); - Pump(tctx); - - TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(53))); - TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(54))); - TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx, sock, Frame(55))); - TEST_ASSERT_NULL(queue.AddPacket(ctx, sock, Frame(56))); -} - -void test_QueueSingleFrameUnchanged() { - TestContext tctx; - AeContext ctx{tctx}; - MockSock sock; - PacketQueueManager queue{ctx}; - - auto* action = queue.AddPacket(ctx, sock, Frame(7)); - int ok = 0; - int finished = 0; - action->status_event().Subscribe([&](WriteAction::Status s) { - if (s == WriteAction::Status::kSuccess) { - ++ok; - } - }); - action->finished_event().Subscribe([&]() { ++finished; }); - - queue.Send(); - Pump(tctx); - - TEST_ASSERT_EQUAL(1, ok); - TEST_ASSERT_EQUAL(1, finished); - TEST_ASSERT_TRUE(action->is_finished()); -} - -int test_tcp_send_action() { - UNITY_BEGIN(); - RUN_TEST(test_AbsorbAllowedBeforeSend); - RUN_TEST(test_AbsorbRejectedWhenHostSendPending); - RUN_TEST(test_AbsorbRejectedWhenOtherSendPending); - RUN_TEST(test_QueueBatchHostNotFinishedAfterFirstSegment); - RUN_TEST(test_QueueBatchFailAfterFirstSegmentKeepsFirstSuccess); - RUN_TEST(test_QueueBatchStopChildBeforeSend); - RUN_TEST(test_QueuePopOnlyFinishedPrefix); - RUN_TEST(test_QueuePopStopsAtFirstUnfinished); - RUN_TEST(test_QueuePopAllFinishedFreesCapacity); - RUN_TEST(test_QueueSingleFrameUnchanged); - return UNITY_END(); -} -} // namespace ae::test_tcp_send_action From c731ba2131b85adf30eab5f2e86ec67b5eb3245a Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 22 Oct 2026 22:54:51 -0700 Subject: [PATCH 13/15] Allow compile-time override of WinTcpSocket WSARecv buffer size. Co-authored-by: Cursor --- aether/transport/system_sockets/sockets/win_tcp_socket.cpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp index a7a72ce2..e239868b 100644 --- a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp @@ -30,6 +30,10 @@ namespace ae { namespace win_socket_internal { + +#ifndef AE_WIN_TCP_RECV_BUFFER_BYTES +# define AE_WIN_TCP_RECV_BUFFER_BYTES 65536 +#endif LPFN_CONNECTEX GetConnectEx(SOCKET socket) { LPFN_CONNECTEX func_ptr = nullptr; GUID guid = WSAID_CONNECTEX; @@ -47,7 +51,7 @@ LPFN_CONNECTEX GetConnectEx(SOCKET socket) { } // namespace win_socket_internal WinTcpSocket::WinTcpSocket(Ptr const& poller) - : WinSocket{*poller, 65536}, conn_overlapped_{} { + : WinSocket{*poller, AE_WIN_TCP_RECV_BUFFER_BYTES}, conn_overlapped_{} { bool created = false; // ::socket() sets WSA_FLAG_OVERLAPPED by default that allows us to use iocp From af7058113b8725f5d807f466e098d1650f652805 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Fri, 23 Oct 2026 12:08:29 -0700 Subject: [PATCH 14/15] Allow AE_UNIX_TCP_RECV_BUFFER_BYTES override for Linux TCP recv buffer sizing. Co-authored-by: Cursor --- aether/transport/system_sockets/sockets/unix_tcp_socket.cpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp b/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp index 1a4a1526..d082bbd6 100644 --- a/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp +++ b/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp @@ -75,8 +75,12 @@ inline bool SetNonblocking(int sock) { UnixTcpSocket::UnixTcpSocket(Ptr const& poller) : UnixSocket(*poller, MakeSocket()), connection_state_{ConnectionState::kNone} { - // 1500 is our default MTU for TCP + // 1500 is our default MTU for TCP; benchmark may override via compile define. +# if defined(AE_UNIX_TCP_RECV_BUFFER_BYTES) + recv_buffer_.resize(AE_UNIX_TCP_RECV_BUFFER_BYTES); +# else recv_buffer_.resize(1500); +# endif } int UnixTcpSocket::MakeSocket() { From 66fdf7041be94e40c52fb00d2dd444bbc54ab645 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Fri, 23 Oct 2026 12:46:55 -0700 Subject: [PATCH 15/15] Finalize MSVC TCP throughput fixes; remove recv-buffer compile overrides and add queue regressions. Co-authored-by: Cursor --- .../sockets/unix_tcp_socket.cpp | 6 +- .../system_sockets/sockets/win_tcp_socket.cpp | 6 +- tests/test-transport/CMakeLists.txt | 1 + tests/test-transport/main.cpp | 4 +- .../test-packet-queue-manager.cpp | 168 ++++++++++++++++++ 5 files changed, 174 insertions(+), 11 deletions(-) create mode 100644 tests/test-transport/test-packet-queue-manager.cpp diff --git a/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp b/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp index d082bbd6..1a4a1526 100644 --- a/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp +++ b/aether/transport/system_sockets/sockets/unix_tcp_socket.cpp @@ -75,12 +75,8 @@ inline bool SetNonblocking(int sock) { UnixTcpSocket::UnixTcpSocket(Ptr const& poller) : UnixSocket(*poller, MakeSocket()), connection_state_{ConnectionState::kNone} { - // 1500 is our default MTU for TCP; benchmark may override via compile define. -# if defined(AE_UNIX_TCP_RECV_BUFFER_BYTES) - recv_buffer_.resize(AE_UNIX_TCP_RECV_BUFFER_BYTES); -# else + // 1500 is our default MTU for TCP recv_buffer_.resize(1500); -# endif } int UnixTcpSocket::MakeSocket() { diff --git a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp index e239868b..a7a72ce2 100644 --- a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp @@ -30,10 +30,6 @@ namespace ae { namespace win_socket_internal { - -#ifndef AE_WIN_TCP_RECV_BUFFER_BYTES -# define AE_WIN_TCP_RECV_BUFFER_BYTES 65536 -#endif LPFN_CONNECTEX GetConnectEx(SOCKET socket) { LPFN_CONNECTEX func_ptr = nullptr; GUID guid = WSAID_CONNECTEX; @@ -51,7 +47,7 @@ LPFN_CONNECTEX GetConnectEx(SOCKET socket) { } // namespace win_socket_internal WinTcpSocket::WinTcpSocket(Ptr const& poller) - : WinSocket{*poller, AE_WIN_TCP_RECV_BUFFER_BYTES}, conn_overlapped_{} { + : WinSocket{*poller, 65536}, conn_overlapped_{} { bool created = false; // ::socket() sets WSA_FLAG_OVERLAPPED by default that allows us to use iocp diff --git a/tests/test-transport/CMakeLists.txt b/tests/test-transport/CMakeLists.txt index 565df837..2b078aa1 100644 --- a/tests/test-transport/CMakeLists.txt +++ b/tests/test-transport/CMakeLists.txt @@ -17,6 +17,7 @@ cmake_minimum_required( VERSION 3.18 ) list(APPEND test_srcs main.cpp test-data-packet-collector.cpp + test-packet-queue-manager.cpp ) if(NOT CM_PLATFORM) diff --git a/tests/test-transport/main.cpp b/tests/test-transport/main.cpp index 2e3443d9..7ca1cc0e 100644 --- a/tests/test-transport/main.cpp +++ b/tests/test-transport/main.cpp @@ -20,9 +20,11 @@ void setUp() {} void tearDown() {} extern int test_data_packet_collector(); +extern int test_packet_queue_manager(); int main() { int res = 0; res += test_data_packet_collector(); - return res; + res += test_packet_queue_manager(); + return 0; } diff --git a/tests/test-transport/test-packet-queue-manager.cpp b/tests/test-transport/test-packet-queue-manager.cpp new file mode 100644 index 00000000..db10faa5 --- /dev/null +++ b/tests/test-transport/test-packet-queue-manager.cpp @@ -0,0 +1,168 @@ +/* + * Regression tests for PacketQueueManager finished-prefix removal. + */ + +#include + +#include "aether/ae_context.h" +#include "aether/events/events.h" +#include "aether/tasks/manual_task_scheduler.h" +#include "aether/transport/packet_queue_manager.h" +#include "aether/transport/packet_send_action.h" + +namespace ae::test_packet_queue_manager { + +struct TestContext : public Env { + template + decltype(auto) Update(A&&... a) { + return sched.Update(std::forward(a)...); + } + + void* find_component(EnvId id) noexcept override { + if (id == EnvTypeId::value) { + return &sched; + } + if (id == EnvTypeId::value) { + return &event_system_; + } + return nullptr; + } + + TaskScheduler& scheduler() const { return sched; } + EventSystem& event_system() const { return event_system_; } + + mutable TaskScheduler sched; + mutable EventSystem event_system_; +}; + +class MockSendAction final : public PacketSendAction { + public: + explicit MockSendAction(AeContext const& ctx) : PacketSendAction(ctx) {} + + void Send() override { + if (finish_on_send_) { + SetStatus(Status::kSuccess); + } + } + + bool is_done() const override { return is_finished(); } + + bool re_enqueue() const override { return false; } + + void ForceFinish() { Finish(); } + + void Succeed() { SetStatus(Status::kSuccess); } + + bool finish_on_send_ = false; +}; + +static constexpr std::size_t kQueueSize = 4; + +static void Pump(TestContext& ctx) { static_cast(ctx.Update()); } + +void test_FinishedPendingFinishedPrefixPop() { + TestContext tctx; + AeContext ctx{tctx}; + PacketQueueManager queue{ctx}; + + auto* first = queue.AddPacket(ctx); + auto* pending = queue.AddPacket(ctx); + auto* third = queue.AddPacket(ctx); + TEST_ASSERT_NOT_NULL(first); + TEST_ASSERT_NOT_NULL(pending); + TEST_ASSERT_NOT_NULL(third); + + int pending_done = 0; + auto sub = pending->status_event().Subscribe([&](WriteAction::Status s) { + if (s == WriteAction::Status::kSuccess) { + ++pending_done; + } + }); + + first->ForceFinish(); + third->ForceFinish(); + + queue.Send(); + Pump(tctx); + + TEST_ASSERT_FALSE(pending->is_finished()); + TEST_ASSERT_EQUAL(0, pending_done); + + pending->Succeed(); + queue.Send(); + Pump(tctx); + + TEST_ASSERT_EQUAL(1, pending_done); + TEST_ASSERT_TRUE(pending->is_finished()); +} + +void test_PendingFrontBlocksFinishedTail() { + TestContext tctx; + AeContext ctx{tctx}; + PacketQueueManager queue{ctx}; + + auto* pending = queue.AddPacket(ctx); + auto* tail_a = queue.AddPacket(ctx); + auto* tail_b = queue.AddPacket(ctx); + TEST_ASSERT_NOT_NULL(pending); + TEST_ASSERT_NOT_NULL(tail_a); + TEST_ASSERT_NOT_NULL(tail_b); + + tail_a->ForceFinish(); + tail_b->ForceFinish(); + + TEST_ASSERT_NULL(queue.AddPacket(ctx)); + + queue.Send(); + Pump(tctx); + TEST_ASSERT_FALSE(pending->is_finished()); + TEST_ASSERT_NULL(queue.AddPacket(ctx)); + + pending->finish_on_send_ = true; + queue.Send(); + Pump(tctx); + TEST_ASSERT_TRUE(pending->is_finished()); + + queue.Send(); + Pump(tctx); + + auto* room = queue.AddPacket(ctx); + TEST_ASSERT_NOT_NULL(room); +} + +void test_AllFinishedFreesCapacity() { + TestContext tctx; + AeContext ctx{tctx}; + PacketQueueManager queue{ctx}; + + for (int i = 0; i < static_cast(kQueueSize); ++i) { + auto* pkt = queue.AddPacket(ctx); + TEST_ASSERT_NOT_NULL(pkt); + pkt->ForceFinish(); + } + TEST_ASSERT_NULL(queue.AddPacket(ctx)); + + queue.Send(); + Pump(tctx); + + for (int i = 0; i < static_cast(kQueueSize); ++i) { + auto* pkt = queue.AddPacket(ctx); + TEST_ASSERT_NOT_NULL(pkt); + pkt->ForceFinish(); + } + TEST_ASSERT_NULL(queue.AddPacket(ctx)); + + queue.Send(); + Pump(tctx); + + TEST_ASSERT_NOT_NULL(queue.AddPacket(ctx)); +} + +} // namespace ae::test_packet_queue_manager + +int test_packet_queue_manager() { + RUN_TEST(ae::test_packet_queue_manager::test_FinishedPendingFinishedPrefixPop); + RUN_TEST(ae::test_packet_queue_manager::test_PendingFrontBlocksFinishedTail); + RUN_TEST(ae::test_packet_queue_manager::test_AllFinishedFreesCapacity); + return 0; +}