diff --git a/weed/admin/plugin/plugin_monitor.go b/weed/admin/plugin/plugin_monitor.go index 9a5327ca18..c8a58e298c 100644 --- a/weed/admin/plugin/plugin_monitor.go +++ b/weed/admin/plugin/plugin_monitor.go @@ -887,6 +887,37 @@ func (r *Plugin) trackExecutionQueued(job *plugin_pb.JobSpec) { }) } +// bindQueuedJobWorker records the home worker while the job stays pending/queued. +func (r *Plugin) bindQueuedJobWorker(job *plugin_pb.JobSpec, workerID string) { + if job == nil || strings.TrimSpace(job.JobId) == "" || strings.TrimSpace(workerID) == "" { + return + } + + now := time.Now().UTC() + r.jobsMu.Lock() + tracked := r.jobs[job.JobId] + if tracked == nil { + r.jobsMu.Unlock() + return + } + if isTerminalTrackedJobState(tracked.State) { + r.jobsMu.Unlock() + return + } + tracked.WorkerID = workerID + tracked.Message = "queued on worker " + workerID + tracked.UpdatedAt = timeToPtr(now) + snap := cloneTrackedJob(*tracked) + r.dirtyJobs = true + r.jobsMu.Unlock() + + r.persistJobDetailSnapshot(job.JobId, func(detail *TrackedJob) { + detail.WorkerID = snap.WorkerID + detail.Message = snap.Message + detail.UpdatedAt = snap.UpdatedAt + }) +} + func (r *Plugin) cancelQueuedJob(job *plugin_pb.JobSpec, cause error) { reason := "job canceled" if cause != nil { diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 9871722a23..f1590c9dc7 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -31,13 +31,13 @@ const ( defaultScheduledPerWorkerConcurrency = 1 // Hold a job on replica-host workers until their reservation load // reaches this; 0 disables the pin and spills immediately. - defaultLocalQueuePin = 5 - maxScheduledExecutionConcurrency = 128 - defaultScheduledRetryBackoff = 5 * time.Second - defaultClusterContextTimeout = 10 * time.Second - defaultWaitingBacklogFloor = 8 - defaultWaitingBacklogMultiplier = 4 - maxEstimatedRuntimeCap = 8 * time.Hour + defaultLocalQueuePin = 5 + maxScheduledExecutionConcurrency = 128 + defaultScheduledRetryBackoff = 5 * time.Second + defaultClusterContextTimeout = 10 * time.Second + defaultWaitingBacklogFloor = 8 + defaultWaitingBacklogMultiplier = 4 + maxEstimatedRuntimeCap = 8 * time.Hour // How long started jobs may drain past the run window close. scheduledExecutionDrainGrace = 30 * time.Minute ) @@ -824,19 +824,22 @@ func (r *Plugin) dispatchScheduledProposals( // whole dispatch lets manual shell operations interleave between jobs. jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType)) - jobQueue := make(chan *plugin_pb.JobSpec, len(proposals)) + jobs := make([]*plugin_pb.JobSpec, 0, len(proposals)) for index, proposal := range proposals { - job := buildScheduledJobSpec(jobType, proposal, index) - r.trackExecutionQueued(job) select { case <-r.shutdownCh: - close(jobQueue) - return 0, 0, 0 + for _, job := range jobs { + r.cancelQueuedJob(job, errSchedulerShutdown) + } + return 0, 0, len(jobs) default: - jobQueue <- job } + job := buildScheduledJobSpec(jobType, proposal, index) + r.trackExecutionQueued(job) + jobs = append(jobs, job) } - close(jobQueue) + + queues := r.assignJobsToWorkerQueues(jobType, jobs, policy) var wg sync.WaitGroup var statsMu sync.Mutex @@ -853,130 +856,110 @@ func (r *Plugin) dispatchScheduledProposals( wg.Add(1) go func() { defer wg.Done() - - jobLoop: - for job := range jobQueue { + for { select { case <-r.shutdownCh: return default: } - if ctx.Err() != nil { - r.cancelQueuedJob(job, ctx.Err()) - statsMu.Lock() - canceledCount++ - statsMu.Unlock() + return + } + if queues.isEmpty() { + return + } + + item, targetID := r.takeQueuedJob(queues, jobType, policy) + if item == nil { + if queues.isEmpty() { + return + } + if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { + return + } + continue + } + if ctx.Err() != nil { + queues.pushFront(item) + return + } + + executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID) + if reserveErr != nil { + queues.pushFront(item) + if ctx.Err() != nil { + return + } + if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { + return + } continue } - for { - select { - case <-r.shutdownCh: - return - default: + var releaseJobLock func() + if jobsRequireLock { + var lockErr error + releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, item.job.JobId)) + if lockErr != nil { + release() + statsMu.Lock() + errorCount++ + statsMu.Unlock() + r.appendActivity(JobActivity{ + JobID: item.job.JobId, + JobType: item.job.JobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled execution lock failed: %v", lockErr), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + continue } if ctx.Err() != nil { - r.cancelQueuedJob(job, ctx.Err()) + releaseJobLock() + release() + queues.pushFront(item) + return + } + } + + err := r.executeScheduledJobWithExecutor(ctx, executor, item.job, clusterContext, policy) + if releaseJobLock != nil { + releaseJobLock() + } + release() + if errors.Is(err, errExecutorAtCapacity) { + queues.pushFront(item) + if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { + return + } + continue + } + if err != nil { + if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { + r.cancelQueuedJob(item.job, err) statsMu.Lock() canceledCount++ statsMu.Unlock() - continue jobLoop - } - - executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy, JobReplicaHosts(job)) - if reserveErr != nil { - if ctx.Err() != nil { - r.cancelQueuedJob(job, ctx.Err()) - statsMu.Lock() - canceledCount++ - statsMu.Unlock() - continue jobLoop - } - statsMu.Lock() - errorCount++ - statsMu.Unlock() - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled execution reservation failed: %v", reserveErr), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - break - } - - var releaseJobLock func() - if jobsRequireLock { - var lockErr error - releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, job.JobId)) - if lockErr != nil { - release() - statsMu.Lock() - errorCount++ - statsMu.Unlock() - r.appendActivity(JobActivity{ - JobID: job.JobId, - JobType: job.JobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled execution lock failed: %v", lockErr), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - break - } - // The lock may have been held by an operator for a - // while; re-check the execution window. - if ctx.Err() != nil { - releaseJobLock() - release() - r.cancelQueuedJob(job, ctx.Err()) - statsMu.Lock() - canceledCount++ - statsMu.Unlock() - continue jobLoop - } - } - - err := r.executeScheduledJobWithExecutor(ctx, executor, job, clusterContext, policy) - if releaseJobLock != nil { - releaseJobLock() - } - release() - if errors.Is(err, errExecutorAtCapacity) { - r.trackExecutionQueued(job) - if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { - return - } continue } - if err != nil { - if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { - r.cancelQueuedJob(job, err) - statsMu.Lock() - canceledCount++ - statsMu.Unlock() - continue jobLoop - } - statsMu.Lock() - errorCount++ - statsMu.Unlock() - r.appendActivity(JobActivity{ - JobID: job.JobId, - JobType: job.JobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled execution failed: %v", err), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - break - } - statsMu.Lock() - successCount++ + errorCount++ statsMu.Unlock() - break + r.appendActivity(JobActivity{ + JobID: item.job.JobId, + JobType: item.job.JobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled execution failed: %v", err), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + continue } + + statsMu.Lock() + successCount++ + statsMu.Unlock() } }() } @@ -987,8 +970,8 @@ func (r *Plugin) dispatchScheduledProposals( if drainErr == nil { drainErr = errSchedulerShutdown } - for job := range jobQueue { - r.cancelQueuedJob(job, drainErr) + for _, item := range queues.drain() { + r.cancelQueuedJob(item.job, drainErr) canceledCount++ } @@ -1047,15 +1030,6 @@ func (r *Plugin) reserveScheduledExecutor( if executor, release, ok := r.tryReserveFirst(pref, jobType, policy); ok { return executor, release, nil } - if policy.LocalQueuePin > 0 && loadOf(pref[0]) < policy.LocalQueuePin { - if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { - if ctx.Err() != nil { - return nil, nil, ctx.Err() - } - return nil, nil, fmt.Errorf("plugin is shutting down") - } - continue - } } sortExecutorsByLoad(rest, loadOf) @@ -1090,6 +1064,28 @@ func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType string, pol return nil, nil, false } +func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, workerID string) (*WorkerSession, func(), error) { + workerID = strings.TrimSpace(workerID) + if workerID == "" { + return nil, nil, fmt.Errorf("executor id is empty") + } + executors, err := r.registry.ListExecutors(jobType) + if err != nil { + return nil, nil, err + } + for _, executor := range executors { + if executor == nil || executor.WorkerID != workerID { + continue + } + release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy) + if !ok { + return nil, nil, errExecutorAtCapacity + } + return executor, release, nil + } + return nil, nil, fmt.Errorf("executor %s is not available for %s", workerID, jobType) +} + func (r *Plugin) executorReservationLoad(executor *WorkerSession, lane SchedulerLane) int { if executor == nil { return 0 diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index afa0e6b3b3..36f48f7e53 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -996,135 +996,6 @@ func TestLaneStatus_LifecycleNextDetectionSurvivesDefaultLanePrune(t *testing.T) } } -func TestReservePinsToPreferredWhileLoadBelowK(t *testing.T) { - t.Parallel() - - pluginSvc, err := New(Options{}) - if err != nil { - t.Fatalf("New: %v", err) - } - defer pluginSvc.Shutdown() - - pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ - WorkerId: "worker-local", - Address: "10.0.0.1:9334", - Capabilities: []*plugin_pb.JobTypeCapability{ - {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1}, - }, - }) - pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ - WorkerId: "worker-remote", - Address: "10.0.0.2:9334", - Capabilities: []*plugin_pb.JobTypeCapability{ - {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 4}, - }, - }) - - policy := schedulerPolicy{ - ExecutionTimeout: time.Second, - PerWorkerConcurrency: 1, - ExecutorReserveBackoff: 5 * time.Millisecond, - LocalQueuePin: 5, - } - preferred := []string{"10.0.0.1:8080"} - - first, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred) - if err != nil { - t.Fatalf("reserve local: %v", err) - } - defer release1() - if first.WorkerID != "worker-local" { - t.Fatalf("first reserve want worker-local, got %s", first.WorkerID) - } - - type reserveResult struct { - id string - err error - } - secondCh := make(chan reserveResult, 1) - go func() { - exec, release, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred) - id := "" - if exec != nil { - id = exec.WorkerID - } - if release != nil { - release() - } - secondCh <- reserveResult{id: id, err: reserveErr} - }() - - select { - case result := <-secondCh: - t.Fatalf("expected pin wait, got worker=%s err=%v", result.id, result.err) - case <-time.After(40 * time.Millisecond): - } - - release1() - - select { - case result := <-secondCh: - if result.err != nil { - t.Fatalf("second reserve after release: %v", result.err) - } - if result.id != "worker-local" { - t.Fatalf("after pin wait want worker-local, got %s", result.id) - } - case <-time.After(200 * time.Millisecond): - t.Fatalf("second reserve did not complete after local release") - } -} - -func TestReserveSpillsWhenPinDisabled(t *testing.T) { - t.Parallel() - - pluginSvc, err := New(Options{}) - if err != nil { - t.Fatalf("New: %v", err) - } - defer pluginSvc.Shutdown() - - pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ - WorkerId: "worker-local", - Address: "10.0.0.1:9334", - Capabilities: []*plugin_pb.JobTypeCapability{ - {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1}, - }, - }) - pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ - WorkerId: "worker-remote", - Address: "10.0.0.2:9334", - Capabilities: []*plugin_pb.JobTypeCapability{ - {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 4}, - }, - }) - - policy := schedulerPolicy{ - PerWorkerConcurrency: 1, - ExecutorReserveBackoff: time.Millisecond, - LocalQueuePin: 0, - } - preferred := []string{"10.0.0.1:8080"} - - first, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred) - if err != nil { - t.Fatalf("reserve local: %v", err) - } - defer release1() - if first.WorkerID != "worker-local" { - t.Fatalf("first reserve want worker-local, got %s", first.WorkerID) - } - - second, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "erasure_coding", policy, preferred) - if err != nil { - t.Fatalf("expected immediate spill, got %v", err) - } - defer release2() - if second.WorkerID != "worker-remote" { - t.Fatalf("K=0 should spill to worker-remote, got %s", second.WorkerID) - } -} - func TestReserveWaterfillsPreferredHosts(t *testing.T) { t.Parallel() diff --git a/weed/admin/plugin/worker_job_queues.go b/weed/admin/plugin/worker_job_queues.go new file mode 100644 index 0000000000..f34986e6dc --- /dev/null +++ b/weed/admin/plugin/worker_job_queues.go @@ -0,0 +1,235 @@ +package plugin + +import ( + "sync" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +// queuedJob is one scheduled spec parked on a home worker's admin-side queue. +type queuedJob struct { + job *plugin_pb.JobSpec + home string +} + +// workerJobQueues holds per-worker FIFOs for one dispatch batch. +// Home assignment is done once up front; run goroutines only pop/steal. +type workerJobQueues struct { + mu sync.Mutex + q map[string][]*queuedJob + leftover []*queuedJob + remaining int +} + +func newWorkerJobQueues() *workerJobQueues { + return &workerJobQueues{q: make(map[string][]*queuedJob)} +} + +func (qs *workerJobQueues) push(home string, job *plugin_pb.JobSpec) { + if qs == nil || job == nil { + return + } + item := &queuedJob{job: job, home: home} + qs.mu.Lock() + defer qs.mu.Unlock() + qs.pushLocked(item, false) +} + +func (qs *workerJobQueues) pushFront(item *queuedJob) { + if qs == nil || item == nil { + return + } + qs.mu.Lock() + defer qs.mu.Unlock() + qs.pushLocked(item, true) +} + +func (qs *workerJobQueues) pushLocked(item *queuedJob, front bool) { + qs.remaining++ + if item.home == "" { + if front { + qs.leftover = append([]*queuedJob{item}, qs.leftover...) + return + } + qs.leftover = append(qs.leftover, item) + return + } + if front { + qs.q[item.home] = append([]*queuedJob{item}, qs.q[item.home]...) + return + } + qs.q[item.home] = append(qs.q[item.home], item) +} + +func (qs *workerJobQueues) isEmpty() bool { + if qs == nil { + return true + } + qs.mu.Lock() + defer qs.mu.Unlock() + return qs.remaining == 0 +} + +func (qs *workerJobQueues) queueLen(workerID string) int { + if qs == nil { + return 0 + } + qs.mu.Lock() + defer qs.mu.Unlock() + return len(qs.q[workerID]) +} + +func (qs *workerJobQueues) drain() []*queuedJob { + if qs == nil { + return nil + } + qs.mu.Lock() + defer qs.mu.Unlock() + out := make([]*queuedJob, 0, qs.remaining) + for id, qq := range qs.q { + out = append(out, qq...) + qs.q[id] = nil + } + out = append(out, qs.leftover...) + qs.leftover = nil + qs.remaining = 0 + return out +} + +func popFront(qq []*queuedJob) (*queuedJob, []*queuedJob) { + item := qq[0] + if len(qq) == 1 { + return item, nil + } + return item, qq[1:] +} + +// assignJobsToWorkerQueues parks each job on the least-loaded replica-host +// worker (or any worker if none match). Single-threaded; no steal here. +func (r *Plugin) assignJobsToWorkerQueues( + jobType string, + jobs []*plugin_pb.JobSpec, + _ schedulerPolicy, +) *workerJobQueues { + qs := newWorkerJobQueues() + executors, err := r.registry.ListExecutors(jobType) + lane := JobTypeLane(jobType) + queued := make(map[string]int, len(executors)) + + for _, job := range jobs { + if job == nil { + continue + } + if err != nil || len(executors) == 0 { + qs.push("", job) + continue + } + pref, rest := partitionByPreferredHost(executors, JobReplicaHosts(job)) + cands := pref + if len(cands) == 0 { + cands = rest + } + sortExecutorsByLoad(cands, func(e *WorkerSession) int { + return r.executorReservationLoad(e, lane) + queued[e.WorkerID] + }) + home := cands[0].WorkerID + qs.push(home, job) + queued[home]++ + r.bindQueuedJobWorker(job, home) + } + return qs +} + +// takeQueuedJob pops a home-runnable job, or steals from a queue at/over K +// onto a free worker. Returns nil when nothing can start now. +func (r *Plugin) takeQueuedJob( + qs *workerJobQueues, + jobType string, + policy schedulerPolicy, +) (item *queuedJob, targetID string) { + if qs == nil { + return nil, "" + } + executors, err := r.registry.ListExecutors(jobType) + if err != nil || len(executors) == 0 { + return nil, "" + } + lane := JobTypeLane(jobType) + pin := policy.LocalQueuePin + + qs.mu.Lock() + defer qs.mu.Unlock() + if qs.remaining == 0 { + return nil, "" + } + + var bestHome *WorkerSession + bestLoad := 0 + for _, exec := range executors { + if len(qs.q[exec.WorkerID]) == 0 { + continue + } + limit := schedulerWorkerExecutionLimit(exec, jobType, policy) + if limit <= 0 { + continue + } + load := r.executorReservationLoad(exec, lane) + if load >= limit { + continue + } + if bestHome == nil || load < bestLoad || (load == bestLoad && exec.WorkerID < bestHome.WorkerID) { + bestHome = exec + bestLoad = load + } + } + if bestHome != nil { + id := bestHome.WorkerID + item, qs.q[id] = popFront(qs.q[id]) + qs.remaining-- + return item, id + } + + var thief *WorkerSession + for _, exec := range executors { + limit := schedulerWorkerExecutionLimit(exec, jobType, policy) + if limit <= 0 { + continue + } + if r.executorReservationLoad(exec, lane) >= limit { + continue + } + if thief == nil || exec.WorkerID < thief.WorkerID { + thief = exec + } + } + if thief == nil { + return nil, "" + } + + if len(qs.leftover) > 0 { + item, qs.leftover = popFront(qs.leftover) + qs.remaining-- + return item, thief.WorkerID + } + + victimID := "" + victimLen := 0 + for id, qq := range qs.q { + if len(qq) == 0 { + continue + } + if pin > 0 && len(qq) < pin { + continue + } + if victimID == "" || len(qq) > victimLen || (len(qq) == victimLen && id < victimID) { + victimID = id + victimLen = len(qq) + } + } + if victimID == "" { + return nil, "" + } + item, qs.q[victimID] = popFront(qs.q[victimID]) + qs.remaining-- + return item, thief.WorkerID +} diff --git a/weed/admin/plugin/worker_job_queues_test.go b/weed/admin/plugin/worker_job_queues_test.go new file mode 100644 index 0000000000..d1f7929249 --- /dev/null +++ b/weed/admin/plugin/worker_job_queues_test.go @@ -0,0 +1,231 @@ +package plugin + +import ( + "fmt" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +func replicaJob(id, source string) *plugin_pb.JobSpec { + return &plugin_pb.JobSpec{ + JobId: id, + JobType: "erasure_coding", + Summary: id, + DedupeKey: id, + Parameters: map[string]*plugin_pb.ConfigValue{ + "source_server": { + Kind: &plugin_pb.ConfigValue_StringValue{StringValue: source}, + }, + }, + } +} + +func registerECWorkers(t *testing.T, pluginSvc *Plugin, workers map[string]string) { + t.Helper() + for id, addr := range workers { + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: id, + Address: addr, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 8}, + }, + }) + } +} + +func TestAssignJobsWaterfillsPreferredWorkers(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + registerECWorkers(t, pluginSvc, map[string]string{ + "worker-a": "10.0.0.1:9334", + "worker-b": "10.0.0.2:9334", + "worker-c": "10.0.0.3:9334", + "worker-d": "10.0.0.4:9334", + }) + + jobs := []*plugin_pb.JobSpec{ + replicaJob("j1", "10.0.0.1:8080"), + replicaJob("j2", "10.0.0.1:8080"), + replicaJob("j3", "10.0.0.1:8080"), + replicaJob("j4", "10.0.0.1:8080"), + } + // All four replica IPs listed so every worker is preferred. + for _, job := range jobs { + job.Parameters["replica_servers"] = &plugin_pb.ConfigValue{ + Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{ + Values: []string{"10.0.0.1:8080", "10.0.0.2:8080", "10.0.0.3:8080", "10.0.0.4:8080"}, + }}, + } + } + + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, schedulerPolicy{LocalQueuePin: 5}) + if qs.queueLen("worker-a")+qs.queueLen("worker-b")+qs.queueLen("worker-c")+qs.queueLen("worker-d") != 4 { + t.Fatalf("expected 4 assigned jobs") + } + for _, id := range []string{"worker-a", "worker-b", "worker-c", "worker-d"} { + if qs.queueLen(id) != 1 { + t.Fatalf("want 1 job on %s, got %d", id, qs.queueLen(id)) + } + } +} + +func TestAssignAndTakeDoesNotStarveOtherHomes(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + registerECWorkers(t, pluginSvc, map[string]string{ + "worker-local": "10.0.0.1:9334", + "worker-remote": "10.0.0.2:9334", + }) + + jobs := []*plugin_pb.JobSpec{ + replicaJob("local-1", "10.0.0.1:8080"), + replicaJob("local-2", "10.0.0.1:8080"), + replicaJob("remote-1", "10.0.0.2:8080"), + } + policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5} + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) + if qs.queueLen("worker-local") != 2 || qs.queueLen("worker-remote") != 1 { + t.Fatalf("homes local=%d remote=%d", qs.queueLen("worker-local"), qs.queueLen("worker-remote")) + } + + _, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + _, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1) + if err != nil { + t.Fatalf("reserve 1: %v", err) + } + defer release1() + + item2, target2 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if item2 == nil || target2 != "worker-remote" { + t.Fatalf("second take should start remote home, got job=%v target=%s", item2, target2) + } +} + +func TestTakeStealsOnlyWhenHomeQueueReachesK(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + registerECWorkers(t, pluginSvc, map[string]string{ + "worker-local": "10.0.0.1:9334", + "worker-remote": "10.0.0.2:9334", + }) + + jobs := make([]*plugin_pb.JobSpec, 0, 6) + for i := 0; i < 6; i++ { + jobs = append(jobs, replicaJob(fmt.Sprintf("j%d", i), "10.0.0.1:8080")) + } + policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5} + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) + if qs.queueLen("worker-local") != 6 { + t.Fatalf("all jobs should home on local, got %d", qs.queueLen("worker-local")) + } + + _, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if target != "worker-local" { + t.Fatalf("first take want local, got %s", target) + } + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + if err != nil { + t.Fatalf("reserve local: %v", err) + } + defer release() + + // 5 still queued on local (>=K) and local has no slot → steal to remote. + stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if stolen == nil || stealTarget != "worker-remote" { + t.Fatalf("expected steal to remote, got job=%v target=%s", stolen, stealTarget) + } + if qs.queueLen("worker-local") != 4 { + t.Fatalf("after steal want 4 left on local, got %d", qs.queueLen("worker-local")) + } + + // 4 < K: do not steal again. + if item, tgt := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil { + t.Fatalf("pin should hold remaining jobs, got target=%s", tgt) + } +} + +func TestTakeStealsImmediatelyWhenPinDisabled(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + registerECWorkers(t, pluginSvc, map[string]string{ + "worker-local": "10.0.0.1:9334", + "worker-remote": "10.0.0.2:9334", + }) + + jobs := []*plugin_pb.JobSpec{ + replicaJob("a", "10.0.0.1:8080"), + replicaJob("b", "10.0.0.1:8080"), + } + policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 0} + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) + + _, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if target != "worker-local" { + t.Fatalf("first take want local, got %s", target) + } + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + if err != nil { + t.Fatalf("reserve local: %v", err) + } + defer release() + + stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if stolen == nil || stealTarget != "worker-remote" { + t.Fatalf("K=0 should steal immediately, got job=%v target=%s", stolen, stealTarget) + } +} + +func TestBindQueuedJobWorkerKeepsPendingState(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + job := replicaJob("bind-1", "10.0.0.1:8080") + pluginSvc.trackExecutionQueued(job) + pluginSvc.bindQueuedJobWorker(job, "worker-local") + + pluginSvc.jobsMu.Lock() + tracked := pluginSvc.jobs[job.JobId] + pluginSvc.jobsMu.Unlock() + if tracked == nil { + t.Fatal("missing tracked job") + } + if tracked.State != "job_state_pending" { + t.Fatalf("state=%s", tracked.State) + } + if tracked.Stage != "queued" { + t.Fatalf("stage=%s", tracked.Stage) + } + if tracked.WorkerID != "worker-local" { + t.Fatalf("worker=%s", tracked.WorkerID) + } +}