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 和命名空间内部设置，不改变宿主现有流量：

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'
    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_sack',1);setting('target','net.ipv4.tcp_sack',1)
    setting('client','net.ipv4.tcp_autocorking',0)
    server_code=r'''
import socket,json,hashlib
listener=socket.socket();listener.bind(('192.0.2.2',18080));listener.listen();print('listening',flush=True)
with listener.accept()[0] as connection:
    connection.settimeout(6);data=b''
    while len(data)<12*256:
        chunk=connection.recv(4096);assert chunk;data+=chunk
    expected=b''.join(bytes([i])*256 for i in range(12));assert data==expected
    result={'bytes':len(data),'sha256':hashlib.sha256(data).hexdigest()}
    connection.sendall(json.dumps(result).encode()+b'\n');print(json.dumps(result),flush=True)
'''
    service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='listening'
    nft('target','''table ip one_gap {
 chain input { type filter hook input priority 0; policy accept;
  tcp dport 18080 ip length > 100 numgen inc mod 2147483647 == 1 counter drop comment "second_data_packet"
 }
}
''')
    tap,path=capture(stack,root,'client','c0','tcp port 18080','sack')
    client_code=r'''
import socket,json,time,hashlib
s=socket.socket();s.settimeout(6);s.setsockopt(socket.IPPROTO_TCP,socket.TCP_NODELAY,1)
s.bind(('192.0.2.1',40500));s.connect(('192.0.2.2',18080))
for i in range(12):s.sendall(bytes([i])*256);time.sleep(.003)
with s.makefile('rb') as stream:result=json.loads(stream.readline())
expected=b''.join(bytes([i])*256 for i in range(12))
assert result=={'bytes':len(expected),'sha256':hashlib.sha256(expected).hexdigest()}
print(json.dumps(result));s.close()
'''
    result=json.loads(ns('client','python3','-c',client_code).stdout);assert json.loads(line(service))==result
    dropped=counters('target');assert dropped['second_data_packet']==1,dropped
    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)
    assert syn['options'].get('sack_ok') and synack['options'].get('sack_ok')
    reports=[p for p in observed if p['sport']==18080 and p['options'].get('sack')]
    assert reports,observed
    data_start=(syn['seq']+1)%2**32
    def relative(sequence):return (sequence-data_start)%2**32
    hole=(data_start+256)%2**32
    assert any(relative(p['ack'])==256 and any(256<relative(left)<relative(right)<=3072 for left,right in p['options']['sack']) for p in reports),reports
    retransmissions=[p for p in observed if p['sport']==40500 and p['length'] and p['seq']==hole]
    assert len(retransmissions)>=2,retransmissions
    print(json.dumps({'sack_negotiated_both_directions':True,'missing_byte_sequence':hole,'sack_reports':reports,
        'original_and_retransmitted_hole':retransmissions,'drop_counter':dropped,'verified_delivery':result}))
"""
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)

两端启用 SACK。客户端启用 TCP_NODELAY，并关闭自己命名空间的自动 corking，然后间隔短暂时间发送十二块可区分数据，每块256字节。目标输入链使用递增表达式，仅丢弃第二个匹配数据包。很大的取模值使这段短传输只出现一次受控丢弃，计数器必须恰好报告一次，避免把不可控随机丢失冒充固定实验条件。

捕获确认两个 SYN 消息均包含 SACK-Permitted。第一块到达后，第二块缺失形成空洞。接收端累计确认停在缺口的首字节，而 SACK 块描述后来已经接收的范围。脚本使用本次实际初始序列号检查这些关系，不要求每次运行采用同一个序列值。数字大小本身没有固定意义，范围之间的对应关系才是需要验证的协议事实。

缺失范围在捕获中先以原始发送段出现，随后以相同序列号重传，本次长度为256字节。服务器最终按预期顺序收到全部3072字节，并检查内容，再返回 SHA-256 摘要。客户端独立将该摘要与计划发送的载荷比较。因此，结果不只是“连接没报错”，而是同时确认重传后真正交付了完整且正确的数据。

这些观察证明真实缺口、有效确认信息、重传与成功交付，却没有对比关闭 SACK 的性能，也不能建立统一重传次数改善比例。单独这份捕获，同样无法说明 Linux 丢失检测决策的每个细节。协商 SACK 与选择何时重传，是相关但不同的传输实现部分；不能把出现一个选择确认块，直接解释成某个特定恢复算法必定触发。

解释反馈时，不承诺重传判断绝对完美

RFC 2018 将 SACK 定义为建议性信息，接收端之后可能丢弃此前报告的数据。因此，发送端不能把 SACK 当作永久应用收据，也不能仅因某个范围出现在选项里，就放弃全部重传责任。累计确认仍是判断哪些字节已被 TCP 按序接受的核心依据，业务是否处理或持久保存数据，还属于更高一层的确认契约。

乱序也可能产生缺口和选择报告，即使报文没有永久丢失。应关联序列范围、重传时间、累计进展与捕获位置，再判断是否存在网络损坏。本例的明确输入过滤和精确丢弃计数，为刻意丢失提供直接证据；原因不明的生产抓包通常没有这种实验控制，不能照搬故障归因，需要补充对应观察。

事件排查应保留握手选项，以及足以重建相关范围的数据交换，分别比较应用交付和传输恢复。正确重组可能使短暂网络丢失不影响应用内容，却仍影响延迟或吞吐量。性能问题需要代表性负载测量，本例则验证协议信息与最终字节交付。这样既能说明 SACK 的作用，也不会用一个小型成功样本替代完整网络性能结论。

来源