fix and test for: https://issues.apache.org/jira/browse/AMQ-4361

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1464165 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -20,9 +20,9 @@ package org.apache.activemq.usage;
  * Used to keep track of how much of something is being used so that a
  * productive working set usage can be controlled. Main use case is manage
  * memory usage.
- * 
+ *
  * @org.apache.xbean.XBean
- * 
+ *
  */
 public class MemoryUsage extends Usage<MemoryUsage> {
 
@@ -36,7 +36,7 @@ public class MemoryUsage extends Usage<MemoryUsage> {
      * Create the memory manager linked to a parent. When the memory manager is
      * linked to a parent then when usage increased or decreased, the parent's
      * usage is also increased or decreased.
-     * 
+     *
      * @param parent
      */
     public MemoryUsage(MemoryUsage parent) {
@@ -58,14 +58,19 @@ public class MemoryUsage extends Usage<MemoryUsage> {
     /**
      * @throws InterruptedException
      */
+    @Override
     public void waitForSpace() throws InterruptedException {
         if (parent != null) {
             parent.waitForSpace();
         }
         synchronized (usageMutex) {
-            for (int i = 0; percentUsage >= 100; i++) {
+            while (percentUsage >= 100 && isStarted()) {
                 usageMutex.wait();
             }
+
+            if (percentUsage >= 100 && !isStarted()) {
+                throw new InterruptedException("waitForSpace stopped during wait.");
+            }
         }
     }
 
@@ -74,6 +79,7 @@ public class MemoryUsage extends Usage<MemoryUsage> {
      * @throws InterruptedException
      * @return true if space
      */
+    @Override
     public boolean waitForSpace(long timeout) throws InterruptedException {
         if (parent != null) {
             if (!parent.waitForSpace(timeout)) {
@@ -88,6 +94,7 @@ public class MemoryUsage extends Usage<MemoryUsage> {
         }
     }
 
+    @Override
     public boolean isFull() {
         if (parent != null && parent.isFull()) {
             return true;
@@ -100,7 +107,7 @@ public class MemoryUsage extends Usage<MemoryUsage> {
     /**
      * Tries to increase the usage by value amount but blocks if this object is
      * currently full.
-     * 
+     *
      * @param value
      * @throws InterruptedException
      */
@@ -111,7 +118,7 @@ public class MemoryUsage extends Usage<MemoryUsage> {
 
     /**
      * Increases the usage by the value amount.
-     * 
+     *
      * @param value
      */
     public void increaseUsage(long value) {
@@ -125,13 +132,13 @@ public class MemoryUsage extends Usage<MemoryUsage> {
         }
         setPercentUsage(percentUsage);
         if (parent != null) {
-            ((MemoryUsage)parent).increaseUsage(value);
+            parent.increaseUsage(value);
         }
     }
 
     /**
      * Decreases the usage by the value amount.
-     * 
+     *
      * @param value
      */
     public void decreaseUsage(long value) {
@@ -149,10 +156,12 @@ public class MemoryUsage extends Usage<MemoryUsage> {
         }
     }
 
+    @Override
     protected long retrieveUsage() {
         return usage;
     }
 
+    @Override
     public long getUsage() {
         return usage;
     }

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

@@ -31,9 +31,9 @@ import org.slf4j.LoggerFactory;
  * Used to keep track of how much of something is being used so that a
  * productive working set usage can be controlled. Main use case is manage
  * memory usage.
- * 
+ *
  * @org.apache.xbean.XBean
- * 
+ *
  */
 public abstract class Usage<T extends Usage> implements Service {
 
@@ -74,7 +74,7 @@ public abstract class Usage<T extends Usage> implements Service {
     public boolean waitForSpace(long timeout) throws InterruptedException {
         return waitForSpace(timeout, 100);
     }
-    
+
     /**
      * @param timeout
      * @throws InterruptedException
@@ -108,7 +108,7 @@ public abstract class Usage<T extends Usage> implements Service {
     public boolean isFull() {
         return isFull(100);
     }
-    
+
     public boolean isFull(int highWaterMark) {
         if (parent != null && parent.isFull(highWaterMark)) {
             return true;
@@ -138,7 +138,7 @@ public abstract class Usage<T extends Usage> implements Service {
      * usagePortion to 0 since the UsageManager is not going to be portion based
      * off the parent.
      * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used
-     * 
+     *
      * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
      */
     public void setLimit(long limit) {
@@ -201,7 +201,7 @@ public abstract class Usage<T extends Usage> implements Service {
     /**
      * Sets the minimum number of percentage points the usage has to change
      * before a UsageListener event is fired by the manager.
-     * 
+     *
      * @param percentUsageMinDelta
      * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
      */
@@ -242,9 +242,9 @@ public abstract class Usage<T extends Usage> implements Service {
 
     private void fireEvent(final int oldPercentUsage, final int newPercentUsage) {
         if (debug) {
-            LOG.debug(getName() + ": usage change from: " + oldPercentUsage + "% of available memory, to: " 
+            LOG.debug(getName() + ": usage change from: " + oldPercentUsage + "% of available memory, to: "
                 + newPercentUsage + "% of available memory");
-        }   
+        }
         if (started.get()) {
             // Switching from being full to not being full..
             if (oldPercentUsage >= 100 && newPercentUsage < 100) {
@@ -262,6 +262,7 @@ public abstract class Usage<T extends Usage> implements Service {
             if (!listeners.isEmpty()) {
                 // Let the listeners know on a separate thread
                 Runnable listenerNotifier = new Runnable() {
+                    @Override
                     public void run() {
                         for (Iterator<UsageListener> iter = listeners.iterator(); iter.hasNext();) {
                             UsageListener l = iter.next();
@@ -290,14 +291,15 @@ public abstract class Usage<T extends Usage> implements Service {
                 + (parent != null ? ";Parent:" + parent.toString() : "");
     }
 
+    @Override
     @SuppressWarnings("unchecked")
     public void start() {
         if (started.compareAndSet(false, true)){
             if (parent != null) {
                 parent.addChild(this);
                 if(getLimit() > parent.getLimit()) {
-                	LOG.info("Usage({}) limit={} should be smaller than its parent limit={}", 
+                    LOG.info("Usage({}) limit={} should be smaller than its parent limit={}",
-                			 new Object[]{getName(), getLimit(), parent.getLimit()});
+                             new Object[]{getName(), getLimit(), parent.getLimit()});
                 }
             }
             for (T t:children) {
@@ -306,13 +308,14 @@ public abstract class Usage<T extends Usage> implements Service {
         }
     }
 
+    @Override
     @SuppressWarnings("unchecked")
     public void stop() {
         if (started.compareAndSet(true, false)){
             if (parent != null) {
                 parent.removeChild(this);
             }
-            
+
             //clear down any callbacks
             synchronized (usageMutex) {
                 usageMutex.notifyAll();
@@ -348,6 +351,7 @@ public abstract class Usage<T extends Usage> implements Service {
         if (parent != null) {
             Runnable r = new Runnable() {
 
+                @Override
                 public void run() {
                     synchronized (usageMutex) {
                         if (percentUsage >= 100) {
@@ -411,11 +415,16 @@ public abstract class Usage<T extends Usage> implements Service {
     public void setParent(T parent) {
         this.parent = parent;
     }
-    
+
     public void setExecutor (ThreadPoolExecutor executor) {
         this.executor = executor;
     }
+
     public ThreadPoolExecutor getExecutor() {
         return executor;
     }
+
+    public boolean isStarted() {
+        return started.get();
+    }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java

@@ -0,0 +1,157 @@
+/**
+ * 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.bugs;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.MessageProducer;
+import javax.jms.ObjectMessage;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.broker.region.policy.VMPendingQueueMessageStoragePolicy;
+import org.apache.activemq.broker.region.policy.VMPendingSubscriberMessageStoragePolicy;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AMQ4361Test {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AMQ4361Test.class);
+
+    private BrokerService service;
+    private String brokerUrlString;
+
+    @Before
+    public void setUp() throws Exception {
+        service = new BrokerService();
+        service.setDeleteAllMessagesOnStartup(true);
+        service.setUseJmx(false);
+
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry policy = new PolicyEntry();
+        policy.setMemoryLimit(1);
+        policy.setPendingSubscriberPolicy(new VMPendingSubscriberMessageStoragePolicy());
+        policy.setPendingQueuePolicy(new VMPendingQueueMessageStoragePolicy());
+        policy.setProducerFlowControl(true);
+        policyMap.setDefaultEntry(policy);
+        service.setDestinationPolicy(policyMap);
+
+        service.setAdvisorySupport(false);
+        brokerUrlString = service.addConnector("tcp://localhost:0").getPublishableConnectString();
+        service.start();
+        service.waitUntilStarted();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        if (service != null) {
+            service.stop();
+            service.waitUntilStopped();
+        }
+    }
+
+    @Test
+    public void testCloseWhenHunk() throws Exception {
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrlString);
+        connectionFactory.setProducerWindowSize(1024);
+
+        // TINY QUEUE is flow controlled after 1024 bytes
+        final ActiveMQDestination destination =
+            ActiveMQDestination.createDestination("queue://TINY_QUEUE", (byte) 0xff);
+
+        Connection connection = connectionFactory.createConnection();
+        connection.start();
+        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final MessageProducer producer = session.createProducer(destination);
+        producer.setTimeToLive(0);
+        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+        final AtomicReference<Exception> publishException = new AtomicReference<Exception>(null);
+        final AtomicReference<Exception> closeException = new AtomicReference<Exception>(null);
+        final AtomicLong lastLoop = new AtomicLong(System.currentTimeMillis() + 100);
+
+        Thread pubThread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    byte[] data = new byte[1000];
+                    new Random(0xdeadbeef).nextBytes(data);
+                    for (int i = 0; i < 10000; i++) {
+                        lastLoop.set(System.currentTimeMillis());
+                        ObjectMessage objMsg = session.createObjectMessage();
+                        objMsg.setObject(data);
+                        producer.send(destination, objMsg);
+                    }
+                } catch (Exception e) {
+                    publishException.set(e);
+                }
+            }
+        }, "PublishingThread");
+        pubThread.start();
+
+        // wait for publisher to deadlock
+        while (System.currentTimeMillis() - lastLoop.get() < 2000) {
+            Thread.sleep(100);
+        }
+        LOG.info("Publisher deadlock detected.");
+
+        Thread closeThread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    LOG.info("Attempting close..");
+                    producer.close();
+                } catch (Exception e) {
+                    closeException.set(e);
+                }
+            }
+        }, "ClosingThread");
+        closeThread.start();
+
+        try {
+            closeThread.join(30000);
+        } catch (InterruptedException ie) {
+            assertFalse("Closing thread didn't complete in 10 seconds", true);
+        }
+
+        try {
+            pubThread.join(30000);
+        } catch (InterruptedException ie) {
+            assertFalse("Publishing thread didn't complete in 10 seconds", true);
+        }
+
+        assertNull(closeException.get());
+        assertNotNull(publishException.get());
+    }
+}
+