rangefeed: reduce goroutines for rangefeed registrations #110432

erikgrinaker · 2023-09-12T13:51:02Z

With the mux rangefeed protocol and the rangefeed scheduler in #107553, the only remaining O(ranges) goroutine scaling in rangefeeds is with registrations. These are basically responsible for buffering and sending events from the processor to the gRPC stream. We currently spin up one registration goroutine per range/client pair both for the plain and mux rangefeed protocols. In both cases, this is wasteful:

plain rangefeed: there's already a goroutine spawned by gRPC to handle the Node.RangeFeed() request that's just sitting around waiting for a future to complete. This goroutine could be reappropriated to do the actual event processing instead of spawning an additional one.
mux rangefeed: all the registrations fan in to a single stream anyway, so we could just have a single goroutine per mux that receives events from the processors and passes it on.

There is a draft PR for this in #109667, but it unfortunately won't make it in for 23.2. We should get it into 24.1 instead -- in that case, we may remove the legacy processor and only use the scheduler (and possibly even the mux), which would significantly reduce the complexity here since we won't have to support all permutations.

Touches #96395.

Jira issue: CRDB-31426

Epic CRDB-37519

The text was updated successfully, but these errors were encountered:

blathers-crl · 2023-09-12T13:51:04Z

cc @cockroachdb/replication

blathers-crl · 2023-12-14T12:00:53Z

cc @cockroachdb/cdc

blathers-crl · 2024-03-13T15:04:04Z

Hi @exalate-issue-sync[bot], please add branch-* labels to identify which branch(es) this GA-blocker affects.

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

Overview This change reduces the goroutines in the rangefeed server side code (mainly the rangefeed registry and scheduled processor). This is done by removing one goroutine per-registration and instead having a shared goroutine per client. Note, for the non-mux protocol, which is to be deprecated, this ends up adding back one goroutine per registration because there is one client per range. In the mux protocol, the number of clients is bounded from above by the number of nodes in the cluster, meaning this change reduces the number of goroutines from 1 goroutine per range to 1 goroutine per node, which is very significant. Well, actually, this change still creates one goroutine per range, but they are short lived. Each goroutine does the catchup scan for the range before it terminates. The catchup scan semaphore exists which helps the scheduling overhead in this scenario. More info below. Implementation Details Before this change, each registration would have a long lived goroutine. This goroutine was responsible for completing the catchup scan for the range and then volleying events from the processor to the RPC stream. Upon context cancellation or completion, the goroutine would shut down and the registration would unregister from the processor. Processor -> Registration Buffer + Goroutine - > RPC Stream This change updates these registration goroutines to be short lived. They are created when the registration is created and perform the catchup scan. While they do the catchup scan, the processor can place events in the registration buffer. Upon completing the catchup scan, these goroutines block the processor, empty the buffered events onto the RPC stream, then unblock the processor. After emptying their buffers, they terminate and signal to the processors that they can emit events onto the stream directly. Since there is no goroutine to detect context cancellation nor a goroutine to unregister the registration upon completing, futures/callbacks are used. See `ctxutil.WhenDone` and `done.WhenReady` in `(*ScheduledProcessor).Register`. Coordination here is very tricky because these callbacks are run by the goroutine which cancels the context / completes the future. There's some careful considerations taken here. For example, `ctxutil.WhenDone` is tricky because it actually can spawn a goroutine in some environments (ex. running tests in your ide) which might leak, so there's some extra work done to ensure the context is always cancelled. Another example is that the `WhenReady` on the future might get called by the single-threaded processor and might end up blocking on itself when it sends an unregister request to the processor. Having these goroutines, albiet short-lived, might be concerning. As discussed with Erik, there's a limit on concurrent catchup scans so most of these should not be runnable at the same time. IIRC, goroutines are created in the runnable state, so upon creation, there might be scheduling overhead. I'm unsure if this is a huge concern, I need some input here. Having the processors emit to the RPC stream is concerning because the RPC stream can block. To solve this problem, I introduced the `BufferedSender` in `pkg/kv/kvserver/rangefeed/stream_sender.go`. It's basically a linked list and a goroutine (events are memory accounted and the linked list has low overhead, so we expect this to have little to no memory impact). It implements the `BufferedSend()` method to emit events to a buffer. Then there's a goroutine which volleys these events onto the stream. This is the "one goroutine per client" mentioned above. It is shared by all the registrations for that client. `BufferedSender` also implements the blocking `Send` which emits to the RPC stream. This `BufferedSender` is created once per rangefeed client. Each registration and it's associated processor both have to be careful about when it calls `BufferedSend` and `Send` to not mess up event ordering. The contract should be that if one of those two things calls `BufferedSend`, `Send` should never be called after. At the moment `Send` is only used for the catchup scan part done by the registration. After that, `BufferedSend` is called by the processor and it should never block. For `Send`, the implementation is the same as before. The registration must free the alloc and then send the event on the RPC stream. For `BufferedSend`, we must hold on to the alloc and free it in the `BufferedSender`. Some changes have been made to accomodate this. Also, since sending is asynchronous now, error handling for sending events is done asynchronously. The `BufferedSend` method calls a callback which sends the error to the registration and disconnects it. This change is not too bad as registrations already used futures and can handle async termination. As mentioned before, one has to be careful of deadlock because the `BufferedSender` goroutine runs the callback and disconnects the reg. By the way, at the time of writing, `TestProcessorMemoryBudgetReleased` fails, so something might be wrong here. Testing To merge this PR, we need CI to be green. Since this change is so foundational, there's a lot of updates in tests that need to be done, but they aren't too bad. Firstly, there's many implementations of the RPC Stream (the `Stream` interface in `pkg/kv/kvserver/rangefeed/registry.go`). I updated the `*testStream` one to have a goroutine and implement buffered send. `*consumer` is something I'm working on right now. A lot of tests assume that registration goroutines are long lived and will have synchronization points to ensure these buffers are empty. See `syncEventAndRegistrations` in `processor_test.go`. This method is changed to wait for registration goroutines to terminate - not empty their buffers. It's a pretty small change but it changes the meaning behind what tests are doing generally. You only need to call that once per registration (waiting for the goroutine to finish) and can just call `syncEvents` every time after to wait for the processor to emit to the stream. Now, the processor is called `BufferedSend`, so you also need to wait for the `testStream` to catch up before reading the sent events. If you look at `(*testStream).Events()`, this actually does synchronization for you. Next Steps I think going test by test in the kv/kvserver package is a good start. I've been working through `processor_test.go`. After getting CI green, I think polish and scale testing is in order. Running some roachtests (cdcbench) with this code and comparing metrics to the old code should point out any performance issues. Informs: cockroachdb#110432 Release note: None Epic: None

This patch adds a node level BufferedSender which uses a queue to buffer events before forwarding events to underlying grpc stream. Part of: cockroachdb#110432 Release note: note Co-authored-by: Steven Danna [email protected]

This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: cockroachdb#110432 Release note: A new cluster setting `kv.rangefeed.buffered_stream_sender.enabled` can now be used to allow rangefeed to use buffered sender for all rangefeed feeds instead of buffering events separately per client per range. Co-authored-by: Steven Danna [email protected]

This patch adds a node level BufferedSender which uses a queue to buffer events before forwarding events to underlying grpc stream. Part of: cockroachdb#110432 Release note: None Co-authored-by: Steven Danna <[email protected]>

This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: cockroachdb#110432 Release note: A new cluster setting `kv.rangefeed.buffered_stream_sender.enabled` can now be used to allow rangefeed to use buffered sender for all rangefeed feeds instead of buffering events separately per client per range. Co-authored-by: Steven Danna [email protected]

This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: cockroachdb#110432 Release note: A new cluster setting `kv.rangefeed.buffered_stream_sender.enabled` can now be used to allow rangefeed to use buffered sender for all rangefeed feeds instead of buffering events separately per client per range. Co-authored-by: Steven Danna <[email protected]>

This patch adds a node level BufferedSender which uses a queue to buffer events before forwarding events to underlying grpc stream. Part of: cockroachdb#110432 Release note: None Co-authored-by: Steven Danna <[email protected]>

136309: kvserver/rangefeed: add node level buffered sender r=stevendanna a=wenyihu6 This patch adds a node level BufferedSender which uses a queue to buffer events before forwarding events to underlying grpc stream. Part of: #110432 Release note: None Co-authored-by: Steven Danna <[email protected]> Co-authored-by: Wenyi Hu <[email protected]>

This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: cockroachdb#110432 Release note: A new cluster setting `kv.rangefeed.buffered_stream_sender.enabled` can now be used to allow rangefeed to use buffered sender for all rangefeed feeds instead of buffering events separately per client per range. Co-authored-by: Steven Danna [email protected]

This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: cockroachdb#110432 Release note: none Co-authored-by: Steven Danna <[email protected]>

136310: kvserver/rangefeed: add unbuffered registration r=tbg,stevendanna a=wenyihu6 This patch adds unbufferedRegistration. UnbufferedRegistration is like BufferedRegistration but uses BufferedSender to buffer live raft updates instead of a using fixed size channel and having a dedicated per-registration goroutine to volley events to underlying gRPC stream. Instead, there is only one BufferedSender for each incoming node.MuxRangefeed gRPC call. BufferedSender is responsible for buffering and sending its updates to the underlying gRPC stream in a dedicated goroutine O(node). Resolved: #110432 Release note: none Co-authored-by: Steven Danna <[email protected]> 136545: clusterversion: remove old gates r=RaduBerinde a=RaduBerinde Remove 24.3 gates and corresponding code that is now obsolete. Epic: REL-1292 Release note: None 136919: build: add TC build configuration for running roachtests with profiling r=jlinder a=rickystewart Epic: CRDB-41952 Part of: CRDB-44692 Release note: None Co-authored-by: Wenyi Hu <[email protected]> Co-authored-by: Radu Berinde <[email protected]> Co-authored-by: Ricky Stewart <[email protected]>

erikgrinaker added C-performance Perf of queries or internals. Solution not expected to change functional behavior. T-kv-replication A-kv-rangefeed Rangefeed infrastructure, server+client labels Sep 12, 2023

erikgrinaker mentioned this issue Sep 12, 2023

kvserver,cdc: high rangefeed count leads to scheduler overload #96395

Closed

exalate-issue-sync bot assigned aliher1911 Sep 12, 2023

exalate-issue-sync bot assigned erikgrinaker and unassigned aliher1911 Oct 13, 2023

erikgrinaker mentioned this issue Oct 25, 2023

rangefeed: enable mux and scheduler by default #113029

Closed

erikgrinaker added the O-support Would prevent or help troubleshoot a customer escalation - bugs, missing observability/tooling, docs label Oct 25, 2023

erikgrinaker removed their assignment Dec 14, 2023

erikgrinaker added T-cdc and removed T-kv-replication labels Dec 14, 2023

blathers-crl bot added the A-cdc Change Data Capture label Dec 14, 2023

exalate-issue-sync bot assigned miretskiy Dec 20, 2023

miretskiy added the P-2 Issues/test failures with a fix SLA of 3 months label Jan 10, 2024

exalate-issue-sync bot assigned jayshrivastava and unassigned miretskiy Jan 29, 2024

exalate-issue-sync bot added the GA-blocker label Mar 13, 2024

rharding6373 added the branch-release-24.1 Used to mark GA and release blockers, technical advisories, and bugs for 24.1 label Mar 13, 2024

erikgrinaker mentioned this issue Mar 25, 2024

rangefeed: revert removal of non-mux/scheduler rangefeeds #121054

Closed

rharding6373 removed GA-blocker branch-release-24.1 Used to mark GA and release blockers, technical advisories, and bugs for 24.1 labels Mar 26, 2024

exalate-issue-sync bot unassigned jayshrivastava Apr 2, 2024

exalate-issue-sync bot assigned wenyihu6 Jun 7, 2024

craig bot closed this as completed in da8dc48 Dec 9, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

rangefeed: reduce goroutines for rangefeed registrations #110432

rangefeed: reduce goroutines for rangefeed registrations #110432

erikgrinaker commented Sep 12, 2023 •

edited by exalate-issue-sync bot

Loading

blathers-crl bot commented Sep 12, 2023

blathers-crl bot commented Dec 14, 2023

blathers-crl bot commented Mar 13, 2024

rangefeed: reduce goroutines for rangefeed registrations #110432

rangefeed: reduce goroutines for rangefeed registrations #110432

Comments

erikgrinaker commented Sep 12, 2023 • edited by exalate-issue-sync bot Loading

blathers-crl bot commented Sep 12, 2023

blathers-crl bot commented Dec 14, 2023

blathers-crl bot commented Mar 13, 2024

erikgrinaker commented Sep 12, 2023 •

edited by exalate-issue-sync bot

Loading