Function TestRing_ShuffleShardWithLookback_CachingConcurrency

ring/ring_test.go:3607–3662

(t *testing.T)

}

func TestRing_ShuffleShardWithLookback_CachingConcurrency(t *testing.T) {
	const (
		numWorkers           = 10
		numRequestsPerWorker = 1000
	)

	now := time.Now()
	cfg := Config{KVStore: kv.Config{}, ReplicationFactor: 1, ZoneAwarenessEnabled: true}
	registry := prometheus.NewRegistry()
	ring, err := NewWithStoreClientAndStrategy(cfg, testRingName, testRingKey, nil, NewDefaultReplicationStrategy(), registry, log.NewNopLogger())
	require.NoError(t, err)

	gen := initTokenGenerator(t)

	// Add some instances to the ring.
	ringDesc := &Desc{Ingesters: map[string]InstanceDesc{
		"instance-1": generateRingInstanceWithInfo("instance-1", "zone-a", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)),
		"instance-2": makeReadOnly(generateRingInstanceWithInfo("instance-2", "zone-a", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)), now),
		"instance-3": generateRingInstanceWithInfo("instance-3", "zone-b", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)),
		"instance-4": generateRingInstanceWithInfo("instance-4", "zone-b", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)),
		"instance-5": generateRingInstanceWithInfo("instance-5", "zone-c", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)),
		"instance-6": makeReadOnly(generateRingInstanceWithInfo("instance-6", "zone-c", gen.GenerateTokens(128, nil), now.Add(-2*time.Hour)), now.Add(-2*time.Hour)),
	}}

	ring.updateRingState(ringDesc)

	for _, shardSize := range []int{3, 0} {
		t.Run(fmt.Sprintf("shardSize=%d", shardSize), func(t *testing.T) {
			// Start the workers.
			wg := sync.WaitGroup{}
			wg.Add(numWorkers)

			for w := 0; w < numWorkers; w++ {
				go func(workerID int) {
					defer wg.Done()

					// Get the subring once. This is the one expected from subsequent requests.
					userID := fmt.Sprintf("user-%d", workerID)
					expected := ring.ShuffleShardWithLookback(userID, shardSize, time.Hour, now)

					for r := 0; r < numRequestsPerWorker; r++ {
						actual := ring.ShuffleShardWithLookback(userID, shardSize, time.Hour, now)
						require.Equal(t, expected, actual)

						// Get the subring for a new user each time too, in order to stress the setter too
						// (if we only read from the cache there's no read/write concurrent access).
						ring.ShuffleShardWithLookback(fmt.Sprintf("stress-%d", r), shardSize, time.Hour, now)
					}
				}(w)
			}

			// Wait until all workers have done.
			wg.Wait()
		})
	}
}

func BenchmarkRing_ShuffleShard(b *testing.B) {