Author: tabish
Date: Wed Apr 3 19:22:16 2013
New Revision: 1464165
URL: http://svn.apache.org/r1464165
Log:
fix and test for: https://issues.apache.org/jira/browse/AMQ-4361
Added:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java
(with props)
Modified:
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java
Modified:
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java?rev=1464165&r1=1464164&r2=1464165&view=diff
==============================================================================
---
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java
(original)
+++
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java
Wed Apr 3 19:22:16 2013
@@ -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<M
* 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<M
/**
* @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<M
* @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<M
}
}
+ @Override
public boolean isFull() {
if (parent != null && parent.isFull()) {
return true;
@@ -100,7 +107,7 @@ public class MemoryUsage extends Usage<M
/**
* 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<M
/**
* Increases the usage by the value amount.
- *
+ *
* @param value
*/
public void increaseUsage(long value) {
@@ -125,13 +132,13 @@ public class MemoryUsage extends Usage<M
}
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<M
}
}
+ @Override
protected long retrieveUsage() {
return usage;
}
+ @Override
public long getUsage() {
return usage;
}
Modified:
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java?rev=1464165&r1=1464164&r2=1464165&view=diff
==============================================================================
---
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java
(original)
+++
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/usage/Usage.java
Wed Apr 3 19:22:16 2013
@@ -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 Us
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 Us
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 Us
* 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 Us
/**
* 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 Us
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 Us
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 Us
+ (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={}",
- new Object[]{getName(), getLimit(),
parent.getLimit()});
+ LOG.info("Usage({}) limit={} should be smaller than its
parent limit={}",
+ new Object[]{getName(), getLimit(),
parent.getLimit()});
}
}
for (T t:children) {
@@ -306,13 +308,14 @@ public abstract class Usage<T extends Us
}
}
+ @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 Us
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 Us
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();
+ }
}
Added:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java?rev=1464165&view=auto
==============================================================================
---
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java
(added)
+++
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java
Wed Apr 3 19:22:16 2013
@@ -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());
+ }
+}
+
Propchange:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4361Test.java
------------------------------------------------------------------------------
svn:eol-style = native
