209 lines
8 KiB
Python
209 lines
8 KiB
Python
"""Check gbn.py and sr.py over the simulated link, offline. Given; do not change it.
|
|
|
|
python3 selfcheck.py # both protocols
|
|
python3 selfcheck.py gbn # one
|
|
|
|
Every check prints PASS, FAIL (with what went wrong), or SKIP while a class
|
|
still raises NotImplementedError. The link is seeded, so a failure repeats the
|
|
same way every run until you change your code.
|
|
"""
|
|
|
|
import os
|
|
import random
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import channel
|
|
import packet
|
|
import stopwait
|
|
|
|
TIMEOUT = 0.03
|
|
LIMIT = 30 # seconds one transfer may take before it counts as stuck
|
|
results = []
|
|
|
|
|
|
def report(name, outcome, why=""):
|
|
results.append(outcome)
|
|
print(f"{outcome:4} {name}" + (f": {why}" if why else ""))
|
|
|
|
|
|
def transfer(proto, data, window=8, watch=None, drop=None, **link):
|
|
"""Run one transfer over the simulated link; return (received bytes, stats, seconds)."""
|
|
mod = __import__(proto)
|
|
receiver_cls = getattr(mod, "Receiver", stopwait.Receiver)
|
|
a, b = channel.link(watch=watch, drop=drop, **link)
|
|
out = {}
|
|
|
|
def run_receiver():
|
|
try:
|
|
kw = {"window": window} if receiver_cls is not stopwait.Receiver else {}
|
|
out["data"] = receiver_cls(b, linger=0.15, **kw).receive()
|
|
except Exception as e: # shown as a FAIL below
|
|
out["error"] = e
|
|
|
|
def run_sender():
|
|
try:
|
|
out["stats"] = mod.Sender(a, "B", window=window, timeout=TIMEOUT).send(data)
|
|
except Exception as e:
|
|
out["error"] = e
|
|
|
|
receiving = threading.Thread(target=run_receiver, daemon=True)
|
|
sending = threading.Thread(target=run_sender, daemon=True)
|
|
receiving.start()
|
|
start = time.monotonic()
|
|
sending.start()
|
|
sending.join(LIMIT)
|
|
receiving.join(max(1.0, LIMIT - (time.monotonic() - start)))
|
|
if "error" in out:
|
|
raise out["error"]
|
|
if sending.is_alive():
|
|
raise AssertionError(f"the sender was still running after {LIMIT} s: is it waiting for an ACK that never comes?")
|
|
if receiving.is_alive():
|
|
raise AssertionError("the receiver never finished: did it get a FIN it accepted?")
|
|
return out.get("data"), out["stats"], time.monotonic() - start
|
|
|
|
|
|
def delivered(proto, name, data, **kw):
|
|
try:
|
|
got, stats, _ = transfer(proto, data, **kw)
|
|
except NotImplementedError:
|
|
report(f"{proto} {name}", "SKIP", "not written yet")
|
|
return None
|
|
except AssertionError as e:
|
|
report(f"{proto} {name}", "FAIL", str(e))
|
|
return None
|
|
if got != data:
|
|
size = "nothing" if not got else f"{len(got)} bytes"
|
|
report(f"{proto} {name}", "FAIL", f"sent {len(data)} bytes, received {size}"
|
|
+ ("" if not got or len(got) != len(data) else " with different contents"))
|
|
return None
|
|
report(f"{proto} {name}", "PASS")
|
|
return stats
|
|
|
|
|
|
def window_watch(proto, window, problems):
|
|
"""Flags any DATA packet sent beyond the window the ACKs seen so far allow."""
|
|
state = {"base": 0, "acked": set()}
|
|
|
|
def watch(sender, raw):
|
|
got = packet.decode(raw)
|
|
if got is None:
|
|
return
|
|
kind, seq, _ = got
|
|
if sender == "B" and kind == packet.ACK:
|
|
if proto == "gbn":
|
|
state["base"] = max(state["base"], seq)
|
|
else:
|
|
state["acked"].add(seq)
|
|
while state["base"] in state["acked"]:
|
|
state["base"] += 1
|
|
elif sender == "A" and kind == packet.DATA and seq >= state["base"] + window:
|
|
problems.append(f"sent packet {seq} while the window was {state['base']} to {state['base'] + window - 1}")
|
|
return watch
|
|
|
|
|
|
def check(proto):
|
|
rng = random.Random(4220)
|
|
blob = bytes(rng.randrange(256) for _ in range(30_000))
|
|
delivered(proto, "clean link", blob, seed=1)
|
|
delivered(proto, "20% loss", blob, loss=0.2, seed=2)
|
|
delivered(proto, "reordering, duplicates and corruption", blob,
|
|
jitter=0.02, duplicate=0.1, corrupt=0.1, seed=3)
|
|
delivered(proto, "an empty file", b"", seed=4)
|
|
delivered(proto, "exactly three full packets", blob[:3 * packet.MSS], loss=0.1, seed=5)
|
|
delivered(proto, "one byte past a full packet", blob[:3 * packet.MSS + 1], loss=0.1, seed=6)
|
|
delivered(proto, "window of 1", blob[:8000], window=1, loss=0.1, seed=7)
|
|
delivered(proto, "40% loss, window 32", blob, window=32, loss=0.4, seed=8)
|
|
|
|
problems = []
|
|
if delivered(proto, "never more than the window in flight", blob, window=4, loss=0.15,
|
|
jitter=0.01, seed=9, watch=window_watch(proto, 4, problems)) is not None:
|
|
if problems:
|
|
results[-1] = "FAIL"
|
|
print(f" ...but {problems[0]}")
|
|
|
|
try:
|
|
data = blob[:50 * packet.MSS]
|
|
_, _, slow = transfer("stopwait", data, window=1, delay=0.01, seed=10)
|
|
_, _, fast = transfer(proto, data, window=8, delay=0.01, seed=10)
|
|
if fast < slow / 2:
|
|
report(f"{proto} pipelines: window 8 beats stop-and-wait", "PASS")
|
|
else:
|
|
report(f"{proto} pipelines: window 8 beats stop-and-wait", "FAIL",
|
|
f"took {fast:.2f} s against stop-and-wait's {slow:.2f} s; is more than one packet in flight?")
|
|
except NotImplementedError:
|
|
report(f"{proto} pipelines", "SKIP", "not written yet")
|
|
|
|
|
|
def one_loss(proto):
|
|
"""Lose data packet 5 once, and nothing else. What does the sender resend?"""
|
|
lost = []
|
|
|
|
def drop(sender, raw):
|
|
got = packet.decode(raw)
|
|
if sender == "A" and got and got[0] == packet.DATA and got[1] == 5 and not lost:
|
|
lost.append(5)
|
|
return True
|
|
return False
|
|
|
|
try:
|
|
_, stats, _ = transfer(proto, bytes(20 * packet.MSS), window=8, drop=drop, seed=11)
|
|
except NotImplementedError:
|
|
report(f"{proto} after one lost packet", "SKIP", "not written yet")
|
|
return
|
|
except AssertionError as e:
|
|
report(f"{proto} after one lost packet", "FAIL", str(e))
|
|
return
|
|
n = stats["retransmissions"]
|
|
if proto == "sr" and n <= 2:
|
|
report("sr after one lost packet, resends about one", "PASS", f"resent {n}")
|
|
elif proto == "sr":
|
|
report("sr after one lost packet, resends about one", "FAIL",
|
|
f"resent {n}; selective repeat resends only the packets whose own timer ran out")
|
|
elif n >= 4:
|
|
report("gbn after one lost packet, resends the window behind it", "PASS", f"resent {n}")
|
|
else:
|
|
report("gbn after one lost packet, resends the window behind it", "FAIL",
|
|
f"resent {n}; when the timer runs out, go-back-N resends every unACKed packet")
|
|
|
|
|
|
def sr_receiver_buffers():
|
|
import sr
|
|
a, b = channel.link(seed=12)
|
|
out = {}
|
|
try:
|
|
receiver = sr.Receiver(b, window=4, linger=0.1)
|
|
except NotImplementedError:
|
|
report("sr receiver keeps packets that arrive early", "SKIP", "not written yet")
|
|
return
|
|
t = threading.Thread(target=lambda: out.setdefault("data", receiver.receive()), daemon=True)
|
|
t.start()
|
|
acks = []
|
|
a.settimeout(0.2)
|
|
for seq, body in ((1, b"world"), (0, b"hello "), (1, b"world")):
|
|
a.sendto(packet.encode(packet.DATA, seq, body), "B")
|
|
try:
|
|
got = packet.decode(a.recvfrom(2048)[0])
|
|
acks.append(got[1] if got and got[0] == packet.ACK else None)
|
|
except OSError:
|
|
acks.append(None)
|
|
a.sendto(packet.encode(packet.FIN, 2), "B")
|
|
t.join(2)
|
|
if acks[:2] == [1, 0] and acks[2] == 1 and out.get("data") == b"hello world":
|
|
report("sr receiver keeps packets that arrive early", "PASS")
|
|
else:
|
|
report("sr receiver keeps packets that arrive early", "FAIL",
|
|
f"sent 1, 0, then 1 again: expected ACKs 1, 0, 1 and b'hello world'; got ACKs {acks} and {out.get('data')!r}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
wanted = sys.argv[1:] or ["gbn", "sr"]
|
|
for proto in wanted:
|
|
check(proto)
|
|
for proto in wanted:
|
|
one_loss(proto)
|
|
if "sr" in wanted:
|
|
sr_receiver_buffers()
|
|
print(f"\n{results.count('PASS')} passed, {results.count('FAIL')} failed, {results.count('SKIP')} not written yet")
|
|
raise SystemExit(1 if "FAIL" in results else 0)
|