Measure one sending loop against a thread a link, with and without push (028)

Five rounds of 1,000 bursts by one link, four links and one wire, four
builds in turns: a thread a link (966b82f), those threads with push
(threads-push.patch), the loop (99b2219), the loop with push (94e7bb8).
Medians: one link 597k, 923k, 525k, 897k sends/s; four links 506k,
799k, 268k, 800k; the wire as it was, 789k to 794k.

7 files changed, 338 insertions, 0 deletions

diff --git a/research/experiments/028-one-sending-loop/2026-10-01-links-linux-aarch64.jsonl b/research/experiments/028-one-sending-loop/2026-10-01-links-linux-aarch64.jsonl
new file mode 100644
index 0000000..7c7fa54
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/2026-10-01-links-linux-aarch64.jsonl
@@ -0,0 +1,60 @@
+{"round": 1, "version": "before", "way": "link", "sends": 1000000, "seconds": 1.88, "sends_per_second": 531822, "cpu_us_per_send": 3.24, "threads": 5, "packets": 38836, "load": 0.98}
+{"round": 1, "version": "threads-push", "way": "link", "sends": 1000000, "seconds": 1.083, "sends_per_second": 923011, "cpu_us_per_send": 1.91, "threads": 5, "packets": 16655, "load": 0.98}
+{"round": 1, "version": "loop", "way": "link", "sends": 1000000, "seconds": 1.904, "sends_per_second": 525093, "cpu_us_per_send": 3.29, "threads": 5, "packets": 41802, "load": 0.98}
+{"round": 1, "version": "push", "way": "link", "sends": 1000000, "seconds": 1.153, "sends_per_second": 867612, "cpu_us_per_send": 1.93, "threads": 5, "packets": 16533, "load": 1.22}
+{"round": 1, "version": "before", "way": "links4", "sends": 4000000, "seconds": 7.369, "sends_per_second": 542778, "cpu_us_per_send": 3.26, "threads": 8, "packets": 142572, "load": 1.22}
+{"round": 1, "version": "threads-push", "way": "links4", "sends": 4000000, "seconds": 5.009, "sends_per_second": 798600, "cpu_us_per_send": 2.26, "threads": 8, "packets": 64269, "load": 1.56}
+{"round": 1, "version": "loop", "way": "links4", "sends": 4000000, "seconds": 15.733, "sends_per_second": 254244, "cpu_us_per_send": 7.3, "threads": 5, "packets": 450723, "load": 1.68}
+{"round": 1, "version": "push", "way": "links4", "sends": 4000000, "seconds": 4.998, "sends_per_second": 800265, "cpu_us_per_send": 2.27, "threads": 5, "packets": 64064, "load": 1.98}
+{"round": 1, "version": "before", "way": "wire", "sends": 1000000, "seconds": 1.367, "sends_per_second": 731746, "cpu_us_per_send": 2.35, "threads": 4, "packets": 0, "load": 2.06}
+{"round": 1, "version": "threads-push", "way": "wire", "sends": 1000000, "seconds": 1.295, "sends_per_second": 772398, "cpu_us_per_send": 2.29, "threads": 4, "packets": 0, "load": 2.06}
+{"round": 1, "version": "loop", "way": "wire", "sends": 1000000, "seconds": 1.369, "sends_per_second": 730324, "cpu_us_per_send": 2.34, "threads": 4, "packets": 0, "load": 2.06}
+{"round": 1, "version": "push", "way": "wire", "sends": 1000000, "seconds": 1.354, "sends_per_second": 738455, "cpu_us_per_send": 2.3, "threads": 4, "packets": 0, "load": 2.21}
+{"round": 2, "version": "threads-push", "way": "link", "sends": 1000000, "seconds": 1.127, "sends_per_second": 887633, "cpu_us_per_send": 1.91, "threads": 5, "packets": 16556, "load": 2.21}
+{"round": 2, "version": "loop", "way": "link", "sends": 1000000, "seconds": 1.803, "sends_per_second": 554743, "cpu_us_per_send": 3.12, "threads": 5, "packets": 37365, "load": 2.21}
+{"round": 2, "version": "push", "way": "link", "sends": 1000000, "seconds": 1.053, "sends_per_second": 949220, "cpu_us_per_send": 1.88, "threads": 5, "packets": 16575, "load": 2.12}
+{"round": 2, "version": "before", "way": "link", "sends": 1000000, "seconds": 1.686, "sends_per_second": 593221, "cpu_us_per_send": 2.96, "threads": 5, "packets": 34215, "load": 2.12}
+{"round": 2, "version": "threads-push", "way": "links4", "sends": 4000000, "seconds": 4.95, "sends_per_second": 808043, "cpu_us_per_send": 2.23, "threads": 8, "packets": 64258, "load": 2.12}
+{"round": 2, "version": "loop", "way": "links4", "sends": 4000000, "seconds": 14.94, "sends_per_second": 267735, "cpu_us_per_send": 6.84, "threads": 5, "packets": 426383, "load": 2.59}
+{"round": 2, "version": "push", "way": "links4", "sends": 4000000, "seconds": 5.042, "sends_per_second": 793260, "cpu_us_per_send": 2.27, "threads": 5, "packets": 64078, "load": 2.75}
+{"round": 2, "version": "before", "way": "links4", "sends": 4000000, "seconds": 6.869, "sends_per_second": 582333, "cpu_us_per_send": 3.05, "threads": 8, "packets": 132057, "load": 2.61}
+{"round": 2, "version": "threads-push", "way": "wire", "sends": 1000000, "seconds": 1.254, "sends_per_second": 797526, "cpu_us_per_send": 2.23, "threads": 4, "packets": 0, "load": 2.69}
+{"round": 2, "version": "loop", "way": "wire", "sends": 1000000, "seconds": 1.28, "sends_per_second": 781221, "cpu_us_per_send": 2.25, "threads": 4, "packets": 0, "load": 2.69}
+{"round": 2, "version": "push", "way": "wire", "sends": 1000000, "seconds": 1.244, "sends_per_second": 804107, "cpu_us_per_send": 2.23, "threads": 4, "packets": 0, "load": 2.69}
+{"round": 2, "version": "before", "way": "wire", "sends": 1000000, "seconds": 1.257, "sends_per_second": 795259, "cpu_us_per_send": 2.23, "threads": 4, "packets": 0, "load": 2.63}
+{"round": 3, "version": "loop", "way": "link", "sends": 1000000, "seconds": 1.735, "sends_per_second": 576408, "cpu_us_per_send": 3.09, "threads": 5, "packets": 39367, "load": 2.63}
+{"round": 3, "version": "push", "way": "link", "sends": 1000000, "seconds": 1.115, "sends_per_second": 896986, "cpu_us_per_send": 1.91, "threads": 5, "packets": 16481, "load": 2.63}
+{"round": 3, "version": "before", "way": "link", "sends": 1000000, "seconds": 1.597, "sends_per_second": 626191, "cpu_us_per_send": 2.81, "threads": 5, "packets": 33474, "load": 2.66}
+{"round": 3, "version": "threads-push", "way": "link", "sends": 1000000, "seconds": 1.052, "sends_per_second": 950572, "cpu_us_per_send": 1.86, "threads": 5, "packets": 16631, "load": 2.66}
+{"round": 3, "version": "loop", "way": "links4", "sends": 4000000, "seconds": 10.343, "sends_per_second": 386728, "cpu_us_per_send": 4.79, "threads": 5, "packets": 281303, "load": 2.66}
+{"round": 3, "version": "push", "way": "links4", "sends": 4000000, "seconds": 4.898, "sends_per_second": 816731, "cpu_us_per_send": 2.17, "threads": 5, "packets": 64041, "load": 2.63}
+{"round": 3, "version": "before", "way": "links4", "sends": 4000000, "seconds": 7.902, "sends_per_second": 506173, "cpu_us_per_send": 3.58, "threads": 8, "packets": 178829, "load": 2.66}
+{"round": 3, "version": "threads-push", "way": "links4", "sends": 4000000, "seconds": 5.065, "sends_per_second": 789711, "cpu_us_per_send": 2.26, "threads": 8, "packets": 64157, "load": 2.48}
+{"round": 3, "version": "loop", "way": "wire", "sends": 1000000, "seconds": 1.338, "sends_per_second": 747207, "cpu_us_per_send": 2.32, "threads": 4, "packets": 0, "load": 2.52}
+{"round": 3, "version": "push", "way": "wire", "sends": 1000000, "seconds": 1.452, "sends_per_second": 688484, "cpu_us_per_send": 2.4, "threads": 4, "packets": 0, "load": 2.52}
+{"round": 3, "version": "before", "way": "wire", "sends": 1000000, "seconds": 1.268, "sends_per_second": 788620, "cpu_us_per_send": 2.25, "threads": 4, "packets": 0, "load": 2.72}
+{"round": 3, "version": "threads-push", "way": "wire", "sends": 1000000, "seconds": 1.251, "sends_per_second": 799460, "cpu_us_per_send": 2.23, "threads": 4, "packets": 0, "load": 2.72}
+{"round": 4, "version": "push", "way": "link", "sends": 1000000, "seconds": 1.126, "sends_per_second": 888061, "cpu_us_per_send": 1.94, "threads": 5, "packets": 16594, "load": 2.72}
+{"round": 4, "version": "before", "way": "link", "sends": 1000000, "seconds": 1.676, "sends_per_second": 596795, "cpu_us_per_send": 2.92, "threads": 5, "packets": 33997, "load": 2.72}
+{"round": 4, "version": "threads-push", "way": "link", "sends": 1000000, "seconds": 1.09, "sends_per_second": 917379, "cpu_us_per_send": 1.93, "threads": 5, "packets": 16683, "load": 2.74}
+{"round": 4, "version": "loop", "way": "link", "sends": 1000000, "seconds": 2.203, "sends_per_second": 454016, "cpu_us_per_send": 3.69, "threads": 5, "packets": 44087, "load": 2.74}
+{"round": 4, "version": "push", "way": "links4", "sends": 4000000, "seconds": 5.052, "sends_per_second": 791775, "cpu_us_per_send": 2.24, "threads": 5, "packets": 64099, "load": 2.84}
+{"round": 4, "version": "before", "way": "links4", "sends": 4000000, "seconds": 8.141, "sends_per_second": 491343, "cpu_us_per_send": 3.69, "threads": 8, "packets": 186973, "load": 2.7}
+{"round": 4, "version": "threads-push", "way": "links4", "sends": 4000000, "seconds": 5.029, "sends_per_second": 795442, "cpu_us_per_send": 2.27, "threads": 8, "packets": 64239, "load": 2.72}
+{"round": 4, "version": "loop", "way": "links4", "sends": 4000000, "seconds": 11.684, "sends_per_second": 342346, "cpu_us_per_send": 5.49, "threads": 5, "packets": 340844, "load": 2.82}
+{"round": 4, "version": "push", "way": "wire", "sends": 1000000, "seconds": 1.259, "sends_per_second": 794427, "cpu_us_per_send": 2.23, "threads": 4, "packets": 0, "load": 3.0}
+{"round": 4, "version": "before", "way": "wire", "sends": 1000000, "seconds": 1.28, "sends_per_second": 781423, "cpu_us_per_send": 2.28, "threads": 4, "packets": 0, "load": 3.0}
+{"round": 4, "version": "threads-push", "way": "wire", "sends": 1000000, "seconds": 1.33, "sends_per_second": 751745, "cpu_us_per_send": 2.29, "threads": 4, "packets": 0, "load": 3.0}
+{"round": 4, "version": "loop", "way": "wire", "sends": 1000000, "seconds": 1.252, "sends_per_second": 798741, "cpu_us_per_send": 2.24, "threads": 4, "packets": 0, "load": 3.08}
+{"round": 5, "version": "before", "way": "link", "sends": 1000000, "seconds": 1.664, "sends_per_second": 600797, "cpu_us_per_send": 2.98, "threads": 5, "packets": 35869, "load": 3.08}
+{"round": 5, "version": "threads-push", "way": "link", "sends": 1000000, "seconds": 1.066, "sends_per_second": 938003, "cpu_us_per_send": 1.89, "threads": 5, "packets": 16682, "load": 3.08}
+{"round": 5, "version": "loop", "way": "link", "sends": 1000000, "seconds": 1.918, "sends_per_second": 521488, "cpu_us_per_send": 3.29, "threads": 5, "packets": 40237, "load": 3.08}
+{"round": 5, "version": "push", "way": "link", "sends": 1000000, "seconds": 1.072, "sends_per_second": 932938, "cpu_us_per_send": 1.92, "threads": 5, "packets": 16517, "load": 3.23}
+{"round": 5, "version": "before", "way": "links4", "sends": 4000000, "seconds": 8.036, "sends_per_second": 497734, "cpu_us_per_send": 3.6, "threads": 8, "packets": 169233, "load": 3.23}
+{"round": 5, "version": "threads-push", "way": "links4", "sends": 4000000, "seconds": 4.938, "sends_per_second": 810046, "cpu_us_per_send": 2.25, "threads": 8, "packets": 64235, "load": 2.89}
+{"round": 5, "version": "loop", "way": "links4", "sends": 4000000, "seconds": 18.845, "sends_per_second": 212262, "cpu_us_per_send": 8.82, "threads": 5, "packets": 568061, "load": 2.82}
+{"round": 5, "version": "push", "way": "links4", "sends": 4000000, "seconds": 4.861, "sends_per_second": 822797, "cpu_us_per_send": 2.18, "threads": 5, "packets": 64017, "load": 2.79}
+{"round": 5, "version": "before", "way": "wire", "sends": 1000000, "seconds": 1.259, "sends_per_second": 794126, "cpu_us_per_send": 2.25, "threads": 4, "packets": 0, "load": 2.73}
+{"round": 5, "version": "threads-push", "way": "wire", "sends": 1000000, "seconds": 1.318, "sends_per_second": 758603, "cpu_us_per_send": 2.29, "threads": 4, "packets": 0, "load": 2.73}
+{"round": 5, "version": "loop", "way": "wire", "sends": 1000000, "seconds": 1.258, "sends_per_second": 794856, "cpu_us_per_send": 2.24, "threads": 4, "packets": 0, "load": 2.73}
+{"round": 5, "version": "push", "way": "wire", "sends": 1000000, "seconds": 1.253, "sends_per_second": 798253, "cpu_us_per_send": 2.24, "threads": 4, "packets": 0, "load": 2.73}
diff --git a/research/experiments/028-one-sending-loop/2026-10-01-links-load.txt b/research/experiments/028-one-sending-loop/2026-10-01-links-load.txt
new file mode 100644
index 0000000..9fd693c
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/2026-10-01-links-load.txt
@@ -0,0 +1,2 @@
+ 08:54:09 up  6:06,  3 users,  load average: 0.98, 0.86, 0.66
+ 08:57:54 up  6:10,  3 users,  load average: 2.67, 1.96, 1.16
diff --git a/research/experiments/028-one-sending-loop/dst.aup b/research/experiments/028-one-sending-loop/dst.aup
new file mode 100644
index 0000000..c1c8acb
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/dst.aup
@@ -0,0 +1,9 @@
+state n = 0
+state total = 0
+input take = message { k: u32 }
+on take {
+  n = n + 1
+  total = total + take.k
+}
+view count = n
+view sum = total
diff --git a/research/experiments/028-one-sending-loop/runs.sh b/research/experiments/028-one-sending-loop/runs.sh
new file mode 100755
index 0000000..65eb51d
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/runs.sh
@@ -0,0 +1,36 @@
+#!/bin/sh
+# The runs of docs/design/017's 2b, from the repository's root, in steps
+# (a step at a time: a background job may be stopped after half an hour).
+#   runs.sh build   the builds, each in /tmp/pure-028/NAME with its
+#                   pure-crypto.so: before (a thread a link, 3791ab8),
+#                   threads-push (before with threads-push.patch: the
+#                   owner pushes a turn's sends at once), loop (one
+#                   sending loop, 32f96b3), push (the loop with push,
+#                   6d3c3c2)
+#   runs.sh links   five rounds of 1,000 bursts by one link, four links and
+#                   one wire, the builds in turns, once the last minute's
+#                   load is below 1.5
+set -e
+out=research/experiments/028-one-sending-loop
+builds=/tmp/pure-028
+case "$1" in
+build)
+    for pair in before:3791ab8 threads-push:3791ab8 loop:32f96b3 push:6d3c3c2; do
+        name=${pair%%:*} commit=${pair#*:}
+        rm -rf $builds/src-$name
+        git worktree add -f -q --detach $builds/src-$name $commit
+        if [ $name = threads-push ]; then git -C $builds/src-$name apply "$PWD/$out/threads-push.patch"; fi
+        make -s -C $builds/src-$name build/pure-serve build/pure-crypto.so
+        mkdir -p $builds/$name
+        cp $builds/src-$name/build/pure-serve $builds/src-$name/build/pure-crypto.so $builds/$name/
+        git worktree remove --force $builds/src-$name
+    done ;;
+links)
+    until awk '{ exit !($1 < 1.5) }' /proc/loadavg; do sleep 10; done
+    echo "$(uptime)" > $out/2026-10-01-links-load.txt
+    python3 $out/sends.py 5 1000 link,links4,wire before=$builds/before/pure-serve \
+        threads-push=$builds/threads-push/pure-serve loop=$builds/loop/pure-serve push=$builds/push/pure-serve \
+        > $out/2026-10-01-links-linux-aarch64.jsonl
+    echo "$(uptime)" >> $out/2026-10-01-links-load.txt ;;
+*) echo "runs.sh build|links" >&2; exit 2 ;;
+esac
diff --git a/research/experiments/028-one-sending-loop/sends.py b/research/experiments/028-one-sending-loop/sends.py
new file mode 100644
index 0000000..16c1eac
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/sends.py
@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""Sends between apps of one host over links or HTTP wires, before and
+after one sending loop (docs/design/017, 2b). src holds 1,000 items; a
+burst walks them and sends one message each to one output (link, wire),
+or to each of four outputs, each to its own app (links4, wires4). A link
+goes to the host's own --link-port, sealed with its key; a wire posts to
+the host's own HTTP port. A run: BURSTS bursts posted one after another,
+then the time until every receiving app has stored every send, the host's
+CPU over it (utime + stime), its threads, and the packets its links sent;
+the receivers' sums say each send came once. In RAM, the host on cores 0-2.
+The builds take turns, their order turned each round.
+
+sends.py ROUNDS BURSTS WAY[,WAY...] NAME=PURE_SERVE [NAME=PURE_SERVE ...]
+    a JSON line per run (each pure-serve in a directory of its own, with
+    its pure-crypto.so)
+"""
+import http.client
+import json
+import os
+import select
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+AUP = ROOT.parent / 'aup' / 'build' / 'aup'
+ITEMS = 1000
+
+
+def free_port(kind):
+    with socket.socket(socket.AF_INET, kind) as probe:
+        probe.bind(('127.0.0.1', 0))
+        return probe.getsockname()[1]
+
+
+def started(args):
+    process = subprocess.Popen(['taskset', '-c', '0-2', *map(str, args)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    ready, _, _ = select.select([process.stdout], [], [], 10)
+    assert ready, process.stderr.read()
+    return process, json.loads(process.stdout.readline())
+
+
+def stopped(process):
+    process.terminate()
+    _, error = process.communicate(timeout=60)
+    assert process.returncode == 0, error[-500:]
+    return json.loads(error.decode().strip().splitlines()[-1])
+
+
+def cpu_seconds(pid):
+    fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
+    return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+
+def threads(pid):
+    return int(Path(f'/proc/{pid}/status').read_text().split('Threads:')[1].split()[0])
+
+
+def run(way, serve, work, modules, key, public, bursts):
+    link_port, http_port = free_port(socket.SOCK_DGRAM), free_port(socket.SOCK_STREAM)
+    outputs = ['o1', 'o2', 'o3', 'o4'] if way.endswith('4') else ['out']
+    lines = []
+    for n, output in enumerate(outputs, 1):
+        to = (f'pure://127.0.0.1:{link_port}/d{n}/take seal={output}={public}' if way.startswith('link')
+              else f'http://127.0.0.1:{http_port}/d{n}/inputs/take')
+        lines.append(f'send={output}={to}')
+    conf = f"src {modules['src']} {work / 'src'} {' '.join(lines)}\n" + \
+        ''.join(f"d{n} {modules['dst']} {work / f'd{n}'}\n" for n in range(1, len(outputs) + 1))
+    (work / 'apps.conf').write_text(conf)
+    args = [serve, '--apps', work / 'apps.conf', '--threads', 1, '--workers', 2, '--port', http_port]
+    if way.startswith('link'):
+        args += ['--key', key, '--link-port', link_port, *sum((['--sender', f'src.{o}={public}'] for o in outputs), [])]
+    load = os.getloadavg()[0]
+    process, info = started(args)
+    client = http.client.HTTPConnection('127.0.0.1', info['port'], timeout=60)
+    counts = {}
+
+    def ask(method, path):
+        client.request(method, path, b'' if method == 'POST' else None)
+        answer = client.getresponse()
+        body = answer.read()
+        assert answer.status == 200, (path, answer.status, answer.getheader('X-Pure-Error'))
+        return body
+
+    try:
+        for k in range(1, ITEMS + 1):
+            ask('POST', f'/src/inputs/fill?k={k}')
+        alive = threads(process.pid)
+        began, cpu = time.perf_counter(), cpu_seconds(process.pid)
+        for _ in range(bursts):
+            ask('POST', '/src/inputs/burst4?step=1' if len(outputs) == 4 else '/src/inputs/burst?step=1')
+        while any(int(ask('GET', f'/d{n}/views/count')) < bursts * ITEMS for n in range(1, len(outputs) + 1)):
+            time.sleep(0.005)
+        took, used = time.perf_counter() - began, cpu_seconds(process.pid) - cpu
+        for n in range(1, len(outputs) + 1):
+            assert int(ask('GET', f'/d{n}/views/sum')) == bursts * (ITEMS * (ITEMS + 1) // 2 + ITEMS)
+    finally:
+        client.close()
+        counts = stopped(process)
+    sent = bursts * ITEMS * len(outputs)
+    return {'way': way, 'sends': sent, 'seconds': round(took, 3), 'sends_per_second': round(sent / took),
+            'cpu_us_per_send': round(used * 1e6 / sent, 2), 'threads': alive,
+            'packets': counts.get('link_packets', 0), 'load': round(load, 2)}
+
+
+def main():
+    rounds, bursts, ways = int(sys.argv[1]), int(sys.argv[2]), sys.argv[3].split(',')
+    serves = {name: Path(path).resolve() for name, path in (a.split('=', 1) for a in sys.argv[4:])}
+    names = list(serves)
+    work = Path(tempfile.mkdtemp(prefix='pure-loop-', dir='/dev/shm'))
+    try:
+        modules = {}
+        for name in ('src', 'dst'):
+            modules[name] = work / f'{name}.pmod'
+            subprocess.run([str(AUP), 'build', str(HERE / f'{name}.aup'), '-o', str(modules[name])], check=True)
+        key = work / 'host.key'
+        (work / 'one.conf').write_text(f"x {modules['dst']} {work / 'keys'}\n")
+        process, info = started([serves[names[-1]], '--apps', work / 'one.conf', '--key', key, '--port', 0])
+        public = info['public_key']
+        stopped(process)
+        for round_ in range(1, rounds + 1):
+            for way in ways:
+                turn = (round_ - 1) % len(names)
+                for version in names[turn:] + names[:turn]:
+                    runs = work / f'run-{round_}-{way}-{version}'
+                    runs.mkdir()
+                    row = run(way, serves[version], runs, modules, key, public, bursts)
+                    print(json.dumps({'round': round_, 'version': version, **row}), flush=True)
+                    shutil.rmtree(runs)
+    finally:
+        shutil.rmtree(work)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/research/experiments/028-one-sending-loop/src.aup b/research/experiments/028-one-sending-loop/src.aup
new file mode 100644
index 0000000..6bb9c42
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/src.aup
@@ -0,0 +1,29 @@
+record Item {
+  k = 0
+}
+state items: list of Item
+input fill = message { k: u32 }
+input burst = message { step: u32 }
+input burst4 = message { step: u32 }
+output out = message { k: u32 }
+output o1 = message { k: u32 }
+output o2 = message { k: u32 }
+output o3 = message { k: u32 }
+output o4 = message { k: u32 }
+on fill {
+  items.append(Item { k: fill.k })
+}
+on burst {
+  for i in items {
+    send out { k: i.k + burst.step }
+  }
+}
+on burst4 {
+  for i in items {
+    send o1 { k: i.k + burst4.step }
+    send o2 { k: i.k + burst4.step }
+    send o3 { k: i.k + burst4.step }
+    send o4 { k: i.k + burst4.step }
+  }
+}
+view count = count(i in items where i.k > 0)
diff --git a/research/experiments/028-one-sending-loop/threads-push.patch b/research/experiments/028-one-sending-loop/threads-push.patch
new file mode 100644
index 0000000..5120135
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/threads-push.patch
@@ -0,0 +1,62 @@
+diff --git a/link/link.c b/link/link.c
+index 8f36e21..6287ae6 100644
+--- a/link/link.c
++++ b/link/link.c
+@@ -185,15 +185,18 @@ static void wake(int fd) {
+ 
+ bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
+     pthread_mutex_lock(&l->lock);
+-    bool ok = pure_link_sender_queue(l->sender, seq, bytes, size), signal = false;
+-    if (ok) {
+-        l->queued_new = true;
+-        signal = l->sleeping && !l->signaled;
+-        l->signaled = l->signaled || signal;
+-    }
++    bool ok = pure_link_sender_queue(l->sender, seq, bytes, size);
++    pthread_mutex_unlock(&l->lock);
++    return ok;
++}
++
++void pure_link_push(PureLink *l) {
++    pthread_mutex_lock(&l->lock);
++    l->queued_new = true;
++    bool signal = l->sleeping && !l->signaled;
++    l->signaled = l->signaled || signal;
+     pthread_mutex_unlock(&l->lock);
+     if (signal) wake(l->wake);
+-    return ok;
+ }
+ 
+ uint64_t pure_link_queued(PureLink *l) {
+diff --git a/link/link.h b/link/link.h
+index c8d8635..6dae11f 100644
+--- a/link/link.h
++++ b/link/link.h
+@@ -23,6 +23,7 @@ void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]);
+  * them by name (docs/design/014-fields-by-name.md). */
+ bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size);
+ bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size);
++void pure_link_push(PureLink *);
+ uint64_t pure_link_queued(PureLink *);
+ /* The last number the receiver saved; any thread. */
+ uint64_t pure_link_delivered(PureLink *);
+diff --git a/serve/serve.c b/serve/serve.c
+index b5112d2..b0200a4 100644
+--- a/serve/serve.c
++++ b/serve/serve.c
+@@ -1829,6 +1829,7 @@ static void feed_outlets(Executor *ex) {
+                  delivered = wire ? pure_wire_delivered(wire) : link ? pure_link_delivered(link) : local_delivered(local);
+         size_t held = wire ? pure_wire_held(wire) : link ? pure_link_held(link) : local_held(local), at = ex->outlet_from[i];
+         at = at > gone ? at - (size_t)gone : 0;
++        uint64_t before = queued;
+         for (; at < ex->app.sent_saved; ++at) {
+             uint32_t of;
+             uint64_t seq;
+@@ -1844,6 +1845,7 @@ static void feed_outlets(Executor *ex) {
+             queued = seq;
+             held += payload.size;
+         }
++        if (link && queued != before) pure_link_push(link);
+         ex->outlet_from[i] = at;
+     }
+     uint64_t now = ex->local_count ? now_ms() : 0;