Put ngtcp2 beside quinn in M6, as it is in M5

drain.c takes a rate and seconds as quinn's client does (paced messages
stamped with the time they were written, and no more written past the
seconds), and its service writes quinn's line for each stream: the
messages read in each 100 ms and how long they took. 020's run.py runs
it as a third, on port 40005 behind the same bottleneck, and --who
picks whom to run.

2 files changed, 210 insertions, 66 deletions

diff --git a/research/experiments/019-idle-links/ngtcp2/drain.c b/research/experiments/019-idle-links/ngtcp2/drain.c
index 43942e7..b376255 100644
--- a/research/experiments/019-idle-links/ngtcp2/drain.c
+++ b/research/experiments/019-idle-links/ngtcp2/drain.c
@@ -1,15 +1,21 @@
-/* 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.
+/* M5 and M6 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
+ *   drain drain PORT COUNT SIZE [RATE [SECONDS]]
+ *                                         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. */
+ * the messages (no store) and answers the count (u64). RATE: messages a
+ * second (0: as fast as it may), each with the time it was written (ns
+ * since the epoch) in its first 8 bytes, after "pace"; SECONDS: no more
+ * written after that long. The service writes a line of JSON for each
+ * stream, as quinn's does: the messages read in each 100 ms, and (paced)
+ * how long they took. 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>
@@ -46,6 +52,12 @@ static uint64_t now_ns(void) {
     return (uint64_t)t.tv_sec * NGTCP2_SECONDS + (uint64_t)t.tv_nsec;
 }
 
+static uint64_t epoch_ns(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_REALTIME, &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);
@@ -63,19 +75,28 @@ typedef struct {
     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. */
+     * "pure" (or "pace"), then the messages up to end (a pattern of one
+     * period, or those written so far when paced), then a length of 0 once
+     * there are no more (the stream's total). */
     int64_t stream;
-    uint64_t count, size, offset, total;
-    unsigned char *pattern;
+    uint64_t count, size, offset, total, end;
+    unsigned char *pattern, *paced;
     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;
+    double rate, seconds;
+    uint64_t began, written;
+    bool finished;
+    /* The service's count, where it is in a message (and its stamp, when
+     * paced), the messages read in each 100 ms and how long each took. */
+    uint64_t messages, need, skip, stream_began;
+    unsigned char length[4], head[4], stamp[8];
+    unsigned have, stamped;
+    bool header, pacing, answered;
     unsigned char answer[8];
     uint64_t read;
+    uint32_t *timeline;
+    size_t buckets;
+    uint64_t *waited;
+    size_t waits, waits_capacity;
 } Side;
 
 static void log_line(void *user, const char *format, ...) {
@@ -102,16 +123,55 @@ static int new_cid(ngtcp2_conn *conn, ngtcp2_cid *cid, ngtcp2_stateless_reset_to
     return 0;
 }
 
-/* The client: its stream opened once the service allows one. */
+/* The client: its stream opened once the service allows one; its messages
+ * are timed from then. */
 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;
+    if (c->stream >= 0 && !c->began) c->began = now_ns();
     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 by_value(const void *a, const void *b) {
+    uint64_t x = *(const uint64_t *)a, y = *(const uint64_t *)b;
+    return x < y ? -1 : x > y;
+}
+
+/* The service's line for a stream: as quinn's. */
+static void report(Side *s) {
+    qsort(s->waited, s->waits, sizeof(*s->waited), by_value);
+    double at[2] = {0, 0};
+    for (int q = 0; q < 2 && s->waits; ++q) {
+        size_t k = s->waits * (q ? 99 : 50) / 100;
+        at[q] = (double)s->waited[k < s->waits ? k : s->waits - 1] / 1e6;
+    }
+    printf("{\"drained\":%llu,\"waited_ms_p50\":%.3f,\"waited_ms_p99\":%.3f,\"waited_ms_max\":%.3f,\"timeline_100ms\":[",
+           (unsigned long long)s->messages, at[0], at[1], s->waits ? (double)s->waited[s->waits - 1] / 1e6 : 0.0);
+    size_t last = s->buckets; /* to the last with a message, as quinn's */
+    while (last && !s->timeline[last - 1]) --last;
+    for (size_t b = 0; b < last; ++b) printf("%s%u", b ? "," : "", s->timeline[b]);
+    printf("]}\n");
+    fflush(stdout);
+}
+
+/* A message read whole: counted in its 100 ms. */
+static void message_read(Side *s) {
+    size_t bucket = (size_t)((now_ns() - s->stream_began) / (100 * NGTCP2_MILLISECONDS));
+    if (bucket >= s->buckets) {
+        size_t buckets = bucket + 64;
+        uint32_t *grown = realloc(s->timeline, buckets * sizeof(*grown));
+        if (!grown) fail("memory");
+        memset(grown + s->buckets, 0, (buckets - s->buckets) * sizeof(*grown));
+        s->timeline = grown, s->buckets = buckets;
+    }
+    ++s->timeline[bucket];
+    ++s->messages;
+}
+
+/* The service: messages counted as they come, over any cut (with the time
+ * each was written, when paced); 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;
@@ -120,14 +180,33 @@ static int stream_data(ngtcp2_conn *conn, uint32_t flags, int64_t stream, uint64
         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;
+        for (size_t i = 0; i < n && !s->answered;) {
+            if (!s->header) { /* "pure" or "pace" */
+                s->head[s->have++] = data[i++];
+                if (s->have < 4) continue;
+                s->header = true, s->have = 0;
+                s->pacing = !memcmp(s->head, "pace", 4);
+                s->stream_began = now_ns();
             } else if (s->skip) {
-                size_t take = n - i < s->skip ? n - i : (size_t)s->skip;
-                s->skip -= take;
-                i += take;
+                if (s->pacing && s->stamped < 8) { /* the time it was written */
+                    s->stamp[s->stamped++] = data[i++];
+                    --s->skip;
+                    if (s->stamped == 8) {
+                        uint64_t stamp = 0;
+                        for (int b = 7; b >= 0; --b) stamp = stamp << 8 | s->stamp[b];
+                        if (s->waits == s->waits_capacity) {
+                            s->waits_capacity = s->waits_capacity ? 2 * s->waits_capacity : 1024;
+                            if (!(s->waited = realloc(s->waited, s->waits_capacity * sizeof(*s->waited)))) fail("memory");
+                        }
+                        uint64_t now = epoch_ns();
+                        s->waited[s->waits++] = now > stamp ? now - stamp : 0;
+                    }
+                } else {
+                    size_t take = n - i < s->skip ? n - i : (size_t)s->skip;
+                    s->skip -= take;
+                    i += take;
+                }
+                if (!s->skip) message_read(s);
             } else {
                 s->length[s->have++] = data[i++];
                 if (s->have < 4) continue;
@@ -138,10 +217,10 @@ static int stream_data(ngtcp2_conn *conn, uint32_t flags, int64_t stream, uint64
                     s->answered = true;
                     s->stream = stream;
                     for (int b = 0; b < 8; ++b) s->answer[b] = (unsigned char)(s->messages >> 8 * b);
+                    report(s);
                     break;
                 }
-                ++s->messages;
-                s->skip = length;
+                s->skip = length, s->stamped = 0;
             }
         }
     ngtcp2_conn_extend_max_stream_offset(conn, stream, n);
@@ -185,19 +264,46 @@ static void windows(ngtcp2_transport_params *p) {
 
 /* 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;
+    static const unsigned char pure[4] = {'p', 'u', 'r', 'e'}, pace[4] = {'p', 'a', 'c', 'e'}, tail[4] = {0};
     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;
+    if (c->offset < 4) v->base = (uint8_t *)(c->rate > 0 ? pace : pure) + c->offset, v->len = 4 - c->offset;
+    else if (c->offset < c->end) {
+        uint64_t left = c->end - c->offset;
+        v->base = c->rate > 0 ? c->paced + (c->offset - 4) : c->pattern + (c->offset - 4) % c->period;
         v->len = left < CHUNK ? left : CHUNK;
     } else
-        v->base = (uint8_t *)tail + (c->offset - messages_end), v->len = c->total - c->offset;
+        v->base = (uint8_t *)tail + (c->offset - c->end), v->len = c->total - c->offset;
     return 1;
 }
 
+/* The client's messages to now: paced, those due written (stamped); then,
+ * past its seconds (or all written), the stream ends after the message it
+ * is in. When the next is due (ns), or 0. */
+static uint64_t write_due(Side *c) {
+    if (!c->began || c->finished) return 0;
+    uint64_t now = now_ns(), limit = c->seconds > 0 ? c->began + (uint64_t)(c->seconds * 1e9) : UINT64_MAX;
+    if (c->rate > 0) {
+        for (;;) {
+            uint64_t due = c->began + (uint64_t)((double)c->written / c->rate * 1e9);
+            if (c->written == c->count || due >= limit) break;
+            if (due > now) return due;
+            unsigned char *m = c->paced + c->written * c->period, *stamp = m + 4;
+            uint64_t time = epoch_ns();
+            for (int b = 0; b < 8; ++b) stamp[b] = (unsigned char)(time >> 8 * b);
+            ++c->written;
+            c->end = c->total = 4 + c->written * c->period;
+        }
+    } else if (now < limit)
+        return limit;
+    else if (c->offset > 4) /* the message it is in, whole */
+        c->end = 4 + ((c->offset - 4 + c->period - 1) / c->period) * c->period;
+    else
+        c->end = 4;
+    c->total = c->end + 4;
+    c->finished = true;
+    return 0;
+}
+
 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;
@@ -208,7 +314,7 @@ static ngtcp2_ssize write_one(ngtcp2_conn *conn, ngtcp2_path *path, ngtcp2_pkt_i
         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;
+            if (s->finished && 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;
@@ -307,8 +413,10 @@ static void read_in(Side *s) {
 
 static void run(Side *s) {
     while (!s->done) {
+        uint64_t due = s->server ? 0 : write_due(s);
         if (s->conn) write_out(s);
         ngtcp2_tstamp expiry = s->conn ? ngtcp2_conn_get_expiry2(s->conn) : UINT64_MAX, now = now_ns();
+        if (due && due < expiry) expiry = due;
         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");
@@ -383,15 +491,30 @@ static _Noreturn void serve(const char *cert, const char *key, unsigned port) {
         ngtcp2_conn_del(s.conn);
         ngtcp2_crypto_ossl_ctx_del(s.tls);
         SSL_free(s.ssl);
+        free(s.timeline);
+        free(s.waited);
         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")};
+static int drain(unsigned port, uint64_t count, uint64_t size, double rate, double seconds) {
+    Side c = {.stream = -1, .count = count, .size = size, .rate = rate, .seconds = seconds,
+              .gso = !getenv("DRAIN_GSO") || strcmp(getenv("DRAIN_GSO"), "0")};
     c.period = 4 + size;
-    c.total = 4 + count * c.period + 4;
+    c.end = 4 + count * c.period;
+    c.total = c.end + 4;
+    c.finished = rate <= 0 && seconds <= 0;
+    if (rate > 0) { /* the messages written as they are due: room for them all, none yet */
+        if (seconds > 0 && (double)count > rate * seconds + 1) c.count = count = (uint64_t)(rate * seconds) + 1;
+        if (size < 8 || !(c.paced = malloc(count * c.period))) fail("paced messages of 8 bytes or more, in memory");
+        for (uint64_t k = 0; k < count; ++k) {
+            unsigned char *m = c.paced + k * c.period;
+            for (int b = 0; b < 4; ++b) m[b] = (unsigned char)(size >> 8 * b);
+            memset(m + 4, 'x', size);
+        }
+        c.end = c.total = 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) {
@@ -428,11 +551,12 @@ static int drain(unsigned port, uint64_t count, uint64_t size) {
     ngtcp2_conn_set_tls_native_handle(c.conn, c.tls);
     uint64_t began = now_ns();
     run(&c);
-    double seconds = (double)(now_ns() - began) / 1e9;
+    double took = (double)(now_ns() - began) / 1e9;
     uint64_t read = 0;
     for (int b = 7; b >= 0; --b) read = read << 8 | c.answer[b];
+    uint64_t sent = c.rate > 0 ? c.written : (c.end - 4) / c.period;
     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);
+           (unsigned long long)sent, (unsigned long long)size, (unsigned long long)read, took, (double)sent / took);
     ngtcp2_path_storage ps;
     ngtcp2_path_storage_zero(&ps);
     ngtcp2_pkt_info pi;
@@ -440,15 +564,16 @@ static int drain(unsigned port, uint64_t count, uint64_t size) {
     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;
+    return read == sent ? 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);
+    if (argc >= 5 && argc <= 7 && !strcmp(argv[1], "drain"))
+        return drain((unsigned)strtoul(argv[2], NULL, 10), strtoull(argv[3], NULL, 10), strtoull(argv[4], NULL, 10),
+                     argc > 5 ? strtod(argv[5], NULL) : 0, argc > 6 ? strtod(argv[6], NULL) : 0);
+    fputs("drain serve CERT.der KEY.der [PORT] | drain drain PORT COUNT SIZE [RATE [SECONDS]]\n", stderr);
     return 2;
 }
diff --git a/research/experiments/020-lossy-paths/run.py b/research/experiments/020-lossy-paths/run.py
index 9dbb53f..542360a 100644
--- a/research/experiments/020-lossy-paths/run.py
+++ b/research/experiments/020-lossy-paths/run.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 """M6 of docs/design/011-pure-transport.md: a link on a path of delay, loss,
-reordering and a bottleneck, against quinn, and beside TCP CUBIC.
+reordering and a bottleneck, against quinn and ngtcp2, and beside TCP CUBIC.
 
 Each case runs in `unshare -rn` (an unprivileged network namespace) whose
 loopback is the path: MTU 1500, and TCP's segments no larger
@@ -8,15 +8,15 @@ loopback is the path: MTU 1500, and TCP's segments no larger
 RTT), jitter (packets reorder) and random loss; the way to the receivers
 (ports 40000-40007) then goes through a 50 Mbit/s bottleneck (tbf) whose
 queue holds a round trip of it, and the way back (answers, TCP's
-acknowledgments) does not queue behind it. Pure and quinn send without UDP
-GSO: a qdisc takes a GSO burst as one packet.
+acknowledgments) does not queue behind it. Pure and the QUICs send without
+UDP GSO: a qdisc takes a GSO burst as one packet.
 
   bulk   a backlog of 1 KiB records for --seconds (Pure: a link and a
-         listener, transport.c; quinn: one stream); the goodput from the
-         receiver's records in each 100 ms, the first 2 s left out.
+         listener, transport.c; quinn and ngtcp2: one stream); the goodput
+         from the receiver's records in each 100 ms, the first 2 s left out.
   paced  50 records a second for --seconds (well within what each carries); the time each took from
          queued (written) to handed over (read): p50, p99, max.
-  share  Pure, quinn or a second TCP flow beside a TCP CUBIC flow, each a
+  share  Pure, quinn, ngtcp2 or a second TCP flow beside a TCP CUBIC flow, each a
          process of its own, started together, for --share-seconds; each
          one's share of the goodput from 20 s on (CUBIC's flows converge
          slowly at an RTT of 100 ms). The path's delay wavers by 1 ms: with
@@ -24,7 +24,7 @@ GSO: a qdisc takes a GSO burst as one packet.
          than the other's by their phases alone (Floyd and Jacobson, 1992),
          and which flow that is says nothing of either.
 
-    python3 research/experiments/020-lossy-paths/run.py [--rounds 3] [--out FILE]
+    python3 research/experiments/020-lossy-paths/run.py [--rounds 3] [--who pure,ngtcp2] [--out FILE]
 """
 import argparse
 import json
@@ -42,15 +42,16 @@ from pathlib import Path
 ROOT = Path(__file__).resolve().parents[3]
 HERE = Path(__file__).resolve().parent
 QUINN = ROOT / 'research/experiments/019-idle-links/quic/target/release/idle-quic'
+NGTCP2 = ROOT / 'build/third/ngtcp2-drain'
 TC = shutil.which('tc') or '/usr/sbin/tc'
 IP = shutil.which('ip') or '/usr/sbin/ip'
 RATE_MBIT, SIZE = 50, 1000
-PURE_PORT, QUINN_PORT, TCP_PORT, OTHER_TCP_PORT = 40001, 40002, 40003, 40004  # the receivers, beyond the bottleneck
+PURE_PORT, QUINN_PORT, TCP_PORT, OTHER_TCP_PORT, NGTCP2_PORT = 40001, 40002, 40003, 40004, 40005  # the receivers, beyond the bottleneck
 sys.path.insert(0, str(ROOT / 'tests'))
 
 
 def build(work):
-    """The transport harness, quinn's, the two modules and a certificate."""
+    """The transport harness, quinn's, ngtcp2's, the two modules and a certificate."""
     import check
     import importlib.util
     spec = importlib.util.spec_from_file_location('idle_links', ROOT / 'research/experiments/019-idle-links/run.py')
@@ -63,6 +64,7 @@ def build(work):
                     str(ROOT / 'build/libpure.a'), '-lm', '-pthread', '-ldl', '-o', str(ROOT / 'build/link-transport')],
                    check=True)
     subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=QUINN.parents[2], check=True)
+    subprocess.run(['sh', str(ROOT / 'research/experiments/019-idle-links/ngtcp2/build.sh')], check=True)
     for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)):
         (work / f'{name}.aup').write_text(text)
         check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), work / f'{name}.aup',
@@ -144,8 +146,20 @@ def finish_quinn(started, seconds):
     return json.loads(line)
 
 
-def quinn(work, seconds, rate=0):
-    return finish_quinn(start_quinn(work, seconds, rate), seconds)
+def start_ngtcp2(work, seconds, rate=0, count=2000000):
+    """As quinn's: the same messages over ngtcp2 (019's ngtcp2/drain.c)."""
+    env = {**os.environ, 'DRAIN_GSO': '0'}
+    server = subprocess.Popen([str(NGTCP2), 'serve', str(work / 'cert.der'), str(work / 'key.der'), str(NGTCP2_PORT)],
+                              stdout=subprocess.PIPE, text=True, env=env)
+    port = json.loads(server.stdout.readline())['port']
+    client = subprocess.Popen([str(NGTCP2), 'drain', str(port), str(count), str(SIZE + 8 + 5), str(rate), str(seconds)],
+                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
+    return server, client
+
+
+def quic(who, work, seconds, rate=0):
+    started = (start_quinn if who == 'quinn' else start_ngtcp2)(work, seconds, rate)
+    return finish_quinn(started, seconds)
 
 
 def start_tcp(port, seconds):
@@ -216,7 +230,7 @@ def inner(case):
     del row['work']
     if kind in ('bulk', 'paced'):
         rate = 50 if kind == 'paced' else 0
-        result = pure(work, seconds, rate) if who == 'pure' else quinn(work, seconds, rate)
+        result = pure(work, seconds, rate) if who == 'pure' else quic(who, work, seconds, rate)
         timeline = result['timeline_100ms']
         row['mbit'] = goodput(timeline, SIZE + 13, 2.0, seconds)
         for key in ('waited_ms_p50', 'waited_ms_p99', 'waited_ms_max', 'resent', 'lost', 'late', 'undone', 'packets'):
@@ -229,7 +243,7 @@ def inner(case):
         elif who == 'pure':
             theirs = finish_pure(start_pure(work, seconds), seconds)['timeline_100ms']
         else:
-            theirs = finish_quinn(start_quinn(work, seconds), seconds)['timeline_100ms']
+            theirs = finish_quinn((start_quinn if who == 'quinn' else start_ngtcp2)(work, seconds), seconds)['timeline_100ms']
         mine = finish_tcp(tcp, seconds)
         row['mbit'] = goodput(theirs, SIZE + 13, 20.0, seconds - 1)
         row['tcp_mbit'] = goodput(mine, 1, 20.0, seconds - 1)
@@ -241,17 +255,21 @@ def inner(case):
 # ---- Outside
 
 
-def cases(seconds, share_seconds):
+def cases(seconds, share_seconds, whom=('pure', 'quinn', 'ngtcp2', 'tcp')):
     for rtt in (20, 100):
         for loss in (0, 1, 5):
             for jitter in (0, 3):
-                for who in ('pure', 'quinn'):
-                    yield {'kind': 'bulk', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': jitter, 'seconds': seconds}
+                for who in ('pure', 'quinn', 'ngtcp2'):
+                    if who in whom:
+                        yield {'kind': 'bulk', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': jitter,
+                               'seconds': seconds}
         for loss in (0, 1, 5):
-            for who in ('pure', 'quinn'):
-                yield {'kind': 'paced', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': 0, 'seconds': seconds}
-        for who in ('pure', 'quinn', 'tcp'):
-            yield {'kind': 'share', 'who': who, 'rtt_ms': rtt, 'loss': 0, 'jitter_ms': 1, 'seconds': share_seconds}
+            for who in ('pure', 'quinn', 'ngtcp2'):
+                if who in whom:
+                    yield {'kind': 'paced', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': 0, 'seconds': seconds}
+        for who in ('pure', 'quinn', 'ngtcp2', 'tcp'):
+            if who in whom:
+                yield {'kind': 'share', 'who': who, 'rtt_ms': rtt, 'loss': 0, 'jitter_ms': 1, 'seconds': share_seconds}
 
 
 def main():
@@ -260,6 +278,7 @@ def main():
     parser.add_argument('--seconds', type=int, default=20)
     parser.add_argument('--share-seconds', type=int, default=60)
     parser.add_argument('--only', default=None, help='kinds to run, for example bulk,share')
+    parser.add_argument('--who', default='pure,quinn,ngtcp2,tcp', help='whom to run, for example pure,ngtcp2')
     parser.add_argument('--out', type=Path, default=None)
     parser.add_argument('--inner', default=None, help=argparse.SUPPRESS)
     parser.add_argument('--tcp', default=None, help=argparse.SUPPRESS)
@@ -278,7 +297,7 @@ def main():
         build(work)
         only = set(args.only.split(',')) if args.only else None
         for round_ in range(args.rounds):
-            for case in cases(args.seconds, args.share_seconds):
+            for case in cases(args.seconds, args.share_seconds, tuple(args.who.split(','))):
                 if only and case['kind'] not in only:
                     continue
                 case = {**case, 'round': round_, 'work': str(work)}