rangefeed: backpressure senders on EventChanTimeout=0

This patch makes rangefeeds backpressure senders instead of
disconnecting clients when `EventChanTimeout` is set to 0.

This will be useful for benchmarks, where backpressuring the workload
will more accurately measure the overhead of rangefeeds instead of
randomly introducing disconnects and catchup scans.

Epic: none
Release note: None

Author: erikgrinaker

pkg/kv/kvserver/rangefeed/processor.go

@@ -64,9 +64,9 @@ type Config struct {
 	// EventChanCap specifies the capacity to give to the Processor's input
 	// channel.
 	EventChanCap int
-	// EventChanTimeout specifies the maximum duration that methods will
-	// wait to send on the Processor's input channel before giving up and
-	// shutting down the Processor. 0 for no timeout.
+	// EventChanTimeout specifies the maximum time to wait when sending on the
+	// Processor's input channel before giving up and shutting down the Processor.
+	// 0 disables the timeout, backpressuring writers up through Raft (for tests).
 	EventChanTimeout time.Duration
 
 	// Metrics is for production monitoring of RangeFeeds.
@@ -476,10 +476,10 @@ func (p *Processor) Register(
 	// it should see these events during its catch up scan.
 	p.syncEventC()
 
+	blockWhenFull := p.Config.EventChanTimeout == 0 // for testing
 	r := newRegistration(
 		span.AsRawSpanWithNoLocals(), startTS, catchUpIterConstructor, withDiff,
-		p.Config.EventChanCap, p.Metrics, stream, disconnectFn, done,
-	)
+		p.Config.EventChanCap, blockWhenFull, p.Metrics, stream, disconnectFn, done)
 	select {
 	case p.regC <- r:
 		// Wait for response.

pkg/kv/kvserver/rangefeed/processor_test.go

@@ -142,6 +142,7 @@ func rangeFeedCheckpoint(span roachpb.Span, ts hlc.Timestamp) *kvpb.RangeFeedEve
 }
 
 const testProcessorEventCCap = 16
+const testProcessorEventCTimeout = 10 * time.Millisecond
 
 func newTestProcessorWithTxnPusher(
 	t *testing.T, rtsIter storage.SimpleMVCCIterator, txnPusher TxnPusher,
@@ -161,6 +162,7 @@ func newTestProcessorWithTxnPusher(
 		TxnPusher:        txnPusher,
 		PushTxnsInterval: pushTxnInterval,
 		PushTxnsAge:      pushTxnAge,
+		EventChanTimeout: testProcessorEventCTimeout,
 		EventChanCap:     testProcessorEventCCap,
 		Metrics:          NewMetrics(),
 	})
@@ -1128,6 +1130,7 @@ func TestBudgetReleaseOnProcessorStop(t *testing.T) {
 		PushTxnsInterval: pushTxnInterval,
 		PushTxnsAge:      pushTxnAge,
 		EventChanCap:     channelCapacity,
+		EventChanTimeout: time.Millisecond,
 		MemBudget:        fb,
 		Metrics:          NewMetrics(),
 	})
@@ -1218,6 +1221,7 @@ func TestBudgetReleaseOnLastStreamError(t *testing.T) {
 		PushTxnsInterval: pushTxnInterval,
 		PushTxnsAge:      pushTxnAge,
 		EventChanCap:     channelCapacity,
+		EventChanTimeout: time.Millisecond,
 		MemBudget:        fb,
 		Metrics:          NewMetrics(),
 	})
@@ -1297,6 +1301,7 @@ func TestBudgetReleaseOnOneStreamError(t *testing.T) {
 		PushTxnsInterval: pushTxnInterval,
 		PushTxnsAge:      pushTxnAge,
 		EventChanCap:     channelCapacity,
+		EventChanTimeout: time.Millisecond,
 		MemBudget:        fb,
 		Metrics:          NewMetrics(),
 	})
@@ -1458,3 +1463,81 @@ func TestSizeOfEvent(t *testing.T) {
 	size := int(unsafe.Sizeof(e))
 	require.Equal(t, 72, size)
 }
+
+// TestProcessorBackpressure tests that a processor with EventChanTimeout set to
+// 0 will backpressure senders when a consumer isn't keeping up.
+func TestProcessorBackpressure(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	span := roachpb.RSpan{Key: roachpb.RKey("a"), EndKey: roachpb.RKey("z")}
+
+	// Set up processor.
+	p := NewProcessor(Config{
+		AmbientContext:   log.MakeTestingAmbientContext(nil),
+		Clock:            hlc.NewClockForTesting(nil),
+		Metrics:          NewMetrics(),
+		Span:             span,
+		MemBudget:        newTestBudget(math.MaxInt64),
+		EventChanCap:     1,
+		EventChanTimeout: 0,
+	})
+	require.NoError(t, p.Start(stopper, nil))
+	defer p.Stop()
+
+	// Add a registration.
+	stream := newTestStream()
+	done := &future.ErrorFuture{}
+	ok, _ := p.Register(span, hlc.MinTimestamp, nil, false, stream, nil, done)
+	require.True(t, ok)
+
+	// Wait for the initial checkpoint.
+	p.syncEventAndRegistrations()
+	require.Len(t, stream.Events(), 1)
+
+	// Block the registration consumer, and spawn a goroutine to post events to
+	// the stream, which should block. The rangefeed pipeline buffers a few
+	// additional events in intermediate goroutines between channels, so post 10
+	// events to be sure.
+	unblock := stream.BlockSend()
+	defer unblock()
+
+	const numEvents = 10
+	doneC := make(chan struct{})
+	go func() {
+		for i := 0; i < numEvents; i++ {
+			assert.True(t, p.ForwardClosedTS(ctx, hlc.Timestamp{WallTime: int64(i + 1)}))
+		}
+		close(doneC)
+	}()
+
+	// The sender should be blocked for at least 3 seconds.
+	select {
+	case <-doneC:
+		t.Fatal("send unexpectely succeeded")
+	case <-time.After(3 * time.Second):
+	case <-ctx.Done():
+	}
+
+	// Unblock the sender, and wait for it to complete.
+	unblock()
+	select {
+	case <-doneC:
+	case <-time.After(time.Second):
+		t.Fatal("sender did not complete")
+	}
+
+	// Wait for the final checkpoint event.
+	p.syncEventAndRegistrations()
+	events := stream.Events()
+	require.Equal(t, &kvpb.RangeFeedEvent{
+		Checkpoint: &kvpb.RangeFeedCheckpoint{
+			Span:       span.AsRawSpanWithNoLocals(),
+			ResolvedTS: hlc.Timestamp{WallTime: numEvents},
+		},
+	}, events[len(events)-1])
+}

pkg/kv/kvserver/rangefeed/registry.go

@@ -95,9 +95,10 @@ type registration struct {
 	done   *future.ErrorFuture
 	unreg  func()
 	// Internal.
-	id   int64
-	keys interval.Range
-	buf  chan *sharedEvent
+	id            int64
+	keys          interval.Range
+	buf           chan *sharedEvent
+	blockWhenFull bool // if true, block when buf is full (for tests)
 
 	mu struct {
 		sync.Locker
@@ -125,6 +126,7 @@ func newRegistration(
 	catchUpIterConstructor CatchUpIteratorConstructor,
 	withDiff bool,
 	bufferSz int,
+	blockWhenFull bool,
 	metrics *Metrics,
 	stream Stream,
 	unregisterFn func(),
@@ -140,6 +142,7 @@ func newRegistration(
 		done:                   done,
 		unreg:                  unregisterFn,
 		buf:                    make(chan *sharedEvent, bufferSz),
+		blockWhenFull:          blockWhenFull,
 	}
 	r.mu.Locker = &syncutil.Mutex{}
 	r.mu.caughtUp = true
@@ -167,6 +170,22 @@ func (r *registration) publish(
 	case r.buf <- e:
 		r.mu.caughtUp = false
 	default:
+		// If we're asked to block (in tests), do a blocking send after releasing
+		// the mutex -- otherwise, the output loop won't be able to consume from the
+		// channel. We optimistically attempt the non-blocking send above first,
+		// since we're already holding the mutex.
+		if r.blockWhenFull {
+			r.mu.Unlock()
+			select {
+			case r.buf <- e:
+				r.mu.Lock()
+				r.mu.caughtUp = false
+			case <-ctx.Done():
+				r.mu.Lock()
+				alloc.Release(ctx)
+			}
+			return
+		}
 		// Buffer exceeded and we are dropping this event. Registration will need
 		// a catch-up scan.
 		r.mu.overflowed = true

pkg/kv/kvserver/rangefeed/registry_test.go

@@ -13,6 +13,7 @@ package rangefeed
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 
 	_ "github.com/cockroachdb/cockroach/pkg/keys" // hook up pretty printer
@@ -88,7 +89,10 @@ func (s *testStream) Events() []*kvpb.RangeFeedEvent {
 
 func (s *testStream) BlockSend() func() {
 	s.mu.Lock()
-	return s.mu.Unlock
+	var once sync.Once
+	return func() {
+		once.Do(s.mu.Unlock) // safe to call multiple times, e.g. defer and explicit
+	}
 }
 
 type testRegistration struct {
@@ -119,6 +123,7 @@ func newTestRegistration(
 		makeCatchUpIteratorConstructor(catchup),
 		withDiff,
 		5,
+		false, /* blockWhenFull */
 		NewMetrics(),
 		s,
 		func() {},