89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""A simulated unreliable network for testing without Mininet. Given; do not change it.
|
|
|
|
a, b = link(loss=0.1, delay=0.005, seed=1)
|
|
|
|
gives two endpoints that behave like UDP sockets connected to each other:
|
|
sendto(), recvfrom(), and settimeout(), where recvfrom raises socket.timeout
|
|
when nothing arrives in time. The link can drop, delay, reorder, duplicate, or
|
|
corrupt packets, each with its own probability, and the same seed always
|
|
gives the same run. Your sender and receiver only ever call those three
|
|
methods, so the same code runs over this link and over a real UDP socket.
|
|
|
|
Reference: https://docs.python.org/3/library/socket.html
|
|
"""
|
|
|
|
import heapq
|
|
import itertools
|
|
import random
|
|
import socket
|
|
import threading
|
|
import time
|
|
|
|
|
|
class _Endpoint:
|
|
def __init__(self, name, link):
|
|
self.name = name
|
|
self._link = link
|
|
self._queue = [] # (arrival time, order, bytes, from)
|
|
self._timeout = None
|
|
|
|
def settimeout(self, seconds):
|
|
self._timeout = seconds
|
|
|
|
def sendto(self, data, addr):
|
|
self._link._carry(self, bytes(data))
|
|
return len(data)
|
|
|
|
def recvfrom(self, bufsize):
|
|
deadline = None if self._timeout is None else time.monotonic() + self._timeout
|
|
with self._link._cond:
|
|
while True:
|
|
now = time.monotonic()
|
|
if self._queue and self._queue[0][0] <= now:
|
|
_, _, data, sender = heapq.heappop(self._queue)
|
|
return data[:bufsize], sender
|
|
waits = [t for t in (deadline, self._queue[0][0] if self._queue else None) if t is not None]
|
|
if deadline is not None and now >= deadline:
|
|
raise socket.timeout("timed out")
|
|
self._link._cond.wait(None if not waits else max(0.0, min(waits) - now))
|
|
|
|
|
|
class _Link:
|
|
def __init__(self, loss, delay, jitter, duplicate, corrupt, seed, watch, drop):
|
|
self.loss, self.delay, self.jitter = loss, delay, jitter
|
|
self.duplicate, self.corrupt = duplicate, corrupt
|
|
self._rng = random.Random(seed)
|
|
self._order = itertools.count()
|
|
self._cond = threading.Condition()
|
|
self._watch = watch
|
|
self._drop = drop
|
|
self.a = _Endpoint("A", self)
|
|
self.b = _Endpoint("B", self)
|
|
|
|
def _carry(self, source, data):
|
|
dest = self.b if source is self.a else self.a
|
|
with self._cond:
|
|
if self._watch is not None:
|
|
self._watch(source.name, data)
|
|
if self._drop is not None and self._drop(source.name, data):
|
|
self._cond.notify_all()
|
|
return
|
|
copies = 2 if self._rng.random() < self.duplicate else 1
|
|
for _ in range(copies):
|
|
if self._rng.random() < self.loss:
|
|
continue
|
|
packet = data
|
|
if packet and self._rng.random() < self.corrupt:
|
|
i = self._rng.randrange(len(packet))
|
|
packet = packet[:i] + bytes([packet[i] ^ 0xFF]) + packet[i + 1:]
|
|
arrive = time.monotonic() + self.delay + self._rng.uniform(0, self.jitter)
|
|
heapq.heappush(dest._queue, (arrive, next(self._order), packet, source.name))
|
|
self._cond.notify_all()
|
|
|
|
|
|
def link(loss=0.0, delay=0.005, jitter=0.0, duplicate=0.0, corrupt=0.0, seed=0, watch=None, drop=None):
|
|
"""Two connected endpoints (A, B). A jitter larger than the gap between
|
|
packets reorders them. `watch(sender_name, bytes)` sees every packet sent;
|
|
`drop(sender_name, bytes)` returning True loses that one packet."""
|
|
made = _Link(loss, delay, jitter, duplicate, corrupt, seed, watch, drop)
|
|
return made.a, made.b
|