MSVC TCP throughput fixes for P2pStream (IOCP recv/send pending, queue prefix, 65536 B Win recv)

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>{*api_}
-                           : ApiContext<Api>{*api_, *protocol_context_};
-    auto promise = std::invoke(fn_, api_context);
-    if (!promise) {
-      ex::set_stopped(std::move(receiver_));
-      return;
-    }
+    auto run = [this](ApiContext<Api>& 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<decltype(result)>(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<decltype(result)>(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> api_context{*api_};
+      run(api_context);
+    } else {
+      ApiContext<Api> api_context{*api_, *protocol_context_};
+      run(api_context);
+    }
   }
 
  private:

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_++) {

aether/transport/system_sockets/sockets/win_socket.cpp

@@ -73,6 +73,8 @@ std::optional<std::size_t> WinSocket::Send(Span<std::uint8_t> 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;
 }
 

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<IPoller> 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

aether/transport/system_sockets/tcp/tcp.h

@@ -68,7 +68,6 @@ class SendAction final : public PacketSendAction {
       return;
     }
     if (*res == 0) {
-      reenqueue_ = true;
       return;
     }
 

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)

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;
 }

tests/test-transport/test-packet-queue-manager.cpp

@@ -0,0 +1,168 @@
+/*
+ * Regression tests for PacketQueueManager finished-prefix removal.
+ */
+
+#include <unity.h>
+
+#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 <typename... A>
+  decltype(auto) Update(A&&... a) {
+    return sched.Update(std::forward<A>(a)...);
+  }
+
+  void* find_component(EnvId id) noexcept override {
+    if (id == EnvTypeId<TaskScheduler>::value) {
+      return &sched;
+    }
+    if (id == EnvTypeId<EventSystem>::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<void>(ctx.Update()); }
+
+void test_FinishedPendingFinishedPrefixPop() {
+  TestContext tctx;
+  AeContext ctx{tctx};
+  PacketQueueManager<MockSendAction, kQueueSize> 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<MockSendAction, 3> 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<MockSendAction, kQueueSize> queue{ctx};
+
+  for (int i = 0; i < static_cast<int>(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<int>(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;
+}