Repository navigation
Opt-in topic fan-out pool with non-blocking park and atomic memory accounting - #2728
Draft
mwashburn2 wants to merge 5 commits into
Draft
mwashburn2 wants to merge 5 commits into
mwashburn2 wants to merge 5 commits into
Conversation
JMH benchmark isolating the broker's topic send path: non-persistent
broker + NON_PERSISTENT BytesMessage, vm:// transport, async producer
send with unbounded window, producer flow control off, advisories/JMX/
scheduler/statistics/audit off, zero consumers. Parameters toggle a
shared topic vs a distinct topic per producer thread (distinctTopics)
and vm async dispatch vs direct dispatch on the calling thread
(vmAsync). Named *Benchmark so surefire skips it in CI.
Build (JDK 23+ needs proc=full for the JMH annotation processor):
mvn -pl activemq-unit-tests -am install -DskipTests -Dmaven.compiler.proc=full
Run (example: direct dispatch, both topic modes, thread sweep):
CP="activemq-unit-tests/target/test-classes:activemq-unit-tests/target/classes:$(mvn -q -pl activemq-unit-tests dependency:build-classpath -Dmdep.outputFile=/dev/stdout)"
java -cp "$CP" org.openjdk.jmh.Main TopicRawThroughputBenchmark \
-p numConsumers=0 -p useVirtualThread=false -p vmAsync=false \
-p distinctTopics=false,true -f 2 -wi 3 -w 2s -i 5 -r 2s \
-jvmArgs "-Xms4g -Xmx4g -XX:+AlwaysPreTouch"
On this commit (stock code) throughput DEGRADES as producer threads are
added, even with a topic per producer (e.g. ~5.2M msgs/sec at 1 thread
falling to ~1.1M at 22 threads on an 11-core machine). JMH stack
profiling attributes ~86% of lock wait time to the MemoryUsage
ReentrantReadWriteLock: every send takes exclusive write locks on a
usage hierarchy rooted in the broker-global SystemUsage, so destination
sharding cannot avoid the contention. The following commit removes it.
Replace the exclusive-lock counter in MemoryUsage with an AtomicLong and cache the current percentUsage bucket as absolute value bounds (PercentBounds) in base Usage. Mutations and reads compare the usage value against the cached bounds and only take the usageLock when a bound is crossed, where setPercentUsage recalculates the percent and bounds and fires the usual listener events and waitForSpace signals. The locked path recomputes after publishing until stable so a concurrent update cannot leave the published percent stale. Storage usage values change externally via store.size(), so base isFull(int) and getPercentUsage() use the same bounds check to skip the write lock on reads; the StoreUsage and TempUsage per-read recompute overrides are removed and JobSchedulerUsage now refreshes its percent on read.
- Topic: async fan-out of non-persistent, non-transacted sends via org.apache.activemq.broker.region.Topic.fanoutThreads (default 0 = unchanged), tunable with fanoutQueueLimit and fanoutBatch - CountStatisticImpl: LongAdder - TimeStatisticImpl: LongAdder/LongAccumulator instead of synchronized addTime - SizeStatisticImpl: LongAdder count/total, CAS min/max instead of synchronized addSize - StatisticImpl: drop synchronized from sample-time accessors Builds on the apache#2276 backport (mattrpav).
TopicSubscription.add waited on matched.isFull() without ever retrying direct dispatch. With the fan-out pool, messages queued behind the add hold the memory it is waiting on, so the wait never ends. Dispatch directly once the consumer has caught up. Fan-out threads now skip the wait and park the message in the pending cursor (which spills to temp storage), so one frozen subscriber no longer stalls the topic's drain thread.
Producers check isFull() in Topic.send and the fan-out enqueue charged memory afterwards, so every producer that passed the check at 99% could charge and push usage past the limit (about 1% at a 4 MB limit with 32 producers). Add MemoryUsage.tryEnqueueUsage, which checks and charges with a single CAS against a limit published with the percent bounds, and Message.tryIncrementReferenceCount on top of it. enqueueFanout waits for space until the reservation succeeds, so at most one charge crosses the limit, the same as isFull() reports.
Contributor
|
Note: LongAdder can use a lot more memory than AtomicLong (32 bytes vs several kb) This will be a problem in a broker with 10,000s to 1M destinations each with a dozen or so metrics. Perhaps we need a factory to allow different backing options for MemoryUsage and Statistics |
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 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.
Stacked on #2276 (lock-free
MemoryUsage). The first two commits belong to that PR.Adds an opt-in topic fan-out pool (
fanoutThreads, default off) so one slow subscriber can't hold up a producer dispatching to many subscribers.Topichands dispatch to a bounded worker pool. Size statistics are now lock-free.TopicSubscription.addwaiting for space. If a subscriber is full, the message is parked for that subscriber and the worker moves on.MemoryUsage.tryEnqueueUsagechecks the limit and charges it in one atomic step (CAS against the published limit), andMessage.tryIncrementReferenceCountuses it. Before, many producers could each passisFull()and then all charge memory, going about 1% over the limit. Now at most one message goes over.Tests: three new
MemoryUsageTestcases.