activemq-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Stirling Chow (JIRA)" <>
Subject [jira] [Updated] (AMQ-4138) Network bridges can deadlock when memory limit exceeded
Date Fri, 26 Oct 2012 21:53:12 GMT


Stirling Chow updated AMQ-4138:


Includes some logging statements on TopicSubscription.add and an artificial delay on TopicSubscription.acknowledge
> Network bridges can deadlock when memory limit exceeded
> -------------------------------------------------------
>                 Key: AMQ-4138
>                 URL:
>             Project: ActiveMQ
>          Issue Type: Bug
>    Affects Versions: 5.7.0
>            Reporter: Stirling Chow
>            Priority: Critical
>         Attachments: BridgeMemoryLimitDeadlockTest.txt,
> Symptom
> =======
> We have a network of 4 brokers that share messages using distributed queues via demand
forwarding bridges.  We were validating the behaviour of the system when memory usage approached
and exceeded the out-of-box memory limit (64MB).
> We discovered that with great frequency the bridges would appear to stop functioning
--- no messages were being produced or consumed.  We've experienced similar behaviour when
producer flow control is activated, but in our tests, we'd turned producer flow control off
(both to avoid bridges stalling due to producer flow control and so that we could produce
enough messages to exceed the memory limit).
> The system would never recover from this deadlock. 
> Cause
> =====
> We found a number of threads looping indefinitely with the following stack:
> {code}
> Daemon Thread [ActiveMQ VMTransport: vm://broker1#7-1] (Suspended)	
> 	owns: Topic  (id=109)	
> 	waiting for: Object  (id=110)	
> 	Object.wait(long) line: not available [native method]	
> 	TopicSubscription.add(MessageReference) line: 135	
> 	SimpleDispatchPolicy.dispatch(MessageReference, MessageEvaluationContext, List<Subscription>)
line: 48	
> 	Topic.dispatch(ConnectionContext, Message) line: 680	
> 	Topic.doMessageSend(ProducerBrokerExchange, Message) line: 491	
> 	Topic.send(ProducerBrokerExchange, Message) line: 427	
> 	ManagedTopicRegion(AbstractRegion).send(ProducerBrokerExchange, Message) line: 407	
> 	ManagedRegionBroker(RegionBroker).send(ProducerBrokerExchange, Message) line: 503	
> 	ManagedRegionBroker.send(ProducerBrokerExchange, Message) line: 311	
> 	AdvisoryBroker.fireAdvisory(ConnectionContext, ActiveMQTopic, Command, ConsumerId, ActiveMQMessage)
line: 551	
> 	AdvisoryBroker.fireConsumerAdvisory(ConnectionContext, ActiveMQDestination, ActiveMQTopic,
Command, ConsumerId) line: 500	
> 	AdvisoryBroker.fireConsumerAdvisory(ConnectionContext, ActiveMQDestination, ActiveMQTopic,
Command) line: 486	
> 	AdvisoryBroker.addConsumer(ConnectionContext, ConsumerInfo) line: 98	
> 	CompositeDestinationBroker(BrokerFilter).addConsumer(ConnectionContext, ConsumerInfo)
line: 89	
> 	TransactionBroker(BrokerFilter).addConsumer(ConnectionContext, ConsumerInfo) line: 89

> 	BrokerService$5(MutableBrokerFilter).addConsumer(ConnectionContext, ConsumerInfo) line:
> 	ManagedTransportConnection(TransportConnection).processAddConsumer(ConsumerInfo) line:
> 	ConsumerInfo.visit(CommandVisitor) line: 332	
> 	ManagedTransportConnection(TransportConnection).service(Command) line: 294	
> 	TransportConnection$1.onCommand(Object) line: 152	
> 	ResponseCorrelator.onCommand(Object) line: 116	
> 	MutexTransport.onCommand(Object) line: 50	
> 	VMTransport.iterate() line: 241	
> 	PooledTaskRunner.runTask() line: 129	
> 	PooledTaskRunner$ line: 47	
> 	ThreadPoolExecutor$Worker.runTask(Runnable) line: 886	
> 	ThreadPoolExecutor$ line: 908	
> line: 662	
> {code}
> The spinning threads were associated with the VMTransport TaskRunner from {{DemandForwardingBridgeSupport.localBroker}}.
 Since the TaskRunner was essentially blocked processing one message, all other messages being
forwarded from the remote end of the bridge (e.g., ACKs) were getting queued, but not processed,
which made the bridge appear to be stalled.
> The message being processed by the spinning thread was a ConsumerInfo representing a
demand subscription from the remote broker, and was being forwarded to a subscription on the
associated consumer advisory topic.  The subscription was waiting for memory to become available
in the {{matchedListMutex}} loop:
> {}
>     public void add(MessageReference node) throws Exception {
>         if (isDuplicate(node)) {
>             return;
>         }
>         enqueueCounter.incrementAndGet();
>         if (!isFull() && matched.isEmpty()  && !isSlave()) {
>             // if maximumPendingMessages is set we will only discard messages which
>             // have not been dispatched (i.e. we allow the prefetch buffer to be filled)
>             dispatch(node);
>             setSlowConsumer(false);
>         } else {
>             if (info.getPrefetchSize() > 1 && matched.size() > info.getPrefetchSize())
>                 // Slow consumers should log and set their state as such.
>                 if (!isSlowConsumer()) {
>                     LOG.warn(toString() + ": has twice its prefetch limit pending, without
an ack; it appears to be slow");
>                     setSlowConsumer(true);
>                     for (Destination dest: destinations) {
>                         dest.slowConsumer(getContext(), this);
>                     }
>                 }
>             }
>             if (maximumPendingMessages != 0) {
>                 boolean warnedAboutWait = false;
>                 while (active) {
>                     synchronized (matchedListMutex) {
>                         while (matched.isFull()) {
>                             if (getContext().getStopping().get()) {
>                                 LOG.warn(toString() + ": stopped waiting for space in
pendingMessage cursor for: "
>                                         + node.getMessageId());
>                                 enqueueCounter.decrementAndGet();
>                                 return;
>                             }
>                             if (!warnedAboutWait) {
>                        + ": Pending message cursor [" +
>                                         + "] is full, temp usage ("
>                                         + +matched.getSystemUsage().getTempUsage().getPercentUsage()
>                                         + "%) or memory usage ("
>                                         + matched.getSystemUsage().getMemoryUsage().getPercentUsage()
>                                         + "%) limit reached, blocking message add() pending
the release of resources.");
>                                 warnedAboutWait = true;
>                             }
>                             matchedListMutex.wait(20);
>                         }
>                         // Temporary storage could be full - so just try to add the message
>                         // see
>                         if (matched.tryAddMessageLast(node, 10)) {
>                             break;
>                         }
>                     }
>                 }
> {code}
> The {{matched.isFull()}} check that was keeping the thread in the loop is only cleared
once the memory usage is reduced:
> {}
> public boolean isFull() {
>     return systemUsage != null ? systemUsage.getMemoryUsage().isFull() : false;
> }
> {code}
> Since the looping thread is essentially stalling the VMTransport on the local side of
the bridge, no dispatch ACKs can be processed for messages sent from the local bridge to the
remote broker.  If all consumers are on the remote broker and ACKs are not being processed,
then memory usage on the local broker is never reduced, thus creating a deadlock.
> In order for {{TopicSubscription.add(...)}} to enter the (essentially infinite) loop,
the following conditions must *not* be met:
> {}
> public void add(MessageReference node) throws Exception {
>         if (isDuplicate(node)) {
>             return;
>         }
>         enqueueCounter.incrementAndGet();
>         if (!isFull() && matched.isEmpty()  && !isSlave()) {
> {code}
> {{isFull()}} is true if the number of unacknowledged messages dispatched to the subscription
is greater than the subscription's prefetch size.  So for the deadlock to occur, two things
must happen:
> # There must be multiple consumers being added to the queue, thus generating multiple
adds to the consumer advisory topic subscription
> # There must be a delay in processing ACKs to the topic subscription so that it becomes
> For reasons to do with distribution management, our network connectors have a prefetch
size of 1, so under load, the deadlock occurs easily.
> I've attached a test case that clearly demonstrates the deadlock.  The test case is simple:
> # Two brokers (broker1 and broker2) are bidirectionally bridged with a network prefetch
of 1
> # broker1 (with producer flow control disabled), produces enough messages to a test queue
so that the memory limit is exceeded
> # broker2 starts two consumers of the test queue, and the broker1->broker2 bridge
forwards two demand subscriptions to broker1
> # broker1 processes the demand subscriptions and starts dispatching messages to broker2
> # Since broker2 has a bridge back to broker1, broker1's processing of the demand subscriptions
generates two consumer advisory messages that are sent over the consumer advisory topic to
broker2 (of course, broker2 ignores them since they are represent its consumers)
> # As messages are dispatched to broker2's instance of the test queue, ACKs are forwarded
by the broker1->broker2 bridge and processed by broker1, reducing the memory usage
> # Eventually all messages are consumed by broker2 and broker1's memory usage is 0.
> This test case generally passes since the deadlock requires a specific race condition:
namely, the first consumer advisory message needs to be "in flight" when the second consumer
advisory message is sent.  Since the network prefetch is 1, when the second advisory message
is processed, the topic subscription is "full", and the thread sending the advisory will wait
for matched.isFull() to be false.
> In order to increase the chancethe first consumer advisory message is "in flight", simply
add a small sleep to TopicSubscription's acknowledge method:
> {}
> public synchronized void acknowledge(final ConnectionContext context, final MessageAck
ack) throws Exception {
>"Acknowledge subscription to " + ack.getDestination().getPhysicalName());
>     Thread.sleep(1000);
> {code}
> The sleep increases the window and pretty much guarantees that the test case will fail
(i.e., messages remain in broker1's test queue since the bridge is stalled).
> Even with the sleep in place, if the number of consumers on broker2 is reduced to 1,
the test case will pass.  Again, this is because at least two consumer advisory messages are
needed to fill the subscription prefetch.
> The use of prefetch=1 for the network connector is simply so that the unit test can demonstrate
the deadlock with 2 consumers.  The deadlock can occur with any prefetch if the number of
consumers is at least prefetch + 1.
> This is a serious issue for us as our deployment involves many networked brokers and
a lot of dynamic consumer creation, which produces frequent consumer advisory messages.  When
a network bridge locks up, it cripples our system.
> Solution
> ========
> The essential problem is that sending a consumer advisory message to a topic can take
an indefinite amount of time (i.e., waits indefinitely until memory is available), and during
this time, no other messages sent to the VMTransport are processed.
> The principle tenant of the TaskRunners used in AMQ is that they implement "cooperative
multi-tasking", this means that task execution has to be reasonably quick so that other tasks
can be run.  A task that runs indefinitely breaks the utility of the model.
> While it would be possible to use {{TopicSubscription.maximumPendingMessages}} to prevent
the {{matchedListMutex}} loop from being entered, this would result in the consumer advisory
message being discarded thus the loss of any demand subscriptions that would have resulted
--- so it's not an option.
> Unfortunately, without understanding the system further, I can't offer a specific solution.

This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see:

View raw message