perf(api): SSE 导航层返工 —— 5+ 次边界压成 1 次(三项赢,仍默认关闭)

按 sse-codec-c.md §6.4 的架构改造方向返工。**部分成功**:从「五项全输」
变成「三项赢 / 一项持平 / 一项输」,且所有场景分配数都下降。

## 改造内容
| 项 | 前 | 后 |
|---|---|---|
| cgo 边界次数 | 5+(每字段一次 findKey) | 1(ha_sse_chunk_locate) |
| 键查找 | 每键各扫一遍对象(6 趟) | 单趟分派(遍历成员表一次即分发) |
| 解码 | 每字段一次往返 + 各自 decBuf | 同一趟内写进一块 sbuf(1 次分配) |
| 成员表遍历 | 6 趟 | 2 趟(顶层 + delta) |

顺带修掉两处自造的浪费(都是「先扫一遍拿个数、再扫第二遍拿首元素」):
choices 数组的「数个数 + 取首元素」合一趟;choice0 内的 delta/finish_reason
合一趟。成员遍历实测 107ns/趟,省一趟就是省 107ns。

新增 ha_sse_chunk_locate:一次调用完成根校验 + 顶层分派 + choices[0] +
delta 分派 + content/reasoning/finish 解码,输出写调用方持有的 C 结构体
(C 结构体无 Go 指针 ⇒ 可安全传指针,消除 out-param 逃逸)。
choices_count>1 时直接回退(Go 侧 Unmarshal 会解析全部元素,本层只认 [0],
其余元素可能类型不符而让 Go 整块作废 ⇒ 无法保证等价)。

## 实测(50000 次 × 3 轮取中位)
| 场景 | Entry | GoOnly | 判定 |
|---|---|---|---|
| content_zh | 1540ns / 5allocs | 1871ns / 13allocs | 快 18%,分配 -62% |
| content_ascii | 1250ns / 5allocs | 1304ns / 13allocs | 持平,分配 -62% |
| finish | 820ns / 6allocs | 921ns / 12allocs | 快 11% |
| usage | 2530ns / 9allocs | 2591ns / 12allocs | 持平偏快 |
| toolcall | 3450ns / 20allocs | 3000ns / 21allocs | 慢 15% |

## toolcall 仍输的根因(已定位,非猜测)
分解测量:C 侧纯 C 零边界 = 766ns;Go 侧 []openAIToolCall unmarshal =
1305ns/15allocs;对照 Go 整块 unmarshal ≈ 2980ns。
问题在第二行:tool_calls 元素是对象,Arguments interface{} 需要真实的
map[string]interface{},必须走 encoding/json 的反射建树。
而为了定位已先做了一遍 C 扫描 ⇒ 同一份数据被解析了两次。
⇒ 不是 C 慢,是「扫两遍 vs 扫一遍」。
标量字段(content/reasoning/finish)C 能一次到位 ⇒ 那些场景赢;
需要建树的字段(tool_calls/usage)C 的定位是纯开销。

## 为什么仍默认关闭(理由充分,不是保守)
1. toolcall 是真实负载最常见的一类块(任何一次工具调用流),仍慢 15%
2. 18% 收益不足以抵消「与 encoding/json 语义并存的第二实现」的风险
3. 本刀原始动机在 toolcall 场景没有兑现:分配数 20 vs 21 几乎没降
⇒ 前提是先做「按字段类型决定是否 C 化」,让 toolcall 也不输,再重测。

## 正确性
6 万+ 差分用例(协议形态/真实负载/随机 JSON 3 万/随机字节 3 万)全过。
基准测量也修了:先前 C 基准脚本用 CLOCK_MONOTONIC 却只取 tv_nsec,
算出 -4201ns 的负值 —— 测量工具本身出错会直接毁掉结论。

## 验证
ASan+UBSan PASS;gcc+clang 零告警;arm64 交叉 0 告警;
libFuzzer 66 万次零崩溃;全量 go test 38 包 ok / 0 FAIL
This commit is contained in:
JianFeeeee
2026-09-26 10:41:52 +08:00
parent 7748ec450e
commit 7a1322c97f
5 changed files with 742 additions and 190 deletions

View File

@ -155,199 +155,75 @@ func chunkAssemble(choices []chunkChoice, usage chunkUsage) (StreamChunk, bool)
// ---------------------------------------------------------------------
// chunkParseFast 尝试 C 快速路径。
// 返回 (chunk, handled, decided):
// handled=false ⇒ 调用方必须用 chunkParseGo
// handled=true,decided=true ⇒ 结果是最终答案
//
// ============================ 第三刀的重做:一次 cgo 调用 ============================
// 上一版逐字段往返(5+ 次 findKey,每次 ~168ns 边界 + 2 allocs)造成固定成本
// 约 1µs,比原实现更慢。本版把全部定位压进**一次** C 调用
// (ha_sse_chunk_locate),并在同一趟里完成键分派与字符串解码。
//
// 返回 (chunk, handled, decided)。handled=false ⇒ 调用方用 chunkParseGo。
func chunkParseFast(data string) (StreamChunk, bool, bool) {
root := rootSpan(data)
if !sseRootObject(root) {
return StreamChunk{}, false, false
}
// ---- usage:整棵子树交给 encoding/json ----
// ★ 为什么逐个整数取是错的:Go 侧 usage 有 9 个字段,且**任一类型不符
// 就让整块作废**(实测 {"prompt_cache_hit_tokens":"x","prompt_tokens":1}
// → 整块 false)。整棵 unmarshal 到**同一个 Go 类型** ⇒ 语义自动一致。
var usage chunkUsage
us, ufound, udup, ubad := findKeyCI(root, "usage")
if ubad || udup {
return StreamChunk{}, false, false
}
if ufound {
switch us.firstByte() {
case 'n':
// null ⇒ 零值 struct(不产出 usage)
case '{':
if err := json.Unmarshal(us.bytes(), &usage); err != nil {
// ★ 类型不符 ⇒ 与 Go 一样「整块作废」,**不需要回退**
return StreamChunk{}, false, true
}
default:
return StreamChunk{}, false, false // 交回 Go 决定
}
}
// ---- choices:只取 [0],但要先判整切片的长度语义 ----
cs, cfound, cdup, cbad := findKeyCI(root, "choices")
if cbad || cdup {
return StreamChunk{}, false, false
}
if !cfound {
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
}
switch cs.firstByte() {
case 'n':
// null ⇒ 零值切片(长度 0)⇒ 走「无 choices」分支
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
case '[':
loc := locateChunkBatch(data)
switch loc.status {
case chunkTypeFail:
// C 已判定「与 Go 一致的整块作废」⇒ 直接给答案,无需回退
return StreamChunk{}, false, true
case chunkOK:
// 继续
default:
return StreamChunk{}, false, false
}
el, has := firstElem(cs)
if !has {
// 空数组:len(choices)==0 ⇒ 与 Go 相同
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
// ---- usage:整棵子树交给 encoding/json(9 个字段 + 类型规则)----
var usage chunkUsage
switch loc.usageKind {
case kindAbsent, kindNull:
// 零值
case kindObject:
if err := json.Unmarshal(loc.usageSpan.bytes(), &usage); err != nil {
return StreamChunk{}, false, true // 类型不符 ⇒ 整块作废
}
default:
return StreamChunk{}, false, false
}
ch, ok := fastChoice(el)
if !ok {
return StreamChunk{}, false, false // 任何不确定 ⇒ 整体回退
// ---- delta 非对象 ⇒ 与 Go 的 Unmarshal 失败一致 ----
if loc.deltaKind == kindOther {
return StreamChunk{}, false, true
}
ck, ok2 := chunkAssemble([]chunkChoice{ch}, usage)
return ck, true, ok2
}
// fastChoice 解析 choices[0]。ok=false ⇒ 必须回退 Go。
//
// ★ 键匹配方式按 Go 那一跳的实际类型选择:
// - choices / delta / finish_reason / tool_calls 是 **struct 字段** ⇒ 大小写不敏感
// - content / reasoning_content / "text" 是 **interface{} → map key** ⇒ 大小写敏感
// (实测:{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]} 有效;
// {"content":[{"TEXT":"up"}]} 取不到 text)
func fastChoice(el strSpan) (chunkChoice, bool) {
var ch chunkChoice
if el.firstByte() != '{' {
return ch, false
}
ds, dfound, ddup, dbad := findKeyCI(el, "delta")
if dbad || ddup {
return ch, false
}
if dfound {
switch ds.firstByte() {
case 'n':
// delta:null ⇒ 零值 struct
case '{':
// ★ delta 是 **struct**(不是 map!)——
// 原实现:Delta struct { Content interface{}; ... } `json:"delta"`
// 故它的字段名匹配是**大小写不敏感**。
// 实测 `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}` → content="ci"。
// 只有 content 的**值**(若为对象/数组)才成为 map/[]interface{},
// 那时里面的键(如 "text")才是大小写敏感。
//
// 我一度把这里改成 CS 并认为「差分测试会通过」——那是错的推理:
// Go 侧给的是 "ci"(CI 匹配成功),改成 CS 反而把快速路径弄丢。
// 教训:**「哪一层是 struct、哪一层是 map」要回原实现读类型,
// 不能凭字段名像 map 就推断它是 map。**
// reasoning_content:Go 侧是 **string**(强类型)。
// 用同样的 Go 类型 unmarshal ⇒ 123 会报错,与原实现一致。
rs, rfound, rdup, rbad := findKeyCI(ds, "reasoning_content")
if rbad || rdup {
return ch, false
}
if rfound && rs.firstByte() != 'n' {
var s string
if err := json.Unmarshal(rs.bytes(), &s); err != nil {
return ch, false // 类型不符 ⇒ 回退(Go 会整块作废)
}
ch.reasoning = s
}
// content 字段名:CI(struct 字段)。
// 其**值**若是数组/对象,内部键由 ha_sse_stringify 按 CS 处理。
cs, cfound, cdup, cbad := findKeyCI(ds, "content")
if cbad || cdup {
return ch, false
}
if cfound {
if s, handled := stringifyC(cs); handled {
ch.content = s
} else {
return ch, false // 需 json.Marshal 重新编码(§5.2)
}
}
// tool_calls:逐个元素整体 unmarshal 成 openAIToolCall,
// 使 arguments 的 interface{} 形态 / 类型检查全由 encoding/json 负责。
tcs, tfound, tdup, tbad := findKeyCI(ds, "tool_calls")
if tbad || tdup {
return ch, false
}
if tfound {
switch tcs.firstByte() {
case 'n':
// null ⇒ 零值切片
case '[':
t, ok := fastToolCalls(tcs)
if !ok {
return ch, false
}
ch.toolCalls = t
default:
return ch, false
}
}
default:
return ch, false
}
}
// finish_reason:struct 字段 ⇒ 大小写不敏感;Go 侧是 *string
fs, ffound, fdup, fbad := findKeyCI(el, "finish_reason")
if fbad || fdup {
return ch, false
}
if ffound && fs.firstByte() != 'n' {
if fs.firstByte() != '"' {
return ch, false
}
var s string
if err := json.Unmarshal(fs.bytes(), &s); err != nil {
return ch, false
}
ch.finishPtr = &s
}
return ch, true
}
// fastToolCalls 解析 tool_calls 数组。
//
// ★ 逐元素整体 unmarshal 成 openAIToolCall 是刻意的:这样 arguments 的
// interface{} 形态、字符串/对象/数组/数字各分支、重复键,全部由
// encoding/json 处理(§5.2 的重新编码语义不必在 C 复刻)。
// 归一化也走**同一个** normalizeStreamToolCall ⇒ 与 Go 路径不分叉。
func fastToolCalls(arr strSpan) ([]ToolCall, bool) {
elems, ok := scanArray(arr)
if !ok {
return nil, false
}
if len(elems) == 0 {
return nil, true // 空数组 ⇒ nil(与 Go 的 normalizeStreamToolCalls 一致)
}
out := make([]ToolCall, 0, len(elems))
for _, elem := range elems {
if elem.firstByte() != '{' {
return nil, false
}
var raw openAIToolCall
if err := json.Unmarshal(elem.bytes(), &raw); err != nil {
return nil, false
}
out = append(out, normalizeStreamToolCall(raw))
}
return out, true
// ---- tool_calls:整段 unmarshal 成 []openAIToolCall ----
// ★ 用与 Go 完全相同的类型 ⇒ arguments 的 interface{} 形态与类型检查
// 全部由 encoding/json 负责;归一化共用 normalizeStreamToolCall。
var toolCalls []ToolCall
switch loc.toolCallsKind {
case kindAbsent, kindNull:
// nil
case kindArray:
var raw []openAIToolCall
if err := json.Unmarshal(loc.toolCallsSpan.bytes(), &raw); err != nil {
return StreamChunk{}, false, true // 元素类型不符 ⇒ 整块作废
}
toolCalls = normalizeStreamToolCalls(raw)
default:
return StreamChunk{}, false, true
}
// ---- 拼装(与 Go 路径共用 chunkAssemble)----
var choices []chunkChoice
if loc.choicesPresent && loc.choicesCount > 0 {
ch := chunkChoice{
content: loc.content,
reasoning: loc.reasoning,
toolCalls: toolCalls,
}
if loc.finishKind == kindString {
// 保留三态:缺失/null ⇒ nil;"" ⇒ 非 nil 空串(不算终止信号)
f := loc.finish
ch.finishPtr = &f
}
choices = []chunkChoice{ch}
}
ck, ok := chunkAssemble(choices, usage)
return ck, true, ok
}

View File

@ -118,6 +118,11 @@ static int go_arr_all(const char *p, size_t n, ha_span *out, int cap) {
return count;
}
static int go_chunk_locate(const char *p, size_t n, ha_chunk_out *out,
char *sbuf, size_t scap, size_t *sused) {
return ha_sse_chunk_locate(p, n, out, sbuf, scap, sused);
}
static int go_sse_abi(void) { return ha_sse_abi_version(); }
*/
import "C"
@ -326,3 +331,134 @@ func scanArray(arr strSpan) ([]strSpan, bool) {
}
return out, true
}
// ---------------------------------------------------------------------
// 批量定位(第三刀的重做:一次 cgo 调用代替 5+ 次)
// ---------------------------------------------------------------------
// chunkLocateResult 是 C 侧 ha_chunk_out 的 Go 视图。
type chunkLocateResult struct {
status int
// 原始 span(用于交回 encoding/json 的那些字段)
usageSpan strSpan
usageKind int
toolCallsSpan strSpan
toolCallsKind int
// C 已解码的字符串(sbuf 的副本)
content string
reasoning string
finish string
// 标志
choicesPresent bool
choicesKind int
choicesCount int
choice0Span strSpan
hasDelta bool
deltaKind int
contentKind int
reasoningKind int
finishKind int
}
// 槽位/类型常量(与 ha_sse.h 保持一致;改动必须同步 ABI 版本)
const (
slotDelta = 0
slotContent = 1
slotReasoning = 2
slotToolCalls = 3
slotFinishReason = 4
slotUsage = 5
slotCount = 6
kindAbsent = 0
kindNull = 1
kindString = 2
kindObject = 3
kindArray = 4
kindOther = 5
chunkOK = 0
chunkFallback = -1
chunkTypeFail = -2
)
// locateChunkBatch 一次调用完成整块定位。
func locateChunkBatch(data string) chunkLocateResult {
var out chunkLocateResult
if len(data) == 0 {
out.status = chunkFallback
return out
}
p, n := cstr(data)
var co C.ha_chunk_out
// ★ 单块缓冲:整块解码输出(content+reasoning+finish)都写这一块。
// 尺寸按输入上界(每字节最坏 3 字节 U+FFFD)——1 次分配,
// 替代原来「每个字段一次 decBuf」的多次分配。
sbuf := make([]byte, len(data)*3+16)
var used C.size_t
st := C.go_chunk_locate(p, n, &co, cstrb(sbuf), C.size_t(len(sbuf)), &used)
out.status = int(st)
if st != C.int(chunkOK) {
return out
}
out.usageKind = int(co.slot[slotUsage].kind)
out.usageSpan = strSpan{co.slot[slotUsage].span.p, co.slot[slotUsage].span.len}
out.toolCallsKind = int(co.slot[slotToolCalls].kind)
out.toolCallsSpan = strSpan{co.slot[slotToolCalls].span.p, co.slot[slotToolCalls].span.len}
out.contentKind = int(co.slot[slotContent].kind)
out.reasoningKind = int(co.slot[slotReasoning].kind)
out.finishKind = int(co.slot[slotFinishReason].kind)
out.hasDelta = int(co.slot[slotDelta].kind) == kindObject
out.deltaKind = int(co.slot[slotDelta].kind)
out.choicesPresent = co.has_choices == 1
out.choicesKind = int(co.choices_kind)
out.choicesCount = int(co.choices_count)
s := sbuf[:int(used)]
out.content = string(s[co.content_off : co.content_off+co.content_len])
out.reasoning = string(s[co.reasoning_off : co.reasoning_off+co.reasoning_len])
out.finish = string(s[co.finish_off : co.finish_off+co.finish_len])
return out
}
// locateChunkBatchInto 是 locateChunkBatch 的零分配内核(基准用):
// 复用调用方提供的 sbuf,不自己 make。
func locateChunkBatchInto(data string, sbuf []byte) chunkLocateResult {
var out chunkLocateResult
if len(data) == 0 {
out.status = chunkFallback
return out
}
p, n := cstr(data)
var co C.ha_chunk_out
var used C.size_t
st := C.go_chunk_locate(p, n, &co, cstrb(sbuf), C.size_t(len(sbuf)), &used)
out.status = int(st)
if st != C.int(chunkOK) {
return out
}
out.usageKind = int(co.slot[slotUsage].kind)
out.usageSpan = strSpan{co.slot[slotUsage].span.p, co.slot[slotUsage].span.len}
out.toolCallsKind = int(co.slot[slotToolCalls].kind)
out.toolCallsSpan = strSpan{co.slot[slotToolCalls].span.p, co.slot[slotToolCalls].span.len}
out.contentKind = int(co.slot[slotContent].kind)
out.reasoningKind = int(co.slot[slotReasoning].kind)
out.finishKind = int(co.slot[slotFinishReason].kind)
out.hasDelta = int(co.slot[slotDelta].kind) == kindObject
out.deltaKind = int(co.slot[slotDelta].kind)
out.choicesPresent = co.has_choices == 1
out.choicesKind = int(co.choices_kind)
out.choicesCount = int(co.choices_count)
out.choice0Span = strSpan{co.choice0_span.p, co.choice0_span.len}
s := sbuf[:int(used)]
out.content = string(s[co.content_off : co.content_off+co.content_len])
out.reasoning = string(s[co.reasoning_off : co.reasoning_off+co.reasoning_len])
out.finish = string(s[co.finish_off : co.finish_off+co.finish_len])
return out
}