Tcp

一个监听中的 TCP 套接字有两个队列，不是一个；而本文标题里那个 sysctl 管的，恰恰是几乎从不丢东西的那一个。传给 listen(2) 的 backlog 是 accept 队列，装的是已经完成握手、正等应用来取走的连接；SYN 队列装的才是半开请求。把两者混为一谈，就会开错药方。 这件事值得掰开，因为那句著名的建议——把 tcp_max_syn_backlog 调大——基本已经过时。tcp_syncookies 默认就是 1，而按 listen(2) 的说法，一旦启用 syncookies，SYN 队列就没有逻辑上的长度上限，这个设置会被忽略。真正会卡住你的是 accept 队列，以及应用调用 accept() 的速度。 backlog 参数管的是 accept 队列 listen(2) 明确写着，这个行为在 Linux 2.2 就变了：backlog 现在指的是「已完成建立、正等待被 accept 的套接字队列长度」，而不是半开请求的数量。半开队列由 /proc/sys/net/ipv4/tcp_max_syn_backlog 单独限制。而如果 backlog 参数超过 /proc/sys/net/core/somaxconn，它会被静默截断到那个值——Linux 5.4 之后是 4096，之前是 128。所以在默认配置的机器上写 listen(8192)，实际拿到的是 4096，而且不会有任何日志告诉你。 ...

慢启动在 TCP 发送端探索路径可承受的未确认数据量时，逐步扩大拥塞窗口。数据充足且确认模式合适时，经典行为是在多个往返间近似指数增长。“每个 RTT 翻倍”是有用的简化模式，不是无条件计时器规则。窗口变化依赖传输反馈和实现的增长条件，不能只根据连接已经打开多久，就推断它获得了多少发送预算。 坏掉的样子 connect 之后等待，不会自动把拥塞窗口预热。应用没有发送数据时，就没有支持这种增长的数据确认。很小的交换，也可能在用满已有发送预算之前便结束。期待每个短请求都出现大幅翻倍序列，是把受应用数据量限制的传输误认成持续积压的发送端。缺少明显增长，并不自动说明慢启动没有工作或连接存在故障。 现代系统初始窗口也不统一为一到两个段。RFC 6928 描述了允许更大初始窗口的实验性扩展，本次 Linux 发送端起步为十个段。单位同样重要：Linux TCP_INFO 的 snd_cwnd 以段为单位，许多协议说明则用字节描述拥塞窗口。推算未确认数据量之前，应同时查看发送 MSS，避免将段数直接当成字节数或当前实际在途段数。 慢启动并非总要持续到第一次丢失，之后统一彻底缩回起点。慢启动阈值帮助决定启动增长与拥塞避免，具体实现还可能采用额外启动检测；重传超时恢复也不同于快速恢复。把同一条示意曲线套到所有算法、丢失信号和工作负载，即使读到了真实指标，也可能作出错误诊断。首先需要确认当前连接处于哪个阶段。 快速验证 将脚本保存为 check-slow-start.py，准备 Docker 后运行 python3 check-slow-start.py。一次性环境使用 Python 3.11.16 和 iproute2 6.1.0，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套命名空间。客户端和目标通过新建 veth 相连，netem 在两个方向的出口各添加十毫秒延迟，全部变更仅作用于虚拟端点： 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']) 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,sys size=int(sys.argv[1]);listener=socket.socket();listener.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) listener.bind(('192.0.2.2',18080));listener.listen();print('listening',flush=True) with listener.accept()[0] as connection: connection.settimeout(6);remaining=size;digest=hashlib.sha256() while remaining: chunk=connection.recv(min(65536,remaining));assert chunk;remaining-=len(chunk);digest.update(chunk) assert digest.hexdigest()==hashlib.sha256(b'x'*size).hexdigest() connection.sendall((digest.hexdigest()+'\n').encode()) ''' client_code=info_code+r''' import socket,json,time,threading,sys,hashlib size=int(sys.argv[1]);stream=socket.socket();stream.settimeout(6) stream.setsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,b'reno');stream.connect(('192.0.2.2',18080)) initial=tcp_info(stream);time.sleep(.06);idle=tcp_info(stream);assert idle['snd_cwnd']==initial['snd_cwnd'] done=threading.Event();errors=[];samples=[initial,idle];reply=[] def transfer(): try: stream.sendall(b'x'*size) with stream.makefile('rb') as incoming:reply.append(incoming.readline().decode().strip()) except BaseException as error:errors.append(repr(error)) finally:done.set() worker=threading.Thread(target=transfer);worker.start();deadline=time.monotonic()+7 while not done.wait(.005): assert time.monotonic()<deadline,'transfer deadline exceeded' snapshot=tcp_info(stream) if snapshot!=samples[-1]:samples.append(snapshot) worker.join(timeout=1);assert not worker.is_alive() and not errors,errors samples.append(tcp_info(stream));assert reply==[hashlib.sha256(b'x'*size).hexdigest()] assert all(row['total_retrans']==0 for row in samples),samples if size>1:assert max(row['snd_cwnd'] for row in samples)>initial['snd_cwnd']*2,samples else:assert max(row['snd_cwnd'] for row in samples)==initial['snd_cwnd'],samples print(json.dumps({'bytes':size,'algorithm':stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_CONGESTION,16).split(b'\0')[0].decode(), 'before_data':initial,'after_idle_without_data':idle,'samples':samples,'sha256':reply[0]}));stream.close() ''' results={} for label,size in [('short',1),('bulk',1048576)]: service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code,str(size)]);assert line(service).strip()=='listening' results[label]=json.loads(ns('client','python3','-c',client_code,str(size),timeout=10).stdout) service.wait(timeout=2);assert service.returncode==0 print(json.dumps({'egress_delay_each_direction_ms':10,'transfers':results})) """ 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，避免实验悄悄受不同默认拥塞控制器影响。发送之前记录 TCP_INFO，在没有数据的情况下等待六十毫秒，然后再次记录。两次观察窗口均为十个段。这直接检验“新连接只要保持打开，拥塞预算就会按时间增长”的错误预期。增加等待时间与提供可确认的数据，属于不同输入，不能互相替代。 ...

TCP 头部的接收窗口字段只有十六位。窗口缩放让已建立连接依据握手期间确定的移位解释这个字段，从而允许接收端通告超过65535字节的空间。两个方向可以采用不同移位值。它扩展的是接收窗口信息的编码，不会直接扩大每个缓冲区，也不保证发送端始终让那么多数据处于未确认状态。 坏掉的样子 高带宽且 RTT 较长的路径，可能需要远多于未缩放窗口所能描述的在途数据。例如每秒一百兆比特、往返八十毫秒，对应大约一百万字节在途数据；65535字节窗口会形成明显更低的理想速率上限。带宽时延积用于容量估算，不是吞吐量实测，也不保证链路不存在其他约束，不能用一个乘法直接宣布应用应达到某个速度。 发送端还受到拥塞窗口限制，并且必须有应用数据可发。扩大接收窗口编码，无法修复过小的拥塞预算、停止读取的应用、丢包或被阻断的路由。零窗口经过缩放仍然是零。应判断究竟哪个资源限制进展，不能仅因握手里存在缩放选项，就断言连接理应跑满链路，也不能把所有低吞吐问题都归因于缓冲配置。 对协商的误解还会导致抓包解释错误。双方都必须提供选项，缩放才能生效；每个因子用于解释提供它的端点所通告的窗口。SYN 与 SYN-ACK 自身的窗口字段从不缩放。使用反方向的移位值，或把 SYN 窗口也乘上因子，即使数据包字节完全正确，也会算出错误容量。报文值需要结合方向和连接阶段解释。 快速验证 将脚本保存为 check-window-scaling.py，准备 Docker 后运行 python3 check-window-scaling.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套命名空间及临时 proc 挂载。端点通过新建 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 socket,json def read_exact(stream,length): data=b'' while len(data)<length: chunk=stream.recv(length-len(data));assert chunk,'premature EOF';data+=chunk return data listener=socket.socket();listener.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) listener.setsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF,262144) listener.bind(('192.0.2.2',18080));listener.listen();print('listening',flush=True) with listener.accept()[0] as connection: connection.settimeout(4);print(json.dumps({'rcvbuf':connection.getsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF)}),flush=True) assert read_exact(connection,4)==b'ping';connection.sendall(b'pong') assert connection.recv(1)==b'' ''' client_code=r''' import socket,json,sys def read_exact(stream,length): data=b'' while len(data)<length: chunk=stream.recv(length-len(data));assert chunk,'premature EOF';data+=chunk return data s=socket.socket();s.settimeout(4);s.setsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF,65536) s.bind(('192.0.2.1',40400+int(sys.argv[1])));s.connect(('192.0.2.2',18080));s.sendall(b'ping') assert read_exact(s,4)==b'pong' print(json.dumps({'rcvbuf':s.getsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF),'reply':'pong'}),flush=True) assert sys.stdin.readline().strip()=='close';s.close() ''' outcomes={} for enabled in [1,0]: setting('client','net.ipv4.tcp_window_scaling',enabled);setting('target','net.ipv4.tcp_window_scaling',enabled) service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='listening' tap,path=capture(stack,root,'client','c0','tcp port 18080','window'+str(enabled)) client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code,str(enabled)]) client_result=json.loads(line(client));server_result=json.loads(line(service)) diagnostics=ns('client','ss','-tin','state','established').stdout client.stdin.write('close\n');client.stdin.flush();client.wait(timeout=2);service.wait(timeout=2) assert client.returncode==service.returncode==0 observed=packets(tap,path);syn=next(p for p in observed if p['flags']&0x12==0x02) synack=next(p for p in observed if p['flags']&0x12==0x12) advertised=next(p for p in observed if p['sport']==18080 and p['length']==4) if enabled: assert 'wscale' in syn['options'] and 'wscale' in synack['options'] effective=advertised['window']<<synack['options']['wscale'];assert effective>65535 else: assert 'wscale' not in syn['options'] and 'wscale' not in synack['options'];effective=advertised['window'];assert effective<=65535 outcomes[enabled]={'client':client_result,'server':server_result,'syn':syn,'synack':synack, 'server_post_handshake_window_field':advertised['window'],'effective_receive_window':effective,'ss':diagnostics} print(json.dumps(outcomes)) """ 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) 连接建立之前，客户端请求65536字节接收缓冲，服务器监听套接字请求262144字节。脚本读取最终套接字选项值，不假定请求值会原样使用。Linux 的缓冲记账和上限会影响结果，本次客户端报告131072，服务器接受后的套接字报告425984。二者均不能直接等同于线上通告窗口。 ...

Nagle 算法减少 TCP 小报文：当前面数据尚未确认时，它会暂缓额外的小块数据，除非已经积累到足以构成完整段。延迟确认则是接收端的另一种行为，可以短暂推迟 ACK。当应用把一个请求拆成多个很小的写入，而对端必须取得后续字节才能生成响应时，两者的相互作用就会影响延迟。需要明确是哪一部分数据已经发出、哪一部分仍在等待。 坏掉的样子 单独一个小请求，不会自动被 Nagle 延迟。如果没有更早的未确认数据，第一个小段可以直接发送。问题场景是第一部分已在途中，第二个短小部分排在后面等待。如果接收端延迟确认，同时应用又在等完整请求，进展可能暂停，直到确认计时器到期。不能把所有“小请求慢”都归到这个机制，也不能省略前面存在未确认数据这一条件。 这属于有边界的延迟，不是协议无限死锁。ACK 可以释放排队数据，积累足够多数据也可能改变发送决策。观察到的等待既不统一等于额外一个 RTT，也不固定为四十毫秒。接收策略、已有确认状态、请求分帧和网络条件，共同决定真正的报文顺序与时间。把一次实验数值当成所有系统的常数，会误导后续故障判断。 TCP_NODELAY 关闭指定套接字上的 Nagle，却不会关闭所有缓冲、拥塞控制、接收窗口流控或调度延迟。Linux TCP_QUICKACK 请求改变确认模式，但这种模式不是永久设置。把任意一个选项当作万能延迟开关，可能掩盖应用缓冲或其他传输限制；这些问题本来就不在该选项职责内，需要通过对应层次的证据定位。 快速验证 将脚本保存为 check-nagle.py，准备 Docker 后运行 python3 check-nagle.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0 和 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']) setting('client','net.ipv4.tcp_autocorking',0) server_code=r''' import socket,json def read_exact(stream,length): data=b'' while len(data)<length: chunk=stream.recv(length-len(data));assert chunk,'premature EOF';data+=chunk return data listener=socket.socket();listener.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) listener.bind(('192.0.2.2',18080));listener.listen();print('listening',flush=True) for trial in range(2): with listener.accept()[0] as connection: connection.settimeout(3);connection.setsockopt(socket.IPPROTO_TCP,socket.TCP_QUICKACK,0) print('configured',flush=True) request=read_exact(connection,2);assert request==b'ab',request connection.sendall(b'OK');print(json.dumps({'request':request.decode()}),flush=True) ''' service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='listening' tap,path=capture(stack,root,'client','c0','tcp port 18080','nagle') client_code=r''' import json,socket,sys,time def read_exact(stream,length): data=b'' while len(data)<length: chunk=stream.recv(length-len(data));assert chunk,'premature EOF';data+=chunk return data nodelay=int(sys.argv[1]);s=socket.socket();s.settimeout(3) s.setsockopt(socket.IPPROTO_TCP,socket.TCP_NODELAY,nodelay);s.bind(('192.0.2.1',40300+nodelay));s.connect(('192.0.2.2',18080)) print('connected',flush=True);assert sys.stdin.readline().strip()=='go' start=time.monotonic();s.sendall(b'a');time.sleep(.002);s.sendall(b'b');reply=read_exact(s,2) assert reply==b'OK';print(json.dumps({'nodelay':nodelay,'reply':reply.decode(),'elapsed_ms':(time.monotonic()-start)*1000}),flush=True);s.close() ''' outcomes=[] for nodelay in [0,1]: client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code,str(nodelay)]) assert line(client).strip()=='connected' and line(service).strip()=='configured' client.stdin.write('go\n');client.stdin.flush();outcomes.append(json.loads(line(client))) assert json.loads(line(service))['request']=='ab';client.wait(timeout=2);assert client.returncode==0 observed=packets(tap,path);timelines={} for nodelay in [0,1]: port=40300+nodelay;flow=[p for p in observed if port in (p['sport'],p['dport'])] first=next(p for p in flow if p['sport']==port and p['preview']=='61') second=next(p for p in flow if p['sport']==port and p['preview']=='62') assert (second['seq']-first['seq'])%2**32==1 acks=[p for p in flow if p['dport']==port and p['flags']&0x10 and 0<(p['ack']-first['seq'])%2**32<=2] assert acks if not nodelay: intermediate=next(p for p in acks if (p['ack']-first['seq'])%2**32==1) assert first['t']<intermediate['t']<=second['t'] and second['t']-first['t']>.01 else:assert second['t']<=acks[0]['t'] timelines[nodelay]={'first_byte':first,'second_byte':second,'first_ack_covering_data':acks[0], 'data_segment_gap_ms':(second['t']-first['t'])*1000} print(json.dumps({'trials':outcomes,'packet_timelines':timelines,'receiver_quickack_requested':0,'client_autocorking':0})) """ 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) 服务器等待完整请求 ab 才回复 OK。每次试验之前，它在已接受连接上请求非快速确认模式。客户端关闭本命名空间自动 corking，以排除另一种合并机制对比较的影响。父子进程通过明确消息同步，确保接收端先完成套接字设置，再让客户端开始发送应用字节，避免把配置时序竞争误认为算法差异。 ...

监听套接字维护的是两个队列，不是一个。半开连接先待在 SYN 队列里，直到客户端的最后一个 ACK 到达；握手完成的连接随后进入 accept 队列，等着应用调用 accept()。两者各有各的上限，而究竟是哪一个溢出，决定了客户端体验到什么。 SYN Cookie 是为第一个队列准备的。它满了以后，服务器不再分配状态，转而把稍后需要的东西编进序号里，于是握手在零内存占用下完成。这确实是一次救援，同时也是一次降级：被救下来的那条连接，比正常情况下本该建立的那条更差。 两个队列，两种症状 一个 SYN 会分配一个 request socket，把连接推进 SYN_RECV 并回一个 SYN-ACK。客户端的 ACK 再把它移进 accept 队列、状态变成 ESTABLISHED。SYN 队列由 tcp_max_syn_backlog 按监听套接字限流——最小值 128，随内存放大，每个表项大约占 304 字节。accept 队列的上限来自 somaxconn（当前内核默认 4096，Linux 5.4 之前是 128），再与应用程序传给 listen() 的 backlog 共同起作用。 两种溢出长得完全不一样。SYN 队列溢出正是 syncookie 为之而生的场景：连接照样能完成。accept 队列溢出更糟也更安静——在 tcp_abort_on_overflow 取默认值 0 的情况下，服务器直接忽略那个最后的 ACK，于是 connect() 返回成功，客户端把请求发出去，然后再也等不到任何回应。实测：backlog 设为 1、客户端 400 个，400 个全都以为自己连上了，服务器实际只能接受 3 个，另外 397 个在两端都没有任何报错的情况下丢掉了。 ...

对于 Linux TCP 套接字，listen() 的 backlog 关注已经建立、等待应用接受的连接；未完成握手的队列是另一种概念。调用 accept() 会移出一条待处理连接，并返回供应用使用的已连接套接字。即使这个套接字之后长时间保持打开或忙碌，它也不再占用监听器的 accept 队列。因此，等待接受与已经接受后的活动状态需要分别计数。 坏掉的样子 如果连接到达速度超过应用接受速度，待处理队列会填满。根据内核行为和设置，新客户端可能遇到重传延迟或超时，而已经接受的请求仍可能正常。进程活着，或者部分请求成功，都不能证明监听器正足够快地接受新连接。问题发生在接受之前时，仅查看业务处理成功率可能看不到完整失败路径。 请求的 backlog 还受到 net.core.somaxconn 限制；只把应用值调得更大，不能超过这个上限。但可观察的精确队列边界与实现有关，因此不能把整数直接当成跨平台承诺，认为必定恰好有那么多同时 connect 成功。应测量实际部署内核，而不是把一个很小的实验结果推广成通用的加一或减一规则。 队列计数也需要上下文。ListenOverflows 和 ListenDrops 是网络命名空间计数，不会自动解释到某个具体服务。其他监听器也能贡献计数，客户端超时也可能来自 accept 队列之外。将生产失败归因于某个 backlog 设置，或者宣称扩大队列能修复问题之前，应结合队列占用、接受进度和计数共同判断。 快速验证 把脚本保存为 check-listen-backlog.py，准备宿主 Python、Docker 和固定 Python 镜像后运行。真实套接字操作发生在可丢弃 Linux 容器内，网络只保留回环。记录内核为 6.12.76-linuxkit。脚本读取相关设置但不修改它们，并要求 tcp_abort_on_overflow=0，使本次超时观察对应一个明确、可以检查的前提： import json,subprocess image='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' program=r''' from contextlib import ExitStack from pathlib import Path import json,platform,socket def counters(): lines=Path('/proc/net/netstat').read_text().splitlines() for index,line in enumerate(lines): if line.startswith('TcpExt:'): return dict(zip(line.split()[1:],map(int,lines[index+1].split()[1:]))) def queued(port): for line in Path('/proc/net/tcp').read_text().splitlines()[1:]: fields=line.split() if fields[3]=='0A' and int(fields[1].split(':')[1],16)==port: return int(fields[4].split(':')[1],16) raise AssertionError('listener missing') with ExitStack() as stack: def fresh(): s=stack.enter_context(socket.socket());s.settimeout(.25);return s def listener(backlog): s=fresh();s.bind(('127.0.0.1',0));s.listen(backlog);return s somaxconn=int(Path('/proc/sys/net/core/somaxconn').read_text()) abort=int(Path('/proc/sys/net/ipv4/tcp_abort_on_overflow').read_text());assert abort==0 small=listener(1);address=small.getsockname();before=counters();clients=[];timed_out=False for _ in range(8): client=fresh() try:client.connect(address) except TimeoutError:timed_out=True;client.close();break clients.append(client) depth=queued(address[1]);after=counters() assert timed_out and depth==len(clients) and 1<=depth<8,(depth,len(clients)) changes={name:after[name]-before[name] for name in ['ListenOverflows','ListenDrops']} assert changes['ListenOverflows']>0 and changes['ListenDrops']>0,changes accepted=[] for client in clients: server,_=small.accept();stack.enter_context(server);accepted.append(server);server.sendall(b'ready') reader=stack.enter_context(client.makefile('rb'));assert reader.read(5)==b'ready' assert queued(address[1])==0 for _ in range(4): client=fresh();client.connect(address);server,_=small.accept();stack.enter_context(server) server.sendall(b'ready');reader=stack.enter_context(client.makefile('rb'));assert reader.read(5)==b'ready' accepted.append(server);clients.append(client) assert len(accepted)>1 and all(s.fileno()>=0 for s in accepted) larger=listener(16) for _ in range(6):client=fresh();client.connect(larger.getsockname()) assert queued(larger.getsockname()[1])==6 print(json.dumps({'kernel':platform.release(),'somaxconn':somaxconn,'tcp_abort_on_overflow':abort, 'requested_small_backlog':1,'queued_before_accept':depth,'next_connect_timed_out':timed_out, 'counter_changes':changes,'accepted_connections_still_open':len(accepted),'larger_backlog_queued':6},sort_keys=True)) ''' result=subprocess.run(['docker','run','--rm','-i','--network','none',image,'python','-'], input=program,capture_output=True,text=True,timeout=30) assert result.returncode==0,result.stderr record=json.loads(result.stdout);print(json.dumps(record,sort_keys=True)) 第一个监听器请求 backlog 为一，并故意不接受连接。在记录环境里，两次客户端 connect 完成，/proc/net/tcp 也报告两条排队连接；下一次 connect 超过客户端设定的二百五十毫秒超时。脚本记录实际深度，而不是断言每个 Linux 版本在这个参数下都必须接受恰好两条。配置参数、实际队列和客户端结果分别被观察。 ...

TCP keepalive 询问的是：连接安静了一段时间后，对面的 TCP 端点是否仍然响应。它不询问 HTTP handler 是否正常、数据库查询是否推进、后台工作线程是否卡住。即使应用已经停顿，内核仍然可能确认收到的数据。因此，“传输对端能够回应”和“应用给出了有用结果”必须作为两个不同的观察项。 传输层探针不是应用心跳 保活有助于暴露远端连接状态已经消失，或者网络路径悄悄停止传递报文的情况。实现会在满足空闲条件后发送探针，并按照未响应探针的策略最终报告失败；但单个探针丢失不等于对端死亡，不能把网络中的临时丢包直接解释成确定性故障。 探针在应用字节流以下处理，不会变成 HTTP 请求、数据库查询或 WebSocket 心跳。如果连接终止在代理上，探测对象就是代理自身的 TCP 端点，不是代理后面的源站。某些中间设备的连接跟踪可能受探针影响，但不能由此承诺刷新应用的空闲计时，更不能保证所有 NAT 映射都会一直保留。 已经发出而尚未确认的数据还有自己的重传与失败规则，因此保活参数不是繁忙连接的通用总截止时间。正常 FIN、连接重置、应用读取超时和静默丢包是不同事件，恢复动作也未必相同。知道是哪一层报告了什么错误，比统一把它们写成“连接死了”更有用。 配置真正承载请求的 socket RFC 1122 对可选保活机制要求默认关闭，并规定默认空闲间隔不得短于两小时且应可配置。但应用和库可以显式启用并选择更短间隔，Go 的拨号器就有自己的设置。因此，查到操作系统默认值，并不能直接推出当前应用实际使用的策略；还需要沿着创建连接的代码检查有效配置。 Linux 的 tcp_keepalive_time、tcp_keepalive_intvl 和 tcp_keepalive_probes 分别表示探测前的空闲秒数、探针间隔和未响应探针数量，文档默认值为 7200、75、9。启用 SO_KEEPALIVE 后，还可以通过 TCP_KEEPIDLE、TCP_KEEPINTVL、TCP_KEEPCNT 为单个 socket 定制策略。这些名称与接口不是 macOS、Windows 通用的系统参数，不能跨平台照搬命令。 ...

TCP 累计确认标识按序接收时下一个期待的字节。如果某段缺失、后续字节却已到达，累计确认不能跨过这个空洞。选择确认 SACK 补充报告累计边界之后已经收到的字节范围，帮助发送端区分真正缺失的数据与接收端已经排队的数据。它改善恢复算法获得的信息，而不是替代全部丢包判断、拥塞控制或应用交付语义。 坏掉的样子 没有 SACK，并不意味着累计 ACK 强制发送端重传后面的每一个段。发送端仍可根据重复确认、计时器和恢复算法推断丢失，只是反馈不够明确，尤其在存在多个缺口时。把普通 TCP 描述为必须重发整个后缀，是把信息能力的限制误写成强制恢复策略，会夸大差异，也无法准确解释真实抓包中的局部重传。 SACK 块报告已经收到的范围，不是直接列出缺失范围。每块在序列空间中具有包含的左边界和不包含的右边界。缺口需要结合这些块、累计确认及尚未确认数据推导。SACK 不改变普通 ACK 字段含义，应用仍获得有序字节流，而不是乱序的选择确认块。传输层暂存后续字节，并不等于已经把它们交给应用。 该能力还需要对当前连接取得许可。SYN 中的 SACK-Permitted 表示发送该 SYN 的一端允许对端向自己发送 SACK 选项，不证明一次无丢失传输中实际出现过 SACK 块。反过来，缺少 SACK 也不自动证明防火墙剥离选项。判断原因之前应核对端点设置与真实握手，不能把某种中间设备问题当成唯一解释，忽略连接本身的协商条件。 快速验证 将脚本保存为 check-sack.py，准备 Docker 后运行 python3 check-sack.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 支持嵌套网络命名空间及临时 proc 挂载。实验仅使用新建 veth 和命名空间内部设置，不改变宿主现有流量： ...