diff --git a/core/cli/worker.go b/core/cli/worker.go index 2d6ffc605..537cae3e0 100644 --- a/core/cli/worker.go +++ b/core/cli/worker.go @@ -417,6 +417,14 @@ type backendSupervisor struct { processes map[string]*backendProcess // key: backend name nextPort int // next available port for new backends freePorts []int // ports freed by stopBackend, reused before nextPort + + // backendLocks serializes gallery operations against the same on-disk + // artifact. Two installs of different backends on the same worker run + // concurrently (their handlers are each in a goroutine); two operations + // on the same backend (install vs upgrade, or two parallel installs of + // the same not-yet-cached backend) are serialized here so the gallery + // download path doesn't race itself on the same directory. + backendLocks map[string]*sync.Mutex } // startBackend starts a gRPC backend process on a dynamically allocated port. @@ -792,6 +800,24 @@ func (s *backendSupervisor) findBackend(backend string) string { return "" } +// lockBackend returns a release function for a per-backend mutex. Different +// backend names lock independently. The first caller for a name allocates +// the mutex under s.mu; subsequent callers for the same name reuse it. +func (s *backendSupervisor) lockBackend(name string) func() { + s.mu.Lock() + if s.backendLocks == nil { + s.backendLocks = make(map[string]*sync.Mutex) + } + m, ok := s.backendLocks[name] + if !ok { + m = &sync.Mutex{} + s.backendLocks[name] = m + } + s.mu.Unlock() + m.Lock() + return m.Unlock +} + // subscribeLifecycleEvents subscribes to NATS backend lifecycle events. func (s *backendSupervisor) subscribeLifecycleEvents() { // backend.install — install backend + start gRPC process (request-reply) diff --git a/core/cli/worker_concurrency_test.go b/core/cli/worker_concurrency_test.go new file mode 100644 index 000000000..4dabfbac3 --- /dev/null +++ b/core/cli/worker_concurrency_test.go @@ -0,0 +1,72 @@ +package cli + +import ( + "sync" + "sync/atomic" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("backendSupervisor.lockBackend", func() { + It("serializes operations on the same backend name", func() { + s := &backendSupervisor{processes: map[string]*backendProcess{}} + + var inflight, peak int32 + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + release := s.lockBackend("llama-cpp") + defer release() + + now := atomic.AddInt32(&inflight, 1) + for { + p := atomic.LoadInt32(&peak) + if now <= p || atomic.CompareAndSwapInt32(&peak, p, now) { + break + } + } + time.Sleep(20 * time.Millisecond) + atomic.AddInt32(&inflight, -1) + }() + } + wg.Wait() + + Expect(atomic.LoadInt32(&peak)).To(Equal(int32(1)), + "only one goroutine should hold the per-backend lock at a time") + }) + + It("allows different backend names to run in parallel", func() { + s := &backendSupervisor{processes: map[string]*backendProcess{}} + + var inflight, peak int32 + var wg sync.WaitGroup + names := []string{"llama-cpp", "vllm", "whisper", "speaker-recognition"} + for _, n := range names { + n := n + wg.Add(1) + go func() { + defer wg.Done() + release := s.lockBackend(n) + defer release() + + now := atomic.AddInt32(&inflight, 1) + for { + p := atomic.LoadInt32(&peak) + if now <= p || atomic.CompareAndSwapInt32(&peak, p, now) { + break + } + } + time.Sleep(50 * time.Millisecond) + atomic.AddInt32(&inflight, -1) + }() + } + wg.Wait() + + Expect(atomic.LoadInt32(&peak)).To(BeNumerically(">=", int32(2)), + "distinct backends should be able to run concurrently") + }) +})