feat(plugins): 分片调用 —— 让阻塞型导出函数真正并行

## 问题

quota 页面 20 个源,刷新要 29 秒。页面用 `Promise.all` 并发发 20 个请求,
**看起来是并发的,实际不是**。

`Plugins.Call` 持 `p.mu.Lock()`,锁一直持有到 `L.Call` 返回 —— **包含 Lua 函数
内部的全部网络等待**。20 个请求到了网关后排队,串行累加。

生产实测:

    单源探测(无竞争)      1.4 – 3.3 s
    20 源 Promise.all       29 s 墙钟
    串行理论值              ≈ 29 s   ← 与实测吻合

代码里其实已经知道这个问题:quota.lua 第 104 行写着「the whole page shares ONE
Lua state under a write lock」,并为此把超时从 10s 压到 4s。但那只把 29s 变成
约 12s,没有解决「刷新不可用」。

而当初改成并发的理由(页面注释)是错的 —— 它假设并发能绕开串行。锁让它退化
回串行了。

## 不能去掉锁

一个 Lua VM 只有一个栈。两个并发调用会破坏栈。锁是必要的,答案只能是**让每个
并发调用有自己的 VM,跑同一份插件源码**。

先量,再写实现(8 分片 × 300ms 本地 HTTP 端点):

    共享 VM,8 次串行      2.407 s
    8 个独立 VM 并行       0.304 s    → 7.9x

顺带一个测量教训:这个 spike 第一版用**忙等**而不是网络 I/O,报出 2.3x。
那是量到了 12 核机器上的 CPU 争抢(负载 8.95),和锁无关。插件做的是网络等待,
spike 就必须做网络等待 —— 和上一条提交里 os.clock 那个错误是同一类。

## 实现

- `callOn(w, fn, args)`:从 `Call` 里抽出的「在某 VM 上调一个导出函数」。`Call`
  从 66 行缩到 18 行,仍负责查找、export 白名单和共享 VM 的锁。
- `shardPool`:私有 VM 池,上限 32。开机约 0.1ms,池化省不了什么但省两次刷新
  的重复开销。
- `ShardCall(name, fn, argsList)`:结果与输入**按位置对齐**。
- `POST /api/plugins/{name}/fanout/{fn}`:`{"args": [...]}`,上限 64 条。
- `pluginAPI.fanout()`:页面一次请求。
- quota.lua 改用 fanout。

## 分片的硬约束(写进注释,不是靠约定)

分片 VM **不能碰 PLUGIN.state** —— state 在 VM 里,每个分片一份副本,billing
的累加会各分叉,saver 写出去的是随机某一份。分片只能用于只读、幂等的工作。
所以 `callOn` 既不查 export 白名单(`Call` 那层才是安全边界)也不落盘。

## 一个分片失败不拖垮整批

一个上游连不上,不该让 20 张卡片变成一个错误。失败的分片在自己的位置留
null,其余照常返回。**但全部分片失败时报错** —— 那不是「某个上游挂了」,
是结构性问题,报成功会把它藏在满屏红卡片后面。

## 顺手修掉一个既有缺陷:热重载是坏的

写分片源码来源时,判据直接把它顶了出来:`LoadSource` 无条件 append,而
`findLocked` 返回**第一个**匹配。实测:

    LoadDir 后 1 条 → LoadSource 后 2 条
    [0] version=1   [1] version=2
    find 返回 version=1

也就是说,**编辑插件并保存看起来什么都没发生** —— 旧版本一直被返回,新版本
躺在旁边不生效。修法是先 Unload 再加载。这不是为分片做的,是分片正好照到了
这条路径,已单独补判据 `TestReloadReplacesRatherThanAppends`。

## 判据

七条:并行性(有界而非计时断言,8×200ms 串行是 1.6s,超过 1.0s 即红)、
位置对齐、未导出与未知插件被拒、单个失败不拖批、全失败报错、用已安装源码而
非磁盘、以及热重载替换而非追加。

全量 9 包通过。

Co-Authored-By: ModelRouter <noreply@modelrouter.dev>
This commit is contained in:
JianFeeeee
2026-10-04 21:56:45 +08:00
parent 345cbb30d6
commit 33b778faee
5 changed files with 741 additions and 8 deletions

View File

@ -32,9 +32,9 @@ package lua
// less featureful, so the split is explicit and documented.
import (
"log"
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"sort"
@ -265,6 +265,10 @@ type Plugins struct {
// visibility holds the operator's per-plugin audience overrides, keyed by
// plugin name. Consulted by UIForRole and the call endpoint's authorisation.
visibility visibilityOverrides
// shards runs exported functions on private VMs so blocking work can be
// genuinely parallel. See plugin_shard.go for why the shared VM's lock makes
// concurrency impossible and what a shard must therefore never touch.
shards *shardPool
// stateDirDisabled turns persistence off. Used by tests that assert state
// starts empty, and by an embedder that has no writable plugin dir.
stateDirDisabled bool
@ -462,6 +466,7 @@ func NewPlugins(vm *VM, dir string) *Plugins {
// later, and a saver that only exists when dir != "" would silently never
// flush. markDirtyLocked and flush both no-op when the dir is empty.
ps.save = newStateSaver(ps)
ps.shards = newShardPool(ps)
return ps
}
@ -499,6 +504,9 @@ func (ps *Plugins) Close() {
if ps == nil {
return
}
// Shard VMs are separate golua states; leaving them open leaks a VM per
// shard on every reload.
ps.shards.Shutdown()
ps.save.Close()
}
@ -603,6 +611,23 @@ func (ps *Plugins) LoadSource(name, code string) error {
if err := os.MkdirAll(ps.dir, 0755); err != nil && ps.dir != "" {
return err
}
// Reloading must REPLACE, not append.
//
// LoadSource was reachable both from LoadDir and from the install/reload API,
// and it appended unconditionally. Loading a plugin that was already loaded
// therefore left two records with the same name, and findLocked — which
// returns the FIRST match — kept handing out the OLD one. Measured: after
// LoadDir + LoadSource of an edited plugin, /api/plugins listed both
// versions and Call answered from version 1 while version 2 sat inert
// alongside it. Editing a plugin and saving appeared to do nothing.
//
// Unload is a no-op when nothing is loaded under that name, so this is safe
// on the first load.
if ps.find(name) != nil {
if err := ps.Unload(name); err != nil {
return fmt.Errorf("replace %s: %w", name, err)
}
}
p := &Plugin{
Info: PluginInfo{Name: name},
Hooks: map[Stage]string{},
@ -1935,6 +1960,17 @@ func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{})
// quota report is a nested structure with numbers and booleans that must not be
// flattened. Hook results still go through invoke, whose map-typed contract the
// state merge depends on.
// Call invokes an exported function on a plugin's SHARED VM, serialised by the
// plugin's mutex.
//
// The lock is what makes one Lua VM safe to share, and it is held for the whole
// call — including any network wait inside it. That is correct for a hook and
// wrong for a function that blocks: twenty concurrent quota probes serialise
// into the sum of their timeouts (measured: 29s on a 20-source gateway, against
// 1.4–3.3s for a single probe with no contention).
//
// callOn below is the same work without the shared-VM assumption, so a caller
// that owns a private VM can use it directly.
func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface{}, error) {
ps.mu.RLock()
var p *Plugin
@ -1952,12 +1988,27 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
}
p.mu.Lock()
defer p.mu.Unlock()
w := p.state
if w == nil {
p.mu.Unlock()
return nil, fmt.Errorf("no state for plugin %s", name)
}
defer p.mu.Unlock()
return callOn(w, fn, args)
}
// callOn runs one exported function on a specific VM.
//
// It does NOT check the export allowlist and does NOT persist state: both belong
// to the shared-VM path in Call. A shard VM has no plugin identity to check
// against and no state file of its own — the export allowlist in Call is the
// security boundary, and calling this with a function the plugin does not export
// is a programming error, not an attack.
//
// The caller must own w exclusively. No lock is taken here.
func callOn(w *worker, fn string, args map[string]interface{}) (interface{}, error) {
if w == nil || w.L == nil {
return nil, fmt.Errorf("no lua state")
}
L := w.L
L.SetTop(0)
defer L.SetTop(0)
@ -1990,14 +2041,14 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
L.Pop(1)
}
if errMsg != "" {
return nil, fmt.Errorf("plugin %s: %s", name, errMsg)
return nil, fmt.Errorf("plugin: %s", errMsg)
}
if L.GetTop() < 1 || L.IsNil(-1) {
return nil, nil
}
var out interface{}
if err := luaToJSON(L, -1, &out); err != nil {
return nil, nil
return nil, err
}
return out, nil
}