Merge cockroachdb#157259

157259: kvcoord: deflake rangefeed tests r=pav-kv,tbg,sumeerbhola a=stevendanna

This aims at fixing recent failures in

    TestRangeFeedMetricsManagement
    TestMuxRangeFeedCatchupScanQuotaReleased

A clear root cause was found for the former while the latter appears
to be related to slowdowns as a result of recreating catchup scan iterators
when running under the race-detector.

Fixes cockroachdb#156852
Fixes cockroachdb#157003
Fixes cockroachdb#157064



Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed_test.go

@@ -29,7 +29,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
-	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -129,14 +128,19 @@ func rangeFeed(
 	startFrom hlc.Timestamp,
 	onValue func(event kvcoord.RangeFeedMessage),
 	opts ...kvcoord.RangeFeedOption,
-) func() {
+) (chan error, func()) {
 	ds := dsI.(*kvcoord.DistSender)
 	events := make(chan kvcoord.RangeFeedMessage)
 	ctx, cancel := context.WithCancel(context.WithValue(context.Background(), testFeedCtxKey{}, struct{}{}))
 
+	errCh := make(chan error, 1)
 	g := ctxgroup.WithContext(ctx)
-	g.GoCtx(func(ctx context.Context) (err error) {
-		return ds.RangeFeed(ctx, []kvcoord.SpanTimePair{{Span: sp, StartAfter: startFrom}}, events, opts...)
+	g.GoCtx(func(ctx context.Context) error {
+		err := ds.RangeFeed(ctx, []kvcoord.SpanTimePair{{Span: sp, StartAfter: startFrom}}, events, opts...)
+		if err != nil {
+			errCh <- err
+		}
+		return err
 	})
 	g.GoCtx(func(ctx context.Context) error {
 		for {
@@ -149,7 +153,7 @@ func rangeFeed(
 		}
 	})
 
-	return func() {
+	return errCh, func() {
 		cancel()
 		_ = g.Wait()
 	}
@@ -176,19 +180,17 @@ func observeNValues(n int) (chan struct{}, func(ev kvcoord.RangeFeedMessage)) {
 	}
 }
 
-func channelWaitWithTimeout(t *testing.T, ch chan struct{}) {
+func channelWaitWithTimeout(t *testing.T, ch chan struct{}, errCh chan error) {
 	t.Helper()
-	timeOut := 30 * time.Second
-	if util.RaceEnabled {
-		timeOut *= 10
-	}
-	if syncutil.DeadlockEnabled {
-		timeOut = 2 * deadlock.Opts.DeadlockTimeout
-	}
+	timeOut := testutils.SucceedsSoonDuration()
 	select {
 	case <-ch:
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("unexpected error while waiting on channel: %v", err)
+		}
 	case <-time.After(timeOut):
-		t.Fatal("test timed out")
+		t.Fatalf("test timed out after %s", timeOut)
 	}
 }
 
@@ -220,7 +222,7 @@ func TestMuxRangeFeedConnectsToNodeOnce(t *testing.T) {
 	// test cluster nodes.
 	sqlDB.ExecMultiple(t,
 		`SET CLUSTER SETTING kv.rangefeed.enabled = true`,
-		`ALTER DATABASE defaultdb  CONFIGURE ZONE USING num_replicas = 1`,
+		`ALTER DATABASE defaultdb CONFIGURE ZONE USING num_replicas = 1`,
 		`CREATE TABLE foo (key INT PRIMARY KEY)`,
 		`INSERT INTO foo (key) SELECT * FROM generate_series(1, 1000)`,
 		`ALTER TABLE foo SPLIT AT (SELECT * FROM generate_series(100, 900, 100))`,
@@ -241,9 +243,9 @@ func TestMuxRangeFeedConnectsToNodeOnce(t *testing.T) {
 	fooSpan := fooDesc.PrimaryIndexSpan(keys.SystemSQLCodec)
 
 	allSeen, onValue := observeNValues(1000)
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, onValue)
+	errCh, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, onValue)
 	defer closeFeed()
-	channelWaitWithTimeout(t, allSeen)
+	channelWaitWithTimeout(t, allSeen, errCh)
 	closeFeed() // Explicitly shutdown the feed to make sure counters no longer change.
 
 	// Verify we connected to each node once.
@@ -254,7 +256,7 @@ func TestMuxRangeFeedConnectsToNodeOnce(t *testing.T) {
 	}
 }
 
-func TestMuxRangeCatchupScanQuotaReleased(t *testing.T) {
+func TestMuxRangeFeedCatchupScanQuotaReleased(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -289,20 +291,23 @@ func TestMuxRangeCatchupScanQuotaReleased(t *testing.T) {
 	noValuesExpected := func(event kvcoord.RangeFeedMessage) {
 		panic("received value when none expected")
 	}
-	const numErrsToReturn = 100
+	const numErrsToReturn = 42
 	var numErrors atomic.Int32
 	enoughErrors := make(chan struct{})
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, noValuesExpected,
+	errCh, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, noValuesExpected,
 		kvcoord.TestingWithOnRangefeedEvent(
 			func(_ context.Context, _ roachpb.Span, _ int64, event *kvpb.RangeFeedEvent) (skip bool, _ error) {
+				if event.Error != nil {
+					return false, nil
+				}
 				*event = transientErrEvent
 				if numErrors.Add(1) == numErrsToReturn {
 					close(enoughErrors)
 				}
 				return false, nil
 			}))
 	defer closeFeed()
-	channelWaitWithTimeout(t, enoughErrors)
+	channelWaitWithTimeout(t, enoughErrors, errCh)
 }
 
 // TestMuxRangeFeedDoesNotStallOnError tests that the mux rangefeed
@@ -413,14 +418,14 @@ func TestMuxRangeFeedDoesNotStallOnError(t *testing.T) {
 
 	shouldError.Store(true)
 	allSeen, onValue := observeNValues(100)
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startFrom, onValue)
+	errCh, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startFrom, onValue)
 	defer closeFeed()
-	channelWaitWithTimeout(t, allSeen)
+	channelWaitWithTimeout(t, allSeen, errCh)
 }
 
 // Test to make sure the various metrics used by rangefeed are correctly
 // updated during the lifetime of the rangefeed and when the rangefeed completes.
-func TestRangeFeedMetricsManagement(t *testing.T) {
+func TestMuxRangeFeedMetricsManagement(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -434,6 +439,9 @@ func TestRangeFeedMetricsManagement(t *testing.T) {
 	sqlDB := sqlutils.MakeSQLRunner(tc.ServerConn(0))
 	startTime := ts.Clock().Now()
 
+	kvserver.RangefeedEnabled.Override(
+		context.Background(), &tc.SystemLayer(0).ClusterSettings().SV, true)
+
 	// Insert 1000 rows, and split them into 10 ranges.
 	const numRanges = 10
 	sqlDB.ExecMultiple(t,
@@ -502,8 +510,9 @@ func TestRangeFeedMetricsManagement(t *testing.T) {
 		return skipSet.stuck.Contains(k)
 	}
 
+	frontierAdvanced := make(chan struct{}, 1)
 	ignoreValues := func(event kvcoord.RangeFeedMessage) {}
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, ignoreValues,
+	errCh, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startTime, ignoreValues,
 		kvcoord.TestingWithRangeFeedMetrics(&metrics),
 		kvcoord.TestingWithOnRangefeedEvent(
 			func(ctx context.Context, s roachpb.Span, _ int64, event *kvpb.RangeFeedEvent) (skip bool, _ error) {
@@ -535,10 +544,16 @@ func TestRangeFeedMetricsManagement(t *testing.T) {
 							return false, nil
 						}
 
-						_, err := frontier.Forward(checkpoint.Span, checkpoint.ResolvedTS)
+						advanced, err := frontier.Forward(checkpoint.Span, checkpoint.ResolvedTS)
 						if err != nil {
 							return false, err
 						}
+						if advanced {
+							select {
+							case frontierAdvanced <- struct{}{}:
+							default:
+							}
+						}
 
 						if numCatchupBlocked.Add(1) <= numCatchupToBlock {
 							// Mux rangefeed can't block single range, so just skip this event
@@ -558,12 +573,7 @@ func TestRangeFeedMetricsManagement(t *testing.T) {
 
 	// Wait for the test frontier to advance.  Once it advances,
 	// we know the rangefeed is started, all ranges are running (even if some of them are blocked).
-	testutils.SucceedsSoon(t, func() error {
-		if frontier.Frontier().IsEmpty() {
-			return errors.Newf("waiting for frontier advance: %s", frontier.String())
-		}
-		return nil
-	})
+	channelWaitWithTimeout(t, frontierAdvanced, errCh)
 
 	// At this point, we know the rangefeed for all ranges are running.
 	require.EqualValues(t, numRanges, metrics.RangefeedRanges.Value(), frontier.String())
@@ -575,9 +585,9 @@ func TestRangeFeedMetricsManagement(t *testing.T) {
 	require.EqualValues(t, numCatchupToBlock, metrics.RangefeedCatchupRanges.Value())
 }
 
-// TestRangefeedRangeObserver ensures the kvcoord.WithRangeObserver option
+// TestMuxRangefeedRangeObserver ensures the kvcoord.WithRangeObserver option
 // works correctly.
-func TestRangefeedRangeObserver(t *testing.T) {
+func TestMuxRangefeedRangeObserver(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -640,7 +650,7 @@ func TestRangefeedRangeObserver(t *testing.T) {
 		})
 	}
 
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, ts.Clock().Now(), ignoreValues,
+	_, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, ts.Clock().Now(), ignoreValues,
 		kvcoord.WithRangeObserver(observer))
 	defer closeFeed()
 
@@ -735,7 +745,7 @@ func TestMuxRangeFeedCanCloseStream(t *testing.T) {
 	ignoreValues := func(event kvcoord.RangeFeedMessage) {}
 	var numRestartStreams atomic.Int32
 
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, ts.Clock().Now(), ignoreValues,
+	_, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, ts.Clock().Now(), ignoreValues,
 		kvcoord.TestingWithMuxRangeFeedRequestSenderCapture(
 			// We expect a single mux sender since we have 1 node in this test.
 			func(nodeID roachpb.NodeID, capture func(request *kvpb.RangeFeedRequest) error) {
@@ -863,13 +873,13 @@ func TestMuxRangeFeedDoesNotDeadlockWithLocalStreams(t *testing.T) {
 	fooSpan := fooDesc.PrimaryIndexSpan(keys.SystemSQLCodec)
 
 	allSeen, onValue := observeNValues(1000)
-	closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startFrom, onValue,
+	_, closeFeed := rangeFeed(ts.DistSenderI(), fooSpan, startFrom, onValue,
 		kvcoord.TestingWithBeforeSendRequest(func() {
 			// Prior to sending rangefeed request, block for just a bit
 			// to make deadlock more likely.
 			time.Sleep(100 * time.Millisecond)
 		}),
 	)
 	defer closeFeed()
-	channelWaitWithTimeout(t, allSeen)
+	channelWaitWithTimeout(t, allSeen, nil)
 }

pkg/kv/kvserver/rangefeed/catchup_scan.go

@@ -9,6 +9,7 @@ import (
 	"bytes"
 	"context"
 	"time"
+	"unsafe"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -238,6 +239,7 @@ func (i *CatchUpSnapshot) CatchUpScan(
 		unsafeKey := iter.UnsafeKey()
 		sameKey := bytes.Equal(unsafeKey.Key, lastKey)
 		if !sameKey {
+
 			// If so, output events for the last key encountered.
 			if err := outputEvents(); err != nil {
 				return err
@@ -253,6 +255,23 @@ func (i *CatchUpSnapshot) CatchUpScan(
 			recreateIter := false
 			// recreateIter => now is initialized.
 			var now crtime.Mono
+
+			// Under race tests we want to recreate the iterator more often to
+			// increase test coverage of this code. Doing this on every iteration
+			// proved a bit slow.
+			forceRecreateUnderTest := false
+			if util.RaceEnabled {
+				// Hash the pointer of the current iterator into the range [0, 3] to
+				// give us a ~25% chance of recreating the iterator.
+				//
+				// unsafe.Pointer's documentation implies that this uintptr is not
+				// technically guaranteed to be stable across multiple calls. So, we
+				// could make a different decision for the same iterator. That's OK, the
+				// goal here is to force recreation occasionally, nothing more.
+				ptr := uintptr(unsafe.Pointer(&iter))
+				forceRecreateUnderTest = util.FibHash(uint64(ptr), 2) == 0
+			}
+
 			// We sample the current time only when readmitted is true, to avoid
 			// doing it in every iteration of the loop. In practice, readmitted is
 			// true after every 100ms of cpu time, and there is no wall time limit
@@ -262,12 +281,13 @@ func (i *CatchUpSnapshot) CatchUpScan(
 			// Pace. We also need to consider the CPU cost of recreating the
 			// iterator, and using the cpu time to amortize that cost seems
 			// reasonable.
-			if (readmitted && i.iterRecreateDuration > 0) || util.RaceEnabled {
+			shouldSampleTime := (readmitted && i.iterRecreateDuration > 0) || forceRecreateUnderTest
+			if shouldSampleTime {
 				// If enough walltime has elapsed, close iter and reopen. This
 				// prevents the iter from holding on to Pebble memtables for too long,
 				// which can cause OOMs.
 				now = crtime.NowMono()
-				recreateIter = now.Sub(lastIterCreateTime) > i.iterRecreateDuration || util.RaceEnabled
+				recreateIter = now.Sub(lastIterCreateTime) > i.iterRecreateDuration || forceRecreateUnderTest
 			}
 			if recreateIter {
 				lastIterCreateTime = now

pkg/util/BUILD.bazel

@@ -47,6 +47,7 @@ go_test(
     srcs = [
         "every_n_test.go",
         "fast_int_map_test.go",
+        "hash_test.go",
         "slices_test.go",
         "smalltrace_test.go",
         "strings_test.go",

pkg/util/hash.go

@@ -56,3 +56,12 @@ func (f *FNV64) Add(c uint64) {
 func (f *FNV64) Sum() uint64 {
 	return f.sum
 }
+
+// Fibonacci Hash is a multiplicative hash that maps the given `v` into 2^bits
+// values.
+//
+// Reference:
+//   - https://probablydance.com/2018/06/16/fibonacci-hashing-the-optimization-that-the-world-forgot-or-a-better-alternative-to-integer-modulo/
+func FibHash(v uint64, bits int) uint64 {
+	return (v * 11400714819323198485) >> (64 - bits)
+}

pkg/util/hash_test.go

@@ -0,0 +1,40 @@
+// Copyright 2016 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package util
+
+import (
+	"testing"
+	"unsafe"
+
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFibHash(t *testing.T) {
+	rng, _ := randutil.NewTestRand()
+	counts := make(map[uint64]int)
+	for range 1000 {
+		h := FibHash(rng.Uint64(), 3)
+		counts[h]++
+	}
+
+	// Let's test using a ptr value. We use this slice to keep the pointers alive
+	// so that we don't just get the same pointer each time.
+	vs := make([]*int, 0, 1000)
+	for range 1000 {
+		v := new(int)
+		vs = append(vs, v)
+		h := FibHash(uint64(uintptr(unsafe.Pointer(v))), 3)
+		counts[h]++
+	}
+	require.Equal(t, 1000, len(vs))
+	// We can only make some weak assertions about the distribution here.
+	for k, v := range counts {
+		require.True(t, k < 8)
+		perfectDistributionCount := 2000 / 8
+		require.Less(t, v, perfectDistributionCount*2)
+	}
+}