diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index 9ee003e81c..13db50ab62 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -79,16 +79,19 @@ type Plugin struct { lanes map[SchedulerLane]*schedulerLaneState - schedulerExecMu sync.Mutex - schedulerExecReservations map[string]int - adminScriptRunMu sync.RWMutex - schedulerDetectionMu sync.Mutex - schedulerDetection map[string]*schedulerDetectionInfo - schedulerRunMu sync.Mutex - schedulerRun map[string]*schedulerRunInfo - schedulerLoopMu sync.Mutex - schedulerLoopState schedulerLoopState - schedulerWakeCh chan struct{} + // Execution slots this admin holds on workers, one entry per job in + // flight; see executorLoadLocked for how they meet the heartbeat. + execResMu sync.Mutex + execRes map[uint64]execReservation + execResSeq uint64 + adminScriptRunMu sync.RWMutex + schedulerDetectionMu sync.Mutex + schedulerDetection map[string]*schedulerDetectionInfo + schedulerRunMu sync.Mutex + schedulerRun map[string]*schedulerRunInfo + schedulerLoopMu sync.Mutex + schedulerLoopState schedulerLoopState + schedulerWakeCh chan struct{} dedupeMu sync.Mutex recentDedupeByType map[string]map[string]time.Time @@ -115,6 +118,9 @@ type Plugin struct { // serialize stale job cleanup to avoid duplicate expirations staleJobsMu sync.Mutex + undeliveredCancelsMu sync.Mutex + undeliveredCancels map[string]undeliveredCancel + jobDetailsMu sync.Mutex activitiesMu sync.RWMutex @@ -216,7 +222,7 @@ func New(options Options) (*Plugin, error) { nextDetectionAt: make(map[string]time.Time), detectionInFlight: make(map[string]bool), detectorLeases: make(map[string]string), - schedulerExecReservations: make(map[string]int), + execRes: make(map[uint64]execReservation), schedulerDetection: make(map[string]*schedulerDetectionInfo), schedulerRun: make(map[string]*schedulerRunInfo), recentDedupeByType: make(map[string]map[string]time.Time), @@ -1034,9 +1040,9 @@ func (r *Plugin) clearDetectorLease(jobType string, workerID string) { delete(r.detectorLeases, jobType) } -func (r *Plugin) sendCancel(workerID, targetID string, kind plugin_pb.WorkKind, cause error) { +func (r *Plugin) sendCancel(workerID, targetID string, kind plugin_pb.WorkKind, cause error) error { if strings.TrimSpace(workerID) == "" || strings.TrimSpace(targetID) == "" { - return + return nil } requestID, err := newRequestID("cancel") @@ -1059,9 +1065,11 @@ func (r *Plugin) sendCancel(workerID, targetID string, kind plugin_pb.WorkKind, }, }, } - if err := r.sendToWorker(workerID, message); err != nil { + err = r.sendToWorker(workerID, message) + if err != nil { glog.V(1).Infof("Plugin failed to send cancel request to worker=%s target=%s: %v", workerID, targetID, err) } + return err } func (r *Plugin) sendAdminHello(session *streamSession) error { @@ -1148,6 +1156,7 @@ func (r *Plugin) handleWorkerMessage(session *streamSession, message *plugin_pb. if !r.updateRegistryFromSession(session, func() { r.registry.UpdateHeartbeat(workerID, body.Heartbeat) }) { return } + r.resendUndeliveredCancels(workerID, body.Heartbeat) case *plugin_pb.WorkerToAdminMessage_ConfigSchemaResponse: r.handleConfigSchemaResponse(body.ConfigSchemaResponse) case *plugin_pb.WorkerToAdminMessage_DetectionProposals: @@ -1440,6 +1449,12 @@ func (r *Plugin) handleJobCompleted(completed *plugin_pb.JobCompleted) { if completed.RequestId != "" { r.safeSendJobCompleted(completed.RequestId, completed) } + r.jobsMu.RLock() + ended := attemptEnded(r.jobs[completed.JobId], completed.RequestId) + r.jobsMu.RUnlock() + if ended { + return + } tracked := r.trackExecutionCompletion(completed) workerID := "" diff --git a/weed/admin/plugin/plugin_monitor.go b/weed/admin/plugin/plugin_monitor.go index c8a58e298c..eb360b639e 100644 --- a/weed/admin/plugin/plugin_monitor.go +++ b/weed/admin/plugin/plugin_monitor.go @@ -2,6 +2,7 @@ package plugin import ( "encoding/json" + "errors" "fmt" "sort" "strings" @@ -143,12 +144,14 @@ func (r *Plugin) ExpireJob(jobID, reason string) (*TrackedJob, bool, error) { var jobType string var requestID string + var workerID string active := false r.jobsMu.RLock() if tracked := r.jobs[normalizedJobID]; tracked != nil { jobType = tracked.JobType requestID = tracked.RequestID + workerID = tracked.WorkerID active = isActiveTrackedJobState(tracked.State) } r.jobsMu.RUnlock() @@ -163,6 +166,9 @@ func (r *Plugin) ExpireJob(jobID, reason string) (*TrackedJob, bool, error) { if requestID == "" { requestID = detail.RequestID } + if workerID == "" { + workerID = detail.WorkerID + } if !active && isActiveTrackedJobState(detail.State) { active = true } @@ -185,14 +191,12 @@ func (r *Plugin) ExpireJob(jobID, reason string) (*TrackedJob, bool, error) { } now := time.Now().UTC() - r.handleJobCompleted(&plugin_pb.JobCompleted{ - JobId: normalizedJobID, - JobType: jobType, - RequestId: requestID, - Success: false, - ErrorMessage: reason, - CompletedAt: timestamppb.New(now), - }) + r.expireActiveJob(activeJobSnapshot{ + jobID: normalizedJobID, + jobType: jobType, + workerID: workerID, + requestID: requestID, + }, reason, now) r.appendActivity(JobActivity{ JobID: normalizedJobID, JobType: jobType, @@ -235,20 +239,108 @@ func (r *Plugin) expireStaleJobs(now time.Time) int { continue } - r.handleJobCompleted(&plugin_pb.JobCompleted{ - JobId: snap.jobID, - JobType: snap.jobType, - RequestId: snap.requestID, - Success: false, - ErrorMessage: reason, - CompletedAt: timestamppb.New(now), - }) + r.expireActiveJob(snap, reason, now) expired++ } return expired } +// expireActiveJob fails the tracked job and tells its worker to stop it. +// Failing the record alone leaves the worker running the job and its slot +// taken, which is what kept a stuck job's slot out of reach for days. +func (r *Plugin) expireActiveJob(snap activeJobSnapshot, reason string, now time.Time) { + r.handleJobCompleted(&plugin_pb.JobCompleted{ + JobId: snap.jobID, + JobType: snap.jobType, + RequestId: snap.requestID, + Success: false, + ErrorMessage: reason, + CompletedAt: timestamppb.New(now), + }) + // A retry reuses the job id, so the request id is what names this attempt. + target := snap.requestID + if target == "" { + target = snap.jobID + } + if err := r.sendCancel(snap.workerID, target, plugin_pb.WorkKind_WORK_KIND_EXECUTION, errors.New(reason)); err != nil { + r.holdUndeliveredCancel(snap.jobID, undeliveredCancel{workerID: snap.workerID, target: target, reason: reason}) + } +} + +// undeliveredCancel is an expiry cancel its worker never received: it was +// disconnected, or its outgoing buffer stayed full past the send timeout. +type undeliveredCancel struct { + workerID string + target string + reason string +} + +// holdUndeliveredCancel keeps the cancel for the worker's next heartbeat. +// Entries whose job is no longer tracked are dropped here, so a worker that +// never comes back does not keep them forever. +func (r *Plugin) holdUndeliveredCancel(jobID string, pending undeliveredCancel) { + r.undeliveredCancelsMu.Lock() + defer r.undeliveredCancelsMu.Unlock() + if r.undeliveredCancels == nil { + r.undeliveredCancels = make(map[string]undeliveredCancel) + } + r.jobsMu.RLock() + for id := range r.undeliveredCancels { + if _, tracked := r.jobs[id]; !tracked { + delete(r.undeliveredCancels, id) + } + } + r.jobsMu.RUnlock() + r.undeliveredCancels[jobID] = pending +} + +// resendUndeliveredCancels sends a held cancel again while the worker's +// heartbeat still lists the job, and forgets it otherwise. A cancel that +// can only name the job id is skipped once the job is active again, since +// it would stop a new attempt too. +func (r *Plugin) resendUndeliveredCancels(workerID string, heartbeat *plugin_pb.WorkerHeartbeat) { + r.undeliveredCancelsMu.Lock() + var held map[string]undeliveredCancel + for jobID, pending := range r.undeliveredCancels { + if pending.workerID != workerID { + continue + } + delete(r.undeliveredCancels, jobID) + if held == nil { + held = make(map[string]undeliveredCancel) + } + held[jobID] = pending + } + r.undeliveredCancelsMu.Unlock() + if len(held) == 0 { + return + } + + for _, work := range heartbeat.GetRunningWork() { + jobID := strings.TrimSpace(work.GetWorkId()) + pending, ok := held[jobID] + if !ok || work.GetKind() != plugin_pb.WorkKind_WORK_KIND_EXECUTION { + continue + } + delete(held, jobID) + if pending.target == jobID { + r.jobsMu.RLock() + tracked := r.jobs[jobID] + active := tracked == nil || isActiveTrackedJobState(tracked.State) + r.jobsMu.RUnlock() + if active { + continue + } + } + go func() { + if err := r.sendCancel(workerID, pending.target, plugin_pb.WorkKind_WORK_KIND_EXECUTION, errors.New(pending.reason)); err != nil { + r.holdUndeliveredCancel(jobID, pending) + } + }() + } +} + func (r *Plugin) snapshotActiveJobs() []activeJobSnapshot { r.jobsMu.RLock() defer r.jobsMu.RUnlock() @@ -276,6 +368,13 @@ func (r *Plugin) snapshotActiveJobs() []activeJobSnapshot { return out } +// attemptEnded reports whether the job's record already closed the attempt +// requestID names, as an expiry does while the worker is still running it. +// That attempt's late reports must not reopen the record or log a second run. +func attemptEnded(job *TrackedJob, requestID string) bool { + return job != nil && requestID != "" && job.RequestID == requestID && !isActiveTrackedJobState(job.State) +} + func jobLastUpdated(job *TrackedJob) time.Time { if job == nil { return time.Time{} @@ -692,6 +791,10 @@ func (r *Plugin) handleJobProgressUpdate(workerID string, update *plugin_pb.JobP if strings.TrimSpace(update.JobId) != "" { r.jobsMu.Lock() job := r.jobs[update.JobId] + if attemptEnded(job, update.RequestId) { + r.jobsMu.Unlock() + return + } if job == nil { job = &TrackedJob{ JobID: update.JobId, diff --git a/weed/admin/plugin/plugin_monitor_test.go b/weed/admin/plugin/plugin_monitor_test.go index d44272bc1b..b4a6b82ca2 100644 --- a/weed/admin/plugin/plugin_monitor_test.go +++ b/weed/admin/plugin/plugin_monitor_test.go @@ -629,3 +629,275 @@ func TestBuildJobDetailLoadsFromDiskWhenMemoryCleared(t *testing.T) { t.Fatalf("unexpected disk detail payload: %q", detail.Job.Detail) } } + +func expectCancelFor(t *testing.T, session *streamSession, target string) { + t.Helper() + select { + case msg := <-session.outgoing: + cancel := msg.GetCancelRequest() + if cancel == nil || cancel.TargetId != target || cancel.TargetKind != plugin_pb.WorkKind_WORK_KIND_EXECUTION { + t.Fatalf("expected a cancel for %s, got %+v", target, msg) + } + case <-time.After(time.Second): + t.Fatalf("expected the worker to be told to stop %s", target) + } +} + +// Expiring a job reaches the worker too: writing a job off in the admin +// while the worker keeps running it would leave the slot taken. +func TestExpireJobCancelsTheJobOnItsWorker(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const jobID = "job-operator-expired" + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + pluginSvc.trackExecutionStart("req-operator-expired", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: "s3_lifecycle"}, 1) + + updated, expired, err := pluginSvc.ExpireJob(jobID, "") + if err != nil || !expired { + t.Fatalf("ExpireJob: expired=%v err=%v", expired, err) + } + if updated == nil || updated.State != StateFailed { + t.Fatalf("expected the job to be failed, got %+v", updated) + } + expectCancelFor(t, session, "req-operator-expired") +} + +// The stale sweep sends the same cancel, so the worker's slot comes back +// on its next heartbeat rather than staying taken by a job the admin +// already wrote off. +func TestExpireStaleJobsCancelsTheJobOnItsWorker(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const jobID = "job-stale" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{WorkerId: workerID}) + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsTotal: 4}) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + pluginSvc.trackExecutionStart("req-stale", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: "s3_lifecycle"}, 1) + + startedAt := time.Now().UTC() + if n := pluginSvc.expireStaleJobs(startedAt.Add(defaultStaleActiveJobTimeout - time.Minute)); n != 0 { + t.Fatalf("expired %d jobs before the stale limit", n) + } + if n := pluginSvc.expireStaleJobs(startedAt.Add(defaultStaleActiveJobTimeout + time.Minute)); n != 1 { + t.Fatalf("expired %d jobs past the stale limit, want 1", n) + } + got, _ := pluginSvc.GetTrackedJob(jobID) + if got == nil || got.State != StateFailed { + t.Fatalf("expected the stale job to be failed, got %+v", got) + } + expectCancelFor(t, session, "req-stale") +} + +func expectNoCancel(t *testing.T, session *streamSession) { + t.Helper() + select { + case msg := <-session.outgoing: + t.Fatalf("expected no message, got %+v", msg) + case <-time.After(100 * time.Millisecond): + } +} + +func heartbeatListing(workerID string, jobIDs ...string) *plugin_pb.WorkerToAdminMessage { + heartbeat := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsTotal: 4} + for _, jobID := range jobIDs { + heartbeat.RunningWork = append(heartbeat.RunningWork, &plugin_pb.RunningWork{ + WorkId: jobID, + Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, + JobType: "s3_lifecycle", + State: plugin_pb.JobState_JOB_STATE_RUNNING, + }) + } + return &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: heartbeat}} +} + +// A cancel the worker could not receive when its job expired goes out with +// the worker's next heartbeat that still lists the job, once. +func TestExpireJobResendsAnUndeliveredCancelOnHeartbeat(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const jobID = "job-expired-while-away" + pluginSvc.trackExecutionStart("req-away", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: "s3_lifecycle"}, 1) + if _, expired, err := pluginSvc.ExpireJob(jobID, ""); err != nil || !expired { + t.Fatalf("ExpireJob: expired=%v err=%v", expired, err) + } + + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{WorkerId: workerID}) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + + pluginSvc.handleWorkerMessage(session, heartbeatListing(workerID, jobID)) + expectCancelFor(t, session, "req-away") + + pluginSvc.handleWorkerMessage(session, heartbeatListing(workerID, jobID)) + expectNoCancel(t, session) +} + +// A held cancel names the expired attempt, so it neither stops a retry +// that reuses the job id nor gets lost when the expired attempt's late +// progress marks the job running again. +func TestUndeliveredCancelNamesTheExpiredAttempt(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const retriedJobID = "job-retried" + const revivedJobID = "job-revived" + const stoppedJobID = "job-stopped" + pluginSvc.trackExecutionStart("req-retried-1", workerID, &plugin_pb.JobSpec{JobId: retriedJobID, JobType: "s3_lifecycle"}, 1) + pluginSvc.trackExecutionStart("req-revived", workerID, &plugin_pb.JobSpec{JobId: revivedJobID, JobType: "s3_lifecycle"}, 1) + pluginSvc.trackExecutionStart("req-stopped", workerID, &plugin_pb.JobSpec{JobId: stoppedJobID, JobType: "s3_lifecycle"}, 1) + for _, jobID := range []string{retriedJobID, revivedJobID, stoppedJobID} { + if _, expired, err := pluginSvc.ExpireJob(jobID, ""); err != nil || !expired { + t.Fatalf("ExpireJob(%s): expired=%v err=%v", jobID, expired, err) + } + } + pluginSvc.trackExecutionStart("req-retried-2", workerID, &plugin_pb.JobSpec{JobId: retriedJobID, JobType: "s3_lifecycle"}, 2) + pluginSvc.handleJobProgressUpdate(workerID, &plugin_pb.JobProgressUpdate{ + JobId: revivedJobID, + JobType: "s3_lifecycle", + RequestId: "req-revived", + State: plugin_pb.JobState_JOB_STATE_RUNNING, + }) + + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{WorkerId: workerID}) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + + pluginSvc.handleWorkerMessage(session, heartbeatListing(workerID, retriedJobID, revivedJobID)) + targets := map[string]bool{} + for range 2 { + select { + case msg := <-session.outgoing: + targets[msg.GetCancelRequest().GetTargetId()] = true + case <-time.After(time.Second): + t.Fatalf("expected two cancels, got %v", targets) + } + } + if !targets["req-retried-1"] || !targets["req-revived"] { + t.Fatalf("expected cancels for the expired attempts, got %v", targets) + } + + pluginSvc.handleWorkerMessage(session, heartbeatListing(workerID, stoppedJobID)) + expectNoCancel(t, session) +} + +// Without a request id the cancel can only name the job, so it is held +// back once the job is active again rather than stopping the new attempt. +func TestUndeliveredCancelByJobIDSkipsAnActiveJob(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const jobID = "job-without-request" + pluginSvc.trackExecutionStart("", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: "s3_lifecycle"}, 1) + if _, expired, err := pluginSvc.ExpireJob(jobID, ""); err != nil || !expired { + t.Fatalf("ExpireJob: expired=%v err=%v", expired, err) + } + pluginSvc.trackExecutionStart("", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: "s3_lifecycle"}, 2) + + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{WorkerId: workerID}) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + + pluginSvc.handleWorkerMessage(session, heartbeatListing(workerID, jobID)) + expectNoCancel(t, session) +} + +// Once the admin expires an attempt, that attempt's late progress and its +// completion from the cancel neither reopen the record nor log a second run, +// while a retry under a new request id is still recorded. +func TestExpiredAttemptIgnoresItsLateReports(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + const jobID = "job-late-reports" + const jobType = "s3_lifecycle" + pluginSvc.trackExecutionStart("req-late-1", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: jobType}, 1) + if _, expired, err := pluginSvc.ExpireJob(jobID, "stuck"); err != nil || !expired { + t.Fatalf("ExpireJob: expired=%v err=%v", expired, err) + } + + pluginSvc.handleJobProgressUpdate(workerID, &plugin_pb.JobProgressUpdate{ + JobId: jobID, + JobType: jobType, + RequestId: "req-late-1", + State: plugin_pb.JobState_JOB_STATE_RUNNING, + Stage: "copying", + }) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + JobId: jobID, + JobType: jobType, + RequestId: "req-late-1", + ErrorMessage: "context canceled", + }) + + got, _ := pluginSvc.GetTrackedJob(jobID) + if got == nil || got.State != StateFailed || got.ErrorMessage != "stuck" { + t.Fatalf("expected the expiry to stand, got %+v", got) + } + history, err := pluginSvc.LoadRunHistory(jobType) + if err != nil { + t.Fatalf("LoadRunHistory: %v", err) + } + if len(history.ErrorRuns) != 1 || len(history.SuccessfulRuns) != 0 { + t.Fatalf("expected the expiry's run record only, got %+v", history) + } + + pluginSvc.trackExecutionStart("req-late-2", workerID, &plugin_pb.JobSpec{JobId: jobID, JobType: jobType}, 2) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + JobId: jobID, + JobType: jobType, + RequestId: "req-late-2", + Success: true, + }) + got, _ = pluginSvc.GetTrackedJob(jobID) + if got == nil || got.State != StateSucceeded { + t.Fatalf("expected the retry to complete, got %+v", got) + } + history, err = pluginSvc.LoadRunHistory(jobType) + if err != nil { + t.Fatalf("LoadRunHistory: %v", err) + } + if len(history.ErrorRuns) != 1 || len(history.SuccessfulRuns) != 1 { + t.Fatalf("expected the retry's run to be recorded, got %+v", history) + } +} diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index af62a5d788..ceed99b7e6 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -36,6 +36,7 @@ const ( // Wait this long after a home worker disappears before stealing // its queue; 0 steals immediately. Covers a reconnect blink. defaultOrphanRehomeGrace = 30 * time.Second + completedJobEchoGrace = 30 * time.Second maxScheduledExecutionConcurrency = 128 defaultScheduledRetryBackoff = 5 * time.Second defaultClusterContextTimeout = 10 * time.Second @@ -888,6 +889,7 @@ func (r *Plugin) dispatchScheduledProposals( wg.Add(1) go func() { defer wg.Done() + var blockedSince time.Time for { select { case <-r.shutdownCh: @@ -906,6 +908,16 @@ func (r *Plugin) dispatchScheduledProposals( if queues.isEmpty() { return } + if blockedSince.IsZero() { + blockedSince = time.Now() + } + if wait, reason := r.executorCapacityOutlook(jobType, policy, time.Since(blockedSince)); !wait { + given := r.giveUpQueuedJobs(queues, jobType, reason) + statsMu.Lock() + canceledCount += given + statsMu.Unlock() + return + } if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { return } @@ -916,17 +928,21 @@ func (r *Plugin) dispatchScheduledProposals( return } - executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID) + executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID, item.job.JobId) if reserveErr != nil { queues.pushFront(item) if ctx.Err() != nil { return } + if blockedSince.IsZero() { + blockedSince = time.Now() + } if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { return } continue } + blockedSince = time.Time{} var releaseJobLock func() if jobsRequireLock { @@ -1010,6 +1026,76 @@ func (r *Plugin) dispatchScheduledProposals( return successCount, errorCount, canceledCount } +// executorCapacityOutlook decides whether a dispatch goroutine that found +// no executor for jobType should keep waiting. Capacity held by this lane's +// own jobs comes back when they complete, so that is worth the wait. +// Capacity held by other lanes or by work this admin never dispatched +// clears on a schedule the lane has no view of, and a lane that serialises +// its job types would idle every other type in it for the whole runtime +// budget; those jobs are given up now, with the reason, and re-proposed by +// the next detection. A missing worker, or a free one that the queue's pin +// keeps off a busy home, is waited for only as long as the orphan grace. +func (r *Plugin) executorCapacityOutlook(jobType string, policy schedulerPolicy, blockedFor time.Duration) (bool, string) { + executors, err := r.registry.ListExecutors(jobType) + if err != nil || len(executors) == 0 { + if blockedFor < policy.OrphanRehomeGrace { + return true, "" + } + return false, fmt.Sprintf("no executor worker available for %s", jobType) + } + + lane := JobTypeLane(jobType) + held := make([]string, 0, len(executors)) + free := false + r.execResMu.Lock() + defer r.execResMu.Unlock() + for _, executor := range executors { + if executor == nil { + continue + } + fits, blocking, limit := r.executorCapacityLocked(executor, jobType, policy) + if fits { + free = true + continue + } + if blocking.reservedLanes[lane] > 0 { + return true, "" + } + if limit <= 0 { + held = append(held, fmt.Sprintf("worker %s accepts no %s jobs", executor.WorkerID, jobType)) + continue + } + held = append(held, fmt.Sprintf("worker %s has %d/%d slots used by %s", + executor.WorkerID, blocking.total(), limit, blocking.holders())) + } + if free && (len(held) == 0 || blockedFor < policy.OrphanRehomeGrace) { + return true, "" + } + return false, fmt.Sprintf("no executor capacity for %s: %s", jobType, strings.Join(held, "; ")) +} + +// giveUpQueuedJobs cancels whatever is still queued for jobType with the +// reason no executor can take it, so the lane moves on and the next +// detection proposes the work again. Returns how many jobs it canceled. +func (r *Plugin) giveUpQueuedJobs(queues *workerJobQueues, jobType, reason string) int { + items := queues.drain() + if len(items) == 0 { + return 0 + } + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: reason, + Stage: "skipped_no_executor", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + cause := errors.New(reason) + for _, item := range items { + r.cancelQueuedJob(item.job, cause) + } + return len(items) +} + func (r *Plugin) reserveScheduledExecutor( ctx context.Context, jobType string, @@ -1054,18 +1140,16 @@ func (r *Plugin) reserveScheduledExecutor( } pref, rest := partitionByPreferredHost(executors, preferredHosts) - lane := JobTypeLane(jobType) - loadOf := func(e *WorkerSession) int { return r.executorReservationLoad(e, lane) } if len(pref) > 0 { - sortExecutorsByLoad(pref, loadOf) - if executor, release, ok := r.tryReserveFirst(pref, jobType, policy); ok { + sortExecutorsByLoad(pref, r.executorWorkerLoad) + if executor, release, ok := r.tryReserveFirst(pref, jobType, "", policy); ok { return executor, release, nil } } - sortExecutorsByLoad(rest, loadOf) - if executor, release, ok := r.tryReserveFirst(rest, jobType, policy); ok { + sortExecutorsByLoad(rest, r.executorWorkerLoad) + if executor, release, ok := r.tryReserveFirst(rest, jobType, "", policy); ok { return executor, release, nil } @@ -1078,17 +1162,9 @@ func (r *Plugin) reserveScheduledExecutor( } } -func (r *Plugin) tryReserveExecutorCapacity( - executor *WorkerSession, - jobType string, - policy schedulerPolicy, -) (func(), bool) { - return r.tryReserveExecutorCapacityForLane(executor, jobType, policy, JobTypeLane(jobType)) -} - -func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType string, policy schedulerPolicy) (*WorkerSession, func(), bool) { +func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType, jobID string, policy schedulerPolicy) (*WorkerSession, func(), bool) { for _, executor := range executors { - release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy) + release, ok := r.reserveExecutorSlot(executor, jobType, jobID, policy) if ok { return executor, release, true } @@ -1096,7 +1172,7 @@ 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) { +func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, workerID, jobID string) (*WorkerSession, func(), error) { workerID = strings.TrimSpace(workerID) if workerID == "" { return nil, nil, fmt.Errorf("executor id is empty") @@ -1109,7 +1185,7 @@ func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, wor if executor == nil || executor.WorkerID != workerID { continue } - release, ok := r.tryReserveExecutorCapacity(executor, jobType, policy) + release, ok := r.reserveExecutorSlot(executor, jobType, jobID, policy) if !ok { return nil, nil, errExecutorAtCapacity } @@ -1118,27 +1194,6 @@ func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, wor 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 - } - load := 0 - if executor.Heartbeat != nil && executor.Heartbeat.ExecutionSlotsUsed > 0 { - load = int(executor.Heartbeat.ExecutionSlotsUsed) - } - workerID := strings.TrimSpace(executor.WorkerID) - if ls := r.lanes[lane]; ls != nil { - ls.execMu.Lock() - load += ls.execRes[workerID] - ls.execMu.Unlock() - return load - } - r.schedulerExecMu.Lock() - load += r.schedulerExecReservations[workerID] - r.schedulerExecMu.Unlock() - return load -} - func sortExecutorsByLoad(executors []*WorkerSession, load func(*WorkerSession) int) { sort.SliceStable(executors, func(i, j int) bool { if executors[i] == nil { @@ -1155,100 +1210,264 @@ func sortExecutorsByLoad(executors []*WorkerSession, load func(*WorkerSession) i }) } -// tryReserveExecutorCapacityForLane reserves an execution slot on the -// per-lane reservation pool so that lanes cannot starve each other. -func (r *Plugin) tryReserveExecutorCapacityForLane( +// execReservation is an execution slot this admin holds on a worker for one +// job, from the moment the job is picked until its completion is handled. +// The job id is what tells a slot the admin already counts from one only the +// worker's heartbeat knows about. The worker reports the completion before +// the heartbeat that frees the slot, so a released reservation is kept for +// completedJobEchoGrace to claim that echo. +type execReservation struct { + workerID string + jobType string + jobID string + releasedAt time.Time +} + +// executorLoad is how many execution slots on one worker the admin must +// treat as taken. reserved is this admin's own reservations, including the +// echo of a job it just completed; foreign is work the worker reports that +// no reservation covers: manual runs and work still running from before a +// reconnect. +type executorLoad struct { + reserved int + foreign int + unitemized int + // The lane of each reservation and the job types behind the whole load, + // so a blocked dispatch can tell whether the load is its own lane's and + // name what holds the worker when it is not. + reservedLanes map[SchedulerLane]int + byType map[string]int +} + +func (l executorLoad) total() int { + return l.reserved + l.foreign +} + +// holders names the job types behind the load, busiest first. +func (l executorLoad) holders() string { + if len(l.byType) == 0 { + return "work the worker did not itemize" + } + types := make([]string, 0, len(l.byType)) + for jobType := range l.byType { + types = append(types, jobType) + } + sort.Slice(types, func(i, j int) bool { + if l.byType[types[i]] != l.byType[types[j]] { + return l.byType[types[i]] > l.byType[types[j]] + } + return types[i] < types[j] + }) + parts := make([]string, 0, len(types)+1) + add := func(holder string, n int) { + if n > 1 { + holder = fmt.Sprintf("%s x%d", holder, n) + } + parts = append(parts, holder) + } + for _, jobType := range types { + add(jobType, l.byType[jobType]) + } + if l.unitemized > 0 { + add("unitemized work", l.unitemized) + } + out := strings.Join(parts, ", ") + if l.reserved == 0 { + out += " this admin did not dispatch" + } + return out +} + +// executorLoadLocked tallies the worker's slots as the admin sees them: +// jobType "" counts work of every type, otherwise only that type. A +// heartbeat entry whose job id matches a reservation is the same slot seen +// twice and counts once. Slots the worker counts without itemizing them go +// first to this admin's jobs it does not list, a completed one only until the +// worker's next heartbeat; the rest are of unknown type and count against +// every type. Caller holds r.execResMu. +func (r *Plugin) executorLoadLocked(executor *WorkerSession, jobType string) executorLoad { + load := executorLoad{ + reservedLanes: make(map[SchedulerLane]int), + byType: make(map[string]int), + } + if executor == nil { + return load + } + countReserved := func(res execReservation) { + if jobType != "" && res.jobType != jobType { + return + } + load.reserved++ + load.reservedLanes[JobTypeLane(res.jobType)]++ + load.byType[res.jobType]++ + } + running := executor.Heartbeat.GetRunningWork() + listed := make(map[string]struct{}, len(running)) + unitemized := int(executor.Heartbeat.GetExecutionSlotsUsed()) + for _, work := range running { + if work.GetKind() == plugin_pb.WorkKind_WORK_KIND_EXECUTION { + listed[work.GetWorkId()] = struct{}{} + unitemized-- + } + } + workerID := strings.TrimSpace(executor.WorkerID) + claimed := make(map[string]struct{}) + var unlistedEchoes []execReservation + now := time.Now() + for token, res := range r.execRes { + released := !res.releasedAt.IsZero() + if released && now.Sub(res.releasedAt) > completedJobEchoGrace { + delete(r.execRes, token) + continue + } + if res.workerID != workerID { + continue + } + isListed := false + if res.jobID != "" { + claimed[res.jobID] = struct{}{} + _, isListed = listed[res.jobID] + } + switch { + case !released: + if !isListed { + unitemized-- + } + countReserved(res) + case isListed: + countReserved(res) + case executor.LastSeenAt.Before(res.releasedAt): + unlistedEchoes = append(unlistedEchoes, res) + } + } + sort.Slice(unlistedEchoes, func(i, j int) bool { + return unlistedEchoes[i].releasedAt.After(unlistedEchoes[j].releasedAt) + }) + for _, res := range unlistedEchoes { + if unitemized <= 0 { + break + } + unitemized-- + countReserved(res) + } + if unitemized > 0 { + load.foreign += unitemized + load.unitemized = unitemized + } + for _, work := range running { + if work.GetKind() != plugin_pb.WorkKind_WORK_KIND_EXECUTION { + continue + } + if _, ok := claimed[work.GetWorkId()]; ok { + continue + } + if jobType != "" && work.GetJobType() != jobType { + continue + } + load.foreign++ + load.byType[work.GetJobType()]++ + } + return load +} + +// executorCapacityLocked reports whether one more jobType job fits on the +// worker. per_worker_execution_concurrency is a per-job-type limit, so it is +// checked against jobType work alone; the worker's advertised slot total is +// checked against work of every type. When the job does not fit, the load +// and the limit that refused it come back for the caller to act on or +// report. Caller holds r.execResMu. +func (r *Plugin) executorCapacityLocked(executor *WorkerSession, jobType string, policy schedulerPolicy) (bool, executorLoad, int) { + limit := schedulerWorkerExecutionLimit(executor, jobType, policy) + if limit <= 0 { + return false, executorLoad{}, 0 + } + if load := r.executorLoadLocked(executor, jobType); load.total() >= limit { + return false, load, limit + } + if total := int(executor.Heartbeat.GetExecutionSlotsTotal()); total > 0 { + if load := r.executorLoadLocked(executor, ""); load.total() >= total { + return false, load, total + } + } + return true, executorLoad{}, limit +} + +func (r *Plugin) executorHasCapacity(executor *WorkerSession, jobType string, policy schedulerPolicy) bool { + if executor == nil || strings.TrimSpace(executor.WorkerID) == "" { + return false + } + r.execResMu.Lock() + defer r.execResMu.Unlock() + fits, _, _ := r.executorCapacityLocked(executor, jobType, policy) + return fits +} + +// executorWorkerLoad is the worker's load across every job type, the +// number to balance by. +func (r *Plugin) executorWorkerLoad(executor *WorkerSession) int { + if executor == nil { + return 0 + } + r.execResMu.Lock() + defer r.execResMu.Unlock() + return r.executorLoadLocked(executor, "").total() +} + +// reserveExecutorSlot holds a slot on the worker for jobID if one more +// jobType job fits. The release returned gives it back; the reservation +// is the admin's own count of the slot until then. +func (r *Plugin) reserveExecutorSlot( executor *WorkerSession, jobType string, + jobID string, policy schedulerPolicy, - lane SchedulerLane, ) (func(), bool) { if executor == nil || strings.TrimSpace(executor.WorkerID) == "" { return nil, false } - limit := schedulerWorkerExecutionLimit(executor, jobType, policy) - if limit <= 0 { + r.execResMu.Lock() + defer r.execResMu.Unlock() + if fits, _, _ := r.executorCapacityLocked(executor, jobType, policy); !fits { return nil, false } - heartbeatUsed := 0 - if executor.Heartbeat != nil && executor.Heartbeat.ExecutionSlotsUsed > 0 { - heartbeatUsed = int(executor.Heartbeat.ExecutionSlotsUsed) - } - - workerID := strings.TrimSpace(executor.WorkerID) - - ls := r.lanes[lane] - if ls == nil { - // Fallback to global reservations if lane state is missing. - r.schedulerExecMu.Lock() - reserved := r.schedulerExecReservations[workerID] - if heartbeatUsed+reserved >= limit { - r.schedulerExecMu.Unlock() - return nil, false + jobID = strings.TrimSpace(jobID) + if jobID != "" { + for token, res := range r.execRes { + if res.jobID == jobID && !res.releasedAt.IsZero() { + delete(r.execRes, token) + } } - r.schedulerExecReservations[workerID] = reserved + 1 - r.schedulerExecMu.Unlock() - release := func() { r.releaseExecutorCapacity(workerID) } - return release, true } - - ls.execMu.Lock() - reserved := ls.execRes[workerID] - if heartbeatUsed+reserved >= limit { - ls.execMu.Unlock() - return nil, false + r.execResSeq++ + token := r.execResSeq + r.execRes[token] = execReservation{ + workerID: strings.TrimSpace(executor.WorkerID), + jobType: jobType, + jobID: jobID, } - ls.execRes[workerID] = reserved + 1 - ls.execMu.Unlock() - release := func() { - r.releaseExecutorCapacityForLane(workerID, lane) + r.execResMu.Lock() + defer r.execResMu.Unlock() + res, ok := r.execRes[token] + if !ok || !res.releasedAt.IsZero() { + return + } + if res.jobID == "" { + delete(r.execRes, token) + return + } + res.releasedAt = time.Now() + r.execRes[token] = res } return release, true } -// releaseExecutorCapacityForLane releases a reservation from the per-lane pool. -func (r *Plugin) releaseExecutorCapacityForLane(workerID string, lane SchedulerLane) { - workerID = strings.TrimSpace(workerID) - if workerID == "" { - return - } - - ls := r.lanes[lane] - if ls == nil { - r.releaseExecutorCapacity(workerID) - return - } - - ls.execMu.Lock() - defer ls.execMu.Unlock() - - current := ls.execRes[workerID] - if current <= 1 { - delete(ls.execRes, workerID) - return - } - ls.execRes[workerID] = current - 1 -} - -func (r *Plugin) releaseExecutorCapacity(workerID string) { - workerID = strings.TrimSpace(workerID) - if workerID == "" { - return - } - - r.schedulerExecMu.Lock() - defer r.schedulerExecMu.Unlock() - - current := r.schedulerExecReservations[workerID] - if current <= 1 { - delete(r.schedulerExecReservations, workerID) - return - } - r.schedulerExecReservations[workerID] = current - 1 -} - +// schedulerWorkerExecutionLimit is the most jobType jobs one worker may run +// at once for this admin: per_worker_execution_concurrency, capped by what +// the worker declared for the type and by its slot total. It bounds jobType +// work alone; other job types on the same worker count only against the +// slot total (see executorCapacityLocked). func schedulerWorkerExecutionLimit(executor *WorkerSession, jobType string, policy schedulerPolicy) int { limit := policy.PerWorkerConcurrency if limit <= 0 { diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index 8b7a384f23..d5ce9e77fe 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -1048,3 +1048,642 @@ func TestReserveWaterfillsPreferredHosts(t *testing.T) { t.Fatalf("want one job per worker, got %v", seen) } } + +// ---------- executor capacity ---------- + +// registerBusyWorker connects one worker that can run every job type in +// caps with four slots, one of which the worker reports taken by running +// work of busyType that this admin never dispatched. +func registerBusyWorker(t *testing.T, pluginSvc *Plugin, workerID, busyType, busyJobID string, caps ...string) { + t.Helper() + hello := &plugin_pb.WorkerHello{WorkerId: workerID} + for _, jobType := range caps { + hello.Capabilities = append(hello.Capabilities, &plugin_pb.JobTypeCapability{ + JobType: jobType, CanExecute: true, MaxExecutionConcurrency: 4, + }) + } + pluginSvc.registry.UpsertFromHello(hello) + heartbeat := &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, + ExecutionSlotsUsed: 1, + ExecutionSlotsTotal: 4, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: busyJobID, + Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, + JobType: busyType, + State: plugin_pb.JobState_JOB_STATE_RUNNING, + Stage: "starting", + }}, + } + pluginSvc.registry.UpdateHeartbeat(workerID, heartbeat) +} + +// The field case: the only admin_script-capable worker has 4 slots, one held +// by an s3_lifecycle job the worker reports running that this admin never +// dispatched, and admin_script runs at global 1 / per-worker 1. +// per_worker_execution_concurrency bounds admin_script work, so the +// lifecycle job must not make admin_script undispatchable; the limit still +// holds within admin_script itself and against foreign work of the same type. +func TestExecutorCapacityCountsOnlyTheJobTypeBeingDispatched(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "w-penso--7aa8" + registerBusyWorker(t, pluginSvc, workerID, "s3_lifecycle", "s3_lifecycle-scheduled-1793477386659262943-0", + "admin_script", "s3_lifecycle", "vacuum", "ec_balance") + policy := schedulerPolicy{ExecutionConcurrency: 1, PerWorkerConcurrency: 1, LocalQueuePin: 5} + + job := &plugin_pb.JobSpec{JobId: "admin_script-scheduled-1793738609215134308-0", JobType: "admin_script", DedupeKey: "admin_script"} + qs := pluginSvc.assignJobsToWorkerQueues("admin_script", []*plugin_pb.JobSpec{job}, policy) + item, target := pluginSvc.takeQueuedJob(qs, "admin_script", policy) + if item == nil || target != workerID { + t.Fatalf("admin_script should dispatch to %s past the foreign s3_lifecycle job, got item=%v target=%q", workerID, item, target) + } + executor, release, err := pluginSvc.reserveExecutorByID("admin_script", policy, target, job.JobId) + if err != nil { + t.Fatalf("reserve admin_script: %v", err) + } + defer release() + + // The per-type limit of 1 is now spent on admin_script's own job. + if pluginSvc.executorHasCapacity(executor, "admin_script", policy) { + t.Fatalf("a second admin_script job must wait for the first") + } + // Unrelated types on the same worker still have their own per-type budget. + if !pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("vacuum should still fit: 2 of 4 slots used, none by vacuum") + } + // Foreign work of the same type counts against that type's limit. + if pluginSvc.executorHasCapacity(executor, "s3_lifecycle", policy) { + t.Fatalf("a second s3_lifecycle job must not fit beside the foreign one at per-worker 1") + } +} + +// Per-type budgets never add up to more than the worker's slot total: a +// worker whose slots are all taken by other job types takes nothing, whether +// it itemized that work or only reported the count. +func TestExecutorCapacityHonorsWorkerSlotTotalAcrossJobTypes(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "admin_script", CanExecute: true, MaxExecutionConcurrency: 4}, + {JobType: "s3_lifecycle", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + policy := schedulerPolicy{ExecutionConcurrency: 1, PerWorkerConcurrency: 1} + executor := func() *WorkerSession { + session, ok := pluginSvc.registry.Get(workerID) + if !ok { + t.Fatalf("worker missing") + } + return session + } + + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 1, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "lc-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle", + }}, + }) + if pluginSvc.executorHasCapacity(executor(), "admin_script", policy) { + t.Fatalf("the worker's only slot is taken by s3_lifecycle") + } + + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 1, + }) + if pluginSvc.executorHasCapacity(executor(), "admin_script", policy) { + t.Fatalf("a slot the worker counts as used is used even when not itemized") + } + + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1, + }) + if !pluginSvc.executorHasCapacity(executor(), "admin_script", policy) { + t.Fatalf("a free worker should take the job") + } +} + +// A job this admin dispatched shows up twice, as its reservation and in the +// worker's heartbeat. It is one slot, and must not be charged as two. +func TestExecutorCapacityDoesNotDoubleCountReservedJobsEchoedByHeartbeat(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + policy := schedulerPolicy{ExecutionConcurrency: 2, PerWorkerConcurrency: 2} + + _, release, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + defer release() + + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "vacuum-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "vacuum", + }}, + }) + executor, _ := pluginSvc.registry.Get(workerID) + if got := pluginSvc.executorWorkerLoad(executor); got != 1 { + t.Fatalf("worker load = %d, want 1: the reservation and its heartbeat echo are one slot", got) + } + if !pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("one vacuum job in flight leaves room for a second at per-worker 2") + } +} + +// ---------- undispatchable jobs do not hold the lane ---------- + +// A worker whose slots are all held by work this admin never dispatched +// will not free them on any schedule the lane can see. The run gives the +// job up at once, says why, and leaves the window for the lane's other +// job types instead of idling until job_type_max_runtime. +func TestDispatchGivesUpWhenCapacityIsHeldByForeignWork(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "admin_script", CanExecute: true, MaxExecutionConcurrency: 1}, + }, + }) + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 1, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "lc-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle", + }}, + }) + + window, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + ExecutionTimeout: time.Second, + ExecutorReserveBackoff: time.Millisecond, + OrphanRehomeGrace: time.Second, + } + proposals := []*plugin_pb.JobProposal{{ProposalId: "p1", JobType: "admin_script", DedupeKey: "k1"}} + + started := time.Now() + success, errCount, canceled := pluginSvc.dispatchScheduledProposals(window, "admin_script", proposals, &plugin_pb.ClusterContext{}, policy) + if elapsed := time.Since(started); elapsed > 5*time.Second { + t.Fatalf("dispatch held the lane for %v with no executor able to take the job", elapsed) + } + if success != 0 || errCount != 0 || canceled != 1 { + t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", success, errCount, canceled) + } + + activity := findActivity(pluginSvc.ListActivities("admin_script", 100), "admin_scheduler", "skipped_no_executor") + if activity == nil { + t.Fatalf("expected a skipped_no_executor activity, got %+v", pluginSvc.ListActivities("admin_script", 100)) + } + want := "no executor capacity for admin_script: worker worker-a has 1/1 slots used by s3_lifecycle this admin did not dispatch" + if activity.Message != want { + t.Fatalf("activity message = %q, want %q", activity.Message, want) + } + for _, job := range pluginSvc.ListTrackedJobs("admin_script", "", 10) { + if job.State != StateCanceled || job.Message != want { + t.Fatalf("queued job should be canceled with the reason, got %+v", job) + } + } +} + +// Capacity held by the lane's own jobs is a different matter: they finish +// and release, so the run waits for them as before. +func TestDispatchWaitsForItsOwnLaneToReleaseCapacity(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "admin_script", CanExecute: true, MaxExecutionConcurrency: 1}, + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 1}, + }, + }) + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1, + }) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + ExecutionTimeout: time.Hour, + ExecutorReserveBackoff: time.Millisecond, + OrphanRehomeGrace: time.Second, + } + // A vacuum job of the same (default) lane holds the worker's only slot. + _, releaseVacuum, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum: %v", err) + } + + resultCh := make(chan [3]int, 1) + go func() { + success, errCount, canceled := pluginSvc.dispatchScheduledProposals( + context.Background(), "admin_script", []*plugin_pb.JobProposal{{ProposalId: "p1", JobType: "admin_script", DedupeKey: "k1"}}, + &plugin_pb.ClusterContext{}, policy) + resultCh <- [3]int{success, errCount, canceled} + }() + + select { + case result := <-resultCh: + t.Fatalf("dispatch gave up on capacity its own lane holds: %v", result) + case <-time.After(50 * time.Millisecond): + } + + releaseVacuum() + first := <-session.outgoing + execReq := first.GetExecuteJobRequest() + if execReq == nil { + t.Fatalf("expected execute_job_request after the vacuum slot was released, got %+v", first) + } + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + RequestId: first.RequestId, JobId: execReq.Job.JobId, JobType: "admin_script", Success: true, CompletedAt: timestamppb.Now(), + }) + if result := <-resultCh; result != [3]int{1, 0, 0} { + t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", result[0], result[1], result[2]) + } +} + +// With no capable worker connected the run waits out the reconnect grace, +// then gives up rather than holding the lane for the whole window. +func TestDispatchGivesUpWhenNoExecutorReturnsWithinGrace(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + window, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + ExecutorReserveBackoff: time.Millisecond, + OrphanRehomeGrace: 20 * time.Millisecond, + } + + started := time.Now() + _, _, canceled := pluginSvc.dispatchScheduledProposals(window, "admin_script", + []*plugin_pb.JobProposal{{ProposalId: "p1", JobType: "admin_script", DedupeKey: "k1"}}, &plugin_pb.ClusterContext{}, policy) + if elapsed := time.Since(started); elapsed < policy.OrphanRehomeGrace || elapsed > 5*time.Second { + t.Fatalf("dispatch took %v, want about the %v grace", elapsed, policy.OrphanRehomeGrace) + } + if canceled != 1 { + t.Fatalf("canceled = %d, want 1", canceled) + } + activity := findActivity(pluginSvc.ListActivities("admin_script", 100), "admin_scheduler", "skipped_no_executor") + if activity == nil || activity.Message != "no executor worker available for admin_script" { + t.Fatalf("expected the no-worker reason, got %+v", activity) + } +} + +// The worker reports a job's completion before the heartbeat that frees its +// slot, so for a moment the lane's own finished job still shows as running, +// itemized by the Go worker and only as a slot count by the Rust one. That +// echo is the lane's slot coming back, not foreign work. +func TestDispatchWaitsForTheEchoOfItsOwnCompletedJob(t *testing.T) { + t.Parallel() + for _, itemized := range []bool{true, false} { + itemized := itemized + t.Run(fmt.Sprintf("itemized=%v", itemized), func(t *testing.T) { + t.Parallel() + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 1}, + }, + }) + idle := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1} + pluginSvc.registry.UpdateHeartbeat(workerID, idle) + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + ExecutionTimeout: time.Hour, + ExecutorReserveBackoff: time.Millisecond, + OrphanRehomeGrace: time.Second, + } + proposals := []*plugin_pb.JobProposal{ + {ProposalId: "p1", JobType: "vacuum", DedupeKey: "k1"}, + {ProposalId: "p2", JobType: "vacuum", DedupeKey: "k2"}, + } + resultCh := make(chan [3]int, 1) + go func() { + success, errCount, canceled := pluginSvc.dispatchScheduledProposals( + context.Background(), "vacuum", proposals, &plugin_pb.ClusterContext{}, policy) + resultCh <- [3]int{success, errCount, canceled} + }() + + complete := func(message *plugin_pb.AdminToWorkerMessage) { + t.Helper() + execReq := message.GetExecuteJobRequest() + if execReq == nil { + t.Fatalf("expected execute_job_request, got %+v", message) + } + running := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 1} + if itemized { + running.RunningWork = []*plugin_pb.RunningWork{{ + WorkId: execReq.Job.JobId, Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "vacuum", + }} + } + pluginSvc.registry.UpdateHeartbeat(workerID, running) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + RequestId: message.RequestId, JobId: execReq.Job.JobId, JobType: "vacuum", Success: true, CompletedAt: timestamppb.Now(), + }) + } + + complete(<-session.outgoing) + select { + case result := <-resultCh: + t.Fatalf("dispatch gave up on the echo of its own completed job: %v", result) + case <-time.After(50 * time.Millisecond): + } + + pluginSvc.registry.UpdateHeartbeat(workerID, idle) + complete(<-session.outgoing) + pluginSvc.registry.UpdateHeartbeat(workerID, idle) + if result := <-resultCh; result != [3]int{2, 0, 0} { + t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", result[0], result[1], result[2]) + } + }) + } +} + +// A worker may count a slot before it itemizes the work in it (the Go worker +// for a moment after taking a job, the Rust worker always). The admin's own +// jobs claim such slots first, so they are not charged twice; what is left +// is of unknown type and holds every type's per-worker limit. +func TestExecutorCapacityCountsUnitemizedSlots(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + executor := func() *WorkerSession { + session, ok := pluginSvc.registry.Get(workerID) + if !ok { + t.Fatalf("worker missing") + } + return session + } + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + if pluginSvc.executorHasCapacity(executor(), "vacuum", schedulerPolicy{PerWorkerConcurrency: 1}) { + t.Fatalf("a slot the worker counts but has not itemized may hold vacuum work") + } + + policy := schedulerPolicy{PerWorkerConcurrency: 2} + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsTotal: 4}) + _, release, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + defer release() + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + if got := pluginSvc.executorWorkerLoad(executor()); got != 1 { + t.Fatalf("worker load = %d, want 1: the unitemized slot is the reserved job", got) + } + if !pluginSvc.executorHasCapacity(executor(), "vacuum", policy) { + t.Fatalf("one vacuum job in flight leaves room for a second at per-worker 2") + } +} + +// A job homed on a replica-host worker stays pinned there while its home +// queue is short, so a free remote worker cannot take it. When the home is +// held by work the lane cannot free, the run gives the job up after the +// orphan grace instead of waiting out the window beside the free worker. +func TestDispatchGivesUpPinnedJobsWhoseHomeIsHeldByForeignWork(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", + }) + pluginSvc.registry.UpdateHeartbeat("worker-local", &plugin_pb.WorkerHeartbeat{ + WorkerId: "worker-local", ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 1, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "vac-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "vacuum", + }}, + }) + pluginSvc.registry.UpdateHeartbeat("worker-remote", &plugin_pb.WorkerHeartbeat{ + WorkerId: "worker-remote", ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1, + }) + + window, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + policy := schedulerPolicy{ + ExecutionConcurrency: 1, + PerWorkerConcurrency: 1, + LocalQueuePin: 5, + ExecutionTimeout: time.Hour, + ExecutorReserveBackoff: time.Millisecond, + OrphanRehomeGrace: 50 * time.Millisecond, + } + proposals := []*plugin_pb.JobProposal{{ + ProposalId: "p1", JobType: "erasure_coding", DedupeKey: "k1", + Parameters: replicaJob("j1", "10.0.0.1:8080").Parameters, + }} + + started := time.Now() + success, errCount, canceled := pluginSvc.dispatchScheduledProposals(window, "erasure_coding", proposals, &plugin_pb.ClusterContext{}, policy) + if elapsed := time.Since(started); elapsed < policy.OrphanRehomeGrace || elapsed > 4*time.Second { + t.Fatalf("dispatch took %v, want about the %v grace", elapsed, policy.OrphanRehomeGrace) + } + if success != 0 || errCount != 0 || canceled != 1 { + t.Fatalf("unexpected dispatch counts: success=%d errors=%d canceled=%d", success, errCount, canceled) + } + activity := findActivity(pluginSvc.ListActivities("erasure_coding", 100), "admin_scheduler", "skipped_no_executor") + want := "no executor capacity for erasure_coding: worker worker-local has 1/1 slots used by vacuum this admin did not dispatch" + if activity == nil || activity.Message != want { + t.Fatalf("expected %q, got %+v", want, activity) + } +} + +// A job the worker turned away is released and queued again under the same +// id; its new reservation replaces what the first one left behind, so the +// running job is one slot. +func TestExecutorCapacityCountsARedispatchedJobOnce(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + policy := schedulerPolicy{PerWorkerConcurrency: 2} + _, rejected, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + rejected() + _, release, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1 again: %v", err) + } + defer release() + + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "vacuum-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "vacuum", + }}, + }) + executor, _ := pluginSvc.registry.Get(workerID) + if got := pluginSvc.executorWorkerLoad(executor); got != 1 { + t.Fatalf("worker load = %d, want 1", got) + } + if !pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("one vacuum job in flight leaves room for a second at per-worker 2") + } +} + +// A count-only heartbeat sent after a job completed no longer holds that +// job: a slot it still counts belongs to other work, of unknown type. +func TestExecutorCapacityStopsCreditingACompletedJobAfterTheNextHeartbeat(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + {JobType: "balance", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + policy := schedulerPolicy{PerWorkerConcurrency: 1} + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + _, release, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + release() + + executor, _ := pluginSvc.registry.Get(workerID) + if !pluginSvc.executorHasCapacity(executor, "balance", policy) { + t.Fatalf("the counted slot is still vacuum-1's echo, not balance work") + } + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + executor, _ = pluginSvc.registry.Get(workerID) + if pluginSvc.executorHasCapacity(executor, "balance", policy) { + t.Fatalf("a slot counted after vacuum-1 completed may hold balance work") + } +} + +// The reason a dispatch gives up names every holder of the worker's slots, +// the unitemized ones included. +func TestCapacityReasonNamesUnitemizedSlots(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "admin_script", CanExecute: true, MaxExecutionConcurrency: 2}, + }, + }) + pluginSvc.registry.UpdateHeartbeat(workerID, &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, ExecutionSlotsUsed: 2, ExecutionSlotsTotal: 2, + RunningWork: []*plugin_pb.RunningWork{{ + WorkId: "lc-1", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle", + }}, + }) + + wait, reason := pluginSvc.executorCapacityOutlook("admin_script", schedulerPolicy{PerWorkerConcurrency: 2}, 0) + want := "no executor capacity for admin_script: worker worker-a has 2/2 slots used by s3_lifecycle, unitemized work this admin did not dispatch" + if wait || reason != want { + t.Fatalf("outlook = (%v, %q), want (false, %q)", wait, reason, want) + } +} diff --git a/weed/admin/plugin/scheduler_lane.go b/weed/admin/plugin/scheduler_lane.go index 070b0025f0..ec960e4a14 100644 --- a/weed/admin/plugin/scheduler_lane.go +++ b/weed/admin/plugin/scheduler_lane.go @@ -127,19 +127,12 @@ type schedulerLaneState struct { loopMu sync.Mutex loop schedulerLoopState - - // Per-lane execution reservation pool. Each lane tracks how many - // execution slots it has reserved on each worker independently, - // so lanes cannot starve each other. - execMu sync.Mutex - execRes map[string]int } // newLaneState creates a schedulerLaneState for the given lane. func newLaneState(lane SchedulerLane) *schedulerLaneState { return &schedulerLaneState{ - lane: lane, - wakeCh: make(chan struct{}, 1), - execRes: make(map[string]int), + lane: lane, + wakeCh: make(chan struct{}, 1), } } diff --git a/weed/admin/plugin/worker_job_queues.go b/weed/admin/plugin/worker_job_queues.go index 1e036a816c..0683e43ee8 100644 --- a/weed/admin/plugin/worker_job_queues.go +++ b/weed/admin/plugin/worker_job_queues.go @@ -141,7 +141,6 @@ func (r *Plugin) assignJobsToWorkerQueues( ) *workerJobQueues { qs := newWorkerJobQueues() executors, err := r.registry.ListExecutors(jobType) - lane := JobTypeLane(jobType) queued := make(map[string]int, len(executors)) for _, job := range jobs { @@ -164,7 +163,7 @@ func (r *Plugin) assignJobsToWorkerQueues( } } sortExecutorsByLoad(pref, func(e *WorkerSession) int { - return r.executorReservationLoad(e, lane) + queued[e.WorkerID] + return r.executorWorkerLoad(e) + queued[e.WorkerID] }) home := pref[0].WorkerID qs.push(home, job) @@ -194,7 +193,6 @@ func (r *Plugin) takeQueuedJob( if err != nil || len(executors) == 0 { return nil, "" } - lane := JobTypeLane(jobType) pin := policy.LocalQueuePin defer func() { if item != nil && targetID != "" { @@ -214,14 +212,10 @@ func (r *Plugin) takeQueuedJob( 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 { + if !r.executorHasCapacity(exec, jobType, policy) { continue } + load := r.executorWorkerLoad(exec) if bestHome == nil || load < bestLoad || (load == bestLoad && exec.WorkerID < bestHome.WorkerID) { bestHome = exec bestLoad = load @@ -236,11 +230,7 @@ func (r *Plugin) takeQueuedJob( var thief *WorkerSession for _, exec := range executors { - limit := schedulerWorkerExecutionLimit(exec, jobType, policy) - if limit <= 0 { - continue - } - if r.executorReservationLoad(exec, lane) >= limit { + if !r.executorHasCapacity(exec, jobType, policy) { continue } if thief == nil || exec.WorkerID < thief.WorkerID { diff --git a/weed/admin/plugin/worker_job_queues_test.go b/weed/admin/plugin/worker_job_queues_test.go index 662ac4e7b7..c31aab954f 100644 --- a/weed/admin/plugin/worker_job_queues_test.go +++ b/weed/admin/plugin/worker_job_queues_test.go @@ -103,7 +103,7 @@ func TestAssignAndTakeDoesNotStarveOtherHomes(t *testing.T) { } _, target1 := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) - _, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1) + _, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1, "") if err != nil { t.Fatalf("reserve 1: %v", err) } @@ -143,7 +143,7 @@ func TestTakeStealsOnlyWhenHomeQueueReachesK(t *testing.T) { if target != "worker-local" { t.Fatalf("first take want local, got %s", target) } - _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local", "") if err != nil { t.Fatalf("reserve local: %v", err) } @@ -189,7 +189,7 @@ func TestTakeStealsImmediatelyWhenPinDisabled(t *testing.T) { if target != "worker-local" { t.Fatalf("first take want local, got %s", target) } - _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local", "") if err != nil { t.Fatalf("reserve local: %v", err) } @@ -381,7 +381,7 @@ func TestHomelessAllRemoteWorkersUseFIFOWithoutPinOrGrace(t *testing.T) { if item1 == nil || item1.job.JobId != "h1" { t.Fatalf("FIFO first want h1, got %v", item1) } - _, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1) + _, release1, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, target1, "") if err != nil { t.Fatalf("reserve 1: %v", err) } @@ -433,7 +433,7 @@ func TestMixedReplicaLocalJobsStayHomeBeforeHomeless(t *testing.T) { if item1 == nil || item1.job.JobId != "local-1" || target1 != "worker-local" { t.Fatalf("first take want local-1 on home, got job=%v target=%s", item1, target1) } - _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local", "") if err != nil { t.Fatalf("reserve local: %v", err) } @@ -523,7 +523,7 @@ func TestHomelessWorkerTakesFIFOBeforeStealableHome(t *testing.T) { if target != "worker-local" { t.Fatalf("first take want local, got %s", target) } - _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local", "") if err != nil { t.Fatalf("reserve local: %v", err) } @@ -579,7 +579,7 @@ func TestHomeWorkerStealsBeforeHomelessAfterOwnQueueEmpty(t *testing.T) { if item1 == nil || target1 != "worker-local" { t.Fatalf("first take want local home, got job=%v target=%s", item1, target1) } - _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local") + _, release, err := pluginSvc.reserveExecutorByID("erasure_coding", policy, "worker-local", "") if err != nil { t.Fatalf("reserve local: %v", err) }