[QPID-5793] [C++ broker] Memory leak _and_ deadlock due to qpid.auto_delete_timeout queues

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.26
• Component/s: C++ Broker
• Assignee: Gordon Sim
• Reporter: Pavel Moravec
• Created: 30/May/14 09:52
• Updated: 06/Sep/16 16:22
• Resolved: 03/Jun/14 19:20

Description

Description of problem:
There is a memory leak in the broker when a client declares auto-delete queue with qpid.auto_delete_timeout and re-connects just at the time the broker is about to delete the queue.

Version-Release number of selected component (if applicable):
any

How reproducible:
100% within some time

Steps to Reproduce:
1) Run script:

queues=100
sleep=0
while true; do
for i in $(seq 1 $queues); do
qpid-receive -a "autoDelQueue_${i}; {create:always, node:{ x-declare:{auto-delete:True, arguments:{'qpid.auto_delete_timeout':1}}, x-bindings:[{exchange:'amq.fanout', queue:autoDelQueue_${i}}] }}" --print-content=no -m 10 &
done
wait
sleep $sleep
date
done

Modify parameters (esp. sleep) such that the loop is executed once every second.

2) Monitor qpid broker memory utilization over time.

Actual results:
Both VSZ and RSS of qpidd process grows.

Expected results:
No memory grow.

Additional info:
the memory consumption increase is followed after a while by deadlock (see below) that - depending on number of worker threads - can practically freeze the broker. Backtraces of relevant threads:

Thread 4 (Thread 0x7fc08e803700 (LWP 26576)):
#0 0x00000032c4c0e264 in __lll_lock_wait () from /lib64/libpthread.so.0
#1 0x00000032c4c09523 in _L_lock_892 () from /lib64/libpthread.so.0
#2 0x00000032c4c09407 in pthread_mutex_lock () from /lib64/libpthread.so.0
#3 0x00007fc08f2e4842 in qpid::sys::Mutex::lock() ()
#4 0x00007fc08f2e4900 in qpid::sys::ScopedLock<qpid::sys::Mutex>::ScopedLock(qpid::sys::Mutex&) ()
#5 0x00007fc08f38bec5 in qpid::broker::Queue::canAutoDelete() const ()
#6 0x00007fc08f39d0bd in bool boost::_mfi::cmf0<bool, qpid::broker::Queue>::call<boost::shared_ptr<qpid::broker::Queue> const>(boost::shared_ptr<qpid::broker::Queue> const&, void const*) const ()
#7 0x00007fc08f39c2ef in bool boost::_mfi::cmf0<bool, qpid::broker::Queue>::operator()<boost::shared_ptr<qpid::broker::Queue> >(boost::shared_ptr<qpid::broker::Queue> const&) const ()
#8 0x00007fc08f39b117 in bool boost::_bi::list1<boost::_bi::value<boost::shared_ptr<qpid::broker::Queue> > >::operator()<bool, boost::_mfi::cmf0<bool, qpid::broker::Queue>, boost::_bi::list0>(boost::_bi::type<bool>, boost::_mfi::cmf0<bool, qpid::broker::Queue>&, boost::_bi::list0&, long) ()
#9 0x00007fc08f39974b in boost::_bi::bind_t<boost::_bi::unspecified, boost::_mfi::cmf0<bool, qpid::broker::Queue>, boost::_bi::list1<boost::_bi::value<boost::shared_ptr<qpid::broker::Queue> > > >::operator()() ()
#10 0x00007fc08f397593 in bool qpid::broker::QueueRegistry::destroyIf<boost::_bi::bind_t<boost::_bi::unspecified, boost::_mfi::cmf0<bool, qpid::broker::Queue>, boost::_bi::list1<boost::_bi::value<boost::shared_ptr<qpid::broker::Queue> > > > >(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, boost::_bi::bind_t<boost::_bi::unspecified, boost::_mfi::cmf0<bool, qpid::broker::Queue>, boost::_bi::list1<boost::_bi::value<boost::shared_ptr<qpid::broker::Queue> > > >) ()
#11 0x00007fc08f38ed3e in qpid::broker::Queue::tryAutoDelete() ()
#12 0x00007fc08f395a78 in qpid::broker::AutoDeleteTask::fire() ()
#13 0x00007fc08ecb9847 in qpid::sys::TimerTask::fireTask() ()
#14 0x00007fc08ecbaaa1 in qpid::sys::Timer::fire(boost::intrusive_ptr<qpid::sys::TimerTask>) ()
#15 0x00007fc08ecba3d5 in qpid::sys::Timer::run() ()
#16 0x00007fc08ec3b597 in qpid::sys::(anonymous namespace)::runRunnable(void*) ()
#17 0x00000032c4c079d1 in start_thread () from /lib64/libpthread.so.0
#18 0x00000032c44e8b6d in clone () from /lib64/libc.so.6

Thread 3 (Thread 0x7fc08cdcd700 (LWP 26577)):
#0 0x00000032c4c0e264 in __lll_lock_wait () from /lib64/libpthread.so.0
#1 0x00000032c4c09523 in _L_lock_892 () from /lib64/libpthread.so.0
#2 0x00000032c4c09407 in pthread_mutex_lock () from /lib64/libpthread.so.0
#3 0x00007fc08f2e4842 in qpid::sys::Mutex::lock() ()
#4 0x00007fc08f2e4900 in qpid::sys::ScopedLock<qpid::sys::Mutex>::ScopedLock(qpid::sys::Mutex&) ()
#5 0x00007fc08f38f3a3 in qpid::broker::Queue::hasExclusiveOwner() const ()
#6 0x00007fc08f346459 in qpid::broker::Broker::bind(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, qpid::framing::FieldTable const&, qpid::broker::OwnershipToken const*, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) ()
#7 0x00007fc08f462d55 in qpid::broker::SessionAdapter::ExchangeHandlerImpl::bind(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, qpid::framing::FieldTable const&) ()
#8 0x00007fc08ebd15d5 in void qpid::framing::ExchangeBindBody::invoke<qpid::framing::AMQP_ServerOperations::ExchangeHandler>(qpid::framing::AMQP_ServerOperations::ExchangeHandler&) const ()
#9 0x00007fc08ebcf1f5 in qpid::framing::AMQP_ServerOperations::ExchangeHandler::Invoker::visit(qpid::framing::ExchangeBindBody const&) ()
#10 0x00007fc08ebdcb7c in qpid::framing::ExchangeBindBody::accept(qpid::framing::MethodBodyConstVisitor&) const ()
#11 0x00007fc08ebcd5a2 in qpid::framing::AMQP_ServerOperations::Invoker::visit(qpid::framing::ExchangeBindBody const&) ()
#12 0x00007fc08ebdcb7c in qpid::framing::ExchangeBindBody::accept(qpid::framing::MethodBodyConstVisitor&) const ()
#13 0x00007fc08f475545 in qpid::framing::Invoker::Result qpid::framing::invoke<qpid::broker::SessionAdapter>(qpid::broker::SessionAdapter&, qpid::framing::AMQMethodBody const&) ()
#14 0x00007fc08f47152f in qpid::broker::SessionState::handleCommand(qpid::framing::AMQMethodBody*) ()
#15 0x00007fc08f4722ba in qpid::broker::SessionState::handleIn(qpid::framing::AMQFrame&) ()
#16 0x00007fc08f47a0a5 in qpid::framing::Handler<qpid::framing::AMQFrame&>::MemFunRef<qpid::framing::Handler<qpid::framing::AMQFrame&>::InOutHandlerInterface, &(qpid::framing::Handler<qpid::framing::AMQFrame&>::InOutHandlerInterface::handleIn(qpid::framing::AMQFrame&))>::handle(qpid::framing::AMQFrame&) () #17 0x00007fc08ec7b6c7 in qpid::amqp_0_10::SessionHandler::handleIn(qpid::framing::AMQFrame&) ()
#18 0x00007fc08f47a0a5 in qpid::framing::Handler<qpid::framing::AMQFrame&>::MemFunRef<qpid::framing::Handler<qpid::framing::AMQFrame&>::InOutHandlerInterface, &(qpid::framing::Handler<qpid::framing::AMQFrame&>::InOutHandlerInterface::handleIn(qpid::framing::AMQFrame&))>::handle(qpid::framing::AMQFrame&) () #19 0x00007fc08f3ce456 in qpid::framing::Handler<qpid::framing::AMQFrame&>::operator()(qpid::framing::AMQFrame&) ()
#20 0x00007fc08f3ca064 in qpid::broker::ConnectionHandler::handle(qpid::framing::AMQFrame&) ()
#21 0x00007fc08f3c0a2e in qpid::broker::amqp_0_10::Connection::received(qpid::framing::AMQFrame&) ()
#22 0x00007fc08f329780 in qpid::amqp_0_10::Connection::decode(char const*, unsigned long) ()
#23 0x00007fc08f43b015 in qpid::broker::SecureConnection::decode(char const*, unsigned long) ()
#24 0x00007fc08ecb031b in qpid::sys::AsynchIOHandler::readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*) ()
#25 0x00007fc08f4d8534 in boost::_mfi::mf2<void, qpid::sys::AsynchIOHandler, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>::operator()(qpid::sys::AsynchIOHandler*, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*) const ()
#26 0x00007fc08f4d7272 in void boost::_bi::list3<boost::_bi::value<qpid::sys::AsynchIOHandler*>, boost::arg<1>, boost::arg<2> >::operator()<boost::_mfi::mf2<void, qpid::sys::AsynchIOHandler, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>, boost::_bi::list2<qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*&> >(boost::_bi::type<void>, boost::_mfi::mf2<void, qpid::sys::AsynchIOHandler, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>&, boost::_bi::list2<qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*&>&, int) ()
#27 0x00007fc08f4d6704 in void boost::_bi::bind_t<void, boost::_mfi::mf2<void, qpid::sys::AsynchIOHandler, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>, boost::_bi::list3<boost::_bi::value<qpid::sys::AsynchIOHandler*>, boost::arg<1>, boost::arg<2> > >::operator--Type <return> to continue, or q <return> to quit--
()<qpid::sys::AsynchIO, qpid::sys::AsynchIOBufferBase*>(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*&) ()
#28 0x00007fc08f4d59dd in boost::detail::function::void_function_obj_invoker2<boost::_bi::bind_t<void, boost::_mfi::mf2<void, qpid::sys::AsynchIOHandler, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>, boost::_bi::list3<boost::_bi::value<qpid::sys::AsynchIOHandler*>, boost::arg<1>, boost::arg<2> > >, void, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>::invoke(boost::detail::function::function_buffer&, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*) ()
#29 0x00007fc08ec233dc in boost::function2<void, qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*>::operator()(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*) const ()
#30 0x00007fc08ec204cc in qpid::sys::posix::AsynchIO::readable(qpid::sys::DispatchHandle&) ()
#31 0x00007fc08ec283da in boost::_mfi::mf1<void, qpid::sys::posix::AsynchIO, qpid::sys::DispatchHandle&>::operator()(qpid::sys::posix::AsynchIO*, qpid::sys::DispatchHandle&) const ()
#32 0x00007fc08ec2772b in void boost::_bi::list2<boost::_bi::value<qpid::sys::posix::AsynchIO*>, boost::arg<1> >::operator()<boost::_mfi::mf1<void, qpid::sys::posix::AsynchIO, qpid::sys::DispatchHandle&>, boost::_bi::list1<qpid::sys::DispatchHandle&> >(boost::_bi::type<void>, boost::_mfi::mf1<void, qpid::sys::posix::AsynchIO, qpid::sys::DispatchHandle&>&, boost::_bi::list1<qpid::sys::DispatchHandle&>&, int) ()
#33 0x00007fc08ec26a1c in void boost::_bi::bind_t<void, boost::_mfi::mf1<void, qpid::sys::posix::AsynchIO, qpid::sys::DispatchHandle&>, boost::_bi::list2<boost::_bi::value<qpid::sys::posix::AsynchIO*>, boost::arg<1> > >::operator()<qpid::sys::DispatchHandle>(qpid::sys::DispatchHandle&) ()
#34 0x00007fc08ec259f9 in boost::detail::function::void_function_obj_invoker1<boost::_bi::bind_t<void, boost::_mfi::mf1<void, qpid::sys::posix::AsynchIO, qpid::sys::DispatchHandle&>, boost::_bi::list2<boost::_bi::value<qpid::sys::posix::AsynchIO*>, boost::arg<1> > >, void, qpid::sys::DispatchHandle&>::invoke(boost::detail::function::function_buffer&, qpid::sys::DispatchHandle&) ()
#35 0x00007fc08ecb4332 in boost::function1<void, qpid::sys::DispatchHandle&>::operator()(qpid::sys::DispatchHandle&) const ()
#36 0x00007fc08ecb38f8 in qpid::sys::DispatchHandle::processEvent(qpid::sys::Poller::EventType) ()
#37 0x00007fc08ec48692 in qpid::sys::Poller::Event::process() ()
#38 0x00007fc08ec47990 in qpid::sys::Poller::run() ()
#39 0x00007fc08ecb24d9 in qpid::sys::Dispatcher::run() ()
#40 0x00007fc08ec3b597 in qpid::sys::(anonymous namespace)::runRunnable(void*) ()
#41 0x00000032c4c079d1 in start_thread () from /lib64/libpthread.so.0
#42 0x00000032c44e8b6d in clone () from /lib64/libc.so.6

(I saw variants of Thread 3 waiting on the lock acquire, e.g. within:
#3 0x00007fc08ecb9c00 in qpid::sys::TimerTask::cancel() ()
#4 0x00007fc08f389cb8 in qpid::broker::Queue::consume(..)

)