[#2275] Lock-free MemoryUsage accounting via LongAdder

Replace the exclusive-lock byte counter in MemoryUsage with a striped
java.util.concurrent.atomic.LongAdder and make percentUsage volatile.
increaseUsage/decreaseUsage become lock-free adds; the existing
usageLock/setPercentUsage path (listener events, waitForSpace
signalling) is entered only when the rounded percentUsage actually
changes - at most ~100/percentUsageMinDelta locked updates per limit
traversal instead of one exclusive write lock (plus the parent chain's)
per message. isFull() becomes a volatile read. waitForSpace and all
public APIs are unchanged.

Rationale: profiling the previous commit's benchmark showed ~86% of
lock wait time in this class - Topic.send -> isFull (read lock),
Message.incrementReferenceCount -> increaseUsage (write lock) and
decrementReferenceCount -> decreaseUsage (write lock), each recursing
into the broker-global SystemUsage parent. That made usage accounting a
broker-wide serialization point that destination sharding cannot avoid.

The multi-producer degradation curve is eliminated; the shared-topic
and uncontended single-producer cases also improve (~+22% / ~+11%).
Allocation is unchanged (JMH gc profiler: ~480 vs ~527 B/op, GC count
~0 in both), so the gain is purely lock behaviour.

Semantics validation (all pass against this change):
ProducerFlowControlTest (7), ProducerFlowControlSendFailTest (8),
TopicProducerFlowControlTest (3), CompositeMessageCursorUsageTest (1),
QueueMemoryAndStoreUsageCleanupTest (1).

Known trade-off: between an add and the locked percent update there is
a sub-percentUsageMinDelta staleness window; rapid crossings may
coalesce listener events (pairs stay consistent and the locked update
recomputes from the live sum, so it self-corrects). StoreUsage and
TempUsage still use the base-class locked path and can be converted the
same way as a follow-up.

Fix setUsage() concurrent safety - delta set instead of LongAdder.reset()

LongAdder.reset() is documented as safe only when there are no
concurrent updates: a racing increaseUsage/decreaseUsage could be lost
outright (or, in narrow cell-sweep interleavings, applied out of
program order), permanently corrupting the accounting. Implement
setUsage(value) as a single delta add instead:

    usage.add(value - usage.sum());

Every interleaving with a concurrent update now maps to a legal
linearization (the update is preserved, as if ordered after the set).
Also rename the parameter to avoid shadowing the field, and document
that setUsage - like the historical plain field assignment - does not
propagate an adjustment to the parent usage.

Tests: adds a sequential setUsage semantics test and a 200-round
concurrent drift-bound hammer (16 threads of balanced increase/decrease
pairs racing a mid-flight setUsage; final usage must stay within one
in-flight op per thread of the set value). Honest note: the hammer did
NOT empirically catch the reset() implementation in 200 rounds - a
same-thread increase/decrease pair usually lands in the same adder
cell, so both are wiped or both survive and the loss window is
vanishingly narrow. The fix is justified by LongAdder's documented
contract; the test remains as a regression guard on the delta
implementation's bound.

Validation: MemoryUsageConcurrencyTest (3) and MemoryUsageTest (5)
pass; ProducerFlowControlTest, ProducerFlowControlSendFailTest and
TopicProducerFlowControlTest (18 total) pass against the patched
client. setUsage has no production callers (test/config only), so this
is hardening, not a behavior change for the broker runtime.

Harden lock-free usage accounting - liveness soak, invariant docs

Untimed waitForSpace() has no polling fallback: it blocks on
waitForSpaceCondition and depends entirely on a lasting full ->
not-full transition reaching the locked setPercentUsage() path. With
lock-free accounting that liveness rests on an ordering invariant
(every usage.add() is unconditionally followed by maybeUpdatePercent(),
so the temporally last mutation of a lasting drop recomputes from the
complete sum and signals). This commit:

- documents the invariant loudly at both mutation sites in MemoryUsage
  so a future edit cannot silently break it, and
- adds testUntimedWaitForSpaceLivenessSoak: 150 rounds parking untimed
  waiters at exactly 100% usage, racing balanced increase/decrease
  churn (usage never drops below full), then issuing the lasting
  one-byte drop WHILE churn runs; after churn quiesces every waiter
  must be released. Passes consistently.

GC evidence for the review record (send_08_threads, distinctTopics,
direct dispatch, JDK 25, -prof gc): per-op allocation is unchanged by
the LongAdder change - 467.5 +/- 36.4 B/op stock vs 440.0 +/- 0.2 B/op
patched. Absolute alloc rate rises only in proportion to throughput
(1.34M -> 5.69M ops/s in the profiled runs); gc.time was 17ms vs 19ms
over the measurement window. The change adds no per-operation
allocation and no meaningful GC pressure.

Author: mattrpav

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.LongAdder;
 
 /**
  * Used to keep track of how much of something is being used so that a
@@ -28,7 +29,11 @@
  */
 public class MemoryUsage extends Usage<MemoryUsage> {
 
-    private long usage;
+    // PROTOTYPE: lock-free usage accounting. The counter is a striped LongAdder so
+    // increase/decrease never take an exclusive lock; the usageLock is only taken when the
+    // rounded percentUsage actually changes (at most ~100/percentUsageMinDelta times per
+    // limit traversal), which preserves listener events and waitForSpace signalling.
+    private final LongAdder usage = new LongAdder();
 
     public MemoryUsage() {
         this(null, null);
@@ -129,12 +134,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,13 +160,14 @@ public void increaseUsage(long value) {
             return;
         }
 
-        usageLock.writeLock().lock();
-        try {
-            usage += value;
-            setPercentUsage(caclPercentUsage());
-        } finally {
-            usageLock.writeLock().unlock();
-        }
+        // INVARIANT: every usage.add() MUST be followed unconditionally by
+        // maybeUpdatePercent() in the same method (no early return or throw between them).
+        // The liveness of untimed waitForSpace() depends on it: the temporally last mutation
+        // recomputes the percent from a sum that includes all completed updates, so a lasting
+        // 100% -> <100% transition always reaches the locked setPercentUsage() path, which
+        // signals waitForSpaceCondition. Breaking this ordering can strand waiters forever.
+        usage.add(value);
+        maybeUpdatePercent();
 
         if (parent != null) {
             parent.increaseUsage(value);
@@ -182,31 +184,55 @@ public void decreaseUsage(long value) {
             return;
         }
 
-        usageLock.writeLock().lock();
-        try {
-            usage -= value;
-            setPercentUsage(caclPercentUsage());
-        } finally {
-            usageLock.writeLock().unlock();
-        }
+        // INVARIANT: add() must be followed unconditionally by maybeUpdatePercent()
+        // (see increaseUsage for the full liveness rationale).
+        usage.add(-value);
+        maybeUpdatePercent();
 
         if (parent != null) {
             parent.decreaseUsage(value);
         }
     }
 
+    /**
+     * Fast-path percent maintenance: a dirty compare against the volatile percentUsage; only
+     * when the rounded percent has actually changed do we take the writeLock and run the
+     * existing setPercentUsage() (which fires listener events and signals waitForSpace
+     * waiters). setPercentUsage recomputes from the live sum under the lock, so the last
+     * writer always stores a fresh value and transient races self-correct on the next update.
+     */
+    private void maybeUpdatePercent() {
+        if (caclPercentUsage() != percentUsage) {
+            usageLock.writeLock().lock();
+            try {
+                setPercentUsage(caclPercentUsage());
+            } finally {
+                usageLock.writeLock().unlock();
+            }
+        }
+    }
+
     @Override
     protected long retrieveUsage() {
-        return usage;
+        return usage.sum();
     }
 
     @Override
     public long getUsage() {
-        return usage;
+        return usage.sum();
     }
 
-    public void setUsage(long usage) {
-        this.usage = usage;
+    /**
+     * Sets the usage to the given value. Implemented as a delta adjustment because
+     * LongAdder.reset() is only safe when there are no concurrent updates - a racing
+     * increase/decrease could be lost outright. With a delta add, a concurrent update is
+     * always preserved, equivalent to it linearizing after the set. Note: as with the
+     * historical field assignment, this does not propagate an adjustment to the parent
+     * usage.
+     */
+    public void setUsage(long value) {
+        this.usage.add(value - this.usage.sum());
+        maybeUpdatePercent();
     }
 
     public void setPercentOfJvmHeap(int percentOfJvmHeap) {

activemq-client/src/main/java/org/apache/activemq/usage/Usage.java

@@ -42,7 +42,9 @@ public abstract class Usage<T extends 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;
     protected T parent;
     protected String name;
 

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,177 @@ 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<Thread> waiterThreads = new ArrayList<>();
+            final List<Thread> 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();
+            }
+        }
+    }
+
+    @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<Thread> 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);