Merge 050f786 into blathers/backport-release-26.1-159403

Author: blathers-crl[bot]

pkg/util/admission/admission.go

@@ -155,6 +155,15 @@ func (bq burstQualification) String() string {
 	}
 }
 
+// Mutex ordering between requester and granter:
+//
+// The requester and granter call into each other. To prevent deadlock due to
+// mutex cycles, any mutex in granter is ordered before any mutex in
+// requester. Therefore, a requester must not hold its own mutex when calling
+// into granter. Of course, a granter could choose to release its own mutex
+// before calling into requester, but it is not necessary for deadlock
+// prevention.
+
 // requester is an interface implemented by an object that orders admission
 // work for a particular WorkKind. See WorkQueue for the implementation of
 // requester.

pkg/util/admission/io_load_listener.go

@@ -583,6 +583,8 @@ func (io *ioLoadListener) pebbleMetricsTick(ctx context.Context, metrics StoreMe
 	io.cumFlushWriteThroughput = metrics.Flush.WriteThroughput
 	// We assume that the system is loaded if there is less than unlimited tokens
 	// available.
+	//
+	// TODO(sumeer): this condition should also incorporate disk byte tokens.
 	return io.totalNumByteTokens < unlimitedTokens || io.totalNumElasticByteTokens < unlimitedTokens
 }
 

pkg/util/admission/snapshot_queue.go

@@ -89,6 +89,9 @@ type snapshotRequester interface {
 type SnapshotQueue struct {
 	snapshotGranter granter
 	mu              struct {
+		// Reminder: this mutex must not be held when calling into
+		// snapshotGranter, as documented near the declaration of requester and
+		// granter.
 		syncutil.Mutex
 		q *queue.Queue[*snapshotWorkItem]
 	}
@@ -164,6 +167,13 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error {
 		return nil
 	}
 	// We were unable to get tokens for admission, so we queue.
+	//
+	// Reminder: there is a race here where a call to hasWaitingRequests after
+	// tryGet and before the mutex is acquired and the item is added to the
+	// queue will not see the waiting request. Such a race would result in an
+	// end state where the granter has resources and the requester has waiting
+	// requests. This is harmless since hasWaitingRequests is called
+	// periodically at a high enough frequency when tokens are limited.
 	shouldRelease := true
 	item := newSnapshotWorkItem(count)
 	defer func() {
@@ -182,11 +192,15 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error {
 	select {
 	case <-ctx.Done():
 		waitDur := timeutil.Since(item.enqueueingTime).Nanoseconds()
+		// INVARIANT: tokensToReturn >= 0, since item.count >= 0.
+		var tokensToReturn int64
 		func() {
 			s.mu.Lock()
 			defer s.mu.Unlock()
 			if !item.mu.inQueue {
-				s.snapshotGranter.returnGrant(item.count)
+				// NB: we must call snapshotGranter.returnGrant after releasing the
+				// mutex.
+				tokensToReturn = item.count
 			}
 			// TODO(aaditya): Ideally, we also remove the item from the actual queue.
 			// Right now, if we cancel the work, it remains in the queue. A call to
@@ -196,6 +210,9 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error {
 			// non-ideal behavior, but still provides accurate token accounting.
 			item.mu.cancelled = true
 		}()
+		if tokensToReturn != 0 {
+			s.snapshotGranter.returnGrant(tokensToReturn)
+		}
 		shouldRelease = false
 		s.metrics.WaitDurations.RecordValue(waitDur)
 		var deadlineSubstring string