diff options
| -rw-r--r-- | docs/design/011-pure-transport.md | 7 | ||||
| -rw-r--r-- | docs/research/096-per-delivery-cpu.md | 122 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/2026-09-29-drain-linux-aarch64.jsonl | 16 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/2026-09-29-transport-linux-aarch64.jsonl | 30 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/quic/src/main.rs | 49 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/run.py | 193 |
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(','): |