mirror of
https://github.com/mudler/LocalAI.git
synced 2026-10-28 09:08:54 -04:00
feat(galleryop): limit concurrent backend operations (#12888)
* feat(config): add a limit for concurrent backend operations Add --max-concurrent-backend-operations (LOCALAI_MAX_CONCURRENT_BACKEND_OPERATIONS) and the runtime setting max_concurrent_backend_operations. It sets how many backend install, upgrade and delete operations run at the same time. The default is 1, which keeps the current behaviour. Values below 1 become 1. The runtime value is also kept in an atomic, so the scheduler that reads it never races with a settings change. The setting shows in the Settings page under Backends. This commit only adds the setting; the next one uses it. Assisted-by: Claude:claude-sonnet-5-5 go-vet Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(galleryop): run backend operations in a bounded pool One goroutine read both gallery channels and ran each operation inline, so one slow backend operation held every operation behind it, model operations included. In distributed mode the manager also visited the nodes one at a time and waited for each up to the install timeout. Backend operations now go through a scheduler that runs up to max_concurrent_backend_operations of them at once. Operations start in the order they arrive. Install, upgrade and delete of one backend share a key and never overlap; an operation that waits for its backend does not hold a slot, so independent operations pass it. Cancelling a queued operation drops it at once. The limit is read on every scheduling step, so a runtime change applies without a restart. With the default of 1 the behaviour is the old single worker. Model operations get their own loop and stay serial. In distributed mode the manager sends one operation to its nodes in parallel, up to the same limit, and keeps the result in node order. A per (node, backend) lock keeps a live operation and the reconciler from sending the same operation to a node together; the reconciler skips a row that is locked and checks that the row still exists before it sends. Timeouts are unchanged. The download progress timers were plain package variables. They are atomics now, because operations can run in parallel. Assisted-by: Claude:claude-sonnet-5-5 go-vet go-test Signed-off-by: Ettore Di Giacinto <mudler@localai.io> --------- Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
mudler-agentandEttore Di Giacinto authored and GitHub committed
1 parent
21096764e2
commit
0b5cab414a
17 files changed
+1195
-221
No files matched your search
+13
-9
@@ -56,15 +56,18 @@ type RunCMD struct {
|
||||
NoColor string `env:"NO_COLOR" hidden:""`
|
||||
HFToken string `env:"HF_TOKEN" hidden:""`
|
||||
|
||||
ExternalBackends []string `env:"LOCALAI_EXTERNAL_BACKENDS,EXTERNAL_BACKENDS" help:"A list of external backends to load from gallery on boot" group:"backends"`
|
||||
WebRTCNAT1To1IPs []string `env:"LOCALAI_WEBRTC_NAT_1TO1_IPS,WEBRTC_NAT_1TO1_IPS" help:"IPs advertised as the host ICE candidates for /v1/realtime WebRTC instead of every local interface. Set to the reachable host/LAN IP when running under Docker host networking or NAT, where pion otherwise offers unreachable bridge addresses and the connection drops after ICE consent checks fail." group:"api"`
|
||||
WebRTCICEInterfaces []string `env:"LOCALAI_WEBRTC_ICE_INTERFACES,WEBRTC_ICE_INTERFACES" help:"Restrict /v1/realtime WebRTC ICE candidate gathering to these network interfaces (e.g. eth0), filtering out docker0/veth noise." group:"api"`
|
||||
WebRTCUDPPort int `env:"LOCALAI_WEBRTC_UDP_PORT" help:"Shared UDP port for /v1/realtime WebRTC ICE traffic. Publish this port as UDP and allow it through the firewall." group:"api" name:"web-rtc-udp-port"`
|
||||
BackendsPath string `env:"LOCALAI_BACKENDS_PATH,BACKENDS_PATH" type:"path" default:"${basepath}/backends" help:"Path containing backends used for inferencing" group:"backends"`
|
||||
BackendsSystemPath string `env:"LOCALAI_BACKENDS_SYSTEM_PATH,BACKEND_SYSTEM_PATH" type:"path" default:"/var/lib/local-ai/backends" help:"Path containing system backends used for inferencing" group:"backends"`
|
||||
ModelsPath string `env:"LOCALAI_MODELS_PATH,MODELS_PATH" type:"path" default:"${basepath}/models" help:"Path containing models used for inferencing" group:"storage"`
|
||||
DownloadStagingPath string `env:"LOCALAI_DOWNLOAD_STAGING_PATH,DOWNLOAD_STAGING_PATH" type:"path" default:"${basepath}/downloading" help:"Path where in-flight downloads are staged before extraction" group:"storage"`
|
||||
ArtifactDownloadConcurrency int `env:"LOCALAI_ARTIFACT_DOWNLOAD_CONCURRENCY" help:"How many files of a model artifact to download at once. 1 (the default) downloads sequentially. Raising it helps artifacts split into many files on a fast link, at the cost of more concurrent load on the models volume" group:"storage" default:"1"`
|
||||
ExternalBackends []string `env:"LOCALAI_EXTERNAL_BACKENDS,EXTERNAL_BACKENDS" help:"A list of external backends to load from gallery on boot" group:"backends"`
|
||||
WebRTCNAT1To1IPs []string `env:"LOCALAI_WEBRTC_NAT_1TO1_IPS,WEBRTC_NAT_1TO1_IPS" help:"IPs advertised as the host ICE candidates for /v1/realtime WebRTC instead of every local interface. Set to the reachable host/LAN IP when running under Docker host networking or NAT, where pion otherwise offers unreachable bridge addresses and the connection drops after ICE consent checks fail." group:"api"`
|
||||
WebRTCICEInterfaces []string `env:"LOCALAI_WEBRTC_ICE_INTERFACES,WEBRTC_ICE_INTERFACES" help:"Restrict /v1/realtime WebRTC ICE candidate gathering to these network interfaces (e.g. eth0), filtering out docker0/veth noise." group:"api"`
|
||||
WebRTCUDPPort int `env:"LOCALAI_WEBRTC_UDP_PORT" help:"Shared UDP port for /v1/realtime WebRTC ICE traffic. Publish this port as UDP and allow it through the firewall." group:"api" name:"web-rtc-udp-port"`
|
||||
BackendsPath string `env:"LOCALAI_BACKENDS_PATH,BACKENDS_PATH" type:"path" default:"${basepath}/backends" help:"Path containing backends used for inferencing" group:"backends"`
|
||||
BackendsSystemPath string `env:"LOCALAI_BACKENDS_SYSTEM_PATH,BACKEND_SYSTEM_PATH" type:"path" default:"/var/lib/local-ai/backends" help:"Path containing system backends used for inferencing" group:"backends"`
|
||||
ModelsPath string `env:"LOCALAI_MODELS_PATH,MODELS_PATH" type:"path" default:"${basepath}/models" help:"Path containing models used for inferencing" group:"storage"`
|
||||
DownloadStagingPath string `env:"LOCALAI_DOWNLOAD_STAGING_PATH,DOWNLOAD_STAGING_PATH" type:"path" default:"${basepath}/downloading" help:"Path where in-flight downloads are staged before extraction" group:"storage"`
|
||||
ArtifactDownloadConcurrency int `env:"LOCALAI_ARTIFACT_DOWNLOAD_CONCURRENCY" help:"How many files of a model artifact to download at once. 1 (the default) downloads sequentially. Raising it helps artifacts split into many files on a fast link, at the cost of more concurrent load on the models volume" group:"storage" default:"1"`
|
||||
|
||||
MaxConcurrentBackendOperations int `env:"LOCALAI_MAX_CONCURRENT_BACKEND_OPERATIONS" help:"How many backend install, upgrade and delete operations run at once. 1 (the default) runs them one after the other. Operations on the same backend always run one at a time" group:"backends" default:"1"`
|
||||
|
||||
GeneratedContentPath string `env:"LOCALAI_GENERATED_CONTENT_PATH,GENERATED_CONTENT_PATH" type:"path" default:"${generatedcontentpath}" help:"Location for generated content (e.g. images, audio, videos)" group:"storage"`
|
||||
UploadPath string `env:"LOCALAI_UPLOAD_PATH,UPLOAD_PATH" type:"path" default:"${uploadpath}" help:"Path to store uploads from files api" group:"storage"`
|
||||
DataPath string `env:"LOCALAI_DATA_PATH" type:"path" default:"${basepath}/data" help:"Path for persistent data (collectiondb, agent state, tasks, jobs). Separates mutable data from configuration" group:"storage"`
|
||||
@@ -338,6 +341,7 @@ func (r *RunCMD) Run(ctx *cliContext.Context) (result error) {
|
||||
config.WithProxyAPIKeyEnvLookup(os.Getenv),
|
||||
config.WithContext(runCtx),
|
||||
config.WithArtifactDownloadConcurrency(r.ArtifactDownloadConcurrency),
|
||||
config.WithMaxConcurrentBackendOperations(r.MaxConcurrentBackendOperations),
|
||||
config.WithModelArtifactMaterializer(modelartifacts.NewDefaultManager(
|
||||
modelartifacts.WithHuggingFaceToken(r.HFToken),
|
||||
modelartifacts.WithDownloadConcurrency(r.ArtifactDownloadConcurrency),
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"regexp"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/mudler/LocalAI/core/services/addons/gateway"
|
||||
@@ -123,14 +124,22 @@ type ApplicationConfig struct {
|
||||
WebRTCUDPPort int
|
||||
UploadLimitMB, Threads, ContextSize int
|
||||
ArtifactDownloadConcurrency int
|
||||
F16 bool
|
||||
Debug bool
|
||||
EnableTracing bool
|
||||
MaxConcurrentBackendRequests int
|
||||
TracingMaxItems int
|
||||
TracingMaxBodyBytes int // Per-body cap for captured request/response bodies; 0 disables the cap
|
||||
EnableBackendLogging bool
|
||||
GeneratedContentDir string
|
||||
|
||||
// MaxConcurrentBackendOperations bounds how many backend install, upgrade
|
||||
// and delete operations run at once. Always >= 1.
|
||||
MaxConcurrentBackendOperations int
|
||||
// liveMaxBackendOps mirrors MaxConcurrentBackendOperations for readers on
|
||||
// other goroutines, since a runtime settings change can land at any time.
|
||||
liveMaxBackendOps atomic.Int64
|
||||
|
||||
F16 bool
|
||||
Debug bool
|
||||
EnableTracing bool
|
||||
MaxConcurrentBackendRequests int
|
||||
TracingMaxItems int
|
||||
TracingMaxBodyBytes int // Per-body cap for captured request/response bodies; 0 disables the cap
|
||||
EnableBackendLogging bool
|
||||
GeneratedContentDir string
|
||||
|
||||
UploadDir string
|
||||
DataPath string // Persistent data directory for collectiondb, agents, etc.
|
||||
@@ -388,10 +397,13 @@ func NewApplicationConfig(o ...AppOption) *ApplicationConfig {
|
||||
EnableBackendLogging: true,
|
||||
VRAMPersistentCache: true,
|
||||
ArtifactDownloadConcurrency: modelartifacts.DefaultDownloadConcurrency,
|
||||
AgentJobRetentionDays: 30, // Default: 30 days
|
||||
LRUEvictionMaxRetries: 30, // Default: 30 retries
|
||||
LRUEvictionRetryInterval: 1 * time.Second, // Default: 1 second
|
||||
ModelLoadFailureCooldown: 10 * time.Second, // Default: 10s base cooldown after a failed load
|
||||
|
||||
MaxConcurrentBackendOperations: 1,
|
||||
|
||||
AgentJobRetentionDays: 30, // Default: 30 days
|
||||
LRUEvictionMaxRetries: 30, // Default: 30 retries
|
||||
LRUEvictionRetryInterval: 1 * time.Second, // Default: 1 second
|
||||
ModelLoadFailureCooldown: 10 * time.Second, // Default: 10s base cooldown after a failed load
|
||||
// WatchDogInterval is intentionally left at the zero value here.
|
||||
// The startup loader applies a persisted runtime_settings.json value
|
||||
// only when the interval is still 0 (its "not set by env var"
|
||||
@@ -416,6 +428,7 @@ func NewApplicationConfig(o ...AppOption) *ApplicationConfig {
|
||||
for _, oo := range o {
|
||||
oo(opt)
|
||||
}
|
||||
opt.SetMaxConcurrentBackendOperations(opt.MaxConcurrentBackendOperations)
|
||||
return opt
|
||||
}
|
||||
|
||||
@@ -791,6 +804,38 @@ func WithModelArtifactMaterializer(materializer ArtifactMaterializer) AppOption
|
||||
}
|
||||
}
|
||||
|
||||
// WithMaxConcurrentBackendOperations sets how many backend operations run at
|
||||
// once. Values below 1 fall back to 1.
|
||||
func WithMaxConcurrentBackendOperations(n int) AppOption {
|
||||
return func(o *ApplicationConfig) {
|
||||
if n < 1 {
|
||||
n = 1
|
||||
}
|
||||
o.SetMaxConcurrentBackendOperations(n)
|
||||
}
|
||||
}
|
||||
|
||||
// SetMaxConcurrentBackendOperations stores the limit, clamped to at least 1.
|
||||
func (o *ApplicationConfig) SetMaxConcurrentBackendOperations(n int) {
|
||||
if n < 1 {
|
||||
n = 1
|
||||
}
|
||||
o.MaxConcurrentBackendOperations = n
|
||||
o.liveMaxBackendOps.Store(int64(n))
|
||||
}
|
||||
|
||||
// BackendOperationLimit returns how many backend operations may run at once.
|
||||
// It is safe to call while a runtime settings change is applied.
|
||||
func (o *ApplicationConfig) BackendOperationLimit() int {
|
||||
if n := o.liveMaxBackendOps.Load(); n >= 1 {
|
||||
return int(n)
|
||||
}
|
||||
if o.MaxConcurrentBackendOperations >= 1 {
|
||||
return o.MaxConcurrentBackendOperations
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
func WithArtifactDownloadConcurrency(concurrency int) AppOption {
|
||||
return func(o *ApplicationConfig) {
|
||||
if concurrency < 1 {
|
||||
|
||||
@@ -35,16 +35,19 @@ type RuntimeSettings struct {
|
||||
LRUEvictionRetryInterval *string `json:"lru_eviction_retry_interval,omitempty"` // Interval between retries when waiting for busy models (e.g., 1s, 2s) (default: 1s)
|
||||
|
||||
// Performance settings
|
||||
Threads *int `json:"threads,omitempty"`
|
||||
ContextSize *int `json:"context_size,omitempty"`
|
||||
ArtifactDownloadConcurrency *int `json:"artifact_download_concurrency,omitempty"`
|
||||
VRAMBudget *string `json:"vram_budget,omitempty"` // Cap VRAM for allocation ("80%" or "12GB"; "" = no cap)
|
||||
F16 *bool `json:"f16,omitempty"`
|
||||
Debug *bool `json:"debug,omitempty"`
|
||||
EnableTracing *bool `json:"enable_tracing,omitempty"`
|
||||
TracingMaxItems *int `json:"tracing_max_items,omitempty"`
|
||||
TracingMaxBodyBytes *int `json:"tracing_max_body_bytes,omitempty"` // Per-body cap in bytes; 0 disables the cap
|
||||
EnableBackendLogging *bool `json:"enable_backend_logging,omitempty"`
|
||||
Threads *int `json:"threads,omitempty"`
|
||||
ContextSize *int `json:"context_size,omitempty"`
|
||||
ArtifactDownloadConcurrency *int `json:"artifact_download_concurrency,omitempty"`
|
||||
|
||||
MaxConcurrentBackendOperations *int `json:"max_concurrent_backend_operations,omitempty"`
|
||||
|
||||
VRAMBudget *string `json:"vram_budget,omitempty"` // Cap VRAM for allocation ("80%" or "12GB"; "" = no cap)
|
||||
F16 *bool `json:"f16,omitempty"`
|
||||
Debug *bool `json:"debug,omitempty"`
|
||||
EnableTracing *bool `json:"enable_tracing,omitempty"`
|
||||
TracingMaxItems *int `json:"tracing_max_items,omitempty"`
|
||||
TracingMaxBodyBytes *int `json:"tracing_max_body_bytes,omitempty"` // Per-body cap in bytes; 0 disables the cap
|
||||
EnableBackendLogging *bool `json:"enable_backend_logging,omitempty"`
|
||||
|
||||
// Security/CORS settings
|
||||
CORS *bool `json:"cors,omitempty"`
|
||||
|
||||
@@ -245,6 +245,12 @@ var runtimeSettingsFields = []fieldSpec{
|
||||
}
|
||||
o.ArtifactDownloadConcurrency = v
|
||||
}),
|
||||
field("max_concurrent_backend_operations",
|
||||
func(s *RuntimeSettings) **int { return &s.MaxConcurrentBackendOperations },
|
||||
func(o *ApplicationConfig) int { return o.MaxConcurrentBackendOperations },
|
||||
func(o *ApplicationConfig, v int) {
|
||||
o.SetMaxConcurrentBackendOperations(v)
|
||||
}),
|
||||
// VRAM budget: the cap string ("80%"/"12GB"/"" = uncapped). The live
|
||||
// side effect (xsysinfo.SetDefaultVRAMBudget) is post-processing in the
|
||||
// apply loop, not here - the row only owns the config member, matching
|
||||
|
||||
@@ -72,6 +72,7 @@ var _ = Describe("runtime settings registry", func() {
|
||||
src.Threads = 7
|
||||
src.ContextSize = 8192
|
||||
src.ArtifactDownloadConcurrency = 6
|
||||
src.MaxConcurrentBackendOperations = 3
|
||||
src.VRAMBudget = "12GiB"
|
||||
src.F16 = true
|
||||
src.Debug = true
|
||||
|
||||
@@ -112,6 +112,8 @@ export const FIELDS = [
|
||||
desc: 'Context window size for models that do not set their own.', placeholder: '2048', min: 0 },
|
||||
{ key: 'artifact_download_concurrency', group: 'speed', was: 'performance', kind: 'int', label: 'Artifact Download Concurrency',
|
||||
desc: 'Maximum artifact files downloaded at once. 1 downloads one after the other.', default: 1, min: 1 },
|
||||
{ key: 'max_concurrent_backend_operations', group: 'backends', was: 'backends', kind: 'int', label: 'Concurrent backend operations',
|
||||
desc: 'Backend installs, upgrades and deletes that run at once. 1 runs them one after the other. Operations on the same backend never overlap.', default: 1, min: 1 },
|
||||
{ key: 'f16', group: 'speed', was: 'performance', kind: 'bool', label: 'F16 precision',
|
||||
desc: 'Use 16-bit floating point for reduced memory use.', default: false },
|
||||
|
||||
|
||||
@@ -0,0 +1,341 @@
|
||||
package galleryop_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mudler/LocalAI/core/config"
|
||||
"github.com/mudler/LocalAI/core/gallery"
|
||||
"github.com/mudler/LocalAI/core/services/distributed"
|
||||
"github.com/mudler/LocalAI/core/services/galleryop"
|
||||
)
|
||||
|
||||
// gatedBackendManager parks every operation until its backend is released. It
|
||||
// records how many operations ran at once, and per backend, so a spec can tell
|
||||
// a pool that really runs in parallel from one that only looks like it.
|
||||
type gatedBackendManager struct {
|
||||
mu sync.Mutex
|
||||
gates map[string]chan struct{}
|
||||
started []string
|
||||
running int
|
||||
maxRunning int
|
||||
perName map[string]int
|
||||
overlapped bool
|
||||
}
|
||||
|
||||
func newGatedBackendManager() *gatedBackendManager {
|
||||
return &gatedBackendManager{gates: map[string]chan struct{}{}, perName: map[string]int{}}
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) gate(name string) chan struct{} {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
g, ok := m.gates[name]
|
||||
if !ok {
|
||||
g = make(chan struct{})
|
||||
m.gates[name] = g
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Release lets every operation on name finish.
|
||||
func (m *gatedBackendManager) Release(name string) { close(m.gate(name)) }
|
||||
|
||||
func (m *gatedBackendManager) Started() []string {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return append([]string(nil), m.started...)
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) MaxRunning() int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.maxRunning
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) Overlapped() bool {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.overlapped
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) run(name string) {
|
||||
g := m.gate(name)
|
||||
m.mu.Lock()
|
||||
m.started = append(m.started, name)
|
||||
m.running++
|
||||
m.perName[name]++
|
||||
if m.perName[name] > 1 {
|
||||
m.overlapped = true
|
||||
}
|
||||
if m.running > m.maxRunning {
|
||||
m.maxRunning = m.running
|
||||
}
|
||||
m.mu.Unlock()
|
||||
|
||||
<-g
|
||||
|
||||
m.mu.Lock()
|
||||
m.running--
|
||||
m.perName[name]--
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) InstallBackend(_ context.Context, op *galleryop.ManagementOp[gallery.GalleryBackend, any], _ galleryop.ProgressCallback) error {
|
||||
m.run(op.GalleryElementName)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) UpgradeBackend(_ context.Context, op *galleryop.ManagementOp[gallery.GalleryBackend, any], _ galleryop.ProgressCallback) error {
|
||||
m.run(op.GalleryElementName)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *gatedBackendManager) DeleteBackend(name string) error {
|
||||
m.run(name)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*gatedBackendManager) ListBackends() (gallery.SystemBackends, error) {
|
||||
return gallery.SystemBackends{}, nil
|
||||
}
|
||||
|
||||
func (*gatedBackendManager) CheckUpgrades(context.Context) (map[string]gallery.UpgradeInfo, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (*gatedBackendManager) IsDistributed() bool { return false }
|
||||
|
||||
func (*gatedBackendManager) BackendAvailable(string) (bool, error) { return false, nil }
|
||||
|
||||
var _ = Describe("concurrent backend operations", func() {
|
||||
var (
|
||||
mgr *gatedBackendManager
|
||||
svc *galleryop.GalleryService
|
||||
cfg *config.ApplicationConfig
|
||||
ctx context.Context
|
||||
)
|
||||
|
||||
start := func(limit int) {
|
||||
cfg = config.NewApplicationConfig(config.WithMaxConcurrentBackendOperations(limit))
|
||||
mgr = newGatedBackendManager()
|
||||
svc = galleryop.NewGalleryService(cfg, nil)
|
||||
svc.SetBackendManager(mgr)
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
Expect(svc.Start(ctx, nil, nil)).To(Succeed())
|
||||
}
|
||||
|
||||
enqueue := func(id, name string, opts ...func(*galleryop.ManagementOp[gallery.GalleryBackend, any])) {
|
||||
op := galleryop.ManagementOp[gallery.GalleryBackend, any]{ID: id, GalleryElementName: name}
|
||||
for _, o := range opts {
|
||||
o(&op)
|
||||
}
|
||||
svc.EnqueueBackendOp(op)
|
||||
}
|
||||
|
||||
processed := func(id string) func() bool {
|
||||
return func() bool {
|
||||
st := svc.GetStatus(id)
|
||||
return st != nil && st.Processed
|
||||
}
|
||||
}
|
||||
|
||||
It("runs one operation at a time by default", func() {
|
||||
start(0)
|
||||
enqueue("a", "alpha")
|
||||
enqueue("b", "beta")
|
||||
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
Consistently(mgr.Started, "300ms", "10ms").Should(HaveLen(1))
|
||||
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("beta")
|
||||
Eventually(processed("a"), "5s", "10ms").Should(BeTrue())
|
||||
Eventually(processed("b"), "5s", "10ms").Should(BeTrue())
|
||||
Expect(mgr.MaxRunning()).To(Equal(1))
|
||||
})
|
||||
|
||||
It("reaches the configured number of parallel operations and no more", func() {
|
||||
start(3)
|
||||
for _, n := range []string{"alpha", "beta", "gamma", "delta", "epsilon"} {
|
||||
enqueue("op-"+n, n)
|
||||
}
|
||||
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(3))
|
||||
Consistently(mgr.Started, "300ms", "10ms").Should(HaveLen(3))
|
||||
Expect(mgr.MaxRunning()).To(Equal(3))
|
||||
|
||||
for _, n := range []string{"alpha", "beta", "gamma", "delta", "epsilon"} {
|
||||
mgr.Release(n)
|
||||
}
|
||||
for _, n := range []string{"alpha", "beta", "gamma", "delta", "epsilon"} {
|
||||
Eventually(processed("op-"+n), "5s", "10ms").Should(BeTrue())
|
||||
}
|
||||
Expect(mgr.MaxRunning()).To(Equal(3))
|
||||
})
|
||||
|
||||
It("keeps operations on one backend serialized, whatever their kind", func() {
|
||||
start(4)
|
||||
enqueue("install", "alpha")
|
||||
enqueue("upgrade", "alpha", func(o *galleryop.ManagementOp[gallery.GalleryBackend, any]) { o.Upgrade = true })
|
||||
enqueue("delete", "gallery@alpha", func(o *galleryop.ManagementOp[gallery.GalleryBackend, any]) { o.Delete = true })
|
||||
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
Consistently(mgr.Started, "300ms", "10ms").Should(HaveLen(1))
|
||||
// The ones that wait are still queued, not failed.
|
||||
Eventually(func() int {
|
||||
queued := 0
|
||||
for _, id := range []string{"install", "upgrade", "delete"} {
|
||||
if svc.GetStatus(id).IsQueued() {
|
||||
queued++
|
||||
}
|
||||
}
|
||||
return queued
|
||||
}, "2s", "10ms").Should(Equal(2))
|
||||
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("gallery@alpha") // DeleteBackend receives the name as given
|
||||
for _, id := range []string{"install", "upgrade", "delete"} {
|
||||
Eventually(processed(id), "5s", "10ms").Should(BeTrue())
|
||||
}
|
||||
Expect(mgr.Overlapped()).To(BeFalse())
|
||||
Expect(mgr.Started()).To(HaveLen(3))
|
||||
})
|
||||
|
||||
It("lets an independent operation pass a slow one", func() {
|
||||
start(2)
|
||||
enqueue("slow", "alpha")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
enqueue("fast", "beta")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(2))
|
||||
|
||||
mgr.Release("beta")
|
||||
Eventually(processed("fast"), "5s", "10ms").Should(BeTrue())
|
||||
Expect(processed("slow")()).To(BeFalse())
|
||||
|
||||
mgr.Release("alpha")
|
||||
Eventually(processed("slow"), "5s", "10ms").Should(BeTrue())
|
||||
})
|
||||
|
||||
It("does not let a same-backend op that waits block an independent one", func() {
|
||||
start(2)
|
||||
enqueue("first", "alpha")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
enqueue("second", "alpha")
|
||||
enqueue("third", "beta")
|
||||
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(ConsistOf("alpha", "beta"))
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("beta")
|
||||
for _, id := range []string{"first", "second", "third"} {
|
||||
Eventually(processed(id), "5s", "10ms").Should(BeTrue())
|
||||
}
|
||||
})
|
||||
|
||||
It("does not hold model operations behind a slow backend operation", func() {
|
||||
start(1)
|
||||
models := newGatedModelManager()
|
||||
svc.SetModelManager(models)
|
||||
close(models.gate)
|
||||
|
||||
enqueue("slow", "alpha")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
|
||||
svc.EnqueueModelOp(galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{ID: "m", GalleryElementName: "some-model"})
|
||||
Eventually(processed("m"), "5s", "10ms").Should(BeTrue())
|
||||
Expect(processed("slow")()).To(BeFalse())
|
||||
|
||||
mgr.Release("alpha")
|
||||
})
|
||||
|
||||
It("cancels a queued operation without running it", func() {
|
||||
start(1)
|
||||
enqueue("running", "alpha")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
|
||||
cctx, cancel, pause := galleryop.NewUserCancellableContext(context.Background())
|
||||
enqueue("queued", "beta", func(o *galleryop.ManagementOp[gallery.GalleryBackend, any]) {
|
||||
o.Context, o.CancelFunc, o.PauseFunc = cctx, cancel, pause
|
||||
})
|
||||
svc.StoreCancellationActions("queued", cancel, pause)
|
||||
Eventually(func() bool { return svc.GetStatus("queued").IsQueued() }, "2s", "10ms").Should(BeTrue())
|
||||
|
||||
// A second queued op for the same name goes through the service API,
|
||||
// with a context the service creates.
|
||||
enqueue("queued-by-id", "gamma")
|
||||
Eventually(func() bool { return svc.GetStatus("queued-by-id").IsQueued() }, "2s", "10ms").Should(BeTrue())
|
||||
|
||||
Expect(svc.CancelOperation("queued")).To(Succeed())
|
||||
Eventually(processed("queued"), "5s", "10ms").Should(BeTrue())
|
||||
Expect(svc.GetStatus("queued").Cancelled).To(BeTrue())
|
||||
|
||||
// The service stores the cancel func of this one when it takes the op
|
||||
// off the channel, which is a moment after admission.
|
||||
Eventually(func() error { return svc.CancelOperation("queued-by-id") }, "2s", "10ms").Should(Succeed())
|
||||
Eventually(processed("queued-by-id"), "5s", "10ms").Should(BeTrue())
|
||||
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("beta")
|
||||
mgr.Release("gamma")
|
||||
Eventually(processed("running"), "5s", "10ms").Should(BeTrue())
|
||||
Consistently(mgr.Started, "300ms", "10ms").Should(Equal([]string{"alpha"}))
|
||||
})
|
||||
|
||||
It("follows a raised limit while operations wait", func() {
|
||||
start(1)
|
||||
enqueue("a", "alpha")
|
||||
enqueue("b", "beta")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(1))
|
||||
Consistently(mgr.Started, "300ms", "10ms").Should(HaveLen(1))
|
||||
|
||||
applied := 2
|
||||
cfg.ApplyRuntimeSettings(&config.RuntimeSettings{MaxConcurrentBackendOperations: &applied})
|
||||
Eventually(mgr.Started, "5s", "50ms").Should(HaveLen(2))
|
||||
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("beta")
|
||||
})
|
||||
|
||||
It("clamps a runtime value below 1 to 1", func() {
|
||||
cfg = config.NewApplicationConfig(config.WithMaxConcurrentBackendOperations(4))
|
||||
zero := 0
|
||||
cfg.ApplyRuntimeSettings(&config.RuntimeSettings{MaxConcurrentBackendOperations: &zero})
|
||||
Expect(cfg.MaxConcurrentBackendOperations).To(Equal(1))
|
||||
})
|
||||
|
||||
It("persists one row per concurrently running operation", func() {
|
||||
start(2)
|
||||
db, err := gorm.Open(sqlite.Open(filepath.Join(GinkgoT().TempDir(), "ops.db")+"?_busy_timeout=5000"), &gorm.Config{})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
store, err := distributed.NewGalleryStore(db)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
svc.SetGalleryStore(store)
|
||||
|
||||
enqueue("row-a", "alpha")
|
||||
enqueue("row-b", "beta")
|
||||
Eventually(mgr.Started, "5s", "10ms").Should(HaveLen(2))
|
||||
for _, id := range []string{"row-a", "row-b"} {
|
||||
Eventually(func() error { _, err := store.Get(id); return err }, "5s", "20ms").Should(Succeed())
|
||||
}
|
||||
|
||||
mgr.Release("alpha")
|
||||
mgr.Release("beta")
|
||||
for _, id := range []string{"row-a", "row-b"} {
|
||||
Eventually(func() string {
|
||||
rec, err := store.Get(id)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return rec.Status
|
||||
}, "5s", "20ms").Should(Equal("completed"))
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,152 @@
|
||||
package galleryop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// opTask is one unit of work for the opScheduler.
|
||||
type opTask struct {
|
||||
// key serializes tasks: two tasks with the same key never run together.
|
||||
key string
|
||||
// ctx is the operation context. When it ends while the task still waits,
|
||||
// the task is dropped and abandon runs instead of run.
|
||||
ctx context.Context
|
||||
run func()
|
||||
abandon func()
|
||||
|
||||
stopWatch func() bool
|
||||
}
|
||||
|
||||
// opScheduler runs tasks with a bounded number of workers.
|
||||
//
|
||||
// Tasks start in submit order, except that a task whose key is busy waits and
|
||||
// lets later tasks with other keys go first. The limit is read each time the
|
||||
// scheduler looks for work, so it follows a runtime setting change without a
|
||||
// restart. With a limit of 1 the behaviour is the old single worker: one task
|
||||
// at a time, in submit order.
|
||||
type opScheduler struct {
|
||||
limit func() int
|
||||
|
||||
mu sync.Mutex
|
||||
pending []*opTask
|
||||
running int
|
||||
busy map[string]struct{}
|
||||
}
|
||||
|
||||
func newOpScheduler(limit func() int) *opScheduler {
|
||||
return &opScheduler{limit: limit, busy: map[string]struct{}{}}
|
||||
}
|
||||
|
||||
func (s *opScheduler) maxRunning() int {
|
||||
if n := s.limit(); n > 1 {
|
||||
return n
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// submit queues a task. It never blocks.
|
||||
func (s *opScheduler) submit(t *opTask) {
|
||||
s.mu.Lock()
|
||||
s.pending = append(s.pending, t)
|
||||
if t.ctx != nil {
|
||||
// A cancel of a queued task must free it at once, not when its turn
|
||||
// comes, so a cancelled operation never holds a place in the queue.
|
||||
t.stopWatch = context.AfterFunc(t.ctx, func() { s.dropPending(t) })
|
||||
}
|
||||
s.dispatchLocked()
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// kick starts waiting tasks if the limit has grown since the last look.
|
||||
func (s *opScheduler) kick() {
|
||||
s.mu.Lock()
|
||||
s.dispatchLocked()
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// hasPending reports whether any task waits for a slot.
|
||||
func (s *opScheduler) hasPending() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return len(s.pending) > 0
|
||||
}
|
||||
|
||||
// abandonAll drops every waiting task. Running tasks are not touched.
|
||||
func (s *opScheduler) abandonAll() {
|
||||
s.mu.Lock()
|
||||
dropped := s.pending
|
||||
s.pending = nil
|
||||
s.mu.Unlock()
|
||||
for _, t := range dropped {
|
||||
if t.stopWatch != nil {
|
||||
t.stopWatch()
|
||||
}
|
||||
t.abandon()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *opScheduler) dropPending(t *opTask) {
|
||||
s.mu.Lock()
|
||||
found := false
|
||||
for i, p := range s.pending {
|
||||
if p == t {
|
||||
s.pending = append(s.pending[:i], s.pending[i+1:]...)
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
if found {
|
||||
t.abandon()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *opScheduler) dispatchLocked() {
|
||||
max := s.maxRunning()
|
||||
for i := 0; i < len(s.pending) && s.running < max; {
|
||||
t := s.pending[i]
|
||||
if _, held := s.busy[t.key]; held {
|
||||
i++
|
||||
continue
|
||||
}
|
||||
s.pending = append(s.pending[:i], s.pending[i+1:]...)
|
||||
if t.stopWatch != nil && !t.stopWatch() {
|
||||
// The context already ended and its callback is on the way; it
|
||||
// finds the task gone and does nothing, so run the abandon here.
|
||||
go t.abandon()
|
||||
continue
|
||||
}
|
||||
s.busy[t.key] = struct{}{}
|
||||
s.running++
|
||||
go func() {
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
s.running--
|
||||
delete(s.busy, t.key)
|
||||
s.dispatchLocked()
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
t.run()
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// backendOpKey names the backend an operation touches, so install, upgrade and
|
||||
// delete of one backend share a key. A gallery prefix ("gallery@name") is
|
||||
// dropped and the case is folded: the same backend can be addressed both ways.
|
||||
func backendOpKey(elementName, externalName string) string {
|
||||
name := externalName
|
||||
if name == "" {
|
||||
name = elementName
|
||||
}
|
||||
if i := strings.LastIndex(name, "@"); i >= 0 {
|
||||
name = name[i+1:]
|
||||
}
|
||||
return strings.ToLower(strings.TrimSpace(name))
|
||||
}
|
||||
|
||||
// schedulerPollInterval is how often a waiting queue re-reads the limit.
|
||||
const schedulerPollInterval = time.Second
|
||||
@@ -30,6 +30,9 @@ type GalleryService struct {
|
||||
statuses map[string]*OpStatus
|
||||
cancellations map[string]cancellationActions
|
||||
|
||||
// backendScheduler runs backend operations; set by Start.
|
||||
backendScheduler *opScheduler
|
||||
|
||||
// Distributed mode (nil when not in distributed mode).
|
||||
// natsClient is a messaging.Broadcaster (Publisher + Subscribe)
|
||||
// when wired by the distributed startup path; broadcastSubs holds the
|
||||
@@ -602,11 +605,26 @@ func (g *GalleryService) Start(c context.Context, cl *config.ModelConfigLoader,
|
||||
}
|
||||
}
|
||||
|
||||
// Backend and model operations have their own loops. A slow backend
|
||||
// operation no longer holds back model operations, and the other way round.
|
||||
sched := newOpScheduler(func() int { return g.maxConcurrentBackendOps() })
|
||||
g.backendScheduler = sched
|
||||
|
||||
go func() {
|
||||
ticker := time.NewTicker(schedulerPollInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-c.Done():
|
||||
// Operations that never started are failed, not left as
|
||||
// "queued" forever. Running ones follow their own contexts.
|
||||
sched.abandonAll()
|
||||
return
|
||||
case <-ticker.C:
|
||||
// Picks up a raised limit while operations wait.
|
||||
if sched.hasPending() {
|
||||
sched.kick()
|
||||
}
|
||||
case op := <-g.BackendGalleryChannel:
|
||||
// Create context if not provided
|
||||
if op.Context == nil {
|
||||
@@ -615,41 +633,24 @@ func (g *GalleryService) Start(c context.Context, cl *config.ModelConfigLoader,
|
||||
} else if op.CancelFunc != nil {
|
||||
g.storeCancellation(op.ID, op.CancelFunc, op.PauseFunc)
|
||||
}
|
||||
// Create DB record for distributed tracking
|
||||
if g.galleryStore != nil {
|
||||
opType := "backend_install"
|
||||
if op.Delete {
|
||||
opType = "backend_delete"
|
||||
}
|
||||
if err := g.galleryStore.Create(&distributed.GalleryOperationRecord{
|
||||
ID: op.ID,
|
||||
GalleryElementName: op.GalleryElementName,
|
||||
OpType: opType,
|
||||
Status: "pending",
|
||||
// Create runs at dequeue, so this is the running-phase
|
||||
// value: a running delete is not cancellable, an install
|
||||
// is, matching the model channel. The queued phase before
|
||||
// this point is cancellable either way (see markQueued).
|
||||
Cancellable: !op.Delete,
|
||||
}); err != nil {
|
||||
// Not fatal: the install still runs and the in-memory
|
||||
// status still updates. Logged because without the row
|
||||
// the cross-replica dedup guard and hydration cannot
|
||||
// see this operation at all.
|
||||
xlog.Warn("Failed to create gallery operation record", "op_id", op.ID, "error", err)
|
||||
}
|
||||
}
|
||||
err := runOpHandler(func() error { return g.backendHandler(&op, systemState) })
|
||||
if err != nil {
|
||||
updateError(op.ID, err)
|
||||
} else if g.OnBackendOpCompleted != nil {
|
||||
// Let listeners (e.g. UpgradeChecker) refresh their view of
|
||||
// installed state. Run off the worker goroutine so a slow
|
||||
// callback doesn't stall the next queued operation.
|
||||
go g.OnBackendOpCompleted()
|
||||
}
|
||||
g.removeCancellation(op.ID)
|
||||
sched.submit(&opTask{
|
||||
key: backendOpKey(op.GalleryElementName, op.ExternalName),
|
||||
ctx: op.Context,
|
||||
run: func() { g.runBackendOp(&op, systemState, updateError) },
|
||||
abandon: func() {
|
||||
g.abandonQueued(op.ID, op.GalleryElementName)
|
||||
g.removeCancellation(op.ID)
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-c.Done():
|
||||
return
|
||||
case op := <-g.ModelGalleryChannel:
|
||||
// Create context if not provided
|
||||
if op.Context == nil {
|
||||
@@ -690,6 +691,53 @@ func (g *GalleryService) Start(c context.Context, cl *config.ModelConfigLoader,
|
||||
return nil
|
||||
}
|
||||
|
||||
// maxConcurrentBackendOps is the number of backend operations allowed to run
|
||||
// at once. It follows the runtime setting.
|
||||
func (g *GalleryService) maxConcurrentBackendOps() int {
|
||||
if g.appConfig == nil {
|
||||
return 1
|
||||
}
|
||||
return g.appConfig.BackendOperationLimit()
|
||||
}
|
||||
|
||||
// runBackendOp runs one backend operation once the scheduler gives it a slot.
|
||||
func (g *GalleryService) runBackendOp(op *ManagementOp[gallery.GalleryBackend, any], systemState *system.SystemState, updateError func(string, error)) {
|
||||
// Create DB record for distributed tracking
|
||||
if g.galleryStore != nil {
|
||||
opType := "backend_install"
|
||||
if op.Delete {
|
||||
opType = "backend_delete"
|
||||
}
|
||||
if err := g.galleryStore.Create(&distributed.GalleryOperationRecord{
|
||||
ID: op.ID,
|
||||
GalleryElementName: op.GalleryElementName,
|
||||
OpType: opType,
|
||||
Status: "pending",
|
||||
// Create runs at dequeue, so this is the running-phase
|
||||
// value: a running delete is not cancellable, an install
|
||||
// is, matching the model channel. The queued phase before
|
||||
// this point is cancellable either way (see markQueued).
|
||||
Cancellable: !op.Delete,
|
||||
}); err != nil {
|
||||
// Not fatal: the install still runs and the in-memory
|
||||
// status still updates. Logged because without the row
|
||||
// the cross-replica dedup guard and hydration cannot
|
||||
// see this operation at all.
|
||||
xlog.Warn("Failed to create gallery operation record", "op_id", op.ID, "error", err)
|
||||
}
|
||||
}
|
||||
err := runOpHandler(func() error { return g.backendHandler(op, systemState) })
|
||||
if err != nil {
|
||||
updateError(op.ID, err)
|
||||
} else if g.OnBackendOpCompleted != nil {
|
||||
// Let listeners (e.g. UpgradeChecker) refresh their view of
|
||||
// installed state. Run off the worker goroutine so a slow
|
||||
// callback doesn't stall the next queued operation.
|
||||
go g.OnBackendOpCompleted()
|
||||
}
|
||||
g.removeCancellation(op.ID)
|
||||
}
|
||||
|
||||
// SubscribeBroadcasts opens the wildcard subscriptions that keep this
|
||||
// replica's in-memory statuses + cancellation state in sync with peers.
|
||||
// Returns an error if the progress subscription fails; cancel-sub failures
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
package nodes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// backendOpLocks hands out one lock per (node, backend). It keeps a live
|
||||
// backend operation and the reconciler's retry of the same (node, backend)
|
||||
// from running at the same time. The zero value is ready to use.
|
||||
//
|
||||
// The locks are per process. Several frontends can still send to one node at
|
||||
// once; the worker is the last guard for that case.
|
||||
type backendOpLocks struct {
|
||||
mu sync.Mutex
|
||||
slots map[string]chan struct{}
|
||||
}
|
||||
|
||||
func backendOpLockKey(nodeID, backend string) string {
|
||||
return nodeID + "\x00" + backend
|
||||
}
|
||||
|
||||
func (l *backendOpLocks) slot(key string) chan struct{} {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.slots == nil {
|
||||
l.slots = map[string]chan struct{}{}
|
||||
}
|
||||
s, ok := l.slots[key]
|
||||
if !ok {
|
||||
s = make(chan struct{}, 1)
|
||||
l.slots[key] = s
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// lock waits for the (node, backend) lock or for ctx. The returned func
|
||||
// releases it. When ctx ends first, it returns ctx's error.
|
||||
func (l *backendOpLocks) lock(ctx context.Context, nodeID, backend string) (func(), error) {
|
||||
s := l.slot(backendOpLockKey(nodeID, backend))
|
||||
select {
|
||||
case s <- struct{}{}:
|
||||
return func() { <-s }, nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// tryLock takes the lock only if it is free.
|
||||
func (l *backendOpLocks) tryLock(nodeID, backend string) (func(), bool) {
|
||||
s := l.slot(backendOpLockKey(nodeID, backend))
|
||||
select {
|
||||
case s <- struct{}{}:
|
||||
return func() { <-s }, true
|
||||
default:
|
||||
return nil, false
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/mudler/LocalAI/core/config"
|
||||
"github.com/mudler/LocalAI/core/gallery"
|
||||
@@ -64,6 +65,8 @@ type DistributedBackendManager struct {
|
||||
backendGalleries []config.Gallery
|
||||
systemState *system.SystemState
|
||||
progressSink nodeProgressSink
|
||||
// maxConcurrent returns max_concurrent_backend_operations. nil means 1.
|
||||
maxConcurrent func() int
|
||||
}
|
||||
|
||||
// NewDistributedBackendManager creates a DistributedBackendManager.
|
||||
@@ -77,6 +80,7 @@ func NewDistributedBackendManager(appConfig *config.ApplicationConfig, ml *model
|
||||
backendGalleries: appConfig.BackendGalleries,
|
||||
systemState: appConfig.SystemState,
|
||||
progressSink: progressSink,
|
||||
maxConcurrent: appConfig.BackendOperationLimit,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -160,7 +164,9 @@ func (d *DistributedBackendManager) enqueueAndDrainBackendOp(ctx context.Context
|
||||
})
|
||||
}
|
||||
|
||||
result := BackendOpResult{Nodes: make([]NodeOpStatus, 0, len(allNodes))}
|
||||
// Pick the nodes first, in registry order, so the result keeps that order
|
||||
// whatever the send order is.
|
||||
targets := make([]BackendNode, 0, len(allNodes))
|
||||
for _, node := range allNodes {
|
||||
// Pending nodes haven't been approved yet - no intent to apply.
|
||||
if node.Status == StatusPending {
|
||||
@@ -176,78 +182,104 @@ func (d *DistributedBackendManager) enqueueAndDrainBackendOp(ctx context.Context
|
||||
if targetNodeIDs != nil && !targetNodeIDs[node.ID] {
|
||||
continue
|
||||
}
|
||||
if err := d.registry.UpsertPendingBackendOp(ctx, node.ID, backend, op, galleriesJSON); err != nil {
|
||||
xlog.Warn("Failed to enqueue backend op", "op", op, "node", node.Name, "backend", backend, "error", err)
|
||||
errMsg := fmt.Sprintf("enqueue failed: %v", err)
|
||||
result.Nodes = append(result.Nodes, NodeOpStatus{
|
||||
NodeID: node.ID, NodeName: node.Name, Status: galleryop.NodeStatusError,
|
||||
Error: errMsg,
|
||||
})
|
||||
emitNodeProgress(node, galleryop.NodeStatusError, errMsg)
|
||||
continue
|
||||
}
|
||||
|
||||
if node.Status != StatusHealthy {
|
||||
// Intent is recorded; reconciler will retry when the node recovers.
|
||||
errMsg := fmt.Sprintf("node %s, will retry when healthy", node.Status)
|
||||
result.Nodes = append(result.Nodes, NodeOpStatus{
|
||||
NodeID: node.ID, NodeName: node.Name, Status: galleryop.NodeStatusQueued,
|
||||
Error: errMsg,
|
||||
})
|
||||
emitNodeProgress(node, galleryop.NodeStatusQueued, errMsg)
|
||||
continue
|
||||
}
|
||||
|
||||
applyErr := apply(node)
|
||||
if applyErr == nil {
|
||||
// Find the row we just upserted and delete it; cheap but requires
|
||||
// a lookup since UpsertPendingBackendOp doesn't return the ID.
|
||||
if err := d.deletePendingRow(ctx, node.ID, backend, op); err != nil {
|
||||
xlog.Debug("Failed to clear pending backend op after success", "error", err)
|
||||
}
|
||||
result.Nodes = append(result.Nodes, NodeOpStatus{
|
||||
NodeID: node.ID, NodeName: node.Name, Status: galleryop.NodeStatusSuccess,
|
||||
})
|
||||
emitNodeProgress(node, galleryop.NodeStatusSuccess, "")
|
||||
continue
|
||||
}
|
||||
|
||||
// Record failure for backoff. If there is no route to the node, mark it
|
||||
// unhealthy so the router stops picking it too.
|
||||
errMsg := applyErr.Error()
|
||||
|
||||
// Worker-still-installing is a "soft" failure: the worker is most
|
||||
// likely still pulling the OCI image. Keep the row, push NextRetryAt
|
||||
// out so the reconciler does not immediately re-fire another install
|
||||
// while the worker is still busy, and report the in-progress state
|
||||
// to the caller. The next reconciler pass / backend.list confirms
|
||||
// the actual outcome.
|
||||
if errors.Is(applyErr, galleryop.ErrWorkerStillInstalling) {
|
||||
if id, err := d.findPendingRow(ctx, node.ID, backend, op); err == nil {
|
||||
_ = d.registry.RecordPendingBackendOpInFlight(ctx, id, errMsg, d.adapter.InstallTimeout())
|
||||
}
|
||||
result.Nodes = append(result.Nodes, NodeOpStatus{
|
||||
NodeID: node.ID, NodeName: node.Name, Status: galleryop.NodeStatusRunningOnWorker, Error: errMsg,
|
||||
})
|
||||
emitNodeProgress(node, galleryop.NodeStatusRunningOnWorker, errMsg)
|
||||
continue
|
||||
}
|
||||
|
||||
if errors.Is(applyErr, ErrNoRoute) {
|
||||
xlog.Warn("No route to node, marking unhealthy", "node", node.Name, "nodeID", node.ID)
|
||||
d.registry.MarkUnhealthy(ctx, node.ID)
|
||||
}
|
||||
if id, err := d.findPendingRow(ctx, node.ID, backend, op); err == nil {
|
||||
_ = d.registry.RecordPendingBackendOpFailure(ctx, id, errMsg)
|
||||
}
|
||||
result.Nodes = append(result.Nodes, NodeOpStatus{
|
||||
NodeID: node.ID, NodeName: node.Name, Status: galleryop.NodeStatusError, Error: errMsg,
|
||||
})
|
||||
emitNodeProgress(node, galleryop.NodeStatusError, errMsg)
|
||||
targets = append(targets, node)
|
||||
}
|
||||
|
||||
// Nodes are independent, so they are served in parallel, up to the
|
||||
// max_concurrent_backend_operations setting. With the default of 1 they are
|
||||
// visited one after the other, as before. Each result goes to its own slot.
|
||||
statuses := make([]NodeOpStatus, len(targets))
|
||||
limit := d.fanOutLimit()
|
||||
sem := make(chan struct{}, limit)
|
||||
var wg sync.WaitGroup
|
||||
for i, node := range targets {
|
||||
sem <- struct{}{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
statuses[i] = d.applyToNode(ctx, node, backend, op, galleriesJSON, apply, emitNodeProgress)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
result := BackendOpResult{Nodes: statuses}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// fanOutLimit is how many nodes one operation may talk to at once.
|
||||
func (d *DistributedBackendManager) fanOutLimit() int {
|
||||
if d.maxConcurrent != nil {
|
||||
if n := d.maxConcurrent(); n > 1 {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// applyToNode records the intent for one node and, when the node is healthy,
|
||||
// sends the operation and settles the pending row. It holds the (node,
|
||||
// backend) lock for the whole time so the reconciler does not send the same
|
||||
// operation in parallel.
|
||||
func (d *DistributedBackendManager) applyToNode(ctx context.Context, node BackendNode, backend, op string, galleriesJSON []byte, apply func(node BackendNode) error, emitNodeProgress func(node BackendNode, status, errMsg string)) NodeOpStatus {
|
||||
fail := func(status, errMsg string) NodeOpStatus {
|
||||
emitNodeProgress(node, status, errMsg)
|
||||
return NodeOpStatus{NodeID: node.ID, NodeName: node.Name, Status: status, Error: errMsg}
|
||||
}
|
||||
|
||||
unlock, lockErr := d.registry.backendOpLocks.lock(ctx, node.ID, backend)
|
||||
if lockErr != nil {
|
||||
return fail(galleryop.NodeStatusError, fmt.Sprintf("waiting for the node: %v", lockErr))
|
||||
}
|
||||
defer unlock()
|
||||
|
||||
if err := d.registry.UpsertPendingBackendOp(ctx, node.ID, backend, op, galleriesJSON); err != nil {
|
||||
xlog.Warn("Failed to enqueue backend op", "op", op, "node", node.Name, "backend", backend, "error", err)
|
||||
return fail(galleryop.NodeStatusError, fmt.Sprintf("enqueue failed: %v", err))
|
||||
}
|
||||
|
||||
if node.Status != StatusHealthy {
|
||||
// Intent is recorded; reconciler will retry when the node recovers.
|
||||
return fail(galleryop.NodeStatusQueued, fmt.Sprintf("node %s, will retry when healthy", node.Status))
|
||||
}
|
||||
|
||||
applyErr := apply(node)
|
||||
if applyErr == nil {
|
||||
// Find the row we just upserted and delete it; cheap but requires
|
||||
// a lookup since UpsertPendingBackendOp doesn't return the ID.
|
||||
if err := d.deletePendingRow(ctx, node.ID, backend, op); err != nil {
|
||||
xlog.Debug("Failed to clear pending backend op after success", "error", err)
|
||||
}
|
||||
return fail(galleryop.NodeStatusSuccess, "")
|
||||
}
|
||||
|
||||
// Record failure for backoff. If there is no route to the node, mark it
|
||||
// unhealthy so the router stops picking it too.
|
||||
errMsg := applyErr.Error()
|
||||
|
||||
// Worker-still-installing is a "soft" failure: the worker is most
|
||||
// likely still pulling the OCI image. Keep the row, push NextRetryAt
|
||||
// out so the reconciler does not immediately re-fire another install
|
||||
// while the worker is still busy, and report the in-progress state
|
||||
// to the caller. The next reconciler pass / backend.list confirms
|
||||
// the actual outcome.
|
||||
if errors.Is(applyErr, galleryop.ErrWorkerStillInstalling) {
|
||||
if id, err := d.findPendingRow(ctx, node.ID, backend, op); err == nil {
|
||||
_ = d.registry.RecordPendingBackendOpInFlight(ctx, id, errMsg, d.adapter.InstallTimeout())
|
||||
}
|
||||
return fail(galleryop.NodeStatusRunningOnWorker, errMsg)
|
||||
}
|
||||
|
||||
if errors.Is(applyErr, ErrNoRoute) {
|
||||
xlog.Warn("No route to node, marking unhealthy", "node", node.Name, "nodeID", node.ID)
|
||||
d.registry.MarkUnhealthy(ctx, node.ID)
|
||||
}
|
||||
if id, err := d.findPendingRow(ctx, node.ID, backend, op); err == nil {
|
||||
_ = d.registry.RecordPendingBackendOpFailure(ctx, id, errMsg)
|
||||
}
|
||||
return fail(galleryop.NodeStatusError, errMsg)
|
||||
}
|
||||
|
||||
// findPendingRow looks up the ID of a pending_backend_ops row by its
|
||||
// composite key. Used to hand off to RecordPendingBackendOpFailure /
|
||||
// DeletePendingBackendOp after UpsertPendingBackendOp upserts by the same
|
||||
|
||||
@@ -0,0 +1,246 @@
|
||||
package nodes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mudler/LocalAI/core/config"
|
||||
"github.com/mudler/LocalAI/core/services/galleryop"
|
||||
"github.com/mudler/LocalAI/core/services/messaging"
|
||||
"github.com/mudler/LocalAI/core/services/workerctl"
|
||||
"github.com/mudler/LocalAI/pkg/system"
|
||||
)
|
||||
|
||||
// gatedNATS answers backend install requests only after the test opens the
|
||||
// gate of the target node. It records how many requests are in flight, in
|
||||
// total and per node.
|
||||
type gatedNATS struct {
|
||||
*scriptedMessagingClient
|
||||
mu sync.Mutex
|
||||
gates map[string]chan struct{}
|
||||
arrived []string
|
||||
inFlight int
|
||||
maxInFlt int
|
||||
perNode map[string]int
|
||||
nodeClash bool
|
||||
allOpen bool
|
||||
requestsBy map[string]int
|
||||
nodeIDs []string
|
||||
}
|
||||
|
||||
func newGatedNATS() *gatedNATS {
|
||||
return &gatedNATS{
|
||||
scriptedMessagingClient: newScriptedMessagingClient(),
|
||||
gates: map[string]chan struct{}{},
|
||||
perNode: map[string]int{},
|
||||
requestsBy: map[string]int{},
|
||||
}
|
||||
}
|
||||
|
||||
func (g *gatedNATS) gate(nodeID string) chan struct{} {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
c, ok := g.gates[nodeID]
|
||||
if !ok {
|
||||
c = make(chan struct{})
|
||||
g.gates[nodeID] = c
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func (g *gatedNATS) open(nodeID string) { close(g.gate(nodeID)) }
|
||||
|
||||
func (g *gatedNATS) arrivals() []string {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return append([]string(nil), g.arrived...)
|
||||
}
|
||||
|
||||
func (g *gatedNATS) Request(subject string, _ []byte, _ time.Duration) ([]byte, error) {
|
||||
var nodeID string
|
||||
g.mu.Lock()
|
||||
for _, id := range g.nodeIDs {
|
||||
if subject == messaging.SubjectNodeBackendInstall(id) {
|
||||
nodeID = id
|
||||
}
|
||||
}
|
||||
g.mu.Unlock()
|
||||
gate := g.gate(nodeID)
|
||||
g.mu.Lock()
|
||||
g.arrived = append(g.arrived, nodeID)
|
||||
g.requestsBy[nodeID]++
|
||||
g.inFlight++
|
||||
g.perNode[nodeID]++
|
||||
if g.perNode[nodeID] > 1 {
|
||||
g.nodeClash = true
|
||||
}
|
||||
if g.inFlight > g.maxInFlt {
|
||||
g.maxInFlt = g.inFlight
|
||||
}
|
||||
g.mu.Unlock()
|
||||
|
||||
<-gate
|
||||
|
||||
g.mu.Lock()
|
||||
g.inFlight--
|
||||
g.perNode[nodeID]--
|
||||
g.mu.Unlock()
|
||||
return json.Marshal(workerctl.BackendInstallReply{Success: true})
|
||||
}
|
||||
|
||||
// Backend operations to several nodes at once. These specs use SQLite, so they
|
||||
// need no container.
|
||||
var _ = Describe("DistributedBackendManager fan-out", func() {
|
||||
var (
|
||||
registry *NodeRegistry
|
||||
nats *gatedNATS
|
||||
mgr *DistributedBackendManager
|
||||
ctx context.Context
|
||||
limit int
|
||||
nodes []*BackendNode
|
||||
)
|
||||
|
||||
register := func(name, address string) *BackendNode {
|
||||
n := &BackendNode{Name: name, NodeType: NodeTypeBackend, Address: address}
|
||||
Expect(registry.Register(ctx, n, true)).To(Succeed())
|
||||
got, err := registry.GetByName(ctx, name)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
nats.mu.Lock()
|
||||
nats.nodeIDs = append(nats.nodeIDs, got.ID)
|
||||
nats.mu.Unlock()
|
||||
return got
|
||||
}
|
||||
|
||||
BeforeEach(func() {
|
||||
ctx = context.Background()
|
||||
db, err := gorm.Open(sqlite.Open(filepath.Join(GinkgoT().TempDir(), "nodes.db")), &gorm.Config{})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
sqlDB, err := db.DB()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
registry, err = NewNodeRegistry(db)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
nats = newGatedNATS()
|
||||
limit = 1
|
||||
mgr = &DistributedBackendManager{
|
||||
local: stubLocalBackendManager{},
|
||||
adapter: NewRemoteUnloaderAdapter(nil, nats, 3*time.Minute, 15*time.Minute),
|
||||
registry: registry,
|
||||
systemState: &system.SystemState{},
|
||||
maxConcurrent: func() int { return limit },
|
||||
}
|
||||
nodes = []*BackendNode{register("node-a", "10.0.0.1:1"), register("node-b", "10.0.0.2:1"), register("node-c", "10.0.0.3:1")}
|
||||
})
|
||||
|
||||
install := func(name string) chan BackendOpResult {
|
||||
done := make(chan BackendOpResult, 1)
|
||||
go func() {
|
||||
defer GinkgoRecover()
|
||||
res, err := mgr.enqueueAndDrainBackendOp(ctx, "", OpBackendInstall, name, nil, nil, func(node BackendNode) error {
|
||||
reply, err := mgr.adapter.InstallBackend(node.ID, name, "", "", "", "", "", 0, "", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !reply.Success {
|
||||
return context.Canceled
|
||||
}
|
||||
return nil
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
done <- res
|
||||
}()
|
||||
return done
|
||||
}
|
||||
|
||||
It("visits the nodes one after the other with the default limit of 1", func() {
|
||||
done := install("llama-cpp")
|
||||
Eventually(nats.arrivals, "5s", "10ms").Should(HaveLen(1))
|
||||
Consistently(nats.arrivals, "300ms", "10ms").Should(HaveLen(1))
|
||||
|
||||
for i := range nodes {
|
||||
Eventually(nats.arrivals, "5s", "10ms").Should(HaveLen(i + 1))
|
||||
nats.open(nats.arrivals()[i])
|
||||
}
|
||||
var res BackendOpResult
|
||||
Eventually(done, "5s").Should(Receive(&res))
|
||||
Expect(res.Nodes).To(HaveLen(3))
|
||||
Expect(nats.maxInFlt).To(Equal(1))
|
||||
})
|
||||
|
||||
It("sends to the nodes in parallel up to the limit, and keeps the node order of the result", func() {
|
||||
limit = 2
|
||||
done := install("llama-cpp")
|
||||
Eventually(nats.arrivals, "5s", "10ms").Should(HaveLen(2))
|
||||
Consistently(nats.arrivals, "300ms", "10ms").Should(HaveLen(2))
|
||||
|
||||
// A slow node does not hold the others: finish the second one first.
|
||||
first, second := nats.arrivals()[0], nats.arrivals()[1]
|
||||
nats.open(second)
|
||||
Eventually(nats.arrivals, "5s", "10ms").Should(HaveLen(3))
|
||||
nats.open(nats.arrivals()[2])
|
||||
Consistently(done, "200ms").ShouldNot(Receive())
|
||||
nats.open(first)
|
||||
|
||||
var res BackendOpResult
|
||||
Eventually(done, "5s").Should(Receive(&res))
|
||||
Expect(res.Nodes).To(HaveLen(3))
|
||||
for i, n := range res.Nodes {
|
||||
Expect(n.Status).To(Equal(galleryop.NodeStatusSuccess))
|
||||
Expect(n.NodeID).To(Equal(nodes[i].ID), "results keep the order of the registry")
|
||||
}
|
||||
Expect(nats.maxInFlt).To(Equal(2))
|
||||
})
|
||||
|
||||
It("never runs two operations on one backend and node at once", func() {
|
||||
limit = 3
|
||||
first := install("llama-cpp")
|
||||
Eventually(nats.arrivals, "5s", "10ms").Should(HaveLen(3))
|
||||
second := install("llama-cpp")
|
||||
Consistently(nats.arrivals, "300ms", "10ms").Should(HaveLen(3), "the second operation waits for the first on every node")
|
||||
|
||||
for _, n := range nodes {
|
||||
nats.open(n.ID)
|
||||
}
|
||||
Eventually(first, "5s").Should(Receive())
|
||||
Eventually(second, "5s").Should(Receive())
|
||||
Expect(nats.nodeClash).To(BeFalse())
|
||||
for _, n := range nodes {
|
||||
Expect(nats.requestsBy[n.ID]).To(Equal(2))
|
||||
}
|
||||
})
|
||||
|
||||
It("lets the reconciler skip a (node, backend) that a live operation holds", func() {
|
||||
node := nodes[0]
|
||||
Expect(registry.UpsertPendingBackendOp(ctx, node.ID, "llama-cpp", OpBackendInstall, nil)).To(Succeed())
|
||||
rc := NewReplicaReconciler(ReplicaReconcilerOptions{Registry: registry, Adapter: mgr.adapter})
|
||||
|
||||
unlock, ok := registry.backendOpLocks.tryLock(node.ID, "llama-cpp")
|
||||
Expect(ok).To(BeTrue())
|
||||
rc.drainPendingBackendOps(ctx)
|
||||
Expect(nats.arrivals()).To(BeEmpty(), "the live operation owns the node and backend")
|
||||
|
||||
unlock()
|
||||
nats.open(node.ID)
|
||||
rc.drainPendingBackendOps(ctx)
|
||||
Expect(nats.arrivals()).To(Equal([]string{node.ID}))
|
||||
rows, err := registry.ListPendingBackendOps(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(rows).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("reads the limit from the application config when built from one", func() {
|
||||
cfg := config.NewApplicationConfig(config.WithMaxConcurrentBackendOperations(5))
|
||||
m := NewDistributedBackendManager(cfg, nil, nil, registry, nil)
|
||||
Expect(m.fanOutLimit()).To(Equal(5))
|
||||
Expect(NewDistributedBackendManager(config.NewApplicationConfig(), nil, nil, registry, nil).fanOutLimit()).To(Equal(1))
|
||||
})
|
||||
})
|
||||
@@ -332,91 +332,113 @@ func (rc *ReplicaReconciler) drainPendingBackendOps(ctx context.Context) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return
|
||||
}
|
||||
var applyErr error
|
||||
switch op.Op {
|
||||
case OpBackendDelete:
|
||||
_, applyErr = rc.adapter.DeleteBackend(op.NodeID, op.Backend)
|
||||
case OpBackendInstall:
|
||||
// Pending-op drain for admin install — not a per-replica load.
|
||||
// Replica 0 is the conventional admin slot. Install is idempotent:
|
||||
// the worker short-circuits if the backend is already running.
|
||||
reply, err := rc.adapter.InstallBackend(op.NodeID, op.Backend, "", string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if err != nil {
|
||||
applyErr = err
|
||||
} else if !reply.Success {
|
||||
applyErr = fmt.Errorf("install failed: %s", reply.Error)
|
||||
}
|
||||
case OpBackendUpgrade:
|
||||
// Pending-op drain for admin upgrade — fires backend.upgrade so
|
||||
// the slow re-pull doesn't head-of-line-block install traffic on
|
||||
// the same worker. Falls back to the legacy backend.install
|
||||
// Force=true path on ErrNoRoute for old workers that
|
||||
// don't subscribe to backend.upgrade yet (rolling-update window).
|
||||
// Reconciler retries are background reconciliation with no live
|
||||
// admin watching a progress bar, so opID/onProgress are empty —
|
||||
// the adapter skips the progress subscription entirely.
|
||||
reply, err := rc.adapter.UpgradeBackend(op.NodeID, op.Backend, string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrNoRoute) {
|
||||
instReply, instErr := rc.adapter.installWithForceFallback(op.NodeID, op.Backend, string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if instErr != nil {
|
||||
applyErr = instErr
|
||||
} else if !instReply.Success {
|
||||
applyErr = fmt.Errorf("upgrade (legacy fallback) failed: %s", instReply.Error)
|
||||
}
|
||||
} else {
|
||||
applyErr = err
|
||||
}
|
||||
} else if !reply.Success {
|
||||
applyErr = fmt.Errorf("upgrade failed: %s", reply.Error)
|
||||
}
|
||||
default:
|
||||
xlog.Warn("Reconciler: unknown pending op", "op", op.Op, "id", op.ID)
|
||||
// A live operation holds this (node, backend). Leave the row for the
|
||||
// next pass instead of sending the same operation twice at once.
|
||||
unlock, free := rc.registry.backendOpLocks.tryLock(op.NodeID, op.Backend)
|
||||
if !free {
|
||||
xlog.Debug("Reconciler: backend op in progress, skipping", "op", op.Op, "backend", op.Backend, "node", op.NodeID)
|
||||
continue
|
||||
}
|
||||
|
||||
if applyErr == nil {
|
||||
if err := rc.registry.DeletePendingBackendOp(ctx, op.ID); err != nil {
|
||||
xlog.Warn("Reconciler: failed to delete drained op row", "id", op.ID, "error", err)
|
||||
} else {
|
||||
xlog.Info("Reconciler: pending backend op applied",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID, "attempts", op.Attempts+1)
|
||||
func() {
|
||||
defer unlock()
|
||||
// The live operation may have settled this row while we waited
|
||||
// for the lock. Do not send an operation that already finished.
|
||||
var row PendingBackendOp
|
||||
if err := rc.registry.db.WithContext(ctx).First(&row, op.ID).Error; err != nil {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// ErrNoRoute means nothing is listening for this subject on the node.
|
||||
// Either its connection dropped, or it's the wrong node type
|
||||
// entirely. Mark unhealthy so the health monitor's
|
||||
// heartbeat-only pass doesn't immediately flip it back — and so
|
||||
// ListDuePendingBackendOps (which filters by status=healthy) stops
|
||||
// picking the row until the node genuinely recovers.
|
||||
if errors.Is(applyErr, ErrNoRoute) {
|
||||
xlog.Warn("Reconciler: no route to node, marking it unhealthy",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID)
|
||||
_ = rc.registry.MarkUnhealthy(ctx, op.NodeID)
|
||||
}
|
||||
|
||||
// Dead-letter cap: after maxAttempts the row is the reconciler
|
||||
// equivalent of a poison message. Delete it loudly so the queue
|
||||
// doesn't churn NATS every tick forever — operators can re-issue
|
||||
// the op from the UI if they still want it applied.
|
||||
if op.Attempts+1 >= maxPendingBackendOpAttempts {
|
||||
xlog.Error("Reconciler: abandoning pending backend op after max attempts",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID,
|
||||
"attempts", op.Attempts+1, "last_error", applyErr)
|
||||
if err := rc.registry.DeletePendingBackendOp(ctx, op.ID); err != nil {
|
||||
xlog.Warn("Reconciler: failed to delete abandoned op row", "id", op.ID, "error", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
_ = rc.registry.RecordPendingBackendOpFailure(ctx, op.ID, applyErr.Error())
|
||||
xlog.Warn("Reconciler: pending backend op retry failed",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID, "attempts", op.Attempts+1, "error", applyErr)
|
||||
rc.drainPendingBackendOp(ctx, op)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// drainPendingBackendOp applies one queued backend op and settles its row.
|
||||
// The caller holds the (node, backend) lock.
|
||||
func (rc *ReplicaReconciler) drainPendingBackendOp(ctx context.Context, op PendingBackendOp) {
|
||||
var applyErr error
|
||||
switch op.Op {
|
||||
case OpBackendDelete:
|
||||
_, applyErr = rc.adapter.DeleteBackend(op.NodeID, op.Backend)
|
||||
case OpBackendInstall:
|
||||
// Pending-op drain for admin install — not a per-replica load.
|
||||
// Replica 0 is the conventional admin slot. Install is idempotent:
|
||||
// the worker short-circuits if the backend is already running.
|
||||
reply, err := rc.adapter.InstallBackend(op.NodeID, op.Backend, "", string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if err != nil {
|
||||
applyErr = err
|
||||
} else if !reply.Success {
|
||||
applyErr = fmt.Errorf("install failed: %s", reply.Error)
|
||||
}
|
||||
case OpBackendUpgrade:
|
||||
// Pending-op drain for admin upgrade — fires backend.upgrade so
|
||||
// the slow re-pull doesn't head-of-line-block install traffic on
|
||||
// the same worker. Falls back to the legacy backend.install
|
||||
// Force=true path on ErrNoRoute for old workers that
|
||||
// don't subscribe to backend.upgrade yet (rolling-update window).
|
||||
// Reconciler retries are background reconciliation with no live
|
||||
// admin watching a progress bar, so opID/onProgress are empty —
|
||||
// the adapter skips the progress subscription entirely.
|
||||
reply, err := rc.adapter.UpgradeBackend(op.NodeID, op.Backend, string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrNoRoute) {
|
||||
instReply, instErr := rc.adapter.installWithForceFallback(op.NodeID, op.Backend, string(op.Galleries), "", "", "", 0, "", nil)
|
||||
if instErr != nil {
|
||||
applyErr = instErr
|
||||
} else if !instReply.Success {
|
||||
applyErr = fmt.Errorf("upgrade (legacy fallback) failed: %s", instReply.Error)
|
||||
}
|
||||
} else {
|
||||
applyErr = err
|
||||
}
|
||||
} else if !reply.Success {
|
||||
applyErr = fmt.Errorf("upgrade failed: %s", reply.Error)
|
||||
}
|
||||
default:
|
||||
xlog.Warn("Reconciler: unknown pending op", "op", op.Op, "id", op.ID)
|
||||
return
|
||||
}
|
||||
|
||||
if applyErr == nil {
|
||||
if err := rc.registry.DeletePendingBackendOp(ctx, op.ID); err != nil {
|
||||
xlog.Warn("Reconciler: failed to delete drained op row", "id", op.ID, "error", err)
|
||||
} else {
|
||||
xlog.Info("Reconciler: pending backend op applied",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID, "attempts", op.Attempts+1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// ErrNoRoute means nothing is listening for this subject on the node.
|
||||
// Either its connection dropped, or it's the wrong node type
|
||||
// entirely. Mark unhealthy so the health monitor's
|
||||
// heartbeat-only pass doesn't immediately flip it back — and so
|
||||
// ListDuePendingBackendOps (which filters by status=healthy) stops
|
||||
// picking the row until the node genuinely recovers.
|
||||
if errors.Is(applyErr, ErrNoRoute) {
|
||||
xlog.Warn("Reconciler: no route to node, marking it unhealthy",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID)
|
||||
_ = rc.registry.MarkUnhealthy(ctx, op.NodeID)
|
||||
}
|
||||
|
||||
// Dead-letter cap: after maxAttempts the row is the reconciler
|
||||
// equivalent of a poison message. Delete it loudly so the queue
|
||||
// doesn't churn NATS every tick forever — operators can re-issue
|
||||
// the op from the UI if they still want it applied.
|
||||
if op.Attempts+1 >= maxPendingBackendOpAttempts {
|
||||
xlog.Error("Reconciler: abandoning pending backend op after max attempts",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID,
|
||||
"attempts", op.Attempts+1, "last_error", applyErr)
|
||||
if err := rc.registry.DeletePendingBackendOp(ctx, op.ID); err != nil {
|
||||
xlog.Warn("Reconciler: failed to delete abandoned op row", "id", op.ID, "error", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
_ = rc.registry.RecordPendingBackendOpFailure(ctx, op.ID, applyErr.Error())
|
||||
xlog.Warn("Reconciler: pending backend op retry failed",
|
||||
"op", op.Op, "backend", op.Backend, "node", op.NodeID, "attempts", op.Attempts+1, "error", applyErr)
|
||||
}
|
||||
|
||||
// maxPendingBackendOpAttempts caps how many times the reconciler retries a
|
||||
// failing row before dead-lettering it. Ten attempts at exponential backoff
|
||||
// (30s → 15m cap) is >1h of wall-clock patience — well past any transient
|
||||
|
||||
@@ -399,6 +399,9 @@ type NodeRegistry struct {
|
||||
clock func() time.Time
|
||||
// leaseTTL overrides loadJobLeaseTTL when set (tests).
|
||||
leaseTTL time.Duration
|
||||
// backendOpLocks keeps live backend operations and the reconciler off the
|
||||
// same (node, backend) at the same time.
|
||||
backendOpLocks backendOpLocks
|
||||
// incarnations caches the last worker incarnation seen per node.
|
||||
incarnations sync.Map
|
||||
// replicaRemovedHooks are invoked after a replica row for (modelName, nodeID)
|
||||
|
||||
@@ -80,6 +80,7 @@ You can configure these settings via the web UI or through environment variables
|
||||
|
||||
- **Threads**: Number of threads used for parallel computation (recommended: number of physical cores)
|
||||
- **Context Size**: Default context size for models (default: `512`)
|
||||
- **Concurrent Backend Operations**: How many backend installs, upgrades and deletes run at once. `1` runs them one after the other (default: `1`, minimum `1`). Operations on the same backend never overlap, whatever the value. Model operations run in their own queue and are not held up by backend operations. In distributed mode the same limit bounds how many nodes receive one operation at the same time. The limit is per frontend. A change applies to operations that start after the change
|
||||
- **Artifact Download Concurrency**: Maximum number of artifact files downloaded at once. `1` downloads sequentially (default: `1`)
|
||||
- **F16**: Enable GPU acceleration using 16-bit floating point
|
||||
- **VRAM Budget**: Cap on VRAM used for model allocation (for example `80%` or `12GB`; empty means no cap). See [VRAM Management]({{%relref "advanced/vram-management" %}})
|
||||
@@ -171,6 +172,7 @@ The `runtime_settings.json` file follows this structure:
|
||||
"threads": 8,
|
||||
"context_size": 2048,
|
||||
"artifact_download_concurrency": 4,
|
||||
"max_concurrent_backend_operations": 2,
|
||||
"f16": false,
|
||||
"debug": false,
|
||||
"cors": true,
|
||||
|
||||
@@ -68,6 +68,7 @@ Application shutdown has no context parameter; work that ignores cancellation ca
|
||||
| `--addons-owner-address` | | `https://host:port` of the add-on owner carrier, with no path. Required for the `owner` and `frontend` roles | `$LOCALAI_ADDONS_OWNER_ADDRESS` |
|
||||
| `--addons-owner-listen` | | `owner` role only: the address the carrier listens on. Empty listens on all interfaces, on the port of `--addons-owner-address` | `$LOCALAI_ADDONS_OWNER_LISTEN` |
|
||||
| `--addons-carrier-dir` | | Directory with the carrier key set from `local-ai util addons-carrier-keys`. Required for the `owner` and `frontend` roles | `$LOCALAI_ADDONS_CARRIER_DIR` |
|
||||
| `--max-concurrent-backend-operations` | `1` | How many backend install, upgrade and delete operations run at once. `1` runs them one after the other. Operations on the same backend always run one at a time. In distributed mode it also bounds how many nodes one operation is sent to at the same time | `$LOCALAI_MAX_CONCURRENT_BACKEND_OPERATIONS` |
|
||||
| `--artifact-download-concurrency` | `1` | How many files of a model artifact to download at once. `1` downloads sequentially. Raising it helps artifacts split into many files on a fast link, at the cost of more concurrent load on the models volume. Whole files only — a single file is never split, so resume and per-file checksum verification are unaffected | `$LOCALAI_ARTIFACT_DOWNLOAD_CONCURRENCY` |
|
||||
|
||||
Backend processes receive a private scratch directory through `TMPDIR`, `TMP`,
|
||||
|
||||
+16
-8
@@ -1,30 +1,38 @@
|
||||
package utils
|
||||
|
||||
import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/mudler/xlog"
|
||||
)
|
||||
|
||||
var lastProgress time.Time = time.Now()
|
||||
var startTime time.Time = time.Now()
|
||||
// The timers are shared by every download progress callback. Gallery
|
||||
// operations can run concurrently, so they are stored as atomic Unix
|
||||
// nanoseconds.
|
||||
var lastProgress atomic.Int64
|
||||
var startTime atomic.Int64
|
||||
|
||||
func init() {
|
||||
ResetDownloadTimers()
|
||||
}
|
||||
|
||||
func ResetDownloadTimers() {
|
||||
lastProgress = time.Now()
|
||||
startTime = time.Now()
|
||||
now := time.Now().UnixNano()
|
||||
lastProgress.Store(now)
|
||||
startTime.Store(now)
|
||||
}
|
||||
|
||||
func DisplayDownloadFunction(fileName string, current string, total string, percentage float64) {
|
||||
currentTime := time.Now()
|
||||
|
||||
if currentTime.Sub(lastProgress) >= 5*time.Second {
|
||||
|
||||
lastProgress = currentTime
|
||||
last := lastProgress.Load()
|
||||
if currentTime.Sub(time.Unix(0, last)) >= 5*time.Second && lastProgress.CompareAndSwap(last, currentTime.UnixNano()) {
|
||||
|
||||
// calculate ETA based on percentage and elapsed time
|
||||
var eta time.Duration
|
||||
if percentage > 0 {
|
||||
elapsed := currentTime.Sub(startTime)
|
||||
elapsed := currentTime.Sub(time.Unix(0, startTime.Load()))
|
||||
eta = time.Duration(float64(elapsed)*(100/percentage) - float64(elapsed))
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user