rangefeed: move back to static (but configurable) send queue limit

The interactions here probably requires more thought before doing
anything more complicated than a simple static limit.

Epic: none
Release note: None

Author: stevendanna

pkg/kv/kvserver/kvserverbase/base.go

@@ -383,3 +383,16 @@ var MaxCommandSize = settings.RegisterByteSizeSetting(
 	MaxCommandSizeDefault,
 	settings.ByteSizeWithMinimum(MaxCommandSizeFloor),
 )
+
+// DefaultRangefeedEventCap is the channel capacity of the rangefeed processor
+// and each registration. It is also used to calculate the default capacity
+// limit for the buffered sender.
+//
+// The size of an event is 72 bytes, so this will result in an allocation on the
+// order of ~300KB per RangeFeed. That's probably ok given the number of ranges
+// on a node that we'd like to support with active rangefeeds, but it's
+// certainly on the upper end of the range.
+//
+// Note that processors also must reserve memory from one of two memory monitors
+// for each event.
+const DefaultRangefeedEventCap = 4096

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/kv/kvserver/concurrency/lock",
+        "//pkg/kv/kvserver/kvserverbase",
         "//pkg/roachpb",
         "//pkg/settings",
         "//pkg/settings/cluster",

pkg/kv/kvserver/rangefeed/buffered_sender.go

@@ -10,6 +10,9 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
+	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -45,6 +48,38 @@ import (
 //               BufferedPerRangeEventSink.Send    BufferedPerRangeEventSink.SendError
 //
 
+// RangefeedSingleBufferedSenderQueueMaxSize is the maximum number of events
+// that the buffered sender will buffer before it starts returning capacity
+// exceeded errors. Updates to this setting are only applied to need
+// MuxRangefeedCalls, existing streams will use the previous value until
+// restarted.
+//
+// # The main goal of this limit is to provide a backstop against the
+//
+// The default here has been somewhat arbitrarily chosen considering that:
+//
+//   - We want to avoid capacity exceeded errors that wouldn't have occurred
+//     when the buffered registrations were in use.
+//
+//   - We don't want to drastically increase the amount of queueing allowed for a
+//     single registration.
+//
+//   - One buffered sender is feeding a single gRPC client.
+//
+//   - Events emitted during catchup scans have their own per-registration buffer
+//     still.
+//
+// TODO(ssd): This is a bit of a stop-gap so that we have a knob to turn if we
+// need to. We probably want each buffered sender (or each consumerID) to be
+// able to hold up to some fraction of the total rangefeed budget. But we are
+// starting here for now.
+var RangefeedSingleBufferedSenderQueueMaxSize = settings.RegisterIntSetting(
+	settings.SystemOnly,
+	"kv.rangefeed.buffered_sender.queue_max_size",
+	"max size of a buffered senders event queue (0 for no max)",
+	kvserverbase.DefaultRangefeedEventCap*8,
+)
+
 // BufferedSender is embedded in every rangefeed.BufferedPerRangeEventSink,
 // serving as a helper which buffers events before forwarding events to the
 // underlying gRPC stream.
@@ -58,21 +93,7 @@ type BufferedSender struct {
 		syncutil.Mutex
 		stopped bool
 		buffer  *eventQueue
-		// capacity is the maximum number of events that can be buffered. It
-		// dynamically scales based on the number of active registrations.
-		//
-		// The capacity is calculated as:
-		// - Minimum: minBufferedSenderQueueCapacity (20 registrations)
-		// - Active streams: perUnbufferedRegCapacity * number of active
-		// registrations
-		//
-		// This scaling is based on the intuition that each stream should have
-		// equivalent buffer space to what unbuffered registrations receive (4096
-		// events per registration). Since all registrations share the same buffered
-		// sender, registrations may affect each other.
-		//
-		// Note that when capacity shrinks, events already buffered will not be
-		// dropped. Capacity is not adjusted during shutdown.
+		// capacity is the maximum number of events that can be buffered.
 		capacity   int64
 		overflowed bool
 	}
@@ -88,14 +109,8 @@ type BufferedSender struct {
 	metrics *BufferedSenderMetrics
 }
 
-// TODO(wenyihu6): This value is set to the same value as
-// https://github.com/cockroachdb/cockroach/blob/5536f0828f50bb21ec1577c77d388c4303d124a4/pkg/kv/kvserver/replica_rangefeed.go#L121.
-// Should I move this const to kvserverbase and import this value?
-const perUnbufferedRegCapacity = int64(4096)
-const minBufferedSenderQueueCapacity = int64(4096 * 20)
-
 func NewBufferedSender(
-	sender ServerStreamSender, bsMetrics *BufferedSenderMetrics,
+	sender ServerStreamSender, settings *cluster.Settings, bsMetrics *BufferedSenderMetrics,
 ) *BufferedSender {
 	bs := &BufferedSender{
 		sender:  sender,
@@ -104,7 +119,7 @@ func NewBufferedSender(
 	bs.queueMu.buffer = newEventQueue()
 	bs.notifyDataC = make(chan struct{}, 1)
 	bs.queueMu.buffer = newEventQueue()
-	bs.queueMu.capacity = minBufferedSenderQueueCapacity
+	bs.queueMu.capacity = RangefeedSingleBufferedSenderQueueMaxSize.Get(&settings.SV)
 	return bs
 }
 
@@ -123,7 +138,7 @@ func (bs *BufferedSender) sendBuffered(
 	if bs.queueMu.overflowed {
 		return newRetryErrBufferCapacityExceeded()
 	}
-	if bs.queueMu.buffer.len() >= bs.queueMu.capacity {
+	if bs.queueMu.capacity > 0 && bs.queueMu.buffer.len() >= bs.queueMu.capacity {
 		bs.queueMu.overflowed = true
 		return newRetryErrBufferCapacityExceeded()
 	}
@@ -208,21 +223,16 @@ func (bs *BufferedSender) cleanup(ctx context.Context) {
 	bs.metrics.BufferedSenderQueueSize.Dec(remaining)
 }
 
-// onStreamConnectOrDisconnect is called when a stream is added or removed. This
-// is currently used to dynamically adjust its queue capacity based on number of
-// active registrations. Note that this is not called during shutdown, so strict
-// dependency on this contract is discouraged. And note that we do not drop
-// events if the capacity is shrunk.
-func (bs *BufferedSender) onStreamConnectOrDisconnect(activeStreamCount int64) {
+func (bs *BufferedSender) len() int {
 	bs.queueMu.Lock()
 	defer bs.queueMu.Unlock()
-	bs.queueMu.capacity = max(minBufferedSenderQueueCapacity, activeStreamCount*perUnbufferedRegCapacity)
+	return int(bs.queueMu.buffer.len())
 }
 
-func (bs *BufferedSender) len() int {
+func (bs *BufferedSender) overflowed() bool {
 	bs.queueMu.Lock()
 	defer bs.queueMu.Unlock()
-	return int(bs.queueMu.buffer.len())
+	return bs.queueMu.overflowed
 }
 
 // Used for testing only.

pkg/kv/kvserver/rangefeed/buffered_sender_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -35,7 +36,8 @@ func TestBufferedSenderDisconnectStream(t *testing.T) {
 	defer stopper.Stop(ctx)
 	testServerStream := newTestServerStream()
 	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
+	st := cluster.MakeTestingClusterSettings()
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
 	sm := NewStreamManager(bs, smMetrics)
 	require.NoError(t, sm.Start(ctx, stopper))
 	defer sm.Stop(ctx)
@@ -88,7 +90,8 @@ func TestBufferedSenderChaosWithStop(t *testing.T) {
 	testServerStream := newTestServerStream()
 
 	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
+	st := cluster.MakeTestingClusterSettings()
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
 	sm := NewStreamManager(bs, smMetrics)
 	require.NoError(t, sm.Start(ctx, stopper))
 
@@ -176,29 +179,33 @@ func TestBufferedSenderOnOverflow(t *testing.T) {
 	stopper := stop.NewStopper()
 	defer stopper.Stop(ctx)
 	testServerStream := newTestServerStream()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
-	require.Equal(t, minBufferedSenderQueueCapacity, bs.queueMu.capacity)
+	st := cluster.MakeTestingClusterSettings()
+
+	queueCap := int64(24)
+	RangefeedSingleBufferedSenderQueueMaxSize.Override(ctx, &st.SV, queueCap)
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
+	require.Equal(t, queueCap, bs.queueMu.capacity)
 
 	val1 := roachpb.Value{RawBytes: []byte("val"), Timestamp: hlc.Timestamp{WallTime: 1}}
 	ev1 := new(kvpb.RangeFeedEvent)
 	ev1.MustSetValue(&kvpb.RangeFeedValue{Key: keyA, Value: val1})
 	muxEv := &kvpb.MuxRangeFeedEvent{RangeFeedEvent: *ev1, RangeID: 0, StreamID: 1}
 
-	for i := int64(0); i < minBufferedSenderQueueCapacity; i++ {
+	for range queueCap {
 		require.NoError(t, bs.sendBuffered(muxEv, nil))
 	}
-	require.Equal(t, minBufferedSenderQueueCapacity, int64(bs.len()))
+	require.Equal(t, queueCap, int64(bs.len()))
 	e, success, overflowed, remains := bs.popFront()
 	require.Equal(t, sharedMuxEvent{
 		ev:    muxEv,
 		alloc: nil,
 	}, e)
 	require.True(t, success)
 	require.False(t, overflowed)
-	require.Equal(t, minBufferedSenderQueueCapacity-1, remains)
-	require.Equal(t, minBufferedSenderQueueCapacity-1, int64(bs.len()))
+	require.Equal(t, queueCap-1, remains)
+	require.Equal(t, queueCap-1, int64(bs.len()))
 	require.NoError(t, bs.sendBuffered(muxEv, nil))
-	require.Equal(t, minBufferedSenderQueueCapacity, int64(bs.len()))
+	require.Equal(t, queueCap, int64(bs.len()))
 
 	// Overflow now.
 	require.Equal(t, bs.sendBuffered(muxEv, nil).Error(),
@@ -207,7 +214,6 @@ func TestBufferedSenderOnOverflow(t *testing.T) {
 
 // TestBufferedSenderOnStreamShutdown tests that BufferedSender and
 // StreamManager handle overflow and shutdown properly.
-
 func TestBufferedSenderOnStreamShutdown(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -217,101 +223,70 @@ func TestBufferedSenderOnStreamShutdown(t *testing.T) {
 	defer stopper.Stop(ctx)
 	testServerStream := newTestServerStream()
 	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
-	require.Equal(t, minBufferedSenderQueueCapacity, bs.queueMu.capacity)
+	st := cluster.MakeTestingClusterSettings()
+
+	queueCap := int64(24)
+	RangefeedSingleBufferedSenderQueueMaxSize.Override(ctx, &st.SV, queueCap)
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
+	require.Equal(t, queueCap, bs.queueMu.capacity)
+
 	sm := NewStreamManager(bs, smMetrics)
 	require.NoError(t, sm.Start(ctx, stopper))
 	defer sm.Stop(ctx)
 
 	p, h, pStopper := newTestProcessor(t, withRangefeedTestType(scheduledProcessorWithBufferedSender))
 	defer pStopper.Stop(ctx)
 
+	streamID := int64(42)
+
 	val1 := roachpb.Value{RawBytes: []byte("val"), Timestamp: hlc.Timestamp{WallTime: 1}}
 	ev1 := new(kvpb.RangeFeedEvent)
 	ev1.MustSetValue(&kvpb.RangeFeedValue{Key: keyA, Value: val1})
-	muxEv := &kvpb.MuxRangeFeedEvent{RangeFeedEvent: *ev1, RangeID: 0, StreamID: 1}
+	muxEv := &kvpb.MuxRangeFeedEvent{RangeFeedEvent: *ev1, RangeID: 0, StreamID: streamID}
 
-	// Add 21 streams and overflow the buffer.
-	t.Run("add 21 streams", func(t *testing.T) {
-		numStreams := int64(21)
-		expectedCapacity := perUnbufferedRegCapacity * numStreams
-		// Block the stream to help the queue to overflow.
-		unblock := testServerStream.BlockSend()
-		for id := int64(0); id < numStreams; id++ {
-			registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpIter */
-				false /* withDiff */, false /* withFiltering */, false /* withOmitRemote */, noBulkDelivery,
-				sm.NewStream(id, 1 /*rangeID*/))
-			require.True(t, registered)
-			sm.AddStream(id, d)
-		}
-		require.Equal(t, expectedCapacity, bs.queueMu.capacity)
-		for int64(bs.len()) != expectedCapacity {
-			require.NoError(t, sm.sender.sendBuffered(muxEv, nil))
-		}
-		require.Equal(t, expectedCapacity, int64(bs.len()))
-		require.Equal(t, bs.sendBuffered(muxEv, nil).Error(),
-			newRetryErrBufferCapacityExceeded().Error())
-		require.True(t, bs.queueMu.overflowed)
-		unblock()
-	})
+	// Block the stream so that we can overflow later.
+	unblock := testServerStream.BlockSend()
+	defer unblock()
 
-	t.Run("overflow clean up", func(t *testing.T) {
-		// All events buffered should still be sent to the stream.
+	waitForQueueLen := func(len int) {
 		testutils.SucceedsSoon(t, func() error {
-			if bs.len() == 0 {
+			if bs.len() == len {
 				return nil
 			}
-			return errors.Newf("expected 0 registrations, found %d", bs.len())
+			return errors.Newf("expected %d events, found %d", len, bs.len())
 		})
-		// Overflow cleanup.
-		err := <-sm.Error()
-		require.Equal(t, newRetryErrBufferCapacityExceeded().Error(), err.Error())
-		// Note that we expect the stream manager to shut down here, but no actual
-		// error events would be sent during the shutdown.
-		require.Equal(t, bs.sendBuffered(muxEv, nil).Error(), newRetryErrBufferCapacityExceeded().Error())
-	})
-}
-
-// TestBufferedSenderOnStreamDisconnect tests that BufferedSender dynamically
-// adjusts its capacity when streams are connected or disconnected.
-func TestBufferedSenderOnStreamDisconnect(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-	ctx := context.Background()
-
-	stopper := stop.NewStopper()
-	defer stopper.Stop(ctx)
-	testServerStream := newTestServerStream()
-	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
-	require.Equal(t, minBufferedSenderQueueCapacity, bs.queueMu.capacity)
-
-	sm := NewStreamManager(bs, smMetrics)
-	require.NoError(t, sm.Start(ctx, stopper))
-
-	p, h, pStopper := newTestProcessor(t, withRangefeedTestType(scheduledProcessorWithBufferedSender))
-	defer pStopper.Stop(ctx)
-	defer sm.Stop(ctx)
+	}
 
-	numStreams := int64(21)
-	expectedCapacity := perUnbufferedRegCapacity * numStreams
-	for id := int64(0); id < numStreams; id++ {
-		registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpIter */
-			false /* withDiff */, false /* withFiltering */, false /* withOmitRemote */, noBulkDelivery,
-			sm.NewStream(id, 1 /*rangeID*/))
-		require.True(t, registered)
-		sm.AddStream(id, d)
+	// Add our stream to the stream manager.
+	registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpIter */
+		false /* withDiff */, false /* withFiltering */, false /* withOmitRemote */, noBulkDelivery,
+		sm.NewStream(streamID, 1 /*rangeID*/))
+	require.True(t, registered)
+	sm.AddStream(streamID, d)
+
+	require.NoError(t, sm.sender.sendBuffered(muxEv, nil))
+	// At this point we actually have sent 2 events. 1 checkpoint event sent by
+	// register and 1 event sent on the line above. We wait for 1 of these events
+	// to be pulled off the queue and block in the sender, leaving 1 in the queue.
+	waitForQueueLen(1)
+	// Now fill the rest of the queue.
+	for range queueCap - 1 {
+		require.NoError(t, sm.sender.sendBuffered(muxEv, nil))
 	}
-	require.Equal(t, expectedCapacity, bs.queueMu.capacity)
-	sm.DisconnectStream(int64(0), newErrBufferCapacityExceeded())
-	testServerStream.waitForEvent(t, makeMuxRangefeedErrorEvent(0, 1, newErrBufferCapacityExceeded()))
-	testutils.SucceedsSoon(t, func() error {
-		bs.queueMu.Lock()
-		defer bs.queueMu.Unlock()
-		if bs.queueMu.capacity == expectedCapacity-perUnbufferedRegCapacity {
-			return nil
-		}
-		return errors.Newf("expected %d cap to be %d", bs.queueMu.capacity,
-			expectedCapacity-perUnbufferedRegCapacity)
-	})
+
+	// The next write should overflow.
+	capExceededErrStr := newRetryErrBufferCapacityExceeded().Error()
+	err := sm.sender.sendBuffered(muxEv, nil)
+	require.EqualError(t, err, capExceededErrStr)
+	require.True(t, bs.overflowed())
+
+	unblock()
+	waitForQueueLen(0)
+	// Overflow cleanup.
+	err = <-sm.Error()
+	require.EqualError(t, err, capExceededErrStr)
+	// Note that we expect the stream manager to shut down here, but no actual
+	// error events would be sent during the shutdown.
+	err = sm.sender.sendBuffered(muxEv, nil)
+	require.EqualError(t, err, capExceededErrStr)
 }