From 6f4565549e6c047c054f034e781ed2565c7ec330 Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Sat, 1 Aug 2026 21:03:33 -0500 Subject: [PATCH 1/2] Add TopicRawThroughputBenchmark - JMH raw topic-send throughput 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. --- .../region/TopicRawThroughputBenchmark.java | 197 ++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/TopicRawThroughputBenchmark.java diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/TopicRawThroughputBenchmark.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/TopicRawThroughputBenchmark.java new file mode 100644 index 00000000000..cb6a0a62f38 --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/TopicRawThroughputBenchmark.java @@ -0,0 +1,197 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.broker.region; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import jakarta.jms.Connection; +import jakarta.jms.DeliveryMode; +import jakarta.jms.MessageConsumer; +import jakarta.jms.MessageProducer; +import jakarta.jms.Session; +import jakarta.jms.BytesMessage; +import jakarta.jms.Topic; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.region.policy.PolicyEntry; +import org.apache.activemq.broker.region.policy.PolicyMap; +import org.apache.activemq.command.ActiveMQTopic; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; + +/** + * RAW topic-send throughput. Everything that adds latency is removed: + * non-persistent broker + NON_PERSISTENT messages (no store), async send with an + * unbounded producer window (no per-send round trip), vm:// transport (no network), + * producer flow control off, advisories/JMX/scheduler/stats/audit off, a single + * reused small message with ids/timestamps disabled. What remains is the cost of + * {@code Topic.doMessageSend()} itself. + * + *

{@code numConsumers=0} = pure send ceiling (empty dispatch); {@code >=1} = + * raw end-to-end delivery to fast, high-prefetch, no-op consumers. + * + *

Named {@code *Benchmark.java} so surefire's {@code **}/{@code *Test.*} skips it in CI. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@State(Scope.Benchmark) +public class TopicRawThroughputBenchmark { + + static final String TOPIC = "RAW.BENCH"; + + @Param({"false", "true"}) + boolean useVirtualThread; + + /** false = one shared topic (single send lock); true = a distinct topic per producer thread. */ + @Param({"false", "true"}) + boolean distinctTopics; + + private final java.util.concurrent.atomic.AtomicInteger threadCounter = + new java.util.concurrent.atomic.AtomicInteger(); + + /** vm transport async dispatch: true = hand off to task-runner pool (SynchronousQueue); + * false = process the send on the calling producer thread (no handoff). */ + @Param({"true", "false"}) + boolean vmAsync; + + /** 0 = send ceiling (no dispatch); >=1 = end-to-end to that many fast consumers. */ + @Param({"0", "1"}) + int numConsumers; + + private BrokerService broker; + private ActiveMQConnectionFactory connectionFactory; + private final List consumerConnections = new ArrayList<>(); + + @State(Scope.Thread) + public static class ThreadState { + Connection connection; + Session session; + MessageProducer producer; + BytesMessage message; // reused every send + + @Setup(Level.Trial) + public void setup(final TopicRawThroughputBenchmark b) throws Exception { + connection = b.connectionFactory.createConnection(); + connection.start(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + final int id = b.threadCounter.getAndIncrement(); + final String topicName = b.distinctTopics ? TOPIC + "." + id : TOPIC; + final Topic topic = session.createTopic(topicName); + producer = session.createProducer(topic); + producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); + producer.setDisableMessageID(true); + producer.setDisableMessageTimestamp(true); + // small raw byte body; created once and reused (marshalled once) + final byte[] payload = new byte[128]; + java.util.Arrays.fill(payload, (byte) 'x'); + final BytesMessage bm = session.createBytesMessage(); + bm.writeBytes(payload); + message = bm; + } + + @TearDown(Level.Trial) + public void tearDown() throws Exception { + if (producer != null) producer.close(); + if (session != null) session.close(); + if (connection != null) connection.close(); + } + } + + @Setup(Level.Trial) + public void setupBroker() throws Exception { + broker = new BrokerService(); + broker.setBrokerName("rawbench"); + broker.setPersistent(false); // no journal / no store + broker.setUseJmx(false); + broker.setAdvisorySupport(false); + broker.setSchedulerSupport(false); + broker.setEnableStatistics(false); + broker.setUseShutdownHook(false); + broker.setDeleteAllMessagesOnStartup(true); + + if (useVirtualThread) { + broker.setVirtualThreadTaskRunner(true); + } else { + broker.setDedicatedTaskRunner(false); + } + + final PolicyEntry topicPolicy = new PolicyEntry(); + topicPolicy.setTopic(">"); + topicPolicy.setProducerFlowControl(false); // never throttle producers + topicPolicy.setEnableAudit(false); + topicPolicy.setExpireMessagesPeriod(0L); + topicPolicy.setOptimizedDispatch(true); + topicPolicy.setMemoryLimit(256L * 1024 * 1024); + final PolicyMap policyMap = new PolicyMap(); + policyMap.put(new ActiveMQTopic(">"), topicPolicy); + broker.setDestinationPolicy(policyMap); + + broker.addConnector("vm://rawbench?async=" + vmAsync); + broker.start(); + broker.waitUntilStarted(); + + connectionFactory = new ActiveMQConnectionFactory("vm://rawbench?create=false&async=" + vmAsync); + connectionFactory.setUseAsyncSend(true); // pipeline sends, no per-send round trip + connectionFactory.setProducerWindowSize(0); // unbounded async window + connectionFactory.setWatchTopicAdvisories(false); + connectionFactory.setAlwaysSyncSend(false); + connectionFactory.getPrefetchPolicy().setTopicPrefetch(50000); + + for (int i = 0; i < numConsumers; i++) { + final Connection conn = connectionFactory.createConnection(); + conn.start(); + final Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + final MessageConsumer consumer = sess.createConsumer(sess.createTopic(TOPIC)); + consumer.setMessageListener(m -> { }); // no-op, drains as fast as dispatched + consumerConnections.add(conn); + } + } + + @TearDown(Level.Trial) + public void tearDownBroker() throws Exception { + for (final Connection c : consumerConnections) { + try { c.close(); } catch (final Exception ignore) { } + } + consumerConnections.clear(); + if (broker != null) { broker.stop(); broker.waitUntilStopped(); } + } + + private void send(final ThreadState s) throws Exception { + s.producer.send(s.message); + } + + @Benchmark @Threads(1) public void send_01_thread(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(4) public void send_04_threads(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(8) public void send_08_threads(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(16) public void send_16_threads(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(2) public void send_02_threads(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(11) public void send_11_threads(final ThreadState s) throws Exception { send(s); } + @Benchmark @Threads(22) public void send_22_threads(final ThreadState s) throws Exception { send(s); } +} From 47261e7b433ff77bf6a60a0c7afb1b1e7b6bb356 Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Sat, 1 Aug 2026 21:04:06 -0500 Subject: [PATCH 2/2] [#2275] Lock-free Usage accounting via AtomicLong 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. --- .../org/apache/activemq/usage/StoreUsage.java | 12 - .../org/apache/activemq/usage/TempUsage.java | 13 -- .../apache/activemq/usage/MemoryUsage.java | 84 +++++-- .../apache/activemq/usage/PercentBounds.java | 88 ++++++++ .../java/org/apache/activemq/usage/Usage.java | 69 ++++-- .../usage/MemoryUsageConcurrencyTest.java | 212 ++++++++++++++++++ .../usage/StorageUsagePercentBoundsTest.java | 89 ++++++++ 7 files changed, 502 insertions(+), 65 deletions(-) create mode 100644 activemq-client/src/main/java/org/apache/activemq/usage/PercentBounds.java create mode 100644 activemq-unit-tests/src/test/java/org/apache/activemq/usage/StorageUsagePercentBoundsTest.java diff --git a/activemq-broker/src/main/java/org/apache/activemq/usage/StoreUsage.java b/activemq-broker/src/main/java/org/apache/activemq/usage/StoreUsage.java index a639cd6abf1..3e56fc8fc79 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/usage/StoreUsage.java +++ b/activemq-broker/src/main/java/org/apache/activemq/usage/StoreUsage.java @@ -67,18 +67,6 @@ public void setStore(PersistenceAdapter store) { } } - @Override - public int getPercentUsage() { - usageLock.writeLock().lock(); - try { - percentUsage = caclPercentUsage(); - return super.getPercentUsage(); - } finally { - usageLock.writeLock().unlock(); - } - } - - @Override protected void updateLimitBasedOnPercent() { usageLock.writeLock().lock(); diff --git a/activemq-broker/src/main/java/org/apache/activemq/usage/TempUsage.java b/activemq-broker/src/main/java/org/apache/activemq/usage/TempUsage.java index 4f75d35b5cb..4f42953401e 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/usage/TempUsage.java +++ b/activemq-broker/src/main/java/org/apache/activemq/usage/TempUsage.java @@ -47,19 +47,6 @@ public TempUsage(TempUsage parent, String name) { updateLimitBasedOnPercent(); } - @Override - public int getPercentUsage() { - if (store != null) { - usageLock.writeLock().lock(); - try { - percentUsage = caclPercentUsage(); - } finally { - usageLock.writeLock().unlock(); - } - } - return super.getPercentUsage(); - } - @Override protected long retrieveUsage() { if (store == null) { diff --git a/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java b/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java index 40d2f1acc24..2a1a7681ca7 100644 --- a/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java +++ b/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java @@ -17,6 +17,7 @@ package org.apache.activemq.usage; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** * Used to keep track of how much of something is being used so that a @@ -28,7 +29,15 @@ */ public class MemoryUsage extends Usage { - private long usage; + // Lock-free usage accounting: the counter is an AtomicLong so increase/decrease never + // take an exclusive lock; the usageLock is only taken when the counter crosses out of the + // current percent bucket (at most ~100/percentUsageMinDelta times per limit traversal), + // which preserves listener events and waitForSpace signalling. AtomicLong was chosen over + // a striped LongAdder after benchmarking showed equal throughput at 1-22 producer threads + // on an 11-core machine, while AtomicLong keeps get() exact, makes setUsage() a plain + // atomic set, and avoids per-instance cell inflation. + private final AtomicLong usage = new AtomicLong(); + public MemoryUsage() { this(null, null); @@ -129,12 +138,8 @@ public boolean isFull() { if (parent != null && parent.isFull()) { return true; } - usageLock.readLock().lock(); - try { - return percentUsage >= 100; - } finally { - usageLock.readLock().unlock(); - } + // percentUsage is volatile; no lock needed for a read. + return percentUsage >= 100; } /** @@ -159,12 +164,15 @@ public void increaseUsage(long value) { return; } - usageLock.writeLock().lock(); - try { - usage += value; - setPercentUsage(caclPercentUsage()); - } finally { - usageLock.writeLock().unlock(); + // INVARIANT: every usage.addAndGet() MUST be followed unconditionally by the bounds + // check in the same method (no early return or throw between them). The liveness of + // untimed waitForSpace() depends on it: the temporally last mutation compares the + // complete counter value against the current bucket bounds, so a lasting + // 100% -> <100% transition always reaches the locked updatePercent() path, which + // signals waitForSpaceCondition. Breaking this ordering can strand waiters forever. + final long v = usage.addAndGet(value); + if (!bounds.contains(v)) { + updatePercent(); } if (parent != null) { @@ -182,12 +190,11 @@ public void decreaseUsage(long value) { return; } - usageLock.writeLock().lock(); - try { - usage -= value; - setPercentUsage(caclPercentUsage()); - } finally { - usageLock.writeLock().unlock(); + // INVARIANT: addAndGet() must be followed unconditionally by the bounds check + // (see increaseUsage for the full liveness rationale). + final long v = usage.addAndGet(-value); + if (!bounds.contains(v)) { + updatePercent(); } if (parent != null) { @@ -195,18 +202,49 @@ public void decreaseUsage(long value) { } } + /** + * Cold path, entered only when the counter crosses out of the cached percent bucket. + * Recomputes percentUsage from the live counter and publishes it via setPercentUsage() + * (firing listener events and signalling waitForSpace waiters), which also installs the + * new bucket bounds. The recompute-after-publish loop makes the update race-proof: after + * publishing we re-read the live counter, and either we observe a concurrent mutation + * (loop and correct), or that mutation's addAndGet follows our read in the counter's + * synchronization order - in which case its bounds check is guaranteed to see the bounds + * we just published and takes this path itself. + */ + private void updatePercent() { + usageLock.writeLock().lock(); + try { + int p; + do { + p = caclPercentUsage(); + setPercentUsage(p); + } while (caclPercentUsage() != p); + } finally { + usageLock.writeLock().unlock(); + } + } + + + @Override protected long retrieveUsage() { - return usage; + return usage.get(); } @Override public long getUsage() { - return usage; + return usage.get(); } - public void setUsage(long usage) { - this.usage = usage; + /** + * Sets the usage to the given value as a single atomic store; a concurrent + * increase/decrease linearizes cleanly before or after it. Note: as with the historical + * field assignment, this does not propagate an adjustment to the parent usage. + */ + public void setUsage(long value) { + this.usage.set(value); + updatePercent(); } public void setPercentOfJvmHeap(int percentOfJvmHeap) { diff --git a/activemq-client/src/main/java/org/apache/activemq/usage/PercentBounds.java b/activemq-client/src/main/java/org/apache/activemq/usage/PercentBounds.java new file mode 100644 index 00000000000..5e0c778af7d --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/usage/PercentBounds.java @@ -0,0 +1,88 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.usage; + +/** + * Internal API. The absolute usage-value bounds {@code [lower, upper)} of the range that maps + * to a Usage's current percentUsage bucket. Lock-free hot paths compare a usage value against + * these two longs - no division or percent math - and only enter the locked + * percent-recompute path when the value crosses out of the bucket. + * + *

Immutable, and held by {@link Usage} in a single volatile reference so the pair can never + * tear (two independent volatile longs could be read as a wider interval and miss a real + * crossing). + * + *

The bucket math matches {@code Usage.caclPercentUsage()} truncating-division semantics: + * percent P (a multiple of percentUsageMinDelta d, against limit L) covers values in + * {@code [ceil(P*L/100), ceil((P+d)*L/100))}. Bounds only need to be conservative - the locked + * path always derives the percent from {@code caclPercentUsage()}, so an imprecise bound costs + * at most an extra locked recompute, never a wrong percent. + */ +public final class PercentBounds { + + /** + * Sentinel whose range is empty, so any value registers as a crossing - forces the first + * observation to take the locked initialization path. + */ + public static final PercentBounds ALWAYS_CROSS = new PercentBounds(0, 0); + + public final long lower; + public final long upper; + + PercentBounds(long lower, long upper) { + this.lower = lower; + this.upper = upper; + } + + public boolean contains(long value) { + return value >= lower && value < upper; + } + + /** + * Bounds of the usage-value range that maps to the given percent bucket. + * {@code limit == 0} pins the percent at 0 (matching caclPercentUsage), so the bucket is + * unbounded. Negative percents (negative usage is an accounting-error state) collapse into + * one bucket below zero so any recovery to {@code >= 0} re-enters the locked path. + */ + public static PercentBounds compute(int percent, long limit, int minDelta) { + if (limit == 0) { + return new PercentBounds(Long.MIN_VALUE, Long.MAX_VALUE); + } + if (percent < 0) { + return new PercentBounds(Long.MIN_VALUE, 0); + } + final int delta = Math.max(1, minDelta); + final long lower = percent == 0 ? 0 : ceilDivSaturated(percent, limit); + final long upper = ceilDivSaturated((long) percent + delta, limit); + return new PercentBounds(lower, upper); + } + + /** ceil(percent * limit / 100), saturating to Long.MAX_VALUE on overflow. */ + private static long ceilDivSaturated(long percent, long limit) { + try { + final long product = Math.multiplyExact(percent, limit); + return product / 100 + (product % 100 == 0 ? 0 : 1); + } catch (ArithmeticException overflow) { + return Long.MAX_VALUE; + } + } + + @Override + public String toString() { + return "PercentBounds[" + lower + "," + upper + ")"; + } +} diff --git a/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java b/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java index 9216e84b46d..e61a96e01f7 100644 --- a/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java +++ b/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java @@ -42,18 +42,26 @@ public abstract class Usage implements Service { protected final ReentrantReadWriteLock usageLock = new ReentrantReadWriteLock(); protected final Condition waitForSpaceCondition = usageLock.writeLock().newCondition(); - protected int percentUsage; + // volatile so lock-free hot paths (isFull, percent-change detection) can read it without + // taking the usageLock; all writes still happen under the writeLock via setPercentUsage(). + protected volatile int percentUsage; + // The absolute usage-value bounds of the current percentUsage bucket, kept in one volatile + // reference (immutable pair, cannot tear). Installed under the write lock whenever + // percentUsage is published (setPercentUsage/refreshPercentUsage), which also covers limit + // and percentUsageMinDelta changes via onLimitChange/setPercentUsageMinDelta. Lock-free + // hot paths compare a usage value against it to decide if a locked recompute is needed. + protected volatile PercentBounds bounds = PercentBounds.ALWAYS_CROSS; protected T parent; protected String name; private UsageCapacity limiter = new DefaultUsageCapacity(); - private int percentUsageMinDelta = 1; + private volatile int percentUsageMinDelta = 1; private final List listeners = new CopyOnWriteArrayList(); private final boolean debug = LOG.isDebugEnabled(); private float usagePortion = 1.0f; private final List children = new CopyOnWriteArrayList(); private final List callbacks = new LinkedList(); - private int pollingTime = 100; + private volatile int pollingTime = 100; private final AtomicBoolean started = new AtomicBoolean(); private ThreadPoolExecutor executor; @@ -93,12 +101,12 @@ public boolean waitForSpace(long timeout, int highWaterMark) throws InterruptedE } usageLock.writeLock().lock(); try { - percentUsage = caclPercentUsage(); + refreshPercentUsage(caclPercentUsage()); if (percentUsage >= highWaterMark) { long deadline = timeout > 0 ? System.currentTimeMillis() + timeout : Long.MAX_VALUE; long timeleft = deadline; while (timeleft > 0) { - percentUsage = caclPercentUsage(); + refreshPercentUsage(caclPercentUsage()); if (percentUsage >= highWaterMark) { waitForSpaceCondition.await(pollingTime, TimeUnit.MILLISECONDS); timeleft = deadline - System.currentTimeMillis(); @@ -121,9 +129,15 @@ public boolean isFull(int highWaterMark) { if (parent != null && parent.isFull(highWaterMark)) { return true; } + // Fast path: while the usage value stays inside the cached percent bucket the + // published percentUsage is still valid - no lock, no division. retrieveUsage() is + // safe to call unlocked for every implementation (atomic counters or constant). + if (bounds.contains(retrieveUsage())) { + return percentUsage >= highWaterMark; + } usageLock.writeLock().lock(); try { - percentUsage = caclPercentUsage(); + refreshPercentUsage(caclPercentUsage()); return percentUsage >= highWaterMark; } finally { usageLock.writeLock().unlock(); @@ -216,21 +230,26 @@ public void setUsagePortion(float usagePortion) { } public int getPercentUsage() { - usageLock.readLock().lock(); - try { - return percentUsage; - } finally { - usageLock.readLock().unlock(); + // Fresh-on-read without a lock: if the usage value has crossed out of the cached + // percent bucket, take the write lock once and silently refresh (no listener events - + // preserving the historical behavior of read-driven recomputes). Subclasses whose + // usage value changes externally (StoreUsage/TempUsage/JobSchedulerUsage) get accurate + // reads from this shared path instead of per-class write-locked overrides. + if (!bounds.contains(retrieveUsage())) { + usageLock.writeLock().lock(); + try { + refreshPercentUsage(caclPercentUsage()); + } finally { + usageLock.writeLock().unlock(); + } } + return percentUsage; } public int getPercentUsageMinDelta() { - usageLock.readLock().lock(); - try { - return percentUsageMinDelta; - } finally { - usageLock.readLock().unlock(); - } + // volatile field - no lock needed; also avoids a nested read-lock acquisition when + // called from subclass code already holding the write lock (MemoryUsage.computeBounds) + return percentUsageMinDelta; } /** @@ -268,6 +287,7 @@ protected void setPercentUsage(int value) { try { int oldValue = percentUsage; percentUsage = value; + bounds = PercentBounds.compute(value, limiter.getLimit(), percentUsageMinDelta); if (oldValue != value) { fireEvent(oldValue, value); } @@ -276,6 +296,17 @@ protected void setPercentUsage(int value) { } } + /** + * Silently refresh the cached percentUsage (no listener events, no waiter signalling) - + * used by the internal recompute sites in waitForSpace(long,int) and isFull(int). + * Must be called with the usageLock write lock held. Subclasses that cache values derived + * from percentUsage override this to refresh them in the same critical section. + */ + protected void refreshPercentUsage(int value) { + percentUsage = value; + bounds = PercentBounds.compute(value, limiter.getLimit(), percentUsageMinDelta); + } + protected int caclPercentUsage() { if (limiter.getLimit() == 0) { return 0; @@ -432,6 +463,10 @@ public UsageCapacity getLimiter() { } /** + * Creation-time setter. Swapping the limiter on a live Usage does not trigger + * onLimitChange(): percentUsage - and any subclass caches derived from it, such as + * MemoryUsage's percent bucket bounds - remain stale until the next recompute. + * * @param limiter * the limiter to set */ diff --git a/activemq-client/src/test/java/org/apache/activemq/usage/MemoryUsageConcurrencyTest.java b/activemq-client/src/test/java/org/apache/activemq/usage/MemoryUsageConcurrencyTest.java index b8475fcafa5..c5c6f8e8eb2 100644 --- a/activemq-client/src/test/java/org/apache/activemq/usage/MemoryUsageConcurrencyTest.java +++ b/activemq-client/src/test/java/org/apache/activemq/usage/MemoryUsageConcurrencyTest.java @@ -18,6 +18,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.List; @@ -37,6 +38,217 @@ public class MemoryUsageConcurrencyTest { private static final Logger LOG = LoggerFactory.getLogger(MemoryUsageConcurrencyTest.class); + /** + * Liveness soak for the untimed waitForSpace(): unlike the timed variants (which poll), + * it blocks on waitForSpaceCondition and depends entirely on the 100% -> below-100% + * transition reaching the locked setPercentUsage() path that signals the condition. + * With lock-free accounting this rests on the invariant that every usage mutation is + * unconditionally followed by a percent re-check, so the temporally last mutation of a + * lasting drop always signals. Each round parks waiters at exactly 100%, races balanced + * churn pairs (usage never drops below full) against the bookkeeping, then issues a + * lasting one-byte drop WHILE churn is still running. After churn quiesces the usage is + * lastingly below the limit, so every waiter must be released. + */ + @Test + public void testUntimedWaitForSpaceLivenessSoak() throws Exception { + final int rounds = 150; + final int waiters = 4; + final int churners = 6; + + for (int round = 0; round < rounds; round++) { + final MemoryUsage u = new MemoryUsage(); + u.setLimit(100); + u.start(); + final AtomicBoolean churnRunning = new AtomicBoolean(true); + final CountDownLatch released = new CountDownLatch(waiters); + final List waiterThreads = new ArrayList<>(); + final List churnThreads = new ArrayList<>(); + try { + u.increaseUsage(100); // exactly full + + for (int w = 0; w < waiters; w++) { + final Thread t = new Thread(() -> { + try { + u.waitForSpace(); // untimed: no polling fallback + released.countDown(); + } catch (InterruptedException ignored) { + } + }); + t.setDaemon(true); + waiterThreads.add(t); + t.start(); + } + + // wait until every waiter is parked on the condition + final long deadline = System.currentTimeMillis() + 5000; + for (Thread t : waiterThreads) { + while (t.getState() != Thread.State.WAITING && System.currentTimeMillis() < deadline) { + Thread.yield(); + } + assertEquals("round " + round + " waiter failed to park", Thread.State.WAITING, t.getState()); + } + + // balanced churn: increase-then-decrease pairs keep usage >= 100 while + // hammering the lock-free percent bookkeeping + for (int c = 0; c < churners; c++) { + final Thread t = new Thread(() -> { + while (churnRunning.get()) { + u.increaseUsage(3); + u.decreaseUsage(3); + } + }); + t.setDaemon(true); + churnThreads.add(t); + t.start(); + } + Thread.sleep(2); + + // the lasting drop below 100%, deliberately concurrent with churn + u.decreaseUsage(1); + + Thread.sleep(1); + churnRunning.set(false); + for (Thread t : churnThreads) { + t.join(5000); + } + + // usage is now lastingly 99/100: every untimed waiter must have been signalled + assertTrue("round " + round + " untimed waitForSpace waiters not released: " + u, + released.await(10, TimeUnit.SECONDS)); + } finally { + churnRunning.set(false); + u.stop(); + } + } + } + + /** + * The cached percent-bucket bounds must keep getPercentUsage() tracking the exact + * calculated percent across bucket boundaries, over the limit, back down, and after a + * runtime limit change (which refreshes bounds via onLimitChange -> setPercentUsage). + */ + @Test + public void testPercentBoundsTracking() { + final MemoryUsage u = new MemoryUsage(); + u.setLimit(1000); + u.start(); + try { + assertEquals(0, u.getPercentUsage()); + u.increaseUsage(5); // 5/1000 -> 0% + assertEquals(0, u.getPercentUsage()); + u.increaseUsage(5); // 10/1000 -> 1% + assertEquals(1, u.getPercentUsage()); + u.increaseUsage(490); // 500/1000 -> 50% + assertEquals(50, u.getPercentUsage()); + u.increaseUsage(499); // 999/1000 -> 99% + assertEquals(99, u.getPercentUsage()); + u.increaseUsage(1); // 1000/1000 -> 100% + assertEquals(100, u.getPercentUsage()); + u.increaseUsage(50); // 1050/1000 -> 105% (over limit) + assertEquals(105, u.getPercentUsage()); + u.decreaseUsage(51); // 999/1000 -> 99% + assertEquals(99, u.getPercentUsage()); + + u.setLimit(2000); // 999/2000 -> 49%; bounds must refresh + assertEquals(49, u.getPercentUsage()); + u.increaseUsage(1); // 1000/2000 -> 50% + assertEquals(50, u.getPercentUsage()); + + u.decreaseUsage(1000); // 0/2000 -> 0% + assertEquals(0, u.getPercentUsage()); + assertEquals(0, u.getUsage()); + } finally { + u.stop(); + } + } + + @Test + public void testSetUsageSequential() { + final MemoryUsage u = new MemoryUsage(); + u.setLimit(1000); + u.start(); + try { + u.increaseUsage(100); + assertEquals(100, u.getUsage()); + u.setUsage(500); + assertEquals(500, u.getUsage()); + assertEquals(50, u.getPercentUsage()); + u.setUsage(0); + assertEquals(0, u.getUsage()); + assertEquals(0, u.getPercentUsage()); + } finally { + u.stop(); + } + } + + /** + * setUsage() racing balanced increase/decrease pairs must leave the final usage within + * one in-flight operation per thread of the set value. Each worker performs complete + * increase(v);decrease(v) pairs, so after joining, the only legal deviations from the + * set target come from pairs that straddle the set's linearization point (or its + * non-atomic LongAdder.sum() sweep): at most one op of at most maxOp per thread, in + * either direction. A setUsage() built on LongAdder.reset() can additionally lose + * concurrent updates outright (reset() is documented as safe only with no concurrent + * updates), allowing drift beyond this bound. + */ + @Test + public void testConcurrentSetUsageDriftBounded() throws Exception { + final int threads = 16; + final int maxOp = 100; + final int rounds = 200; + final long target = 123456; + + for (int round = 0; round < rounds; round++) { + final MemoryUsage u = new MemoryUsage(); + u.setLimit(1L << 40); + u.start(); + final AtomicBoolean running = new AtomicBoolean(true); + final CountDownLatch startLatch = new CountDownLatch(1); + final List workers = new ArrayList<>(); + try { + for (int t = 0; t < threads; t++) { + final int seed = round * 31 + t; + final Thread w = new Thread(() -> { + final Random r = new Random(seed); + try { + startLatch.await(); + } catch (InterruptedException e) { + return; + } + while (running.get()) { + final int v = r.nextInt(maxOp) + 1; + u.increaseUsage(v); + u.decreaseUsage(v); + } + }); + w.setDaemon(true); + workers.add(w); + w.start(); + } + + startLatch.countDown(); + Thread.sleep(2); + u.setUsage(target); + Thread.sleep(2); + running.set(false); + for (Thread w : workers) { + w.join(5000); + } + + final long drift = u.getUsage() - target; + final long bound = (long) threads * maxOp; + if (Math.abs(drift) > bound) { + LOG.info("Round {} drift {} exceeds bound {} : {}", round, drift, bound, u); + } + assertEquals("round " + round + " drift " + drift + " exceeds per-thread in-flight bound " + bound, + 0, Math.abs(drift) > bound ? drift : 0); + } finally { + running.set(false); + u.stop(); + } + } + } + @Test public void testCycle() throws Exception { final Random r = new Random(0xb4a14); diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usage/StorageUsagePercentBoundsTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usage/StorageUsagePercentBoundsTest.java new file mode 100644 index 00000000000..16721bc43e9 --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usage/StorageUsagePercentBoundsTest.java @@ -0,0 +1,89 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.usage; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.activemq.store.memory.MemoryPersistenceAdapter; +import org.junit.Test; + +/** + * Storage usage values change externally (the store grows without the Usage layer being + * told), so percent freshness comes from the read-driven bounds check in base + * Usage.getPercentUsage()/isFull(int) rather than a mutation hook. This test drives an + * externally-mutated store size and asserts the percent tracks exactly across bucket + * boundaries, over the limit, back down, and across a runtime limit change. TempUsage and + * JobSchedulerUsage share the identical base-class read path exercised here. + */ +public class StorageUsagePercentBoundsTest { + + static class SettableSizeAdapter extends MemoryPersistenceAdapter { + final AtomicLong size = new AtomicLong(); + + @Override + public long size() { + return size.get(); + } + } + + @Test + public void testStoreUsagePercentTracksExternalSizeChanges() throws Exception { + final SettableSizeAdapter adapter = new SettableSizeAdapter(); + final StoreUsage u = new StoreUsage(); + u.setLimit(1000); + u.setStore(adapter); + u.start(); + try { + assertEquals(0, u.getPercentUsage()); + adapter.size.set(5); // 5/1000 -> 0% + assertEquals(0, u.getPercentUsage()); + adapter.size.set(10); // 10/1000 -> 1% + assertEquals(1, u.getPercentUsage()); + adapter.size.set(500); // 50% + assertEquals(50, u.getPercentUsage()); + assertFalse(u.isFull(90)); + adapter.size.set(999); // 99% + assertEquals(99, u.getPercentUsage()); + assertTrue(u.isFull(90)); + assertFalse(u.isFull(100)); + adapter.size.set(1000); // 100% + assertTrue(u.isFull(100)); + assertEquals(100, u.getPercentUsage()); + adapter.size.set(1050); // 105% (over limit) + assertEquals(105, u.getPercentUsage()); + assertTrue(u.isFull(100)); + adapter.size.set(999); // back down + assertEquals(99, u.getPercentUsage()); + assertFalse(u.isFull(100)); + + u.setLimit(2000); // 999/2000 -> 49%; bounds must refresh + assertEquals(49, u.getPercentUsage()); + adapter.size.set(1000); // 50% + assertEquals(50, u.getPercentUsage()); + + adapter.size.set(0); + assertEquals(0, u.getPercentUsage()); + assertFalse(u.isFull(1)); + } finally { + u.stop(); + } + } +}