diff options
4 files changed, 505 insertions, 0 deletions
diff --git a/research/experiments/019-idle-links/2026-09-30-transport-three-linux-aarch64.jsonl b/research/experiments/019-idle-links/2026-09-30-transport-three-linux-aarch64.jsonl new file mode 100644 index 0000000..3e43568 --- /dev/null +++ b/research/experiments/019-idle-links/2026-09-30-transport-three-linux-aarch64.jsonl @@ -0,0 +1,15 @@ +{"item": 48, "records": 300000, "round": 0, "pure": {"us": 0.321, "per_second": 4802771, "packets": 13889}, "quinn": {"us": 0.32, "per_second": 5644023}, "ngtcp2": {"us": 0.27, "per_second": 6164660}} +{"item": 48, "records": 300000, "round": 1, "pure": {"us": 0.28200000000000003, "per_second": 5876868, "packets": 13902}, "quinn": {"us": 0.31000000000000005, "per_second": 5991396}, "ngtcp2": {"us": 0.27, "per_second": 6081678}} +{"item": 48, "records": 300000, "round": 2, "pure": {"us": 0.278, "per_second": 5810147, "packets": 13835}, "quinn": {"us": 0.31000000000000005, "per_second": 6066156}, "ngtcp2": {"us": 0.27, "per_second": 6174980}} +{"item": 48, "records": 300000, "round": 3, "pure": {"us": 0.28, "per_second": 5781982, "packets": 13835}, "quinn": {"us": 0.31000000000000005, "per_second": 6091074}, "ngtcp2": {"us": 0.27, "per_second": 6209375}} +{"item": 48, "records": 300000, "round": 4, "pure": {"us": 0.274, "per_second": 5879728, "packets": 13847}, "quinn": {"us": 0.31000000000000005, "per_second": 6069152}, "ngtcp2": {"us": 0.27, "per_second": 6189593}} +{"item": 1000, "records": 100000, "round": 0, "pure": {"us": 2.612, "per_second": 756596, "packets": 75008}, "quinn": {"us": 3.84, "per_second": 520495}, "ngtcp2": {"us": 3.5, "per_second": 542464}} +{"item": 1000, "records": 100000, "round": 1, "pure": {"us": 3.2969999999999997, "per_second": 490797, "packets": 75008}, "quinn": {"us": 3.94, "per_second": 508117}, "ngtcp2": {"us": 3.5700000000000003, "per_second": 532466}} +{"item": 1000, "records": 100000, "round": 2, "pure": {"us": 2.521, "per_second": 786189, "packets": 75008}, "quinn": {"us": 3.94, "per_second": 507144}, "ngtcp2": {"us": 3.52, "per_second": 542998}} +{"item": 1000, "records": 100000, "round": 3, "pure": {"us": 2.6390000000000002, "per_second": 751117, "packets": 75008}, "quinn": {"us": 3.82, "per_second": 522353}, "ngtcp2": {"us": 3.54, "per_second": 536007}} +{"item": 1000, "records": 100000, "round": 4, "pure": {"us": 2.6239999999999997, "per_second": 752571, "packets": 75008}, "quinn": {"us": 4.43, "per_second": 442694}, "ngtcp2": {"us": 3.54, "per_second": 538312}} +{"item": 16000, "records": 20000, "round": 0, "pure": {"us": 44.4, "per_second": 44166, "packets": 233920}, "quinn": {"us": 60.16, "per_second": 32745}, "ngtcp2": {"us": 54.129999999999995, "per_second": 35859}} +{"item": 16000, "records": 20000, "round": 1, "pure": {"us": 40.21, "per_second": 49579, "packets": 233920}, "quinn": {"us": 61.55, "per_second": 32043}, "ngtcp2": {"us": 54.989999999999995, "per_second": 35156}} +{"item": 16000, "records": 20000, "round": 2, "pure": {"us": 40.06, "per_second": 49735, "packets": 233920}, "quinn": {"us": 59.16, "per_second": 33280}, "ngtcp2": {"us": 54.81, "per_second": 35190}} +{"item": 16000, "records": 20000, "round": 3, "pure": {"us": 39.324, "per_second": 50714, "packets": 233920}, "quinn": {"us": 60.26, "per_second": 32420}, "ngtcp2": {"us": 53.99, "per_second": 35958}} +{"item": 16000, "records": 20000, "round": 4, "pure": {"us": 40.421, "per_second": 49209, "packets": 233920}, "quinn": {"us": 60.14, "per_second": 32044}, "ngtcp2": {"us": 54.43, "per_second": 35632}} diff --git a/research/experiments/019-idle-links/ngtcp2/build.sh b/research/experiments/019-idle-links/ngtcp2/build.sh new file mode 100755 index 0000000..e76017a --- /dev/null +++ b/research/experiments/019-idle-links/ngtcp2/build.sh @@ -0,0 +1,16 @@ +#!/bin/sh +# ngtcp2 1.25.0 with its OpenSSL 3.5 backend, into build/third (not kept in +# the repository), and drain.c against it: build/third/ngtcp2-drain. +set -e +root=$(cd "$(dirname "$0")/../../../.." && pwd) +third=$root/build/third +if [ ! -f "$third/prefix/lib/libngtcp2.a" ]; then + mkdir -p "$third" + [ -d "$third/ngtcp2" ] || git clone -q --depth 1 --branch v1.25.0 https://github.com/ngtcp2/ngtcp2 "$third/ngtcp2" + mkdir -p "$third/ngtcp2/b" + (cd "$third/ngtcp2/b" && cmake -DENABLE_LIB_ONLY=ON -DENABLE_OPENSSL=ON -DENABLE_SHARED_LIB=OFF -DBUILD_TESTING=OFF \ + -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX="$third/prefix" .. > /dev/null && make -s -j4 > /dev/null && + make -s install > /dev/null) +fi +cc -O2 -g -std=c17 -Wall -Wextra -Werror -I"$third/prefix/include" "$(dirname "$0")/drain.c" \ + "$third/prefix/lib/libngtcp2_crypto_ossl.a" "$third/prefix/lib/libngtcp2.a" -lssl -lcrypto -o "$third/ngtcp2-drain" diff --git a/research/experiments/019-idle-links/ngtcp2/drain.c b/research/experiments/019-idle-links/ngtcp2/drain.c new file mode 100644 index 0000000..43942e7 --- /dev/null +++ b/research/experiments/019-idle-links/ngtcp2/drain.c @@ -0,0 +1,454 @@ +/* M5 over ngtcp2 (docs/design/011-pure-transport.md): what the quinn + * harness (../quic) does, in C, with OpenSSL 3.5's QUIC TLS (ngtcp2's ossl + * backend), so a second QUIC stands beside Pure's links. + * + * drain serve CERT.der KEY.der [PORT] prints {"port":N}, then serves + * drain drain PORT COUNT SIZE one connection, one stream + * + * The client writes "pure", then COUNT messages each after its u32 length + * (SIZE bytes), then a length of 0, and ends the stream; the service counts + * the messages (no store) and answers the count (u64). Packets go in bursts + * with UDP GSO (DRAIN_GSO=0: one by one), as quinn's and Pure's do. Built by + * build.sh against ngtcp2 1.25.0 fetched into build/third. */ +#define _GNU_SOURCE +#include <errno.h> +#include <netdb.h> +#include <netinet/in.h> +#include <netinet/udp.h> +#include <poll.h> +#include <stdarg.h> +#include <stdbool.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/socket.h> +#include <time.h> +#include <unistd.h> + +#include <ngtcp2/ngtcp2.h> +#include <ngtcp2/ngtcp2_crypto.h> +#include <ngtcp2/ngtcp2_crypto_ossl.h> +#include <openssl/err.h> +#include <openssl/rand.h> +#include <openssl/ssl.h> + +#ifndef UDP_SEGMENT +#define UDP_SEGMENT 103 +#endif + +static const unsigned char ALPN[] = "\x04pure"; +enum { BURST = 64 * 1024, CHUNK = 64 * 1024 }; + +static uint64_t now_ns(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (uint64_t)t.tv_sec * NGTCP2_SECONDS + (uint64_t)t.tv_nsec; +} + +static void fail(const char *what) { + fprintf(stderr, "drain: %s\n", what); + exit(1); +} + +typedef struct { + ngtcp2_crypto_conn_ref ref; + ngtcp2_crypto_ossl_ctx *tls; + SSL *ssl; + ngtcp2_conn *conn; + int fd; + bool gso, server, done; + struct sockaddr_storage local, remote; + socklen_t local_size, remote_size; + ngtcp2_ccerr error; + /* The client's stream: its bytes are a function of their offset, so + * what ngtcp2 keeps to send again points into buffers that never change: + * "pure", then the messages (a pattern of one period), then a length of + * 0. */ + int64_t stream; + uint64_t count, size, offset, total; + unsigned char *pattern; + size_t period; + /* The service's count, and where it is in a message. */ + uint64_t messages, need, skip; + unsigned char length[4]; + unsigned have; + bool header, answered; + unsigned char answer[8]; + uint64_t read; +} Side; + +static void log_line(void *user, const char *format, ...) { + (void)user; + va_list list; + va_start(list, format); + vfprintf(stderr, format, list); + va_end(list); + fputc('\n', stderr); +} + +static ngtcp2_conn *conn_of(ngtcp2_crypto_conn_ref *ref) { return ((Side *)ref->user_data)->conn; } + +static void random_bytes(uint8_t *out, size_t n, const ngtcp2_rand_ctx *ctx) { + (void)ctx; + if (RAND_bytes(out, (int)n) != 1) abort(); +} + +static int new_cid(ngtcp2_conn *conn, ngtcp2_cid *cid, ngtcp2_stateless_reset_token *token, size_t n, void *user) { + (void)conn, (void)user; + if (RAND_bytes(cid->data, (int)n) != 1 || RAND_bytes(token->data, sizeof(token->data)) != 1) + return NGTCP2_ERR_CALLBACK_FAILURE; + cid->datalen = n; + return 0; +} + +/* The client: its stream opened once the service allows one. */ +static int streams_allowed(ngtcp2_conn *conn, uint64_t most, void *user) { + (void)most; + Side *c = user; + if (c->stream < 0 && ngtcp2_conn_open_bidi_stream(conn, &c->stream, NULL)) c->stream = -1; + return 0; +} + +/* The service: messages counted as they come, over any cut; at the length of + * 0 the count goes back and the stream ends. */ +static int stream_data(ngtcp2_conn *conn, uint32_t flags, int64_t stream, uint64_t offset, const uint8_t *data, size_t n, + void *user, void *stream_user) { + (void)offset, (void)stream_user; + Side *s = user; + if (!s->server) { /* the client: the count said back */ + for (size_t i = 0; i < n && s->read < 8; ++i) s->answer[s->read++] = data[i]; + if ((flags & NGTCP2_STREAM_DATA_FLAG_FIN) || s->read >= 8) s->done = true; + } else + for (size_t i = 0; i < n;) { + if (!s->header) { /* "pure" */ + if (++s->have == 4) s->header = true, s->have = 0; + ++i; + } else if (s->skip) { + size_t take = n - i < s->skip ? n - i : (size_t)s->skip; + s->skip -= take; + i += take; + } else { + s->length[s->have++] = data[i++]; + if (s->have < 4) continue; + s->have = 0; + uint32_t length = (uint32_t)s->length[0] | (uint32_t)s->length[1] << 8 | (uint32_t)s->length[2] << 16 | + (uint32_t)s->length[3] << 24; + if (!length) { + s->answered = true; + s->stream = stream; + for (int b = 0; b < 8; ++b) s->answer[b] = (unsigned char)(s->messages >> 8 * b); + break; + } + ++s->messages; + s->skip = length; + } + } + ngtcp2_conn_extend_max_stream_offset(conn, stream, n); + ngtcp2_conn_extend_max_offset(conn, n); + return 0; +} + +static ngtcp2_callbacks callbacks(bool server) { + ngtcp2_callbacks c = { + .recv_crypto_data = ngtcp2_crypto_recv_crypto_data_cb, + .encrypt = ngtcp2_crypto_encrypt_cb, + .decrypt = ngtcp2_crypto_decrypt_cb, + .hp_mask = ngtcp2_crypto_hp_mask_cb, + .recv_stream_data = stream_data, + .rand = random_bytes, + .update_key = ngtcp2_crypto_update_key_cb, + .delete_crypto_aead_ctx = ngtcp2_crypto_delete_crypto_aead_ctx_cb, + .delete_crypto_cipher_ctx = ngtcp2_crypto_delete_crypto_cipher_ctx_cb, + .version_negotiation = ngtcp2_crypto_version_negotiation_cb, + .get_new_connection_id2 = new_cid, + .get_path_challenge_data2 = ngtcp2_crypto_get_path_challenge_data2_cb, + }; + if (server) c.recv_client_initial = ngtcp2_crypto_recv_client_initial_cb; + else { + c.client_initial = ngtcp2_crypto_client_initial_cb; + c.recv_retry = ngtcp2_crypto_recv_retry_cb; + c.extend_max_local_streams_bidi = streams_allowed; + } + return c; +} + +/* Windows wide enough that flow control never holds the drain back. */ +static void windows(ngtcp2_transport_params *p) { + ngtcp2_transport_params_default(p); + p->initial_max_streams_bidi = 4; + p->initial_max_stream_data_bidi_local = 16 << 20; + p->initial_max_stream_data_bidi_remote = 16 << 20; + p->initial_max_data = 64 << 20; + p->max_idle_timeout = 60 * NGTCP2_SECONDS; +} + +/* The client's next bytes at its offset: a pointer into what never changes. */ +static size_t next_bytes(Side *c, ngtcp2_vec *v) { + static const unsigned char head[4] = {'p', 'u', 'r', 'e'}, tail[4] = {0}; + uint64_t messages_end = 4 + c->count * c->period; + if (c->offset >= c->total) return 0; + if (c->offset < 4) v->base = (uint8_t *)head + c->offset, v->len = 4 - c->offset; + else if (c->offset < messages_end) { + uint64_t phase = (c->offset - 4) % c->period, left = messages_end - c->offset; + v->base = c->pattern + phase; + v->len = left < CHUNK ? left : CHUNK; + } else + v->base = (uint8_t *)tail + (c->offset - messages_end), v->len = c->total - c->offset; + return 1; +} + +static ngtcp2_ssize write_one(ngtcp2_conn *conn, ngtcp2_path *path, ngtcp2_pkt_info *pi, uint8_t *dest, size_t size, + ngtcp2_tstamp ts, void *user) { + Side *s = user; + for (;;) { + ngtcp2_vec v; + size_t count = 0; + int64_t stream = -1; + uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_MORE | NGTCP2_WRITE_STREAM_FLAG_PADDING; + if (!s->server && s->stream >= 0 && (count = next_bytes(s, &v))) { + stream = s->stream; + if (s->offset + v.len >= s->total) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN; + } else if (s->server && s->answered && s->read < 8) { /* the count, then the end */ + stream = s->stream, v.base = s->answer + s->read, v.len = 8 - s->read, count = 1; + flags |= NGTCP2_WRITE_STREAM_FLAG_FIN; + } + ngtcp2_ssize sent = 0; + ngtcp2_ssize n = ngtcp2_conn_writev_stream(conn, path, pi, dest, size, &sent, flags, stream, &v, count, ts); + if (sent > 0) { + if (s->server) s->read += (uint64_t)sent; + else s->offset += (uint64_t)sent; + } + if (n == NGTCP2_ERR_WRITE_MORE) continue; + if (n == NGTCP2_ERR_STREAM_DATA_BLOCKED || n == NGTCP2_ERR_STREAM_SHUT_WR) { + n = ngtcp2_conn_writev_stream(conn, path, pi, dest, size, NULL, NGTCP2_WRITE_STREAM_FLAG_PADDING, -1, NULL, + 0, ts); + } + return n < 0 ? NGTCP2_ERR_CALLBACK_FAILURE : n; + } +} + +static void send_burst(Side *s, const uint8_t *bytes, size_t n, size_t segment) { + _Alignas(8) unsigned char control[CMSG_SPACE(sizeof(uint16_t))] = {0}; + struct iovec part = {(void *)bytes, n}; + struct msghdr m = {.msg_name = &s->remote, .msg_namelen = s->remote_size, .msg_iov = &part, .msg_iovlen = 1}; + if (s->gso && n > segment) { + m.msg_control = control, m.msg_controllen = sizeof(control); + struct cmsghdr *c = CMSG_FIRSTHDR(&m); + c->cmsg_level = SOL_UDP, c->cmsg_type = UDP_SEGMENT, c->cmsg_len = CMSG_LEN(sizeof(uint16_t)); + uint16_t size = (uint16_t)segment; + memcpy(CMSG_DATA(c), &size, sizeof(size)); + if (sendmsg(s->fd, &m, 0) >= 0) return; + s->gso = false; /* no GSO here: one by one */ + m.msg_control = NULL, m.msg_controllen = 0; + } + for (size_t at = 0; at < n; at += segment) { + part.iov_base = (void *)(bytes + at); + part.iov_len = n - at < segment ? n - at : segment; + while (sendmsg(s->fd, &m, 0) < 0 && errno == EINTR) {} + } +} + +/* What may go now, in bursts. */ +static void write_out(Side *s) { + static uint8_t burst[BURST]; + for (int round = 0; round < 64; ++round) { + ngtcp2_path_storage ps; + ngtcp2_path_storage_zero(&ps); + ngtcp2_pkt_info pi; + size_t segment = 0; + ngtcp2_tstamp ts = now_ns(); + ngtcp2_ssize n = ngtcp2_conn_write_aggregate_pkt(s->conn, &ps.path, &pi, burst, sizeof(burst), &segment, + write_one, ts); + if (n < 0) fail(ngtcp2_strerror((int)n)); + if (!n) return; + send_burst(s, burst, (size_t)n, segment); + } +} + +static void read_in(Side *s) { + uint8_t bytes[65536]; + for (;;) { + struct sockaddr_storage from; + struct iovec part = {bytes, sizeof(bytes)}; + struct msghdr m = {.msg_name = &from, .msg_namelen = sizeof(from), .msg_iov = &part, .msg_iovlen = 1}; + ssize_t n = recvmsg(s->fd, &m, MSG_DONTWAIT); + if (n < 0) return; + if (!s->conn) { /* the service, a client's first packet */ + ngtcp2_pkt_hd hd; + int accepted = ngtcp2_accept(&hd, bytes, (size_t)n); + if (getenv("DRAIN_LOG")) fprintf(stderr, "first packet %zd bytes: %s\n", n, accepted ? ngtcp2_strerror(accepted) : "accepted"); + if (accepted) continue; + memcpy(&s->remote, &from, m.msg_namelen), s->remote_size = m.msg_namelen; + ngtcp2_path path = {.local = {(struct sockaddr *)&s->local, s->local_size}, + .remote = {(struct sockaddr *)&s->remote, s->remote_size}}; + ngtcp2_cid scid = {.datalen = 8}; + RAND_bytes(scid.data, 8); + ngtcp2_settings settings; + ngtcp2_settings_default(&settings); + settings.initial_ts = now_ns(); + if (getenv("DRAIN_LOG")) settings.log_printf = log_line; + ngtcp2_transport_params params; + windows(¶ms); + params.original_dcid = hd.dcid, params.original_dcid_present = 1; + ngtcp2_callbacks cb = callbacks(true); + if (ngtcp2_conn_server_new(&s->conn, &hd.scid, &scid, &path, hd.version, &cb, &settings, ¶ms, NULL, s)) + fail("a connection"); + ngtcp2_conn_set_tls_native_handle(s->conn, s->tls); + } + ngtcp2_path path = {.local = {(struct sockaddr *)&s->local, s->local_size}, + .remote = {(struct sockaddr *)&from, m.msg_namelen}}; + ngtcp2_pkt_info pi = {0}; + int rv = ngtcp2_conn_read_pkt(s->conn, &path, &pi, bytes, (size_t)n, now_ns()); + if (rv == NGTCP2_ERR_DRAINING || rv == NGTCP2_ERR_CLOSING) s->done = true; + else if (rv) fail(ngtcp2_strerror(rv)); + } +} + +static void run(Side *s) { + while (!s->done) { + if (s->conn) write_out(s); + ngtcp2_tstamp expiry = s->conn ? ngtcp2_conn_get_expiry2(s->conn) : UINT64_MAX, now = now_ns(); + int timeout = expiry == UINT64_MAX ? 1000 : expiry <= now ? 0 : (int)((expiry - now) / NGTCP2_MILLISECONDS) + 1; + struct pollfd wait = {s->fd, POLLIN, 0}; + if (poll(&wait, 1, timeout) < 0 && errno != EINTR) fail("poll"); + if (wait.revents & POLLIN) read_in(s); + if (s->conn && ngtcp2_conn_get_expiry2(s->conn) <= now_ns()) { + int rv = ngtcp2_conn_handle_expiry(s->conn, now_ns()); + if (rv == NGTCP2_ERR_IDLE_CLOSE) s->done = true; + else if (rv) fail(ngtcp2_strerror(rv)); + } + } +} + +static int alpn(SSL *ssl, const unsigned char **out, unsigned char *out_size, const unsigned char *in, unsigned in_size, + void *arg) { + (void)ssl, (void)arg; + for (unsigned at = 0; at + 1 + ALPN[0] <= in_size; at += 1u + in[at]) + if (!memcmp(in + at, ALPN, 1u + ALPN[0])) { + *out = in + at + 1, *out_size = in[at]; + return SSL_TLSEXT_ERR_OK; + } + return SSL_TLSEXT_ERR_ALERT_FATAL; +} + +static unsigned char *read_file(const char *path, long *size) { + FILE *f = fopen(path, "rb"); + if (!f || fseek(f, 0, SEEK_END) || (*size = ftell(f)) <= 0 || fseek(f, 0, SEEK_SET)) fail(path); + unsigned char *bytes = malloc((size_t)*size); + if (!bytes || fread(bytes, 1, (size_t)*size, f) != (size_t)*size) fail(path); + fclose(f); + return bytes; +} + +static void tls(Side *s, SSL_CTX *ctx) { + if (!(s->ssl = SSL_new(ctx)) || ngtcp2_crypto_ossl_ctx_new(&s->tls, s->ssl)) fail("TLS"); + if (s->server ? ngtcp2_crypto_ossl_configure_server_session(s->ssl) + : ngtcp2_crypto_ossl_configure_client_session(s->ssl)) + fail("TLS session"); + s->ref = (ngtcp2_crypto_conn_ref){conn_of, s}; + SSL_set_app_data(s->ssl, &s->ref); + if (s->server) SSL_set_accept_state(s->ssl); + else { + SSL_set_connect_state(s->ssl); + SSL_set_alpn_protos(s->ssl, ALPN, sizeof(ALPN) - 1); + SSL_set_tlsext_host_name(s->ssl, "localhost"); + } +} + +static _Noreturn void serve(const char *cert, const char *key, unsigned port) { + Side s = {.server = true, .stream = -1, .gso = !getenv("DRAIN_GSO") || strcmp(getenv("DRAIN_GSO"), "0")}; + long cert_size, key_size; + unsigned char *cert_der = read_file(cert, &cert_size), *key_der = read_file(key, &key_size); + const unsigned char *k = key_der; + EVP_PKEY *pkey = d2i_AutoPrivateKey(NULL, &k, key_size); + SSL_CTX *ctx = SSL_CTX_new(TLS_server_method()); + if (!ctx || !pkey || SSL_CTX_use_certificate_ASN1(ctx, (int)cert_size, cert_der) != 1 || + SSL_CTX_use_PrivateKey(ctx, pkey) != 1) + fail("the certificate"); + SSL_CTX_set_alpn_select_cb(ctx, alpn, NULL); + struct sockaddr_in at = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port), .sin_addr = {htonl(INADDR_LOOPBACK)}}; + s.fd = socket(AF_INET, SOCK_DGRAM, 0); + int buffer = 4 << 20; + setsockopt(s.fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer)); + setsockopt(s.fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer)); + s.local_size = sizeof(struct sockaddr_in); + if (bind(s.fd, (struct sockaddr *)&at, sizeof(at)) || getsockname(s.fd, (struct sockaddr *)&s.local, &s.local_size)) + fail("a port"); + printf("{\"port\":%u}\n", ntohs(((struct sockaddr_in *)&s.local)->sin_port)); + fflush(stdout); + for (;;) { /* a connection at a time, as long as it is let run */ + tls(&s, ctx); + run(&s); + ngtcp2_conn_del(s.conn); + ngtcp2_crypto_ossl_ctx_del(s.tls); + SSL_free(s.ssl); + Side next = {.server = true, .stream = -1, .gso = s.gso, .fd = s.fd, .local = s.local, .local_size = s.local_size}; + s = next; + } +} + +static int drain(unsigned port, uint64_t count, uint64_t size) { + Side c = {.stream = -1, .count = count, .size = size, .gso = !getenv("DRAIN_GSO") || strcmp(getenv("DRAIN_GSO"), "0")}; + c.period = 4 + size; + c.total = 4 + count * c.period + 4; + size_t periods = CHUNK / c.period + 2; + if (!(c.pattern = malloc(periods * c.period))) fail("memory"); + for (size_t p = 0; p < periods; ++p) { + unsigned char *m = c.pattern + p * c.period; + for (int b = 0; b < 4; ++b) m[b] = (unsigned char)(size >> 8 * b); + memset(m + 4, 'x', size); + } + SSL_CTX *ctx = SSL_CTX_new(TLS_client_method()); + if (!ctx) fail("TLS"); + SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL); /* as quinn's client: no certificate checked */ + tls(&c, ctx); + struct sockaddr_in to = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port), .sin_addr = {htonl(INADDR_LOOPBACK)}}; + c.fd = socket(AF_INET, SOCK_DGRAM, 0); + int buffer = 4 << 20; + setsockopt(c.fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer)); + setsockopt(c.fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer)); + memcpy(&c.remote, &to, sizeof(to)), c.remote_size = sizeof(to); + c.local_size = sizeof(c.local); + if (connect(c.fd, (struct sockaddr *)&to, sizeof(to)) || getsockname(c.fd, (struct sockaddr *)&c.local, &c.local_size)) + fail("a socket"); + ngtcp2_path path = {.local = {(struct sockaddr *)&c.local, c.local_size}, + .remote = {(struct sockaddr *)&c.remote, c.remote_size}}; + ngtcp2_cid dcid = {.datalen = NGTCP2_MIN_INITIAL_DCIDLEN}, scid = {.datalen = 8}; + RAND_bytes(dcid.data, (int)dcid.datalen), RAND_bytes(scid.data, (int)scid.datalen); + ngtcp2_settings settings; + ngtcp2_settings_default(&settings); + settings.initial_ts = now_ns(); + if (getenv("DRAIN_LOG")) settings.log_printf = log_line; + ngtcp2_transport_params params; + windows(¶ms); + ngtcp2_callbacks cb = callbacks(false); + if (ngtcp2_conn_client_new(&c.conn, &dcid, &scid, &path, NGTCP2_PROTO_VER_V1, &cb, &settings, ¶ms, NULL, &c)) + fail("a connection"); + ngtcp2_conn_set_tls_native_handle(c.conn, c.tls); + uint64_t began = now_ns(); + run(&c); + double seconds = (double)(now_ns() - began) / 1e9; + uint64_t read = 0; + for (int b = 7; b >= 0; --b) read = read << 8 | c.answer[b]; + printf("{\"messages\":%llu,\"size\":%llu,\"read\":%llu,\"seconds\":%.3f,\"per_second\":%.0f}\n", + (unsigned long long)count, (unsigned long long)size, (unsigned long long)read, seconds, (double)count / seconds); + ngtcp2_path_storage ps; + ngtcp2_path_storage_zero(&ps); + ngtcp2_pkt_info pi; + uint8_t close[1280]; + ngtcp2_ccerr_default(&c.error); + ngtcp2_ssize n = ngtcp2_conn_write_connection_close(c.conn, &ps.path, &pi, close, sizeof(close), &c.error, now_ns()); + if (n > 0) send(c.fd, close, (size_t)n, 0); + return read == count ? 0 : 1; +} + +int main(int argc, char **argv) { + if (ngtcp2_crypto_ossl_init()) fail("ngtcp2's OpenSSL"); + if (argc >= 4 && argc <= 5 && !strcmp(argv[1], "serve")) + serve(argv[2], argv[3], argc == 5 ? (unsigned)strtoul(argv[4], NULL, 10) : 0); + if (argc == 5 && !strcmp(argv[1], "drain")) + return drain((unsigned)strtoul(argv[2], NULL, 10), strtoull(argv[3], NULL, 10), strtoull(argv[4], NULL, 10)); + fputs("drain serve CERT.der KEY.der [PORT] | drain drain PORT COUNT SIZE\n", stderr); + return 2; +} diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py index 7f2647d..0d18c1d 100644 --- a/research/experiments/019-idle-links/run.py +++ b/research/experiments/019-idle-links/run.py @@ -543,6 +543,26 @@ def drain_quic(work, n, size): return row +def drain_ngtcp2(work, n, size): + """The same over ngtcp2 (ngtcp2/drain.c, OpenSSL 3.5's QUIC TLS): one + connection, one stream, n messages; CPU of both sides.""" + import resource + binary = ROOT / 'build/third/ngtcp2-drain' + server, info = start([binary, 'serve', work / 'cert.der', work / 'key.der']) + before, children = thread_cpu(server.pid, None), resource.getrusage(resource.RUSAGE_CHILDREN) + 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 |