"""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)