TCP

TCP 三次握手建立字节流的两个方向。每端提供自己的初始序列号，并确认另一端的序列号。这是两个独立序列空间，不是共享墙上时钟；应用建立连接不需要两台机器时钟同步。之后的序列号用于标识字节，支持连接内部的排序、重复处理、确认和重传。把序列号与时间混为一谈，会误解握手真正建立的协议状态。 坏掉的样子 客户端发送带初始序列号的 SYN，服务器回复 SYN-ACK，给出自己的初始序列号并确认客户端 SYN，客户端再确认服务器响应。即使没有应用数据，SYN 也占用序列空间中的一个位置，所以对应确认号增加一。将这些数值当成时间戳，会错误解释报文，还可能把排障引向无关的时钟同步问题，忽略真正缺失的确认或路由。 两端不会在完全相同的时刻改变状态。客户端收到 SYN-ACK 后，connect 可能成功返回，但最后一个确认在到达服务器前丢失。服务器仍可能停留在 SYN-RECEIVED，Linux 工具显示为 SYN-RECV。后续携带适当确认的数据也能完成交换，因此单独第三个报文缺失，并不必然导致永久失败，不能把抓包中的固定三包模式当成唯一合法过程。 TCP 建立成功也不代表应用健康或对端身份可信。服务可能接受连接却无法生成有效响应，代理可能完成客户端这一侧连接，却无法建立另一个独立上游连接。初始序列号选择有助于区分连接实例、抵御预测攻击，但不是身份认证协议，也不会加密字节流。传输状态、应用结果和安全身份需要各自对应的验证，不能由一项成功代替全部。 快速验证 将脚本保存为 check-tcp-handshake.py，准备 Docker 后运行 python3 check-tcp-handshake.py。一次性容器使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套网络命名空间。客户端和目标由新建 veth 相连，不移动宿主接口或发布端口： from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n' program=r""" from contextlib import ExitStack,contextmanager from pathlib import Path from tempfile import TemporaryDirectory import json,os,select,signal,subprocess,time def run(*args,check=True,input=None,timeout=8): result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout) if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr) return result def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs) def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs) def stop(process): if process.poll() is None: process.terminate() try:process.wait(timeout=2) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2) def line(process,timeout=3): assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded' value=process.stdout.readline();assert value,('child exited without a response',process.stderr.read()) return value def spawn(stack,command): process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) stack.callback(stop,process);return process @contextmanager def lab(names): with TemporaryDirectory(prefix='net-path-') as directory,ExitStack() as stack: root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc)) stack.callback(run,'umount',str(proc)) for name in names: run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False) ip(name,'link','set','lo','up') def setting(name,key,value): assert name in names path=proc/'sys'/key.replace('.','/') return ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r}); p.write_text({str(value)!r}); print(p.read_text().strip())').stdout.strip() yield root,stack,setting def wire(left,left_if,right,right_if,left_addresses,right_addresses): run('ip','link','add',left_if,'type','veth','peer','name',right_if) for name,interface,addresses in [(left,left_if,left_addresses),(right,right_if,right_addresses)]: run('ip','link','set',interface,'netns',name) for address in addresses:ip(name,'address','add',address,'dev',interface,*(['nodad'] if ':' in address else [])) ip(name,'link','set',interface,'up') def nft(name,script):return ns(name,'nft','-f','-',input=script) def counters(name): data=json.loads(ns(name,'nft','-j','list','ruleset').stdout);found={} for entry in data['nftables']: rule=entry.get('rule',{}) if 'comment' in rule: for expression in rule['expr']: if 'counter' in expression:found[rule['comment']]=expression['counter']['packets'] return found def capture(stack,root,name,interface,expression,label): path=root/(label+'.pcap') process=subprocess.Popen(['ip','netns','exec',name,'tcpdump','--immediate-mode','-Z','root','-U','-n','-i',interface,'-w',str(path),expression], stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,text=True) stack.callback(stop,process) deadline=time.monotonic()+3 while time.monotonic()<deadline: assert process.poll() is None,'tcpdump exited before listening' if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path raise AssertionError('tcpdump did not become ready') def packets(process,path): import struct,socket process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3) assert process.returncode==0,diagnostic raw=path.read_bytes();assert raw[:4]==bytes.fromhex('d4c3b2a1') and struct.unpack_from('<I',raw,20)[0]==1 offset=24;result=[] while offset<len(raw): seconds,micros,size,original=struct.unpack_from('<IIII',raw,offset);offset+=16 frame=raw[offset:offset+size];offset+=size if frame[12:14]!=b'\x08\x00' or frame[23]!=6:continue ihl=(frame[14]&15)*4;length=struct.unpack_from('!H',frame,16)[0];tcp=14+ihl sport,dport,seq,ack=struct.unpack_from('!HHII',frame,tcp);header=(frame[tcp+12]>>4)*4 opts={};cursor=tcp+20 while cursor<tcp+header: kind=frame[cursor] if kind==0:break if kind==1:cursor+=1;continue option_length=frame[cursor+1];data=frame[cursor+2:cursor+option_length];assert option_length>=2 if kind==2:opts['mss']=struct.unpack('!H',data)[0] elif kind==3:opts['wscale']=data[0] elif kind==4:opts['sack_ok']=True elif kind==5:opts['sack']=[list(pair) for pair in struct.iter_unpack('!II',data)] cursor+=option_length body=frame[tcp+header:14+length] result.append({'t':seconds+micros/1e6,'src':socket.inet_ntoa(frame[26:30]),'dst':socket.inet_ntoa(frame[30:34]), 'sport':sport,'dport':dport,'seq':seq,'ack':ack,'flags':frame[tcp+13],'window':struct.unpack_from('!H',frame,tcp+14)[0], 'length':len(body),'options':opts,'preview':body[:24].hex()}) return result with lab(['client','target']) as (root,stack,setting): wire('client','c0','target','t0',['192.0.2.1/30'],['192.0.2.2/30']) server_code=r''' import socketserver class Handler(socketserver.StreamRequestHandler): def handle(self): data=self.rfile.readline();self.wfile.write(b'echo:'+data);self.wfile.flush() class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True server=Server(('192.0.2.2',18080),Handler);print('ready',flush=True);server.serve_forever() ''' service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='ready' tap,path=capture(stack,root,'client','c0','tcp port 18080','handshake') normal=ns('client','python3','-c', 'import socket; s=socket.socket();s.settimeout(2);s.bind(("192.0.2.1",40100));s.connect(("192.0.2.2",18080));s.sendall(b"normal\\n");print(s.makefile("rb").readline().decode().strip());s.close()') assert normal.stdout.strip()=='echo:normal' nft('client','''table ip missing_ack { chain output { type filter hook output priority 0; policy accept; tcp dport 18080 tcp flags & (syn|ack|rst|fin) == ack counter drop comment "missing_final_ack" } } ''') client_code=r''' import json,socket,sys s=socket.socket();s.settimeout(.25);s.bind(('192.0.2.1',40101));s.connect(('192.0.2.2',18080)) print(json.dumps({'connect_returned':True}),flush=True);s.sendall(b'deferred\n') try:s.recv(64);raise AssertionError('unexpected application response') except TimeoutError:print(json.dumps({'application_timeout':True}),flush=True) assert sys.stdin.readline().strip()=='restore';s.settimeout(5) received=b'' while not received.endswith(b'\n'): chunk=s.recv(64);assert chunk,'premature EOF';received+=chunk assert received==b'echo:deferred\n',received print(json.dumps({'recovered':received.decode().strip()}),flush=True);s.close() ''' client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code]) connected=json.loads(line(client));failure=json.loads(line(client));assert connected['connect_returned'] and failure['application_timeout'] pending=ns('target','ss','-nHt','state','syn-recv').stdout;assert '192.0.2.1:40101' in pending,pending blocked=counters('client');assert blocked['missing_final_ack']>0 nft('client','delete table ip missing_ack');client.stdin.write('restore\n');client.stdin.flush() recovery=json.loads(line(client,6));assert recovery['recovered']=='echo:deferred' client.wait(timeout=2);assert client.returncode==0 observed=packets(tap,path);flow=[p for p in observed if 40100 in (p['sport'],p['dport'])] syn=next(p for p in flow if p['flags']&0x12==0x02) synack=next(p for p in flow if p['flags']&0x12==0x12) third=next(p for p in flow if p['sport']==40100 and p['flags']&0x12==0x10) assert synack['ack']==(syn['seq']+1)%2**32 assert third['seq']==(syn['seq']+1)%2**32 and third['ack']==(synack['seq']+1)%2**32 print(json.dumps({'three_messages':[syn,synack,third],'normal_reply':normal.stdout.strip(), 'client_connect_returned':connected['connect_returned'],'application_timeout':failure['application_timeout'], 'server_still_syn_recv':pending.strip(),'drop_counters':blocked,'after_ack_filter_removed':recovery})) """ name='xbcatm-tcpflow-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN', '--memory','256m','--pids-limit','128',name,'python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) 第一条连接使用客户端端口 40100 和目标端口 18080，发送换行结束的请求，收到预期回显。抓包提供真实 SYN、SYN-ACK 和最终 ACK 数值。断言检查服务器确认号等于客户端序列号加一，第三条消息确认号等于服务器序列号加一，并对三十二位序列空间使用取模运算。这样验证的是协议关系，而不是要求每次随机数相同。 ...

HTTP 持久连接允许多次请求与响应复用同一条连接，减少重复的 TCP 建连；使用 HTTPS 时，也可能省去新的 TLS 协商。收益取决于网络延迟、响应大小、服务端处理时间以及池中是否已经有可用连接。连接复用不会消除业务计算，也没有承诺某个套接字永远保持打开。讨论优化之前，先明确观察的是哪一段连接、哪个协议版本。 完整报文才允许复用 HTTP/1.1 默认允许持久连接，除非某一方发送 Connection: close，或实际关闭传输连接。通常不必额外添加 Connection: keep-alive。客户端连接池可以保留空闲连接，供符合源、代理、证书和池策略的后续请求使用。连续执行两次独立的命令行程序，通常不会共享内存中的连接池，因而不能凭两次响应成功就断言已经复用。 复用的前提是双方都能识别每条消息的结束位置。响应可以通过准确的 Content-Length、分块传输编码，或无正文响应的协议规则界定边界。如果只能依赖关闭连接表示正文结束，这条连接自然无法留给下一次请求。客户端需要读完响应正文；服务器需要读完请求正文，或者关闭连接。残留字节一旦被当成下一条消息解析，问题就从性能退化变成报文错位。 HTTP 持久连接与 TCP keepalive 也是两回事。后者在传输连接空闲足够久后发送探测，帮助识别失联的对端，不负责 HTTP 消息分界，也不能覆盖代理或应用自己的空闲超时。仅仅打开操作系统的探活选项，并不会让一个已经被代理回收的连接重新可用。排查时应分别记录应用层关闭、传输层错误和连接池回收动作。 顺序复用还不同于流水线请求。客户端可以等前一个响应完成后再发送下一条请求，同时一直使用相同连接。HTTP/1.1 流水线允许提前发送请求，但响应仍须按序返回。因此，仅有持久连接并不具备 HTTP/2 多路复用提供的独立响应流，不能把两者的并发收益混为一谈。 每一跳管理自己的超时 空闲超时按照具体实现的规则衡量无活动时间，请求截止时间则限制一次操作能等待多久，两者需要独立配置。反向代理同时拥有客户端连接和上游连接，两侧可能使用不同协议、连接池和超时。因此，浏览器到边缘复用了连接，并不能证明边缘到源站也复用；看到一侧握手减少，也不能直接解释另一侧的延迟变化。 连接池应结合已知的服务端限制提前淘汰空闲连接，但始终可能发生竞态：客户端刚发送下一条请求，服务器恰好开始关闭。监控应同时包含新建连接数、成功复用次数、连接重置、池占用和请求延迟。保留更久可以减少握手，也会占用文件描述符与内存。应依据实际请求间隔和资源预算确定保留策略，不能仅凭“站点很忙”推导一个固定超时。 坏掉的样子 正文长度错误可能让读取方一直等待，或让下一条响应的开头混进上一条正文。遇到请求挂起时先检查消息分界，再考虑调大超时。另一个常见故障是调用方放弃流式响应，却既不读完也不关闭，导致池中的连接长期占用。此时服务器还有处理能力，新的请求却全在客户端等空位，扩大服务端容量无法解决这个问题。 陈旧连接报错不代表服务器没有执行请求。如果支付已经提交而响应在回程丢失，自动重连并原样重发可能重复扣款。是否重试必须依照方法语义和业务幂等契约判断，不能把传输异常当成回滚证明。HTTP/2 虽然可以在连接上复用多个流，却禁止 Connection、Keep-Alive 等 HTTP/1.x 专属字段；把这些头搬进 HTTP/2 配置没有优化意义。 快速验证 下面是可直接运行的 Python 3 实验：服务器绑定系统分配的回环端口，客户端连续请求四次，结束后关闭测试服务器。它不需要外部服务；内置 HTTP 服务器只用于本地验证，不是部署配置。 ...

TCP 最大段大小选项 MSS，按照其定义的开销规则，通告接收端可接受的 TCP 数据大小。两个方向分别通告，数值不必相同。路由器转发 SYN 消息时可以减小这些值，使后续数据段适应受限路径。这改变的是相关连接的 TCP 行为，不会修改底层链路 MTU。把选项调整与网络本身修复区分开，才能准确说明它解决了什么。 坏掉的样子 中间瓶颈可能比两个端点接口都小。如果经典路径 MTU 反馈被过滤，连接可以成功建立，却在较大数据上停滞。降低 MSS 能帮助该 TCP 流量避开超大报文，但缺失的反馈仍然缺失。若把这个变通措施当作完整网络修复，其他协议与流量模式仍可能暴露原有问题，健康检查也可能继续遗漏受影响场景。 MSS 不是完整 IP 包大小。IPv4 路径 MTU 为 1280、普通 IPv4 与 TCP 基础头各占 20 字节时，基本计算得到通告 MSS 1240。形成真实数据段时，还要为实际 IP 或 TCP 选项另外留空间。把通告值直接等同于最终线路报文长度，可能使看似保守的设置仍然超限。应明确每个数字包含哪些头部，而不是只套一个固定减法。 方向和时机也很重要。只修改一个 SYN 方向，另一端通告仍保持原样；更新规则，也不会改写已有连接早先交换过的选项。真正的限制应降低过大通告，而不是抬高对端原本更小的上限。策略应针对真实瓶颈和相关流量，不能把一个硬编码值视为所有网络都合适，也不能假定旧连接会自动采用新配置。 快速验证 将脚本保存为 check-mss-clamp.py，准备 Docker 后运行 python3 check-mss-clamp.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持四个嵌套命名空间。端点链路 MTU 为 1500，两台路由器之间的链路为 1280： ...

路径 MTU 是某条具体路由上各链路 MTU 的最小值。发送端不能只根据自身接口确定它，因为中间链路可能只能承载更小的数据包。经典 IPv4 路径 MTU 发现设置禁止分片标志，通过 ICMP“需要分片”反馈降低估计包大小。路由能够到达目的地，与发送大小适合整条路径，是两项不同条件，前者成立不能自动证明后者。 坏掉的样子 隧道或其他受限中间链路，可能使大传输失败，而 TCP 握手与小请求仍然成功。小报文能够通过瓶颈，因此这些成功没有验证完整大小的数据。反复只测短 ping 或微小健康响应，就可能漏掉影响上传、大响应或持续传输的路径 MTU 问题。应让检查包含真实负载中会触发限制的大小，避免健康探针与实际业务长期脱节。 过滤必要 ICMP 反馈，会让过大报文进入黑洞。发送端观察到没有进展，却缺少指出更小 MTU 的经典信号。其他探测机制可能提供帮助，取决于传输协议和配置，因此不能断言所有 TCP 实现遇到这种过滤都会永久卡住。作出结论前，应查明实际启用了哪一种发现或黑洞恢复机制，不能把某个实验控制条件省略后推广到所有环境。 路径信息还可能被缓存。成功发现之后进行的测试，即使防火墙改变，也可能与全新客户端表现不同。复用已有状态会隐藏反馈路径损坏。路由变化和非对称返回路径进一步增加复杂性，所以需要记录方向、来源、目的地，以及估计是在什么条件下学到的。配置变化前后的结果，只有在这些前提清楚时才适合直接比较。 快速验证 以下命名空间、抓包与套接字集成原样保留，本轮未重跑；观察属于历史记录，离线检查不能复现网络结果。 将脚本保存为 check-path-mtu.py，准备 Docker 后运行 python3 check-path-mtu.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持四个嵌套命名空间，两台路由器通过新建 veth 连接端点，不修改宿主路由或接口： ...

TIME-WAIT 是 TCP 在某些有序关闭之后继续保留的协议状态。它为必要时重发最后确认、完成关闭交换留出时间，也帮助避免早先连接实例的延迟报文被误认成新连接数据。相关身份包含两端地址和端口，不能把一个 TIME-WAIT 条目简单理解成对某个本地端口号的全局锁。连接四元组和单独端口，是不同范围的概念。 坏掉的样子 普通关闭中，首先发起关闭的一端通常在确认对端 FIN 后进入 TIME-WAIT。这一端可能是客户端，也可能是服务器；同时关闭还可能涉及双方。假定 TIME-WAIT 永远属于服务器，会掩盖真正制造连接周转的工作负载。应查看状态及两端四元组，不能从应用角色推断谁先关闭，也不能把客户端保留该状态直接视为异常。 高频连接周转可能限制到某个目的地的新连接，但 TIME-WAIT 数量很大，本身不证明资源耗尽或应用文件描述符泄漏。应用关闭套接字之后，操作系统仍可保留协议状态。源地址、可用端口、目的地多样性、显式绑定、复用选项与内核行为，都会影响实际限制。诊断资源耗尽之前，应同时测量连接失败与状态分布，而不是只盯一个总数。 混淆 TIME-WAIT 和 FIN-WAIT-2 会导致无效调参。Linux 的 tcp_fin_timeout 控制另一种孤立套接字情形，不是通用的 TIME-WAIT 时长开关。关于 tcp_tw_recycle 的旧建议，也不适用于当前 Linux。复用受到协议安全条件约束，其中包含时间戳相关检查；如果不了解真实套接字路径，仅修改一个参数，就可能产生错误的行为预期。 快速验证 将脚本保存为 check-time-wait.py，准备 Docker 后运行 python3 check-time-wait.py。一次性环境使用 Python 3.11.16 和 iproute2 6.1.0，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套命名空间及临时 proc 挂载。客户端与目标通过新建 veth 相连，参数变更只作用于这些范围，不影响宿主服务或接口： ...

TCP 拥塞控制限制发送端向容量有限、可能被共享的网络注入数据。拥塞窗口约束未确认数据量，采用 pacing 时还会控制发送在时间上的分布。接收端通告窗口施加的是另一类流量控制限制。这些机制相互作用，却描述不同约束，不能笼统合并成一个“发送速度设置”，否则很难判断真正限制当前连接的是哪一部分。 坏掉的样子 两端不需要选择完全相同的拥塞控制算法。每端负责控制自己发送的流量，因此 Reno 客户端能够与 CUBIC 服务器通信。TCP 建立过程协商传输选项，没有要求双方发送算法名称一致。把不同控制器名称诊断成协议不匹配，会分散注意力，掩盖真正的路由、缓冲或应用故障。算法选择需要按方向和具体套接字理解。 算法是否可用，也是需要核对的运行时事实。配置字符串写了某个控制器，不证明内核支持它或允许调用进程选择它。Linux 的 TCP_CONGESTION 按套接字设置，受可用和允许算法约束。应检查调用结果，并回读实际名称，不能默认请求标签已生效。选择失败与选择成功后性能不合适，属于不同问题，需要不同处理。 丢失恢复不存在适用于所有情况的统一曲线。重传超时、快速恢复、显式拥塞通知和基于模型的估计，会因控制器不同而产生不同响应。丢失可能来自拥塞，也可能来自其他原因，发送端仍须按算法规则作出反应。增加缓冲并不能自动修复持续丢包或等待确认的连接，也不能替代对具体恢复状态和触发信号的检查。 快速验证 将脚本保存为 check-congestion-control.py，准备 Docker 后运行 python3 check-congestion-control.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0 和 nftables 1.0.6，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套命名空间。新建 veth 两个方向出口各增加十毫秒延迟，不修改宿主接口或生产流量： from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n' program=r""" from contextlib import ExitStack,contextmanager from pathlib import Path from tempfile import TemporaryDirectory import json,os,select,signal,subprocess,time def run(*args,check=True,input=None,timeout=8): result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout) if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr) return result def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs) def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs) def stop(process): if process.poll() is None: process.terminate() try:process.wait(timeout=2) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2) def line(process,timeout=3): assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded' value=process.stdout.readline();assert value,('child exited without a response',process.stderr.read()) return value def spawn(stack,command): process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) stack.callback(stop,process);return process @contextmanager def lab(names): with TemporaryDirectory(prefix='net-path-') as directory,ExitStack() as stack: root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc)) stack.callback(run,'umount',str(proc)) for name in names: run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False) ip(name,'link','set','lo','up') def setting(name,key,value): assert name in names path=proc/'sys'/key.replace('.','/') return ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r}); p.write_text({str(value)!r}); print(p.read_text().strip())').stdout.strip() yield root,stack,setting def wire(left,left_if,right,right_if,left_addresses,right_addresses): run('ip','link','add',left_if,'type','veth','peer','name',right_if) for name,interface,addresses in [(left,left_if,left_addresses),(right,right_if,right_addresses)]: run('ip','link','set',interface,'netns',name) for address in addresses:ip(name,'address','add',address,'dev',interface,*(['nodad'] if ':' in address else [])) ip(name,'link','set',interface,'up') def nft(name,script):return ns(name,'nft','-f','-',input=script) def counters(name): data=json.loads(ns(name,'nft','-j','list','ruleset').stdout);found={} for entry in data['nftables']: rule=entry.get('rule',{}) if 'comment' in rule: for expression in rule['expr']: if 'counter' in expression:found[rule['comment']]=expression['counter']['packets'] return found info_code=r''' import struct def tcp_info(stream): raw=stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_INFO,104);assert len(raw)==104 offsets={'snd_mss':16,'unacked':24,'rtt_us':68,'snd_ssthresh':76,'snd_cwnd':80,'total_retrans':100} return {'ca_state':raw[1],**{key:struct.unpack_from('=I',raw,offset)[0] for key,offset in offsets.items()}} ''' with lab(['client','target']) as (root,stack,setting): wire('client','c0','target','t0',['192.0.2.1/30'],['192.0.2.2/30']) available=ns('client','sysctl','-n','net.ipv4.tcp_available_congestion_control').stdout.split();assert {'reno','cubic'}<=set(available),available for name,interface in [('client','c0'),('target','t0')]: ns(name,'tc','qdisc','add','dev',interface,'root','netem','delay','10ms','limit','10000') server_code=r''' import socket,json,hashlib listener=socket.socket();listener.setsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,b'cubic') listener.bind(('192.0.2.2',18080));listener.listen();print('listening',flush=True) with listener.accept()[0] as connection: connection.settimeout(8) print(json.dumps({'algorithm':connection.getsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,16).split(b'\0')[0].decode()}),flush=True) for size in [65536,16384]: data=b'' while len(data)<size: chunk=connection.recv(size-len(data));assert chunk;data+=chunk assert data==b'x'*size connection.sendall((hashlib.sha256(data).hexdigest()+'\n').encode()) ''' service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='listening' client_code=info_code+r''' import socket,json,sys,time,hashlib with socket.socket() as unsupported: try:unsupported.setsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,b'xbcatm_absent');raise AssertionError('unexpected available algorithm') except OSError as error:selection_failure={'errno':error.errno,'type':type(error).__name__} assert selection_failure['errno']==2,selection_failure stream=socket.socket();stream.settimeout(8);stream.setsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,b'reno') stream.bind(('192.0.2.1',40600));stream.connect(('192.0.2.2',18080)) with stream.makefile('rb') as incoming: stream.sendall(b'x'*65536);assert incoming.readline().decode().strip()==hashlib.sha256(b'x'*65536).hexdigest() baseline=tcp_info(stream);assert baseline['snd_cwnd']>1 algorithm=stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,16).split(b'\0')[0].decode() print(json.dumps({'baseline':baseline,'algorithm':algorithm,'unavailable_algorithm':selection_failure}),flush=True) assert sys.stdin.readline().strip()=='drop';stream.sendall(b'x'*16384) deadline=time.monotonic()+4;samples=[] while time.monotonic()<deadline: state=tcp_info(stream);samples.append(state) if state['total_retrans']>baseline['total_retrans'] and state['snd_cwnd']==1 and state['ca_state']==4:break time.sleep(.01) else:raise AssertionError(('no timeout recovery observed',samples)) print(json.dumps({'during_blackout':state}),flush=True) assert sys.stdin.readline().strip()=='restore' assert incoming.readline().decode().strip()==hashlib.sha256(b'x'*16384).hexdigest() print(json.dumps({'after_restore':tcp_info(stream),'verified_bytes':65536+16384}),flush=True) stream.close() ''' client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code]) baseline=json.loads(line(client));server_result=json.loads(line(service));assert baseline['algorithm']=='reno' and server_result['algorithm']=='cubic' nft('target','''table ip blackout { chain input { type filter hook input priority 0; policy accept; tcp sport 40600 tcp dport 18080 ip length > 60 counter drop comment "data_dropped" } } ''') client.stdin.write('drop\n');client.stdin.flush();loss=json.loads(line(client,5)) count=counters('target');assert count['data_dropped']>0 nft('target','delete table ip blackout');client.stdin.write('restore\n');client.stdin.flush() recovery=json.loads(line(client,8));client.wait(timeout=2);service.wait(timeout=2);assert client.returncode==service.returncode==0 print(json.dumps({'available':available,'client':baseline,'server':server_result,'loss':loss,'drop_counter':count,'recovery':recovery})) """ name='xbcatm-tcpflow-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN', '--memory','256m','--pids-limit','128',name,'python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) 脚本首先验证 Reno 与 CUBIC 可用。刻意选择一个不存在的控制器会失败，返回 ENOENT，错误号二，Python 表现为 FileNotFoundError。这个异常指控制器选择不可用，不是应用数据文件丢失。真实客户端明确选择 Reno，服务器监听套接字选择 CUBIC，接受后的套接字报告继承了该选择，分别验证配置意图和最终生效状态。 ...

持久连接允许多次请求和响应共享同一条连接，减少重复的 TCP 建立和 TLS 协商。但这是协议提供的能力，不是某个套接字永远可用的承诺。下一次请求能否真正复用，还取决于消息边界是否完整、客户端连接池策略、服务端行为和连接当前状态；单看请求成功或一次响应较快，都不足以证明走了复用路径。 默认允许，不代表永不关闭 HTTP/1.1 默认允许持久连接，客户端通常不需要发送 Connection: keep-alive 才能启用它。Connection: close 则表示当前交换之后不再继续使用这条连接。HTTP/1.0 的持久连接规则不同，不能把旧版本的协商方式复制到所有现代请求里，也不能把“没有 keep-alive 字段”直接判成配置错误。 复用首先需要明确的消息结束位置。准确的 Content-Length、分块编码，或无正文响应规则，都可以界定边界。若响应只能依靠关闭连接来表示结束，自然无法留下这条连接继续使用。接收方必须先消费完当前消息，再把后续字节解释成下一条；只收到 HTTP 200 状态，还不能证明响应正文已经完整到达。 已建立的 HTTPS 连接被复用时，不会重新进行 TLS 握手；使用缓存的 TLS 会话建立另一条连接，则仍然需要握手。TCP keepalive 又是另一回事，它的空闲探测不会刷新 HTTP 消息边界，也不能否定应用主动关闭连接的决定。HTTP/2 同样可以复用连接，但并不使用 HTTP/1.x 的 Connection 字段语义。 分别管理连接池与一次操作 空闲套接字仍占用描述符、缓冲区和实现相关的状态。Go 的 MaxIdleConnsPerHost 控制每个主机保留多少空闲连接，MaxConnsPerHost 则同时限制正在拨号、活动和空闲的连接总数。前者不是总并发上限。可复用连接池由共享的 Transport 持有，连续调用若各自建立无关的 transport，就会丢掉原本可以复用的连接资源。 ...