Deadlock in Queue.java
----------------------
Key: AMQ-1070
URL: https://issues.apache.org/activemq/browse/AMQ-1070
Project: ActiveMQ
Issue Type: Bug
Components: Broker
Affects Versions: 4.0.2
Reporter: Tom Kaitchuck
It is possible to have a deadlock as follows:
"ActiveMQ Transport: tcp:///127.0.0.1:53335":
at
org.apache.activemq.broker.region.PrefetchSubscription.add(PrefetchSubscription.java:66)
- waiting to lock <0x90786240> (a
org.apache.activemq.broker.region.QueueSubscription)
at
org.apache.activemq.broker.region.Queue.addSubscription(Queue.java:192)
- locked <0x908fa480> (a java.util.LinkedList)
at
org.apache.activemq.broker.region.AbstractRegion.addDestination(AbstractRegion.java:93)
- locked <0x903b9b40> (a java.lang.Object)
at
org.apache.activemq.broker.region.RegionBroker.addDestination(RegionBroker.java:221)
at
org.apache.activemq.broker.BrokerFilter.addDestination(BrokerFilter.java:130)
at
org.apache.activemq.advisory.AdvisoryBroker.addDestination(AdvisoryBroker.java:142)
at
org.apache.activemq.broker.BrokerFilter.addDestination(BrokerFilter.java:130)
at
org.apache.activemq.broker.MutableBrokerFilter.addDestination(MutableBrokerFilter.java:143)
at
org.apache.activemq.broker.region.AbstractRegion.addConsumer(AbstractRegion.java:182)
- locked <0x908e6cb8> (a java.lang.Object)
at
org.apache.activemq.broker.region.RegionBroker.addConsumer(RegionBroker.java:297)
at
org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:74)
at
org.apache.activemq.advisory.AdvisoryBroker.addConsumer(AdvisoryBroker.java:78)
at
org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:74)
at
org.apache.activemq.broker.MutableBrokerFilter.addConsumer(MutableBrokerFilter.java:87)
at
org.apache.activemq.broker.AbstractConnection.processAddConsumer(AbstractConnection.java:538)
at org.apache.activemq.command.ConsumerInfo.visit(ConsumerInfo.java:296)
at
org.apache.activemq.broker.AbstractConnection.service(AbstractConnection.java:237)
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:63)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:92)
at
org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:67)
at
org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:124)
at
org.apache.activemq.transport.InactivityMonitor.onCommand(InactivityMonitor.java:123)
at
org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:88)
at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:138)
at java.lang.Thread.run(Thread.java:595)
"ActiveMQ Transport: tcp:///127.0.0.1:53315":
at org.apache.activemq.broker.region.Queue.dropEvent(Queue.java:321)
- waiting to lock <0x908fa480> (a java.util.LinkedList)
at org.apache.activemq.broker.region.Queue.dropEvent(Queue.java:315)
at
org.apache.activemq.broker.region.QueueSubscription.acknowledge(QueueSubscription.java:54)
at
org.apache.activemq.broker.region.PrefetchSubscription.acknowledge(PrefetchSubscription.java:125)
- locked <0x90786240> (a
org.apache.activemq.broker.region.QueueSubscription)
at
org.apache.activemq.broker.region.AbstractRegion.acknowledge(AbstractRegion.java:265)
at
org.apache.activemq.broker.region.RegionBroker.acknowledge(RegionBroker.java:366)
at
org.apache.activemq.broker.TransactionBroker.acknowledge(TransactionBroker.java:177)
at
org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:66)
at
org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:66)
at
org.apache.activemq.broker.MutableBrokerFilter.acknowledge(MutableBrokerFilter.java:79)
at
org.apache.activemq.broker.AbstractConnection.processMessageAck(AbstractConnection.java:445)
at org.apache.activemq.command.MessageAck.visit(MessageAck.java:179)
at
org.apache.activemq.broker.AbstractConnection.service(AbstractConnection.java:237)
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:63)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:92)
at
org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:67)
at
org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:124)
at
org.apache.activemq.transport.InactivityMonitor.onCommand(InactivityMonitor.java:123)
at
org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:88)
at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:138)
at java.lang.Thread.run(Thread.java:595)
The simple solution is in AbstractReagion.java:
@@ -89,10 +89,12 @@
// Add all consumers that are interested in the destination.
for (Iterator iter = subscriptions.values().iterator();
iter.hasNext();) {
Subscription sub = (Subscription) iter.next();
+ synchronized (sub) {
if( sub.matches(destination) ) {
dest.addSubscription(context, sub);
}
}
+ }
return dest;
}
}
@@ -104,11 +106,13 @@
if( timeout == 0 ) {
for(Iterator
iter=subscriptions.values().iterator();iter.hasNext();){
Subscription sub=(Subscription) iter.next();
+ synchronized (sub) {
if(sub.matches(destination)){
throw new JMSException("Destination still has an active
subscription: "+destination);
}
}
}
+ }
if( timeout > 0 ) {
// TODO: implement a way to notify the subscribers that we want to
take the down
@@ -125,10 +129,12 @@
// timeout<0 or we timed out, we now force any remaining
subscriptions to un-subscribe.
for(Iterator
iter=subscriptions.values().iterator();iter.hasNext();){
Subscription sub=(Subscription) iter.next();
+ synchronized (sub) {
if(sub.matches(destination)){
dest.removeSubscription(context, sub);
}
}
+ }
destinationMap.removeAll(destination);
dest.dispose(context);
@@ -173,7 +179,8 @@
// broker will not see a destination that exists in persistent
store. We may want to
// eagerly load all destinations into the broker but have an
inactive state for the
// destination which has reduced memory usage.
- //
+ synchronized (sub)
+ {
if( persistenceAdapter!=null ) {
Set inactiveDests = getInactiveDestinations();
for (Iterator iter = inactiveDests.iterator();
iter.hasNext();) {
@@ -183,6 +190,7 @@
}
}
}
+ }
subscriptions.put(info.getConsumerId(), sub);
@@ -193,16 +201,16 @@
// no mutex held. Remove is only essentially run once
// so everything after this point would be leaked.
+ synchronized (sub) {
// Add the subscription to all the matching queues.
for (Iterator iter =
destinationMap.get(info.getDestination()).iterator(); iter.hasNext();) {
Destination dest = (Destination) iter.next();
dest.addSubscription(context, sub);
}
-
if( info.isBrowser() ) {
((QueueBrowserSubscription)sub).browseDone();
}
-
+ }
return sub;
}
}
--
This message is automatically generated by JIRA.
-
If you think it was sent incorrectly contact one of the administrators:
https://issues.apache.org/activemq/secure/Administrators.jspa
-
For more information on JIRA, see: http://www.atlassian.com/software/jira
