路径 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 连接端点，不修改宿主路由或接口：

from pathlib import Path
from tempfile import TemporaryDirectory
import subprocess,json,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 spawn(stack,command):
    process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    stack.callback(stop,process);return process
def line(process,timeout=3):
    assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded'
    value=process.stdout.readline();assert value,process.stderr.read();return value
@contextmanager
def lab():
    with TemporaryDirectory(prefix='mtu-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 ['client','r1','r2','target']:
            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 ['client','r1','r2','target']
            path=proc/'sys'/key.replace('.','/')
            ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r});p.write_text({str(value)!r});assert p.read_text().strip()=={str(value)!r}')
        yield root,stack,setting
def topology(setting,ipv6=False):
    links=[('client','c0','r1','a0',1,2,1,1500),('r1','a1','r2','b0',2,1,2,1280),('r2','b1','target','t0',3,1,2,1500)]
    for left,left_if,right,right_if,network,left_host,right_host,mtu in links:
        run('ip','link','add',left_if,'type','veth','peer','name',right_if)
        for name,interface,host in [(left,left_if,left_host),(right,right_if,right_host)]:
            run('ip','link','set',interface,'netns',name);ip(name,'link','set',interface,'mtu',str(mtu))
            address=f'2001:db8:{network}::{host}/64' if ipv6 else f'10.210.{network}.{host}/30'
            ip(name,'addr','add',address,'dev',interface,*(['nodad'] if ipv6 else []));ip(name,'link','set',interface,'up')
    family='-6' if ipv6 else '-4'
    for name in ['r1','r2']:setting(name,'net.ipv6.conf.all.forwarding' if ipv6 else 'net.ipv4.ip_forward',1)
    routes=[('client','default','2001:db8:1::1' if ipv6 else '10.210.1.1'),('target','default','2001:db8:3::1' if ipv6 else '10.210.3.1'),
        ('r1','2001:db8:3::/64' if ipv6 else '10.210.3.0/30','2001:db8:2::2' if ipv6 else '10.210.2.2'),
        ('r2','2001:db8:1::/64' if ipv6 else '10.210.1.0/30','2001:db8:2::1' if ipv6 else '10.210.2.1')]
    for name,destination,gateway in routes:ip(name,family,'route','add',destination,'via',gateway)
    if ipv6:
        deadline=time.monotonic()+4
        while any('tentative' in ip(name,'-6','address','show').stdout for name in ['client','r1','r2','target']):
            assert time.monotonic()<deadline,'IPv6 link-local DAD did not finish';time.sleep(.02)
def nft(name,script):return ns(name,'nft','-f','-',input=script)
def counters(name):
    result={}
    for item in json.loads(ns(name,'nft','-j','list','ruleset').stdout)['nftables']:
        rule=item.get('rule',{})
        if 'comment' in rule:
            for expression in rule['expr']:
                if 'counter' in expression:result[rule['comment']]=expression['counter']['packets']
    return result
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,'capture exited'
        if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path
    raise AssertionError('capture did not become ready')
def captured(process,path):
    process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3);assert process.returncode==0,diagnostic
    return run('tcpdump','-nn','-vv','-r',str(path)).stdout


def server(stack):
    code=r'''
import socketserver,hashlib
class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        self.request.settimeout(3)
        header=self.rfile.readline()
        if not header:return
        length=int(header);data=self.rfile.read(length)
        if len(data)!=length:return
        self.wfile.write(hashlib.sha256(data).hexdigest().encode()+b'\n');self.wfile.flush()
class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True
service=Server(('10.210.3.2',18080),Handler);print('ready',flush=True);service.serve_forever()
'''
    process=spawn(stack,['ip','netns','exec','target','python3','-u','-c',code]);assert line(process).strip()=='ready'
def transfer(size):
    code=r'''
import socket,json,hashlib,sys,struct
size=int(sys.argv[1]);stream=socket.socket();stream.settimeout(1.5);connected=False
try:
    stream.connect(('10.210.3.2',18080));connected=True
    stream.sendall(str(size).encode()+b'\n'+b'x'*size)
    with stream.makefile('rb') as incoming:reply=incoming.readline().decode().strip()
    assert reply==hashlib.sha256(b'x'*size).hexdigest()
    print(json.dumps({'ok':True,'connected':True,'bytes':size,'sha256':reply,'snd_mss':stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_MAXSEG)}))
except OSError as error:
    stream.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack('ii',1,0))
    print(json.dumps({'ok':False,'connected':connected,'type':type(error).__name__,'errno':error.errno}))
finally:stream.close()
'''
    return json.loads(ns('client','python3','-c',code,str(size)).stdout)
def hide_fragmentation_feedback():
    nft('r1','''table ip hidden_mtu {
 chain output { type filter hook output priority 0; policy accept;
  icmp type destination-unreachable icmp code frag-needed counter drop comment "fragmentation_feedback_dropped"
 }
}
''')
results={}
for hidden in [False,True]:
    with lab() as (root,stack,setting):
        topology(setting);setting('client','net.ipv4.tcp_mtu_probing',0);server(stack)
        if hidden:hide_fragmentation_feedback()
        small=transfer(128);assert small['ok']
        before=ip('client','route','get','10.210.3.2').stdout
        tap,path=capture(stack,root,'client','c0','icmp or tcp port 18080','pmtu')
        large=transfer(65536);trace=captured(tap,path);after=ip('client','route','get','10.210.3.2').stdout
        if hidden:
            assert not large['ok'] and large['connected'] and large['type']=='TimeoutError',large
            count=counters('r1');assert count['fragmentation_feedback_dropped']>0
        else:
            assert large['ok'] and 'mtu 1280' in after,(large,after)
            assert 'unreachable - need to frag (mtu 1280)' in trace,trace;count={}
        results[str(hidden)]={'small_transfer':small,'large_transfer':large,'route_before':before,'route_after':after,
            'icmp_feedback_filtered':hidden,'counter':count,'capture':trace}
print(json.dumps(results))
"""
name='xbcatm-mtupath-'+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)

两个端点链路 MTU 都是 1500，路由器之间的链路则为 1280。这样放置瓶颈很重要：端点不会仅因自身接口较小，就在握手时直接通告较小 MSS。客户端明确关闭 TCP 的独立 MTU 探测功能，使比较聚焦经典 ICMP 发现。每次试验新建命名空间，防止正常场景学到的目的地估计泄漏到过滤场景，从而掩盖反馈缺失。

允许反馈时，128 字节传输先成功；随后新连接发送 65536 字节，并校验服务器 SHA-256 响应。捕获中出现需要分片、MTU 为 1280 的 ICMP 消息。客户端之后的路由查询显示缓存 MTU 1280，本次启用时间戳的交互中，套接字报告有效发送 MSS 1228。这把控制报文、目的地状态和应用交付连接起来，而不是仅展示配置命令。

第二个拓扑只抑制第一台路由器向外发送的需要分片消息。小传输仍成功，大传输完成 TCP 建立后超时。规则计数器记录实际被抑制的反馈，客户端路由查询则没有学到较小 MTU。这些观察把大小相关故障与刻意缺失的发现信号联系起来，避免错误归因于监听停止，也说明 connect 成功之后仍需要检查真实数据交换。

失败测试套接字会被明确中止，防止排队重传污染后续观察。应用截止时间 1.5 秒只是限制实验长度，不是操作系统完整重试生命周期。成功场景验证准确载荷，失败场景同时保留本地建立结果与超时类型。这样既能说明哪里已经成功，也不会把传输连接建立误报成应用数据已经交付，或把实验截止时间当作协议统一常数。

按正确处理位置解释抓包

本地捕获可能出现很大的卸载表示或看似不完整的校验和。这类记录不一定是最终放上物理链路的分段，因此一个很大的捕获长度，不能证明超大线路报文已经穿过瓶颈。应关联路由状态、ICMP 反馈、有效 MSS 和应用结果，而不是脱离处理上下文解读单个字段。本例确实包含这些容易误读的本地捕获表现，需要保留观察点说明。

诊断路径时使用多个有意义的载荷大小，并计算 IP、传输、选项和隧道开销。ping 载荷不等于完整 IP 包大小，应用一次写入也不等于一个 TCP 段。保留原始失败方向，因为反方向的小响应成功，可能没有触发相同限制。两个方向都可达，不代表它们的包大小条件和缓存状态必然完全一致。

恢复必要反馈或修正实际路径配置后，应从新状态重复代表性传输。针对单一传输协议的变通措施，可能缓解某个应用，却未修复其他协议的发现过程。本实验验证经典 IPv4 发现及其刻意失败，没有认证所有替代探测算法，也没有给任意互联网路径规定统一超时。修复结论应与实际测试协议、方向和负载大小保持一致。

来源