activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Kevin Burton <bur...@spinn3r.com>
Subject Re: Proof that a lock GCing inactive destinations cause ActiveMQ to halt consumers and producers and DoS the broker
Date Mon, 09 Feb 2015 01:34:46 GMT
I could probably make one..

I think it would just be a broker with leveldb config, inactive GC of about
60 seconds, code which writes 1 message to each of about 5000 queues,
consumes them, then closes all consumers.

After once the GC kicks in it will take a long time for ActiveMQ to GC all
the destinations.

Once I get to a state where I have my code at least reliable/stable, I’ll
try to code up something.

I’m wondering if I’m in anti-pattern territory though.  If this is NOT a
use case which ActiveMQ services , then I should really expect a fix.

But it seems that the performance of GCing destinations should be higher
than this.

Seems you should be able to GC something like 100-500 destinations per
second.

On Sun, Feb 8, 2015 at 5:22 PM, James Carman <james@carmanconsulting.com>
wrote:

> Do you have an example project which will exhibit the issue reliably?
>
> On Sunday, February 8, 2015, Kevin Burton <burton@spinn3r.com> wrote:
>
> > OK. Now I have proof that my theory was correct :)
> >
> > If you have a lot of inactive queues that need to be GCd.. these will
> block
> > consumers and producers and effectively shut down ActiveMQ during this
> > process.
> >
> > In my usage, we see 30 minutes of activity where ActiveMQ is unresponsive
> > and effectively dead :-(
> >
> > I spent all of last week trying to work around this by allowing ActiveMQ
> to
> > GC queues by releasing them more aggressively hoping to amortize this
> > process but I don’t think that fixed the issue.
> >
> > This is a report of a ‘stack report’ tool I wrote 5 or so years ago which
> > we use internally.  It runs jstack then builds a graph looking at the
> lock
> > IDs and then ranks them accordingly, sorting by hot spot descending.
> >
> > So here we have 13 threads, which have 12 inbound thread blocking them (3
> > distinct stack traces)
> >
> > Note the addConsumer and addProducer… so no new consumers can be created
> > while this process happens.
> >
> > Is this mutex lock strictly needed? Perhaps one strategy is to have one
> > thread which does the removeDestination and then just an AtomicBoolean
> > flagging this reference as “gone” and pending removal.
> >
> > the main thing I’m worried about is that I’m the only one seeing this
> issue
> > …
> >
> > I don’t really understand why this is happening though and why it’s so
> > slow.
> >
> > I’m looking at my log right now and it has about 3200 queues that it’s
> > GCing at the moment.
> >
> > It looks like it’s able to do about 10 per minute.  So obviously this is
> > just going to take a long time.
> >
> > So I think my main idea is to move to an all memory ActiveMQ broker for
> > now.  I can just get good at QUICKLY rebuilding the queue during total
> > queue failure.  This is just a short term work around though..
> >
> > I’ll either have to figure out away to fix this, completely redesign my
> app
> > (which will be no fun), build my own queue server with special semantics,
> > or implement some sort of snapshotting support and logging.
> >
> > If I were to build this myself doing logs and checkpoints, while keeping
> > the whole thing in memory, I think that would be faster.  But of course
> > that would take more time :-(
> >
> > -------------
> > Threads: 13 , Unique waiting threads: 3 , Total waiting threads: 12
> > -------------
> >     java.lang.Thread.State: WAITING (on object monitor)
> >         at java.lang.Object.wait(Native Method)
> >         at java.lang.Object.wait(Object.java:503)
> >         at
> > org.fusesource.hawtdispatch.SettableFuture$class.await(Future.scala:71)
> >         - locked <0xXXXXXXXXXXXXXXXX> (a
> > org.fusesource.hawtdispatch.SettableFuture$mutex$)
> >         at
> > org.fusesource.hawtdispatch.Future$$anon$1.await(Future.scala:122)
> >         at
> org.fusesource.hawtdispatch.Future$class.apply(Future.scala:28)
> >         at
> > org.fusesource.hawtdispatch.Future$$anon$1.apply(Future.scala:122)
> >         at
> >
> >
> org.fusesource.hawtdispatch.package$RichExecutorTrait$class.sync(hawtdispatch.scala:106)
> >         at
> >
> >
> org.fusesource.hawtdispatch.package$RichExecutor.sync(hawtdispatch.scala:142)
> >         at
> >
> >
> org.apache.activemq.leveldb.DBManager.destroyQueueStore(DBManager.scala:769)
> >         at
> >
> >
> org.apache.activemq.leveldb.LevelDBStore$$anonfun$removeQueueMessageStore$1.apply(LevelDBStore.scala:588)
> >         at
> >
> >
> org.apache.activemq.leveldb.LevelDBStore$$anonfun$removeQueueMessageStore$1.apply(LevelDBStore.scala:587)
> >         at scala.Option.foreach(Option.scala:245)
> >         at
> >
> >
> org.apache.activemq.leveldb.LevelDBStore.removeQueueMessageStore(LevelDBStore.scala:587)
> >         - locked <0xXXXXXXXXXXXXXXXX> (a
> > org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter)
> >         at
> >
> >
> org.apache.activemq.broker.region.DestinationFactoryImpl.removeDestination(DestinationFactoryImpl.java:113)
> >         at
> >
> >
> org.apache.activemq.broker.region.AbstractRegion.dispose(AbstractRegion.java:592)
> >         at
> >
> >
> org.apache.activemq.broker.region.AbstractRegion.removeDestination(AbstractRegion.java:222)
> >         at
> >
> >
> org.apache.activemq.broker.jmx.ManagedQueueRegion.removeDestination(ManagedQueueRegion.java:62)
> >         at
> >
> >
> org.apache.activemq.broker.region.RegionBroker.removeDestination(RegionBroker.java:340)
> >         at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeDestination(BrokerFilter.java:172)
> >         at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeDestination(BrokerFilter.java:172)
> >         at
> >
> >
> org.apache.activemq.advisory.AdvisoryBroker.removeDestination(AdvisoryBroker.java:212)
> >         at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeDestination(BrokerFilter.java:172)
> >         at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeDestination(BrokerFilter.java:172)
> >         at
> >
> >
> org.apache.activemq.broker.MutableBrokerFilter.removeDestination(MutableBrokerFilter.java:177)
> >         at
> >
> >
> org.apache.activemq.broker.region.RegionBroker.purgeInactiveDestinations(RegionBroker.java:860)
> >         at
> >
> org.apache.activemq.broker.region.RegionBroker$1.run(RegionBroker.java:109)
> >         at
> >
> >
> org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33)
> >         at java.util.TimerThread.mainLoop(Timer.java:555)
> >         at java.util.TimerThread.run(Timer.java:505)
> >
> >     Lockable ownable synchronizers:
> >          - <0x00000005cdb4ac00>
> >          - <0x00000005ce152b78>
> >
> >         5 waiting threads:
> >         --------
> >         java.lang.Thread.State: WAITING (parking)
> >             at sun.misc.Unsafe.park(Native Method)
> >             - parking to wait for  <0x00000005ce152b78> (a
> > java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync)
> >             at
> > java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireShared(AbstractQueuedSynchronizer.java:964)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1282)
> >             at
> >
> >
> java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:731)
> >             at
> >
> >
> org.apache.activemq.broker.region.RegionBroker.addConsumer(RegionBroker.java:402)
> >             at
> >
> >
> org.apache.activemq.broker.jmx.ManagedRegionBroker.addConsumer(ManagedRegionBroker.java:244)
> >             at
> > org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:97)
> >             at
> > org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:97)
> >             at
> >
> >
> org.apache.activemq.advisory.AdvisoryBroker.addConsumer(AdvisoryBroker.java:101)
> >             at
> > org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:97)
> >             at
> > org.apache.activemq.broker.BrokerFilter.addConsumer(BrokerFilter.java:97)
> >             at
> >
> >
> org.apache.activemq.broker.MutableBrokerFilter.addConsumer(MutableBrokerFilter.java:102)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.processAddConsumer(TransportConnection.java:618)
> >             at
> > org.apache.activemq.command.ConsumerInfo.visit(ConsumerInfo.java:349)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:294)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:148)
> >             at
> >
> >
> org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
> >             at
> >
> >
> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:113)
> >             at
> >
> >
> org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:270)
> >             at
> >
> >
> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
> >             at
> >
> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:214)
> >             at
> > org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
> >             at java.lang.Thread.run(Thread.java:745)
> >
> >
> >         1 waiting threads:
> >         --------
> >         java.lang.Thread.State: WAITING (parking)
> >             at sun.misc.Unsafe.park(Native Method)
> >             - parking to wait for  <0x00000005ce152b78> (a
> > java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync)
> >             at
> > java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireShared(AbstractQueuedSynchronizer.java:964)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1282)
> >             at
> >
> >
> java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:731)
> >             at
> >
> >
> org.apache.activemq.broker.region.RegionBroker.addProducer(RegionBroker.java:371)
> >             at
> >
> >
> org.apache.activemq.broker.jmx.ManagedRegionBroker.addProducer(ManagedRegionBroker.java:267)
> >             at
> >
> org.apache.activemq.broker.BrokerFilter.addProducer(BrokerFilter.java:102)
> >             at
> >
> org.apache.activemq.broker.BrokerFilter.addProducer(BrokerFilter.java:102)
> >             at
> >
> >
> org.apache.activemq.advisory.AdvisoryBroker.addProducer(AdvisoryBroker.java:172)
> >             at
> >
> >
> org.apache.activemq.broker.CompositeDestinationBroker.addProducer(CompositeDestinationBroker.java:56)
> >             at
> >
> org.apache.activemq.broker.BrokerFilter.addProducer(BrokerFilter.java:102)
> >             at
> >
> >
> org.apache.activemq.broker.MutableBrokerFilter.addProducer(MutableBrokerFilter.java:107)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.processAddProducer(TransportConnection.java:565)
> >             at
> > org.apache.activemq.command.ProducerInfo.visit(ProducerInfo.java:108)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:294)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:148)
> >             at
> >
> >
> org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
> >             at
> >
> >
> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:113)
> >             at
> >
> >
> org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:270)
> >             at
> >
> >
> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
> >             at
> >
> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:214)
> >             at
> > org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
> >             at java.lang.Thread.run(Thread.java:745)
> >
> >
> >         6 waiting threads:
> >         --------
> >         java.lang.Thread.State: WAITING (parking)
> >             at sun.misc.Unsafe.park(Native Method)
> >             - parking to wait for  <0x00000005ce152b78> (a
> > java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync)
> >             at
> > java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireShared(AbstractQueuedSynchronizer.java:964)
> >             at
> >
> >
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1282)
> >             at
> >
> >
> java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:731)
> >             at
> >
> >
> org.apache.activemq.broker.region.RegionBroker.removeConsumer(RegionBroker.java:413)
> >             at
> >
> >
> org.apache.activemq.broker.jmx.ManagedRegionBroker.removeConsumer(ManagedRegionBroker.java:262)
> >             at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:132)
> >             at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:132)
> >             at
> >
> >
> org.apache.activemq.advisory.AdvisoryBroker.removeConsumer(AdvisoryBroker.java:263)
> >             at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:132)
> >             at
> >
> >
> org.apache.activemq.broker.BrokerFilter.removeConsumer(BrokerFilter.java:132)
> >             at
> >
> >
> org.apache.activemq.broker.MutableBrokerFilter.removeConsumer(MutableBrokerFilter.java:137)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.processRemoveConsumer(TransportConnection.java:650)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.processRemoveSession(TransportConnection.java:689)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.processRemoveConnection(TransportConnection.java:801)
> >             - locked <0xXXXXXXXXXXXXXXXX> (a
> > org.apache.activemq.broker.jmx.ManagedTransportConnection)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection.doStop(TransportConnection.java:1138)
> >             at
> >
> >
> org.apache.activemq.broker.TransportConnection$4.run(TransportConnection.java:1068)
> >             at
> >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> >             at
> >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> >             at java.lang.Thread.run(Thread.java:745)
> >
> >
> >
> > --
> >
> > Founder/CEO Spinn3r.com
> > Location: *San Francisco, CA*
> > blog: http://burtonator.wordpress.com
> > … or check out my Google+ profile
> > <https://plus.google.com/102718274791889610666/posts>
> > <http://spinn3r.com>
> >
>



-- 

Founder/CEO Spinn3r.com
Location: *San Francisco, CA*
blog: http://burtonator.wordpress.com
… or check out my Google+ profile
<https://plus.google.com/102718274791889610666/posts>
<http://spinn3r.com>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message