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.

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,8 @@ public void increaseUsage(long value) {
             return;
         }
 
-        usageLock.writeLock().lock();
-        try {
-            usage += value;
-            setPercentUsage(caclPercentUsage());
-        } finally {
-            usageLock.writeLock().unlock();
-        }
+        usage.add(value);
+        maybeUpdatePercent();
 
         if (parent != null) {
             parent.increaseUsage(value);
@@ -182,31 +178,46 @@ public void decreaseUsage(long value) {
             return;
         }
 
-        usageLock.writeLock().lock();
-        try {
-            usage -= value;
-            setPercentUsage(caclPercentUsage());
-        } finally {
-            usageLock.writeLock().unlock();
-        }
+        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;
+        this.usage.reset();
+        this.usage.add(usage);
+        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;