Measure the CPU a record costs a link, against quinn (096)

M5 of design 011. At first a link spent eight times quinn's CPU on a small
record: scans of the backlog, a system call and allocations a record, 19
bytes of framing. After the seven changes before this one, the transport
alone (transport.c; a backlog drained, every record checked) spends per
record 0.30 us for 61 bytes (quinn 0.27: 1.06 to 1.12 times over three
campaigns, at the design's 10 % line), 2.91 us for 1,013 bytes (quinn
3.88) and 45.2 us for 16,013 bytes (quinn 60.9): 25 % less for the larger
ones, and 30 % more records a second. What is left for small records is
per packet: the stream's names and the RECORDS head. quinn's numbers carry
no confirmation of storing and no exactly once.

Between two pure-serve nodes the links spend more per record, woken at
the receiving app's pace. Filling the sender showed the outbox is written
whole at every save, as the numbers heard were (094): the next thing to
change.

run.py gains --transport and --drain; the quinn service reads a drain a
chunk at a time and its client writes 64 KiB at a time.

1 files changed, 29 insertions, 20 deletions

diff --git a/research/experiments/019-idle-links/quic/src/main.rs b/research/experiments/019-idle-links/quic/src/main.rs
index 28cda77..80e663f 100644
--- a/research/experiments/019-idle-links/quic/src/main.rs
+++ b/research/experiments/019-idle-links/quic/src/main.rs
@@ -116,24 +116,27 @@ fn serve(cert: Vec<u8>, key: Vec<u8>, port: u16) {
 /// A drain's stream begins with these bytes; an order's never does (a client number that large).
 const DRAIN: &[u8; 4] = b"pure";
 
-/// Messages each after its u32 length until a length of 0; then how many, as the answer.
+/// Messages each after its u32 length until a length of 0, read a chunk at
+/// a time as they come; then how many, as the answer.
 async fn drain(send: &mut quinn::SendStream, recv: &mut quinn::RecvStream) {
-    let mut count = 0u64;
-    let mut buffer = Vec::new();
-    loop {
-        let mut length = [0u8; 4];
-        if recv.read_exact(&mut length).await.is_err() {
-            return;
-        }
-        let length = u32::from_le_bytes(length) as usize;
-        if length == 0 {
-            break;
-        }
-        buffer.resize(length, 0);
-        if recv.read_exact(&mut buffer).await.is_err() {
-            return;
+    let (mut count, mut pending, mut done) = (0u64, Vec::<u8>::new(), false);
+    while !done {
+        let Ok(Some(chunk)) = recv.read_chunk(1 << 20, true).await else { return };
+        pending.extend_from_slice(&chunk.bytes);
+        let mut at = 0;
+        while pending.len() - at >= 4 {
+            let length = u32::from_le_bytes(pending[at..at + 4].try_into().unwrap()) as usize;
+            if length == 0 {
+                done = true;
+                break;
+            }
+            if pending.len() - at - 4 < length {
+                break;
+            }
+            at += 4 + length;
+            count += 1;
         }
-        count += 1;
+        pending.drain(..at);
     }
     let _ = send.write_all(&count.to_le_bytes()).await;
     let _ = send.finish();
@@ -356,12 +359,18 @@ fn drain_client(port: u16, count: u64, size: usize) {
         endpoint.set_default_client_config(client_config());
         let connection = endpoint.connect(SocketAddr::from(([127, 0, 0, 1], port)), "localhost").unwrap().await.unwrap();
         let (mut send, mut recv) = connection.open_bi().await.unwrap();
-        let message = vec![b'x'; size];
+        // The messages, each after its length, written 64 KiB at a time.
+        let mut frame = (size as u32).to_le_bytes().to_vec();
+        frame.extend(std::iter::repeat_n(b'x', size));
+        let per_chunk = ((64 << 10) / frame.len()).max(1);
+        let chunk: Vec<u8> = frame.iter().copied().cycle().take(frame.len() * per_chunk).collect();
         let began = Instant::now();
         send.write_all(DRAIN).await.unwrap();
-        for _ in 0..count {
-            send.write_all(&(size as u32).to_le_bytes()).await.unwrap();
-            send.write_all(&message).await.unwrap();
+        let mut left = count as usize;
+        while left > 0 {
+            let n = left.min(per_chunk);
+            send.write_all(&chunk[..n * frame.len()]).await.unwrap();
+            left -= n;
         }
         send.write_all(&0u32.to_le_bytes()).await.unwrap();
         send.finish().unwrap();