[#2275] Refactor Usage to address contention

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();

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) {

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<MemoryUsage> {
 
-    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,31 +190,61 @@ 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) {
             parent.decreaseUsage(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) {

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.
+ *
+ * <p>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).
+ *
+ * <p>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 + ")";
+    }
+}

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

@@ -42,18 +42,26 @@ 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;
+    // 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<UsageListener> listeners = new CopyOnWriteArrayList<UsageListener>();
     private final boolean debug = LOG.isDebugEnabled();
     private float usagePortion = 1.0f;
     private final List<T> children = new CopyOnWriteArrayList<T>();
     private final List<Runnable> callbacks = new LinkedList<Runnable>();
-    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
      */