保留每行身份,并分别附加唯一位置和允许并列的名次。

窗口函数在不合并组的前提下排名

排名函数为分区内的每一条输入行附加位置,不会把整个团队、部门或客户的数据压缩成一行聚合结果。真正需要选择的是并列的含义:每行占一个独立位置、允许并列且保留空位的竞赛名次,还是按不同分数组连续计数的名次。三种规则解决的问题不同,应先确定业务规则再写窗口。 三个问题对应三种排名 row_number() 从一开始给每行分配连续整数。如果两行的排序值完全相同,而窗口没有其他区分条件,它们的相对序号没有保证。需要恰好选出两位获胜者时,可以按分数降序,再按稳定且唯一的标识排序。这意味着同分时由标识决定谁占据剩余名额,不能把这个结果当成不影响业务的技术细节。 rank() 给同组并列行分配该组第一行的位置。分数为一百、九十、九十、八十时,名次依次为一、二、二、四。两位第二名占用了两个位置,因此下一档分数前面会留下空位。dense_rank() 则计算同分组的数量,对同一批数据给出一、二、二、三,不会因并列人数增加而跳号。 并列组由窗口 ORDER BY 中的全部表达式决定。如果把唯一标识也加入 rank 的窗口,即使本意只是稳定展示顺序,也会让原来的同分行不再并列。row_number 需要标识裁决、rank 需要保留同分时,应使用不同窗口定义。最终结果的展示顺序可以由外层 ORDER BY 控制,不必因此改掉排名所依据的并列关系。 明确分区与过滤时机 PARTITION BY 让每个组从头编号。不写分区条件时,窗口看到的所有输入行属于同一个分区。这和窗口帧是不同概念:排名依据分区顺序和并列组,不是对某段移动聚合帧计数。添加 ROWS 边界,不会把 rank 变成“这个边界内有多少行”的函数。 窗口看到的是 FROM、WHERE、分组和 HAVING 处理后剩余的行。因此,在同层查询中过滤,可能先删掉部分竞争者,再计算名次。如果需要先给所有人排名,再挑前几个位置,应在子查询或 CTE 中生成窗口值,然后到外层过滤。PostgreSQL 不允许直接把窗口函数放进 WHERE,这也提醒读者区分排名输入与排名输出。 “前两名”必须有明确含义。位置不大于二,每个分区最多返回两行;竞赛名次不大于二,会保留全部并列第二名,结果可能超过两行;密集名次不大于二,则选择前两个不同分数组。当第一档分数本身就有多人并列时,密集排名与竞赛排名的筛选结果也可能不同。这些是不同的结果集合,不是可以随意互换的性能写法。 坏掉的样子 row_number 缺少确定的并列裁决条件时,同分选手可能在重复查询中交换位置。反过来,在所有窗口都添加标识,又可能悄悄消除 rank 本来要保留的并列。测试时必须放入恰好卡在截止名次上的同分数据,不能只测每个分数都不同的情况。如果分数允许 NULL,还应明确空值顺序;PostgreSQL 降序默认把空值放前面,需要时应显式指定 NULLS LAST。 ...

二月 8, 2028 · 2 分钟 · XB Field Notes
每端通告自己的缩放因子,SYN 报文中的窗口字段仍不缩放。

TCP 窗口缩放分别扩展两个方向的接收窗口编码

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。二者均不能直接等同于线上通告窗口。 ...

十二月 2, 2027 · 3 分钟 · XB Field Notes