diff --git a/aether/api_protocol/make_api_call_sender.h b/aether/api_protocol/make_api_call_sender.h index b7058a142..94e5ab1a7 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: diff --git a/aether/transport/packet_queue_manager.h b/aether/transport/packet_queue_manager.h index 0921abf54..59293796b 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/aether/transport/system_sockets/sockets/win_socket.cpp b/aether/transport/system_sockets/sockets/win_socket.cpp index fcc6988aa..dd1f69ae0 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); @@ -168,6 +170,11 @@ 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). 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; } diff --git a/aether/transport/system_sockets/sockets/win_tcp_socket.cpp b/aether/transport/system_sockets/sockets/win_tcp_socket.cpp index f63aa7931..a7a72ce26 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 diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index 546847e97..195426bc2 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; } diff --git a/tests/test-transport/CMakeLists.txt b/tests/test-transport/CMakeLists.txt index 565df8370..2b078aa11 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 2e3443d96..7ca1cc0ec 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 000000000..db10faa5c --- /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; +}