ARTEMIS-4187 made SizeAware attributes consistent

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SizeAwareMetric.java

@@ -42,18 +42,14 @@ public class SizeAwareMetric {
    private static final AtomicIntegerFieldUpdater<SizeAwareMetric> flagUpdater = AtomicIntegerFieldUpdater.newUpdater(SizeAwareMetric.class, "flag");
    private volatile int flag = NOT_USED;
 
-   private long maxElements;
+   private long maxElements = -1; // disabled by default
 
    private long lowerMarkElements;
 
-   private long maxSize;
+   private long maxSize = -1; // disabled by default
 
    private long lowerMarkSize;
 
-   private boolean sizeEnabled = false;
-
-   private boolean elementsEnabled = false;
-
    private AddCallback onSizeCallback;
 
    private Runnable overCallback;
@@ -62,8 +58,6 @@ public class SizeAwareMetric {
 
    /** To be used in a case where we just measure elements */
    public SizeAwareMetric() {
-      this.sizeEnabled = false;
-      this.elementsEnabled = false;
    }
 
 
@@ -78,8 +72,6 @@ public class SizeAwareMetric {
       this.lowerMarkElements = lowerMarkElements;
       this.maxSize = maxSize;
       this.lowerMarkSize = lowerMarkSize;
-      this.sizeEnabled = maxSize >= 0;
-      this.elementsEnabled = maxElements >= 0;
    }
 
    public boolean isOver() {
@@ -99,12 +91,7 @@ public class SizeAwareMetric {
    }
 
    public boolean isElementsEnabled() {
-      return elementsEnabled;
+      return maxElements >= 0;
-   }
-
-   public SizeAwareMetric setElementsEnabled(boolean elementsEnabled) {
-      this.elementsEnabled = elementsEnabled;
-      return this;
    }
 
    public long getElements() {
@@ -112,12 +99,7 @@ public class SizeAwareMetric {
    }
 
    public boolean isSizeEnabled() {
-      return sizeEnabled;
+      return maxSize >= 0;
-   }
-
-   public SizeAwareMetric setSizeEnabled(boolean sizeEnabled) {
-      this.sizeEnabled = sizeEnabled;
-      return this;
    }
 
    public SizeAwareMetric setOnSizeCallback(AddCallback onSize) {
@@ -214,56 +196,69 @@ public class SizeAwareMetric {
    }
 
    private void checkUnder(long currentElements, long currentSize) {
-      if (sizeEnabled) {
+      if (isUnderSize(currentSize) && changeFlag(OVER_SIZE, PENDING_FREE)) {
-         if (currentSize <= lowerMarkSize && changeFlag(OVER_SIZE, PENDING_FREE)) {
+         if (isOverElements(currentElements) && changeFlag(PENDING_FREE, OVER_ELEMENTS)) {
-            // checking if we need to switch from OVER_SIZE to OVER_ELEMENTS, to avoid calling under needless
+            logger.debug("Switch from OVER_SIZE to OVER_ELEMENTS [currentSize={}, currentElements={}, lowerMarkSize={}, maxElements={}]",
-            if (!(elementsEnabled && currentElements >= maxElements && changeFlag(PENDING_FREE, OVER_ELEMENTS))) {
+                    currentSize, currentElements, lowerMarkSize, maxElements);
-               try {
+            return;
-                  under();
+         }
-               } finally {
+
-                  changeFlag(PENDING_FREE, FREE);
+         try {
-               }
+            logger.debug("UnderSize [currentSize={}, lowerMarkSize={}]", currentSize, lowerMarkSize);
-            }
+            under();
-            return; // we must return now as we already checked for the elements portion
+         } finally {
+            changeFlag(PENDING_FREE, FREE);
          }
       }
 
-      if (elementsEnabled) {
+      if (isUnderElements(currentElements) && changeFlag(OVER_ELEMENTS, PENDING_FREE)) {
-         if (currentElements <= lowerMarkElements && changeFlag(OVER_ELEMENTS, PENDING_FREE)) {
+         if (isOverSize(currentSize) && changeFlag(PENDING_FREE, OVER_SIZE)) {
-            // checking if we need to switch from OVER_ELEMENTS to OVER_SIZE, to avoid calling under needless
+            logger.debug("Switch from OVER_ELEMENTS to OVER_SIZE [currentElements={}, currentSize={}, lowerMarkElements={}, maxSize={}]",
-            if (!(sizeEnabled && currentSize >= maxSize && changeFlag(PENDING_FREE, OVER_SIZE))) {
+                    currentElements, currentSize, lowerMarkElements, maxSize);
-               // this is checking the other side from size (elements).
+            return;
-               // on this case we are just switching sides and we should not fire under();
+         }
-               try {
+
-                  under();
+         try {
-               } finally {
+            logger.debug("UnderElements [currentElements={}, lowerMarkElements={}]", currentElements, lowerMarkElements);
-                  changeFlag(PENDING_FREE, FREE);
+            under();
-               }
+         } finally {
-            }
+            changeFlag(PENDING_FREE, FREE);
-            return; // this return here is moot I know. I am keeping it here for consistence with the size portion
-                    //  and in case eventually further checks are added on this method, this needs to be reviewed.
          }
       }
    }
 
+   private boolean isUnderSize(long currentSize) {
+      return isSizeEnabled() && currentSize < lowerMarkSize;
+   }
+
+   private boolean isOverSize(long currentSize) {
+      return isSizeEnabled() && currentSize >= maxSize;
+   }
+
+   private boolean isUnderElements(long currentElements) {
+      return isElementsEnabled() && currentElements < lowerMarkElements;
+   }
+
+   private boolean isOverElements(long currentElements) {
+      return isElementsEnabled() && currentElements >= 0 && currentElements >= maxElements;
+   }
+
    private void checkOver(long currentElements, long currentSize) {
-      if (sizeEnabled) {
+      if (isOverSize(currentSize) && changeFlag(FREE, PENDING_OVER_SIZE)) {
-         if (currentSize >= maxSize && changeFlag(FREE, PENDING_OVER_SIZE)) {
+         try {
-            try {
+            logger.debug("OverSize [currentSize={}, maxSize={}]", currentSize, maxSize);
-               over();
+            over();
-            } finally {
+         } finally {
-               changeFlag(PENDING_OVER_SIZE, OVER_SIZE);
+            changeFlag(PENDING_OVER_SIZE, OVER_SIZE);
-            }
          }
       }
 
-      if (elementsEnabled && currentElements >= 0) {
+      if (isOverElements(currentElements) && changeFlag(FREE, PENDING_OVER_ELEMENTS)) {
-         if (currentElements >= maxElements && changeFlag(FREE, PENDING_OVER_ELEMENTS)) {
+         try {
-            try {
+            logger.debug("currentElements [currentSize={}, maxElements={}]", currentElements, maxElements);
-               over();
+            over();
-            } finally {
+         } finally {
-               changeFlag(PENDING_OVER_ELEMENTS, OVER_ELEMENTS);
+            changeFlag(PENDING_OVER_ELEMENTS, OVER_ELEMENTS);
-            }
          }
       }
    }

artemis-commons/src/test/java/org/apache/activemq/artemis/utils/SizeAwareMetricTest.java

@@ -32,7 +32,7 @@ public class SizeAwareMetricTest {
 
    ExecutorService executor;
 
-   private void setupExecutor(int threads) throws Exception {
+   private void setupExecutor(int threads) {
       if (executor == null) {
          executor = Executors.newFixedThreadPool(threads);
       }
@@ -48,7 +48,7 @@ public class SizeAwareMetricTest {
    }
 
    @Test
-   public void testWithParent() throws Exception {
+   public void testWithParent() {
       AtomicBoolean childBoolean = new AtomicBoolean(false);
       AtomicBoolean parentBoolean = new AtomicBoolean(false);
 
@@ -227,7 +227,7 @@ public class SizeAwareMetricTest {
 
 
    @Test
-   public void testMaxElements() throws Exception {
+   public void testMaxElements() {
       SizeAwareMetric metric = new SizeAwareMetric(10000, 500, 10,10);
 
       AtomicBoolean over = new AtomicBoolean(false);
@@ -251,7 +251,7 @@ public class SizeAwareMetricTest {
 
    }
    @Test
-   public void testMaxElementsReleaseNonSizeParentMetric() throws Exception {
+   public void testMaxElementsReleaseNonSizeParentMetric() {
       SizeAwareMetric metricMain = new SizeAwareMetric(10000, 500, 10,10);
       SizeAwareMetric metric = new SizeAwareMetric(10000, 500, 1000,1000);
 
@@ -299,7 +299,7 @@ public class SizeAwareMetricTest {
 
 
    @Test
-   public void testMaxElementsReleaseNonSize() throws Exception {
+   public void testMaxElementsReleaseNonSize() {
       SizeAwareMetric metric = new SizeAwareMetric(10000, 500, 10,10);
 
       AtomicBoolean over = new AtomicBoolean(false);
@@ -350,7 +350,7 @@ public class SizeAwareMetricTest {
       final AtomicBoolean globalMetricOver = new AtomicBoolean(false);
       final AtomicBoolean[] metricOverArray = new AtomicBoolean[THREADS];
 
-      SizeAwareMetric globalMetric = new SizeAwareMetric(10000, 500, 0, 0);
+      SizeAwareMetric globalMetric = new SizeAwareMetric(10000, 500, 10000, 500);
 
       SizeAwareMetric[] metric = new SizeAwareMetric[THREADS];
 
@@ -369,24 +369,24 @@ public class SizeAwareMetricTest {
       CyclicBarrier flagStart = new CyclicBarrier(THREADS + 1);
       for (int istart = 0; istart < THREADS; istart++) {
          final AtomicBoolean metricOver = new AtomicBoolean(false);
-         final SizeAwareMetric themetric = new SizeAwareMetric(1000, 500, 0, 0);
+         final SizeAwareMetric theMetric = new SizeAwareMetric(1000, 500, 1000, 500);
-         themetric.setOnSizeCallback(globalMetric::addSize);
+         theMetric.setOnSizeCallback(globalMetric::addSize);
-         themetric.setOverCallback(() -> {
+         theMetric.setOverCallback(() -> {
             metricOver.set(true);
             metricOverCalls.incrementAndGet();
          });
-         themetric.setUnderCallback(() -> {
+         theMetric.setUnderCallback(() -> {
             metricOver.set(false);
             metricUnderCalls.incrementAndGet();
          });
-         metric[istart] = themetric;
+         metric[istart] = theMetric;
          metricOverArray[istart] = metricOver;
          executor.execute(() -> {
             try {
                flagStart.await(10, TimeUnit.SECONDS);
 
                for (int iadd = 0; iadd < ELEMENTS; iadd++) {
-                  themetric.addSize(1);
+                  theMetric.addSize(1);
                }
                latchDone.countDown();
             } catch (Throwable e) {
@@ -463,7 +463,7 @@ public class SizeAwareMetricTest {
    }
 
    @Test
-   public void testUpdateMax() throws Exception {
+   public void testUpdateMax() {
       AtomicBoolean over = new AtomicBoolean(false);
       SizeAwareMetric metric = new SizeAwareMetric(1000, 500, -1, -1);
       metric.setOverCallback(() -> over.set(true));
@@ -472,31 +472,31 @@ public class SizeAwareMetricTest {
       metric.addSize(900);
       Assert.assertFalse(over.get());
 
-      metric.setMax(800, 700, 0, 0);
+      metric.setMax(800, 700, -1, -1);
       Assert.assertTrue(over.get());
 
-      metric.addSize(-200);
+      metric.addSize(-201);
       Assert.assertFalse(over.get());
    }
 
    @Test
-   public void testDisabled() throws Exception {
+   public void testDisabled() {
       AtomicBoolean over = new AtomicBoolean(false);
-      SizeAwareMetric metric = new SizeAwareMetric(0, 0, -1, -1);
+      SizeAwareMetric metric = new SizeAwareMetric(-1, -1, -1, -1);
-      metric.setSizeEnabled(false);
       metric.setOverCallback(() -> over.set(true));
       metric.addSize(100);
+
       Assert.assertEquals(100, metric.getSize());
       Assert.assertEquals(1, metric.getElements());
       Assert.assertFalse(over.get());
    }
 
    @Test
-   public void testMultipleNonSized() throws Exception {
+   public void testMultipleNonSized() {
       AtomicBoolean over = new AtomicBoolean(false);
-      final SizeAwareMetric metricMain = new SizeAwareMetric(0, 0, 1, 1);
+      final SizeAwareMetric metricMain = new SizeAwareMetric(-1, -1, -1, -1);
-      SizeAwareMetric metric = new SizeAwareMetric(0, 0, 1, 1);
+      SizeAwareMetric metric = new SizeAwareMetric(-1, -1, -1, -1);
-      metric.setSizeEnabled(false);
+
       metric.setOverCallback(() -> over.set(true));
       metric.setOnSizeCallback(metricMain::addSize);
       for (int i = 0; i  < 10; i++) {
@@ -516,30 +516,29 @@ public class SizeAwareMetricTest {
       Assert.assertEquals(200, metric.getSize());
       Assert.assertEquals(10, metricMain.getElements());
       Assert.assertEquals(10, metric.getElements());
+
+      Assert.assertFalse(over.get());
    }
 
    @Test
-   public void testResetNeverUsed() throws Exception {
+   public void testResetNeverUsed() {
-      SizeAwareMetric metric = new SizeAwareMetric(0, 0, 0, 0);
       AtomicBoolean over = new AtomicBoolean(false);
 
+      SizeAwareMetric metric = new SizeAwareMetric(0, 0, 0, 0);
       metric.setOverCallback(() -> over.set(true));
-      metric.setElementsEnabled(true);
-      metric.setSizeEnabled(true);
       metric.setMax(0, 0, 0, 0);
+
       Assert.assertFalse(over.get());
       Assert.assertFalse(metric.isOver());
    }
 
    @Test
-   public void testSwitchSides() throws Exception {
+   public void testSwitchSides() {
       SizeAwareMetric metric = new SizeAwareMetric(2000, 2000, 1, 1);
       AtomicBoolean over = new AtomicBoolean(false);
 
       metric.setOverCallback(() -> over.set(true));
       metric.setUnderCallback(() -> over.set(false));
-      metric.setElementsEnabled(true);
-      metric.setSizeEnabled(true);
 
       metric.addSize(2500, true);
 
@@ -633,11 +632,18 @@ public class SizeAwareMetricTest {
 
       Assert.assertTrue(done.await(10, TimeUnit.SECONDS));
 
-
       Assert.assertEquals(0, metric.getSize());
       Assert.assertEquals(0, metric.getElements());
       Assert.assertEquals(0, errors.get());
-
    }
 
+   @Test
+   public void testConsistency() {
+      SizeAwareMetric metric = new SizeAwareMetric(-1, -1, -1, -1);
+      Assert.assertFalse(metric.isSizeEnabled());
+      Assert.assertFalse(metric.isElementsEnabled());
+      metric.setMax(1, 1, 1, 1);
+      Assert.assertTrue(metric.isSizeEnabled());
+      Assert.assertTrue(metric.isElementsEnabled());
+   }
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java

@@ -142,8 +142,6 @@ public final class PagingManagerImpl implements PagingManager {
       this.maxSize = maxSize;
       this.maxMessages = maxMessages;
       this.globalSizeMetric = new SizeAwareMetric(maxSize, maxSize, maxMessages, maxMessages);
-      globalSizeMetric.setSizeEnabled(maxSize >= 0);
-      globalSizeMetric.setElementsEnabled(maxMessages >= 0);
       globalSizeMetric.setOverCallback(() -> setGlobalFull(true));
       globalSizeMetric.setUnderCallback(() -> setGlobalFull(false));
       this.managerExecutor = pagingSPI.newExecutor();

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java

@@ -217,9 +217,6 @@ public class PagingStoreImpl implements PagingStore {
 
    private void configureSizeMetric() {
       size.setMax(maxSize, maxSize, maxMessages, maxMessages);
-      size.setSizeEnabled(maxSize >= 0);
-      size.setElementsEnabled(maxMessages >= 0);
-
    }
 
    /**