From c97125287182b230d56cff8b0f012de3b11eb04b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 5 Nov 2026 05:02:41 -0800 Subject: [PATCH] admin: make the cluster lock's fairness window configurable (#12426) * admin: make the cluster lock's fairness window configurable Scheduled jobs hold the cluster admin lock for their whole run. Once the lock had been held for a minute, every new job waited until all running jobs had released it, so with hour-long EC jobs dispatch ran in waves and most workers sat idle between them. -adminLock.fairnessWindow keeps the one-minute default; 0 lets new jobs join the shared hold, and shell clients get the lock when the scheduler next drains. Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi * admin: wait for the blocked acquire before disabling the fairness window The wake-up test slept 50 ms before SetFairnessWindow(0), so a late acquire that started after the change joined without exercising the Broadcast. Wait until it is registered as a waiter instead. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/admin/dash/admin_lock_manager.go | 26 +++++++- weed/admin/dash/admin_lock_manager_test.go | 75 +++++++++++++++++++++- weed/admin/dash/admin_server.go | 9 +++ weed/command/admin.go | 6 ++ 4 files changed, 110 insertions(+), 6 deletions(-) diff --git a/weed/admin/dash/admin_lock_manager.go b/weed/admin/dash/admin_lock_manager.go index 419da8abbf..552dca5a98 100644 --- a/weed/admin/dash/admin_lock_manager.go +++ b/weed/admin/dash/admin_lock_manager.go @@ -23,7 +23,7 @@ const ( // After the lock has been held continuously for this long (overlapping // reference-counted holds can keep it held indefinitely), new Acquire // calls wait for a full release so shell clients are not starved. - defaultAdminLockFairnessWindow = time.Minute + DefaultAdminLockFairnessWindow = time.Minute ) // adminLocker is the subset of exclusive_locks.ExclusiveLocker the manager uses. @@ -67,7 +67,7 @@ func NewAdminLockManager(masterClient *wdclient.MasterClient, clientName string) locker: exclusive_locks.NewExclusiveLocker(masterClient, adminLockName), clientName: clientName, yieldWindow: defaultAdminLockYieldWindow, - fairnessWindow: defaultAdminLockFairnessWindow, + fairnessWindow: DefaultAdminLockFairnessWindow, } manager.cond = sync.NewCond(&manager.mu) return manager @@ -93,7 +93,7 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) { delete(m.waiters, waiter) m.mu.Unlock() }() - for m.acquiring || m.releasing || (m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow) { + for m.acquiring || m.releasing || m.fairnessDueLocked() { m.cond.Wait() } if reason != "" { @@ -134,6 +134,26 @@ func (m *AdminLockManager) Acquire(reason string) (func(), error) { return m.Release, nil } +// fairnessDueLocked reports whether new holders must wait for a full release +// because the lock has been held past the fairness window. A window of zero +// or less never forces one. Caller holds m.mu. +func (m *AdminLockManager) fairnessDueLocked() bool { + return m.fairnessWindow > 0 && m.holdCount > 0 && time.Since(m.heldSince) > m.fairnessWindow +} + +// SetFairnessWindow changes how long overlapping holds may keep the lock +// before new Acquire calls wait for a full release; zero or less disables +// the forced release, leaving shell clients the gaps between holds. +func (m *AdminLockManager) SetFairnessWindow(window time.Duration) { + if m == nil { + return + } + m.mu.Lock() + m.fairnessWindow = window + m.cond.Broadcast() + m.mu.Unlock() +} + func (m *AdminLockManager) Release() { if m == nil || m.locker == nil { return diff --git a/weed/admin/dash/admin_lock_manager_test.go b/weed/admin/dash/admin_lock_manager_test.go index de08c78602..746a447c66 100644 --- a/weed/admin/dash/admin_lock_manager_test.go +++ b/weed/admin/dash/admin_lock_manager_test.go @@ -89,7 +89,7 @@ func TestAdminLockManagerRefCountsOverlappingHolds(t *testing.T) { t.Parallel() locker := &fakeAdminLocker{} - m := newTestLockManager(locker, defaultAdminLockYieldWindow, defaultAdminLockFairnessWindow) + m := newTestLockManager(locker, defaultAdminLockYieldWindow, DefaultAdminLockFairnessWindow) release1, _ := m.Acquire("a") release2, _ := m.Acquire("b") @@ -112,7 +112,7 @@ func TestAdminLockManagerYieldsBeforeReacquire(t *testing.T) { t.Parallel() locker := &fakeAdminLocker{} - m := newTestLockManager(locker, 100*time.Millisecond, defaultAdminLockFairnessWindow) + m := newTestLockManager(locker, 100*time.Millisecond, DefaultAdminLockFairnessWindow) release, _ := m.Acquire("first") release() @@ -178,7 +178,7 @@ func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) { t.Parallel() locker := &fakeAdminLocker{releaseDelay: 50 * time.Millisecond} - m := newTestLockManager(locker, time.Millisecond, defaultAdminLockFairnessWindow) + m := newTestLockManager(locker, time.Millisecond, DefaultAdminLockFairnessWindow) release1, _ := m.Acquire("first") releaseDone := make(chan struct{}) @@ -201,3 +201,72 @@ func TestAdminLockManagerAcquireWaitsForReleaseInFlight(t *testing.T) { t.Fatalf("RequestLock ran while ReleaseLock was still in flight") } } + +func TestAdminLockManagerZeroFairnessWindowKeepsJoining(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{} + m := newTestLockManager(locker, time.Millisecond, 0) + + release1, _ := m.Acquire("long-job") + m.mu.Lock() + m.heldSince = time.Now().Add(-time.Hour) + m.mu.Unlock() + + acquired := make(chan func(), 1) + go func() { + release2, _ := m.Acquire("late-joiner") + acquired <- release2 + }() + + select { + case release2 := <-acquired: + release2() + case <-time.After(time.Second): + t.Fatalf("late acquire waited for a full release with the fairness window disabled") + } + release1() + + if requests, releases := locker.counts(); requests != 1 || releases != 1 { + t.Fatalf("expected one shared hold, got requests=%d releases=%d", requests, releases) + } +} + +func TestAdminLockManagerSetFairnessWindowWakesWaiters(t *testing.T) { + t.Parallel() + + locker := &fakeAdminLocker{} + m := newTestLockManager(locker, time.Millisecond, time.Millisecond) + + release1, _ := m.Acquire("long-job") + m.mu.Lock() + m.heldSince = time.Now().Add(-time.Second) + m.mu.Unlock() + + acquired := make(chan func(), 1) + go func() { + release2, _ := m.Acquire("late-joiner") + acquired <- release2 + }() + + require.Eventually(t, func() bool { return len(m.Status().Waiters) == 1 }, time.Second, time.Millisecond) + select { + case <-acquired: + t.Fatalf("late acquire joined a hold older than the fairness window") + default: + } + + m.SetFairnessWindow(0) + + select { + case release2 := <-acquired: + release2() + case <-time.After(time.Second): + t.Fatalf("disabling the fairness window did not admit the waiting acquire") + } + release1() + + if requests, releases := locker.counts(); requests != 1 || releases != 1 { + t.Fatalf("expected one shared hold, got requests=%d releases=%d", requests, releases) + } +} diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 27958dec6d..52c518cca2 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -1708,6 +1708,15 @@ func (s *AdminServer) GetPlugin() *adminplugin.Plugin { return s.plugin } +// SetAdminLockFairnessWindow sets the cluster admin lock's fairness window +// (see AdminLockManager.SetFairnessWindow). +func (s *AdminServer) SetAdminLockFairnessWindow(window time.Duration) { + if s == nil || s.pluginLock == nil { + return + } + s.pluginLock.SetFairnessWindow(window) +} + func (s *AdminServer) acquirePluginLock(reason string) (func(), error) { if s == nil || s.pluginLock == nil { return func() {}, nil diff --git a/weed/command/admin.go b/weed/command/admin.go index 59ac89bdb0..b2e6e7854e 100644 --- a/weed/command/admin.go +++ b/weed/command/admin.go @@ -65,6 +65,8 @@ type AdminOptions struct { cpuProfile *string memProfile *string uploadMaxSizeMB *int + // nil for callers other than runAdmin; the lock keeps its default then + lockFairnessWindow *time.Duration // workerGrpcListener, when set, is a listener already bound to grpcPort by // the caller. `weed mini` reserves the port this way because the admin @@ -127,6 +129,7 @@ func init() { a.cpuProfile = cmdAdmin.Flag.String("cpuprofile", "", "cpu profile output file") a.memProfile = cmdAdmin.Flag.String("memprofile", "", "memory profile output file") a.uploadMaxSizeMB = cmdAdmin.Flag.Int("uploadMaxSizeMB", dash.DefaultUploadMaxSizeMB, "maximum total size in MB of one file browser upload (0 for no limit)") + a.lockFairnessWindow = cmdAdmin.Flag.Duration("adminLock.fairnessWindow", dash.DefaultAdminLockFairnessWindow, "once scheduled jobs have held the cluster admin lock this long, new jobs wait until every holder releases it so shell clients get a turn; 0 lets jobs keep joining the hold, and shell clients then get the lock when the scheduler next drains") } var cmdAdmin = &Command{ @@ -505,6 +508,9 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool, if options.uploadMaxSizeMB != nil { adminServer.SetUploadMaxSize(int64(*options.uploadMaxSizeMB) << 20) } + if options.lockFairnessWindow != nil { + adminServer.SetAdminLockFairnessWindow(*options.lockFairnessWindow) + } if err := adminServer.ApplyPluginConfigFromToml(util.GetViper()); err != nil { return fmt.Errorf("apply admin.toml to plugin config: %w", err)