net: tcp: change unsent queue from packets to temporary_buffer:s

The unsent queue is currently a queue of packets. This is awkward
because the packets queued don't correspond to packets to be sent;
they have to be cut and pasted to fit the send window or maximum
packet size, whichever is greater.

The recent change to data_sink to work in terms of temporary_buffer:s
rather than packets provides an additional incentive to fix this.

The fix is simple: change the queue type to hold temporary buffers.
The compatibility code to construct a packet out of the data_sink
temporary_buffer:s is no longer needed.

The code to reshape the unsent packets to a packet to be sent is
simplified: we append buffers to the packet until a complete
buffer will overflow the maximum send size, then split the last
buffer (if available and needed) to complete the packet.

Tested with httpd on a tap device, seems to work.

Author: avikivity

demos/tcp_demo.cc

@@ -45,7 +45,8 @@ struct tcp_test {
                     return;
                 }
                 fmt::print("read {:d} bytes\n", p.len());
-                (void)tcp_conn.send(std::move(p));
+                auto v = std::move(p).release();
+                (void)tcp_conn.send(std::span(v));
                 run();
             });
         }

include/seastar/net/tcp.hh

@@ -28,6 +28,7 @@
 #include <deque>
 #include <chrono>
 #include <random>
+#include <span>
 #include <stdexcept>
 #include <system_error>
 #include <gnutls/crypto.h>
@@ -339,7 +340,7 @@ private:
             tcp_seq wl2;
             tcp_seq initial;
             std::deque<unacked_segment> data;
-            std::deque<packet> unsent;
+            std::deque<temporary_buffer<char>> unsent;
             uint32_t unsent_len = 0;
             bool closed = false;
             promise<> _window_opened;
@@ -436,7 +437,7 @@ private:
         void abort_reader() noexcept;
         future<> wait_for_all_data_acked();
         future<> wait_send_available();
-        future<> send(packet p);
+        future<> send(std::span<temporary_buffer<char>> data);
         void connect();
         packet read();
         void close() noexcept;
@@ -689,8 +690,8 @@ public:
         future<> connected() {
             return _tcb->connect_done();
         }
-        future<> send(packet p) {
-            return _tcb->send(std::move(p));
+        future<> send(std::span<temporary_buffer<char>> data) {
+            return _tcb->send(data);
         }
         future<> wait_for_data() {
             return _tcb->wait_for_data();
@@ -1590,27 +1591,9 @@ packet tcp<InetTraits>::tcb::get_transmit_packet() {
         len = std::min(uint16_t(_tcp.hw_features().mtu - net::tcp_hdr_len_min - InetTraits::ip_hdr_len_min), _snd.mss);
     }
     can_send = std::min(can_send, len);
-    // easy case: one small packet
-    if (_snd.unsent.size() == 1 && _snd.unsent.front().len() <= can_send) {
-        auto p = std::move(_snd.unsent.front());
-        _snd.unsent.pop_front();
-        _snd.unsent_len -= p.len();
-        return p;
-    }
-    // moderate case: need to split one packet
-    if (_snd.unsent.front().len() > can_send) {
-        auto p = _snd.unsent.front().share(0, can_send);
-        _snd.unsent.front().trim_front(can_send);
-        _snd.unsent_len -= p.len();
-        return p;
-    }
-    // hard case: merge some packets, possibly split last
-    auto p = std::move(_snd.unsent.front());
-    _snd.unsent.pop_front();
-    can_send -= p.len();
-    while (!_snd.unsent.empty()
-            && _snd.unsent.front().len() <= can_send) {
-        can_send -= _snd.unsent.front().len();
+    auto p = packet();
+    while (!_snd.unsent.empty() && _snd.unsent.front().size() <= can_send) {
+        can_send -= _snd.unsent.front().size();
         p.append(std::move(_snd.unsent.front()));
         _snd.unsent.pop_front();
     }
@@ -1813,16 +1796,19 @@ future<> tcp<InetTraits>::tcb::wait_send_available() {
 }
 
 template <typename InetTraits>
-future<> tcp<InetTraits>::tcb::send(packet p) {
+future<> tcp<InetTraits>::tcb::send(std::span<temporary_buffer<char>> data) {
     // We can not send after the connection is closed
     if (_snd.closed || in_state(CLOSED)) {
         return make_exception_future<>(tcp_reset_error());
     }
 
-    auto len = p.len();
+    auto sizes = data | std::views::transform(&temporary_buffer<char>::size);
+    auto len = std::accumulate(sizes.begin(), sizes.end(), size_t(0));
+
     _snd.current_queue_space += len;
     _snd.unsent_len += len;
-    _snd.unsent.push_back(std::move(p));
+    // FIXME: use append_range with C++23
+    std::ranges::move(data, std::back_inserter(_snd.unsent));
 
     if (can_send() > 0) {
         output();

src/net/native-stack-impl.hh

@@ -207,17 +207,19 @@ public:
         : _conn(std::move(conn)) {}
 #if SEASTAR_API_LEVEL >= 9
     future<> put(std::span<temporary_buffer<char>> bufs) override {
-        net::packet p = net::packet(bufs);
         auto sg_id = internal::scheduling_group_index(current_scheduling_group());
-        internal::native_stack_net_stats::bytes_sent[sg_id] += p.len();
-        return _conn->send(std::move(p));
+        auto sizes = bufs | std::views::transform(&temporary_buffer<char>::size);
+        auto len = std::accumulate(sizes.begin(), sizes.end(), size_t(0));
+        internal::native_stack_net_stats::bytes_sent[sg_id] += len;
+        return _conn->send(bufs);
     }
 #else
     using data_sink_impl::put;
     virtual future<> put(packet p) override {
         auto sg_id = internal::scheduling_group_index(current_scheduling_group());
         internal::native_stack_net_stats::bytes_sent[sg_id] += p.len();
-        return _conn->send(std::move(p));
+        auto v = std::move(p).release();
+        return _conn->send(v);
     }
 #endif
     virtual future<> close() override {