backport-2.1: storage: use FIFO order in queues on equal priorities

pkg/storage/queue.go

@@ -62,6 +62,7 @@ type processCallback func(error)
 // processing state.
 type replicaItem struct {
 	value roachpb.RangeID
+	seq   int // enforce FIFO order for equal priorities
 
 	// fields used when a replicaItem is enqueued in a priority queue.
 	priority float64
@@ -87,34 +88,45 @@ func (i *replicaItem) registerCallback(cb processCallback) {
 }
 
 // A priorityQueue implements heap.Interface and holds replicaItems.
-type priorityQueue []*replicaItem
+type priorityQueue struct {
+	seqGen int
+	sl     []*replicaItem
+}
 
-func (pq priorityQueue) Len() int { return len(pq) }
+func (pq priorityQueue) Len() int { return len(pq.sl) }
 
 func (pq priorityQueue) Less(i, j int) bool {
+	a, b := pq.sl[i], pq.sl[j]
+	if a.priority == b.priority {
+		// When priorities are equal, we want the lower sequence number to show
+		// up first (FIFO).
+		return a.seq < b.seq
+	}
 	// We want Pop to give us the highest, not lowest, priority so we use greater than here.
-	return pq[i].priority > pq[j].priority
+	return a.priority > b.priority
 }
 
 func (pq priorityQueue) Swap(i, j int) {
-	pq[i], pq[j] = pq[j], pq[i]
-	pq[i].index, pq[j].index = i, j
+	pq.sl[i], pq.sl[j] = pq.sl[j], pq.sl[i]
+	pq.sl[i].index, pq.sl[j].index = i, j
 }
 
 func (pq *priorityQueue) Push(x interface{}) {
-	n := len(*pq)
+	n := len(pq.sl)
 	item := x.(*replicaItem)
 	item.index = n
-	*pq = append(*pq, item)
+	pq.seqGen++
+	item.seq = pq.seqGen
+	pq.sl = append(pq.sl, item)
 }
 
 func (pq *priorityQueue) Pop() interface{} {
-	old := *pq
+	old := pq.sl
 	n := len(old)
 	item := old[n-1]
 	item.index = -1 // for safety
 	old[n-1] = nil  // for gc
-	*pq = old[0 : n-1]
+	pq.sl = old[0 : n-1]
 	return item
 }
 
@@ -525,7 +537,7 @@ func (bq *baseQueue) addInternalLocked(
 	// If adding this replica has pushed the queue past its maximum size,
 	// remove the lowest priority element.
 	if pqLen := bq.mu.priorityQ.Len(); pqLen > bq.maxSize {
-		bq.removeLocked(bq.mu.priorityQ[pqLen-1])
+		bq.removeLocked(bq.mu.priorityQ.sl[pqLen-1])
 	}
 	// Signal the processLoop that a replica has been added.
 	select {

pkg/storage/queue_test.go

@@ -128,14 +128,15 @@ func TestQueuePriorityQueue(t *testing.T) {
 	// establish the priority queue (heap) invariants.
 	const count = 3
 	expRanges := make([]roachpb.RangeID, count+1)
-	pq := make(priorityQueue, count)
+	pq := priorityQueue{}
+	pq.sl = make([]*replicaItem, count)
 	for i := 0; i < count; {
-		pq[i] = &replicaItem{
+		pq.sl[i] = &replicaItem{
 			value:    roachpb.RangeID(i),
 			priority: float64(i),
 			index:    i,
 		}
-		expRanges[3-i] = pq[i].value
+		expRanges[3-i] = pq.sl[i].value
 		i++
 	}
 	heap.Init(&pq)
@@ -293,6 +294,48 @@ func TestBaseQueueAddUpdateAndRemove(t *testing.T) {
 	}
 }
 
+// TestBaseQueueSamePriorityFIFO verifies that if multiple items are queued at
+// the same priority, they will be processes in first-in-first-out order.
+// This avoids starvation scenarios, in particular in the Raft snapshot queue.
+//
+// See:
+// https://github.com/cockroachdb/cockroach/issues/31947#issuecomment-434383267
+func TestBaseQueueSamePriorityFIFO(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	tc := testContext{}
+	stopper := stop.NewStopper()
+	ctx := context.Background()
+	defer stopper.Stop(ctx)
+	tc.Start(t, stopper)
+
+	repls := createReplicas(t, &tc, 5)
+
+	testQueue := &testQueueImpl{
+		shouldQueueFn: func(now hlc.Timestamp, r *Replica) (shouldQueue bool, priority float64) {
+			t.Fatal("unexpected call to shouldQueue")
+			return false, 0.0
+		},
+	}
+
+	bq := makeTestBaseQueue("test", testQueue, tc.store, tc.gossip, queueConfig{maxSize: 100})
+
+	for _, repl := range repls {
+		added, err := bq.Add(repl, 0.0)
+		if err != nil {
+			t.Fatal(errors.Wrap(err, repl.String()))
+		}
+		if !added {
+			t.Fatalf("%v not added", repl)
+		}
+	}
+	for _, expRepl := range repls {
+		actRepl := bq.pop()
+		if actRepl != expRepl {
+			t.Fatalf("expected %v, got %v", expRepl, actRepl)
+		}
+	}
+}
+
 // TestBaseQueueAdd verifies that calling Add() directly overrides the
 // ShouldQueue method.
 func TestBaseQueueAdd(t *testing.T) {