Add round robin policy for the cc > 3

pkg/activator/net/lb_policy.go

@@ -21,6 +21,7 @@ package net
 import (
 	"context"
 	"math/rand"
+	"sync"
 )
 
 // lbPolicy is a functor that selects a target pod from the list, or (noop, nil) if
@@ -84,3 +85,34 @@ func firstAvailableLBPolicy(ctx context.Context, targets []*podTracker) (func(),
 	}
 	return noop, nil
 }
+
+type roundRobinPolicyT struct {
+	mu  sync.Mutex
+	idx int
+}
+
+func newRoundRobinPolicy() lbPolicy {
+	rrp := roundRobinPolicyT{}
+	return func(ctx context.Context, targets []*podTracker) (func(), *podTracker) {
+		rrp.mu.Lock()
+		defer rrp.mu.Unlock()
+		// The number of trackers might have shrunk, so reset to 0.
+		l := len(targets)
+		if rrp.idx >= l {
+			rrp.idx = 0
+		}
+
+		// Now for |targets| elements and check every next one in
+		// round robin fashion.
+		for i := 0; i < l; i++ {
+			p := (rrp.idx + i) % l
+			if cb, ok := targets[p].Reserve(ctx); ok {
+				// We want to start with the next index.
+				rrp.idx = p + 1
+				return cb, targets[p]
+			}
+		}
+		// We exhausted all the options...
+		return noop, nil
+	}
+}

pkg/activator/net/lb_policy_test.go

@@ -158,6 +158,38 @@ func TestFirstAvailable(t *testing.T) {
 	})
 }
 
+func TestRoundRobin(t *testing.T) {
+	rrp := newRoundRobinPolicy()
+	podTrackers := makeTrackers(3, 1)
+	cb, pt := rrp(context.Background(), podTrackers)
+	t.Cleanup(cb)
+	if got, want := pt, podTrackers[0]; got != want {
+		t.Fatalf("Tracker = %v, want: %v", got, want)
+	}
+	cb, pt = rrp(context.Background(), podTrackers)
+	t.Cleanup(cb)
+	if got, want := pt, podTrackers[1]; got != want {
+		t.Fatalf("Tracker = %v, want: %v", got, want)
+	}
+	cb, pt = rrp(context.Background(), podTrackers)
+	if got, want := pt, podTrackers[2]; got != want {
+		t.Fatalf("Tracker = %v, want: %v", got, want)
+	}
+	_, pt = rrp(context.Background(), podTrackers)
+	t.Cleanup(cb)
+	if pt != nil {
+		t.Fatal("Wanted nil, got: ", pt)
+	}
+
+	// Reset last one.
+	cb()
+	cb, pt = rrp(context.Background(), podTrackers)
+	if got, want := pt, podTrackers[2]; got != want {
+		t.Fatalf("Tracker = %v, want: %v", got, want)
+	}
+	t.Cleanup(cb)
+}
+
 func BenchmarkPolicy(b *testing.B) {
 	for _, test := range []struct {
 		name   string
@@ -171,6 +203,9 @@ func BenchmarkPolicy(b *testing.B) {
 	}, {
 		name:   "first-available",
 		policy: firstAvailableLBPolicy,
+	}, {
+		name:   "round-robin",
+		policy: newRoundRobinPolicy(),
 	}} {
 		for _, n := range []int{1, 2, 3, 10, 100} {
 			b.Run(fmt.Sprintf("%s-%d-trackers-sequential", test.name, n), func(b *testing.B) {

pkg/activator/net/throttler.go

@@ -163,12 +163,18 @@ func newRevisionThrottler(revID types.NamespacedName,
 		revBreaker breaker
 		lbp        lbPolicy
 	)
-	if containerConcurrency == 0 {
+	switch {
+	case containerConcurrency == 0:
 		revBreaker = newInfiniteBreaker(logger)
 		lbp = randomChoice2Policy
-	} else {
+	case containerConcurrency <= 3:
+		// For very low CC values use first available pod.
 		revBreaker = queue.NewBreaker(breakerParams)
 		lbp = firstAvailableLBPolicy
+	default:
+		// Otherwise RR.
+		revBreaker = queue.NewBreaker(breakerParams)
+		lbp = newRoundRobinPolicy()
 	}
 	return &revisionThrottler{
 		revID:                revID,