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.

6 files changed, 395 insertions, 22 deletions

diff --git a/docs/design/011-pure-transport.md b/docs/design/011-pure-transport.md
index a49d702..25a3b66 100644
--- a/docs/design/011-pure-transport.md
+++ b/docs/design/011-pure-transport.md
@@ -1,6 +1,6 @@
 # 011 — 연결 없는 전송: 전송의 상태를 앱의 데이터로
 
-작성일: 2026-09-28(같은 날 첫 판을 버리고 다시 씀). 상태: 설계. 1단계를 구현했고 M1, M2, M4를 쟀다.
+작성일: 2026-09-28(같은 날 첫 판을 버리고 다시 씀). 상태: 설계. 1단계를 구현했고 M1, M2, M4, M5를 쟀다.
 [외부 피드백 002](../outside-feedback/feedback-002.md)의 "QUIC을 능가할 자체 raw socket 프로토콜"에서 시작한다.
 
 1단계(링크 시제품과 kill -9 검사)는 [093](../research/093-connectionless-link.md)에 있다. 어느 쪽을 죽여도 핸드셰이크 없이
@@ -17,6 +17,11 @@ M2와 M4는 [095](../research/095-restart-and-first-contact.md)에 있다. 비�
 연산이 들지 않으며, 잃거나 겹친 기록이 없다. quinn은 1초 멈추고 10.9초 뒤에 따라잡는다. 그동안 핸드셰이크 1만 번에 서버가 CPU
 약 7초를 더 쓴다. 처음 만남은 1 RTT로, QUIC의 2 RTT의 절반이다.
 
+M5는 [096](../research/096-per-delivery-cpu.md)에 있다. 처음에는 작은 기록 하나에 quinn의 8배를 썼다. 쌓인 것을 훑는 탐색,
+기록마다의 시스템 호출과 할당, 19 B의 기록 머리 때문이었다. 고친 뒤 전송이 기록 하나에 쓰는 CPU는 1 KiB와 16 KiB에서 quinn보다
+25 % 적고, 61 B에서 1.06–1.12배다. 설계가 적은 송수신 경로(GSO, GRO, 묶어 받기)를 이때 넣었고, 작은 기록을 담는 RECORDS 프레임을
+더했다.
+
 ## 첫 판을 버린 까닭
 
 첫 판(1953051)은 QUIC의 층 몇 개를 Pure가 이미 가진 장치로 대신하는 설계였다. 저장소를 재전송 버퍼로 쓰고, 봉인을 한 번만
diff --git a/docs/research/096-per-delivery-cpu.md b/docs/research/096-per-delivery-cpu.md
new file mode 100644
index 0000000..3a0fecb
--- /dev/null
+++ b/docs/research/096-per-delivery-cpu.md
@@ -0,0 +1,122 @@
+# 096 — 전달당 CPU: 작은 기록에서 QUIC만큼, 큰 기록에서 QUIC보다 적게
+
+2026-09-28–29. [설계 011](../design/011-pure-transport.md)의 M5다. [094](094-idle-links.md)와 [095](095-restart-and-first-contact.md)가
+연결이 없어서 얻는 것을 봤다면, 여기서는 활성 링크 하나가 기록 하나에 쓰는 CPU를 QUIC과 댄다.
+
+착수 기준:
+
+- **바꿀 결정.** 011의 H4와 가드레일이다. 활성 링크의 전달당 CPU가 QUIC보다 10 % 넘게 나쁘면, 고친 뒤에야 채택한다.
+- **최소 확인.** 한 링크에 쌓인 보내기를 푼다. 기록은 64 B, 1 KiB, 16 KiB짜리다. 기록 하나에 보내는 쪽과 받는 쪽이 쓰는 CPU를
+  잰다. quinn은 한 연결의 한 스트림으로 같은 크기의 메시지를 보낸다. 메시지마다 u32 길이를 앞에 붙인다.
+- **중단 기준.** 위의 10 %다.
+
+## 쉬운 말로
+
+- 알아낸 것:
+  - 처음 쟀을 때 Pure 링크는 작은 기록 하나에 quinn의 8배를 썼다. 쌓인 양이 많을수록 느려지는 선형 탐색이 여럿 있었고, 기록마다
+    시스템 호출과 할당을 했다.
+  - 일곱 번 고친 뒤에는 61 B 기록 하나에 0.30 µs를 쓴다. quinn의 1.06–1.12배로, 가드레일(1.10) 선 위다.
+  - 1 KiB와 16 KiB 기록에서는 Pure가 quinn보다 25 % 적게 쓰고, 초당 30 % 넘게 더 보낸다.
+  - 모든 측정에서 넘겨받은 기록은 순서대로 한 번씩, 보낸 바이트 그대로였다. 손실과 재정렬 아래에서도 그렇다.
+- 아직 모르는 것: 손실과 재정렬 아래의 효율, 병목에서의 공정성(M6).
+- 새로 찾은 문제: 보내는 앱의 outbox가 저장할 때마다 통째로 다시 쓰인다. 쌓인 보내기가 많으면 저장이 그만큼 커진다.
+- 사용 경험의 차이: 같은 Pi가 같은 CPU로 더 많은 기기의 기록을 받는다. 무엇보다 기록마다 저장됐다는 확인이 붙는다. quinn
+  쪽에는 그것이 없다.
+
+## 처음 잰 값
+
+pure-serve 두 대를 잇고, 보내는 앱에 주문 2만 건을 쌓은 뒤 받는 노드를 띄워 풀었다. 61 B 기록 하나에 보내는 쪽 링크
+스레드가 1.1 µs, 받는 쪽 리스너 스레드가 2.4 µs를 썼다. quinn은 0.43 µs였다(2만 건, 연결 맺기 포함).
+
+## 고친 것
+
+전송만 따로 재는 벤치마크를 먼저 만들었다([transport.c](../../research/experiments/019-idle-links/transport.c)). 링크와
+리스너가 한 프로세스에 있고, 앱도 저장도 없다. 리스너는 넘기는 즉시 저장된 것으로 친다. 아래 수는 61 B 기록 20만 개를 푼
+값이다(보내는 쪽 + 받는 쪽, 기록당).
+
+| 커밋 | 무엇 | 기록당 CPU |
+|---|---|---:|
+| 처음(pure-serve, 2만 건) | | 3.5 µs |
+| [0f365f0](../../serve/link.c) | 쌓인 것을 훑지 않는다. GSO 묶음과 recvmmsg | 1.0 µs(pure-serve) |
+| e19aa30 | 링크 스레드를 잠들었을 때만 한 번 깨운다 | 0.49 µs |
+| 54035fe | 작은 기록을 RECORDS 프레임에 담는다(머리 19 B → 1–2 B) | 0.447 µs |
+| 803626a | 받을 때 UDP GRO, 리스너의 AES 키는 run이 바뀔 때만 넣는다 | 0.394 µs |
+| bebdecb | 받는 쪽: 순서대로 온 기록을 스트림의 줄에 붙여 그대로 넘긴다 | 0.31 µs |
+| ba835b2 | 보내는 쪽: 쌓인 기록의 바이트를 한 고리에 둔다 | 0.28 µs |
+| dac7ba0 | 패킷에 통째로 들어가는 기록은 조각 수와 상관없이 통째로 보낸다 | 0.28 µs(1,013 B: 3.18 → 2.85 µs) |
+
+- **훑기.** 보내는 쪽은 매 차례 쌓인 기록의 모든 조각을 두 번 훑어 저장을 기다리는 것이 있는지 봤다. 받는 쪽은 저장된
+  기록을 앞에서 뺄 때마다 뒤의 기록을 모두 당겼다. 둘 다 쌓인 양의 제곱으로 느려진다. 이제 기다리는 조각의 수를 세고, 받는
+  쪽 배열은 머리 위치를 둔다.
+- **시스템 호출.** 패킷 하나를 보내는 데 2.4 µs가 들었다. GSO로 54개를 한 번에 보내면 패킷당 0.45 µs다. 가득 찬 패킷은
+  PADDING으로 1200 B를 채운다. QUIC과 같은 방법이다. 받는 쪽은 GRO로 묶음을 한 버퍼로 받는다.
+- **깨우기.** 기록을 넣을 때마다 eventfd를 써서 링크 스레드를 깨웠다. 스레드가 잠들어 있을 때만 한 번 깨운다.
+- **틀.** RECORD 프레임은 기록마다 번호, 크기, 위치, 길이로 19 B를 붙였다. RECORDS 프레임은 첫 번호와 개수를 한 번 적고,
+  기록마다 크기를 가변 길이로 적는다. 패킷당 기록이 13.7개에서 17.2개가 됐다.
+- **할당과 복사.** 받는 쪽은 기록마다 할당하고, 넘길 때 다시 복사하고, 저장되면 풀었다. 이제 순서대로 온 온전한 기록은
+  전달이 싣는 모양 그대로(u32 크기, 바이트) 스트림의 줄에 붙는다. 줄은 그대로 넘기고, 저장되면 머리를 옮긴다. 보내는 쪽도
+  기록의 바이트를 들어온 순서대로 고리 하나에 두고, 저장되면 머리를 옮긴다.
+- **조각.** 1,013 B 기록은 조각(1000 B)보다 커서 두 조각으로 갔고, 받는 쪽은 그것을 느린 길로 모았다. 이제 패킷에 통째로
+  들어가는 기록은 통째로 가서 줄에 곧바로 붙는다. 패킷보다 큰 기록과, 조각 일부를 잃어 다시 보내는 기록만 조각으로 간다.
+- **확인.** 벤치마크는 넘겨받은 기록을 모두 확인한다. 순서대로 한 번씩 오는지, 바이트가 보낸 그대로인지 본다. 지연 1 ms,
+  손실 3 %, 재정렬 10 %인 네트워크 이름공간에서 세 크기 모두 틀린 것이 없었다.
+
+## 결과
+
+라즈베리 파이 5다. 원시 결과는 `research/experiments/019-idle-links/`의
+[2026-09-29-transport-linux-aarch64.jsonl](../../research/experiments/019-idle-links/2026-09-29-transport-linux-aarch64.jsonl)과
+[2026-09-29-drain-linux-aarch64.jsonl](../../research/experiments/019-idle-links/2026-09-29-drain-linux-aarch64.jsonl)에 있다.
+
+**전송만.** transport.c와 quinn으로 쌓인 기록을 모두 풀었다. 다섯 라운드의 중앙값이다. Pure 쪽은 넘겨받은 기록을 모두
+확인했고, 틀린 것은 없었다.
+
+| 기록(메시지) | 개수 | Pure 보내는 쪽 + 받는 쪽 | quinn 클라이언트 + 서비스 | Pure / quinn | 초당(Pure / quinn) |
+|---:|---:|---:|---:|---:|---:|
+| 61 B | 200,000 | 0.145 + 0.158 = **0.303 µs** | 0.14 + 0.13 = 0.27 µs | 1.12 | 509만 / 772만 |
+| 1,013 B | 50,000 | 1.456 + 1.457 = **2.91 µs** | 1.99 + 1.89 = 3.88 µs | 0.75 | 67만 / 51만 |
+| 16,013 B | 5,000 | 22.6 + 22.6 = **45.2 µs** | 31.1 + 29.7 = 60.9 µs | 0.74 | 4.4만 / 3.2만 |
+
+- quinn의 값은 한 스트림에 메시지를 이어 쓰고, 받는 쪽이 길이를 읽으며 세기만 한 것이다. 메시지마다 저장됐다는 확인도,
+  정확히 한 번도 없다. Pure의 값에는 기록마다의 번호, 두 단계 확인(받음, 저장됨), 기록 단위의 다시 보내기가 들어 있다.
+- 61 B의 비는 측정마다 흔들린다. 이틀 동안 세 번 쟀고 1.06, 1.08, 1.12였다(각 라운드 안에서도 Pure 0.29–0.45 µs, quinn
+  0.27–0.32 µs).
+- 작은 기록에서 남은 차이는 기록마다의 일보다 패킷마다의 짐에 있다. Pure는 패킷마다 스트림의 이름(STREAM 프레임, 여기서 25
+  B)과 RECORDS 머리(10 B)를 싣는다. 그래서 패킷에 기록이 17.4개 든다. quinn은 18.5개다.
+- 큰 기록에서 Pure가 적게 쓰는 까닭은 확인하지 않았다. 둘 다 GSO와 GRO를 쓴다.
+
+**pure-serve 끝단 사이.** 설계의 M5 그대로다. 보내는 앱에 주문 2만 건을 쌓고, 받는 노드를 띄워 풀었다. 받는 앱이 주문마다
+이벤트를 적용하고, 묶음마다 저장한다(tmpfs). 세 라운드의 중앙값이다(16 KiB는 두 라운드다. 셋째 라운드는 앞 라운드의
+저장소가 tmpfs를 채워 멈췄다).
+
+| 기록 | 초당 | 보내는 쪽(링크 스레드) | 받는 쪽(리스너 스레드) | quinn(전송만) |
+|---:|---:|---:|---:|---:|
+| 61 B | 108만 | 0.47(0.27) µs | 1.10(0.33) µs | 0.43 µs |
+| 1,013 B | 35만 | 2.71(2.22) µs | 4.16(2.18) µs | 3.93 µs |
+| 16,013 B | 2.3만 | 43.1(40.6) µs | 59.4(35.7) µs | 61.3 µs |
+
+- 받는 쪽의 나머지는 앱의 적용과 저장이다. quinn 쪽에는 없는 일이라 견주지 않는다.
+- 링크 스레드만 봐도 기록당 CPU가 전송만 쟀을 때보다 크다(1 KiB에 4.4 µs 대 2.9 µs). 다시 보내기 때문은 아니다. 1 KiB에서 풀기
+  중 다시 보낸 것은 받는 노드가 꺼져 있을 때의 몇 개뿐이었다. 받는 앱이 묶음(약 100건)을 저장할 때마다 STORED가 가고, 그때마다
+  링크가 깨어 작은 단위로 일한다. 전송만 잴 때는 받는 쪽이 넘겨받는 즉시 저장된 것으로 친다.
+- 채우는 동안 보내는 앱의 저장이 느려졌다. outbox(덜어 내지 않은 보내기)가 저장 때마다 links 값에 통째로 다시 쓰인다. 쌓인
+  양에 비례하는 저장이다. 16 KiB 2만 건이면 저장 한 번이 수백 MB에 이른다. 094의 들은 번호와 같은 모양의 문제다.
+
+## 판정
+
+- **1 KiB와 16 KiB: 통과.** 전송이 기록 하나에 쓰는 CPU가 quinn보다 25 % 적다. 초당 보내는 수는 30 % 넘게 많다.
+- **61 B: 가드레일 선 위.** quinn의 1.06–1.12배다. 남은 차이는 패킷마다 싣는 스트림 이름과 RECORDS 머리다. 이것을 줄이면
+  넘는다. 연결 없는 모델에서는 패킷마다 자기가 누구의 어느 스트림인지 말해야 하므로, 이름 대신 짧은 이름표를 쓰는 방법을 따로
+  설계해야 한다.
+- H4("활성 링크의 전달당 CPU는 QUIC 기준선보다 나쁘지 않다")는 1 KiB 이상에서 선다. 작은 기록에서는 거의 선다. quinn의 값에는
+  기록마다의 확인과 정확히 한 번이 없다는 점을 함께 본다.
+
+## 한계와 다음
+
+- **outbox의 저장.** 보내는 앱의 outbox가 저장 때마다 통째로 쓰인다. 들은 번호처럼 목록으로 바꿔, 저장이 바뀐 것만 로그에
+  쓰게 한다. 목록에는 이진 바이트 칸이 먼저 필요하다(지금은 텍스트 칸이 UTF-8만 받는다). 다음 일이다.
+- **하나씩 넣으며 보내기.** 기록을 하나씩 넣는 스레드와 링크 스레드가 잠금을 다퉈, 기록당 1.5 µs가 든다. pure-serve는 저장
+  한 번의 보내기를 묶어 넣으므로 덜하다. 묶어 넣는 API를 둔다.
+- **조각의 크기.** 큰 기록은 1000 B 조각으로 나뉜다. 가장 긴 이름에도 맞게 잡은 값이라 패킷마다 100 B 남짓이 빈다. 조각의
+  경계를 패킷에 맞추면 큰 기록에서 더 줄어든다.
+- **손실과 재정렬.** 재정렬을 손실로 잘못 봐서 다시 보내는 것이 많다(093). 이제 한 패킷에 기록이 17개라 잘못 본 손실의 값이
+  커졌다. M6에서 RACK과 적응형 문턱을 본다.
diff --git a/research/experiments/019-idle-links/2026-09-29-drain-linux-aarch64.jsonl b/research/experiments/019-idle-links/2026-09-29-drain-linux-aarch64.jsonl
new file mode 100644
index 0000000..f3758ff
--- /dev/null
+++ b/research/experiments/019-idle-links/2026-09-29-drain-linux-aarch64.jsonl
@@ -0,0 +1,16 @@
+{"side": "pure", "messages": 20000, "item_bytes": 48, "first_seen": 1793, "seconds": 0.0169, "per_second": 1076457, "sender_cpu_us_per_message": 0.47, "sender_link_cpu_us_per_message": 0.26, "receiver_cpu_us_per_message": 1.08, "receiver_link_cpu_us_per_message": 0.33}
+{"side": "quinn", "messages": 20000, "size": 61, "read": 20000, "seconds": 0.004, "per_second": 4676464, "client_cpu_us_per_message": 0.34, "service_cpu_us_per_message": 0.23}
+{"side": "pure", "messages": 20000, "item_bytes": 48, "first_seen": 1281, "seconds": 0.0166, "per_second": 1125905, "sender_cpu_us_per_message": 0.46, "sender_link_cpu_us_per_message": 0.27, "receiver_cpu_us_per_message": 1.1, "receiver_link_cpu_us_per_message": 0.32}
+{"side": "quinn", "messages": 20000, "size": 61, "read": 20000, "seconds": 0.003, "per_second": 6193413, "client_cpu_us_per_message": 0.25, "service_cpu_us_per_message": 0.18}
+{"side": "pure", "messages": 20000, "item_bytes": 48, "first_seen": 769, "seconds": 0.0196, "per_second": 983509, "sender_cpu_us_per_message": 0.48, "sender_link_cpu_us_per_message": 0.27, "receiver_cpu_us_per_message": 1.2, "receiver_link_cpu_us_per_message": 0.33}
+{"side": "quinn", "messages": 20000, "size": 61, "read": 20000, "seconds": 0.003, "per_second": 6330681, "client_cpu_us_per_message": 0.24, "service_cpu_us_per_message": 0.18}
+{"side": "pure", "messages": 20000, "item_bytes": 1000, "first_seen": 306, "seconds": 0.0572, "per_second": 344364, "sender_cpu_us_per_message": 2.71, "sender_link_cpu_us_per_message": 2.19, "receiver_cpu_us_per_message": 4.16, "receiver_link_cpu_us_per_message": 2.18}
+{"side": "quinn", "messages": 20000, "size": 1013, "read": 20000, "seconds": 0.039, "per_second": 511194, "client_cpu_us_per_message": 2.04, "service_cpu_us_per_message": 1.93}
+{"side": "pure", "messages": 20000, "item_bytes": 1000, "first_seen": 331, "seconds": 0.0566, "per_second": 347422, "sender_cpu_us_per_message": 2.63, "sender_link_cpu_us_per_message": 2.22, "receiver_cpu_us_per_message": 4.28, "receiver_link_cpu_us_per_message": 2.18}
+{"side": "quinn", "messages": 20000, "size": 1013, "read": 20000, "seconds": 0.039, "per_second": 511684, "client_cpu_us_per_message": 2.04, "service_cpu_us_per_message": 1.88}
+{"side": "pure", "messages": 20000, "item_bytes": 1000, "first_seen": 84, "seconds": 0.0544, "per_second": 365774, "sender_cpu_us_per_message": 2.72, "sender_link_cpu_us_per_message": 2.25, "receiver_cpu_us_per_message": 4.09, "receiver_link_cpu_us_per_message": 2.22}
+{"side": "quinn", "messages": 20000, "size": 1013, "read": 20000, "seconds": 0.042, "per_second": 480746, "client_cpu_us_per_message": 2.15, "service_cpu_us_per_message": 1.89}
+{"side": "pure", "messages": 20000, "item_bytes": 16000, "first_seen": 7, "seconds": 1.0315, "per_second": 19382, "sender_cpu_us_per_message": 48.02, "sender_link_cpu_us_per_message": 44.77, "receiver_cpu_us_per_message": 66.42, "receiver_link_cpu_us_per_message": 39.39}
+{"side": "quinn", "messages": 20000, "size": 16013, "read": 20000, "seconds": 0.619, "per_second": 32335, "client_cpu_us_per_message": 31.01, "service_cpu_us_per_message": 28.92}
+{"side": "pure", "messages": 20000, "item_bytes": 16000, "first_seen": 17, "seconds": 0.7495, "per_second": 26662, "sender_cpu_us_per_message": 38.13, "sender_link_cpu_us_per_message": 36.46, "receiver_cpu_us_per_message": 52.28, "receiver_link_cpu_us_per_message": 32.05}
+{"side": "quinn", "messages": 20000, "size": 16013, "read": 20000, "seconds": 0.635, "per_second": 31504, "client_cpu_us_per_message": 31.87, "service_cpu_us_per_message": 30.71}
diff --git a/research/experiments/019-idle-links/2026-09-29-transport-linux-aarch64.jsonl b/research/experiments/019-idle-links/2026-09-29-transport-linux-aarch64.jsonl
new file mode 100644
index 0000000..72f3b95
--- /dev/null
+++ b/research/experiments/019-idle-links/2026-09-29-transport-linux-aarch64.jsonl
@@ -0,0 +1,30 @@
+{"side": "pure", "item_bytes": 48, "records": 200000, "record_bytes": 61, "seconds": 0.036, "per_second": 5615011, "packets": 11453, "wrong": 0, "resent": 0, "link_us_per_record": 0.142, "listener_us_per_record": 0.15, "other_us_per_record": 0.011}
+{"side": "quinn", "item_bytes": 48, "messages": 200000, "size": 61, "read": 200000, "seconds": 0.026, "per_second": 7715099, "client_cpu_us_per_message": 0.14, "service_cpu_us_per_message": 0.13}
+{"side": "pure", "item_bytes": 48, "records": 200000, "record_bytes": 61, "seconds": 0.039, "per_second": 5093829, "packets": 11403, "wrong": 0, "resent": 0, "link_us_per_record": 0.145, "listener_us_per_record": 0.158, "other_us_per_record": 0.024}
+{"side": "quinn", "item_bytes": 48, "messages": 200000, "size": 61, "read": 200000, "seconds": 0.026, "per_second": 7729533, "client_cpu_us_per_message": 0.14, "service_cpu_us_per_message": 0.13}
+{"side": "pure", "item_bytes": 48, "records": 200000, "record_bytes": 61, "seconds": 0.054, "per_second": 3701858, "packets": 11398, "wrong": 0, "resent": 0, "link_us_per_record": 0.175, "listener_us_per_record": 0.185, "other_us_per_record": 0.035}
+{"side": "quinn", "item_bytes": 48, "messages": 200000, "size": 61, "read": 200000, "seconds": 0.058, "per_second": 3453915, "client_cpu_us_per_message": 0.17, "service_cpu_us_per_message": 0.15}
+{"side": "pure", "item_bytes": 48, "records": 200000, "record_bytes": 61, "seconds": 0.059, "per_second": 3367697, "packets": 11469, "wrong": 0, "resent": 0, "link_us_per_record": 0.226, "listener_us_per_record": 0.225, "other_us_per_record": 0.038}
+{"side": "quinn", "item_bytes": 48, "messages": 200000, "size": 61, "read": 200000, "seconds": 0.031, "per_second": 6458727, "client_cpu_us_per_message": 0.16, "service_cpu_us_per_message": 0.15}
+{"side": "pure", "item_bytes": 48, "records": 200000, "record_bytes": 61, "seconds": 0.037, "per_second": 5390888, "packets": 11413, "wrong": 0, "resent": 0, "link_us_per_record": 0.14, "listener_us_per_record": 0.148, "other_us_per_record": 0.022}
+{"side": "quinn", "item_bytes": 48, "messages": 200000, "size": 61, "read": 200000, "seconds": 0.026, "per_second": 7795676, "client_cpu_us_per_message": 0.14, "service_cpu_us_per_message": 0.13}
+{"side": "pure", "item_bytes": 1000, "records": 50000, "record_bytes": 1013, "seconds": 0.073, "per_second": 684187, "packets": 50000, "wrong": 0, "resent": 0, "link_us_per_record": 1.435, "listener_us_per_record": 1.437, "other_us_per_record": 0.094}
+{"side": "quinn", "item_bytes": 1000, "messages": 50000, "size": 1013, "read": 50000, "seconds": 0.096, "per_second": 521389, "client_cpu_us_per_message": 1.95, "service_cpu_us_per_message": 1.85}
+{"side": "pure", "item_bytes": 1000, "records": 50000, "record_bytes": 1013, "seconds": 0.074, "per_second": 674802, "packets": 50000, "wrong": 0, "resent": 0, "link_us_per_record": 1.452, "listener_us_per_record": 1.451, "other_us_per_record": 0.078}
+{"side": "quinn", "item_bytes": 1000, "messages": 50000, "size": 1013, "read": 50000, "seconds": 0.099, "per_second": 507322, "client_cpu_us_per_message": 2.01, "service_cpu_us_per_message": 1.94}
+{"side": "pure", "item_bytes": 1000, "records": 50000, "record_bytes": 1013, "seconds": 0.074, "per_second": 674399, "packets": 50000, "wrong": 0, "resent": 0, "link_us_per_record": 1.456, "listener_us_per_record": 1.457, "other_us_per_record": 0.11}
+{"side": "quinn", "item_bytes": 1000, "messages": 50000, "size": 1013, "read": 50000, "seconds": 0.096, "per_second": 519206, "client_cpu_us_per_message": 1.96, "service_cpu_us_per_message": 1.85}
+{"side": "pure", "item_bytes": 1000, "records": 50000, "record_bytes": 1013, "seconds": 0.077, "per_second": 651514, "packets": 50000, "wrong": 0, "resent": 0, "link_us_per_record": 1.505, "listener_us_per_record": 1.49, "other_us_per_record": 0.109}
+{"side": "quinn", "item_bytes": 1000, "messages": 50000, "size": 1013, "read": 50000, "seconds": 0.099, "per_second": 507608, "client_cpu_us_per_message": 2.01, "service_cpu_us_per_message": 1.92}
+{"side": "pure", "item_bytes": 1000, "records": 50000, "record_bytes": 1013, "seconds": 0.076, "per_second": 657794, "packets": 50000, "wrong": 0, "resent": 0, "link_us_per_record": 1.493, "listener_us_per_record": 1.493, "other_us_per_record": 0.129}
+{"side": "quinn", "item_bytes": 1000, "messages": 50000, "size": 1013, "read": 50000, "seconds": 0.098, "per_second": 511069, "client_cpu_us_per_message": 1.99, "service_cpu_us_per_message": 1.89}
+{"side": "pure", "item_bytes": 16000, "records": 5000, "record_bytes": 16013, "seconds": 0.118, "per_second": 42284, "packets": 80000, "wrong": 0, "resent": 0, "link_us_per_record": 23.345, "listener_us_per_record": 23.36, "other_us_per_record": 0.886}
+{"side": "quinn", "item_bytes": 16000, "messages": 5000, "size": 16013, "read": 5000, "seconds": 0.154, "per_second": 32537, "client_cpu_us_per_message": 31.08, "service_cpu_us_per_message": 29.32}
+{"side": "pure", "item_bytes": 16000, "records": 5000, "record_bytes": 16013, "seconds": 0.113, "per_second": 44238, "packets": 80000, "wrong": 0, "resent": 0, "link_us_per_record": 22.325, "listener_us_per_record": 22.354, "other_us_per_record": 1.221}
+{"side": "quinn", "item_bytes": 16000, "messages": 5000, "size": 16013, "read": 5000, "seconds": 0.159, "per_second": 31400, "client_cpu_us_per_message": 32.06, "service_cpu_us_per_message": 29.72}
+{"side": "pure", "item_bytes": 16000, "records": 5000, "record_bytes": 16013, "seconds": 0.115, "per_second": 43645, "packets": 80000, "wrong": 0, "resent": 0, "link_us_per_record": 22.617, "listener_us_per_record": 22.608, "other_us_per_record": 1.144}
+{"side": "quinn", "item_bytes": 16000, "messages": 5000, "size": 16013, "read": 5000, "seconds": 0.154, "per_second": 32407, "client_cpu_us_per_message": 31.07, "service_cpu_us_per_message": 29.11}
+{"side": "pure", "item_bytes": 16000, "records": 5000, "record_bytes": 16013, "seconds": 0.118, "per_second": 42553, "packets": 80000, "wrong": 0, "resent": 0, "link_us_per_record": 23.195, "listener_us_per_record": 23.227, "other_us_per_record": 1.129}
+{"side": "quinn", "item_bytes": 16000, "messages": 5000, "size": 16013, "read": 5000, "seconds": 0.16, "per_second": 31164, "client_cpu_us_per_message": 32.34, "service_cpu_us_per_message": 30.05}
+{"side": "pure", "item_bytes": 16000, "records": 5000, "record_bytes": 16013, "seconds": 0.114, "per_second": 43757, "packets": 80000, "wrong": 0, "resent": 0, "link_us_per_record": 22.538, "listener_us_per_record": 22.583, "other_us_per_record": 1.04}
+{"side": "quinn", "item_bytes": 16000, "messages": 5000, "size": 16013, "read": 5000, "seconds": 0.153, "per_second": 32692, "client_cpu_us_per_message": 30.93, "service_cpu_us_per_message": 29.93}
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();
diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py
index 63f88a2..96a0cbc 100644
--- a/research/experiments/019-idle-links/run.py
+++ b/research/experiments/019-idle-links/run.py
@@ -29,9 +29,17 @@ network namespace of its own (unshare -rn) whose loopback delays UDP:
 senders the node never heard from, and new quinn clients, each send one
 order and wait until it is stored, one at a time.
 
+With --transport SIZE:COUNT,... (M5), transport.c drains a backlog of
+COUNT records through a link and a listener in one process (no app, no
+store), then quinn sends as many messages of the same size over one
+stream; --rounds times each. With --drain SIZES (M5 as the design says),
+two pure-serve nodes: the sender's app saves --counts orders while the
+receiver is away, then the receiver starts and the link drains them;
+quinn does the same over one stream.
+
     python3 research/experiments/019-idle-links/run.py [--counts 1000,10000]
         [--store DIRECTORY] [--out FILE.jsonl] [--quic] [--restart [--no-memo]]
-        [--first-contact 5,25,50]
+        [--first-contact 5,25,50] [--transport 48:200000 --rounds 5] [--drain 48,1000]
 """
 import argparse
 import json
@@ -384,6 +392,143 @@ with tempfile.TemporaryDirectory(prefix='pure-m4-') as d:
             file.write(line + '\n')
 
 
+def thread_cpu(pid, name):
+    """CPU ms of the process's threads named name (all threads when None),
+    from the scheduler's nanoseconds (schedstat), not clock ticks."""
+    total = 0
+    for task in Path(f'/proc/{pid}/task').iterdir():
+        try:
+            if name and (task / 'comm').read_text().strip() != name:
+                continue
+            total += int((task / 'schedstat').read_text().split()[0])
+        except OSError:
+            pass
+    return total / 1e6
+
+
+def post_all(port, n, size, threads=8):
+    """n orders (item of size bytes) posted to the sender app, over threads connections."""
+    import http.client
+    import threading
+    body = json.dumps({'item': 'x' * size, 'count': 1}).encode()
+    left = [n]
+    lock = threading.Lock()
+
+    def work():
+        connection = http.client.HTTPConnection('127.0.0.1', port, timeout=60)
+        while True:
+            with lock:
+                if not left[0]:
+                    break
+                left[0] -= 1
+            connection.request('POST', '/inputs/buy', body, {'Content-Type': 'application/json'})
+            response = connection.getresponse()
+            response.read()
+            assert response.status == 200, response.status
+        connection.close()
+    workers = [threading.Thread(target=work) for _ in range(threads)]
+    for w in workers:
+        w.start()
+    for w in workers:
+        w.join()
+
+
+def drain_pure(work, n, size):
+    """M5 for Pure: n saved sends of an order whose item is size bytes wait
+    in the sender's outbox (the receiver is not there); then the receiver
+    starts, and the link drains them. Rate from the first applied to the
+    last; CPU of both nodes, and of their link threads alone."""
+    sender_module, receiver_module = work / 'drain-sender.pmod', work / 'drain-receiver.pmod'
+    for module, text in ((sender_module, check.LINK_SENDER), (receiver_module, check.LINK_RECEIVER)):
+        source = module.with_suffix('.aup')
+        source.write_text(text)
+        check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, module)
+    keys = {}
+    for name in ('a', 'b'):
+        node, info = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / f'key-{name}',
+                            '--port', free_port(socket.SOCK_STREAM), '--key', work / f'{name}.key'])
+        keys[name] = info['public_key']
+        node.terminate()
+        node.communicate(timeout=60)
+    (work / 'drain-senders').write_text(f'drain-sender.ordered {keys["a"]}\n')
+    link_port, port_a, port_b = free_port(socket.SOCK_DGRAM), free_port(socket.SOCK_STREAM), free_port(socket.SOCK_STREAM)
+    a, _ = start([ROOT / 'build/pure-serve', '--module', sender_module, '--store', work / 'drain-a', '--port', port_a,
+                  '--threads', '2', '--key', work / 'a.key', '--send', f'ordered=pure://127.0.0.1:{link_port}/order',
+                  '--seal', f'ordered={keys["b"]}'])
+    post_all(port_a, n, size)
+    assert view(port_a, 'sent') == n
+    b, _ = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / 'drain-b', '--port', port_b,
+                  '--threads', '2', '--key', work / 'b.key', '--link-port', link_port, '--senders', work / 'drain-senders'])
+    cpu = {name: (thread_cpu(pid, None), thread_cpu(pid, thread)) for name, pid, thread in
+           (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen'))}
+    import http.client
+    connection = http.client.HTTPConnection('127.0.0.1', port_b, timeout=30)
+    first = None
+    while True:  # the count every 2 ms, over one connection
+        connection.request('GET', '/views/count')
+        count = json.loads(connection.getresponse().read())
+        now = time.monotonic()
+        if count and first is None:
+            first = (now, count)
+        if count >= n:
+            break
+        time.sleep(0.002)
+    connection.close()
+    seconds = now - first[0]
+    row = {'messages': n, 'item_bytes': size, 'first_seen': first[1], 'seconds': round(seconds, 4),
+           'per_second': round((n - first[1]) / seconds) if seconds else None}
+    for name, pid, thread in (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen')):
+        total, link = thread_cpu(pid, None) - cpu[name][0], thread_cpu(pid, thread) - cpu[name][1]
+        row[f'{name}_cpu_us_per_message'] = round(total * 1000 / n, 2)
+        row[f'{name}_link_cpu_us_per_message'] = round(link * 1000 / n, 2)
+    for name, node in (('sender', a), ('receiver', b)):
+        node.terminate()
+        _, error = node.communicate(timeout=120)
+        stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1]
+        keys = ('link_packets', 'link_records', 'link_resent', 'link_lost') if name == 'sender' else \
+               ('listener_received', 'listener_deliveries', 'bursts')
+        row[name] = {k: stats[k] for k in keys}
+    subprocess.run(['rm', '-rf', str(work / 'drain-a'), str(work / 'drain-b')], check=True)  # tmpfs: room for the next
+    return row
+
+
+def drain_quic(work, n, size):
+    """The same over quinn: one connection, one stream, n messages; CPU of
+    both sides. The service only counts them (no store)."""
+    binary = Path(__file__).parent / 'quic/target/release/idle-quic'
+    import resource
+    server, info = start([binary, 'serve', work / 'cert.der', work / 'key.der'])
+    before, children = thread_cpu(server.pid, None), resource.getrusage(resource.RUSAGE_CHILDREN)
+    # The same bytes as Pure's record: its prefix (count, item's length), the item and the city ("Seoul").
+    client = subprocess.run([str(binary), 'drain', str(info['port']), str(n), str(size + 8 + 5)], capture_output=True,
+                            timeout=600)
+    after = resource.getrusage(resource.RUSAGE_CHILDREN)
+    row = json.loads(client.stdout)
+    server_cpu = thread_cpu(server.pid, None) - before
+    client_cpu = (after.ru_utime + after.ru_stime - children.ru_utime - children.ru_stime) * 1000
+    row['client_cpu_us_per_message'] = round(client_cpu * 1000 / n, 2)
+    server.kill()
+    server.wait()
+    row['service_cpu_us_per_message'] = round(server_cpu * 1000 / n, 2)
+    return row
+
+
+def transport(work, size, n):
+    """The transport alone (transport.c): a backlog of n records of an order
+    with a size-byte item drained through a link and a listener in one
+    process, no app and no store."""
+    modules = {}
+    for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)):
+        source = work / f'{name}.aup'
+        source.write_text(text)
+        modules[name] = work / f'{name}.pmod'
+        check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, modules[name])
+    done = subprocess.run([str(ROOT / 'build/link-transport'), str(modules['sender']), str(modules['receiver']), str(n),
+                           str(size), '0'], capture_output=True, text=True, timeout=600)
+    assert done.returncode == 0, done.stderr
+    return json.loads(done.stdout)
+
+
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument('--counts', default='1000,10000,100000,1000000')
@@ -399,8 +544,54 @@ def main():
     parser.add_argument('--kill-at', type=int, default=20)
     parser.add_argument('--first-contact', type=str, default=None,
                         help='M4: one-way delays in ms (for example 5,25,50), each in a namespace of its own')
+    parser.add_argument('--drain', type=str, default=None,
+                        help='M5: item sizes in bytes (for example 48,1000,16000); --counts gives the messages')
+    parser.add_argument('--transport', type=str, default=None,
+                        help='M5: SIZE:COUNT pairs (for example 48:200000,1000:50000): the transport alone and quinn')
+    parser.add_argument('--rounds', type=int, default=1)
     args = parser.parse_args()
     counts = [int(c) for c in args.counts.split(',')]
+    if args.transport:
+        subprocess.run(['make', '-s', '-C', str(ROOT), 'build/libpure.a', 'build/pure-crypto.so'], check=True)
+        subprocess.run(['cc', '-O2', '-g', '-std=c17', '-Wall', '-Wextra', '-Werror', f'-I{ROOT}/core', f'-I{ROOT}/serve',
+                        '-DPURE_ASYNC_EFFECTS', str(Path(__file__).parent / 'transport.c'), str(ROOT / 'serve/link.c'),
+                        str(ROOT / 'serve/crypto.c'), str(ROOT / 'build/libpure.a'), '-lm', '-pthread', '-ldl', '-o',
+                        str(ROOT / 'build/link-transport')], check=True)
+        subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True)
+        with tempfile.TemporaryDirectory(prefix='pure-m5-', dir=args.store) as directory:
+            work = Path(directory)
+            make_certificate(work)
+            for pair in args.transport.split(','):
+                size, n = (int(x) for x in pair.split(':'))
+                for _ in range(args.rounds):
+                    for name, measure in (('pure', lambda: transport(work, size, n)), ('quinn', lambda: drain_quic(work, n, size))):
+                        row = {'side': name, 'item_bytes': size, **measure()}
+                        line = json.dumps(row)
+                        print(line, flush=True)
+                        if args.out:
+                            with open(args.out, 'a') as out:
+                                out.write(line + '\n')
+        return
+    if args.drain:
+        subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True)
+        with tempfile.TemporaryDirectory(prefix='pure-m5-', dir=args.store) as directory:
+            work = Path(directory)
+            make_certificate(work)
+            for size in (int(s) for s in args.drain.split(',')):
+                for round_, n in enumerate(counts):
+                    for name, measure in (('pure', drain_pure), ('quinn', drain_quic)):
+                        case = work / f'{name}-{size}-{n}-{round_}'
+                        case.mkdir()
+                        if name == 'quinn':
+                            for f in ('cert.der', 'key.der'):
+                                (case / f).write_bytes((work / f).read_bytes())
+                        row = {'side': name, **measure(case, n, size)}
+                        line = json.dumps(row)
+                        print(line, flush=True)
+                        if args.out:
+                            with open(args.out, 'a') as out:
+                                out.write(line + '\n')
+        return
     if args.first_contact:
         subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True)
         for delay in args.first_contact.split(','):