kvserver/rangefeed: remove lockedRangefeedStream

[wenyihu6]

kvserver: wrap kvpb.RangeFeedEventSink in Stream

Previously, we declared the same interface signature twice: once in
kvpb.RangeFeedEventSink and again in rangefeed.Stream. This patch embeds
kvpb.RangeFeedEventSink inside rangefeed.Stream, making rangefeed.Stream a
superset of kvpb.RangeFeedEventSink. This approach makes sense, as each
rangefeed server stream should be a rangefeed event sink, capable of making
thread-safe rangefeed event sends.

Epic: none
Release note: none

---

kvserver/rangefeed: remove lockedRangefeedStream

Previously, we created separate locked rangefeed streams for each individual
rangefeed stream to ensure Send can be called concurrently as the underlying
grpc stream is not thread safe. However, since the introduction of the mux
rangefeed support, we already have a dedicated lock for the underlying mux
stream, making the Send method on each rangefeed stream thread safe already.
This patch removes the redundant locks from each individual rangefeed stream.

Epic: none
Release note: none

[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 

[wenyihu6]

I’m not sure the performance implication of such changes. The original way has a more fine grained mutex wrapped under every setRangeIDEventSink, but they would need to access the same underlying muxstream lock anyways. I did a small benchmarks for comparison, and it seems that having one mutex is better (less contentions and also faster). I'm not sure how accurate the benchmark is tho since I'm getting different results with code refactoring.

Benchmark
Benchmark/fine_grained_mutexes
Benchmark/fine_grained_mutexes-12         	 1465849	       863.0 ns/op
Benchmark/only_one_mutex
Benchmark/only_one_mutex-12               	 1804202	       644.2 ns/op
PASS

type SharedCounter struct {
	mu    sync.Mutex
	count int
}

func (s *SharedCounter) Increment() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.count++
}

type Counter struct {
	shared *SharedCounter
	mu     sync.Mutex
}

func NewCounter(shared *SharedCounter) *Counter {
	return &Counter{
		shared: shared,
	}
}

func (c *Counter) Increment() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.shared.Increment()
}

type OneCounter struct {
	shared *SharedCounter
}

func NewOne(shared *SharedCounter) *OneCounter {
	return &OneCounter{
		shared: shared,
	}
}

func (c *OneCounter) Increment() {
	c.shared.Increment()
}

func BenchmarkSeparateMutexes(b *testing.B) {
	sharedCounter := &SharedCounter{}
	counter1 := NewCounter(sharedCounter)
	counter2 := NewCounter(sharedCounter)

	b.ResetTimer()
	var wg sync.WaitGroup
	for i := 0; i < b.N; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter1.Increment()
		}()
	}

	for i := 0; i < b.N; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter2.Increment()
		}()
	}

	wg.Wait()
	b.StopTimer()
}

func BenchmarkOneMutexes(b *testing.B) {
	sharedCounter := &SharedCounter{}
	counter1 := NewOne(sharedCounter)
	counter2 := NewOne(sharedCounter)

	b.ResetTimer()
	var wg sync.WaitGroup
	for i := 0; i < b.N; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter1.Increment()
		}()
	}

	for i := 0; i < b.N; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter2.Increment()
		}()
	}

	wg.Wait()
	b.StopTimer()
}

func Benchmark(b *testing.B) {
	b.ReportAllocs()
	b.Run("fine grained mutexes", func(b *testing.B) {
		BenchmarkSeparateMutexes(b)
	})

	b.Run("only one mutex", func(b *testing.B) {
		BenchmarkOneMutexes(b)
	})
}

[nvanbenschoten]

Reviewed 2 of 2 files at r1, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @wenyihu6)

---

-- commits line 8 at r1:
This assumption feels brittle. What do you think about adding an "annotation method" to RangeFeedEventSink called SendIsThreadSafe so that implementors need to actively opt-in to implementing the contract?

---

pkg/kv/kvserver/replica_rangefeed.go line 326 at r1 (raw file):

	var done future.ErrorFuture
	p := r.registerWithRangefeedRaftMuLocked(
		ctx, rSpan, args.Timestamp, catchUpIter, args.WithDiff, args.WithFiltering, stream, &done,

Do you understand why we have kvpb.RangeFeedEventSink and rangefeed.Stream? Should we replace rangefeed.Stream with kvpb.RangeFeedEventSink?

[wenyihu6]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @mgartner and @nvanbenschoten)

---

-- commits line 8 at r1:

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

This assumption feels brittle. What do you think about adding an "annotation method" to RangeFeedEventSink called SendIsThreadSafe so that implementors need to actively opt-in to implementing the contract?

I like the idea! Made the changes accordingly. Lmkif you had something else in mind.

---

pkg/kv/kvserver/replica_rangefeed.go line 326 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

Do you understand why we have kvpb.RangeFeedEventSink and rangefeed.Stream? Should we replace rangefeed.Stream with kvpb.RangeFeedEventSink?

I agree that the current implementation feels redundant. In the future PR, as we add more methods to the rangefeed.Stream interface, such as Disconnect and RegisterRangefeedCleanup, their purposes start diverging. The kvpb.RangeFeedEventSink is a node level concept, and we have other RPC context tests that implement it directly for testing. It doesn’t need to implement functions like Disconnect. On the other hand, rangefeed.Stream is more specific to the rangefeed level and include functions like Disconnect. Wdyt?

[nvanbenschoten]

Reviewed 1 of 1 files at r2, 11 of 11 files at r3, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @wenyihu6)

---

-- commits line 8 at r1:

Previously, wenyihu6 (Wenyi Hu) wrote…

I like the idea! Made the changes accordingly. Lmkif you had something else in mind.

I'd actually advise we get rid of the bool return value, as that makes the property look optional. The role of the method is to indicate that Send must be thread-safe in every implementation of RangeFeedEventSink. Implementors aren't allowed to return false.

---

pkg/kv/kvserver/replica_rangefeed.go line 326 at r1 (raw file):

Previously, wenyihu6 (Wenyi Hu) wrote…

I agree that the current implementation feels redundant. In the future PR, as we add more methods to the rangefeed.Stream interface, such as Disconnect and RegisterRangefeedCleanup, their purposes start diverging. The kvpb.RangeFeedEventSink is a node level concept, and we have other RPC context tests that implement it directly for testing. It doesn’t need to implement functions like Disconnect. On the other hand, rangefeed.Stream is more specific to the rangefeed level and include functions like Disconnect. Wdyt?

That makes sense to me, thanks for explaining. Maybe we should have the rangefeed.Stream interface embed the kvpb.RangeFeedEventSink interface so that it's a clear superset.

[wenyihu6]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @nvanbenschoten)

---

-- commits line 8 at r1:

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

I'd actually advise we get rid of the bool return value, as that makes the property look optional. The role of the method is to indicate that Send must be thread-safe in every implementation of RangeFeedEventSink. Implementors aren't allowed to return false.

I see what you mean. It's a bit tricky here since we actually do have a few test streams that have thread unsafe Send method. For example,

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

Line 206 in c8476d1

func (s *noopStream) Send(*kvpb.RangeFeedEvent) error {

and

cockroach/pkg/rpc/context_test.go

Lines 291 to 293 in c8476d1

	func (s *rangefeedEventSink) Send(event *kvpb.RangeFeedEvent) error {
	return s.stream.Send(&kvpb.MuxRangeFeedEvent{RangeFeedEvent: *event})
	}

. And they need to conform to the RangeFeedEventSink interface. Do you see a clean way to handle this? I could think of

• define another kvpb.TestStream struct that implements the SendIsThreadSafe() interface. Other test streams will embed the *kvpb.TestStream struct to opt in.
• change each test stream to become thread safe

---

pkg/kv/kvserver/replica_rangefeed.go line 326 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

That makes sense to me, thanks for explaining. Maybe we should have the rangefeed.Stream interface embed the kvpb.RangeFeedEventSink interface so that it's a clear superset.

Aha, great idea. I will make the changes in the future pr above.

[wenyihu6]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @nvanbenschoten)

---

-- commits line 8 at r1:

Previously, wenyihu6 (Wenyi Hu) wrote…

I see what you mean. It's a bit tricky here since we actually do have a few test streams that have thread unsafe Send method. For example,

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

Line 206 in c8476d1

func (s *noopStream) Send(*kvpb.RangeFeedEvent) error {

and

cockroach/pkg/rpc/context_test.go

Line 295 in c8476d1

func (s *rangefeedEventSink) SendIsThreadSafe() bool { return false }

. And they need to conform to the RangeFeedEventSink interface. Do you see a clean way to handle this? I could think of

• define another kvpb.TestStream struct that implements the SendIsThreadSafe() interface. Other test streams will embed the *kvpb.TestStream struct to opt in.
• change each test stream to become thread safe

I went with the first proposed idea above. Lmk if you have a better idea in mind.

[nvanbenschoten]

Reviewed 12 of 12 files at r4, 11 of 11 files at r5, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @wenyihu6)

---

pkg/kv/kvpb/api.go line 2519 at r5 (raw file):

}

// Every stream used in prod code implementing the RangeFeedEventSink interface

This type doesn't feel like it belongs in the top-level kvpb package. Arguably, RangeFeedEventSink doesn't even belong here, but it's more reasonable.

I'd suggest that we don't introduce this at all. Just because a stream is used in a test, doesn't necessarily mean it doesn't need to be thread-safe. Instead, we should define a SendIsThreadSafe method on all RangeFeedEventSink structs. That gives us a place to talk about thread safety. Either we then make these implementations thread-safe or we just comment about how the test has organized for the sink to not need thread safety (so in a way, the test is providing the thread safety).

---

pkg/kv/kvpb/api.go line 2531 at r5 (raw file):

	// Context returns the context for this stream.
	Context() context.Context
	// Send blocks until it sends m, the stream is done, or the stream breaks.

nit: The parameter is not named here. We should either name it m, or replace "sends m" with "sends the RangeFeedEvent".

---

pkg/kv/kvpb/api.go line 2535 at r5 (raw file):

	Send(*RangeFeedEvent) error
	// SendIsThreadSafe is a no-op declaration method. It is a contract that the
	// struct has a thread safe Send method.

nit: s/struct/interface/

nit: s/thread safe/thread-safe/

[wenyihu6]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @nvanbenschoten)

---

pkg/kv/kvpb/api.go line 2519 at r5 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

This type doesn't feel like it belongs in the top-level kvpb package. Arguably, RangeFeedEventSink doesn't even belong here, but it's more reasonable.

I'd suggest that we don't introduce this at all. Just because a stream is used in a test, doesn't necessarily mean it doesn't need to be thread-safe. Instead, we should define a SendIsThreadSafe method on all RangeFeedEventSink structs. That gives us a place to talk about thread safety. Either we then make these implementations thread-safe or we just comment about how the test has organized for the sink to not need thread safety (so in a way, the test is providing the thread safety).

I see. This is much cleaner. Done.

---

pkg/kv/kvpb/api.go line 2531 at r5 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

nit: The parameter is not named here. We should either name it m, or replace "sends m" with "sends the RangeFeedEvent".

Ahh oops. Changed it sends the RangeFeedEvent.

---

pkg/kv/kvpb/api.go line 2535 at r5 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

nit: s/struct/interface/

nit: s/thread safe/thread-safe/

Done.

[nvanbenschoten]

Reviewed 11 of 11 files at r7, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @wenyihu6)

[wenyihu6]

Thanks for the thorough review!

bors r=nvanbenschoten

[craig]

Build succeeded:

• linux_amd64_build
• check_generated_code
• linux_amd64_fips_build
• local_roachtest
• lint
• docker_image_amd64
• windows_build
• unit_tests
• linux_arm64_build
• macos_arm64_build
• macos_amd64_build
• acceptance
• local_roachtest_fips
• examples_orms