ARTEMIS-4569 Blocked Producers will hold runnables until messages are…

… consumed.

When initially developed the expectation was that no more producers would keep connecting but in a scenario like this
the consumers could actually give up and things will just accumulate on the server.

We should cleanup these upon disconnect.

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

@@ -18,9 +18,13 @@
 package org.apache.activemq.artemis.utils.runnables;
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.Consumer;
 
 public abstract class AtomicRunnable implements Runnable {
 
+   public static AtomicRunnable delegate(Runnable runnable) {
+      return new AtomicRunnableWithDelegate(runnable);
+   }
 
    public static AtomicRunnable checkAtomic(Runnable run) {
       if (run instanceof AtomicRunnable) {
@@ -30,6 +34,27 @@ public static AtomicRunnable checkAtomic(Runnable run) {
       }
    }
 
+   private RunnableList acceptedList;
+   private Consumer<AtomicRunnable> cancelTask;
+
+   public RunnableList getAcceptedList() {
+      return acceptedList;
+   }
+
+   public AtomicRunnable setAcceptedList(RunnableList acceptedList) {
+      this.acceptedList = acceptedList;
+      return this;
+   }
+
+   public Consumer<AtomicRunnable> getCancel() {
+      return cancelTask;
+   }
+
+   public AtomicRunnable setCancel(Consumer<AtomicRunnable> cancelTask) {
+      this.cancelTask = cancelTask;
+      return this;
+   }
+
    private volatile int ran;
 
    private static final AtomicIntegerFieldUpdater<AtomicRunnable> RAN_UPDATE =
@@ -52,7 +77,21 @@ public boolean isRun() {
    @Override
    public void run() {
       if (RAN_UPDATE.compareAndSet(this, 0, 1)) {
-         atomicRun();
+         try {
+            atomicRun();
+         } finally {
+            if (acceptedList != null) {
+               acceptedList.remove(AtomicRunnable.this);
+            }
+         }
+      }
+   }
+
+   public void cancel() {
+      if (RAN_UPDATE.compareAndSet(this, 0, 1)) {
+         if (cancelTask != null) {
+            cancelTask.accept(AtomicRunnable.this);
+         }
       }
    }
 

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

@@ -0,0 +1,56 @@
+/*
+ * 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.artemis.utils.runnables;
+
+import java.util.HashSet;
+import java.util.function.Consumer;
+
+public class RunnableList {
+
+   private final HashSet<AtomicRunnable> list = new HashSet<>();
+
+   public RunnableList() {
+   }
+
+   public synchronized void add(AtomicRunnable runnable) {
+      runnable.setAcceptedList(this);
+      list.add(runnable);
+   }
+
+   public int size() {
+      return list.size();
+   }
+
+   public synchronized void remove(AtomicRunnable runnable) {
+      list.remove(runnable);
+   }
+
+   public synchronized void cancel() {
+      list.forEach(this::cancel);
+      list.clear();
+   }
+
+   private void cancel(AtomicRunnable atomicRunnable) {
+      atomicRunnable.cancel();
+   }
+
+   public void forEach(Consumer<AtomicRunnable> consumerRunnable) {
+      list.forEach(consumerRunnable);
+   }
+
+}

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

@@ -0,0 +1,135 @@
+/*
+ * 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.artemis.utils.runnables;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RunnableListTest {
+
+   HashSet<AtomicRunnable> masterList = new HashSet<>();
+
+   @Test
+   public void testRunning() {
+      AtomicInteger result = new AtomicInteger();
+
+      RunnableList listA = new RunnableList();
+      RunnableList listB = new RunnableList();
+      RunnableList listC = new RunnableList();
+
+      RunnableList[] lists = new RunnableList[]{listA, listB, listC};
+      for (RunnableList l : lists) {
+         for (int i = 0; i < 10; i++) {
+            AtomicRunnable runnable = new AtomicRunnable() {
+               @Override
+               public void atomicRun() {
+                  result.incrementAndGet();
+                  masterList.remove(this);
+               }
+            };
+            addItem(l, runnable);
+         }
+      }
+
+      Assert.assertEquals(30, masterList.size());
+
+      runList(listA);
+
+      Assert.assertEquals(10, result.get());
+
+      Assert.assertEquals(20, masterList.size());
+      Assert.assertEquals(0, listA.size());
+      Assert.assertEquals(10, listB.size());
+      Assert.assertEquals(10, listC.size());
+
+      HashSet<AtomicRunnable> copyList = new HashSet<>();
+      copyList.addAll(masterList);
+
+      copyList.forEach(r -> r.run());
+
+      for (RunnableList l : lists) {
+         Assert.assertEquals(0, l.size());
+      }
+
+      Assert.assertEquals(30, result.get());
+   }
+
+   @Test
+   public void testCancel() {
+      AtomicInteger result = new AtomicInteger();
+
+      RunnableList listA = new RunnableList();
+      RunnableList listB = new RunnableList();
+      RunnableList listC = new RunnableList();
+
+      RunnableList[] lists = new RunnableList[]{listA, listB, listC};
+      for (RunnableList l : lists) {
+         for (int i = 0; i < 10; i++) {
+            AtomicRunnable runnable = new AtomicRunnable() {
+               @Override
+               public void atomicRun() {
+                  result.incrementAndGet();
+                  masterList.remove(this);
+               }
+            };
+            addItem(l, runnable);
+         }
+      }
+
+      Assert.assertEquals(30, masterList.size());
+
+      listA.cancel();
+
+      Assert.assertEquals(0, result.get());
+
+      Assert.assertEquals(20, masterList.size());
+      Assert.assertEquals(0, listA.size());
+      Assert.assertEquals(10, listB.size());
+      Assert.assertEquals(10, listC.size());
+
+      listB.cancel();
+      listC.cancel();
+
+      for (RunnableList l : lists) {
+         Assert.assertEquals(0, l.size());
+      }
+
+      Assert.assertEquals(0, masterList.size());
+   }
+
+   // runs all AtomicRunnables inside the list
+   private void runList(RunnableList list) {
+      // make a copy of all the tasks to a new list
+      ArrayList<AtomicRunnable> toRun = new ArrayList<>();
+      list.forEach(toRun::add);
+
+      // run all the elements
+      toRun.forEach(r -> r.run());
+   }
+
+   private void addItem(RunnableList list, AtomicRunnable runnable) {
+      list.add(runnable);
+      runnable.setCancel(masterList::remove);
+      masterList.add(runnable);
+   }
+
+}

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java

@@ -64,6 +64,7 @@
 import org.apache.activemq.artemis.utils.SelectorTranslator;
 import org.apache.activemq.artemis.utils.SimpleIDGenerator;
 import org.apache.activemq.artemis.utils.UUIDGenerator;
+import org.apache.activemq.artemis.utils.runnables.RunnableList;
 import org.apache.qpid.proton.amqp.Binary;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.messaging.Accepted;
@@ -112,6 +113,8 @@ public class AMQPSessionCallback implements SessionCallback {
 
    private ProtonTransactionHandler transactionHandler;
 
+   private final RunnableList blockedRunnables = new RunnableList();
+
    public AMQPSessionCallback(AMQPConnectionCallback protonSPI,
                               ProtonProtocolManager manager,
                               AMQPConnectionContext connection,
@@ -384,6 +387,7 @@ public String tempQueueName() {
    }
 
    public void close() throws Exception {
+      blockedRunnables.cancel();
       //need to check here as this can be called if init fails
       if (serverSession != null) {
          // we cannot hold the nettyExecutor on this rollback here, otherwise other connections will be waiting
@@ -610,7 +614,7 @@ public void flow(final SimpleString address,
          } else {
             final PagingStore store = manager.getServer().getPagingManager().getPageStore(address);
             if (store != null) {
-               store.checkMemory(runnable);
+               store.checkMemory(runnable, blockedRunnables::add);
             } else {
                runnable.run();
             }

artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallbackTest.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.artemis.protocol.amqp.broker;
 
 import java.util.concurrent.Executor;
+import java.util.function.Consumer;
 
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.paging.PagingManager;
@@ -157,7 +158,7 @@ public void testOfferProducerWithAddressDoesNotTopOffCreditAboveThreshold() thro
       session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection));
 
       // Run the credit refill code.
-      Mockito.verify(pagingStore).checkMemory(argument.capture());
+      Mockito.verify(pagingStore).checkMemory(argument.capture(), Mockito.isA(Consumer.class));
       assertNotNull(argument.getValue());
       argument.getValue().run();
 
@@ -188,7 +189,7 @@ public void testOfferProducerWithAddressTopsOffCreditAtThreshold() throws Except
       session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection));
 
       // Run the credit refill code.
-      Mockito.verify(pagingStore).checkMemory(argument.capture());
+      Mockito.verify(pagingStore).checkMemory(argument.capture(), Mockito.isA(Consumer.class));
       assertNotNull(argument.getValue());
       argument.getValue().run();
 
@@ -249,7 +250,7 @@ public void testOfferProducerWithAddressDoesNotGrantNegativeCredit() throws Exce
       session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(1, AMQP_LOW_CREDITS_DEFAULT, receiver, connection));
 
       // Run the credit refill code.
-      Mockito.verify(pagingStore).checkMemory(argument.capture());
+      Mockito.verify(pagingStore).checkMemory(argument.capture(), Mockito.isA(Consumer.class));
       assertNotNull(argument.getValue());
       argument.getValue().run();