Fire when_available callbacks via async

QUIC typically acks multiple packets at once, which was resulting in
multiple immediately sequential calls to the unblock callbacks; this
commit schedules it via an async on the event loop instead to batch
them.

Author: jagerman

connection.cpp

@@ -308,9 +308,8 @@ io_result Connection::send() {
 }
 
 
-std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks> Connection::init(Endpoint& ep) {
-    Debug("loop: ", ep.loop);
-    io_trigger = ep.loop->resource<uvw::AsyncHandle>();
+std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks> Connection::init() {
+    io_trigger = endpoint.loop->resource<uvw::AsyncHandle>();
     io_trigger->on<uvw::AsyncEvent>([this] (auto&, auto&) { on_io_ready(); });
 
     auto result = std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks>{};
@@ -362,7 +361,7 @@ std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks> Connectio
 Connection::Connection(Server& s, const ConnectionID& base_cid_, ngtcp2_pkt_hd& header, const Path& path)
         : endpoint{s}, base_cid{base_cid_}, dest_cid{header.scid}, path{path} {
 
-    auto [settings, tparams, cb] = init(s);
+    auto [settings, tparams, cb] = init();
 
     cb.recv_client_initial = recv_client_initial;
 
@@ -417,7 +416,7 @@ Connection::Connection(Server& s, const ConnectionID& base_cid_, ngtcp2_pkt_hd&
 Connection::Connection(Client& c, const ConnectionID& scid, const Path& path, uint16_t tunnel_port)
         : endpoint{c}, base_cid{scid}, dest_cid{ConnectionID::random(c.rng)}, path{path}, tunnel_port{tunnel_port} {
 
-    auto [settings, tparams, cb] = init(c);
+    auto [settings, tparams, cb] = init();
 
     assert(tunnel_port != 0);
 

connection.h

@@ -103,7 +103,7 @@ class Connection : public std::enable_shared_from_this<Connection> {
 
     // Internal base method called invoked during construction to set up common client/server
     // settings.  dest_cid and path must already be set.
-    std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks> init(Endpoint& ep);
+    std::tuple<ngtcp2_settings, ngtcp2_transport_params, ngtcp2_callbacks> init();
 
     // Event trigger used to queue packet processing for this connection
     std::shared_ptr<uvw::AsyncHandle> io_trigger;

endpoint.h

@@ -205,6 +205,8 @@ class Endpoint {
 
     // Default stream buffer size for streams opened through this endpoint.
     size_t default_stream_buffer_size = 64*1024;
+
+    uvw::Loop& get_loop() { return *loop; }
 };
 
 }

stream.cpp

@@ -1,6 +1,8 @@
 #include "stream.h"
 #include "connection.h"
+#include "endpoint.h"
 #include "log.h"
+#include "uvw/async.h"
 
 #include <cassert>
 #include <iostream>
@@ -46,13 +48,19 @@ std::ostream& operator<<(std::ostream& o, const StreamID& s) {
     return o << u8"Str❰" << s.id << u8"❱";
 }
 
-Stream::Stream(Connection& conn, data_callback_t data_cb, close_callback_t close_cb, size_t buffer_size)
-    : conn{conn}, data_callback{std::move(data_cb)}, close_callback{std::move(close_cb)}, buffer{buffer_size}
+Stream::Stream(Connection& conn, data_callback_t data_cb, close_callback_t close_cb, size_t buffer_size, StreamID id) :
+    conn{conn},
+    stream_id{std::move(id)},
+    data_callback{std::move(data_cb)},
+    close_callback{std::move(close_cb)},
+    buffer{buffer_size},
+    avail_trigger{conn.endpoint.get_loop().resource<uvw::AsyncHandle>()}
 {
+    avail_trigger->on<uvw::AsyncEvent>([this] (auto&, auto&) { handle_unblocked(); });
 }
 
 Stream::Stream(Connection& conn, StreamID id, size_t buffer_size)
-    : conn{conn}, stream_id{id}, buffer{buffer_size}
+    : Stream{conn, nullptr, nullptr, buffer_size, std::move(id)}
 {
 }
 
@@ -122,7 +130,7 @@ void Stream::acknowledge(size_t bytes) {
     size -= bytes;
     start = size == 0 ? 0 : (start + bytes) % buffer.size(); // reset start to 0 (to reduce wrapping buffers) if empty
     if (!unblocked_callbacks.empty())
-        handle_unblocked();
+        available_ready();
 }
 
 std::pair<bstring_view, bstring_view> Stream::pending() {
@@ -140,26 +148,24 @@ std::pair<bstring_view, bstring_view> Stream::pending() {
 }
 
 void Stream::when_available(unblocked_callback_t unblocked_cb) {
+    assert(available() == 0);
     unblocked_callbacks.push(std::move(unblocked_cb));
-    handle_unblocked();
 }
 
 void Stream::handle_unblocked() {
     while (!unblocked_callbacks.empty() && available() > 0) {
-#ifndef NDEBUG
-        size_t pre_avail = available();
-#endif
-        bool done = unblocked_callbacks.front()(*this);
-        if (done)
+        if (unblocked_callbacks.front()(*this))
             unblocked_callbacks.pop();
         else
-            assert(available() < pre_avail);
+            assert(available() == 0);
     }
     conn.io_ready();
 }
 
 void Stream::io_ready() { conn.io_ready(); }
 
+void Stream::available_ready() { avail_trigger->send(); }
+
 void Stream::wrote(size_t bytes) {
     // Called to tell us we sent some bytes off, e.g. wrote(3) changes:
     //     [  áaarrrrrr  ]  or  [rr     áaar]

stream.h

@@ -10,6 +10,8 @@
 #include <variant>
 #include <vector>
 
+#include <uvw/async.h>
+
 namespace quic {
 
 class Connection;
@@ -136,11 +138,18 @@ class Stream : public std::enable_shared_from_this<Stream> {
     // callbacks are queued they are invoked in order, space permitting.  The stored std::function
     // will not be moved or copied after being invoked (i.e. if invoked multiple times it will
     // always be invoked on the same instance).
+    //
+    // Available callbacks should only be used when the buffer is full, typically immediately after
+    // an `append_any` call that returns less than the full write.  Similarly a false return from an
+    // unblock function (which keeps the callback alive) should satisfy the same condition.
     void when_available(unblocked_callback_t unblocked_cb);
 
-    // Calls io_ready() on the stream's connection to trigger sending data
+    // Calls io_ready() on the stream's connection to scheduling sending outbound data
     void io_ready();
 
+    // Schedules processing of the "when_available" callbacks
+    void available_ready();
+
     // Lets you stash some arbitrary data in a shared_ptr; this is not used internally.
     void data(std::shared_ptr<void> data);
 
@@ -160,7 +169,7 @@ class Stream : public std::enable_shared_from_this<Stream> {
 private:
     friend class Connection;
 
-    Stream(Connection& conn, data_callback_t data_cb, close_callback_t close_cb, size_t buffer_size);
+    Stream(Connection& conn, data_callback_t data_cb, close_callback_t close_cb, size_t buffer_size, StreamID id = {-1});
     Stream(Connection& conn, StreamID id, size_t buffer_size);
 
     // Non-copyable, non-movable; we manage it via a unique_ptr held by its Connection
@@ -216,6 +225,10 @@ class Stream : public std::enable_shared_from_this<Stream> {
     bool sent_fin{false};
     bool is_shutdown{false};
 
+    // Async trigger we use to schedule when_available callbacks (so that we can make them happen in
+    // batches rather than after each and every packet ack).
+    std::shared_ptr<uvw::AsyncHandle> avail_trigger;
+
     std::variant<std::shared_ptr<void>, std::weak_ptr<void>> user_data;
 };