internal/router/group.go

Rama main · réplica de lectura del repositorio original.

package router

import (
	"fmt"

	"github.com/quartermaster-labs/quartermaster/internal/config"
	"github.com/quartermaster-labs/quartermaster/internal/logmon"
	"github.com/quartermaster-labs/quartermaster/internal/process"
	"github.com/quartermaster-labs/quartermaster/internal/router/scheduler"
)

type Group struct {
	*baseRouter
}

func NewGroup(conf config.Config, proxylog, upstreamlog *logmon.Monitor) (*Group, error) {
	// plan derives the eviction planner and the model→config set of every process
	// the group router should run under a config. Shared by initial construction
	// and ApplyConfig (live reload).
	// One holder for every Swapper this router will ever build: plan() runs again
	// on each ApplyConfig, and the probe is installed once, after construction.
	liveVram := &liveVramHolder{}

	plan := func(c config.Config) (scheduler.Swapper, map[string]config.ModelConfig, error) {
		modelToGroup, err := buildModelToGroup(c)
		if err != nil {
			return nil, nil, err
		}
		want := make(map[string]config.ModelConfig, len(modelToGroup))
		for mid := range modelToGroup {
			modelCfg, _, ok := c.FindConfig(mid)
			if !ok {
				return nil, nil, fmt.Errorf("no model config for %q", mid)
			}
			want[mid] = modelCfg
		}
		return &groupSwapper{config: c, modelToGroup: modelToGroup, logger: proxylog, liveVram: liveVram}, want, nil
	}

	swapper, want, err := plan(conf)
	if err != nil {
		return nil, err
	}

	processes := make(map[string]process.Process, len(want))
	base, err := newBaseRouter("group", conf, processes, proxylog, swapper)
	if err != nil {
		return nil, fmt.Errorf("creating base router: %w", err)
	}
	base.liveVram = liveVram
	base.plan = plan
	base.makeProcess = func(id string, mc config.ModelConfig) (process.Process, error) {
		return process.New(base.procCtx, id, mc, logmon.NewWriter(upstreamlog), proxylog)
	}

	for mid, modelCfg := range want {
		p, err := base.makeProcess(mid, modelCfg)
		if err != nil {
			base.shutdownFn()
			base.procCancel()
			return nil, fmt.Errorf("creating process for %q: %w", mid, err)
		}
		processes[mid] = p
	}

	g := &Group{baseRouter: base}
	go base.run()
	return g, nil
}

// buildModelToGroup maps each grouped model ID to its group, rejecting a model
// that appears in more than one group.
func buildModelToGroup(conf config.Config) (map[string]string, error) {
	modelToGroup := make(map[string]string)
	for gid, gcfg := range conf.Routing.Router.Settings.Groups {
		for _, mid := range gcfg.Members {
			if existing, dup := modelToGroup[mid]; dup {
				return nil, fmt.Errorf("model %q is in multiple groups: %q and %q", mid, existing, gid)
			}
			modelToGroup[mid] = gid
		}
	}
	return modelToGroup, nil
}

// groupSwapper decides evictions from static group configuration.
//
// Same-group siblings are stopped when the group has swap=true. Cross-group
// members are stopped only when the target's group is exclusive; loading a
// model from a non-exclusive group leaves running exclusive groups alone,
// matching the gotcha in the original ProcessGroup behaviour.
type groupSwapper struct {
	config       config.Config
	modelToGroup map[string]string
	logger       *logmon.Monitor
	// liveVram is the live VRAM-ceiling probe budgetEviction tightens against.
	// Read-only here; nil in tests that only exercise the static policy.
	liveVram *liveVramHolder
}

// EvictionFor splits the residents by memory pool (GroupConfig.Pool). A
// ram-pool (CPU-only) model holds no VRAM, so it is invisible to the VRAM
// admission and never evicted for it, and it never evicts a vram-pool model in
// turn: the two pools only meet in the RAM check, where a GPU model's offloaded
// share (EstRamGB) can push CPU models out but not the other way round.
func (p *groupSwapper) EvictionFor(target string, running []string) []string {
	var cpuRunning, gpuRunning []string
	for _, id := range running {
		if p.inRAMPool(id) {
			cpuRunning = append(cpuRunning, id)
		} else {
			gpuRunning = append(gpuRunning, id)
		}
	}
	if p.inRAMPool(target) {
		if evict, ok := p.ramEviction(target, running, nil); ok {
			return evict
		}
		return p.staticEviction(target, cpuRunning)
	}

	evict, ok := p.budgetEviction(target, gpuRunning)
	if !ok {
		evict = p.staticEviction(target, gpuRunning)
	}
	if more, ok := p.ramEviction(target, running, evict); ok {
		evict = append(evict, more...)
	}
	return evict
}

// inRAMPool reports whether a model's group is admitted against the RAM budget.
func (p *groupSwapper) inRAMPool(id string) bool {
	return p.config.Routing.Router.Settings.Groups[p.modelToGroup[id]].IsRAMPool()
}

// ramEviction is budgetEviction over system memory: the target's EstRamGB plus
// every resident's (both pools, minus those already being evicted) must fit
// RamBudgetGB, and only ram-pool, non-persistent residents are evicted to make
// it, least-recently-used first. A ram-pool resident with no estimate is
// evicted outright when the TARGET is ram-pool (same rule as the VRAM side);
// when a GPU model is loading it is left alone, since an unmeasured CPU model
// is no reason to tear down something the GPU load does not collide with.
//
// ok=false: no RAM budget, or a ram-pool target with no estimate. The caller
// then falls back to the static group rule (ram-pool target) or adds nothing
// (GPU target).
func (p *groupSwapper) ramEviction(target string, running, already []string) ([]string, bool) {
	budget := p.config.RamBudgetGB
	if budget <= 0 {
		return nil, false
	}
	targetRAM := p.inRAMPool(target)
	targetGB := p.config.Models[target].EstRamGB
	if targetRAM && targetGB <= 0 {
		return nil, false
	}

	skip := map[string]struct{}{target: {}}
	for _, id := range already {
		skip[id] = struct{}{}
	}
	total := targetGB
	var victims, candidates []string
	for _, id := range running {
		if _, dup := skip[id]; dup {
			continue
		}
		skip[id] = struct{}{}
		estGB := p.config.Models[id].EstRamGB
		gcfg := p.config.Routing.Router.Settings.Groups[p.modelToGroup[id]]
		switch {
		case !gcfg.IsRAMPool() || gcfg.Persistent:
			total += estGB
		case estGB <= 0:
			if targetRAM {
				victims = append(victims, id)
			}
		default:
			total += estGB
			candidates = append(candidates, id)
		}
	}
	for _, id := range candidates {
		if total <= budget {
			break
		}
		total -= p.config.Models[id].EstRamGB
		victims = append(victims, id)
	}
	return victims, true
}

// staticEviction is the legacy group policy, used when no budget applies. The
// caller has already narrowed running to the target's pool where that matters.
func (p *groupSwapper) staticEviction(target string, running []string) []string {
	tg := p.modelToGroup[target]
	tgCfg := p.config.Routing.Router.Settings.Groups[tg]

	seen := make(map[string]struct{})
	var result []string
	consider := func(mID string) {
		if mID == target {
			return
		}
		if _, dup := seen[mID]; dup {
			return
		}
		og := p.modelToGroup[mID]
		switch {
		case og == tg && tgCfg.Swap:
			seen[mID] = struct{}{}
			result = append(result, mID)
		// the previous ProcessGroup behaviour did not unload exclusive groups
		// when loading a non-exclusive model. This maintains that gotcha
		// for backwards compatibility. The newer swap matrix approach does not
		// have this issue.
		case og != tg && tgCfg.Exclusive && !p.inRAMPool(mID):
			if ogCfg := p.config.Routing.Router.Settings.Groups[og]; !ogCfg.Persistent {
				seen[mID] = struct{}{}
				result = append(result, mID)
			}
		}
	}

	for _, mID := range running {
		consider(mID)
	}
	return result
}

// budgetEviction is the VRAM-budget admission rule, which REPLACES the static
// group policy above whenever config.VramBudgetGB is set and the target's
// footprint is known: models stay co-resident as long as the sum of their
// estVramGB fits the budget, and only when the incoming model would push the
// total over it are residents evicted — least-recently-used first, and only as
// many as it takes to fit.
//
// This is also what keeps the fork's cross-port accounting honest once groups
// stop being exclusive: one scheduler sees every group's residents, so the sum
// it admits against covers all listeners rather than relying on group
// exclusivity to serialise the GPU.
//
// ok=false hands the decision back to the static policy. That happens when no
// budget is configured (hand-written config, or multiResident off) or when the
// TARGET has no estimate — admitting a model of unknown size against a budget is
// exactly the case where guessing OOMs the box, so it falls back to the
// conservative "evict per group config" behaviour.
//
// Two rules keep the accounting from lying:
//   - persistent-group members can never be evicted, but they still OCCUPY the
//     budget (they hold real VRAM). ASR/SAM/TTS.cpp models emit no estimate
//     precisely because they are CPU-resident, so they add nothing here.
//   - an evictable resident with NO estimate is always evicted. Its footprint is
//     unknown, so keeping it would silently overrun the budget; this degenerates
//     to the legacy one-at-a-time behaviour for models the sizer can't measure.
//
// Purity: reads only config, the args, and a live VRAM SNAPSHOT (a lock-free
// read of a value the server samples elsewhere) - no mutation (see
// Swapper.EvictionFor).
func (p *groupSwapper) budgetEviction(target string, running []string) ([]string, bool) {
	budget, ok := p.effectiveBudgetGB()
	if !ok {
		return nil, false
	}
	targetGB := p.config.Models[target].EstVramGB
	if targetGB <= 0 {
		return nil, false
	}

	total := targetGB
	var victims []string
	var candidates []string // evictable, known cost, LRU-first (runningSet order)
	seen := map[string]struct{}{target: {}}
	for _, id := range running {
		if _, dup := seen[id]; dup {
			continue
		}
		seen[id] = struct{}{}

		gcfg := p.config.Routing.Router.Settings.Groups[p.modelToGroup[id]]
		estGB := p.config.Models[id].EstVramGB
		switch {
		case gcfg.Persistent:
			total += estGB
		case estGB <= 0:
			victims = append(victims, id)
		default:
			total += estGB
			candidates = append(candidates, id)
		}
	}

	// Free the least-recently-used residents until the target fits. If even
	// evicting every candidate leaves us over budget, they all go — the spawn
	// guard (autogen.LiveOffloadArgs) then sizes or refuses the load against
	// what the GPU actually has free.
	for _, id := range candidates {
		if total <= budget {
			break
		}
		total -= p.config.Models[id].EstVramGB
		victims = append(victims, id)
	}
	return victims, true
}

// effectiveBudgetGB is the VRAM the resident model set is admitted into: the
// configured vramBudgetGB, TIGHTENED to the live ceiling whenever a foreign GPU
// client is holding memory the static budget assumed we had.
//
// This is the OOM guard's admission half. Without it the router happily evicts
// residents to make room inside a 24 GB budget while a game holds 8 GB of the
// card, and the spawn guard (autogen.LiveOffloadArgs) then refuses the load
// anyway - the worst of both outcomes, since the evicted models are gone and
// the requested one never arrives. Charging the foreign VRAM here instead means
// the router either keeps fewer models co-resident or declines to disturb the
// resident set at all.
//
// ok=false hands the decision back to the static group policy: no budget
// configured. A missing or untrustworthy live reading is NOT a reason to fall
// back - it just leaves the static budget in force, which is the behaviour that
// shipped before the probe existed.
func (p *groupSwapper) effectiveBudgetGB() (float64, bool) {
	budget := p.config.VramBudgetGB
	if budget <= 0 {
		return 0, false
	}
	if live, ok := p.liveVram.ceiling(); ok && live < budget {
		budget = live
	}
	return budget, true
}

// OnSwapStart logs the budget arithmetic behind this swap — the one place a
// Swapper may log, since it fires once per real swap while EvictionFor runs on
// every request and every queue drain.
func (p *groupSwapper) OnSwapStart(target string, running []string) {
	if p.logger == nil || p.config.VramBudgetGB <= 0 {
		return
	}
	residentGB := 0.0
	for _, id := range running {
		if id != target {
			residentGB += p.config.Models[id].EstVramGB
		}
	}
	budget, _ := p.effectiveBudgetGB()
	note := ""
	if budget < p.config.VramBudgetGB {
		note = fmt.Sprintf(" (tightened from %.2fGB - foreign GPU use)", p.config.VramBudgetGB)
	}
	p.logger.Debugf("group: admitting %s (est %.2fGB) against resident %.2fGB of %.2fGB budget%s",
		target, p.config.Models[target].EstVramGB, residentGB, budget, note)
}