From 41f22f5c921db06932af6a435ac68cc240b38f52 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Mar 2026 16:41:59 -0800 Subject: [PATCH] plugin: rewrite scheduler to sequential lock-based iteration model Replace the ticker-based parallel goroutine scheduler with a single-lock sequential-processing iteration model: - schedulerLoop now calls runSchedulerIteration in a loop, sleeping idleSleepDuration (default 17m) when no work is found, or looping immediately when work is detected. - runSchedulerIteration acquires the admin lock once, loads cluster context once, then processes all enabled job types sequentially. - runJobTypeIteration runs detection then execution for one job type within a per-job-type budget (default 30m) enforced via context. - Remove markDetectionDue, finishDetection, pruneSchedulerState, runSchedulerTick, and runScheduledDetection which are no longer needed. - dispatchScheduledProposals and executeScheduledJobWithExecutor now accept a parent context for budget enforcement. - Remove DetectionInterval from schedulerPolicy, add MaxJobTypeDuration. Co-Authored-By: Claude Opus 4.6 --- weed/admin/plugin/plugin_scheduler.go | 417 ++++++++++++++------------ 1 file changed, 224 insertions(+), 193 deletions(-) diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 67e463c56..60aac274e 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -17,7 +17,8 @@ var errExecutorAtCapacity = errors.New("executor is at capacity") const ( defaultSchedulerTick = 5 * time.Second - defaultScheduledDetectionInterval = 300 * time.Second + defaultIdleSleepDuration = 17 * time.Minute + defaultMaxJobTypeDuration = 30 * time.Minute defaultScheduledDetectionTimeout = 45 * time.Second defaultScheduledExecutionTimeout = 90 * time.Second defaultScheduledMaxResults int32 = 1000 @@ -31,9 +32,9 @@ const ( ) type schedulerPolicy struct { - DetectionInterval time.Duration DetectionTimeout time.Duration ExecutionTimeout time.Duration + MaxJobTypeDuration time.Duration RetryBackoff time.Duration MaxResults int32 ExecutionConcurrency int @@ -42,33 +43,63 @@ type schedulerPolicy struct { ExecutorReserveBackoff time.Duration } +func (r *Plugin) setSchedulerPhase(phase, jobType string) { + r.schedulerMu.Lock() + r.schedulerPhase = phase + r.currentJobType = jobType + r.schedulerMu.Unlock() +} + func (r *Plugin) schedulerLoop() { defer r.wg.Done() - ticker := time.NewTicker(r.schedulerTick) - defer ticker.Stop() // Try once immediately on startup. - r.runSchedulerTick() + workDetected := r.runSchedulerIteration() for { + if workDetected { + // Immediate re-iteration when work was found. + select { + case <-r.shutdownCh: + return + default: + } + } else { + r.setSchedulerPhase("idle", "") + if !waitForShutdownOrTimer(r.shutdownCh, r.idleSleepDuration) { + return + } + } + select { case <-r.shutdownCh: return - case <-ticker.C: - r.runSchedulerTick() + default: } + + workDetected = r.runSchedulerIteration() } } -func (r *Plugin) runSchedulerTick() { - r.expireStaleJobs(time.Now().UTC()) +func (r *Plugin) runSchedulerIteration() bool { + now := time.Now().UTC() + + r.schedulerMu.Lock() + r.iterationStartedAt = now + r.schedulerMu.Unlock() + + r.expireStaleJobs(now) jobTypes := r.registry.DetectableJobTypes() if len(jobTypes) == 0 { - return + r.finishIteration(false) + return false } active := make(map[string]struct{}, len(jobTypes)) + enabledJobTypes := make([]string, 0, len(jobTypes)) + policies := make(map[string]schedulerPolicy, len(jobTypes)) + for _, jobType := range jobTypes { active[jobType] = struct{}{} @@ -82,19 +113,173 @@ func (r *Plugin) runSchedulerTick() { continue } - if !r.markDetectionDue(jobType, policy.DetectionInterval) { - continue - } - - r.wg.Add(1) - go func(jt string, p schedulerPolicy) { - defer r.wg.Done() - r.runScheduledDetection(jt, p) - }(jobType, policy) + enabledJobTypes = append(enabledJobTypes, jobType) + policies[jobType] = policy } - r.pruneSchedulerState(active) r.pruneDetectorLeases(active) + + if len(enabledJobTypes) == 0 { + r.finishIteration(false) + return false + } + + // Acquire the lock ONCE for the entire iteration. + r.setSchedulerPhase("acquiring_lock", "") + releaseLock, lockErr := r.acquireAdminLock("plugin scheduler iteration") + if lockErr != nil { + glog.Warningf("Plugin scheduler failed to acquire lock: %v", lockErr) + r.appendActivity(JobActivity{ + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduler iteration aborted: failed to acquire lock: %v", lockErr), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + r.finishIteration(false) + return false + } + if releaseLock != nil { + defer releaseLock() + } + + // Load cluster context ONCE for all job types. + clusterContext, err := r.loadSchedulerClusterContext() + if err != nil { + r.appendActivity(JobActivity{ + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduler iteration aborted: %v", err), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + r.finishIteration(false) + return false + } + + // Process each job type sequentially. + anyWorkDetected := false + for _, jobType := range enabledJobTypes { + select { + case <-r.shutdownCh: + r.finishIteration(anyWorkDetected) + return anyWorkDetected + default: + } + + policy := policies[jobType] + r.setSchedulerPhase("processing", jobType) + + if r.runJobTypeIteration(jobType, policy, clusterContext) { + anyWorkDetected = true + } + } + + r.finishIteration(anyWorkDetected) + return anyWorkDetected +} + +func (r *Plugin) finishIteration(workDetected bool) { + now := time.Now().UTC() + r.schedulerMu.Lock() + r.lastIterationEndedAt = now + r.lastIterationWorkDetected = workDetected + r.currentJobType = "" + r.schedulerPhase = "idle" + r.schedulerMu.Unlock() +} + +func (r *Plugin) runJobTypeIteration( + jobType string, + policy schedulerPolicy, + clusterContext *plugin_pb.ClusterContext, +) bool { + budget := policy.MaxJobTypeDuration + if budget <= 0 { + budget = defaultMaxJobTypeDuration + } + ctx, cancel := context.WithTimeout(context.Background(), budget) + defer cancel() + + start := time.Now().UTC() + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: "scheduled detection started", + Stage: "detecting", + OccurredAt: timeToPtr(start), + }) + + if skip, waitingCount, waitingThreshold := r.shouldSkipDetectionForWaitingJobs(jobType, policy); skip { + r.recordSchedulerDetectionSkip(jobType, fmt.Sprintf("waiting backlog %d reached threshold %d", waitingCount, waitingThreshold)) + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection skipped: waiting backlog %d reached threshold %d", waitingCount, waitingThreshold), + Stage: "skipped_waiting_backlog", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + return false + } + + detectionTimeout := policy.DetectionTimeout + if detectionTimeout <= 0 { + detectionTimeout = defaultScheduledDetectionTimeout + } + detCtx, detCancel := context.WithTimeout(ctx, detectionTimeout) + proposals, err := r.RunDetection(detCtx, jobType, clusterContext, policy.MaxResults) + detCancel() + if err != nil { + r.recordSchedulerDetectionError(jobType, err) + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection failed: %v", err), + Stage: "failed", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + return false + } + + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection completed: %d proposal(s)", len(proposals)), + Stage: "detected", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + r.recordSchedulerDetectionSuccess(jobType, len(proposals)) + + filteredByActive, skippedActive := r.filterProposalsWithActiveJobs(jobType, proposals) + if skippedActive > 0 { + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection skipped %d proposal(s) due to active assigned/running jobs", skippedActive), + Stage: "deduped_active_jobs", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + } + + if len(filteredByActive) == 0 { + return false + } + + filtered := r.filterScheduledProposals(filteredByActive) + if len(filtered) != len(filteredByActive) { + r.appendActivity(JobActivity{ + JobType: jobType, + Source: "admin_scheduler", + Message: fmt.Sprintf("scheduled detection deduped %d proposal(s) within this run", len(filteredByActive)-len(filtered)), + Stage: "deduped", + OccurredAt: timeToPtr(time.Now().UTC()), + }) + } + + if len(filtered) == 0 { + return false + } + + r.dispatchScheduledProposals(ctx, jobType, filtered, clusterContext, policy) + return true } func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, error) { @@ -116,9 +301,9 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err } policy := schedulerPolicy{ - DetectionInterval: durationFromSeconds(adminRuntime.DetectionIntervalSeconds, defaultScheduledDetectionInterval), DetectionTimeout: durationFromSeconds(adminRuntime.DetectionTimeoutSeconds, defaultScheduledDetectionTimeout), ExecutionTimeout: defaultScheduledExecutionTimeout, + MaxJobTypeDuration: defaultMaxJobTypeDuration, RetryBackoff: durationFromSeconds(adminRuntime.RetryBackoffSeconds, defaultScheduledRetryBackoff), MaxResults: adminRuntime.MaxJobsPerDetection, ExecutionConcurrency: int(adminRuntime.GlobalExecutionConcurrency), @@ -127,9 +312,6 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err ExecutorReserveBackoff: 200 * time.Millisecond, } - if policy.DetectionInterval < r.schedulerTick { - policy.DetectionInterval = r.schedulerTick - } if policy.MaxResults <= 0 { policy.MaxResults = defaultScheduledMaxResults } @@ -166,28 +348,18 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) { } r.schedulerMu.Lock() - nextDetectionAt := make(map[string]time.Time, len(r.nextDetectionAt)) - for jobType, nextRun := range r.nextDetectionAt { - nextDetectionAt[jobType] = nextRun - } - detectionInFlight := make(map[string]bool, len(r.detectionInFlight)) - for jobType, inFlight := range r.detectionInFlight { - detectionInFlight[jobType] = inFlight - } + currentJobType := r.currentJobType r.schedulerMu.Unlock() states := make([]SchedulerJobTypeState, 0, len(jobTypes)) for _, jobTypeInfo := range jobTypes { jobType := jobTypeInfo.JobType state := SchedulerJobTypeState{ - JobType: jobType, - DetectionInFlight: detectionInFlight[jobType], + JobType: jobType, } - if nextRun, ok := nextDetectionAt[jobType]; ok && !nextRun.IsZero() { - nextRunUTC := nextRun.UTC() - state.NextDetectionAt = &nextRunUTC - } + // Mark as in-flight if this job type is being processed right now. + state.DetectionInFlight = jobType == currentJobType && currentJobType != "" policy, enabled, loadErr := r.loadSchedulerPolicy(jobType) @@ -196,9 +368,9 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) { } else { state.Enabled = enabled if enabled { - state.DetectionIntervalSeconds = secondsFromDuration(policy.DetectionInterval) state.DetectionTimeoutSeconds = secondsFromDuration(policy.DetectionTimeout) state.ExecutionTimeoutSeconds = secondsFromDuration(policy.ExecutionTimeout) + state.MaxJobTypeDurationSeconds = secondsFromDuration(policy.MaxJobTypeDuration) state.MaxJobsPerDetection = policy.MaxResults state.GlobalExecutionConcurrency = policy.ExecutionConcurrency state.PerWorkerExecutionConcurrency = policy.PerWorkerConcurrency @@ -261,49 +433,7 @@ func deriveSchedulerAdminRuntime( } } -func (r *Plugin) markDetectionDue(jobType string, interval time.Duration) bool { - now := time.Now().UTC() - - r.schedulerMu.Lock() - defer r.schedulerMu.Unlock() - - if r.detectionInFlight[jobType] { - return false - } - - nextRun, exists := r.nextDetectionAt[jobType] - if exists && now.Before(nextRun) { - return false - } - - r.nextDetectionAt[jobType] = now.Add(interval) - r.detectionInFlight[jobType] = true - return true -} - -func (r *Plugin) finishDetection(jobType string) { - r.schedulerMu.Lock() - delete(r.detectionInFlight, jobType) - r.schedulerMu.Unlock() -} - -func (r *Plugin) pruneSchedulerState(activeJobTypes map[string]struct{}) { - r.schedulerMu.Lock() - defer r.schedulerMu.Unlock() - - for jobType := range r.nextDetectionAt { - if _, ok := activeJobTypes[jobType]; !ok { - delete(r.nextDetectionAt, jobType) - delete(r.detectionInFlight, jobType) - } - } -} - func (r *Plugin) clearSchedulerJobType(jobType string) { - r.schedulerMu.Lock() - delete(r.nextDetectionAt, jobType) - delete(r.detectionInFlight, jobType) - r.schedulerMu.Unlock() r.clearDetectorLease(jobType, "") } @@ -318,116 +448,6 @@ func (r *Plugin) pruneDetectorLeases(activeJobTypes map[string]struct{}) { } } -func (r *Plugin) runScheduledDetection(jobType string, policy schedulerPolicy) { - defer r.finishDetection(jobType) - - releaseLock, lockErr := r.acquireAdminLock(fmt.Sprintf("plugin scheduled detection %s", jobType)) - if lockErr != nil { - r.recordSchedulerDetectionError(jobType, lockErr) - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection aborted: failed to acquire lock: %v", lockErr), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - return - } - if releaseLock != nil { - defer releaseLock() - } - - start := time.Now().UTC() - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: "scheduled detection started", - Stage: "detecting", - OccurredAt: timeToPtr(start), - }) - - if skip, waitingCount, waitingThreshold := r.shouldSkipDetectionForWaitingJobs(jobType, policy); skip { - r.recordSchedulerDetectionSkip(jobType, fmt.Sprintf("waiting backlog %d reached threshold %d", waitingCount, waitingThreshold)) - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection skipped: waiting backlog %d reached threshold %d", waitingCount, waitingThreshold), - Stage: "skipped_waiting_backlog", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - return - } - - clusterContext, err := r.loadSchedulerClusterContext() - if err != nil { - r.recordSchedulerDetectionError(jobType, err) - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection aborted: %v", err), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - return - } - - ctx, cancel := context.WithTimeout(context.Background(), policy.DetectionTimeout) - proposals, err := r.RunDetection(ctx, jobType, clusterContext, policy.MaxResults) - cancel() - if err != nil { - r.recordSchedulerDetectionError(jobType, err) - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection failed: %v", err), - Stage: "failed", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - return - } - - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection completed: %d proposal(s)", len(proposals)), - Stage: "detected", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - r.recordSchedulerDetectionSuccess(jobType, len(proposals)) - - filteredByActive, skippedActive := r.filterProposalsWithActiveJobs(jobType, proposals) - if skippedActive > 0 { - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection skipped %d proposal(s) due to active assigned/running jobs", skippedActive), - Stage: "deduped_active_jobs", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - } - - if len(filteredByActive) == 0 { - return - } - - filtered := r.filterScheduledProposals(filteredByActive) - if len(filtered) != len(filteredByActive) { - r.appendActivity(JobActivity{ - JobType: jobType, - Source: "admin_scheduler", - Message: fmt.Sprintf("scheduled detection deduped %d proposal(s) within this run", len(filteredByActive)-len(filtered)), - Stage: "deduped", - OccurredAt: timeToPtr(time.Now().UTC()), - }) - } - - if len(filtered) == 0 { - return - } - - r.dispatchScheduledProposals(jobType, filtered, clusterContext, policy) -} - func (r *Plugin) loadSchedulerClusterContext() (*plugin_pb.ClusterContext, error) { if r.clusterContextProvider == nil { return nil, fmt.Errorf("cluster context provider is not configured") @@ -447,6 +467,7 @@ func (r *Plugin) loadSchedulerClusterContext() (*plugin_pb.ClusterContext, error } func (r *Plugin) dispatchScheduledProposals( + parentCtx context.Context, jobType string, proposals []*plugin_pb.JobProposal, clusterContext *plugin_pb.ClusterContext, @@ -460,6 +481,9 @@ func (r *Plugin) dispatchScheduledProposals( case <-r.shutdownCh: close(jobQueue) return + case <-parentCtx.Done(): + close(jobQueue) + return default: jobQueue <- job } @@ -485,6 +509,8 @@ func (r *Plugin) dispatchScheduledProposals( select { case <-r.shutdownCh: return + case <-parentCtx.Done(): + return default: } @@ -492,6 +518,8 @@ func (r *Plugin) dispatchScheduledProposals( select { case <-r.shutdownCh: return + case <-parentCtx.Done(): + return default: } @@ -515,7 +543,7 @@ func (r *Plugin) dispatchScheduledProposals( break } - err := r.executeScheduledJobWithExecutor(executor, job, clusterContext, policy) + err := r.executeScheduledJobWithExecutor(parentCtx, executor, job, clusterContext, policy) release() if errors.Is(err, errExecutorAtCapacity) { r.trackExecutionQueued(job) @@ -680,6 +708,7 @@ func schedulerWorkerExecutionLimit(executor *WorkerSession, jobType string, poli } func (r *Plugin) executeScheduledJobWithExecutor( + parentCtx context.Context, executor *WorkerSession, job *plugin_pb.JobSpec, clusterContext *plugin_pb.ClusterContext, @@ -695,10 +724,12 @@ func (r *Plugin) executeScheduledJobWithExecutor( select { case <-r.shutdownCh: return fmt.Errorf("plugin is shutting down") + case <-parentCtx.Done(): + return parentCtx.Err() default: } - execCtx, cancel := context.WithTimeout(context.Background(), policy.ExecutionTimeout) + execCtx, cancel := context.WithTimeout(parentCtx, policy.ExecutionTimeout) _, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt)) cancel() if err == nil {