197 lines
7.2 KiB
Plaintext
197 lines
7.2 KiB
Plaintext
from stdint import *
|
||
import zlib.pyzlib as pyzlib
|
||
import zlib.zdeflate as zdeflate
|
||
import zlib.zinflate as zinflate
|
||
import zlib.zdef as zdef
|
||
import zlib.zhuff as zhuff
|
||
import string
|
||
import stdlib
|
||
import testcheck
|
||
import t, c
|
||
|
||
|
||
def fill_data(data: BYTEPTR, size: t.CSizeT, seed: t.CSizeT, repeating: t.CInt):
|
||
i: t.CSizeT
|
||
for i in range(size):
|
||
if repeating:
|
||
data[i] = BYTE(65 + (i % 3))
|
||
else:
|
||
data[i] = BYTE((i * 31 + seed * 17 + (i * i) % 251) % 256)
|
||
|
||
|
||
def huffman_probe(seed: t.CSizeT, size: t.CSizeT, repeating: t.CInt):
|
||
"""对 roundtrip 的同一份数据直接用 zhuff 复算字面量码表并打印分布。
|
||
|
||
背景：随机数据的 comp 远小于其字面量熵下限（250 字节数据熵下限约 221 字节，
|
||
实测 comp=132），说明码表/码长分配异常。本探针打印：
|
||
- active：拿到码位（bits>0）的符号数
|
||
- no_code：有频率却拿不到码位的符号数（编码期会被 encode_symbol 静默丢弃）
|
||
- sum15：Σ hist[b]*2^(15-b)，合法完整码表必须恰为 32768（=2^15）
|
||
再逐位宽打印码字数直方图。
|
||
"""
|
||
data: BYTEPTR = BYTEPTR(stdlib.malloc(size))
|
||
fill_data(data, size, seed, repeating)
|
||
freqs: t.CArray[t.CInt, 288]
|
||
string.memset(freqs, 0, freqs.__sizeof__())
|
||
i: t.CSizeT
|
||
for i in range(size):
|
||
freqs[data[i]] += 1
|
||
freqs[256] = 1
|
||
|
||
tree = zhuff.zhuff_tree()
|
||
tree.build_codes(freqs, 257, 15)
|
||
|
||
hist: t.CArray[t.CInt, 16]
|
||
string.memset(hist, 0, hist.__sizeof__())
|
||
active: t.CInt = 0
|
||
no_code: t.CInt = 0
|
||
sum15: t.CInt = 0
|
||
k: t.CInt
|
||
bits: t.CInt
|
||
for k in range(257):
|
||
bits = tree.codes[k].bits
|
||
if freqs[k] > 0 and bits == 0:
|
||
no_code += 1
|
||
if bits > 0 and bits < 16:
|
||
active += 1
|
||
hist[bits] += 1
|
||
sum15 += 1 << (15 - bits)
|
||
|
||
testcheck.info(f"[probe seed={seed} size={size} rep={repeating}] "
|
||
f"active={active} no_code={no_code} sum15={sum15} (期望 32768)")
|
||
for k in range(1, 16):
|
||
if hist[k] > 0:
|
||
testcheck.info(f" bits={k} codes={hist[k]}")
|
||
|
||
stdlib.free(data)
|
||
|
||
|
||
def count_freqs_probe(seed: t.CSizeT, size: t.CSizeT, repeating: t.CInt):
|
||
"""调用真实的 zdeflate_count_freqs，与朴素直方图对照。
|
||
|
||
上面 huffman_probe 证明码表层正确，则 comp 远小于熵下限只能来自
|
||
"频率统计与编码遍历不一致"：count 阶段漏计某些位置 ⇒ 那些字节值 freq=0 ⇒
|
||
码表里 bits=0 ⇒ encode_block_data 仍会写出这些字面量，被 encode_symbol
|
||
静默丢弃 ⇒ 码流偏短（dec < size）。本探针打印：
|
||
- sum_lit：Σlit_freqs，正常应恰为 size+1（每个字节一次 + EOB）
|
||
- active：真实统计下 freq>0 的字面量符号数
|
||
- hlit / hdist / Σdist_freqs（>0 说明 count 阶段找到了 LZ77 匹配）
|
||
"""
|
||
data: BYTEPTR = BYTEPTR(stdlib.malloc(size))
|
||
fill_data(data, size, seed, repeating)
|
||
|
||
lit: t.CArray[t.CInt, 288]
|
||
dist: t.CArray[t.CInt, 32]
|
||
string.memset(lit, 0, lit.__sizeof__())
|
||
string.memset(dist, 0, dist.__sizeof__())
|
||
|
||
s = zdeflate.zdeflate_create(None, pyzlib.Z_DEFAULT_COMPRESSION, pyzlib.MAX_WBITS, 8, 0)
|
||
if s:
|
||
zdeflate.zdeflate_count_freqs(lit, dist, s, data, size)
|
||
s.destroy()
|
||
|
||
# 元素符号性对拍：UINT8PTR 若被当作**有符号 i8**，则字节 ≥128 读成负数
|
||
# （-56 >= 128 恒假），既导致 count 阶段 lit_freqs[负数] 写到数组之前（漏计），
|
||
# 也让 encode_symbol(data[pos]) 收到负 symbol 被静默丢弃 ⇒ 码流截断。
|
||
up: UINT8PTR = UINT8PTR(data)
|
||
ge8_up: t.CInt = 0
|
||
ge8_bp: t.CInt = 0
|
||
i: t.CSizeT
|
||
for i in range(size):
|
||
if up[i] >= 128: ge8_up += 1
|
||
if data[i] >= 128: ge8_bp += 1
|
||
testcheck.info(f"[countprobe] >=128 计数: UINT8PTR={ge8_up} BYTEPTR={ge8_bp} "
|
||
f"(两者应相等且≈一半)")
|
||
|
||
sum_lit: t.CInt = 0
|
||
active: t.CInt = 0
|
||
k: t.CInt
|
||
for k in range(288):
|
||
sum_lit += lit[k]
|
||
if k < 256 and lit[k] > 0:
|
||
active += 1
|
||
sum_dist: t.CInt = 0
|
||
for k in range(32):
|
||
sum_dist += dist[k]
|
||
hlit: t.CInt = 286
|
||
while hlit > 257 and lit[hlit - 1] == 0:
|
||
hlit -= 1
|
||
hdist: t.CInt = 30
|
||
while hdist > 1 and dist[hdist - 1] == 0:
|
||
hdist -= 1
|
||
|
||
testcheck.info(f"[countprobe seed={seed} size={size} rep={repeating}] "
|
||
f"sum_lit={sum_lit} (期望 {size + 1}) active={active} "
|
||
f"sum_dist={sum_dist} hlit={hlit} hdist={hdist}")
|
||
|
||
stdlib.free(data)
|
||
|
||
|
||
def roundtrip(seed: t.CSizeT, size: t.CSizeT, repeating: t.CInt, wbits: t.CInt) -> t.CInt:
|
||
data: BYTEPTR = BYTEPTR(stdlib.malloc(size))
|
||
fill_data(data, size, seed, repeating)
|
||
|
||
out_len: t.CSizeT = 0
|
||
comp: BYTEPTR = pyzlib.compress(None, data, size,
|
||
pyzlib.Z_DEFAULT_COMPRESSION, wbits, c.Addr(out_len))
|
||
if not comp:
|
||
testcheck.fail(f"seed={seed} size={size} rep={repeating} wbits={wbits}: "
|
||
f"compress 失败 err={pyzlib.get_error_code()} {pyzlib.get_error()}")
|
||
stdlib.free(data)
|
||
return -1
|
||
|
||
dec_len: t.CSizeT = 0
|
||
dec: BYTEPTR = pyzlib.decompress(None, comp, out_len, wbits,
|
||
pyzlib.DEF_BUF_SIZE, c.Addr(dec_len))
|
||
ok: t.CInt = 1
|
||
if not dec:
|
||
ok = 0
|
||
testcheck.fail(f"seed={seed} size={size} rep={repeating} wbits={wbits}: "
|
||
f"decompress 失败 comp={out_len} err={pyzlib.get_error_code()} {pyzlib.get_error()}")
|
||
elif dec_len != size or string.memcmp(dec, data, size) != 0:
|
||
ok = 0
|
||
testcheck.fail(f"seed={seed} size={size} rep={repeating} wbits={wbits}: "
|
||
f"内容不匹配 comp={out_len} dec={dec_len}")
|
||
else:
|
||
testcheck.ok(f"seed={seed} size={size} rep={repeating} wbits={wbits}: "
|
||
f"comp={out_len} dec={dec_len}")
|
||
|
||
if dec: stdlib.free(dec)
|
||
stdlib.free(comp)
|
||
stdlib.free(data)
|
||
return 0 if ok else -1
|
||
|
||
|
||
def main() -> t.CInt:
|
||
testcheck.begin("ZlibDbg: zlib 多 seed/size/wbits 往返")
|
||
|
||
# 先跑码表探针：定位 comp 远小于熵下限的根因（码表异常 / 还是别处）
|
||
testcheck.section("字面量码表探针")
|
||
huffman_probe(0, 250, 0)
|
||
huffman_probe(9, 20000, 0)
|
||
huffman_probe(0, 256, 1)
|
||
|
||
# 真实频率统计 vs 朴素直方图对照
|
||
testcheck.section("频率统计探针 (zdeflate_count_freqs)")
|
||
count_freqs_probe(0, 250, 0)
|
||
count_freqs_probe(9, 20000, 0)
|
||
count_freqs_probe(0, 256, 1)
|
||
|
||
testcheck.section("随机数据 (wbits=15)")
|
||
sizes: t.CArray[t.CSizeT, None] = [250, 255, 256, 257, 260, 300, 500, 1000, 4096, 20000]
|
||
n: t.CInt
|
||
for n in range(10):
|
||
roundtrip(t.CSizeT(n), sizes[n], 0, pyzlib.MAX_WBITS)
|
||
|
||
testcheck.section("重复数据 (wbits=15)")
|
||
sizes2: t.CArray[t.CSizeT, None] = [256, 257, 300, 1000, 20000]
|
||
for n in range(5):
|
||
roundtrip(t.CSizeT(n), sizes2[n], 1, pyzlib.MAX_WBITS)
|
||
|
||
testcheck.section("gzip 容器 (wbits=31)")
|
||
sizes3: t.CArray[t.CSizeT, None] = [300, 1000]
|
||
for n in range(2):
|
||
roundtrip(t.CSizeT(n), sizes3[n], 0, pyzlib.MAX_WBITS + 16)
|
||
|
||
return testcheck.end()
|