[enh] Broker - Shared subscription: run filters in a separate (per-subscription) thread (dispatcherDispatchMessagesInSubscriptionThread) #16603
Conversation
This file contains hidden or 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
eolivelli
requested review from
BewareMyPower,
codelipenghui,
hangc0276,
lhotari,
merlimat,
michaeljmarshall and
rdhabalia
eolivelli
added
doc-not-needed
eolivelli
changed the title
| dispatcherMaxReadBatchSize=100 | ||
|
|
||
| # Dispatch messages and execute broker side filters in a per-subscription thread | ||
| dispatcherDispatchMessagesInSubscriptionThread=true |
Contributor
There was a problem hiding this comment.
Should we keep it false by default? Most of the users don't have broker-side filters. Enabled by default will introduce more thread switching.
Contributor
Author
There was a problem hiding this comment.
I am adding this flag only to rollback in case of problem with particular workloads.
We are doing performance testing, I believe that this new behaviour is generally better, not only for the case of server side filters, but especially in these cases:
- multiple subscriptions on the same topic
- concurrent produce and consume
| } | ||
|
|
||
| protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) { | ||
| protected synchronized void sendMessagesToConsumers(ReadType readType, List<Entry> entries) { |
Contributor
There was a problem hiding this comment.
It's better to have a check and only apply the lock if dispatcherDispatchMessagesInSubscriptionThread is enabled.
Contributor
Author
There was a problem hiding this comment.
sendMessagesToConsumers is usually called by readEntriesComplete, that is already synchronized
the JVM is smart enough to understand that there is no need for extra work for synchronized
Member
There was a problem hiding this comment.
Alternatively, I think you could add a synchronized (this) block around the runnable's call to this method.
Contributor
Author
There was a problem hiding this comment.
there is no difference.
I think that this is not a problem.
@codelipenghui do you agree ?
Comment on lines
528
to
530
| topic.getBrokerService().getTopicOrderedExecutor() | ||
| .executeOrdered(name, | ||
| safeRun(() -> sendMessagesToConsumers(readType, entries))); |
Contributor
There was a problem hiding this comment.
We can have a pinned executor instead of selecting the executor for each operation.
Contributor
Author
There was a problem hiding this comment.
this is an interesting point.
Let me investigate.
I am not sure about the threadpool that we can use here as "pinned", I would like to pick a thread from "TopicOrderedExecutor"
Contributor
Author
There was a problem hiding this comment.
@codelipenghui I have pinned the Subscription to a thread.
I think that this is what you suggested
thanks
dlg99
reviewed
Jul 14, 2022
| // sendMessagesToConsumers is responsible for running broker-side filters | ||
| // that may be quite expensive | ||
| topic.getBrokerService().getTopicOrderedExecutor() | ||
| .executeOrdered(name, |
Contributor
There was a problem hiding this comment.
- do we care about potential (unchecked) RejectedExecutionException here?
- we may need to prevent this code from submitting unlimited amount of runnables into that executor / handle long backlog.
Contributor
Author
There was a problem hiding this comment.
do we care about potential (unchecked) RejectedExecutionException here?
we usually don't deal with RejectedExecutionException in Dispatcher code.
in any case even if the execution is rejected the Dispatcher will read more entries for some other reasons
we may need to prevent this code from submitting unlimited amount of runnables into that executor / handle long backlog.
This shouldn't be a problem because there is a "natural" backpressure mechanism here, we trigger this code when new entries are read from storage (BK/Offloaders) but we request new reads from Storage after dispatching them to Consumers.
So usually there is only one of this tasks that is pending for the subscription.
|
@eolivelli Please provide a correct documentation label for your PR. |
eolivelli
force-pushed
the
impl/shared-thread-per-subscription
branch
from
019493d to
7d80f8b
Compare
codelipenghui
approved these changes
Jul 15, 2022
dlg99
approved these changes
Jul 15, 2022
…ead (dispatcherDispatchMessagesInSubscriptionThread) (cherry picked from commit 9d0ae97)
eolivelli
force-pushed
the
impl/shared-thread-per-subscription
branch
from
709c3ad to
5c1cf84
Compare
lhotari
reviewed
Jul 21, 2022
...che/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
Outdated
Show resolved
Hide resolved
…ersistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>
Contributor
Author
|
/pulsarbot rerun-failure-checks |
BewareMyPower
added a commit
to BewareMyPower/pulsar
that referenced
this pull request
Jul 26, 2022
… (per-subscription) thread (dispatcherDispatchMessagesInSubscriptionThread) (apache#16603)" Fixes apache#16802 This reverts commit abe5d15.
4 tasks
BewareMyPower
added a commit
to BewareMyPower/pulsar
that referenced
this pull request
Jul 27, 2022
### Motivation See apache#16802 and the discussion in apache#16803. Before reverting apache#16603, disabling the `dispatcherDispatchMessagesInSubscriptionThread` option first. ### Modifications Change the default value of `dispatcherDispatchMessagesInSubscriptionThread` to false.
4 tasks
Member
|
This enhancement may fix the |
Member
|
And this PR has bug, the fix PR at here: #16812 |
Member
|
Hi, @gaoran10 |
|
@eolivelli Please provide a correct documentation label for your PR. |
Contributor
|
@eolivelli just wanted to check in - any docs required to be updated for this improvement? |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
8 participants
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.
Motivation
Server side filters may be very heavy weight, and currently they are running (very often) in the same thread that handles writes and also all the subscriptions.
We want to improve performances in presence of heavy server side filters:
Modifications
Add a new flag,
dispatcherDispatchMessagesInSubscriptionThread, enabled by default, to dispatch messages to consumers in a Shared subscription using a separate thread than the BK executor thread.With this patch we choose a thread depending on the topic + subscription name and we run the actual dispatching of messages (sendMessagesToConsumers) in such thread.
Verifying this change
I did some manual testing with a dummy filter that slows down (Thread.sleep(100)) consumption and even with a simply workload (50k msg/s) I have verified that: