diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index ceed99b7e6..5d80254d90 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -1220,6 +1220,7 @@ type execReservation struct { workerID string jobType string jobID string + reservedAt time.Time releasedAt time.Time } @@ -1282,9 +1283,10 @@ func (l executorLoad) holders() string { // 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. +// first to this admin's jobs it does not list, and only from a heartbeat +// that came after the job's reservation: a pending job's from then on, a +// completed job's 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), @@ -1330,13 +1332,13 @@ func (r *Plugin) executorLoadLocked(executor *WorkerSession, jobType string) exe } switch { case !released: - if !isListed { + if !isListed && !executor.heartbeatAt.Before(res.reservedAt) { unitemized-- } countReserved(res) case isListed: countReserved(res) - case executor.LastSeenAt.Before(res.releasedAt): + case !executor.heartbeatAt.Before(res.reservedAt) && executor.heartbeatAt.Before(res.releasedAt): unlistedEchoes = append(unlistedEchoes, res) } } @@ -1442,9 +1444,10 @@ func (r *Plugin) reserveExecutorSlot( r.execResSeq++ token := r.execResSeq r.execRes[token] = execReservation{ - workerID: strings.TrimSpace(executor.WorkerID), - jobType: jobType, - jobID: jobID, + workerID: strings.TrimSpace(executor.WorkerID), + jobType: jobType, + jobID: jobID, + reservedAt: time.Now(), } release := func() { r.execResMu.Lock() diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index d5ce9e77fe..cd1622c67f 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -1656,6 +1656,78 @@ func TestExecutorCapacityStopsCreditingACompletedJobAfterTheNextHeartbeat(t *tes } } +// A count-only heartbeat taken before a reservation cannot count that job, +// so the slot it counted is still unknown work rather than the new job's. +// A reconnect's hello keeps that heartbeat and does not change this. +func TestExecutorCapacityKeepsSlotsCountedBeforeAReservation(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + hello := &plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + } + pluginSvc.registry.UpsertFromHello(hello) + policy := schedulerPolicy{PerWorkerConcurrency: 2} + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + if _, _, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1"); err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + pluginSvc.registry.UpsertFromHello(hello) + + executor, _ := pluginSvc.registry.Get(workerID) + if pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("the slot counted before vacuum-1 was reserved may hold vacuum work") + } + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + executor, _ = pluginSvc.registry.Get(workerID) + if !pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("a heartbeat after the reservation counts vacuum-1 itself") + } +} + +// A job that completes before any heartbeat counts it has no echo: the slot +// the last heartbeat counted is still unknown work, not the job's. +func TestExecutorCapacityKeepsSlotsCountedBeforeACompletedJob(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}, + }, + }) + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + _, release, err := pluginSvc.reserveExecutorByID("vacuum", schedulerPolicy{PerWorkerConcurrency: 2}, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + release() + + executor, _ := pluginSvc.registry.Get(workerID) + if pluginSvc.executorHasCapacity(executor, "balance", schedulerPolicy{PerWorkerConcurrency: 1}) { + t.Fatalf("the slot counted before vacuum-1 was reserved 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) { diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 4fa141fb06..3ff48ab397 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -24,6 +24,8 @@ type WorkerSession struct { LastSeenAt time.Time Capabilities map[string]*plugin_pb.JobTypeCapability Heartbeat *plugin_pb.WorkerHeartbeat + // heartbeatAt is when Heartbeat arrived; a hello moves LastSeenAt alone. + heartbeatAt time.Time } // Registry tracks connected plugin workers and capability-based selection. @@ -91,7 +93,8 @@ func (r *Registry) UpdateHeartbeat(workerID string, heartbeat *plugin_pb.WorkerH return } session.Heartbeat = cloneWorkerHeartbeat(heartbeat) - session.LastSeenAt = time.Now() + session.heartbeatAt = time.Now() + session.LastSeenAt = session.heartbeatAt } func (r *Registry) Get(workerID string) (*WorkerSession, bool) {