一千字节切三段
本节的字节序号小工具：
segments(nbytes, mss) 把 nbytes 字节切成段，交回 [(序号, 长度)]；序号是这段第一个字节的编号，SYN 占了 0 号，数据从 1 起
acks(segs, arrived) 接收方的累计确认：每收到一段回一个 ack = 下一个想要的字节号；乱序到的先存着，序号补齐了 ack 才往前跳；arrived 是每段到没到1000 字节、每段最多 400，三段全到。打印每段的序号和每次回的 ack：
TABLE = {
("CLOSED", "listen"): ("LISTEN", None),
("CLOSED", "connect"): ("SYN-SENT", "SYN"),
("LISTEN", "SYN"): ("SYN-RECV", "SYN-ACK"),
("SYN-SENT", "SYN-ACK"): ("ESTAB", "ACK"),
("SYN-RECV", "ACK"): ("ESTAB", None),
("ESTAB", "close"): ("FIN-WAIT-1", "FIN"),
("ESTAB", "FIN"): ("CLOSE-WAIT", "ACK"),
("FIN-WAIT-1", "ACK"): ("FIN-WAIT-2", None),
("FIN-WAIT-1", "FIN"): ("CLOSING", "ACK"),
("FIN-WAIT-2", "FIN"): ("TIME-WAIT", "ACK"),
("CLOSING", "ACK"): ("TIME-WAIT", None),
("CLOSE-WAIT", "close"): ("LAST-ACK", "FIN"),
("LAST-ACK", "ACK"): ("CLOSED", None),
("LAST-ACK", "retrans"): ("LAST-ACK", "FIN"),
("TIME-WAIT", "FIN"): ("TIME-WAIT", "ACK"),
("TIME-WAIT", "timeout"): ("CLOSED", None),
("CLOSED", "FIN"): ("CLOSED", "RST"),
("LAST-ACK", "RST"): ("CLOSED", None),
}
class Host:
def __init__(self, name):
self.name = name
self.state = "CLOSED"
def step(self, event):
"""交回要发出去的段（没有就 None）；没这一行的 (状态, 事件) 一律不动。"""
key = (self.state, event)
if key not in TABLE:
return None
self.state, out = TABLE[key]
return out
def talk(script, drop=()):
"""script: ["B:listen", "A:connect", "A:close", "B:close", "A:timeout", ...]。
一台主机发出的段立刻送到对面（除非它的序号在 drop 里），对面的回信再送回来。
交回 (段的轨迹 ["A>B SYN", ...], {"A": 状态, "B": 状态})。"""
hosts = {"A": Host("A"), "B": Host("B")}
trace = []
def send(frm, seg):
to = "B" if frm == "A" else "A"
trace.append(frm + ">" + to + " " + seg)
if len(trace) in drop:
trace[-1] += " x"
return
reply = hosts[to].step(seg)
if reply:
send(to, reply)
for item in script:
who, event = item.split(":")
seg = hosts[who].step(event)
if seg:
send(who, seg)
return trace, {k: h.state for k, h in hosts.items()}
def segments(nbytes, mss, start=1):
"""把 nbytes 字节切成段，交回 [(序号, 长度)]。序号是这段第一个字节的编号（SYN 占掉了 0 号，数据从 1 起）。"""
out = []
seq = start
while nbytes > 0:
size = min(mss, nbytes)
out.append((seq, size))
seq += size
nbytes -= size
return out
def acks(segs, arrived):
"""接收方的累计确认：每收到一段就回一个 ack = 下一个想要的字节号。乱序到的先存着，序号补齐了 ack 才往前跳。
arrived: 和 segs 等长的 True/False。交回每次到达后回的 ack 列表（丢掉的段没有 ack）。"""
expect = segs[0][0]
have = set()
out = []
for (seq, size), ok in zip(segs, arrived):
if not ok:
continue
have.add((seq, size))
while any(s == expect for s, _ in have):
expect += next(n for s, n in have if s == expect)
out.append(expect)
return out
def rto_trace(samples):
"""RFC 6298 的整数版：第一次 SRTT = R，RTTVAR = R // 2；之后 RTTVAR = (3 * RTTVAR + |SRTT - R|) // 4，SRTT = (7 * SRTT + R) // 8；
RTO = SRTT + 4 * RTTVAR。交回每个样本之后的 (SRTT, RTTVAR, RTO)。"""
out = []
srtt = rttvar = None
for r in samples:
if srtt is None:
srtt, rttvar = r, r // 2
else:
rttvar = (3 * rttvar + abs(srtt - r)) // 4
srtt = (7 * srtt + r) // 8
out.append((srtt, rttvar, srtt + 4 * rttvar))
return out
def backoff(rto, n):
"""连续超时 n 次之后的 RTO：每次翻倍。"""
return rto * (2 ** n)
def resend(segs, plan, rtt, rto):
"""停等 + 超时重传：一段一段发，收到 ack 才发下一段；plan 作用在每一次发送上（. 到 x 丢），丢了等 rto 再发（rto 每连续丢一次翻倍）。
交回 (一共发了几次, 一共花了多少时间)。时间单位随 rtt / rto。"""
sends = 0
clock = 0
k = 0
for _ in segs:
cur = rto
while True:
act = plan[k] if k < len(plan) else "."
k += 1
sends += 1
if act == "x":
clock += cur
cur *= 2
continue
clock += rtt
break
return sends, clock
def fast_retransmit(ack_list):
"""看接收方回的 ack 序列：同一个 ack 连着出现第 4 次（原始 + 3 个重复）时触发快速重传，交回触发时的位置（从 1 数）；没触发交回 0。"""
for i in range(3, len(ack_list)):
if ack_list[i] == ack_list[i - 1] == ack_list[i - 2] == ack_list[i - 3]:
return i + 1
return 0
def rounds(nsegs, wnd):
"""窗口一次能放 wnd 段：nsegs 段要几轮（一轮 = 一个 RTT）。停等就是 wnd = 1。"""
return (nsegs + wnd - 1) // wnd
def flow(nsegs, buf, reads):
"""流控：接收缓冲装得下 buf 段；reads 是应用每轮从缓冲里读走几段的清单（用完就照最后一个值）。
每轮：发送方按上一轮通告的窗口发（不超过剩余），到了先进缓冲，应用再读，接收方通告 rwnd = buf - 缓冲里的。
交回 (每轮通告的 rwnd 列表, 用了几轮)。通告是 0 的那一轮之后，发送方只能干等。最多算 50 轮。"""
inbuf = 0
left = nsegs
rwnd = buf
adv = []
n = 0
while left > 0 or inbuf > 0:
send = min(rwnd, left)
left -= send
inbuf += send
inbuf -= min(reads[n] if n < len(reads) else reads[-1], inbuf)
rwnd = buf - inbuf
adv.append(rwnd)
n += 1
if n >= 50:
break
return adv, n
def cwnd_trace(nrounds, ssthresh, events=None):
"""cwnd 从 1 段起：cwnd < ssthresh 时每轮翻倍（慢启动），否则每轮加 1（拥塞避免）。
events: {轮: "timeout" | "dup"}：超时 → ssthresh = cwnd // 2，cwnd 归 1；三个重复 ACK → ssthresh = cwnd // 2，cwnd = ssthresh（Reno 的快速恢复）。
交回每一轮开始时 cwnd 的列表（长度 nrounds）。"""
events = events or {}
cwnd = 1
out = []
for r in range(1, nrounds + 1):
out.append(cwnd)
ev = events.get(r)
if ev == "timeout":
ssthresh = max(cwnd // 2, 1)
cwnd = 1
elif ev == "dup":
ssthresh = max(cwnd // 2, 1)
cwnd = ssthresh
elif cwnd < ssthresh:
cwnd = min(cwnd * 2, ssthresh)
else:
cwnd += 1
return out
def effective(cwnd, rwnd):
"""真正能发的窗口：拥塞窗口和接收窗口里小的那个。"""
return min(cwnd, rwnd)
SS_SAMPLE = """State Recv-Q Send-Q Local Address:Port Peer Address:Port Process
LISTEN 1 5 127.0.0.1:25001 0.0.0.0:* users:(("python3",pid=46,fd=3))
LISTEN 0 5 127.0.0.1:25000 0.0.0.0:* users:(("python3",pid=27,fd=3))
TIME-WAIT 0 0 127.0.0.1:31010 127.0.0.1:25002
FIN-WAIT-2 5000 0 127.0.0.1:31001 127.0.0.1:25000 users:(("python3",pid=43,fd=3))
CLOSE-WAIT 1 0 127.0.0.1:25000 127.0.0.1:31001 users:(("python3",pid=27,fd=5))
ESTAB 5 0 127.0.0.1:25001 127.0.0.1:31003
TIME-WAIT 0 0 127.0.0.1:25002 127.0.0.1:31011
ESTAB 0 0 127.0.0.1:31003 127.0.0.1:25001 users:(("python3",pid=49,fd=3))
ESTAB 3976 0 127.0.0.1:31000 127.0.0.1:25000 users:(("python3",pid=30,fd=3))
ESTAB 0 0 127.0.0.1:25000 127.0.0.1:31000 users:(("python3",pid=27,fd=4))
"""
def parse_ss(text):
"""把 ss -tanp 的输出拆成一行一个字典：state / recvq / sendq / local / lport / peer / pport / pid（没有进程就是 None）。"""
rows = []
for line in text.splitlines()[1:]:
parts = line.split()
if len(parts) < 5:
continue
lh, lp = parts[3].rsplit(":", 1)
ph, pp = parts[4].rsplit(":", 1)
pid = None
if len(parts) > 5 and "pid=" in parts[5]:
pid = int(parts[5].split("pid=")[1].split(",")[0])
rows.append(dict(state=parts[0], recvq=int(parts[1]), sendq=int(parts[2]),
local=lh, lport=int(lp), peer=ph, pport=pp if pp == "*" else int(pp), pid=pid))
return rows
def by_state(rows):
out = {}
for r in rows:
out[r["state"]] = out.get(r["state"], 0) + 1
return out
def listeners(rows):
return sorted(r["lport"] for r in rows if r["state"] == "LISTEN")
def peers_of(rows, port):
"""服务端口 port 上（Local 是它）的非 LISTEN 连接，交回 [(状态, 对端端口)]。"""
return [(r["state"], r["pport"]) for r in rows if r["lport"] == port and r["state"] != "LISTEN"]
segs = segments(1000, 400)
print(",".join(str(s) for s, _ in segs) + "/" + ",".join(str(a) for a in acks(segs, [True, True, True])))
全部评论