diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index f1590c9dc7..aa97e92737 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -31,7 +31,10 @@ 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 + defaultLocalQueuePin = 5 + // Wait this long after a home worker disappears before stealing + // its queue; 0 steals immediately. Covers a reconnect blink. + defaultOrphanRehomeGrace = 30 * time.Second maxScheduledExecutionConcurrency = 128 defaultScheduledRetryBackoff = 5 * time.Second defaultClusterContextTimeout = 10 * time.Second @@ -56,6 +59,9 @@ type schedulerPolicy struct { // LocalQueuePin is K: do not spill off replica-host workers while // min(load(pref)) < K. Zero is the old immediate-spill path. LocalQueuePin int + // OrphanRehomeGrace is how long a missing home's queue is held + // before steal-without-K. Zero steals as soon as the worker is gone. + OrphanRehomeGrace time.Duration } // laneSchedulerLoop is the main scheduling goroutine for a single lane. @@ -474,6 +480,7 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err RetryLimit: int(adminRuntime.RetryLimit), ExecutorReserveBackoff: 200 * time.Millisecond, LocalQueuePin: defaultLocalQueuePin, + OrphanRehomeGrace: defaultOrphanRehomeGrace, } if policy.DetectionInterval < r.schedulerTick { @@ -1650,6 +1657,7 @@ func (r *Plugin) DispatchProposals( RetryBackoff: defaultScheduledRetryBackoff, ExecutorReserveBackoff: 200 * time.Millisecond, LocalQueuePin: defaultLocalQueuePin, + OrphanRehomeGrace: defaultOrphanRehomeGrace, } } diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index 36f48f7e53..b7aa3757ca 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -64,6 +64,9 @@ func TestLoadSchedulerPolicyUsesAdminConfig(t *testing.T) { if policy.LocalQueuePin != defaultLocalQueuePin { t.Fatalf("unexpected local queue pin: got=%d", policy.LocalQueuePin) } + if policy.OrphanRehomeGrace != defaultOrphanRehomeGrace { + t.Fatalf("unexpected orphan rehome grace: got=%v", policy.OrphanRehomeGrace) + } } func TestLoadSchedulerPolicyUsesDescriptorDefaultsWhenConfigMissing(t *testing.T) { diff --git a/weed/admin/plugin/worker_job_queues.go b/weed/admin/plugin/worker_job_queues.go index 439d28c375..c03cf22f67 100644 --- a/weed/admin/plugin/worker_job_queues.go +++ b/weed/admin/plugin/worker_job_queues.go @@ -2,6 +2,7 @@ package plugin import ( "sync" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" ) @@ -15,14 +16,18 @@ type queuedJob struct { // 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 + mu sync.Mutex + q map[string][]*queuedJob + leftover []*queuedJob + remaining int + missingSince map[string]time.Time } func newWorkerJobQueues() *workerJobQueues { - return &workerJobQueues{q: make(map[string][]*queuedJob)} + return &workerJobQueues{ + q: make(map[string][]*queuedJob), + missingSince: make(map[string]time.Time), + } } func (qs *workerJobQueues) push(home string, job *plugin_pb.JobSpec) { @@ -142,7 +147,8 @@ func (r *Plugin) assignJobsToWorkerQueues( // takeQueuedJob pops a home-runnable job, or steals onto a free worker. // Live homes are stolen from only when their queue is at/over K. -// A home missing from live executors is stolen immediately (no K). +// A home missing from live executors is stolen without K after +// OrphanRehomeGrace (0 = immediately). func (r *Plugin) takeQueuedJob( qs *workerJobQueues, jobType string, @@ -221,20 +227,33 @@ func (r *Plugin) takeQueuedJob( live := make(map[string]struct{}, len(executors)) for _, exec := range executors { live[exec.WorkerID] = struct{}{} + delete(qs.missingSince, exec.WorkerID) } + now := time.Now() + grace := policy.OrphanRehomeGrace victimID := "" victimLen := 0 victimOrphan := false for id, qq := range qs.q { if len(qq) == 0 { + delete(qs.missingSince, id) continue } _, isLive := live[id] - if isLive && pin > 0 && len(qq) < pin { - continue - } orphan := !isLive + if isLive { + if pin > 0 && len(qq) < pin { + continue + } + } else { + if _, ok := qs.missingSince[id]; !ok { + qs.missingSince[id] = now + } + if grace > 0 && now.Sub(qs.missingSince[id]) < grace { + continue + } + } better := victimID == "" if !better && orphan != victimOrphan { better = orphan diff --git a/weed/admin/plugin/worker_job_queues_test.go b/weed/admin/plugin/worker_job_queues_test.go index 2fcb5b5cca..23329104f8 100644 --- a/weed/admin/plugin/worker_job_queues_test.go +++ b/weed/admin/plugin/worker_job_queues_test.go @@ -3,6 +3,7 @@ package plugin import ( "fmt" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" ) @@ -218,7 +219,11 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) { replicaJob("o1", "10.0.0.1:8080"), replicaJob("o2", "10.0.0.1:8080"), } - policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5} + policy := schedulerPolicy{ + PerWorkerConcurrency: 1, + LocalQueuePin: 5, + OrphanRehomeGrace: 30 * time.Second, + } qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) if qs.queueLen("worker-local") != 2 { t.Fatalf("want 2 on local, got %d", qs.queueLen("worker-local")) @@ -226,9 +231,20 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) { pluginSvc.registry.Remove("worker-local") + if item, tgt := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil { + t.Fatalf("within grace should hold orphan queue, got target=%s", tgt) + } + if qs.queueLen("worker-local") != 2 { + t.Fatalf("held queue changed: %d", qs.queueLen("worker-local")) + } + + qs.mu.Lock() + qs.missingSince["worker-local"] = time.Now().Add(-time.Minute) + qs.mu.Unlock() + stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) if stolen == nil || stealTarget != "worker-remote" { - t.Fatalf("dead home should steal without K, got job=%v target=%s", stolen, stealTarget) + t.Fatalf("after grace should steal without K, got job=%v target=%s", stolen, stealTarget) } if qs.queueLen("worker-local") != 1 { t.Fatalf("want 1 orphan left, got %d", qs.queueLen("worker-local")) @@ -240,6 +256,64 @@ func TestTakeStealsOrphanedHomeWithoutK(t *testing.T) { } } +func TestTakeKeepsOrphanWhenHomeReconnectsDuringGrace(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("blink", "10.0.0.1:8080")} + policy := schedulerPolicy{ + PerWorkerConcurrency: 1, + LocalQueuePin: 5, + OrphanRehomeGrace: 30 * time.Second, + } + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) + pluginSvc.registry.Remove("worker-local") + if item, _ := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy); item != nil { + t.Fatal("expected hold during grace") + } + + registerECWorkers(t, pluginSvc, map[string]string{"worker-local": "10.0.0.1:9334"}) + item, target := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if item == nil || target != "worker-local" { + t.Fatalf("reconnected home should run its queue, got job=%v target=%s", item, target) + } +} + +func TestTakeStealsOrphanImmediatelyWhenGraceDisabled(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("o1", "10.0.0.1:8080")} + policy := schedulerPolicy{PerWorkerConcurrency: 1, LocalQueuePin: 5, OrphanRehomeGrace: 0} + qs := pluginSvc.assignJobsToWorkerQueues("erasure_coding", jobs, policy) + pluginSvc.registry.Remove("worker-local") + + stolen, stealTarget := pluginSvc.takeQueuedJob(qs, "erasure_coding", policy) + if stolen == nil || stealTarget != "worker-remote" { + t.Fatalf("grace=0 should steal immediately, got job=%v target=%s", stolen, stealTarget) + } +} + func TestBindQueuedJobWorkerKeepsPendingState(t *testing.T) { t.Parallel()