diff options
Diffstat (limited to 'research/experiments/044-specs')
| -rw-r--r-- | research/experiments/044-specs/2026-10-19-rto.jsonl | 1 | ||||
| -rw-r--r-- | research/experiments/044-specs/rto.py | 40 | ||||
| -rw-r--r-- | research/experiments/044-specs/rto_check.py | 145 |
3 files changed, 186 insertions, 0 deletions
diff --git a/research/experiments/044-specs/2026-10-19-rto.jsonl b/research/experiments/044-specs/2026-10-19-rto.jsonl new file mode 100644 index 0000000..71d73d7 --- /dev/null +++ b/research/experiments/044-specs/2026-10-19-rto.jsonl @@ -0,0 +1 @@ +{"scenarios": 20, "events": 241612, "timeouts": 156, "right": {"batches": 241612, "rounds": 241612, "all": 241612}, "paths_same": true, "costs": {"10": {"ticks": 218070, "timeouts": 78, "ticks_a_timeout": 2795.8, "late_mean_ms": 4.27, "late_most_ms": 9.875}, "100": {"ticks": 21968, "timeouts": 78, "ticks_a_timeout": 281.6, "late_mean_ms": 49.23, "late_most_ms": 99.0}}, "aup_lines": 56, "aup_brace_lines": 11, "python_lines": 30} diff --git a/research/experiments/044-specs/rto.py b/research/experiments/044-specs/rto.py new file mode 100644 index 0000000..e4a4a57 --- /dev/null +++ b/research/experiments/044-specs/rto.py @@ -0,0 +1,40 @@ +# RFC 6298's retransmission timer in plain Python: the reference 140 counts aup's lines against. A sender's segments +# are sent and acknowledged (cumulatively) at given times; the round-trip time of each acknowledged segment that was +# never sent again (Karn) updates SRTT, RTTVAR and RTO (2.2-2.5: 1 s at least, 60 s at most), and the timer runs as +# 5.1-5.6 say: started when one is sent and none runs, stopped when all are acknowledged, restarted when new data is, +# and on expiry the earliest is sent again and RTO doubles. The timer is seen at ticks of period P, as an app sees time. +LOWEST, HIGHEST = 1000, 60000 # milliseconds; the clock's granularity G is the tick's period + + +class Sender: + def __init__(self, period): + self.period, self.srtt, self.rttvar, self.rto, self.measured = period, 0.0, 0.0, 1000.0, False + self.deadline, self.outstanding, self.retransmits = -1.0, [], [] # outstanding: [seq, sent, again] + + def send(self, now, seq): + self.outstanding.append([seq, now, False]) + if self.deadline < 0: + self.deadline = now + self.rto # 5.1 + + def ack(self, now, seq): + acked = [s for s in self.outstanding if s[0] <= seq] + if not acked: + return + self.outstanding = [s for s in self.outstanding if s[0] > seq] + last = acked[-1] + if not last[2]: # Karn: no sample from a segment sent again + r = now - last[1] + if not self.measured: + self.srtt, self.rttvar, self.measured = r, r / 2, True # 2.2 + else: + self.rttvar = 0.75 * self.rttvar + 0.25 * abs(self.srtt - r) # 2.3 + self.srtt = 0.875 * self.srtt + 0.125 * r + self.rto = min(max(self.srtt + max(self.period, 4 * self.rttvar), LOWEST), HIGHEST) # 2.4, 2.5 + self.deadline = now + self.rto if self.outstanding else -1.0 # 5.3, 5.2 + + def tick(self, now): + if self.deadline >= 0 and now >= self.deadline: # 5.4-5.6 + self.outstanding[0][2] = True + self.retransmits.append((self.outstanding[0][0], now)) + self.rto = min(self.rto * 2, HIGHEST) + self.deadline = now + self.rto diff --git a/research/experiments/044-specs/rto_check.py b/research/experiments/044-specs/rto_check.py new file mode 100644 index 0000000..be41ffb --- /dev/null +++ b/research/experiments/044-specs/rto_check.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Research 140: RFC 6298's retransmission timer written in aup (aup's examples/rto.aup), with periodic ticks. + +Each scenario (a seed and a tick period P) is made closed-loop by rto.py's model and a receiver: segments sent at random +gaps, each arriving half a random round trip later unless its first sending is lost (one in five), the receiver +acknowledging cumulatively all it holds without a gap, the acknowledgement arriving after the other half; a segment the +timer sends again arrives. The events (sends, acknowledgements, ticks every P ms) are then given to the app +through pure-run (time MS before each), on Pure's three paths: after every event its SRTT, RTTVAR, RTO, deadline, the +segments outstanding and the retransmissions must be the model's, the paths the same. No known implementation is +compared: the oracle is the model of the same rules. What the ticks cost is measured: ticks a timeout (a one-shot timer +would be one event), and how late a timeout is seen (below P). Then counts lines of rto.aup and of rto.py's model. One +JSON line.""" +import json, os, random, struct, subprocess, sys, tempfile + +here = os.path.dirname(os.path.abspath(__file__)) +pure = os.path.normpath(os.path.join(here, '../../..')) +aup = os.environ.get('AUP', os.path.join(pure, '..', 'aup')) +sys.path.insert(0, here) +import rto # noqa: E402 + +START = 1_000_000 # the first event's time (ms since 1970, as now is) + + +def lines(path, stop=None): + code = braces = 0 + for line in open(path): + if stop and line.startswith(stop): break + text = line.strip() + if not text or text.startswith('#'): continue + code += 1 + braces += text == '}' + return code, braces + + +def scenario(seed, period, segments=40): + """The events the app is given (time, kind, seq), the model's state after each, and the timeouts' lateness. The + network: a sending arrives half a random round trip later unless it is lost (a first sending, one in five), and the + receiver acknowledges at once, cumulatively, all it holds without a gap; the acknowledgement arrives after the other + half.""" + chooser = random.Random(seed) + sender = rto.Sender(period) + pending, t = [], START + period # (time, order, kind, seq): arrivals, then acknowledgements, sends, then ticks + for seq in range(1, segments + 1): + t += chooser.choice([0, 0, 5, 40, 120, 300, 900]) + pending.append((t, 2, 'transmit', seq)) + held, through = set(), 0 # the receiver's segments, and the last one it holds every one up to + + def sent(time, seq, first): + if not first or chooser.random() > 0.2: + half = chooser.randrange(20, 350) + pending.append((time + half, 0, 'arrive', seq * 1000 + half)) + + events, states, late = [], [], [] + tick = START + while pending or sender.outstanding: + pending.sort() + if pending and pending[0][0] < tick + period: # the host ticks every P, the timer running or not + time, _, kind, seq = pending.pop(0) + else: + time, kind, seq = tick + period, 'tick', 0 + tick = time + if kind == 'arrive': # the receiver's: not the app's + seq, half = divmod(seq, 1000) + held.add(seq) + while through + 1 in held: + through += 1 + if through: + pending.append((time + half, 1, 'ack', through)) + continue + if kind == 'transmit': + sender.send(time, seq) + sent(time, seq, True) + elif kind == 'ack': + sender.ack(time, seq) + else: + before, deadline = len(sender.retransmits), sender.deadline + sender.tick(time) + if len(sender.retransmits) > before: + late.append(time - deadline) + sent(time, sender.retransmits[-1][0], False) + events.append((time, kind, seq)) + states.append((sender.srtt, sender.rttvar, sender.rto, sender.deadline, len(sender.outstanding), len(sender.retransmits))) + if len(events) > 200000: + raise RuntimeError('a scenario without end') + return events, states, late + + +def run(module, events, env, work): + commands = os.path.join(work, 'rto.commands') + with open(commands, 'w') as f: + for time, kind, seq in events: + f.write(f'time {time}\n') + f.write('tick tick\n' if kind == 'tick' else f"message {kind} {struct.pack('<I', seq).hex()}\n") + got = [] + with open(commands) as stdin: + process = subprocess.Popen([os.path.join(pure, 'build', 'pure-run'), module], stdin=stdin, stdout=subprocess.PIPE, + text=True, env=dict(os.environ, **env)) + for k, line in enumerate(process.stdout): + if k % 2 == 0: continue # a time command's + state = json.loads(line) + s = state['states'] + got.append((state['fault'], s['srtt'], s['rttvar'], s['rto'], s['deadline'], len(s['outstanding']), + len(s['retransmits']))) + process.wait() + return got + + +if __name__ == '__main__': + report = {'scenarios': 0, 'events': 0, 'timeouts': 0, 'right': {}, 'paths_same': True} + costs = {} + seeds = int(sys.argv[1]) if len(sys.argv) > 1 else 10 + with tempfile.TemporaryDirectory() as work: + source = open(os.path.join(aup, 'examples', 'rto.aup')).read() + for period in (10, 100): + module = os.path.join(work, f'rto-{period}.pmod') # its tick and its G the scenario's period + with open(os.path.join(work, f'rto-{period}.aup'), 'w') as f: + f.write(source.replace('every 10 ms', f'every {period} ms').replace('fn PERIOD = 10 ', f'fn PERIOD = {period} ')) + subprocess.run([os.path.join(aup, 'build', 'aup'), 'build', os.path.join(work, f'rto-{period}.aup'), '-o', module], check=True) + ticks = timeouts = 0 + lateness = [] + for seed in range(seeds): + events, states, late = scenario(seed, period) + outputs = {} + for name, env in (('batches', {'PURE_JIT': '0'}), ('rounds', {'PURE_JIT': '0', 'PURE_BATCH': '0'}), + ('all', {'PURE_JIT': 'all'})): + got = run(module, events, env, work) + want = [('none',) + s for s in states] + report['right'][name] = report['right'].get(name, 0) + sum(g == w for g, w in zip(got, want)) + outputs[name] = got + report['paths_same'] = report['paths_same'] and outputs['rounds'] == outputs['batches'] == outputs['all'] + report['scenarios'] += 1 + report['events'] += len(events) + ticks += sum(kind == 'tick' for _, kind, _ in events) + timeouts += len(late) + lateness += late + report['timeouts'] += timeouts + costs[period] = {'ticks': ticks, 'timeouts': timeouts, 'ticks_a_timeout': round(ticks / max(timeouts, 1), 1), + 'late_mean_ms': round(sum(lateness) / max(len(lateness), 1), 2), 'late_most_ms': max(lateness, default=0)} + report['costs'] = costs + aup_code, aup_braces = lines(os.path.join(aup, 'examples', 'rto.aup')) + python_code, _ = lines(os.path.join(here, 'rto.py')) + report.update({'aup_lines': aup_code, 'aup_brace_lines': aup_braces, 'python_lines': python_code}) + ok = all(n == report['events'] for n in report['right'].values()) and report['paths_same'] + print(json.dumps(report)) + sys.exit(0 if ok else 1) |