feat(engine): mutex free subscription handling #1076
Merged
+426
−278
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jensneuse
requested changes
Feb 18, 2025
|
I answered my own change requests with a draft for improvements, except one where I'm not sure: |
jensneuse
approved these changes
Feb 18, 2025
When a origin completes a subscription, we drain inflight events before closing the connection to the client. In addition, we're only sending the complete event when all events were handled.
jensneuse
approved these changes
Feb 19, 2025
alepane21
requested changes
Feb 20, 2025
jensneuse
approved these changes
Feb 20, 2025
This was referenced Feb 20, 2025
StarpTech
added a commit
that referenced
this pull request
Feb 20, 2025
🤖 I have created a release *beep* *boop* --- ## [2.0.0-rc.157](v2.0.0-rc.156...v2.0.0-rc.157) (2025-02-20) ### Features * **engine:** mutex free subscription handling ([#1076](#1076)) ([21be4ab](21be4ab)) ### Bug Fixes * fix values validation list compatibility check ([#1082](#1082)) ([541be0d](541be0d)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please).
5 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR removes all mutexes needed and uses channels to synchronize access in subscription handling. Every subscription spawns a single goroutine where the fetch and resolve process is run (connection write). Passing a task to a subscription is done on a buffered channel and on the event loop (single threaded). In that way, we can guarantee that we send messages in the order they were received, no matter how high the concurrency is. Before, it could happen that events arrived out of order. By using a buffered channel, we can increase the throughput of how many updates can be offloaded to a subscription worker while other subscriptions can continue. This implementation make use of the natural back-pressure behaviour of go channels. A
MaxSubscriptionFetchTimeoutis added to ensure that a single subscription fetch can't block the event loop forever. This timeout is configurable and set to30sby default.Semaphores have been deleted as well because we now naturally limit the throughput by writing to subscriptions sequentially via channels. Additionally, heartbeat handling is done in the worker goroutine of every subscription. This simplified the implementation and removed another synchronization need.
We also massively improved the way synchronous subscriptions are managed. There is no difference in handling anymore, except that for SSE/Multipart, we wait until the client request is done or the subscription is completed and drained.
I did a few benchmarks with 200 concurrent subscriptions and around 10 triggers. Writing to all subscriptions was still pretty fast (<1ms, ~3ms Max). The next iteration of the implementation should focus on leveraging concurrency better. We should handle triggers individually so that the blast radius of a fast producer and slow writer and can't impact others. Today, a fast producer will result in increased latency but more stable CPU utilization due to backpressure.
Fixes ENG-6513