diff --git a/core/services/worker/config.go b/core/services/worker/config.go index 817f7f217de8..8057e69fe790 100644 --- a/core/services/worker/config.go +++ b/core/services/worker/config.go @@ -22,6 +22,13 @@ type Config struct { Addr string `env:"LOCALAI_ADDR" help:"Address where this worker is reachable (host:port). Port is base for gRPC backends, port-1 for HTTP." group:"server"` ServeAddr string `env:"LOCALAI_SERVE_ADDR" default:"0.0.0.0:50051" help:"(Advanced) gRPC base port bind address" group:"server" hidden:""` + // GRPCMaxPort bounds the dynamic gRPC port allocator at [basePort, this]. + // The width of that range is how many backend processes this worker can run + // concurrently; released ports also sit in a short quarantine before reuse, + // so a worker with heavy start/stop churn needs headroom above its true + // concurrency. 0 = up to 65535. + GRPCMaxPort int `env:"LOCALAI_GRPC_MAX_PORT" default:"0" help:"Highest port the worker may assign to a backend gRPC process. The range is [base port, this]; its width caps concurrent backends on this worker. 0 uses up to 65535." group:"server"` + BackendsPath string `env:"LOCALAI_BACKENDS_PATH,BACKENDS_PATH" type:"path" default:"${basepath}/backends" help:"Path containing backends" group:"server"` BackendsSystemPath string `env:"LOCALAI_BACKENDS_SYSTEM_PATH" type:"path" default:"/var/lib/local-ai/backends" help:"Path containing system backends" group:"server"` BackendGalleries string `env:"LOCALAI_BACKEND_GALLERIES,BACKEND_GALLERIES" help:"JSON list of backend galleries" group:"server" default:"${backends}"` diff --git a/core/services/worker/port_allocator_test.go b/core/services/worker/port_allocator_test.go new file mode 100644 index 000000000000..8677c840a84a --- /dev/null +++ b/core/services/worker/port_allocator_test.go @@ -0,0 +1,268 @@ +package worker + +import ( + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// mustAllocate fails the spec if allocation errors, so the ordinary "give me a +// port" specs below stay readable while still refusing to assert on a zero +// value that the allocator never intended to hand out. +func mustAllocate(s *backendSupervisor, key string) int { + port, err := s.allocatePort(key) + ExpectWithOffset(1, err).NotTo(HaveOccurred()) + return port +} + +var _ = Describe("Backend gRPC port allocator", func() { + Describe("range exhaustion", func() { + It("reports an explicit error instead of handing out an unbindable port", func() { + // The allocator used to increment without an upper bound. Past + // 65535 it kept returning larger integers, which cannot be bound, + // so the operator saw "backend won't start" with nothing pointing + // at the allocator. Exhaustion must be named as exhaustion. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50052, + portQuarantine: time.Hour, + } + + first := mustAllocate(s, "a#0") + second := mustAllocate(s, "b#0") + Expect([]int{first, second}).To(ConsistOf(50051, 50052)) + + _, err := s.allocatePort("c#0") + Expect(err).To(MatchError(ErrNoFreePort)) + // The operator has to learn which knob to turn from the message + // alone; a bare "no free port" sends them reading source. + Expect(err.Error()).To(ContainSubstring("50051")) + Expect(err.Error()).To(ContainSubstring("50052")) + Expect(err.Error()).To(ContainSubstring("LOCALAI_GRPC_MAX_PORT")) + }) + + It("does not hand out a port beyond the end of the range", func() { + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50053, + portQuarantine: time.Hour, + } + + for i := 0; i < 3; i++ { + port := mustAllocate(s, "model#0") + Expect(port).To(BeNumerically(">=", 50051)) + Expect(port).To(BeNumerically("<=", 50053)) + } + }) + }) + + Describe("unexpectedly dead processes", func() { + It("returns the dead process's port to the allocator", func() { + // The death path deleted the map entry without releasing the port, + // so every unexpected exit permanently consumed one port. A + // crash-looping backend leaks one per restart, which is the real + // route to exhausting the range — not the concurrent-peak growth + // the issue assumed. + bp := &backendProcess{port: 50051} + s := &backendSupervisor{ + processes: map[string]*backendProcess{"model#0": bp}, + minPort: 50051, + nextPort: 50052, + maxPort: 50060, + portQuarantine: time.Millisecond, + } + + s.reapDeadProcess("model#0", bp) + + Expect(s.processes).NotTo(HaveKey("model#0")) + // Quarantined, not lost: the port must come back. + Expect(quarantinedPortNumbers(s)).To(ContainElement(50051)) + time.Sleep(5 * time.Millisecond) + Expect(mustAllocate(s, "model#0")).To(Equal(50051)) + }) + + It("does not exhaust the range when one backend crash-loops", func() { + // Restarting the same key repeatedly must not walk the allocator up + // the range. Before the fix this consumed a fresh port per restart. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50053, + portQuarantine: time.Millisecond, + } + + for i := 0; i < 10; i++ { + port, err := s.allocatePort("crashy#0") + Expect(err).NotTo(HaveOccurred(), "restart %d exhausted the range", i) + bp := &backendProcess{port: port} + s.processes["crashy#0"] = bp + s.reapDeadProcess("crashy#0", bp) + time.Sleep(2 * time.Millisecond) + } + }) + }) + + Describe("effectiveMaxPort", func() { + It("uses the full range when unset", func() { + Expect((&Config{}).effectiveMaxPort(50051)).To(Equal(65535)) + }) + + It("honours an operator-set ceiling", func() { + Expect((&Config{GRPCMaxPort: 51000}).effectiveMaxPort(50051)).To(Equal(51000)) + }) + + It("clamps a ceiling above the highest TCP port", func() { + Expect((&Config{GRPCMaxPort: 70000}).effectiveMaxPort(50051)).To(Equal(65535)) + }) + + It("ignores a ceiling below the base port instead of wedging every start", func() { + // An inverted range would leave the allocator with nothing to hand + // out, so a typo here would take the whole worker down rather than + // degrade it. + Expect((&Config{GRPCMaxPort: 40000}).effectiveMaxPort(50051)).To(Equal(65535)) + }) + }) + + Describe("per-key port affinity", func() { + It("never hands one key's released port to a different key", func() { + // This is what makes a stale NodeModel row harmless. Process keys + // (modelID#replica) and NodeModel rows (nodeID, modelName, + // replicaIndex) are isomorphic, so if a port can only ever be + // re-bound by the key that last held it, the only row that can name + // that port is that key's own row — which its re-registration + // overwrites. Misrouting to a *different* model becomes impossible + // by construction rather than by racing a quarantine timer. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50060, + portQuarantine: time.Millisecond, + } + + portA := mustAllocate(s, "alpha#0") + s.releasePortForKey("alpha#0", portA) + time.Sleep(5 * time.Millisecond) + + // beta must NOT be given alpha's port while unused ports remain. + portB := mustAllocate(s, "beta#0") + Expect(portB).NotTo(Equal(portA)) + + // alpha coming back gets its own port again. + Expect(mustAllocate(s, "alpha#0")).To(Equal(portA)) + }) + + It("reuses an unowned port before growing the range", func() { + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50060, + portQuarantine: time.Millisecond, + } + + // A port released without an owning key (legacy/unknown) is free + // for anyone: no row can be attributed to it. + s.releasePort(50051) + time.Sleep(5 * time.Millisecond) + + Expect(mustAllocate(s, "gamma#0")).To(Equal(50051)) + }) + + It("steals another key's port rather than failing an exhausted range", func() { + // Affinity is a preference, not a reservation. Refusing to start a + // backend because a long-gone key still owns the last port trades a + // vanishingly rare misroute for a guaranteed outage. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50051, + portQuarantine: time.Millisecond, + } + + portA := mustAllocate(s, "alpha#0") + s.releasePortForKey("alpha#0", portA) + time.Sleep(5 * time.Millisecond) + + port, err := s.allocatePort("beta#0") + Expect(err).NotTo(HaveOccurred()) + Expect(port).To(Equal(portA)) + }) + + It("stops sequestering the port once the stale-row window has passed", func() { + // Affinity exists to outlive any controller row that could still + // name the port. Past that, holding it for a key that may never + // come back is pure cost: every distinct model the worker has ever + // served would consume a port permanently, so a long-lived worker + // would climb to the end of its range on distinct-key count rather + // than on concurrency, then steal on every subsequent allocation + // while advising the operator to raise a ceiling that is not the + // problem. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50060, + portQuarantine: time.Millisecond, + portAffinityWindow: 20 * time.Millisecond, + } + + portA := mustAllocate(s, "alpha#0") + s.releasePortForKey("alpha#0", portA) + time.Sleep(60 * time.Millisecond) + + // A brand-new key reuses it rather than growing the range. + Expect(mustAllocate(s, "beta#0")).To(Equal(portA)) + }) + + It("applies the default affinity window when none is configured", func() { + // The zero value must not degrade to "no affinity": every + // supervisor built outside the tests leaves the field unset, and + // zero would hand a just-released port straight to another model. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50060, + portQuarantine: time.Millisecond, + } + + portA := mustAllocate(s, "alpha#0") + s.releasePortForKey("alpha#0", portA) + time.Sleep(5 * time.Millisecond) + + Expect(mustAllocate(s, "beta#0")).NotTo(Equal(portA)) + }) + + It("keeps the affinity map bounded by the size of the port range", func() { + // The affinity map must not become the port leak's twin. Handing a + // port to a new key drops the previous owner's entry, so the map is + // injective over ports and can never hold more entries than the + // range has ports, however many distinct keys the worker sees. + s := &backendSupervisor{ + processes: map[string]*backendProcess{}, + minPort: 50051, + nextPort: 50051, + maxPort: 50052, + portQuarantine: time.Nanosecond, + } + + for i := 0; i < 50; i++ { + key := "model" + string(rune('a'+i%26)) + "#0" + port, err := s.allocatePort(key) + Expect(err).NotTo(HaveOccurred()) + s.releasePortForKey(key, port) + time.Sleep(time.Millisecond) + } + + Expect(len(s.portAffinity)).To(BeNumerically("<=", 2)) + }) + }) +}) diff --git a/core/services/worker/port_quarantine_test.go b/core/services/worker/port_quarantine_test.go index 3dcd25d2d5da..c5b48b69c3d0 100644 --- a/core/services/worker/port_quarantine_test.go +++ b/core/services/worker/port_quarantine_test.go @@ -35,12 +35,17 @@ var _ = Describe("Backend port quarantine", func() { Expect(s.finishBackendStop("model#0", bp, nil)).To(Succeed()) Expect(s.freePorts).To(BeEmpty()) - Expect(s.allocatePort()).NotTo(Equal(50051)) + Expect(mustAllocate(s, "next#0")).NotTo(Equal(50051)) }) - It("returns the port to the free pool once the quarantine expires", func() { - // The quarantine must not leak ports: nextPort only ever grows, so a - // port that never comes back is a permanent loss of allocator range. + It("returns the port to its own backend once the quarantine expires", func() { + // The quarantine must not leak ports: a port that never comes back + // is a permanent loss of allocator range. + // + // It comes back to the key that released it, not to a global pool — + // that is the affinity rule in allocatePort, which is what keeps a + // stale controller row for this key from ever resolving to a + // different model's backend. bp := &backendProcess{port: 50051} s := &backendSupervisor{ processes: map[string]*backendProcess{"model#0": bp}, @@ -51,7 +56,7 @@ var _ = Describe("Backend port quarantine", func() { Expect(s.finishBackendStop("model#0", bp, nil)).To(Succeed()) time.Sleep(5 * time.Millisecond) - Expect(s.allocatePort()).To(Equal(50051)) + Expect(mustAllocate(s, "model#0")).To(Equal(50051)) }) }) @@ -70,7 +75,7 @@ var _ = Describe("Backend port quarantine", func() { s.releaseBackendStart("model#0", bp) Expect(s.freePorts).To(BeEmpty()) - Expect(s.allocatePort()).To(Equal(50060)) + Expect(mustAllocate(s, "next#0")).To(Equal(50060)) }) }) @@ -85,8 +90,8 @@ var _ = Describe("Backend port quarantine", func() { s.releasePort(50051) time.Sleep(5 * time.Millisecond) - Expect(s.allocatePort()).To(Equal(50051)) - Expect(s.allocatePort()).To(Equal(50060)) + Expect(mustAllocate(s, "next#0")).To(Equal(50051)) + Expect(mustAllocate(s, "next#0")).To(Equal(50060)) Expect(s.nextPort).To(Equal(50061)) }) @@ -100,8 +105,8 @@ var _ = Describe("Backend port quarantine", func() { s.releasePort(50051) s.releasePort(50052) - Expect(s.allocatePort()).To(Equal(50060)) - Expect(s.allocatePort()).To(Equal(50061)) + Expect(mustAllocate(s, "next#0")).To(Equal(50060)) + Expect(mustAllocate(s, "next#0")).To(Equal(50061)) }) It("applies the default quarantine when none is configured", func() { @@ -114,7 +119,7 @@ var _ = Describe("Backend port quarantine", func() { s.releasePort(50051) - Expect(s.allocatePort()).To(Equal(50060)) + Expect(mustAllocate(s, "next#0")).To(Equal(50060)) }) }) }) diff --git a/core/services/worker/registration.go b/core/services/worker/registration.go index ad8b82e83c90..6b4ce19d6a1e 100644 --- a/core/services/worker/registration.go +++ b/core/services/worker/registration.go @@ -38,6 +38,32 @@ func (cfg *Config) effectiveBasePort() int { return 50051 } +// effectiveMaxPort returns the last port the gRPC backend allocator may hand +// out. The range is [basePort, maxPort]; its width is the number of backend +// processes this worker can run concurrently, minus whatever the port +// quarantine is holding at the time. +// +// An unset, non-positive, or out-of-order value falls back to 65535 — the +// historical (unbounded) behaviour clamped to something bindable. +// Misconfiguring this must not shrink the range to nothing and wedge every +// backend start on the worker. +func (cfg *Config) effectiveMaxPort(basePort int) int { + if cfg.GRPCMaxPort <= 0 { + return defaultMaxPort + } + if cfg.GRPCMaxPort > defaultMaxPort { + xlog.Warn("Configured gRPC max port is above the highest TCP port; clamping", + "configured", cfg.GRPCMaxPort, "max", defaultMaxPort) + return defaultMaxPort + } + if cfg.GRPCMaxPort < basePort { + xlog.Warn("Configured gRPC max port is below the base port; ignoring it and using the full range", + "configured", cfg.GRPCMaxPort, "basePort", basePort, "max", defaultMaxPort) + return defaultMaxPort + } + return cfg.GRPCMaxPort +} + // advertiseAddr returns the address the frontend should use to reach this node. func (cfg *Config) advertiseAddr() string { if cfg.AdvertiseAddr != "" { diff --git a/core/services/worker/supervisor.go b/core/services/worker/supervisor.go index e065518d5982..dab96e4ab235 100644 --- a/core/services/worker/supervisor.go +++ b/core/services/worker/supervisor.go @@ -2,6 +2,7 @@ package worker import ( "context" + "errors" "fmt" "maps" "os" @@ -91,9 +92,20 @@ type backendSupervisor struct { mu sync.Mutex processes map[string]*backendProcess // key: backend name - nextPort int // next available port for new backends + nextPort int // next unhanded-out port; grows within [minPort, maxPort] + minPort int // first port of the allocatable range + maxPort int // last port of the allocatable range (inclusive) freePorts []int // ports out of quarantine, reused before nextPort + // portAffinity remembers which process key last held each port, so a + // released port is offered back to that key before any other, and how long + // that claim outlives the release. See allocatePort for why this is a + // correctness property and not a nice-to-have. + // portAffinityWindow is overridden only by tests; zero means + // defaultPortAffinityWindow. + portAffinity map[string]portOwnership + portAffinityWindow time.Duration + // quarantinedPorts holds ports whose process has terminated but which are // not yet safe to re-bind, and portQuarantine is how long they wait. // Overridden only by tests; zero means defaultPortQuarantine. @@ -144,20 +156,215 @@ func (s *backendSupervisor) quarantineWindow() time.Duration { return s.portQuarantine } -// allocatePort returns a gRPC port for a new backend process, preferring a -// previously released port over growing the range. Callers must hold s.mu. -func (s *backendSupervisor) allocatePort() int { +// ErrNoFreePort reports that every port in this worker's configured gRPC range +// is either bound by a live backend or still in quarantine. +var ErrNoFreePort = errors.New("no free gRPC port in range") + +// defaultPortAffinityWindow is how long a released port stays reserved for the +// process key that last held it. +// +// It must comfortably outlive any controller NodeModel row that could still +// name the port, because that row is the whole reason affinity exists. The +// slowest such row is reaped by the per-model health check after +// perModelMissThreshold consecutive misses (~45s at the default cadence), and +// operators can widen that cadence, so this is set well above it rather than +// derived from it — the same reasoning as defaultPortQuarantine. +// +// It must also expire. Ownership held forever would make every distinct model +// a worker has ever served consume a port permanently, so the allocator would +// climb to the end of its range on distinct-key count rather than concurrency, +// then steal on every allocation while telling the operator to raise a ceiling +// that is not the constraint. +const defaultPortAffinityWindow = 5 * time.Minute + +// portOwnership records the key that last held a port and, once released, when +// that claim lapses. A zero `until` means the port is currently held: it cannot +// be reallocated to anyone, so the claim does not need to age. +type portOwnership struct { + port int + until time.Time +} + +// expired reports whether this claim has lapsed and the port may now be handed +// to an unrelated key. +func (o portOwnership) expired(now time.Time) bool { + return !o.until.IsZero() && now.After(o.until) +} + +// affinityWindow returns the configured affinity lifetime, defaulting when +// unset so a zero-value supervisor never degrades to no affinity at all. +func (s *backendSupervisor) affinityWindow() time.Duration { + if s.portAffinityWindow <= 0 { + return defaultPortAffinityWindow + } + return s.portAffinityWindow +} + +// defaultMaxPort bounds the allocator when the operator sets no explicit +// ceiling. The allocator used to increment without any bound at all, so past +// 65535 it handed out integers that cannot be bound and the operator saw an +// opaque "backend won't start" with nothing implicating the allocator. +const defaultMaxPort = 65535 + +// portBounds resolves the allocatable range, tolerating a zero-value +// supervisor. minPort defaults to wherever the range was seeded so that a +// supervisor built without an explicit floor never scans below its base port. +func (s *backendSupervisor) portBounds() (int, int) { + minPort := s.minPort + if minPort <= 0 { + minPort = s.nextPort + } + maxPort := s.maxPort + if maxPort <= 0 { + maxPort = defaultMaxPort + } + return minPort, maxPort +} + +// allocatePort returns a gRPC port for the process `key` will run under. +// +// Ports are handed back to the key that last held them before they are offered +// to anyone else. That preference is a correctness property, not a tidiness +// one. A process key is `modelID#replica` and a controller NodeModel row is +// keyed (nodeID, modelName, replicaIndex) — the two are isomorphic. So if a +// port can only ever be re-bound by the key that last held it, the only row +// that can name that port belongs to that same key, and that key's +// re-registration overwrites it. A stale row can therefore never resolve to a +// *different* model's backend, which is the silent misroute #10952 describes. +// The quarantine narrows the window for that misroute; affinity removes it. +// +// The claim lapses after defaultPortAffinityWindow, once no controller row can +// still name the port. Holding it forever would make every distinct model the +// worker has ever served consume a port permanently, so the allocator would +// climb to the end of its range on distinct-key count rather than on +// concurrency. Expiry keeps the guarantee for exactly as long as it buys +// anything. +// +// Nor is the preference a reservation: when the range would otherwise be +// exhausted a still-claimed port is stolen rather than failing the start. A +// rare misroute window is a better trade than a guaranteed outage, and by then +// the port is long out of quarantine. Because claims expire, reaching that +// branch means the worker is genuinely out of concurrent capacity, so the +// warning's advice to widen the range is the right advice. +// +// Callers must hold s.mu. +func (s *backendSupervisor) allocatePort(key string) (int, error) { s.sweepQuarantine() + s.sweepAffinity() + minPort, maxPort := s.portBounds() + + // 1. This key's own port, if it is back out of quarantine. + if own, ok := s.portAffinity[key]; ok { + if s.takeFreePort(own.port) { + return s.claimPort(key, own.port), nil + } + } + + // 2. Any released port no live claim covers — either never owned, or owned + // by a key whose window has lapsed, so no controller row can still name it. + owners := s.portOwners() + for i := len(s.freePorts) - 1; i >= 0; i-- { + port := s.freePorts[i] + if _, owned := owners[port]; owned { + continue + } + s.freePorts = slices.Delete(s.freePorts, i, i+1) + return s.claimPort(key, port), nil + } + + // 3. Grow into ports never handed out, staying inside the range. + if s.nextPort >= minPort && s.nextPort <= maxPort { + port := s.nextPort + s.nextPort++ + return s.claimPort(key, port), nil + } + + // 4. Steal another key's port rather than refuse to start a backend. if len(s.freePorts) > 0 { port := s.freePorts[len(s.freePorts)-1] s.freePorts = s.freePorts[:len(s.freePorts)-1] - return port + xlog.Warn("gRPC port range is exhausted; reusing a port that belonged to another backend. A stale controller row for the previous owner could briefly misroute to this backend — raise LOCALAI_GRPC_MAX_PORT to restore headroom", + "backend", key, "port", port, "previousOwner", owners[port], "min", minPort, "max", maxPort) + return s.claimPort(key, port), nil + } + + return 0, fmt.Errorf("%w: %d-%d is fully consumed by %d running backend(s) and %d port(s) still in quarantine; raise LOCALAI_GRPC_MAX_PORT to widen the range", + ErrNoFreePort, minPort, maxPort, len(s.processes), len(s.quarantinedPorts)) +} + +// sweepAffinity drops claims whose window has lapsed, so their ports become +// ordinary free ports again. Swept lazily on allocation for the same reason as +// sweepQuarantine: the only observer is allocation itself, so a timer goroutine +// per released port would buy nothing. Callers must hold s.mu. +func (s *backendSupervisor) sweepAffinity() { + if len(s.portAffinity) == 0 { + return } - port := s.nextPort - s.nextPort++ + now := time.Now() + for key, own := range s.portAffinity { + if own.expired(now) { + delete(s.portAffinity, key) + } + } +} + +// portOwners inverts the affinity map. Building it per allocation keeps a +// single source of truth: a second always-in-sync reverse map would be more +// state to get wrong, and allocation happens once per backend start. +func (s *backendSupervisor) portOwners() map[int]string { + owners := make(map[int]string, len(s.portAffinity)) + for key, own := range s.portAffinity { + owners[own.port] = key + } + return owners +} + +// takeFreePort removes `port` from the free pool, reporting whether it was +// there to take. +func (s *backendSupervisor) takeFreePort(port int) bool { + if i := slices.Index(s.freePorts, port); i >= 0 { + s.freePorts = slices.Delete(s.freePorts, i, i+1) + return true + } + return false +} + +// claimPort records `key` as the owner of `port` and evicts whichever key owned +// it before. +// +// That eviction is what bounds the affinity map. Ownership is kept injective +// over ports, so the map can never hold more entries than the range has ports, +// no matter how many distinct model keys the worker sees over its lifetime. +// Without it the map would grow once per distinct key forever — the same +// unbounded-growth shape as the port leak this change fixes. Callers must hold +// s.mu. +// The claim is recorded without an expiry: the port is in use, so it cannot be +// reallocated to anyone and the claim has nothing to age against. The window +// starts when the port is released. +func (s *backendSupervisor) claimPort(key string, port int) int { + if s.portAffinity == nil { + s.portAffinity = make(map[string]portOwnership) + } + for owner, own := range s.portAffinity { + if own.port == port && owner != key { + delete(s.portAffinity, owner) + } + } + s.portAffinity[key] = portOwnership{port: port} return port } +// releasePortForKey returns `port` to the allocator, reserving it for `key` for +// the affinity window so that no unrelated model can bind it while a controller +// row for this key could still name it. Callers must hold s.mu. +func (s *backendSupervisor) releasePortForKey(key string, port int) { + s.claimPort(key, port) + own := s.portAffinity[key] + own.until = time.Now().Add(s.affinityWindow()) + s.portAffinity[key] = own + s.releasePort(port) +} + // sweepQuarantine moves ports whose quarantine has elapsed into the free pool. // Sweeping lazily on allocation avoids a timer goroutine per stopped backend; // the only observer of the free pool is allocation itself. Callers must hold @@ -208,17 +415,20 @@ func (s *backendSupervisor) startBackend(backend, backendName, backendPath strin return bp.addr, nil } // Process died — clean up and restart - xlog.Warn("Backend process died unexpectedly, restarting", "backend", backend) - delete(s.processes, backend) + s.reapDeadProcess(backend, bp) } - port := s.allocatePort() + port, err := s.allocatePort(backend) + if err != nil { + s.mu.Unlock() + return "", fmt.Errorf("allocating gRPC port for backend %s: %w", backend, err) + } bindAddr := fmt.Sprintf("0.0.0.0:%d", port) clientAddr := fmt.Sprintf("127.0.0.1:%d", port) proc, err := s.ml.StartProcess(backendPath, backend, bindAddr) if err != nil { - s.releasePort(port) + s.releasePortForKey(backend, port) s.mu.Unlock() return "", fmt.Errorf("starting backend process: %w", err) } @@ -302,6 +512,36 @@ func (s *backendSupervisor) backendStartStillValid(key string, bp *backendProces return exists && current == bp && !current.stopping } +// reapDeadProcess drops the bookkeeping for a process that exited without +// anyone asking it to (OOM kill, CUDA fault, a segfaulting shared library). +// Unlike every other teardown path there is no request/reply in flight here: +// nothing on the controller side is waiting to be told, and the death is only +// noticed lazily, when something tries to start this key again. +// +// This path used to drop the map entry without releasing the port, so every +// unexpected exit consumed one port permanently. A crash-looping backend leaks +// one per restart, which is what actually walks a long-lived worker to the end +// of its range — not the concurrent-peak growth #10961 assumed. +// +// Releasing it through the affinity-preserving path is what makes the recycle +// safe: nothing here can tell the controller its row is stale (there is no +// reply to carry the key, and the death is noticed only lazily), so the port +// must come back only to this same key. See allocatePort. +// +// Callers must hold s.mu. +func (s *backendSupervisor) reapDeadProcess(key string, bp *backendProcess) { + xlog.Warn("Backend process died unexpectedly, restarting", "backend", key) + delete(s.processes, key) + if bp == nil { + return + } + if bp.port <= 0 { + xlog.Error("Cannot recycle backend port: dead process has invalid recorded port", "backend", key, "addr", bp.addr, "port", bp.port) + return + } + s.releasePortForKey(key, bp.port) +} + // releaseBackendStart removes a failed startup and recycles its port only when // the map still owns that exact attempt. A concurrent stop or replacement may // already have removed it and recycled the port. @@ -316,7 +556,7 @@ func (s *backendSupervisor) releaseBackendStart(key string, bp *backendProcess) xlog.Error("Cannot recycle backend port: startup has invalid recorded port", "backend", key, "addr", bp.addr, "port", bp.port) return } - s.releasePort(bp.port) + s.releasePortForKey(key, bp.port) } // resolveProcessKeys turns a caller-supplied identifier into the set of @@ -532,7 +772,7 @@ func (s *backendSupervisor) finishBackendStop(key string, bp *backendProcess, st xlog.Error("Cannot recycle backend port: process has invalid recorded port", "backend", key, "addr", bp.addr, "port", bp.port) return nil } - s.releasePort(bp.port) + s.releasePortForKey(key, bp.port) return nil } diff --git a/core/services/worker/worker.go b/core/services/worker/worker.go index 5cdafb7c0f76..04376242ab4b 100644 --- a/core/services/worker/worker.go +++ b/core/services/worker/worker.go @@ -199,15 +199,18 @@ func Run(ctx *cliContext.Context, cfg *Config) error { } supervisor := &backendSupervisor{ - cfg: cfg, - ml: ml, - systemState: systemState, - galleries: galleries, - nodeID: nodeID, - nats: natsClient, - sigCh: sigCh, - processes: make(map[string]*backendProcess), - nextPort: basePort, + cfg: cfg, + ml: ml, + systemState: systemState, + galleries: galleries, + nodeID: nodeID, + nats: natsClient, + sigCh: sigCh, + processes: make(map[string]*backendProcess), + portAffinity: make(map[string]portOwnership), + nextPort: basePort, + minPort: basePort, + maxPort: cfg.effectiveMaxPort(basePort), } if err := supervisor.subscribeLifecycleEvents(); err != nil { nodes.ShutdownFileTransferServer(httpServer) diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index cdeea890e115..e972833462d9 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -205,6 +205,7 @@ local-ai worker \ | Flag | Env Var | Default | Description | |------|---------|---------|-------------| | `--addr` | `LOCALAI_SERVE_ADDR` | `0.0.0.0:50051` | gRPC listen address | +| `--grpc-max-port` | `LOCALAI_GRPC_MAX_PORT` | `65535` | Highest port the worker may assign to a backend gRPC process. Each backend gets its own port, allocated upward from the base port, so the width of `[base port, this]` caps how many backends this worker can run at once (see [Backend gRPC port range](#backend-grpc-port-range)) | | `--advertise-addr` | `LOCALAI_ADVERTISE_ADDR` | *(auto)* | Address the frontend uses to reach this node (see below) | | `--http-addr` | `LOCALAI_HTTP_ADDR` | gRPC port - 1 | HTTP file transfer server bind address | | `--advertise-http-addr` | `LOCALAI_ADVERTISE_HTTP_ADDR` | *(auto)* | HTTP address the frontend uses for file transfer | @@ -253,10 +254,47 @@ For advanced networking scenarios (NAT, load balancers, separate gRPC/HTTP ports | Variable | Description | Default | |----------|-------------|---------| | `LOCALAI_SERVE_ADDR` | gRPC base port bind address | `0.0.0.0:50051` | +| `LOCALAI_GRPC_MAX_PORT` | Highest port assignable to a backend gRPC process | `65535` | | `LOCALAI_HTTP_ADDR` | HTTP file transfer bind address | `0.0.0.0:{gRPC port - 1}` | | `LOCALAI_ADVERTISE_ADDR` | Public gRPC address (if different from `LOCALAI_ADDR`) | Derived from `LOCALAI_ADDR` | | `LOCALAI_ADVERTISE_HTTP_ADDR` | Public HTTP address (if different from gRPC host) | Derived from advertise host + HTTP port | +### Backend gRPC port range + +Every backend process a worker starts listens on its own gRPC port, allocated +upward from the worker's base port (`LOCALAI_SERVE_ADDR`, default `50051`). +`LOCALAI_GRPC_MAX_PORT` sets the top of that range. The width of +`[base port, LOCALAI_GRPC_MAX_PORT]` is therefore a hard cap on how many +backend processes one worker can run concurrently. + +Set it when the worker shares a host with other services and you need to keep +the rest of the ephemeral range clear, or when you want a worker's backend +count bounded explicitly rather than by whatever the host happens to allow: + +```bash +# Confine this worker's backends to 50051-50150 (100 concurrent backends). +LOCALAI_SERVE_ADDR=0.0.0.0:50051 +LOCALAI_GRPC_MAX_PORT=50150 +``` + +Leave it unset (the default) and the worker may use anything up to 65535. + +Budget headroom above your real concurrency. When a backend stops, its port is +held briefly before it can be reused, so a worker with heavy start/stop churn +has more ports tied up than it has running backends at any instant. If the +range does fill, backend starts fail with: + +``` +no free gRPC port in range: 50051-50150 is fully consumed by 100 running +backend(s) and 12 port(s) still in quarantine; raise LOCALAI_GRPC_MAX_PORT to +widen the range +``` + +Raise `LOCALAI_GRPC_MAX_PORT` (or reduce how many models you schedule onto that +worker). A value above 65535 is clamped, and a value below the base port is +ignored in favour of the full range, so a typo degrades the setting rather than +wedging every backend start on the node. + ### NVIDIA GPU support When running workers in a container, two runtime settings affect how VRAM