Files
LocalAI/pkg/model/watchdog.go
mudler's LocalAI [bot] f735cb24c0 fix(worker): reap deleted backends and stop models that live on a worker (#10956)
* fix(worker): reap deleted backends and stop models that live on a worker

Three related backend-lifecycle defects, all reachable from the same
production incident on a Jetson/Thor worker: a deleted backend's gRPC
process survived ~40 minutes with its directory removed from disk, a later
model load was routed to that orphan and failed with a certifi path pointing
into the deleted directory, and the admin could not stop the model because
the frontend reported it as not loaded.

1. backend.delete orphaned the process it claimed to delete
------------------------------------------------------------
s.processes is keyed by `modelID#replicaIndex` (buildProcessKey), so the
backend name never appeared in a key and was recorded nowhere on the
process. backend.delete resolved its target via isRunning/stopBackend, whose
prefix path only matches a bare *modelID* - a delete keyed on a backend name
resolved to zero keys, the stop silently no-op'd, and the files were removed
out from under a live process.

The install fast path then handed that orphan back out: it returns any live
process for the (model, replica) slot without checking which backend started
it, so a reinstalled variant inherited the deleted backend's port.

- Record backendName on backendProcess, threaded installBackend ->
  startBackend.
- Add resolveProcessKeysForBackend, matching the recorded name and resolving
  alias <-> concrete via ListSystemBackends *before* DeleteBackendFromSystem
  erases the metadata that carries the alias. Alias resolution failure
  degrades to name-only matching so a delete never fails on it.
- backend.stop goes through resolveStopTargets, which accepts a backend
  name, a model name, or an exact modelID#replica key. Its payload field is
  named "backend" but is published with all three meanings: the admin UI
  sends a backend name, UnloadRemoteModel sends a model name, and the
  router's abandoned-load reap (#10948) sends an exact replica key.
  Narrowing it to backend names alone would strand the latter two.
  backend.delete stays strict - its identifier is unambiguously a backend.
- Gate the install fast path on processMatchesBackend so a slot held by a
  different backend is restarted rather than reused. Processes with no
  recorded name (pre-upgrade) are accepted, so rollout does not restart
  every running backend.
- stopBackendExact reports a real stop failure - the process still being
  alive afterwards, which is precisely what finishBackendStop already
  detects to keep the entry and its port reserved - and backend.delete no
  longer replies success when it knew about a process and could not kill it.
  "No process was running" stays a success but is logged, so the orphan case
  is visible rather than silent.

2. /backend/shutdown reported a running model as missing
---------------------------------------------------------
ModelLoader.deleteProcess short-circuits on a miss in this replica's
in-memory store. In distributed mode the authoritative record of "is this
model loaded" is the shared node registry: a frontend replica that never
served the model itself (load balancer picked a peer, or the replica
restarted) has no local entry. The remote unload path that pkg/model
documents ("when ShutdownModel is called for a model with no local process,
UnloadRemoteModel is called") sat behind that short-circuit, unreachable in
exactly the case it exists for. #10865 reworked this function but kept the
short-circuit at the top, so the gap survived that refactor.

- deleteProcess consults the remote unloader on a local-store miss, via a
  shared unloadRemote helper so this branch and the existing
  no-local-process branch both prefer #10865's RemoteModelContextUnloader,
  preserving force propagation across the distributed boundary.
- UnloadRemoteModelContext reports ErrRemoteModelNotLoaded when no node has
  the model; it previously returned nil, making a no-op stop
  indistinguishable from a real one. The converse case (nodes have it, none
  could be stopped) already errors since #10865 joined the per-node
  failures, so that half of the original fix was dropped as redundant.
- Only when the model is absent locally AND cluster-wide does the endpoint
  report not-found, now 404 naming both scopes rather than a bare 500.
- modelNotFoundErr becomes the exported ErrModelNotFound so the HTTP layer
  can map it without string matching; watchdog's identity comparison becomes
  errors.Is.

3. Coverage for the bounded Free() that #10865 shipped untested
----------------------------------------------------------------
The original branch also bounded the pre-stop Free(), but #10865 landed that
fix first (workerBackendFreeTimeout, applied in both stopBackendExact and
handleModelUnload). That production change is therefore DROPPED here as
superseded - master's version is strictly better, since it also releases the
supervisor mutex across the call and keeps the port reserved until
termination completes.

What #10865 did not ship is a test, and the bound is load-bearing: the
router-side reap in #10948 sends backend.stop for an abandoned load, and
against a wedged backend an unbounded Free would swallow that stop before it
reached the process. Nothing failed if the bound regressed.

The spec stands up a real gRPC backend server whose Free handler never
returns - what a Python backend looks like when its single worker thread
(PYTHON_GRPC_MAX_WORKERS=1 on 37 backends) is occupied by a stuck LoadModel.
A stub socket is not sufficient and was tried first: without a completed
HTTP/2 handshake, gRPC's own ~20s connect timeout ends the call, so that
version passed against the very bug it targets. With the connection READY,
only the caller's deadline can end it, so the spec hangs to its 60s limit if
the timeout is removed and passes with it.

Its fixture process is deliberately never started. go-processmanager v0.1.1
writes Process.pid from readPID() without synchronization, so a live process
races its own monitor goroutine under -race - reproducible with a bare
Run()+Stop() and unrelated to this spec. Since
scripts/model-lifecycle-conformance.sh runs this package with -race and is
fail-closed, starting one would turn that gate red on an upstream defect. An
unstarted process still proves the point: the stop is reached and the slot
released, which is exactly what an unbounded Free prevents.

Verified: make lint (new-from-merge-base origin/master) reports 0 issues;
scripts/model-lifecycle-conformance.sh passes all three stages including the
FizzBee liveness check (1458 states, IsLive: true).

Assisted-by: Claude:claude-opus-4-8 golangci-lint
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

* fix(distributed): keep remote unload idempotent, ask presence separately

2035a4d25 made UnloadRemoteModel return ErrRemoteModelNotLoaded when no node
holds the model, so ShutdownModel could answer 404 instead of a misleading
500. That narrowed a shared adapter contract to serve one caller and broke
the documented idempotent-unload guarantee, which CI caught on PR #10956:

  [FAIL] Node Backend Lifecycle (NATS-driven) > NATS backend.stop events
         should be no-op for models not on any node [Distributed]
         Expected success, but got: model not loaded on any node

The spec name states the contract outright. The matching unit assertion was
updated in that commit; this e2e one was missed because it lives under
tests/e2e/ with no build tags and does not run in package-scoped test runs.

Caller audit - who breaks when an idempotent unload becomes an error:

- pkg/model/watchdog.go:902 (LRU memory reclaimer) is the serious one. It
  untracks a model ONLY when shutdown returns nil or ErrModelNotFound. A new
  error type means the model is never untracked, so the reclaimer keeps
  re-selecting the same entry and never reclaims - a live wedge whenever a
  local store entry outlives the remote model.
- core/services/galleryop/managers_local.go:43 (DeleteModel) would warn on
  every deletion of an already-unloaded model.
- core/services/modeladmin/{state,config,remote_sync}.go stop instances
  best-effort against models that are frequently not loaded.
- deleteProcess itself: the no-local-process branch returns the unload result
  directly, so a stale local entry for a model no longer on any node turned a
  previously-successful cleanup into a failure.

Only ShutdownModel wants the distinction, and only on the local-store-miss
path. So the distinction moves to the caller instead of the contract:

- UnloadRemoteModel/UnloadRemoteModelContext return nil again when no node
  has the model, and ErrRemoteModelNotLoaded is removed.
- New optional RemoteModelPresenceChecker (HasRemoteModel) answers the
  question directly. deleteProcess consults it BEFORE unloading, because an
  idempotent unload cannot report afterwards whether anything was stopped.
  Absent locally AND cluster-wide is the only case that reports 404.
- A failed registry lookup is surfaced rather than reported as absence: an
  unreachable registry is not evidence a model is gone, and answering a
  confident 404 off a failed lookup is how an operator gets told a running
  model does not exist.
- Unloaders that predate the extension keep working - deleteProcess attempts
  the unload rather than refusing it - and compile-time assertions in the
  nodes package now pin all three optional interfaces, since both are
  consumed by runtime type assertion where drift degrades behavior silently
  instead of failing the build.

The contract is now pinned at both levels that disagreed, each spec pointing
at the other: "with no nodes returns nil" in unloader_test.go and "should be
no-op for models not on any node" in node_lifecycle_test.go.

Verified: full distributed e2e suite 233 passed / 0 failed (the suite that
failed 232/1 in CI); pkg/model and core/services/nodes green; make lint
new-from-merge-base reports 0 issues.

Assisted-by: Claude:claude-opus-4-8 golangci-lint
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

* fix(distributed): drop replica rows when a worker stops a backend

A worker returns a stopped backend's gRPC port to its allocator as soon as
the process is confirmed dead, and hands it to the next backend that starts.
The controller's NodeModel row for the old address survives, and both
SmartRouter.probeHealth and the HealthMonitor per-model probe verify
liveness, not identity, so once an unrelated backend binds the recycled port
the stale row passes every check and the request is served by the wrong
backend instead of failing.

backend.delete is newly able to trigger this: before #10956 a delete never
actually stopped a process, so it never recycled a port. backend.upgrade has
the identical gap and always did — upgradeBackend force-stops every process
using the binary and starts none back up, while
DistributedBackendManager.UpgradeBackend never removes rows. model.unload is
the one path that gets this right today: it calls RemoveAllNodeModelReplicas
straight after StopBackend.

Report the process keys the worker terminated on the delete and upgrade
replies, and drop the matching rows in RemoteUnloaderAdapter, which already
holds a ModelLocator with RemoveNodeModel. All three call sites funnel
through that adapter, so no new interface, DB migration, or proto change is
needed. A key is reported only once its process is confirmed gone, so the
list stays trustworthy on the partial-failure replies too.

Old workers never populate the new fields. ReportsStoppedProcesses tells
"stopped nothing" apart from "does not report", so an old worker's silence
falls back to the pre-existing probe-based staleness recovery instead of
being mistaken for a completed cleanup.

Quarantine released ports for a short window as an interlock covering the
NATS round-trip between the worker freeing the port and the controller
dropping the row. It is deliberately not derived from HealthCheckInterval:
that cadence is operator-tunable and the per-model reaper can be disabled
outright, so coupling a worker-local constant to it would be silently wrong
on some clusters. Eager row removal is the fix; the delay only closes the
handoff gap.

Identity verification in probeHealth was considered and rejected: Health and
Status carry no backend identity, so it needs a proto change plus an
implementation in 36 Python and 4 C++ Health servicers, it is fail-open for
any backend not yet rebuilt, and the probeCache short-circuit means it would
not even execute during the 30s window where the misroute happens.

Fixes #10952
Refs #10954, #10956

Assisted-by: Claude:claude-opus-4-8 golangci-lint
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

* chore(deps): bump go-processmanager, assert real backend termination

go-processmanager wrote Process.PID from readPID() with no synchronization
while its own monitor goroutine cleared the same field on exit, so a bare
Run()+Stop() tripped the race detector without any concurrent access from
the caller. LocalAI hit this on every backend stop.

Upstream fixed it in a94e2b7 by guarding PID with a mutex and adding
CurrentPID() as a race-safe accessor. The exported field was kept to avoid
a breaking change but is now deprecated: a direct read still races the
monitor. No tag carries the fix yet, so pin the pseudo-version.

GetGRPCPID reads through CurrentPID() instead of the field. The accessor
returns the same string under an RLock, so the empty-PID and strconv error
paths are unchanged; it is the only direct field read in the tree.

With the race gone, the Free-timeout spec no longer has to leave its
fixture process unstarted. It now runs a real child and asserts the child
genuinely exits, which is exactly what the earlier workaround gave up: the
spec could show the stop was reached and the slot released, but not that
SIGTERM ever landed. Termination is observed through Done(), which closes
only once the library has waited on the child. The pidfile-based liveness
helpers cannot serve here, because Stop() deletes the pidfile while
releasing the handle and so reports "not alive" even if no signal was sent.

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash]

---------

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
2026-07-20 01:09:22 +02:00

927 lines
30 KiB
Go

package model
import (
"errors"
"slices"
"sync"
"time"
"github.com/mudler/LocalAI/pkg/xsysinfo"
process "github.com/mudler/go-processmanager"
"github.com/mudler/xlog"
)
// WatchDog tracks all the requests from GRPC clients.
// All GRPC Clients created by ModelLoader should have an associated injected
// watchdog that will keep track of the state of each backend (busy or not)
// and for how much time it has been busy.
// If a backend is busy for too long, the watchdog will kill the process and
// force a reload of the model.
// The watchdog also supports LRU (Least Recently Used) eviction when a maximum
// number of active backends is configured.
// The watchdog also supports memory threshold monitoring - when memory usage
// (GPU VRAM if available, otherwise system RAM) exceeds the threshold,
// it will evict backends using the LRU strategy.
// The watchdog runs as a separate go routine,
// and the GRPC client talks to it via a channel to send status updates
type WatchDog struct {
sync.Mutex
busyTime map[string]time.Time
inFlight map[string]int
requestStarts map[string]map[uint64]time.Time
legacyRequests map[string][]uint64
nextRequestID uint64
idleTime map[string]time.Time
lastUsed map[string]time.Time // LRU tracking: when each model was last used
timeout, idletimeout time.Duration
addressMap map[string]*process.Process
addressModelMap map[string]string
pm ProcessManager
stop chan bool
done chan bool // Signals when Run() has completely shut down
busyCheck, idleCheck bool
lruLimit int // Maximum number of active backends (0 = unlimited)
// Memory reclaimer settings (works with GPU if available, otherwise RAM)
memoryReclaimerEnabled bool // Enable memory threshold monitoring
memoryReclaimerThreshold float64 // Threshold 0.0-1.0 (e.g., 0.95 = 95%)
watchdogInterval time.Duration
// Eviction settings
forceEvictionWhenBusy bool // Force eviction even when models have active API calls (default: false for safety)
// Size-aware eviction: sort candidates by model file size (largest first) to maximize freed memory.
// When enabled, bigger models are evicted before smaller ones regardless of recency.
sizeAwareEviction bool
modelSizes map[string]int64 // modelID → file size in bytes
// Pinned models are excluded from idle, LRU, and memory-pressure eviction
pinnedModels map[string]bool
// modelGroups maps a model name to its declared concurrency groups.
// Two loaded models that share at least one group cannot coexist on this
// node — see EnforceGroupExclusivity.
modelGroups map[string][]string
}
type ProcessManager interface {
ShutdownModel(modelName string) error
// ShutdownModelForce stops the backend without waiting for an in-flight
// gRPC call to finish. Used when the watchdog evicts a backend that is
// stuck busy: the graceful ShutdownModel intentionally waits for active
// calls until its bounded deadline.
ShutdownModelForce(modelName string) error
}
// NewWatchDog creates a new WatchDog with the provided options.
// Example usage:
//
// wd := NewWatchDog(
// WithProcessManager(pm),
// WithBusyTimeout(5*time.Minute),
// WithIdleTimeout(15*time.Minute),
// WithBusyCheck(true),
// WithIdleCheck(true),
// WithLRULimit(3),
// WithMemoryReclaimer(true, 0.95),
// )
func NewWatchDog(opts ...WatchDogOption) *WatchDog {
o := NewWatchDogOptions(opts...)
return &WatchDog{
timeout: o.busyTimeout,
idletimeout: o.idleTimeout,
pm: o.processManager,
busyTime: make(map[string]time.Time),
inFlight: make(map[string]int),
requestStarts: make(map[string]map[uint64]time.Time),
legacyRequests: make(map[string][]uint64),
idleTime: make(map[string]time.Time),
lastUsed: make(map[string]time.Time),
addressMap: make(map[string]*process.Process),
busyCheck: o.busyCheck,
idleCheck: o.idleCheck,
lruLimit: o.lruLimit,
addressModelMap: make(map[string]string),
pinnedModels: make(map[string]bool),
modelGroups: make(map[string][]string),
stop: make(chan bool, 1),
done: make(chan bool, 1),
memoryReclaimerEnabled: o.memoryReclaimerEnabled,
memoryReclaimerThreshold: o.memoryReclaimerThreshold,
watchdogInterval: o.watchdogInterval,
forceEvictionWhenBusy: o.forceEvictionWhenBusy,
sizeAwareEviction: o.sizeAwareEviction,
modelSizes: make(map[string]int64),
}
}
// SetLRULimit updates the LRU limit dynamically
func (wd *WatchDog) SetLRULimit(limit int) {
wd.Lock()
defer wd.Unlock()
wd.lruLimit = limit
}
// GetLRULimit returns the current LRU limit
func (wd *WatchDog) GetLRULimit() int {
wd.Lock()
defer wd.Unlock()
return wd.lruLimit
}
// SetMemoryReclaimer updates the memory reclaimer settings dynamically
func (wd *WatchDog) SetMemoryReclaimer(enabled bool, threshold float64) {
wd.Lock()
defer wd.Unlock()
wd.memoryReclaimerEnabled = enabled
wd.memoryReclaimerThreshold = threshold
}
// GetMemoryReclaimerSettings returns the current memory reclaimer settings
func (wd *WatchDog) GetMemoryReclaimerSettings() (enabled bool, threshold float64) {
wd.Lock()
defer wd.Unlock()
return wd.memoryReclaimerEnabled, wd.memoryReclaimerThreshold
}
// SetForceEvictionWhenBusy updates the force eviction when busy setting dynamically
func (wd *WatchDog) SetForceEvictionWhenBusy(force bool) {
wd.Lock()
defer wd.Unlock()
wd.forceEvictionWhenBusy = force
}
// RegisterModelSize records the on-disk file size for a model.
// This is used by size-aware eviction to prefer evicting larger models first.
// Call this after a model has been successfully loaded.
func (wd *WatchDog) RegisterModelSize(modelID string, bytes int64) {
wd.Lock()
defer wd.Unlock()
wd.modelSizes[modelID] = bytes
}
// SetSizeAwareEviction enables or disables size-aware eviction ordering.
// When enabled, eviction candidates are sorted by file size (largest first)
// rather than by recency, maximizing freed memory per eviction.
func (wd *WatchDog) SetSizeAwareEviction(enabled bool) {
wd.Lock()
defer wd.Unlock()
wd.sizeAwareEviction = enabled
}
// GetSizeAwareEviction returns whether size-aware eviction is enabled.
func (wd *WatchDog) GetSizeAwareEviction() bool {
wd.Lock()
defer wd.Unlock()
return wd.sizeAwareEviction
}
// SetPinnedModels replaces the set of pinned model names.
// Pinned models are excluded from idle, LRU, and memory-pressure eviction.
func (wd *WatchDog) SetPinnedModels(models []string) {
wd.Lock()
defer wd.Unlock()
wd.pinnedModels = make(map[string]bool, len(models))
for _, m := range models {
wd.pinnedModels[m] = true
}
}
// IsModelPinned returns true if the given model name is pinned
func (wd *WatchDog) IsModelPinned(modelName string) bool {
wd.Lock()
defer wd.Unlock()
return wd.pinnedModels[modelName]
}
// ReplaceModelGroups replaces the per-model concurrency-group registry. The
// supplied map is copied; callers may mutate it after the call. Passing an
// empty or nil map clears all entries.
func (wd *WatchDog) ReplaceModelGroups(groups map[string][]string) {
wd.Lock()
defer wd.Unlock()
wd.modelGroups = make(map[string][]string, len(groups))
for name, gs := range groups {
if len(gs) == 0 {
continue
}
wd.modelGroups[name] = slices.Clone(gs)
}
}
// GetModelGroups returns a copy of the concurrency groups configured for
// the given model, or nil if the model has no groups. The result may be
// freely mutated by the caller.
func (wd *WatchDog) GetModelGroups(modelName string) []string {
wd.Lock()
defer wd.Unlock()
gs, ok := wd.modelGroups[modelName]
if !ok || len(gs) == 0 {
return nil
}
return slices.Clone(gs)
}
func (wd *WatchDog) Shutdown() {
wd.Lock()
defer wd.Unlock()
xlog.Info("[WatchDog] Shutting down watchdog")
wd.stop <- true
}
// WaitDone blocks until the watchdog's Run() goroutine has completely shut down.
// This should be called after Shutdown() to ensure the watchdog is fully stopped.
func (wd *WatchDog) WaitDone() {
<-wd.done
}
func (wd *WatchDog) AddAddressModelMap(address string, model string) {
wd.Lock()
defer wd.Unlock()
wd.addressModelMap[address] = model
}
func (wd *WatchDog) Add(address string, p *process.Process) {
wd.Lock()
defer wd.Unlock()
wd.addressMap[address] = p
}
// TrackRequest records one request and returns an idempotent completion
// callback. The per-request identity lets the busy watchdog follow the oldest
// request that is still active instead of treating uninterrupted parallel
// traffic as one indefinitely old request.
func (wd *WatchDog) TrackRequest(address string) func() {
wd.Lock()
requestID := wd.beginRequestLocked(address, false)
wd.Unlock()
var once sync.Once
return func() {
once.Do(func() {
wd.Lock()
defer wd.Unlock()
wd.finishRequestLocked(address, requestID)
})
}
}
// Mark and UnMark retain the legacy public API for non-gRPC callers. Callers
// that can keep the completion callback should use TrackRequest so overlapping
// requests can complete in any order without losing their individual ages.
func (wd *WatchDog) Mark(address string) {
wd.Lock()
defer wd.Unlock()
wd.beginRequestLocked(address, true)
}
func (wd *WatchDog) UnMark(address string) {
wd.Lock()
defer wd.Unlock()
requests := wd.legacyRequests[address]
if len(requests) == 0 {
return
}
requestID := requests[len(requests)-1]
requests = requests[:len(requests)-1]
if len(requests) == 0 {
delete(wd.legacyRequests, address)
} else {
wd.legacyRequests[address] = requests
}
wd.finishRequestLocked(address, requestID)
}
func (wd *WatchDog) beginRequestLocked(address string, legacy bool) uint64 {
now := time.Now()
wd.nextRequestID++
requestID := wd.nextRequestID
starts := wd.requestStarts[address]
if starts == nil {
starts = make(map[uint64]time.Time)
wd.requestStarts[address] = starts
}
starts[requestID] = now
if legacy {
wd.legacyRequests[address] = append(wd.legacyRequests[address], requestID)
}
wd.inFlight[address] = len(starts)
if len(starts) == 1 {
wd.busyTime[address] = now
}
wd.lastUsed[address] = now
delete(wd.idleTime, address)
return requestID
}
func (wd *WatchDog) finishRequestLocked(address string, requestID uint64) {
starts := wd.requestStarts[address]
if _, exists := starts[requestID]; !exists {
return
}
delete(starts, requestID)
now := time.Now()
wd.lastUsed[address] = now // Update LRU tracking
if len(starts) > 0 {
wd.inFlight[address] = len(starts)
var oldest time.Time
for _, started := range starts {
if oldest.IsZero() || started.Before(oldest) {
oldest = started
}
}
wd.busyTime[address] = oldest
return
}
delete(wd.requestStarts, address)
delete(wd.inFlight, address)
delete(wd.busyTime, address)
wd.idleTime[address] = now
}
// UpdateLastUsed updates the last used time for a model address (for LRU tracking)
// This should be called when a model is accessed (e.g., when checking if loaded)
func (wd *WatchDog) UpdateLastUsed(address string) {
wd.Lock()
defer wd.Unlock()
wd.lastUsed[address] = time.Now()
}
// GetLoadedModelCount returns the number of currently loaded models tracked by the watchdog
func (wd *WatchDog) GetLoadedModelCount() int {
wd.Lock()
defer wd.Unlock()
return len(wd.addressModelMap)
}
// WatchDogState holds the current state of models tracked by the watchdog
type WatchDogState struct {
AddressModelMap map[string]string
BusyTime map[string]time.Time
InFlight map[string]int
IdleTime map[string]time.Time
LastUsed map[string]time.Time
AddressMap map[string]*process.Process
}
// GetState returns the current state of models tracked by the watchdog
// This can be used to restore state when creating a new watchdog
func (wd *WatchDog) GetState() WatchDogState {
wd.Lock()
defer wd.Unlock()
// Create copies to avoid race conditions
addressModelMap := make(map[string]string, len(wd.addressModelMap))
for k, v := range wd.addressModelMap {
addressModelMap[k] = v
}
busyTime := make(map[string]time.Time, len(wd.busyTime))
for k, v := range wd.busyTime {
busyTime[k] = v
}
inFlight := make(map[string]int, len(wd.inFlight))
for k, v := range wd.inFlight {
inFlight[k] = v
}
idleTime := make(map[string]time.Time, len(wd.idleTime))
for k, v := range wd.idleTime {
idleTime[k] = v
}
lastUsed := make(map[string]time.Time, len(wd.lastUsed))
for k, v := range wd.lastUsed {
lastUsed[k] = v
}
addressMap := make(map[string]*process.Process, len(wd.addressMap))
for k, v := range wd.addressMap {
addressMap[k] = v
}
return WatchDogState{
AddressModelMap: addressModelMap,
BusyTime: busyTime,
InFlight: inFlight,
IdleTime: idleTime,
LastUsed: lastUsed,
AddressMap: addressMap,
}
}
// RestoreState restores the model state from a previous watchdog
// This should be called after the new watchdog is created but before Run() is started
func (wd *WatchDog) RestoreState(state WatchDogState) {
wd.Lock()
defer wd.Unlock()
wd.addressModelMap = state.AddressModelMap
wd.busyTime = state.BusyTime
wd.inFlight = state.InFlight
if wd.inFlight == nil {
wd.inFlight = make(map[string]int, len(wd.busyTime))
for address := range wd.busyTime {
wd.inFlight[address] = 1
}
}
wd.requestStarts = make(map[string]map[uint64]time.Time, len(wd.inFlight))
wd.legacyRequests = make(map[string][]uint64, len(wd.inFlight))
for address, count := range wd.inFlight {
started := wd.busyTime[address]
if started.IsZero() {
started = time.Now()
wd.busyTime[address] = started
}
requests := make(map[uint64]time.Time, count)
for range count {
wd.nextRequestID++
requestID := wd.nextRequestID
requests[requestID] = started
wd.legacyRequests[address] = append(wd.legacyRequests[address], requestID)
}
wd.requestStarts[address] = requests
}
wd.idleTime = state.IdleTime
wd.lastUsed = state.LastUsed
wd.addressMap = state.AddressMap
xlog.Info("[WatchDog] Restored model state", "modelCount", len(wd.addressModelMap))
}
// modelUsageInfo holds information about a model's usage for eviction sorting
type modelUsageInfo struct {
address string
model string
lastUsed time.Time
sizeBytes int64 // on-disk file size; 0 if unknown
}
// evictionTarget is a model selected for eviction, retaining whether it was busy
// at selection time. A busy target must be shut down via ShutdownModelForce:
// the graceful path waits for its in-flight gRPC call to finish, but a busy
// eviction only happens when that call is stuck (busy-killer) or the operator
// opted into forceEvictionWhenBusy — either way, waiting for the graceful
// deadline would defeat prompt eviction.
type evictionTarget struct {
model string
wasBusy bool
}
// EnforceLRULimitResult contains the result of LRU enforcement
type EnforceLRULimitResult struct {
EvictedCount int // Number of models successfully evicted
NeedMore bool // True if more evictions are needed but couldn't be done (e.g., all models are busy)
}
// EnforceLRULimit ensures we're under the LRU limit by evicting least recently used models.
// This should be called before loading a new model.
// pendingLoads is the number of models currently being loaded (to account for concurrent loads).
// Returns the result containing evicted count and whether more evictions are needed.
func (wd *WatchDog) EnforceLRULimit(pendingLoads int) EnforceLRULimitResult {
if wd.lruLimit <= 0 {
return EnforceLRULimitResult{EvictedCount: 0, NeedMore: false} // LRU disabled
}
wd.Lock()
currentCount := len(wd.addressModelMap)
// We need to evict enough to make room for the new model AND any pending loads
// Total after loading = currentCount + pendingLoads + 1 (the new one we're about to load)
// We need: currentCount + pendingLoads + 1 <= lruLimit
// So evict: currentCount + pendingLoads + 1 - lruLimit = currentCount - lruLimit + pendingLoads + 1
modelsToEvict := currentCount - wd.lruLimit + pendingLoads + 1
forceEvictionWhenBusy := wd.forceEvictionWhenBusy
if modelsToEvict <= 0 {
wd.Unlock()
return EnforceLRULimitResult{EvictedCount: 0, NeedMore: false}
}
sizeAwareEviction := wd.sizeAwareEviction
xlog.Debug("[WatchDog] LRU enforcement triggered", "current", currentCount, "pendingLoads", pendingLoads, "limit", wd.lruLimit, "toEvict", modelsToEvict, "sizeAware", sizeAwareEviction)
// Build a list of models to sort for eviction candidates
var models []modelUsageInfo
for address, model := range wd.addressModelMap {
lastUsed := wd.lastUsed[address]
if lastUsed.IsZero() {
lastUsed = time.Time{}
}
models = append(models, modelUsageInfo{
address: address,
model: model,
lastUsed: lastUsed,
sizeBytes: wd.modelSizes[model],
})
}
// Sort eviction candidates: largest-first when size-aware, oldest-first otherwise.
// Tiebreaker in size-aware mode: oldest last-used (LRU) to break ties between
// models of the same size.
if sizeAwareEviction {
slices.SortFunc(models, func(a, b modelUsageInfo) int {
if a.sizeBytes != b.sizeBytes {
return int(b.sizeBytes - a.sizeBytes) // largest first
}
return a.lastUsed.Compare(b.lastUsed) // oldest first as tiebreaker
})
} else {
slices.SortFunc(models, func(a, b modelUsageInfo) int {
return a.lastUsed.Compare(b.lastUsed)
})
}
// Collect models to evict (the oldest ones)
modelsToShutdown, skippedBusyCount := wd.collectEvictionsLocked(models, modelsToEvict, forceEvictionWhenBusy)
needMore := len(modelsToShutdown) < modelsToEvict && skippedBusyCount > 0
wd.Unlock()
// Now shutdown models without holding the watchdog lock to prevent
// deadlock. Busy targets go through the force path so the loader doesn't
// wait on their stuck in-flight call (which would re-deadlock here).
wd.shutdownEvicted(modelsToShutdown, "LRU eviction")
if needMore {
xlog.Warn("[WatchDog] LRU eviction incomplete", "evicted", len(modelsToShutdown), "needed", modelsToEvict, "skippedBusy", skippedBusyCount, "reason", "some models are busy with active API calls")
}
return EnforceLRULimitResult{
EvictedCount: len(modelsToShutdown),
NeedMore: needMore,
}
}
// collectEvictionsLocked walks `candidates` (already in eviction order) and
// untracks up to `maxToEvict` models that are eligible for eviction. Pinned
// models are always skipped; busy models are skipped unless `force` is true.
// Returns the evicted models (with their busy state at selection time) and
// the number skipped because they were busy. Must be called with wd.Lock()
// held.
func (wd *WatchDog) collectEvictionsLocked(candidates []modelUsageInfo, maxToEvict int, force bool) (evicted []evictionTarget, skippedBusy int) {
for i := 0; len(evicted) < maxToEvict && i < len(candidates); i++ {
m := candidates[i]
if wd.pinnedModels[m.model] {
xlog.Debug("[WatchDog] Skipping eviction for pinned model", "model", m.model)
continue
}
_, isBusy := wd.busyTime[m.address]
if isBusy && !force {
xlog.Warn("[WatchDog] Skipping eviction for busy model", "model", m.model, "reason", "model has active API calls")
skippedBusy++
continue
}
xlog.Info("[WatchDog] evicting model", "model", m.model, "busy", isBusy)
evicted = append(evicted, evictionTarget{model: m.model, wasBusy: isBusy})
wd.untrack(m.address)
}
return evicted, skippedBusy
}
// shutdownEvicted shuts down each evicted model, using the force path for any
// that were busy at selection time so a stuck in-flight call does not consume
// the graceful shutdown deadline.
func (wd *WatchDog) shutdownEvicted(targets []evictionTarget, label string) {
for _, t := range targets {
var err error
if t.wasBusy {
err = wd.pm.ShutdownModelForce(t.model)
} else {
err = wd.pm.ShutdownModel(t.model)
}
if err != nil {
xlog.Error("[WatchDog] error shutting down model during "+label, "error", err, "model", t.model, "busy", t.wasBusy)
}
xlog.Debug("[WatchDog] "+label+" complete", "model", t.model, "busy", t.wasBusy)
}
}
// EnforceGroupExclusivity evicts every loaded model that shares at least one
// concurrency group with the requested model. The pinned/busy/retry semantics
// match EnforceLRULimit so the loader's retry loop can stay generic.
func (wd *WatchDog) EnforceGroupExclusivity(requestedModel string) EnforceLRULimitResult {
wd.Lock()
requestedGroups := wd.modelGroups[requestedModel]
if len(requestedGroups) == 0 {
wd.Unlock()
return EnforceLRULimitResult{}
}
forceEvictionWhenBusy := wd.forceEvictionWhenBusy
// Build the conflict candidate list: every loaded model whose groups
// overlap with requestedGroups. Order doesn't affect correctness, but
// sort by lastUsed (oldest first) so logs and behaviour are deterministic.
var conflicts []modelUsageInfo
for address, name := range wd.addressModelMap {
if name == requestedModel {
continue
}
if !groupsOverlap(requestedGroups, wd.modelGroups[name]) {
continue
}
conflicts = append(conflicts, modelUsageInfo{
address: address,
model: name,
lastUsed: wd.lastUsed[address],
})
}
if len(conflicts) == 0 {
wd.Unlock()
return EnforceLRULimitResult{}
}
slices.SortFunc(conflicts, func(a, b modelUsageInfo) int {
return a.lastUsed.Compare(b.lastUsed)
})
xlog.Debug("[WatchDog] Group exclusivity triggered", "requested", requestedModel, "groups", requestedGroups, "conflicts", len(conflicts))
modelsToShutdown, skippedBusyCount := wd.collectEvictionsLocked(conflicts, len(conflicts), forceEvictionWhenBusy)
// For groups any unresolved conflict matters — busy *or* pinned. The loader
// retries on NeedMore; pinned cases will eventually time out and the load
// proceeds with a visible warning, which is the right signal for what is a
// configuration mismatch.
needMore := len(modelsToShutdown) < len(conflicts)
wd.Unlock()
wd.shutdownEvicted(modelsToShutdown, "group eviction")
if needMore {
xlog.Warn("[WatchDog] Group eviction incomplete", "requested", requestedModel, "evicted", len(modelsToShutdown), "needed", len(conflicts), "skippedBusy", skippedBusyCount, "reason", "some conflicts are busy or pinned")
}
return EnforceLRULimitResult{
EvictedCount: len(modelsToShutdown),
NeedMore: needMore,
}
}
// groupsOverlap reports whether the two group lists share any name.
func groupsOverlap(a, b []string) bool {
if len(a) == 0 || len(b) == 0 {
return false
}
for _, x := range a {
if slices.Contains(b, x) {
return true
}
}
return false
}
func (wd *WatchDog) Run() {
xlog.Info("[WatchDog] starting watchdog")
for {
select {
case <-wd.stop:
xlog.Info("[WatchDog] Stopping watchdog")
wd.done <- true
return
case <-time.After(wd.watchdogInterval):
// Check if any monitoring is enabled
wd.Lock()
busyCheck := wd.busyCheck
idleCheck := wd.idleCheck
memoryCheck := wd.memoryReclaimerEnabled
wd.Unlock()
if !busyCheck && !idleCheck && !memoryCheck {
xlog.Info("[WatchDog] No checks enabled, stopping watchdog")
wd.done <- true
return
}
if busyCheck {
wd.checkBusy()
}
if idleCheck {
wd.checkIdle()
}
if memoryCheck {
wd.checkMemory()
}
}
}
}
func (wd *WatchDog) checkIdle() {
wd.Lock()
xlog.Debug("[WatchDog] Watchdog checks for idle connections")
// Collect models to shutdown while holding the lock
var modelsToShutdown []string
for address, t := range wd.idleTime {
xlog.Debug("[WatchDog] idle connection", "address", address)
if time.Since(t) > wd.idletimeout {
model, ok := wd.addressModelMap[address]
if ok {
if wd.pinnedModels[model] {
xlog.Debug("[WatchDog] Skipping idle eviction for pinned model", "model", model)
continue
}
xlog.Warn("[WatchDog] Address is idle for too long, killing it", "address", address)
modelsToShutdown = append(modelsToShutdown, model)
} else {
xlog.Warn("[WatchDog] Address unresolvable", "address", address)
}
wd.untrack(address)
}
}
wd.Unlock()
// Now shutdown models without holding the watchdog lock to prevent deadlock
for _, model := range modelsToShutdown {
if err := wd.pm.ShutdownModel(model); err != nil {
xlog.Error("[watchdog] error shutting down model", "error", err, "model", model)
}
xlog.Debug("[WatchDog] model shut down", "model", model)
}
}
func (wd *WatchDog) checkBusy() {
wd.Lock()
xlog.Debug("[WatchDog] Watchdog checks for busy connections")
// Collect models to shutdown while holding the lock
var modelsToShutdown []string
for address, t := range wd.busyTime {
xlog.Debug("[WatchDog] active connection", "address", address)
if time.Since(t) > wd.timeout {
model, ok := wd.addressModelMap[address]
if ok {
xlog.Warn("[WatchDog] Model is busy for too long, killing it", "model", model)
modelsToShutdown = append(modelsToShutdown, model)
} else {
xlog.Warn("[WatchDog] Address unresolvable", "address", address)
}
wd.untrack(address)
}
}
wd.Unlock()
// The busy-killer targets backends whose in-flight gRPC call has been
// stuck past the busy timeout. Use the force path so the loader stops
// the process FIRST (dropping the stuck call's gRPC connection) instead
// of waiting for the graceful shutdown deadline.
for _, model := range modelsToShutdown {
if err := wd.pm.ShutdownModelForce(model); err != nil {
xlog.Error("[watchdog] error shutting down model", "error", err, "model", model)
}
xlog.Debug("[WatchDog] busy model shut down", "model", model)
}
}
// checkMemory monitors memory usage (GPU VRAM if available, otherwise RAM) and evicts backends when usage exceeds threshold
func (wd *WatchDog) checkMemory() {
wd.Lock()
threshold := wd.memoryReclaimerThreshold
enabled := wd.memoryReclaimerEnabled
modelCount := len(wd.addressModelMap)
wd.Unlock()
if !enabled || threshold <= 0 || modelCount == 0 {
return
}
// Get current memory usage (GPU if available, otherwise RAM)
aggregate := xsysinfo.GetResourceAggregateInfo()
if aggregate.TotalMemory == 0 {
xlog.Debug("[WatchDog] No memory information available for memory reclaimer")
return
}
// Convert threshold from 0.0-1.0 to percentage
thresholdPercent := threshold * 100
memoryType := "GPU"
if aggregate.GPUCount == 0 {
memoryType = "RAM"
}
//xlog.Debug("[WatchDog] Memory check", "type", memoryType, "usage_percent", aggregate.UsagePercent, "threshold_percent", thresholdPercent, "loaded_models", modelCount)
// Check if usage exceeds threshold
if aggregate.UsagePercent > thresholdPercent {
xlog.Warn("[WatchDog] Memory usage exceeds threshold, evicting LRU backend", "type", memoryType, "usage_percent", aggregate.UsagePercent, "threshold_percent", thresholdPercent)
// Evict the least recently used model
wd.evictLRUModel()
}
}
// evictLRUModel evicts the least recently used model
func (wd *WatchDog) evictLRUModel() {
wd.Lock()
if len(wd.addressModelMap) == 0 {
wd.Unlock()
return
}
forceEvictionWhenBusy := wd.forceEvictionWhenBusy
sizeAwareEviction := wd.sizeAwareEviction
// Build a list of models to sort for eviction candidates
var models []modelUsageInfo
for address, model := range wd.addressModelMap {
lastUsed := wd.lastUsed[address]
if lastUsed.IsZero() {
lastUsed = time.Time{}
}
models = append(models, modelUsageInfo{
address: address,
model: model,
lastUsed: lastUsed,
sizeBytes: wd.modelSizes[model],
})
}
if len(models) == 0 {
wd.Unlock()
return
}
// Sort eviction candidates: largest-first when size-aware, oldest-first otherwise.
if sizeAwareEviction {
slices.SortFunc(models, func(a, b modelUsageInfo) int {
if a.sizeBytes != b.sizeBytes {
return int(b.sizeBytes - a.sizeBytes) // largest first
}
return a.lastUsed.Compare(b.lastUsed)
})
} else {
slices.SortFunc(models, func(a, b modelUsageInfo) int {
return a.lastUsed.Compare(b.lastUsed)
})
}
// Find the first non-busy, non-pinned model (or first non-pinned model if forceEvictionWhenBusy is true)
var lruModel *modelUsageInfo
for i := range len(models) {
m := models[i]
if wd.pinnedModels[m.model] {
xlog.Debug("[WatchDog] Skipping memory reclaimer eviction for pinned model", "model", m.model)
continue
}
_, isBusy := wd.busyTime[m.address]
if isBusy && !forceEvictionWhenBusy {
// Skip busy models when forceEvictionWhenBusy is false
xlog.Warn("[WatchDog] Skipping memory reclaimer eviction for busy model", "model", m.model, "reason", "model has active API calls")
continue
}
lruModel = &m
break
}
if lruModel == nil {
// All models are busy and forceEvictionWhenBusy is false
wd.Unlock()
xlog.Warn("[WatchDog] Memory reclaimer cannot evict: all models are busy with active API calls")
return
}
xlog.Info("[WatchDog] Memory reclaimer evicting LRU model", "model", lruModel.model, "lastUsed", lruModel.lastUsed)
// A busy target only gets here when forceEvictionWhenBusy is true, i.e.
// the operator accepted evicting models with active calls. Route it
// through the force path so the loader stops the process first instead
// of blocking on the stuck in-flight call while holding its mutex.
_, wasBusy := wd.busyTime[lruModel.address]
wd.Unlock()
// Shutdown the model
shutdown := wd.pm.ShutdownModel
if wasBusy {
shutdown = wd.pm.ShutdownModelForce
}
if err := shutdown(lruModel.model); err != nil && !errors.Is(err, ErrModelNotFound) {
xlog.Error("[WatchDog] error shutting down model during memory reclamation", "error", err, "model", lruModel.model)
} else {
// Untrack the model
wd.Lock()
wd.untrack(lruModel.address)
wd.Unlock()
xlog.Info("[WatchDog] Memory reclaimer eviction complete", "model", lruModel.model)
}
}
func (wd *WatchDog) untrack(address string) {
if modelID, ok := wd.addressModelMap[address]; ok {
delete(wd.modelSizes, modelID)
}
delete(wd.busyTime, address)
delete(wd.inFlight, address)
delete(wd.requestStarts, address)
delete(wd.legacyRequests, address)
delete(wd.idleTime, address)
delete(wd.lastUsed, address)
delete(wd.addressModelMap, address)
delete(wd.addressMap, address)
}