From e6f7501cad7b527f6417f7b47eb70a384be696a3 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 19 Jul 2026 22:19:51 +0000 Subject: [PATCH] fix(worker): bound the gRPC port allocator and stop leaking dead backends' ports The worker's gRPC port allocator grew monotonically with no upper bound: nextPort started at the base port and incremented whenever freePorts was empty, and nothing checked 65535. Past that it handed out integers that cannot be bound, surfacing as an opaque "backend won't start". #10961 estimated this needed ~15,000 concurrent-peak allocations, i.e. effectively unreachable. It is not, because of a second defect: the "process died unexpectedly" branch in startBackend deleted the process map entry without releasing its port at all. That port was leaked, never quarantined and never reused. A crash-looping backend leaks one port per restart, so a backend dying every 30s walks 50051 to 65535 in about five days. The leak, not concurrent peak, is the realistic route to exhaustion. Fixing the leak alone would have been wrong. Releasing that port makes it re-bindable, and the death path is the one teardown path with no request/reply to carry StoppedProcessKeys back to the controller (#10952's eager row removal), so a stale NodeModel row could then resolve to a live listener belonging to a different backend. probeHealth verifies liveness, not identity, so the request is silently misrouted. The 15s port quarantine does not cover this: the only reaper is the per-model health check at ~45s, and it can be disabled outright. The residual was masked only because the port was never rebound. So both are fixed together: - The allocator takes an explicit [basePort, LOCALAI_GRPC_MAX_PORT] range and returns ErrNoFreePort naming the range, the live backend count, the quarantined count, and the knob to raise. Exhaustion is now diagnosable instead of surfacing as an unbindable port. - Released ports carry per-key affinity: a port is offered back to the process key that last held it before any other key. Process keys (modelID#replica) and NodeModel rows (nodeID, modelName, replicaIndex) are isomorphic, so a port that can only be re-bound by its previous owner can only ever be named by that owner's row, which that key's re-registration overwrites. Misrouting to a different model becomes impossible by construction rather than by racing the quarantine timer. Affinity is a preference, not a reservation: under range pressure an owned port is stolen with a warning, because a guaranteed outage is worse than a rare misroute window on a port long out of quarantine. Claiming a port evicts its previous owner's entry, keeping ownership injective over ports so the affinity map can never exceed the range width regardless of how many distinct model keys the worker sees. Ownership also expires. It is only load-bearing while a controller row could still name the port, which the per-model reaper bounds at roughly 45s, so it lapses after five minutes and the port becomes ordinary free space again. Holding it indefinitely would have made every distinct model the worker ever served consume a port permanently: every release path is keyed, so nothing would ever be unowned, the allocator would climb to the end of its range on distinct-key count rather than concurrency, stealing would become routine, and the steal warning would tell operators to widen a range that was not the constraint. With expiry, reaching the steal branch means the worker is genuinely out of concurrent capacity, so that advice is correct when it appears. Closes #10961 Closes #10952 Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash] --- core/services/worker/config.go | 7 + core/services/worker/port_allocator_test.go | 268 +++++++++++++++++++ core/services/worker/port_quarantine_test.go | 27 +- core/services/worker/registration.go | 26 ++ core/services/worker/supervisor.go | 266 +++++++++++++++++- core/services/worker/worker.go | 21 +- docs/content/features/distributed-mode.md | 38 +++ 7 files changed, 620 insertions(+), 33 deletions(-) create mode 100644 core/services/worker/port_allocator_test.go 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