[AMQCPP-556] setMessageListener on active session may leads to deadlock

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.8.3
• Fix Version/s: 3.8.4, 3.9.0
• Component/s: CMS Impl
• Patch Info: Patch Available
• Assignee: Timothy A. Bish
• Reporter: Jeremy Leung
• Created: 01/Dec/14 06:09
• Updated: 01/Dec/14 20:27
• Resolved: 01/Dec/14 20:27

Description

When we create a new MessageConsumer with a MessageListener from a session, the session will be stopped before the updates of the message listener and restarted afterward if the session already started.
However, if there have more than one consumers already started; calling ActiveMQSessionKernel::stop() may goes into a deadlock if there have many messages coming in.

Stack Trace when the sample program run into deadlock

26989: ./SimpleAsyncConsumer
----------------- lwp# 1 / thread# 1 --------------------
fdb29379 lwp_park (0, 0, 0)
fdb23b6b cond_wait_queue (8078ef0, 808bc28, 0) + 5e
fdb2403f _cond_wait (8078ef0, 808bc28) + 64
fdb24081 cond_wait (8078ef0, 808bc28) + 21
fdb240ba pthread_cond_wait (8078ef0, 808bc28) + 1b
fe7217b6 void decaf::internal::util::concurrent::PlatformThread::interruptibleWaitOnCondition(_pthread_cond*,_pthread_mutex*,decaf::internal::util::concurrent::CompletionCondition&) (8078ef0, 808bc28, 8046e14) + 36
fe71f400 bool decaf::internal::util::concurrent::Threading::join(decaf::internal::util::concurrent::ThreadHandle*,long long,int) (808f8d0, 0, 0, 0, fe758acd) + 160
fe758aeb void decaf::lang::Thread::join() (808ea18) + 2b
fe56b727 void activemq::threads::DedicatedTaskRunner::shutdown() (808f788) + f7
fe3f417a void activemq::core::ActiveMQSessionExecutor::stop() (808cb30) + 2da
fe48af48 void activemq::core::kernels::ActiveMQSessionKernel::stop() (808c240) + 18
fe44362a void activemq::core::kernels::ActiveMQConsumerKernel::setMessageListener(cms::MessageListener*) (80917d8, 8047384) + f2
fe3aa8c3 void activemq::core::ActiveMQConsumer::setMessageListener(cms::MessageListener*) (808f4e8, 8047384) + 73
08055170 void SimpleAsyncConsumer::runConsumer() (8047380) + 610
080530d3 main (1, 80474b8, 80474c0) + 2a3
08052d5a _start (1, 8047618, 0, 804762e, 804763a, 8047674) + 7a
----------------- lwp# 10 / thread# 10 --------------------
fdb29379 lwp_park (0, 0, 0)
fdb23b6b cond_wait_queue (808f188, 808bc48, 0) + 5e
fdb2403f _cond_wait (808f188, 808bc48) + 64
fdb24081 cond_wait (808f188, 808bc48) + 21
fdb240ba pthread_cond_wait (808f188, 808bc48) + 1b
fe721666 void decaf::internal::util::concurrent::PlatformThread::waitOnCondition(_pthread_cond*,_pthread_mutex*) (808f188, 808bc48) + 1e
fe71dab3 void __unnamed_syh4PZ6vdUWCx::doMonitorEnter(decaf::internal::util::concurrent::MonitorHandle*,decaf::internal::util::concurrent::ThreadHandle*) (806c740, 808f8d0) + 8f
fe71fd0b void decaf::internal::util::concurrent::Threading::enterMonitor(decaf::internal::util::concurrent::MonitorHandle*) (806c740) + 27
fe7aaa78 void decaf::util::concurrent::Mutex::lock() (808ec34) + 4c
fe7aa2d4 void decaf::util::concurrent::Lock::lock() (fd71fb84) + 2c
fe7aa023 decaf::util::concurrent::Lock::Lock #Nvariant 1(decaf::util::concurrent::Synchronizable*,const bool) (fd71fb84, 808ec34, 1) + 43
fe44b7a4 void activemq::core::kernels::ActiveMQConsumerKernel::dispatch(const decaf::lang::Pointer<activemq::commands::MessageDispatch,decaf::util::concurrent::atomic::AtomicRefCounter>&) (808ebb8, fd71feb0) + 49c
fe3f4371 void activemq::core::ActiveMQSessionExecutor::dispatch(const decaf::lang::Pointer<activemq::commands::MessageDispatch,decaf::util::concurrent::atomic::AtomicRefCounter>&) (808cb30, fd71feb0) + 11d
fe3f45be bool activemq::core::ActiveMQSessionExecutor::iterate() (808cb30) + a2
fe56b88a void activemq::threads::DedicatedTaskRunner::run() (808f788) + 8e
fe75889e void decaf::lang::Thread::run() (808ea18) + 1a
fe71cfbb void __unnamed_syh4PZ6vdUWCx::runCallback(void*) (808f8d0) + 2b
fe71cf52 void*__unnamed_syh4PZ6vdUWCx::threadEntryMethod(void*) (808f8d0) + ee
fdb28ffb _thr_setup (fd814200) + 4e
fdb29300 _lwp_start (fd814200, 0, 0, fd71fff8, fdb29300, fd814200)
----------------- lwp# 11 / thread# 11 --------------------
fdb29379 lwp_park (0, 0, 0)
fdb23b6b cond_wait_queue (808f5d8, 808bbc8, 0) + 5e
fdb2403f _cond_wait (808f5d8, 808bbc8) + 64
fdb24081 cond_wait (808f5d8, 808bbc8) + 21
fdb240ba pthread_cond_wait (808f5d8, 808bbc8) + 1b
fe721666 void decaf::internal::util::concurrent::PlatformThread::waitOnCondition(_pthread_cond*,_pthread_mutex*) (808f5d8, 808bbc8) + 1e
fe71dab3 void __unnamed_syh4PZ6vdUWCx::doMonitorEnter(decaf::internal::util::concurrent::MonitorHandle*,decaf::internal::util::concurrent::ThreadHandle*) (806c7a0, 8093c18) + 8f
fe71fd0b void decaf::internal::util::concurrent::Threading::enterMonitor(decaf::internal::util::concurrent::MonitorHandle*) (806c7a0) + 27
fe7aaa78 void decaf::util::concurrent::Mutex::lock() (80798b0) + 4c
fe7aa2d4 void decaf::util::concurrent::Lock::lock() (fd73fda4) + 2c
fe7aa023 decaf::util::concurrent::Lock::Lock #Nvariant 1(decaf::util::concurrent::Synchronizable*,const bool) (fd73fda4, 80798b0, 1) + 43
fe434d3d decaf::lang::Pointer<activemq::commands::MessageDispatch,decaf::util::concurrent::atomic::AtomicRefCounter>activemq::core::SimplePriorityMessageDispatchChannel::dequeueNoWait() (fd73fdf0, 80798a8) + 31
fe44e1cb bool activemq::core::kernels::ActiveMQConsumerKernel::iterate() (808ebb8) + bb
fe48de0e bool activemq::core::kernels::ActiveMQSessionKernel::iterateConsumers() (808c240) + 1b6
fe3f4545 bool activemq::core::ActiveMQSessionExecutor::iterate() (808cb30) + 29
fe56b88a void activemq::threads::DedicatedTaskRunner::run() (8091160) + 8e
fe75889e void decaf::lang::Thread::run() (8092a88) + 1a
fe71cfbb void __unnamed_syh4PZ6vdUWCx::runCallback(void*) (8093c18) + 2b
fe71cf52 void*__unnamed_syh4PZ6vdUWCx::threadEntryMethod(void*) (8093c18) + ee
fdb28ffb _thr_setup (fd813a00) + 4e
fdb29300 _lwp_start (fd813a00, 0, 0, fd73fff8, fdb29300, fd813a00)