refactor concurrent work limiting in sync in x/sync

x/sync/syncmanager.go

@@ -25,7 +25,6 @@ const (
 )
 
 var (
-	token                         = struct{}{}
 	ErrAlreadyStarted             = errors.New("cannot start a StateSyncManager that has already been started")
 	ErrAlreadyClosed              = errors.New("StateSyncManager is closed")
 	ErrNoClientProvided           = errors.New("client is a required field of the sync config")
@@ -80,7 +79,7 @@ type StateSyncManager struct {
 	unprocessedWork *syncWorkHeap
 	// Signalled when:
 	// - An item is added to [unprocessedWork].
-	// - There are no more items in [unprocessedWork] and [processingWorkItems] is 0.
+	// - An item is added to [processedWork].
 	// - Close() is called.
 	// [workLock] is its inner lock.
 	unprocessedWorkCond sync.Cond
@@ -93,9 +92,6 @@ type StateSyncManager struct {
 	// - [workToBeDone] and [completedWork] are closed.
 	syncDoneChan chan struct{}
 
-	// Rate-limits the number of concurrently processing work items.
-	workTokens chan struct{}
-
 	errLock sync.Mutex
 	// If non-nil, there was a fatal error.
 	// [errLock] must be held when accessing [fatalError].
@@ -134,14 +130,9 @@ func NewStateSyncManager(config StateSyncConfig) (*StateSyncManager, error) {
 		syncDoneChan:    make(chan struct{}),
 		unprocessedWork: newSyncWorkHeap(2 * config.SimultaneousWorkLimit),
 		processedWork:   newSyncWorkHeap(2 * config.SimultaneousWorkLimit),
-		workTokens:      make(chan struct{}, config.SimultaneousWorkLimit),
 	}
 	m.unprocessedWorkCond.L = &m.workLock
 
-	// fill the work tokens channel with work tokens
-	for i := 0; i < config.SimultaneousWorkLimit; i++ {
-		m.workTokens <- token
-	}
 	return m, nil
 }
 
@@ -170,8 +161,8 @@ func (m *StateSyncManager) StartSyncing(ctx context.Context) error {
 func (m *StateSyncManager) sync(ctx context.Context) {
 	defer func() {
 		// Note we release [m.workLock] before calling Close()
-		// because Close() will try to acquire [m.workLock].
-		// Invariant: [m.workLock] is held when we return from this goroutine.
+		// because Close() will acquire [m.workLock].
+		// Invariant: [m.workLock] is held when this goroutine begins.
 		m.workLock.Unlock()
 		m.Close()
 	}()
@@ -183,6 +174,12 @@ func (m *StateSyncManager) sync(ctx context.Context) {
 		if ctx.Err() != nil { // [m] is closed.
 			return // [m.workLock] released by defer.
 		}
+		if m.processingWorkItems >= m.config.SimultaneousWorkLimit {
+			// We're already processing the maximum number of work items.
+			// Wait until one of them finishes.
+			m.unprocessedWorkCond.Wait()
+			continue
+		}
 		if m.unprocessedWork.Len() == 0 {
 			if m.processingWorkItems == 0 {
 				// There's no work to do, and there are no work items being processed
@@ -231,23 +228,12 @@ func (m *StateSyncManager) Close() {
 // Processes [item] by fetching and applying a change or range proof.
 // Assumes [m.workLock] is not held.
 func (m *StateSyncManager) doWork(ctx context.Context, item *syncWorkItem) {
-	// Wait until we get a work token or we close.
-	select {
-	case <-m.workTokens:
-	case <-ctx.Done():
-		// [m] is closed and sync() is returning so don't care about cleanup.
-		return
-	}
-
 	defer func() {
-		m.workTokens <- token
 		m.workLock.Lock()
+		defer m.workLock.Unlock()
+
 		m.processingWorkItems--
-		if m.processingWorkItems == 0 && m.unprocessedWork.Len() == 0 {
-			// There are no processing or unprocessed work items so we're done.
-			m.unprocessedWorkCond.Signal()
-		}
-		m.workLock.Unlock()
+		m.unprocessedWorkCond.Signal()
 	}()
 
 	if item.LocalRootID == ids.Empty {