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，以排除另一种合并机制对比较的影响。父子进程通过明确消息同步，确保接收端先完成套接字设置，再让客户端开始发送应用字节，避免把配置时序竞争误认为算法差异。

第一次试验把 TCP_NODELAY 设为零。客户端发送 a，等待两毫秒，再发送 b。这些写入是应用动作，并不保证立即产生分离报文。抓包显示第一个字节离开、覆盖它的 ACK 到达，随后第二个字节才离开。程序检查这个顺序，而不是只要响应变慢就宣称 Nagle 一定是原因。应用写入时间与实际传输时间必须分开观察。

本次记录中，两段数据相隔约四十一毫秒，完整交互也花费接近这个时间。第二次启用 TCP_NODELAY，保持相同分帧与刻意加入的等待。第二个字节在首个数据确认之前离开，两段间隔约两毫秒。服务器在两次试验中都取得完整请求，并返回相同有效回复。比较改变的是传输时机，不能将第一种情况写成请求永远无法完成。

这些时间描述受控运行，不是生产延迟保证。脚本要求在所述设置下看到延迟确认交互，并检查对应报文先后关系；其他内核或接收策略需要重新调查。流式读取明确累积所需字节数，因为 recv 即使给定长度，通常也不保证一次返回完整应用消息。这个边界与 Nagle 是否启用无关，应用必须始终正确处理分段读取。

一起决定请求分帧与套接字策略

对于已经完整存在内存中的小型逻辑请求，可以考虑把完整表示一起写出，避免接收端等后续片段，同时保留有价值的传输合并。必须及时交付独立小更新的交互流，则可能适合 TCP_NODELAY。选择应依据负载的消息边界与实测行为，不能对所有连接使用同一条经验规则，也不能通过改变分帧后忘记重新检查应用语义。

请求周期性停顿时，同时捕获数据和确认，判断第二部分仍停在应用缓冲、已排入发送端，还是已经到达对端。应用中的一个超时，无法区分这些阶段。将抓包与套接字配置、接收端认定请求完整的时刻关联，才能知道需要改变哪一处等待，而不是反复调整与问题没有关系的传输参数。

大量发送微小更新时，还要考虑包速率与带宽开销。减少不必要的合并等待，可以改善交互，却也可能增加每次操作产生的段数。应同时测量相关响应延迟和资源成本。这里有用的结果，是对特定交互建立有证据支持的策略：两次请求都成功，而实际观察到的发送顺序不同，足以解释本例等待差异，但不构成统一性能承诺。

来源