mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-22 01:48:01 +00:00
feat(lua): elastic adapter worker pools instead of monotonic growth
ConfigureConcurrency() set each adapter pool's target to the sum of
max_concurrent over its sources (108 on this deployment) and `created` only
ever went UP: once a Lua state was booted it was parked forever, so a
long-running gateway's resident state count was a high-water mark of all
traffic it had ever seen, never of what it currently needs.
Pools are now sized from three live inputs:
* the adapter's MAX CONCURRENCY (sum of max_concurrent) is a ceiling, not a
preallocation, and it sets the growth step:
growStep = clamp(ceil(maxW/8), 1, 8). A 64-wide adapter warms 8 states at
once on a spike; an 8-wide one creeps up one at a time.
* the LIVE CONNECTION COUNT (inUse, i.e. checked-out states) sets the shrink
step: shrinkStep = clamp(ceil(excess/(1+inUse)), 1, excess). With no
connections the slack collapses in a single round; a busy adapter gives up
one state per round so the hot path keeps its warm states.
* how many states already exist (created / len(idle)) decides how much room
is left to grow and how much can be reclaimed.
Batch prewarm only fires on genuine contention (a miss while every existing
state is checked out), so a single sequential caller keeps reusing one state
rather than burning a whole grow step on a cold start. A single VM-level
janitor goroutine (not one per adapter) reclaims idle states every 30s, and
shrinkGraceRounds=2 plus idleHeadroom=1 keep a gap between requests from being
mistaken for the end of a load period; any checkout resets the grace counter.
`created` now decrements on reclaim and on shutdown, and release() closes a
state outright when the ceiling was lowered underneath it, so shrinking
max_concurrent in the config gives memory back immediately instead of parking
orphans until restart.
PoolStats() exposes created/idle/in_use/waiting/max/resident/grow_step/
shrink_step/peak_in_use for the status API.
Measured on the test instance (single mock source, max_concurrent=64):
idle created=1; 50 concurrent requests -> created=10 (peak_in_use=5, ceiling
respected); after 95s of silence -> created=1. On production after deploy: 13
adapters, 1 resident Lua state total with ceilings up to 76.
This commit is contained in:
JianFeeeee
@ -5,7 +5,9 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// freshAdapterDir returns a path under a temp dir that does not exist yet, so
|
||||
@ -484,8 +486,8 @@ func TestAnthropicToolRoundTrip(t *testing.T) {
|
||||
}
|
||||
var r1 struct {
|
||||
Tools []struct {
|
||||
Name string `json:"name"`
|
||||
InputSchema map[string]interface{} `json:"input_schema"`
|
||||
Name string `json:"name"`
|
||||
InputSchema map[string]interface{} `json:"input_schema"`
|
||||
} `json:"tools"`
|
||||
Messages []map[string]interface{} `json:"messages"`
|
||||
}
|
||||
@ -512,12 +514,12 @@ func TestAnthropicToolRoundTrip(t *testing.T) {
|
||||
Messages []struct {
|
||||
Role string `json:"role"`
|
||||
Content []struct {
|
||||
Type string `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Type string `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Input map[string]interface{} `json:"input"`
|
||||
ToolUseID string `json:"tool_use_id"`
|
||||
ContentText string `json:"content"`
|
||||
ToolUseID string `json:"tool_use_id"`
|
||||
ContentText string `json:"content"`
|
||||
} `json:"content"`
|
||||
} `json:"messages"`
|
||||
}
|
||||
@ -746,3 +748,494 @@ func TestAdaptersReportZeroCacheHit(t *testing.T) {
|
||||
t.Fatalf("openai fabricated cache details when upstream reported none: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- elastic pool sizing (plan 阶段 2) ----
|
||||
|
||||
// poolOf reaches into the VM for one adapter's pool. Tests assert on internal
|
||||
// counters because the whole point of the change is that they no longer grow
|
||||
// monotonically.
|
||||
func poolOf(t *testing.T, v *VM, name string) *adapterPool {
|
||||
t.Helper()
|
||||
p := v.pool(name)
|
||||
if p == nil {
|
||||
t.Fatalf("adapter %q not loaded", name)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func poolCounts(p *adapterPool) (created, idle, inUse int) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.created, len(p.idle), p.inUse
|
||||
}
|
||||
|
||||
// TestPoolStartsEmpty is the startup-memory claim: loading adapters must not
|
||||
// boot a single Lua state, no matter how large the configured ceiling is.
|
||||
func TestPoolStartsEmpty(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 64, "deepseek": 32})
|
||||
for _, st := range vm.PoolStats() {
|
||||
if st.Created != 0 || st.Idle != 0 || st.InUse != 0 {
|
||||
t.Fatalf("adapter %q booted eagerly: created=%d idle=%d in_use=%d",
|
||||
st.Name, st.Created, st.Idle, st.InUse)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestGrowStepFollowsMaxConcurrent pins requirement 3: the growth step is
|
||||
// derived from the adapter's maximum concurrency.
|
||||
func TestGrowStepFollowsMaxConcurrent(t *testing.T) {
|
||||
p := newAdapterPool("t", "return {name='t'}", staticInfo{name: "t"})
|
||||
for _, tc := range []struct{ max, want int }{
|
||||
{1, 1}, {4, 1}, {8, 1}, {9, 2}, {16, 2}, {32, 4}, {64, 8}, {200, 8},
|
||||
} {
|
||||
p.setMax(tc.max)
|
||||
p.mu.Lock()
|
||||
got := p.growStepLocked()
|
||||
p.mu.Unlock()
|
||||
if got != tc.want {
|
||||
t.Fatalf("max=%d: grow step = %d, want %d", tc.max, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestShrinkStepFollowsConnections pins requirement 4: the shrink step is
|
||||
// driven by the live connection count — no connections means collapse, busy
|
||||
// means give up one state at a time.
|
||||
func TestShrinkStepFollowsConnections(t *testing.T) {
|
||||
p := newAdapterPool("t", "return {name='t'}", staticInfo{name: "t"})
|
||||
p.setMax(32)
|
||||
p.used = true
|
||||
|
||||
cases := []struct {
|
||||
created, idle, inUse, want int
|
||||
why string
|
||||
}{
|
||||
{created: 9, idle: 9, inUse: 0, want: 8, why: "no connections: collapse to the resident floor in one round"},
|
||||
{created: 9, idle: 8, inUse: 1, want: 4, why: "1 connection: halve the slack"},
|
||||
{created: 9, idle: 6, inUse: 3, want: 2, why: "3 connections: quarter the slack"},
|
||||
{created: 9, idle: 2, inUse: 7, want: 1, why: "busy: release one state at a time"},
|
||||
{created: 4, idle: 0, inUse: 4, want: 0, why: "everything checked out: nothing to reclaim"},
|
||||
{created: 1, idle: 1, inUse: 0, want: 0, why: "at the resident floor: keep the warm state"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
p.mu.Lock()
|
||||
p.created, p.inUse = tc.created, tc.inUse
|
||||
p.idle = make([]*worker, tc.idle)
|
||||
got := p.shrinkStepLocked()
|
||||
p.mu.Unlock()
|
||||
if got != tc.want {
|
||||
t.Fatalf("%s: created=%d idle=%d inUse=%d -> step %d, want %d",
|
||||
tc.why, tc.created, tc.idle, tc.inUse, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPoolGrowsOnDemandAndReclaims is the end-to-end elasticity loop: boot on
|
||||
// demand, stay under the ceiling, then collapse back to the resident floor.
|
||||
func TestPoolGrowsOnDemandAndReclaims(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 16})
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
if created, _, _ := poolCounts(p); created != 0 {
|
||||
t.Fatalf("pool must start empty, created=%d", created)
|
||||
}
|
||||
|
||||
// concurrent load: hold several workers at once so the pool must grow
|
||||
const hold = 6
|
||||
var wg sync.WaitGroup
|
||||
release := make(chan struct{})
|
||||
for i := 0; i < hold; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Errorf("acquire: %v", err)
|
||||
return
|
||||
}
|
||||
<-release
|
||||
p.release(w)
|
||||
}()
|
||||
}
|
||||
// wait until all six are checked out
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
_, _, inUse := poolCounts(p)
|
||||
if inUse >= hold {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("workers never checked out, in_use=%d", inUse)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
created, _, inUse := poolCounts(p)
|
||||
if inUse != hold {
|
||||
t.Fatalf("in_use = %d, want %d", inUse, hold)
|
||||
}
|
||||
if created < hold {
|
||||
t.Fatalf("created = %d, must cover %d concurrent checkouts", created, hold)
|
||||
}
|
||||
if created > 16 {
|
||||
t.Fatalf("created = %d exceeds the ceiling 16", created)
|
||||
}
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
// all connections closed: the janitor collapses the pool to the floor.
|
||||
// shrinkGraceRounds rounds are required, and one extra round proves the
|
||||
// floor is stable rather than shrinking to zero.
|
||||
for i := 0; i < shrinkGraceRounds+2; i++ {
|
||||
vm.ReclaimIdleNow()
|
||||
}
|
||||
created, idle, inUse := poolCounts(p)
|
||||
if inUse != 0 {
|
||||
t.Fatalf("in_use = %d after release, want 0", inUse)
|
||||
}
|
||||
if created != residentWorkers || idle != residentWorkers {
|
||||
t.Fatalf("after reclaim: created=%d idle=%d, want %d/%d (resident floor)",
|
||||
created, idle, residentWorkers, residentWorkers)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReclaimNeedsGraceRounds: a single quiet round must not tear the pool
|
||||
// down — a gap between requests is not the end of a load period.
|
||||
func TestReclaimNeedsGraceRounds(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 8})
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
// build slack: 4 states created, all returned
|
||||
var ws []*worker
|
||||
for i := 0; i < 4; i++ {
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire: %v", err)
|
||||
}
|
||||
ws = append(ws, w)
|
||||
}
|
||||
for _, w := range ws {
|
||||
p.release(w)
|
||||
}
|
||||
created, _, _ := poolCounts(p)
|
||||
if created != 4 {
|
||||
t.Fatalf("created = %d, want 4", created)
|
||||
}
|
||||
|
||||
if n := vm.ReclaimIdleNow(); n != 0 {
|
||||
t.Fatalf("first quiet round must not reclaim (grace), closed %d", n)
|
||||
}
|
||||
if c, _, _ := poolCounts(p); c != 4 {
|
||||
t.Fatalf("created = %d after the grace round, want 4", c)
|
||||
}
|
||||
if n := vm.ReclaimIdleNow(); n == 0 {
|
||||
t.Fatal("second quiet round must reclaim")
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckoutResetsGrace: live traffic between janitor rounds must cancel a
|
||||
// pending shrink decision.
|
||||
func TestCheckoutResetsGrace(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 8})
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
var ws []*worker
|
||||
for i := 0; i < 4; i++ {
|
||||
w, _ := p.acquire()
|
||||
ws = append(ws, w)
|
||||
}
|
||||
for _, w := range ws {
|
||||
p.release(w)
|
||||
}
|
||||
if n := vm.ReclaimIdleNow(); n != 0 {
|
||||
t.Fatalf("grace round reclaimed %d", n)
|
||||
}
|
||||
// a request arrives: the grace counter resets
|
||||
w, _ := p.acquire()
|
||||
p.release(w)
|
||||
if n := vm.ReclaimIdleNow(); n != 0 {
|
||||
t.Fatalf("traffic must reset the grace counter, reclaimed %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPoolCeilingIsRespectedUnderContention: acquires beyond the ceiling block
|
||||
// instead of booting unbounded states.
|
||||
func TestPoolCeilingIsRespectedUnderContention(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 2})
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
w1, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire 1: %v", err)
|
||||
}
|
||||
w2, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire 2: %v", err)
|
||||
}
|
||||
got := make(chan struct{})
|
||||
go func() {
|
||||
w3, err := p.acquire()
|
||||
if err == nil {
|
||||
p.release(w3)
|
||||
}
|
||||
close(got)
|
||||
}()
|
||||
select {
|
||||
case <-got:
|
||||
t.Fatal("a third acquire must block at the ceiling of 2")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
if created, _, _ := poolCounts(p); created != 2 {
|
||||
t.Fatalf("created = %d, must not exceed the ceiling 2", created)
|
||||
}
|
||||
p.release(w1)
|
||||
select {
|
||||
case <-got:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("releasing a state must unblock the waiter")
|
||||
}
|
||||
p.release(w2)
|
||||
}
|
||||
|
||||
// TestLoweredCeilingReleasesStates: shrinking max_concurrent in the config must
|
||||
// actually give memory back rather than leaving orphaned states parked.
|
||||
func TestLoweredCeilingReleasesStates(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 8})
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
var ws []*worker
|
||||
for i := 0; i < 6; i++ {
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire: %v", err)
|
||||
}
|
||||
ws = append(ws, w)
|
||||
}
|
||||
if created, _, _ := poolCounts(p); created != 6 {
|
||||
t.Fatalf("created = %d, want 6", created)
|
||||
}
|
||||
// operator lowers the ceiling while the states are still checked out
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 2})
|
||||
for _, w := range ws {
|
||||
p.release(w)
|
||||
}
|
||||
created, idle, _ := poolCounts(p)
|
||||
if created > 2 || idle > 2 {
|
||||
t.Fatalf("lowering the ceiling must drop the excess: created=%d idle=%d", created, idle)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPoolStatsSurfacesAlgorithm checks the observability payload the status API
|
||||
// exposes.
|
||||
func TestPoolStatsSurfacesAlgorithm(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 32})
|
||||
p := poolOf(t, vm, "openai")
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire: %v", err)
|
||||
}
|
||||
|
||||
var st PoolStats
|
||||
for _, s := range vm.PoolStats() {
|
||||
if s.Name == "openai" {
|
||||
st = s
|
||||
}
|
||||
}
|
||||
if st.Max != 32 {
|
||||
t.Fatalf("Max = %d, want 32", st.Max)
|
||||
}
|
||||
if st.InUse != 1 || st.Created != 1 {
|
||||
t.Fatalf("InUse=%d Created=%d, want 1/1", st.InUse, st.Created)
|
||||
}
|
||||
if st.GrowStep != 4 {
|
||||
t.Fatalf("GrowStep = %d, want ceil(32/8)=4", st.GrowStep)
|
||||
}
|
||||
if st.PeakInUse != 1 {
|
||||
t.Fatalf("PeakInUse = %d, want 1", st.PeakInUse)
|
||||
}
|
||||
p.release(w)
|
||||
|
||||
// stats must be sorted by name for a stable UI
|
||||
names := []string{}
|
||||
for _, s := range vm.PoolStats() {
|
||||
names = append(names, s.Name)
|
||||
}
|
||||
for i := 1; i < len(names); i++ {
|
||||
if names[i-1] > names[i] {
|
||||
t.Fatalf("PoolStats not sorted: %v", names)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestHotReloadKeepsCeiling: replacing an adapter's code must not reset its
|
||||
// elastic ceiling back to 1.
|
||||
func TestHotReloadKeepsCeiling(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 24})
|
||||
if err := vm.LoadAdapterSource("openai", "return {name='openai', version='9'}"); err != nil {
|
||||
t.Fatalf("reload: %v", err)
|
||||
}
|
||||
p := poolOf(t, vm, "openai")
|
||||
p.mu.Lock()
|
||||
maxW := p.maxW
|
||||
p.mu.Unlock()
|
||||
if maxW != 24 {
|
||||
t.Fatalf("ceiling after hot reload = %d, want 24", maxW)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentTransformStress exercises the grow/shrink paths together under
|
||||
// real hook calls, and asserts the accounting stays consistent (no leaked
|
||||
// states, no negative counters).
|
||||
func TestConcurrentTransformStress(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 16})
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 40; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
raw := `{"model":"m","messages":[{"role":"user","content":"hi"}]}`
|
||||
if _, err := vm.Transform("openai", "transform_request", raw); err != nil {
|
||||
t.Errorf("transform: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
p := poolOf(t, vm, "openai")
|
||||
created, idle, inUse := poolCounts(p)
|
||||
if inUse != 0 {
|
||||
t.Fatalf("in_use = %d after the stress run, want 0 (leaked checkout)", inUse)
|
||||
}
|
||||
if created != idle {
|
||||
t.Fatalf("created=%d idle=%d must match when nothing is checked out", created, idle)
|
||||
}
|
||||
if created > 16 {
|
||||
t.Fatalf("created = %d exceeds the ceiling", created)
|
||||
}
|
||||
for i := 0; i < shrinkGraceRounds+2; i++ {
|
||||
vm.ReclaimIdleNow()
|
||||
}
|
||||
created, idle, _ = poolCounts(p)
|
||||
if created != residentWorkers || idle != residentWorkers {
|
||||
t.Fatalf("post-stress reclaim: created=%d idle=%d, want %d/%d",
|
||||
created, idle, residentWorkers, residentWorkers)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopIsIdempotent guards the janitor's stop channel against a double close.
|
||||
func TestStopIsIdempotent(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
vm.Stop()
|
||||
vm.Stop() // must not panic
|
||||
}
|
||||
|
||||
// TestSequentialTrafficDoesNotBatch: a single caller looping must keep reusing
|
||||
// one warm state — batch prewarm is for genuine concurrency only.
|
||||
func TestSequentialTrafficDoesNotBatch(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 64}) // grow step would be 8
|
||||
p := poolOf(t, vm, "openai")
|
||||
for i := 0; i < 20; i++ {
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire %d: %v", i, err)
|
||||
}
|
||||
p.release(w)
|
||||
}
|
||||
created, _, _ := poolCounts(p)
|
||||
if created != 1 {
|
||||
t.Fatalf("sequential traffic must reuse one state, created=%d", created)
|
||||
}
|
||||
}
|
||||
|
||||
// TestContentionBatchPrewarms: when concurrent demand exceeds the warm set, the
|
||||
// pool warms a whole grow step instead of one state per request.
|
||||
func TestContentionBatchPrewarms(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
vm.ConfigureConcurrency(map[string]int{"openai": 32}) // grow step = 4
|
||||
p := poolOf(t, vm, "openai")
|
||||
|
||||
// hold the first state, then miss while it is busy -> contended
|
||||
w1, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire 1: %v", err)
|
||||
}
|
||||
w2, err := p.acquire()
|
||||
if err != nil {
|
||||
t.Fatalf("acquire 2: %v", err)
|
||||
}
|
||||
// the batch is booted in the background; wait for it to land
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
created, _, _ := poolCounts(p)
|
||||
if created >= 5 { // 2 checked out + 3 prewarmed
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("contended miss did not batch prewarm, created=%d", created)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
created, _, _ := poolCounts(p)
|
||||
if created > 32 {
|
||||
t.Fatalf("prewarm exceeded the ceiling, created=%d", created)
|
||||
}
|
||||
p.release(w1)
|
||||
p.release(w2)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user