feat(napi): VM pool + per-request isolation for serial/parallel handling
- fnode capi shim: single VM+mutex -> VM pool (FIVENODE_VM_POOL, default 4; 1=serial, N=parallel). Each request checks out its own VM so PRG runs concurrently across libuv worker threads. - per-request data keyed by VM (FN_NAPI_REQ via ctx.T.VM()) -- no shared capiReq race. - napibridge: per-VM handle tracking; ReleaseAll(vm) auto-ends only that request's npm handles (parallel-safe auto-__end__). FN_AWAIT replaces the reserved Five AWAIT keyword (Clipper-compat, no gengo codegen -> NIL). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
CharlesKWON
12
capi/napi_delay.prg
Normal file
12
capi/napi_delay.prg
Normal file
@@ -0,0 +1,12 @@
|
||||
FUNCTION FN_HANDLE()
|
||||
LOCAL oD := FN_REQUIRE( "/Users/charleskwon/fivenode/fivenode/fivenode/napi/test/delaymod" )
|
||||
LOCAL cR
|
||||
IF ! HB_ISOBJECT( oD )
|
||||
RETURN hb_jsonEncode( { "err" => FN_LASTERROR() } )
|
||||
ENDIF
|
||||
cR := FN_AWAIT( oD:wait( 200 ) )
|
||||
RETURN hb_jsonEncode( { ;
|
||||
"status" => 200, ;
|
||||
"headers" => { "Content-Type" => "application/json" }, ;
|
||||
"body" => hb_CStr( cR ) ;
|
||||
} )
|
||||
@@ -1,20 +1,13 @@
|
||||
FUNCTION FN_HANDLE()
|
||||
LOCAL oQR := FN_REQUIRE( "qr-image" )
|
||||
LOCAL xRet, cSvg
|
||||
LOCAL oQR := FN_REQUIRE( "qrcode" )
|
||||
LOCAL cSvg
|
||||
IF ! HB_ISOBJECT( oQR )
|
||||
RETURN hb_jsonEncode( { "status" => 500, "headers" => { => }, "body" => "require qr-image: " + FN_LASTERROR() } )
|
||||
RETURN hb_jsonEncode( { "step" => "require", "err" => FN_LASTERROR() } )
|
||||
ENDIF
|
||||
// qr-image.imageSync(text, {type:'svg'}) → Buffer → toString
|
||||
xRet := oQR:imageSync( "https://solmade.kr", { "type" => "svg" } )
|
||||
IF HB_ISOBJECT( xRet ) // Buffer 핸들로 옴 → toString
|
||||
cSvg := xRet:toString( "utf8" )
|
||||
xRet:__end__()
|
||||
ELSE
|
||||
cSvg := hb_CStr( xRet )
|
||||
ENDIF
|
||||
oQR:__end__()
|
||||
cSvg := FN_AWAIT( oQR:toString( "https://solmade.kr", { "type" => "svg", "margin" => 2 } ) )
|
||||
// __end__ 수동 호출 없음 — 요청 종료 시 ReleaseAll 이 자동 정리(P3)
|
||||
RETURN hb_jsonEncode( { ;
|
||||
"status" => 200, ;
|
||||
"headers" => { "Content-Type" => "application/json" }, ;
|
||||
"body" => hb_jsonEncode( { "msg" => "PRG -> npm qr-image (real Node) via N-API", "svg_len" => Len( cSvg ), "svg_head" => Left( cSvg, 30 ) } ) ;
|
||||
"headers" => { "Content-Type" => "image/svg+xml" }, ;
|
||||
"body" => hb_CStr( cSvg ) ;
|
||||
} )
|
||||
|
||||
@@ -205,6 +205,8 @@ import "C"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"unsafe"
|
||||
|
||||
@@ -213,30 +215,62 @@ import (
|
||||
"fivenode_go/hbrtl_ext/napibridge"
|
||||
)
|
||||
|
||||
// VM pool — supports both serial (pool=1) and parallel (pool=N) request
|
||||
// handling. Each in-flight request checks out its own VM, so PRG runs
|
||||
// concurrently across libuv worker threads without sharing VM state. The
|
||||
// npm bridge (C side) gives each worker thread its own slot, so one request
|
||||
// awaiting npm I/O does not block others.
|
||||
var (
|
||||
capiInit sync.Once
|
||||
capiVM *hbrt.VM
|
||||
capiMu sync.Mutex
|
||||
capiReq string
|
||||
capiErr string
|
||||
capiInit sync.Once
|
||||
capiPool chan *hbrt.VM
|
||||
capiReqMu sync.Mutex
|
||||
capiReqByVM = map[*hbrt.VM]string{}
|
||||
)
|
||||
|
||||
// FN_NAPI_REQ is registered via HB_FUNC (same mechanism as core/ext RTL) so
|
||||
// PRG direct calls resolve it. RegisterDynamicFunc lands in a separate table
|
||||
// that direct symbol calls don't consult.
|
||||
// FN_NAPI_REQ returns the request JSON for the VM running this PRG. Keyed by
|
||||
// VM (ctx.T.VM()) so parallel requests — each on its own pooled VM — never
|
||||
// see each other's request.
|
||||
func init() {
|
||||
hbrt.HB_FUNC("FN_NAPI_REQ", func(ctx *hbrt.HBContext) {
|
||||
ctx.RetC(capiReq)
|
||||
vm := ctx.T.VM()
|
||||
capiReqMu.Lock()
|
||||
r := capiReqByVM[vm]
|
||||
capiReqMu.Unlock()
|
||||
ctx.RetC(r)
|
||||
})
|
||||
}
|
||||
|
||||
// capiPoolSize: FIVENODE_VM_POOL env (default 4). 1 = serial, N = parallel.
|
||||
func capiPoolSize() int {
|
||||
if v := os.Getenv("FIVENODE_VM_POOL"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return 4
|
||||
}
|
||||
|
||||
func capiEnsure() {
|
||||
capiInit.Do(func() {
|
||||
capiVM = hbrt.NewVM()
|
||||
hbrtl.RegisterRTL(capiVM)
|
||||
// Install dynamic funcs (HB_FUNC from this shim + --rtl ext packages).
|
||||
// vm.Run drains libModules but NOT dynamicFuncs, so do it explicitly.
|
||||
capiVM.RegisterLibModules()
|
||||
// Snapshot the lib registry (PRG modules + every HB_FUNC) ONCE, then
|
||||
// install the SAME set into each pooled VM. RegisterLibModules drains
|
||||
// the global registry, so without snapshotting only the first VM would
|
||||
// have FN_HANDLE and the bridge funcs.
|
||||
mods, dyns := hbrt.LibRegistrySnapshotAndDrain()
|
||||
n := capiPoolSize()
|
||||
capiPool = make(chan *hbrt.VM, n)
|
||||
for i := 0; i < n; i++ {
|
||||
vm := hbrt.NewVM()
|
||||
hbrtl.RegisterRTL(vm)
|
||||
for _, m := range mods {
|
||||
vm.RegisterModule(m)
|
||||
}
|
||||
for j := range dyns {
|
||||
s := dyns[j]
|
||||
vm.RegisterSymbol(&s)
|
||||
}
|
||||
capiPool <- vm
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -251,28 +285,41 @@ func hb_bridge_shutdown() {}
|
||||
|
||||
//export hb_bridge_handle_request
|
||||
func hb_bridge_handle_request(req *C.char) *C.char {
|
||||
capiMu.Lock()
|
||||
defer capiMu.Unlock()
|
||||
capiEnsure()
|
||||
capiReq = C.GoString(req)
|
||||
capiErr = ""
|
||||
vm := <-capiPool
|
||||
defer func() { capiPool <- vm }()
|
||||
|
||||
capiReqMu.Lock()
|
||||
capiReqByVM[vm] = C.GoString(req)
|
||||
capiReqMu.Unlock()
|
||||
|
||||
// On request end (normal or panic): auto-release this request's npm
|
||||
// handles (P3, PRG may skip __end__) and clear its request slot.
|
||||
defer func() {
|
||||
napibridge.ReleaseAll(vm)
|
||||
capiReqMu.Lock()
|
||||
delete(capiReqByVM, vm)
|
||||
capiReqMu.Unlock()
|
||||
}()
|
||||
|
||||
out := ""
|
||||
errStr := ""
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
capiErr = fmt.Sprintf("%v", r)
|
||||
errStr = fmt.Sprintf("%v", r)
|
||||
}
|
||||
}()
|
||||
out = capiVM.Run("FN_HANDLE").AsString()
|
||||
out = vm.Run("FN_HANDLE").AsString()
|
||||
}()
|
||||
if capiErr != "" {
|
||||
if errStr != "" {
|
||||
out = "{\"status\":500,\"headers\":{\"Content-Type\":\"text/plain\"},\"body\":\"PRG error\"}"
|
||||
}
|
||||
return C.CString(out)
|
||||
}
|
||||
|
||||
//export hb_bridge_last_error
|
||||
func hb_bridge_last_error() *C.char { return C.CString(capiErr) }
|
||||
func hb_bridge_last_error() *C.char { return C.CString("") }
|
||||
|
||||
//export hb_bridge_set_auth
|
||||
func hb_bridge_set_auth(a *C.char) {}
|
||||
|
||||
@@ -63,18 +63,76 @@ import (
|
||||
func init() {
|
||||
hbrt.HB_FUNC("FN_REQUIRE", fnRequire)
|
||||
hbrt.HB_FUNC("REQUIRE", fnRequire)
|
||||
// 주의: AWAIT 는 Five 언어 예약어(ASYNC/AWAIT future 모델)라 HB_FUNC 로
|
||||
// 가로챌 수 없다(파서가 AwaitExpr 로 잡고 gengo 백엔드는 코드 미생성→NIL).
|
||||
// 그래서 npm Promise 해소는 FN_AWAIT 로 노출한다.
|
||||
hbrt.HB_FUNC("FN_AWAIT", fnAwait)
|
||||
hbrt.HB_FUNC("FN_LASTERROR", func(ctx *hbrt.HBContext) { ctx.RetC(lastErr) })
|
||||
}
|
||||
|
||||
// FN_AWAIT( oPromise ) — async npm 메서드가 돌려준 Promise 핸들 객체를 해소한다.
|
||||
// 메서드 호출이 Promise 면 npmCall 이 "handle:N" 으로 감싸 오는데(=객체), 그
|
||||
// 객체를 FN_AWAIT 에 넘기면 npmAwait 가 결과를 돌려준다.
|
||||
//
|
||||
// 코루틴식 동작: PRG 핸들러는 libuv 워커 스레드(handleRequestAsync)에서 돌고,
|
||||
// npmAwait 는 워커 분기에서 tfn_async_call({action:await}) 로 메인에 위임한 뒤
|
||||
// cond 로 블록(=yield)한다. 메인 스레드 이벤트루프는 그대로 살아 Promise.then 이
|
||||
// 자연스레 발화하고 fn_deferred_resolve 가 결과를 써 워커를 깨운다. 메인스레드
|
||||
// 바쁜-펌프(uv_run UV_RUN_ONCE)는 쓰지 않는다.
|
||||
// cSvg := FN_AWAIT( oQR:toString( cUrl, { "type" => "svg" } ) )
|
||||
func fnAwait(ctx *hbrt.HBContext) {
|
||||
if ctx.PCount() < 1 {
|
||||
ctx.RetNil()
|
||||
return
|
||||
}
|
||||
v := ctx.Param(1)
|
||||
h := -1
|
||||
if a := v.AsArray(); a != nil && len(a.Items) > 0 {
|
||||
h = int(a.Items[0].AsNumInt())
|
||||
}
|
||||
if h < 0 {
|
||||
lastErr = "AWAIT: not a promise handle"
|
||||
ctx.RetNil()
|
||||
return
|
||||
}
|
||||
ctx.RetVal(resultToValue(ctx.T.VM(), cbAwait(h)))
|
||||
}
|
||||
|
||||
var (
|
||||
lastErr string
|
||||
clsMu sync.Mutex
|
||||
clsBySig = map[string]uint16{}
|
||||
|
||||
// acquired — VM(=요청)별로 발급된 npm 핸들. PRG 가 oH:__end__() 을 안 불러도
|
||||
// 요청이 끝나면 ReleaseAll(vm) 이 그 요청 핸들만 정리한다(누수 방지). 병렬
|
||||
// 요청은 각자 다른 VM 이라 요청끼리 핸들이 섞이지 않는다(VM 풀 모델).
|
||||
acqMu sync.Mutex
|
||||
acquired = map[*hbrt.VM][]int{}
|
||||
)
|
||||
|
||||
// StoreCallbacks — 애드온이 tfn_register_callbacks 로 넘긴 콜백 묶음을 보관.
|
||||
func StoreCallbacks(cb unsafe.Pointer) { C.nb_store(cb) }
|
||||
|
||||
// track — 발급된 핸들을 해당 VM(요청)의 정리 대상에 등록.
|
||||
func track(vm *hbrt.VM, h int) {
|
||||
acqMu.Lock()
|
||||
acquired[vm] = append(acquired[vm], h)
|
||||
acqMu.Unlock()
|
||||
}
|
||||
|
||||
// ReleaseAll — 요청 종료 시 호출(shim 의 defer). 그 VM 에서 발급된 핸들에만
|
||||
// cbEnd 를 보낸다. JS end 액션은 delete(이미 __end__/await 로 지워졌으면 no-op)
|
||||
// 라 수동/중복 호출과 충돌 없음.
|
||||
func ReleaseAll(vm *hbrt.VM) {
|
||||
acqMu.Lock()
|
||||
hs := acquired[vm]
|
||||
delete(acquired, vm)
|
||||
acqMu.Unlock()
|
||||
for _, h := range hs {
|
||||
cbEnd(h)
|
||||
}
|
||||
}
|
||||
|
||||
// ── C 콜백 래퍼 ──────────────────────────────────────────────────────────
|
||||
func cbRequire(module string) int {
|
||||
cs := C.CString(module)
|
||||
@@ -130,11 +188,12 @@ func fnRequire(ctx *hbrt.HBContext) {
|
||||
ctx.RetL(false)
|
||||
return
|
||||
}
|
||||
ctx.RetVal(wrapHandle(h))
|
||||
ctx.RetVal(wrapHandle(ctx.T.VM(), h))
|
||||
}
|
||||
|
||||
// wrapHandle: Node 핸들 → 메서드명을 조회해 그 이름들을 가진 PRG 객체 생성.
|
||||
func wrapHandle(h int) hbrt.Value {
|
||||
func wrapHandle(vm *hbrt.VM, h int) hbrt.Value {
|
||||
track(vm, h) // 요청 종료 시 자동 __end__ 대상에 등록(VM=요청별)
|
||||
var names []string
|
||||
_ = json.Unmarshal([]byte(cbMethods(h)), &names)
|
||||
classID := classForSig(names)
|
||||
@@ -175,7 +234,7 @@ func makeDispatch(jsName string) hbrt.MethodFunc {
|
||||
}
|
||||
argsJSON, _ := json.Marshal(args)
|
||||
res := cbCall(h, jsName, string(argsJSON))
|
||||
t.RetVal(resultToValue(res))
|
||||
t.RetVal(resultToValue(t.VM(), res))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,12 +256,12 @@ func selfHandle(t *hbrt.Thread) int {
|
||||
|
||||
// resultToValue: npmCall 결과 문자열 → PRG 값.
|
||||
// "handle:N" → 객체 핸들(체이닝/버퍼 등) / {..}|[..] → 네이티브 / 그 외 → 문자열
|
||||
func resultToValue(s string) hbrt.Value {
|
||||
func resultToValue(vm *hbrt.VM, s string) hbrt.Value {
|
||||
if strings.HasPrefix(s, "handle:") {
|
||||
var h int
|
||||
_, err := fmtSscan(s[len("handle:"):], &h)
|
||||
if err == nil {
|
||||
return wrapHandle(h)
|
||||
return wrapHandle(vm, h)
|
||||
}
|
||||
return hbrt.MakeNil()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user