kvserver/rangefeed: remove per-registration goroutines

[wenyihu6]

The goal of this PR is to remove the per registration goroutines that move
data from per registration buffers to the client and replace it with a per mux rangefeed goroutine
managing a per mux rangefeed buffer.

Since registrations are O(ranges) and since for large scale users, rangefeeds typically cover many
ranges on a node, this should reduce the number of goroutines needed to run a rangefeed in caught up mode.

In catch-up mode, the per-registration goroutine and buffered is still used.

This per-mux-rangefeed goroutine and buffer is managed by the new BufferedSender component.

Epic: None

Known follow-ups:

• Perf test
• Add buffered capacity as a follow up - add more tests related to memory accounting draining
• Add test to cdc bench as a follow up
• Enable the setting metamorphically as a follow up
• Make catch up scan switch over test better

[blathers-crl]

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[cockroach-teamcity]

This change is 

[blathers-crl]

Your pull request contains more than 1000 changes. It is strongly encouraged to split big PRs into smaller chunks.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[tbg]

I gave this a first pass. Even though I understand the high-level picture, there is too much here, especially in the first commit, to really review in depth; entire files have moved and Reviewable is already extremely confused and this would only get worse. I would like to suggest breaking the first commit up into a sequence of PRs that we can then iterate and improve on, which should now be possible since we know how it should all shake out.
While doing so, I would try to look at ways to avoid the "obfuscation" that arises from sharing the sender interface, and avoid code movement except in commits that don't make any semantic changes. My (perhaps naive) suggestion would be:

1. extract the StreamManager. (There is no buffered stream yet, so it's used only for the unbuffered stream). (Extract it in place, then move it to new file). We merge that.
2. Do the disconnector refactor. (If this can be split up further, great).
3. now unless I've missed something, all the cleanup has happened and we can introduce the unbuffered sender (i.e. we pick up the sequence of commits you have here). First commit we devise a way to avoid sharing the sender interface. Then introduce the interface for the buffered sender. Make sure StreamManager can route it. (buffered sender still unimplemented here, just have a stub). Then replace the stub with the real thing, add tests, etc.
4. add unbuffered registration.

WDYT?

Reviewed 7 of 24 files at r14, 20 of 20 files at r18, 7 of 7 files at r19, 8 of 8 files at r20, 12 of 12 files at r21, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @stevendanna and @wenyihu6)

---

pkg/kv/kvserver/rangefeed/buffered_sender.go line 72 at r20 (raw file):

// sendUnbuffered sends the event directly to the underlying ServerStreamSender.
// It bypasses the buffer.

But this doesn't seem to be called (PerRangeEventSink.SendUnbuffered calls it, but this sink wouldn't wrap a buffered sender I think).

So we could drop this method from here and the sender interface and instead make an interface-widening assertion only in the case of the unbuffered sender.

---

pkg/kv/kvserver/rangefeed/buffered_sender.go line 96 at r20 (raw file):

			for {
				e, success := bs.popFront()
				if success {

nit:

if !success {
  break
}
// do the happy path

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 21 at r18 (raw file):

// +-----------------+               +-----------------+                  +-----------------+
// |                 | Disconnect    |                 |  r.disconnect    |                 |
// |  MuxRangeFeed   +-------------->|   SteamManager  +----------------->|   Registration  |

StreamManager

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 49 at r18 (raw file):

	errCh chan error

	// Note that lockedMuxStream wraps the underlying grpc server stream, ensuring

Patch up the comment to explain that sender is thread-safe (which I think is what it is now saying). I think it's helpful to also mention the possible implementation (as the comment does now).

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 108 at r18 (raw file):

// OnError is a callback that is called when a sender sends a rangefeed
// completion error back to the client. Note that we check for the existence of
// streamID to avoid metrics inaccuracy when the error is sent before the stream

What happens in this case when the stream is added? Why isn't a stream always added before it can generate errors?

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 123 at r18 (raw file):

	if err == nil {
		log.Fatalf(context.Background(),
			"unexpected: DisconnectStream called with non-nil error")

nil, not non-nil.

---

pkg/kv/kvserver/rangefeed/unbuffered_sender.go line 129 at r18 (raw file):

// needs to make sure this is called only with non-nil error events. Important
// to be thread-safe.
func (ubs *UnbufferedSender) sendBuffered(

it's surprising to me that this method is used only in one specific case and the sibling on BufferedSender is the main workhorse. I think this is because one interface is used for two different systems: an UnbufferedSender is used by a PerRangeEventSink and a BufferedSender is used by a BufferedPerRangeEventSink.

I find that keeping legacy impls around tends to greatly obfuscate the code. I am looking at this code:

	case *BufferedSender:
		return &BufferedPerRangeEventSink{
			PerRangeEventSink: NewPerRangeEventSink(rangeID, streamID, sender),
		}
	case *UnbufferedSender:
		return NewPerRangeEventSink(rangeID, streamID, sender)
	default:

and I wonder why we need to use the same sender interface in both. Can't they take different interfaces? Or can we at least make the interface "canonical" for the non-legacy (buffered) case and then just type assert in the legacy case and use the underlying type (a non-buffering sender in the unbuffered case)?

This suggestion is not for this PR. But anything that helps make this code more accessible will help for the future.

[stevendanna]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @tbg and @wenyihu6)

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 108 at r18 (raw file):

Previously, tbg (Tobias Grieger) wrote…

What happens in this case when the stream is added? Why isn't a stream always added before it can generate errors?

Just as a heads up: This has been one of the main points of back and forth in some past review.

A registration can send us error once it is registered with a processor because of an error at the processor level or from the catch-up scan. This can happen before the Disconnector has been sent back up the stack to us.

Previous iterations solved this by having the registration process register the callback with us by sending something down the stack and then setting the callback at the point of registration construction. No one was particularly happy with the complexity this appeared to add.

In this iteration, we've opted to send this Disconnector back up the stack and then solve the race by checking whether the stream is Disconnected before adding it to the map. Because we only increment stats when we added the stream to the map, we only want to decrement them on streams actually deleted from the map.

[wenyihu6]

TYFTR! I started working with Steven to break down the PRs into smaller chunks. I've already started a series here (#134953, #134954, #134955, #134956, #134957, #134958, #134959). The last one is still a bit large, but I'll see if I can split it further - it will require a bit more surgery. Just wanted to get the ball rolling on the first few ones. Let me know if this is along the lines you were thinking.

[wenyihu6]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @stevendanna and @tbg)

---

pkg/kv/kvserver/rangefeed/buffered_sender.go line 72 at r20 (raw file):

Previously, tbg (Tobias Grieger) wrote…

But this doesn't seem to be called (PerRangeEventSink.SendUnbuffered calls it, but this sink wouldn't wrap a buffered sender I think).

So we could drop this method from here and the sender interface and instead make an interface-widening assertion only in the case of the unbuffered sender.

We can discuss the contracts more in #134957 and see if there is a cleaner way to do things. It is a bit confusing since we allow PerRangeEventSink to wrap a BufferedSender

cockroach/pkg/kv/kvserver/rangefeed/stream_manager.go

Line 99 in 6478174

PerRangeEventSink: NewPerRangeEventSink(rangeID, streamID, sender),

. More details in #134957 (comment). iirc, the only reason we need both PerRangeEventSink and BufferedPerRangeEventSink is because we want to have both BufferedStream and Stream interfaces to tell p.Register which registration type to initialize.

---

pkg/kv/kvserver/rangefeed/buffered_sender.go line 96 at r20 (raw file):

Previously, tbg (Tobias Grieger) wrote…

nit:

if !success {
  break
}
// do the happy path

I will try to remember this when we are adding the BufferedSender PR.

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 21 at r18 (raw file):

Previously, tbg (Tobias Grieger) wrote…

StreamManager

Done in

cockroach/pkg/kv/kvserver/rangefeed/stream_manager.go

Line 21 in 39afe4e

// | MuxRangeFeed +-------------->| StreamManager +----------------->| Registration |

.

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 49 at r18 (raw file):

Previously, tbg (Tobias Grieger) wrote…

Patch up the comment to explain that sender is thread-safe (which I think is what it is now saying). I think it's helpful to also mention the possible implementation (as the comment does now).

Done in

cockroach/pkg/kv/kvserver/rangefeed/stream_manager.go

Lines 49 to 50 in 39afe4e

	// Implemented by UnbufferedSender and BufferedSender. Implementations should
	// ensure that sendUnbuffered and sendBuffered are thread-safe.

. I added a contract method to the sender interface to make this clearer.

---

pkg/kv/kvserver/rangefeed/stream_manager.go line 123 at r18 (raw file):

Previously, tbg (Tobias Grieger) wrote…

nil, not non-nil.

Oops, done in

cockroach/pkg/kv/kvserver/rangefeed/stream_manager.go

Line 117 in 39afe4e

"unexpected: DisconnectStream called with nil error")

.

---

pkg/kv/kvserver/rangefeed/unbuffered_sender.go line 129 at r18 (raw file):

Previously, tbg (Tobias Grieger) wrote…

it's surprising to me that this method is used only in one specific case and the sibling on BufferedSender is the main workhorse. I think this is because one interface is used for two different systems: an UnbufferedSender is used by a PerRangeEventSink and a BufferedSender is used by a BufferedPerRangeEventSink.

I find that keeping legacy impls around tends to greatly obfuscate the code. I am looking at this code:

	case *BufferedSender:
		return &BufferedPerRangeEventSink{
			PerRangeEventSink: NewPerRangeEventSink(rangeID, streamID, sender),
		}
	case *UnbufferedSender:
		return NewPerRangeEventSink(rangeID, streamID, sender)
	default:

and I wonder why we need to use the same sender interface in both. Can't they take different interfaces? Or can we at least make the interface "canonical" for the non-legacy (buffered) case and then just type assert in the legacy case and use the underlying type (a non-buffering sender in the unbuffered case)?

This suggestion is not for this PR. But anything that helps make this code more accessible will help for the future.

We can discuss this more in #134957.