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/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); } +} 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(); + } + } +}