[#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.

Author: mattrpav

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
      */