You are viewing a plain text version of this content. The canonical link for it is here.

Posted to users@activemq.apache.org by Henrique Magarotto <ma...@gmail.com> on 2011/03/02 00:22:56 UTC

AMQCPP: unexpected exception

Hi everybody,

First, sorry for long mail and thanks for great job in ActiveMQ and
ActiveMQ-CPP.

I'm testing some failure cases with ActiveMQ and ActiveMQ-CPP.

In my test, an unexpected exception is thrown if broker is stopped
while consumer is in transaction.

Environment:
ActiveMQ 5.4.2
ActiveMQ-CPP 3.2.4
Ubuntu 10.04
JRE 1.6.0_20-b02


I get the following message:

terminate called after throwing an instance of 'cms::CMSException'
  what():  Failover timeout of 1000 ms reached.
Aborted (core dumped)

core backtrace:
#0  0x001ef422 in __kernel_vsyscall ()
#1  0x00c5a651 in *__GI_raise (sig=6) at
../nptl/sysdeps/unix/sysv/linux/raise.c:64
#2  0x00c5da82 in *__GI_abort () at abort.c:92
#3  0x00bf952f in __gnu_cxx::__verbose_terminate_handler() () from
/usr/lib/libstdc++.so.6
#4  0x00bf7465 in ?? () from /usr/lib/libstdc++.so.6
#5  0x00bf74a2 in std::terminate() () from /usr/lib/libstdc++.so.6
#6  0x00bf74c5 in ?? () from /usr/lib/libstdc++.so.6
#7  0x00bf6915 in __cxa_call_unexpected () from /usr/lib/libstdc++.so.6
#8  0x0052f8ae in
activemq::core::TransactionSynhcronization::beforeEnd (this=0x93ac548)
at activemq/core/ActiveMQConsumer.cpp:84
#9  0x00550588 in
activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x93b3e48)
at activemq/core/ActiveMQTransactionContext.cpp:192
#10 0x00550d63 in activemq::core::ActiveMQTransactionContext::commit
(this=0x93b3e48) at activemq/core/ActiveMQTransactionContext.cpp:127
#11 0x0053e460 in activemq::core::ActiveMQSession::commit
(this=0x93b3c10) at activemq/core/ActiveMQSession.cpp:189
#12 0x0042c1a0 in activemq::cmsutil::PooledSession::commit
(this=0x93b3fd8) at activemq/cmsutil/PooledSession.h:87
#13 0x0804c0fb in Consumer::onMessage (this=0x93a8c30,
message=0x93b49b0) at main.cpp:455
#14 0x0804bdc8 in Consumer::consumeLoop (this=0x93a8c30) at main.cpp:415
#15 0x0804bb56 in Consumer::run (this=0x93a8c30) at main.cpp:381
#16 0x00823cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x93abff0) at decaf/lang/Thread.cpp:135
#17 0x00822847 in threadWorker (arg=0x93abff0) at decaf/lang/Thread.cpp:188
#18 0x0015a96e in start_thread (arg=0xb6f8bb70) at pthread_create.c:300
#19 0x00cfda4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130


Exception specification for method
'activemq::core::TransactionSynhcronization::beforeEnd' says: 'throw(
exceptions::ActiveMQException )' but 'ActiveMQConsumer::acknowledge'
throws 'cms::CMSException', so 'std::unexpected' is called and 'abort'
occurs. (http://www.linuxprogrammingblog.com/cpp-exception-specifications-are-evil)


When 'MessageListener' is used, the commit freezes, probability, in thread 5:

Thread 1:
#0  0x00cfd422 in __kernel_vsyscall ()
#1  0x00ddcb5d in pthread_join (threadid=3061869424,
thread_return=0xbf964bdc) at pthread_join.c:89
#2  0x00a923c0 in decaf::lang::Thread::join (this=0x8eba574) at
decaf/lang/Thread.cpp:421
#3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf964c64) at main.cpp:586
#4  0x0804c678 in AppTest::run (this=0xbf964c64) at main.cpp:555
#5  0x0804a33c in main (argc=4, argv=0xbf964d84) at main.cpp:633

Thread 2:
#0  0x00cfd422 in __kernel_vsyscall ()
#1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
#2  0x0034f9dd in __pthread_cond_wait (cond=0x8eb0e38,
mutex=0x8eb0e08) at forward.c:139
#3  0x00a606a9 in
decaf::internal::util::concurrent::ConditionImpl::wait
(condition=0x8eb0e38) at
decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
#4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
(this=0x8eb0d6c) at decaf/util/concurrent/Mutex.cpp:95
#5  0x007fc9c5 in activemq::threads::CompositeTaskRunner::run
(this=0x8eb0d48) at activemq/threads/CompositeTaskRunner.cpp:118
#6  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x8eb0e90) at decaf/lang/Thread.cpp:135
#7  0x00a92847 in threadWorker (arg=0x8eb0e90) at decaf/lang/Thread.cpp:188
#8  0x00ddb96e in start_thread (arg=0xb7808b70) at pthread_create.c:300
#9  0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 3:
#0  0x00cfd422 in __kernel_vsyscall ()
#1  0x00de2af9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8ebb3a8) at
pthread_mutex_lock.c:61
#4  0x0034fba6 in pthread_mutex_lock (mutex=0x8ebb3a8) at forward.c:182
#5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x8ebb3a8) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
(this=0x8ebb2bc) at decaf/util/concurrent/Mutex.cpp:75
#7  0x007c5770 in
decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
(this=0x8ebb2a8)
    at ./decaf/util/StlQueue.h:253
#8  activemq::core::MessageDispatchChannel::lock (this=0x8ebb2a8) at
activemq/core/MessageDispatchChannel.h:153
#9  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
(this=0xb7006fe4) at decaf/util/concurrent/Lock.cpp:54
#10 0x00ad0c08 in Lock (this=0xfffffe00, object=0x8ebb3a8,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#11 0x0078e102 in
activemq::core::ActiveMQConsumer::clearMessagesInProgress
(this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:1112
#12 0x007af15c in
activemq::core::ActiveMQSession::clearMessagesInProgress
(this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:239
#13 0x00778543 in
activemq::core::ActiveMQConnection::transportInterrupted
(this=0x8eb1210) at activemq/core/ActiveMQConnection.cpp:704
#14 0x00803b34 in
activemq::transport::TransportFilter::transportInterrupted
(this=0x8eb11c0) at activemq/transport/TransportFilter.cpp:67
#15 0x008187ae in
activemq::transport::failover::FailoverTransport::handleTransportFailure
(this=0x8eb0a38, error=...) at
activemq/transport/failover/FailoverTransport.cpp:476
#16 0x0082406c in
activemq::transport::failover::FailoverTransportListener::onException
(this=0x8eb0878, ex=...) at
activemq/transport/failover/FailoverTransportListener.cpp:97
#17 0x00803c0b in activemq::transport::TransportFilter::fire
(this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:49
#18 0x00803c64 in activemq::transport::TransportFilter::onException
(this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:41
#19 0x00803c0b in activemq::transport::TransportFilter::fire
(this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:49
#20 0x00803c64 in activemq::transport::TransportFilter::onException
(this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:41
#21 0x00801b13 in activemq::transport::IOTransport::fire
(this=0x8eb27a0, ex=...) at activemq/transport/IOTransport.cpp:73
#22 0x008023bf in activemq::transport::IOTransport::run
(this=0x8eb27a0) at activemq/transport/IOTransport.cpp:246
#23 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x8eb2ea0) at decaf/lang/Thread.cpp:135
#24 0x00a92847 in threadWorker (arg=0x8eb2ea0) at decaf/lang/Thread.cpp:188
#25 0x00ddb96e in start_thread (arg=0xb7007b70) at pthread_create.c:300
#26 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 4:
#0  0x00cfd422 in __kernel_vsyscall ()
#1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
#2  0x0034f9dd in __pthread_cond_wait (cond=0x8eba6c8,
mutex=0x8eba698) at forward.c:139
#3  0x00a606a9 in
decaf::internal::util::concurrent::ConditionImpl::wait
(condition=0x8eba6c8) at
decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
#4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
(this=0x8eba584) at decaf/util/concurrent/Mutex.cpp:95
#5  0x00ad029a in decaf::util::concurrent::CountDownLatch::await
(this=0x8eba57c) at decaf/util/concurrent/CountDownLatch.cpp:56
#6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x8eba574,
timeOut=-1) at main.cpp:226
#7  0x0804bb46 in Consumer::run (this=0x8eba570) at main.cpp:379
#8  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x8eba5b0) at decaf/lang/Thread.cpp:135
#9  0x00a92847 in threadWorker (arg=0x8eba5b0) at decaf/lang/Thread.cpp:188
#10 0x00ddb96e in start_thread (arg=0xb6806b70) at pthread_create.c:300
#11 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 5:
#0  0x00cfd422 in __kernel_vsyscall ()
#1  0x00de2af9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8eb05e0) at
pthread_mutex_lock.c:61
#4  0x0034fba6 in pthread_mutex_lock (mutex=0x8eb05e0) at forward.c:182
#5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x8eb05e0) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
(this=0x8eb0a88) at decaf/util/concurrent/Mutex.cpp:75
#7  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
(this=0xb6004ce0) at decaf/util/concurrent/Lock.cpp:54
#8  0x00ad0c08 in Lock (this=0xfffffe00, object=0x8eb05e0,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#9  0x0081b8fd in
activemq::transport::failover::FailoverTransport::oneway
(this=0x8eb0a38, command=...) at
activemq/transport/failover/FailoverTransport.cpp:186
#10 0x00807f9f in
activemq::transport::correlator::ResponseCorrelator::oneway
(this=0x8eb11c0, command=...) at
activemq/transport/correlator/ResponseCorrelator.cpp:82
#11 0x007713be in activemq::core::ActiveMQConnection::oneway
(this=0x8eb1210, command=...) at
activemq/core/ActiveMQConnection.cpp:741
#12 0x007b0a4f in activemq::core::ActiveMQSession::oneway
(this=0x8ebab90, command=...) at activemq/core/ActiveMQSession.cpp:903
#13 0x00795a1c in activemq::core::ActiveMQConsumer::acknowledge
(this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:860
#14 0x0079f885 in
activemq::core::TransactionSynhcronization::beforeEnd (this=0x8ebb9e8)
at activemq/core/ActiveMQConsumer.cpp:85
#15 0x007c0588 in
activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x8ebadc8)
at activemq/core/ActiveMQTransactionContext.cpp:192
#16 0x007c0d63 in activemq::core::ActiveMQTransactionContext::commit
(this=0x8ebadc8) at activemq/core/ActiveMQTransactionContext.cpp:127
#17 0x007ae460 in activemq::core::ActiveMQSession::commit
(this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:189
#18 0x0069c1a0 in activemq::cmsutil::PooledSession::commit
(this=0x8ebaf38) at activemq/cmsutil/PooledSession.h:87
#19 0x0804c0fb in Consumer::onMessage (this=0x8eba570,
message=0x8ebbec0) at main.cpp:455
#20 0x0079972f in activemq::core::ActiveMQConsumer::dispatch
(this=0x8ebb270, dispatch=...) at
activemq/core/ActiveMQConsumer.cpp:1018
#21 0x007bd5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
(this=0x8ebae78, dispatch=...) at
activemq/core/ActiveMQSessionExecutor.cpp:129
#22 0x007bd993 in activemq::core::ActiveMQSessionExecutor::iterate
(this=0x8ebae78) at activemq/core/ActiveMQSessionExecutor.cpp:166
#23 0x008007f3 in activemq::threads::DedicatedTaskRunner::run
(this=0x8ebb550) at activemq/threads/DedicatedTaskRunner.cpp:111
#24 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x8ebaad8) at decaf/lang/Thread.cpp:135
#25 0x00a92847 in threadWorker (arg=0x8ebaad8) at decaf/lang/Thread.cpp:188
#26 0x00ddb96e in start_thread (arg=0xb6005b70) at pthread_create.c:300
#27 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130



I tried to resolve this issue changing configurations and code, but i
always get an error or client freezes.
Is it a bug? Any suggestions?

Thanks.
Henrique

How to reproduce:
1. Start ActiveMQ
2. Send a massage './activemqTest -p1 -n1'
3. Run consumer './activemqTest -c1 -d10000'
4. Stop activemq when 'Starting delay...' message appears.
5. After activemq is down and delay is finish, consumer try commit
(message 'Try commit') and unexpected exception occurs.

Following test code, based on ActiveMQ-CPP sample
'activemq-cpp-library-3.2.4/src/examples/main.cpp':


/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

// START SNIPPET: demo

#include <activemq/library/ActiveMQCPP.h>
#include <activemq/cmsutil/SessionPool.h>
#include <activemq/cmsutil/PooledSession.h>
#include <activemq/cmsutil/ResourceLifecycleManager.h>
#include <activemq/exceptions/ActiveMQException.h>
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/util/concurrent/Mutex.h>
#include <decaf/lang/Integer.h>
#include <decaf/lang/Long.h>
#include <decaf/lang/System.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/util/Config.h>
#include <cms/Connection.h>
#include <cms/Session.h>
#include <cms/TextMessage.h>
#include <cms/BytesMessage.h>
#include <cms/MapMessage.h>
#include <cms/ExceptionListener.h>
#include <cms/MessageListener.h>
#include <iostream>
#include <memory>
#include <map>
#include <vector>
#include <unistd.h>

using namespace activemq::core;
using namespace activemq::cmsutil;
using namespace decaf::util::concurrent;
using namespace decaf::util;
using namespace decaf::lang;
using namespace cms;


class ConnPoolManager : protected decaf::util::concurrent::Mutex,
                        public ExceptionListener {
private:
    typedef std::map< int, activemq::cmsutil::SessionPool* > tSessionPoolMap;

    std::string _brokerURI;
    ConnectionFactory* _connectionFactory;
    Connection* _connection;
    ResourceLifecycleManager* _lifecycleManager;
    tSessionPoolMap _pools;
    bool _started;

public:
    ConnPoolManager() :
            _brokerURI("tcp://localhost:61616"),
            _connectionFactory(NULL),
            _connection(NULL),
            _lifecycleManager(NULL),
            _started(false)
    {
    }

    virtual ~ConnPoolManager() {
        cleanup();
    }

    void setBrokerURI(const std::string& uri) {
        _brokerURI = uri;
    }

    bool startPools() {
        synchronized(this) {
            if(_started) return true;
            try {
                createConnection();
                createPools();
                _started=true;
                return true;
            } catch(CMSException& e) {
                std::cout<<"(ConnPoolManager::startPools)
CMSException: " << e.getMessage() << std::endl;
            } catch(...) {
                std::cout<<"(ConnPoolManager::startPools) Unknown
exception!!!" << std::endl;
            }
        }
        cleanup();
        return false;
    }

    PooledSession* getSession(cms::Session::AcknowledgeMode ack) {
        if(_started) {
            return _pools[ack]->takeSession();
        } else if(startPools()) {
            return _pools[ack]->takeSession();
        } else {
            return NULL;
        }
    }

private:

    void createConnection(){
        // Create a ConnectionFactory
        _connectionFactory =
ConnectionFactory::createCMSConnectionFactory( _brokerURI );

        // Create a Connection
        _connection = _connectionFactory->createConnection();
        _connection->start();
        _connection->setExceptionListener(this);
    }

    void createPools(){
        // Create lifecycle manager
        _lifecycleManager = new activemq::cmsutil::ResourceLifecycleManager();

        // Create pools, one for each acknowledge type
        for(    int ack = cms::Session::AUTO_ACKNOWLEDGE;
                ack <= cms::Session::INDIVIDUAL_ACKNOWLEDGE;
                ack++)
        {
            _pools[ack] =  new activemq::cmsutil::SessionPool(
                    _connection,
                    static_cast<cms::Session::AcknowledgeMode>(ack),
                    _lifecycleManager);
        }
    }

    void cleanup() {
        synchronized(this) {
            releasePools();
            if(_lifecycleManager!=NULL) {
                try { delete _lifecycleManager;
                } catch(...) {}
            } _lifecycleManager = NULL;

            if(_connection!=NULL) {
                try { delete _connection;
                } catch(...) {}
            } _connection = NULL;
            if(_connectionFactory!=NULL) {
                try { delete _connectionFactory;
                } catch(...) {}
            } _connectionFactory = NULL;
            _started=false;
        }
    }

    void releasePools() {
        for(tSessionPoolMap::iterator it=_pools.begin();
it!=_pools.end(); it++) {
            try {
                delete it->second;
            } catch(...) { }
        }
        _pools.clear();
    }

    // If something bad happens you see it here as this class is also been
    // registered as an ExceptionListener with the connection.
    virtual void onException( const CMSException& ex AMQCPP_UNUSED) {
        std::cout<<"(ConnPoolManager::onException) CMS Exception
occurred.  Shutting down client."<<std::endl;
        ex.printStackTrace();
        exit(1);
    }



};

class JMSEndPointThread  : public Thread {
private:
    CountDownLatch _stopNow;
    ConnPoolManager* _pools;
    std::string _destinationString;
    PooledSession* _session;
    Destination* _destination;
    bool _sessionTransacted;

protected:

    JMSEndPointThread(  ConnPoolManager* pools,
                        const std::string& dest,
                        bool sessionTransacted) :
            _stopNow(1),
            _pools(pools),
            _destinationString(dest),
            _session(NULL),
            _destination(NULL),
            _sessionTransacted(sessionTransacted)
    {
    }

    virtual bool prepareSession() {
        cleanup();
        // Create a Session
        if( _sessionTransacted ) {
            _session = _pools->getSession(Session::SESSION_TRANSACTED);
        } else {
            _session = _pools->getSession(Session::AUTO_ACKNOWLEDGE);
        }
        if(_session==NULL) return false;

        // Create the Queue destination
        _destination = _session->createQueue( _destinationString );
        return true;
    }

    PooledSession* getSession() { return _session; }
    Destination* getDestination() { return _destination; }
    bool getTransacted() { return _sessionTransacted; }

    bool stopCheck(int timeOut=0) {
        if(timeOut>=0) return _stopNow.await(timeOut);
        else _stopNow.await();
        return true;
    }

    virtual void cleanup() {
        // Destroy resources.
        try{
            if( _destination != NULL ) delete _destination;
        } catch(...) { }
        _destination = NULL;

        // Back session to pool
        try{
            if( _session != NULL ) _session->close();
        } catch(...) { }
        _session = NULL;
    }

public:

    typedef std::vector<JMSEndPointThread*> List;

    virtual ~JMSEndPointThread() {
        cleanup();
    }

    void stopNow() {
        _stopNow.countDown();
    }

};

class Producer : public JMSEndPointThread {
private:

    MessageProducer* _producer;
    int _delay;
    int _numMessages;

public:

    Producer(   ConnPoolManager* pools,
                const std::string& dest,
                int delay,
                int numMessages,
                bool sessionTransacted = false ) :
        JMSEndPointThread(pools,dest,sessionTransacted)
    {
        _producer = NULL;
        _delay = delay;
        _numMessages = numMessages;
    }

    virtual ~Producer() {
        cleanup();
    }

    virtual void run() {
        std::cout<<"Producer started!!!! Thread: "<<Thread::getId()<<std::endl;
        while(!stopCheck() && _numMessages) {
            try {
                if(prepareProducer()) {
                    sendMessages();
                    continue;
                }
            } catch(CMSException& e) {
                std::cout<<"(Producer::run) CMSException: " <<
e.getMessage() << std::endl;
            } catch(...) {
                std::cout<<"(Producer::run) Unknown exception!!!" << std::endl;
            }
            cleanup();
            stopCheck(5000); // reconnect delay
        }
        std::cout<<"Producer end!!!! Thread: "<<Thread::getId()<<std::endl;
        cleanup();
    }

private:

    bool prepareProducer() {
        // Get Session
        if(!prepareSession()) return false;
        // Create a MessageProducer from the Session to the Queue
        _producer = getSession()->createProducer( getDestination() );
        _producer->setDeliveryMode( DeliveryMode::PERSISTENT );
        return true;
    }
    void sendMessages() {
        // Create a messages
        std::stringstream text;
        text<<"Hello world! from thread " << Thread::getId();

        while(!stopCheck() && _numMessages) {

            TextMessage* message = getSession()->createTextMessage(
text.str() );
            message->setIntProperty( "Integer", _numMessages );

            // Tell the producer to send the message
            std::cout<<"Sent message #"<<_numMessages<<" from thread
"<<Thread::getId()<<std::endl;
            _producer->send( message );
            delete message;
            --_numMessages;

            stopCheck(_delay);
            if(getTransacted()) getSession()->commit();
        }
    }

    virtual void cleanup() {
        try{
            if( _producer != NULL ) delete _producer;
        } catch(...) { }
        _producer = NULL;

        JMSEndPointThread::cleanup();
    }

};


class Consumer :    public MessageListener,
                    public JMSEndPointThread {

private:

    MessageConsumer* _consumer;
    int _delay;
    bool _useListener;

public:

    Consumer(   ConnPoolManager* pools,
                const std::string& dest,
                int delay,
                bool useListener = true,
                bool sessionTransacted = true ) :
        JMSEndPointThread(pools,dest,sessionTransacted)
    {
        _consumer = NULL;
        _delay = delay;
        _useListener = useListener;
    }
    virtual ~Consumer(){
        cleanup();
    }

    virtual void run() {
        std::cout<<"Consumer started!!!! Thread: "<<Thread::getId()<<std::endl;
        while(!stopCheck()) {
            try {
                if(prepareConsumer()) {
                    // Wait while asynchronous messages come in.
                    if(_useListener) {
                        stopCheck(-1);
                    } else {
                        consumeLoop();
                    }
                    continue;
                }
            } catch (CMSException& e) {
                std::cout<<"(Consumer::run) CMSException: " <<
e.getMessage() << std::endl;
            } catch(...) {
                std::cout<<"(Consumer::run) Unknown exception!!!" << std::endl;
            }
            cleanup();
            stopCheck(5000); // reconnect delay
        }
        std::cout<<"Consumer end!!!! Thread: "<<Thread::getId()<<std::endl;
        cleanup();
    }


private:

    bool prepareConsumer() {
        // Get Session
        if(!prepareSession()) return false;
        // Create a MessageConsumer from the Session to the Queue
        _consumer = getSession()->createConsumer( getDestination() );
        if(_useListener)
            _consumer->setMessageListener( this );

        return true;
    }

    void consumeLoop() {
        while(!stopCheck()) {
            Message *msg = _consumer->receive(500);
            if(msg!=NULL) {
                onMessage(msg);
                delete msg;
            }
        }
    }

    // Called from the consumer since this class is a registered
MessageListener.
    virtual void onMessage( const Message* message ) {

        static int count = 0;

        try
        {
            count++;
            const TextMessage* textMessage =
                dynamic_cast< const TextMessage* >( message );
            std::string text = "";

            if( textMessage != NULL ) {
                text = textMessage->getText();
            } else {
                text = "NOT A TEXTMESSAGE!";
            }

            std::cout<<"Message #"<<count<<" Received: "<<text<<std::endl;
            if(_delay) {
                std::cout<<"Starting delay..."<<std::endl;
                stopCheck(_delay);
            }

        } catch (CMSException& e) {
            std::cout<<"(Consumer::onMessage) CMSException: " <<
e.getMessage() << std::endl;
        } catch(...) {
            std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
std::endl;
        }

        try
        {   // Commit all messages.
            if( getTransacted() ) {
                std::cout<<"Try commit....."<<std::endl;
                getSession()->commit();
                std::cout<<"Commit OK!!!"<<std::endl;
            }
        } catch (CMSException& e) {
            std::cout<<"(Consumer::onMessage) CMSException: " <<
e.getMessage() << std::endl;
        } catch(...) {
            std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
std::endl;
        }

    }

    virtual void cleanup() {
        try{
            if( _consumer != NULL ) delete _consumer;
        } catch (...) { }
        _consumer = NULL;
        JMSEndPointThread::cleanup();
    }
};


class ScopedActiveMQLibrary {
public:
    ScopedActiveMQLibrary() {
        activemq::library::ActiveMQCPP::initializeLibrary();
    }
    virtual ~ScopedActiveMQLibrary() {
        activemq::library::ActiveMQCPP::shutdownLibrary();
    }
};

class AppTest : private ScopedActiveMQLibrary {
private:
    int _consumer;
    int _producer;
    int _numMessages;
    bool _useListener;
    int _delay;
    ConnPoolManager _poolManager;
    JMSEndPointThread::List _endPointList;

    static AppTest* _app;
    static CountDownLatch _terminationRequest;

public:
    AppTest(int argc, char** argv) :
            _consumer(0),
            _producer(0),
            _numMessages(1),
            _useListener(false),
            _delay(0)
    {
        _app=this;
        int c;
        while ((c = getopt (argc, argv, "lc:p:n:d:")) != -1) {
            switch(c) {
            case 'l': _useListener=true; break;
            case 'c': _consumer=atoi(optarg); break;
            case 'p': _producer=atoi(optarg); break;
            case 'n': _numMessages=atoi(optarg); break;
            case 'd': _delay=atoi(optarg); break;
            }
        }
        if(_consumer<0) _consumer=0;
        if(_producer<0) _producer=0;
        _poolManager.setBrokerURI(
                "failover://("
                "tcp://localhost:61616"
                "?transport.useInactivityMonitor=false"
                ")?timeout=1000"
                "&cms.RedeliveryPolicy.maximumRedeliveries=-1"
                );
        _poolManager.startPools();
    }

    virtual ~AppTest() {
    }

    int run() {

        // Install signal handler
        if(!installSigAction())
            return 1;

        // Create consumer/producer objects
        loadEndPoins();

        // Start the producer/consumer thread.
        startEndPoins();

        // start another tasks
        // ...

        // Wait for termination request
        /*if(_consumer) {
            _terminationRequest.await();
            stopEndPoins();
        }*/

        // join threads
        joinEndPoins();

        return 0;
    }

private:

    void loadEndPoins() {
        for(int i=0; i<_consumer;i++) {
            _endPointList.push_back( new
Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
        }
        for(int i=0; i<_producer;i++) {
            _endPointList.push_back( new
Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
        }
    }

    void startEndPoins() {
        for(JMSEndPointThread::List::iterator it=_endPointList.begin();
            it!=_endPointList.end(); it++ )
            (**it).start();
    }

    void stopEndPoins() {
        for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
            it!=_endPointList.rend(); it++ )
            (**it).stopNow();
    }

    void joinEndPoins() {
        for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
            it!=_endPointList.rend(); it++ ) {
            (**it).join();
            delete *it;
            *it = NULL;
        }
        _endPointList.clear();
    }

    static void signalHandler(int sig) {
        _app->_terminationRequest.countDown();
        _app->stopEndPoins();
    }

    bool installSigAction() {
        struct sigaction action;
        memset(&action, 0, sizeof(action));
        action.sa_handler = signalHandler;
        if( sigaction(SIGTERM, &action, NULL)<0 ||
            sigaction(SIGQUIT, &action, NULL)<0 ||
            sigaction(SIGINT,  &action, NULL)<0 )
        { return false; }
        return true;
    }

    int termWait() {
        sigset_t sset;
        sigemptyset(&sset);
        sigaddset(&sset, SIGINT);
        sigaddset(&sset, SIGQUIT);
        sigaddset(&sset, SIGTERM);
        sigprocmask(SIG_BLOCK, &sset, NULL);
        int sig;
        sigwait(&sset, &sig);
        return sig;
    }

};
AppTest* AppTest::_app = NULL;
CountDownLatch AppTest::_terminationRequest(1);


void my_unexpected() {
    throw activemq::exceptions::ActiveMQException();
}

int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
    //std::set_unexpected(my_unexpected);
    AppTest test(argc, argv);
    return test.run();
}

// END SNIPPET: demo

Re: AMQCPP: unexpected exception

Posted by Oscar Pernas <os...@pernas.es>.

Hi all,

Also, if you run a blocking producer (with connection->start() blocking) to
a not existing broker, and try to close the producer after a while, the
connection->start crash.

2011-03-07 02:27:33 DEBUG (ActiveInterface:671) - ActiveInterface::shutdown.
Shutting down singleton manager.
2011-03-07 02:27:33 INFO  (ActiveProducer:901) - Producer::close. Closing
producer failover://(tcp://localhost:61616?connection.useAsyncSend=true)
alarmas...
2011-03-07 02:27:33 INFO  (ActiveProducer:915) - Producer::deleting
destination...
2011-03-07 02:27:33 INFO  (ActiveProducer:927) - Producer::deleting
producer...
2011-03-07 02:27:33 INFO  (ActiveProducer:940) - Producer::closing
session...
2011-03-07 02:27:33 INFO  (ActiveProducer:944) - Producer::closing
connection...
2011-03-07 02:27:48 INFO  (ActiveProducer:948) - Producer::closing
connection... Producer::cleaning up CMSException: No valid response received
for command: RemoveInfo { commandId = 2, responseRequired = true, ObjectId =
ID:artemisa-47349-1299461248159-0:0, LastDeliveredSequenceId = 0 }, check
broker.
2011-03-07 02:27:48 INFO  (ActiveProducer:954) - Producer::deleting temporal
destination...
2011-03-07 02:27:48 INFO  (ActiveProducer:965) - Producer::deleting
session...
2011-03-07 02:27:48 INFO  (ActiveProducer:976) - Producer::deleting
connection...
pure virtual method called
terminate called without an active exception






2011/3/7 Oscar Pernas <os...@pernas.es>

> Yes, It's probably the same issue. I've resolved the destructor stucks
> (partially) deleting connection and session before delete the consumer.
> Perhaps you should catch a decaf exception that sometimes is thrown.
>
> Tell me if it helps you.
>
>
> regards
>
> 2011/3/4 Henrique Magarotto <ma...@gmail.com>
>
> Hi Tim and Oscar,
>> thanks for your attention,
>>
>> So, in my case i got frozen in a commit attempt.
>> But really, testing without transaction and MessageListener i got
>> frozen in consumer destructor too.
>> Maybe it's the same reason for both cases. I tested with several
>> configurations to try avoid freezes for both  cases.
>> I opened the AMQCPP-354 issue with 'unexpected exception' report, but
>> i think that these cases deserve more one (or two) new issues in Jira
>> for later investigation.
>>
>> regards
>> Henrique
>>
>> Following how to reproduce the cases:
>>
>> MESSAGELISTENER WITH TRANSACTION, FREEZES IN COMMIT
>> MessageListener consumer destructor freezes in commit:
>> 1- Start ActiveMQ
>> 2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
>> 3- Send one message './activemqTest -p1'
>> 4- Wait consumer 'Starting delay'
>> 5- Stop ActiveMQ
>> 6- Consumer will try commit
>> 8- Freezes (see below 'BACKTRACE 1')!!!
>>
>>
>> MESSAGELISTENER WITHOUT TRANSACTION, FREEZES IN DESTRUCTOR
>> MessageListener consumer destructor freezes (apply changes in test code):
>> 1- Start ActiveMQ
>> 2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
>> 3- Send one message './activemqTest -p1'
>> 4- Wait consumer 'Starting delay'
>> 5- Stop ActiveMQ
>> 6- Wait consumer delay finishes 'Delay finish'
>> 7- Try kill consumer ( not kill -9 :-) )
>> 8- Freezes (see below 'BACKTRACE 2')!!!
>>
>> Changes in test:
>> @@ -440,6 +440,7 @@
>>              if(_delay) {
>>                  std::cout<<"Starting delay..."<<std::endl;
>>                  stopCheck(_delay);
>> +                std::cout<<"Delay finish..."<<std::endl;
>>              }
>>
>>          } catch (CMSException& e) {
>> @@ -561,7 +562,7 @@
>>
>>      void loadEndPoins() {
>>          for(int i=0; i<_consumer;i++) {
>> -            _endPointList.push_back( new
>> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
>> +            _endPointList.push_back( new
>> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener,false) );
>>          }
>>          for(int i=0; i<_producer;i++) {
>>              _endPointList.push_back( new
>> Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
>>
>>
>> BACKTRACE 1:
>>
>> Thread 1:
>> #0  0x00eed422 in __kernel_vsyscall ()
>> #1  0x00468b5d in pthread_join (threadid=3061414768,
>> thread_return=0xbf951e8c) at pthread_join.c:89
>> #2  0x00be23c0 in decaf::lang::Thread::join (this=0x9551574) at
>> decaf/lang/Thread.cpp:421
>> #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf951f14) at main.cpp:586
>> #4  0x0804c678 in AppTest::run (this=0xbf951f14) at main.cpp:555
>> #5  0x0804a33c in main (argc=4, argv=0xbf952034) at main.cpp:633
>>
>> Thread 2:
>> #0  0x00eed422 in __kernel_vsyscall ()
>> #1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
>> #2  0x003259dd in __pthread_cond_wait (cond=0x9547e38,
>> mutex=0x9547e08) at forward.c:139
>> #3  0x00bb06a9 in
>> decaf::internal::util::concurrent::ConditionImpl::wait
>> (condition=0x9547e38) at
>> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
>> #4  0x00c21353 in decaf::util::concurrent::Mutex::wait
>> (this=0x9547d6c) at decaf/util/concurrent/Mutex.cpp:95
>> #5  0x0094c9c5 in activemq::threads::CompositeTaskRunner::run
>> (this=0x9547d48) at activemq/threads/CompositeTaskRunner.cpp:118
>> #6  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9547e90) at decaf/lang/Thread.cpp:135
>> #7  0x00be2847 in threadWorker (arg=0x9547e90) at
>> decaf/lang/Thread.cpp:188
>> #8  0x0046796e in start_thread (arg=0xb7799b70) at pthread_create.c:300
>> #9  0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 3:
>> #0  0x00eed422 in __kernel_vsyscall ()
>> #1  0x0046eaf9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x00469f61 in __pthread_mutex_lock (mutex=0x95523a8) at
>> pthread_mutex_lock.c:61
>> #4  0x00325ba6 in pthread_mutex_lock (mutex=0x95523a8) at forward.c:182
>> #5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x95523a8) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x00c21412 in decaf::util::concurrent::Mutex::lock
>> (this=0x95522bc) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x00915770 in
>>
>> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
>> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
>> (this=0x95522a8)
>>     at ./decaf/util/StlQueue.h:253
>> #8  activemq::core::MessageDispatchChannel::lock (this=0x95522a8) at
>> activemq/core/MessageDispatchChannel.h:153
>> #9  0x00c20ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb6f97fe4) at decaf/util/concurrent/Lock.cpp:54
>> #10 0x00c20c08 in Lock (this=0xfffffe00, object=0x95523a8,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #11 0x008de102 in
>> activemq::core::ActiveMQConsumer::clearMessagesInProgress
>> (this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:1112
>> #12 0x008ff15c in
>> activemq::core::ActiveMQSession::clearMessagesInProgress
>> (this=0x9551b90) at activemq/core/ActiveMQSession.cpp:239
>> #13 0x008c8543 in
>> activemq::core::ActiveMQConnection::transportInterrupted
>> (this=0x9548210) at activemq/core/ActiveMQConnection.cpp:704
>> #14 0x00953b34 in
>> activemq::transport::TransportFilter::transportInterrupted
>> (this=0x95481c0) at activemq/transport/TransportFilter.cpp:67
>> #15 0x009687ae in
>> activemq::transport::failover::FailoverTransport::handleTransportFailure
>> (this=0x9547a38, error=...) at
>> activemq/transport/failover/FailoverTransport.cpp:476
>> #16 0x0097406c in
>> activemq::transport::failover::FailoverTransportListener::onException
>> (this=0x9547878, ex=...) at
>> activemq/transport/failover/FailoverTransportListener.cpp:97
>> #17 0x00953c0b in activemq::transport::TransportFilter::fire
>> (this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #18 0x00953c64 in activemq::transport::TransportFilter::onException
>> (this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #19 0x00953c0b in activemq::transport::TransportFilter::fire
>> (this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #20 0x00953c64 in activemq::transport::TransportFilter::onException
>> (this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #21 0x00951b13 in activemq::transport::IOTransport::fire
>> (this=0x95497a0, ex=...) at activemq/transport/IOTransport.cpp:73
>> #22 0x009523bf in activemq::transport::IOTransport::run
>> (this=0x95497a0) at activemq/transport/IOTransport.cpp:246
>> #23 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9549ea0) at decaf/lang/Thread.cpp:135
>> #24 0x00be2847 in threadWorker (arg=0x9549ea0) at
>> decaf/lang/Thread.cpp:188
>> #25 0x0046796e in start_thread (arg=0xb6f98b70) at pthread_create.c:300
>> #26 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 4:
>> #0  0x00eed422 in __kernel_vsyscall ()
>> #1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
>> #2  0x003259dd in __pthread_cond_wait (cond=0x95516c8,
>> mutex=0x9551698) at forward.c:139
>> #3  0x00bb06a9 in
>> decaf::internal::util::concurrent::ConditionImpl::wait
>> (condition=0x95516c8) at
>> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
>> #4  0x00c21353 in decaf::util::concurrent::Mutex::wait
>> (this=0x9551584) at decaf/util/concurrent/Mutex.cpp:95
>> #5  0x00c2029a in decaf::util::concurrent::CountDownLatch::await
>> (this=0x955157c) at decaf/util/concurrent/CountDownLatch.cpp:56
>> #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x9551574,
>> timeOut=-1) at main.cpp:226
>> #7  0x0804bb46 in Consumer::run (this=0x9551570) at main.cpp:379
>> #8  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x95515b0) at decaf/lang/Thread.cpp:135
>> #9  0x00be2847 in threadWorker (arg=0x95515b0) at
>> decaf/lang/Thread.cpp:188
>> #10 0x0046796e in start_thread (arg=0xb6797b70) at pthread_create.c:300
>> #11 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 5:
>> #0  0x00eed422 in __kernel_vsyscall ()
>> #1  0x0046eaf9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x00469f61 in __pthread_mutex_lock (mutex=0x95475e0) at
>> pthread_mutex_lock.c:61
>> #4  0x00325ba6 in pthread_mutex_lock (mutex=0x95475e0) at forward.c:182
>> #5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x95475e0) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x00c21412 in decaf::util::concurrent::Mutex::lock
>> (this=0x9547a88) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x00c20ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb5f95ce0) at decaf/util/concurrent/Lock.cpp:54
>> #8  0x00c20c08 in Lock (this=0xfffffe00, object=0x95475e0,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #9  0x0096b8fd in
>> activemq::transport::failover::FailoverTransport::oneway
>> (this=0x9547a38, command=...) at
>> activemq/transport/failover/FailoverTransport.cpp:186
>> #10 0x00957f9f in
>> activemq::transport::correlator::ResponseCorrelator::oneway
>> (this=0x95481c0, command=...) at
>> activemq/transport/correlator/ResponseCorrelator.cpp:82
>> #11 0x008c13be in activemq::core::ActiveMQConnection::oneway
>> (this=0x9548210, command=...) at
>> activemq/core/ActiveMQConnection.cpp:741
>> #12 0x00900a4f in activemq::core::ActiveMQSession::oneway
>> (this=0x9551b90, command=...) at activemq/core/ActiveMQSession.cpp:903
>> #13 0x008e5a1c in activemq::core::ActiveMQConsumer::acknowledge
>> (this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:860
>> #14 0x008ef885 in
>> activemq::core::TransactionSynhcronization::beforeEnd (this=0x95529e8)
>> at activemq/core/ActiveMQConsumer.cpp:85
>> #15 0x00910588 in
>> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x9551dc8)
>> at activemq/core/ActiveMQTransactionContext.cpp:192
>> #16 0x00910d63 in activemq::core::ActiveMQTransactionContext::commit
>> (this=0x9551dc8) at activemq/core/ActiveMQTransactionContext.cpp:127
>> #17 0x008fe460 in activemq::core::ActiveMQSession::commit
>> (this=0x9551b90) at activemq/core/ActiveMQSession.cpp:189
>> #18 0x007ec1a0 in activemq::cmsutil::PooledSession::commit
>> (this=0x9551f38) at activemq/cmsutil/PooledSession.h:87
>> #19 0x0804c0fb in Consumer::onMessage (this=0x9551570,
>> message=0x9552ec0) at main.cpp:455
>> #20 0x008e972f in activemq::core::ActiveMQConsumer::dispatch
>> (this=0x9552270, dispatch=...) at
>> activemq/core/ActiveMQConsumer.cpp:1018
>> #21 0x0090d5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
>> (this=0x9551e78, dispatch=...) at
>> activemq/core/ActiveMQSessionExecutor.cpp:129
>> #22 0x0090d993 in activemq::core::ActiveMQSessionExecutor::iterate
>> (this=0x9551e78) at activemq/core/ActiveMQSessionExecutor.cpp:166
>> #23 0x009507f3 in activemq::threads::DedicatedTaskRunner::run
>> (this=0x9552550) at activemq/threads/DedicatedTaskRunner.cpp:111
>> #24 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9551ad8) at decaf/lang/Thread.cpp:135
>> #25 0x00be2847 in threadWorker (arg=0x9551ad8) at
>> decaf/lang/Thread.cpp:188
>> #26 0x0046796e in start_thread (arg=0xb5f96b70) at pthread_create.c:300
>> #27 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>>
>>
>> BACKTRACE 2:
>>
>> Thread 1:
>> #0  0x0079b422 in __kernel_vsyscall ()
>> #1  0x008bfb5d in pthread_join (threadid=3061021552,
>> thread_return=0xbfbff07c) at pthread_join.c:89
>> #2  0x0132a3c0 in decaf::lang::Thread::join (this=0x9be0574) at
>> decaf/lang/Thread.cpp:421
>> #3  0x0804ca15 in AppTest::joinEndPoins (this=0xbfbff104) at main.cpp:587
>> #4  0x0804c69c in AppTest::run (this=0xbfbff104) at main.cpp:556
>> #5  0x0804a33c in main (argc=4, argv=0xbfbff224) at main.cpp:634
>>
>> Thread 2:
>> #0  0x0079b422 in __kernel_vsyscall ()
>> #1  0x008c3015 in pthread_cond_wait@@GLIBC_2.3.2 () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
>> #2  0x004ed9dd in __pthread_cond_wait (cond=0x9bd6e38,
>> mutex=0x9bd6e08) at forward.c:139
>> #3  0x012f86a9 in
>> decaf::internal::util::concurrent::ConditionImpl::wait
>> (condition=0x9bd6e38) at
>> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
>> #4  0x01369353 in decaf::util::concurrent::Mutex::wait
>> (this=0x9bd6d6c) at decaf/util/concurrent/Mutex.cpp:95
>> #5  0x010949c5 in activemq::threads::CompositeTaskRunner::run
>> (this=0x9bd6d48) at activemq/threads/CompositeTaskRunner.cpp:118
>> #6  0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9bd6e90) at decaf/lang/Thread.cpp:135
>> #7  0x0132a847 in threadWorker (arg=0x9bd6e90) at
>> decaf/lang/Thread.cpp:188
>> #8  0x008be96e in start_thread (arg=0xb7739b70) at pthread_create.c:300
>> #9  0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 3:
>> #0  0x0079b422 in __kernel_vsyscall ()
>> #1  0x008c5af9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
>> pthread_mutex_lock.c:61
>> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
>> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x9be12d8) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
>> (this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x0105d770 in
>>
>> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
>> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
>> (this=0x9be11d8)
>>     at ./decaf/util/StlQueue.h:253
>> #8  activemq::core::MessageDispatchChannel::lock (this=0x9be11d8) at
>> activemq/core/MessageDispatchChannel.h:153
>> #9  0x01368ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb6f37fe4) at decaf/util/concurrent/Lock.cpp:54
>> #10 0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #11 0x01026102 in
>> activemq::core::ActiveMQConsumer::clearMessagesInProgress
>> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:1112
>> #12 0x0104715c in
>> activemq::core::ActiveMQSession::clearMessagesInProgress
>> (this=0x9be0b90) at activemq/core/ActiveMQSession.cpp:239
>> #13 0x01010543 in
>> activemq::core::ActiveMQConnection::transportInterrupted
>> (this=0x9bd7210) at activemq/core/ActiveMQConnection.cpp:704
>> #14 0x0109bb34 in
>> activemq::transport::TransportFilter::transportInterrupted
>> (this=0x9bd71c0) at activemq/transport/TransportFilter.cpp:67
>> #15 0x010b07ae in
>> activemq::transport::failover::FailoverTransport::handleTransportFailure
>> (this=0x9bd6a38, error=...) at
>> activemq/transport/failover/FailoverTransport.cpp:476
>> #16 0x010bc06c in
>> activemq::transport::failover::FailoverTransportListener::onException
>> (this=0x9bd6878, ex=...) at
>> activemq/transport/failover/FailoverTransportListener.cpp:97
>> #17 0x0109bc0b in activemq::transport::TransportFilter::fire
>> (this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #18 0x0109bc64 in activemq::transport::TransportFilter::onException
>> (this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #19 0x0109bc0b in activemq::transport::TransportFilter::fire
>> (this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #20 0x0109bc64 in activemq::transport::TransportFilter::onException
>> (this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #21 0x01099b13 in activemq::transport::IOTransport::fire
>> (this=0x9bd88e8, ex=...) at activemq/transport/IOTransport.cpp:73
>> #22 0x0109a3bf in activemq::transport::IOTransport::run
>> (this=0x9bd88e8) at activemq/transport/IOTransport.cpp:246
>> #23 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9bd8fe8) at decaf/lang/Thread.cpp:135
>> #24 0x0132a847 in threadWorker (arg=0x9bd8fe8) at
>> decaf/lang/Thread.cpp:188
>> #25 0x008be96e in start_thread (arg=0xb6f38b70) at pthread_create.c:300
>> #26 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 4:
>> #0  0x0079b422 in __kernel_vsyscall ()
>> #1  0x008c5af9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
>> pthread_mutex_lock.c:61
>> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
>> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x9be12d8) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
>> (this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x01036830 in
>>
>> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
>> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
>> (this=0x9be11e0)
>>     at ./decaf/util/StlQueue.h:253
>> #8  0x01368ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb6737064) at decaf/util/concurrent/Lock.cpp:54
>> #9  0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #10 0x0105c507 in activemq::core::MessageDispatchChannel::clear
>> (this=0x9be11d8) at activemq/core/MessageDispatchChannel.cpp:143
>> #11 0x0102ec37 in activemq::core::ActiveMQConsumer::doClose
>> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:335
>> #12 0x0102fb00 in activemq::core::ActiveMQConsumer::close
>> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:309
>> #13 0x01030390 in ~ActiveMQConsumer (this=0x9be11a0, __in_chrg=<value
>> optimized out>) at activemq/core/ActiveMQConsumer.cpp:261
>> #14 0x0804c276 in Consumer::cleanup (this=0x9be0570) at main.cpp:469
>> #15 0x0804bcdd in Consumer::run (this=0x9be0570) at main.cpp:394
>> #16 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9be05b0) at decaf/lang/Thread.cpp:135
>> #17 0x0132a847 in threadWorker (arg=0x9be05b0) at
>> decaf/lang/Thread.cpp:188
>> #18 0x008be96e in start_thread (arg=0xb6737b70) at pthread_create.c:300
>> #19 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 5:
>> #0  0x0079b422 in __kernel_vsyscall ()
>> #1  0x008c5af9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9bd65e0) at
>> pthread_mutex_lock.c:61
>> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9bd65e0) at forward.c:182
>> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x9bd65e0) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
>> (this=0x9bd6a88) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x01368ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb5f35ea0) at decaf/util/concurrent/Lock.cpp:54
>> #8  0x01368c08 in Lock (this=0xfffffe00, object=0x9bd65e0,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #9  0x010b38fd in
>> activemq::transport::failover::FailoverTransport::oneway
>> (this=0x9bd6a38, command=...) at
>> activemq/transport/failover/FailoverTransport.cpp:186
>> #10 0x0109ff9f in
>> activemq::transport::correlator::ResponseCorrelator::oneway
>> (this=0x9bd71c0, command=...) at
>> activemq/transport/correlator/ResponseCorrelator.cpp:82
>> #11 0x010093be in activemq::core::ActiveMQConnection::oneway
>> (this=0x9bd7210, command=...) at
>> activemq/core/ActiveMQConnection.cpp:741
>> #12 0x01048a4f in activemq::core::ActiveMQSession::oneway
>> (this=0x9be0b90, command=...) at activemq/core/ActiveMQSession.cpp:903
>> #13 0x01030df4 in
>> activemq::core::ActiveMQConsumer::afterMessageIsConsumed
>> (this=0x9be11a0, message=..., messageExpired=false) at
>> activemq/core/ActiveMQConsumer.cpp:642
>> #14 0x01031743 in activemq::core::ActiveMQConsumer::dispatch
>> (this=0x9be11a0, dispatch=...) at
>> activemq/core/ActiveMQConsumer.cpp:1021
>> #15 0x010555c3 in activemq::core::ActiveMQSessionExecutor::dispatch
>> (this=0x9be0dc8, dispatch=...) at
>> activemq/core/ActiveMQSessionExecutor.cpp:129
>> #16 0x01055993 in activemq::core::ActiveMQSessionExecutor::iterate
>> (this=0x9be0dc8) at activemq/core/ActiveMQSessionExecutor.cpp:166
>> #17 0x010987f3 in activemq::threads::DedicatedTaskRunner::run
>> (this=0x9be1480) at activemq/threads/DedicatedTaskRunner.cpp:111
>> #18 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x9be0ad8) at decaf/lang/Thread.cpp:135
>> #19 0x0132a847 in threadWorker (arg=0x9be0ad8) at
>> decaf/lang/Thread.cpp:188
>> #20 0x008be96e in start_thread (arg=0xb5f36b70) at pthread_create.c:300
>> #21 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> 2011/3/1 Oscar Pernas <os...@pernas.es>
>> >
>> > Hi,
>> >
>> > I ran into this bug before. If I am shutting down and starting the
>> broker so
>> > many times, connection are not reestablished and the consumer gets
>> frozen.
>> > If you try to close the consumer that is frozen, the closing stucks your
>> > thread too? In my case, this problem only happens when I'm debugging the
>> > application (without breaks but with visual studio debug).
>> > I've partially resolved it manually, if I dont received anything over a
>> > topic in a time, I try to close the consumer and reconnect it again.
>> >
>> > regards
>> >
>> > --
>> > Óscar Pernas Plaza.
>>
>
>
>
> --
> Óscar Pernas Plaza.
>



-- 
Óscar Pernas Plaza.

Re: AMQCPP: unexpected exception

Posted by Oscar Pernas <os...@pernas.es>.

Yes, It's probably the same issue. I've resolved the destructor stucks
(partially) deleting connection and session before delete the consumer.
Perhaps you should catch a decaf exception that sometimes is thrown.

Tell me if it helps you.


regards

2011/3/4 Henrique Magarotto <ma...@gmail.com>

> Hi Tim and Oscar,
> thanks for your attention,
>
> So, in my case i got frozen in a commit attempt.
> But really, testing without transaction and MessageListener i got
> frozen in consumer destructor too.
> Maybe it's the same reason for both cases. I tested with several
> configurations to try avoid freezes for both  cases.
> I opened the AMQCPP-354 issue with 'unexpected exception' report, but
> i think that these cases deserve more one (or two) new issues in Jira
> for later investigation.
>
> regards
> Henrique
>
> Following how to reproduce the cases:
>
> MESSAGELISTENER WITH TRANSACTION, FREEZES IN COMMIT
> MessageListener consumer destructor freezes in commit:
> 1- Start ActiveMQ
> 2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
> 3- Send one message './activemqTest -p1'
> 4- Wait consumer 'Starting delay'
> 5- Stop ActiveMQ
> 6- Consumer will try commit
> 8- Freezes (see below 'BACKTRACE 1')!!!
>
>
> MESSAGELISTENER WITHOUT TRANSACTION, FREEZES IN DESTRUCTOR
> MessageListener consumer destructor freezes (apply changes in test code):
> 1- Start ActiveMQ
> 2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
> 3- Send one message './activemqTest -p1'
> 4- Wait consumer 'Starting delay'
> 5- Stop ActiveMQ
> 6- Wait consumer delay finishes 'Delay finish'
> 7- Try kill consumer ( not kill -9 :-) )
> 8- Freezes (see below 'BACKTRACE 2')!!!
>
> Changes in test:
> @@ -440,6 +440,7 @@
>              if(_delay) {
>                  std::cout<<"Starting delay..."<<std::endl;
>                  stopCheck(_delay);
> +                std::cout<<"Delay finish..."<<std::endl;
>              }
>
>          } catch (CMSException& e) {
> @@ -561,7 +562,7 @@
>
>      void loadEndPoins() {
>          for(int i=0; i<_consumer;i++) {
> -            _endPointList.push_back( new
> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
> +            _endPointList.push_back( new
> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener,false) );
>          }
>          for(int i=0; i<_producer;i++) {
>              _endPointList.push_back( new
> Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
>
>
> BACKTRACE 1:
>
> Thread 1:
> #0  0x00eed422 in __kernel_vsyscall ()
> #1  0x00468b5d in pthread_join (threadid=3061414768,
> thread_return=0xbf951e8c) at pthread_join.c:89
> #2  0x00be23c0 in decaf::lang::Thread::join (this=0x9551574) at
> decaf/lang/Thread.cpp:421
> #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf951f14) at main.cpp:586
> #4  0x0804c678 in AppTest::run (this=0xbf951f14) at main.cpp:555
> #5  0x0804a33c in main (argc=4, argv=0xbf952034) at main.cpp:633
>
> Thread 2:
> #0  0x00eed422 in __kernel_vsyscall ()
> #1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x003259dd in __pthread_cond_wait (cond=0x9547e38,
> mutex=0x9547e08) at forward.c:139
> #3  0x00bb06a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x9547e38) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00c21353 in decaf::util::concurrent::Mutex::wait
> (this=0x9547d6c) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x0094c9c5 in activemq::threads::CompositeTaskRunner::run
> (this=0x9547d48) at activemq/threads/CompositeTaskRunner.cpp:118
> #6  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9547e90) at decaf/lang/Thread.cpp:135
> #7  0x00be2847 in threadWorker (arg=0x9547e90) at decaf/lang/Thread.cpp:188
> #8  0x0046796e in start_thread (arg=0xb7799b70) at pthread_create.c:300
> #9  0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 3:
> #0  0x00eed422 in __kernel_vsyscall ()
> #1  0x0046eaf9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00469f61 in __pthread_mutex_lock (mutex=0x95523a8) at
> pthread_mutex_lock.c:61
> #4  0x00325ba6 in pthread_mutex_lock (mutex=0x95523a8) at forward.c:182
> #5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x95523a8) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00c21412 in decaf::util::concurrent::Mutex::lock
> (this=0x95522bc) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x00915770 in
>
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> (this=0x95522a8)
>     at ./decaf/util/StlQueue.h:253
> #8  activemq::core::MessageDispatchChannel::lock (this=0x95522a8) at
> activemq/core/MessageDispatchChannel.h:153
> #9  0x00c20ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb6f97fe4) at decaf/util/concurrent/Lock.cpp:54
> #10 0x00c20c08 in Lock (this=0xfffffe00, object=0x95523a8,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #11 0x008de102 in
> activemq::core::ActiveMQConsumer::clearMessagesInProgress
> (this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:1112
> #12 0x008ff15c in
> activemq::core::ActiveMQSession::clearMessagesInProgress
> (this=0x9551b90) at activemq/core/ActiveMQSession.cpp:239
> #13 0x008c8543 in
> activemq::core::ActiveMQConnection::transportInterrupted
> (this=0x9548210) at activemq/core/ActiveMQConnection.cpp:704
> #14 0x00953b34 in
> activemq::transport::TransportFilter::transportInterrupted
> (this=0x95481c0) at activemq/transport/TransportFilter.cpp:67
> #15 0x009687ae in
> activemq::transport::failover::FailoverTransport::handleTransportFailure
> (this=0x9547a38, error=...) at
> activemq/transport/failover/FailoverTransport.cpp:476
> #16 0x0097406c in
> activemq::transport::failover::FailoverTransportListener::onException
> (this=0x9547878, ex=...) at
> activemq/transport/failover/FailoverTransportListener.cpp:97
> #17 0x00953c0b in activemq::transport::TransportFilter::fire
> (this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:49
> #18 0x00953c64 in activemq::transport::TransportFilter::onException
> (this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:41
> #19 0x00953c0b in activemq::transport::TransportFilter::fire
> (this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:49
> #20 0x00953c64 in activemq::transport::TransportFilter::onException
> (this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:41
> #21 0x00951b13 in activemq::transport::IOTransport::fire
> (this=0x95497a0, ex=...) at activemq/transport/IOTransport.cpp:73
> #22 0x009523bf in activemq::transport::IOTransport::run
> (this=0x95497a0) at activemq/transport/IOTransport.cpp:246
> #23 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9549ea0) at decaf/lang/Thread.cpp:135
> #24 0x00be2847 in threadWorker (arg=0x9549ea0) at decaf/lang/Thread.cpp:188
> #25 0x0046796e in start_thread (arg=0xb6f98b70) at pthread_create.c:300
> #26 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 4:
> #0  0x00eed422 in __kernel_vsyscall ()
> #1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x003259dd in __pthread_cond_wait (cond=0x95516c8,
> mutex=0x9551698) at forward.c:139
> #3  0x00bb06a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x95516c8) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00c21353 in decaf::util::concurrent::Mutex::wait
> (this=0x9551584) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x00c2029a in decaf::util::concurrent::CountDownLatch::await
> (this=0x955157c) at decaf/util/concurrent/CountDownLatch.cpp:56
> #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x9551574,
> timeOut=-1) at main.cpp:226
> #7  0x0804bb46 in Consumer::run (this=0x9551570) at main.cpp:379
> #8  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x95515b0) at decaf/lang/Thread.cpp:135
> #9  0x00be2847 in threadWorker (arg=0x95515b0) at decaf/lang/Thread.cpp:188
> #10 0x0046796e in start_thread (arg=0xb6797b70) at pthread_create.c:300
> #11 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 5:
> #0  0x00eed422 in __kernel_vsyscall ()
> #1  0x0046eaf9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00469f61 in __pthread_mutex_lock (mutex=0x95475e0) at
> pthread_mutex_lock.c:61
> #4  0x00325ba6 in pthread_mutex_lock (mutex=0x95475e0) at forward.c:182
> #5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x95475e0) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00c21412 in decaf::util::concurrent::Mutex::lock
> (this=0x9547a88) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x00c20ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb5f95ce0) at decaf/util/concurrent/Lock.cpp:54
> #8  0x00c20c08 in Lock (this=0xfffffe00, object=0x95475e0,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #9  0x0096b8fd in
> activemq::transport::failover::FailoverTransport::oneway
> (this=0x9547a38, command=...) at
> activemq/transport/failover/FailoverTransport.cpp:186
> #10 0x00957f9f in
> activemq::transport::correlator::ResponseCorrelator::oneway
> (this=0x95481c0, command=...) at
> activemq/transport/correlator/ResponseCorrelator.cpp:82
> #11 0x008c13be in activemq::core::ActiveMQConnection::oneway
> (this=0x9548210, command=...) at
> activemq/core/ActiveMQConnection.cpp:741
> #12 0x00900a4f in activemq::core::ActiveMQSession::oneway
> (this=0x9551b90, command=...) at activemq/core/ActiveMQSession.cpp:903
> #13 0x008e5a1c in activemq::core::ActiveMQConsumer::acknowledge
> (this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:860
> #14 0x008ef885 in
> activemq::core::TransactionSynhcronization::beforeEnd (this=0x95529e8)
> at activemq/core/ActiveMQConsumer.cpp:85
> #15 0x00910588 in
> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x9551dc8)
> at activemq/core/ActiveMQTransactionContext.cpp:192
> #16 0x00910d63 in activemq::core::ActiveMQTransactionContext::commit
> (this=0x9551dc8) at activemq/core/ActiveMQTransactionContext.cpp:127
> #17 0x008fe460 in activemq::core::ActiveMQSession::commit
> (this=0x9551b90) at activemq/core/ActiveMQSession.cpp:189
> #18 0x007ec1a0 in activemq::cmsutil::PooledSession::commit
> (this=0x9551f38) at activemq/cmsutil/PooledSession.h:87
> #19 0x0804c0fb in Consumer::onMessage (this=0x9551570,
> message=0x9552ec0) at main.cpp:455
> #20 0x008e972f in activemq::core::ActiveMQConsumer::dispatch
> (this=0x9552270, dispatch=...) at
> activemq/core/ActiveMQConsumer.cpp:1018
> #21 0x0090d5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
> (this=0x9551e78, dispatch=...) at
> activemq/core/ActiveMQSessionExecutor.cpp:129
> #22 0x0090d993 in activemq::core::ActiveMQSessionExecutor::iterate
> (this=0x9551e78) at activemq/core/ActiveMQSessionExecutor.cpp:166
> #23 0x009507f3 in activemq::threads::DedicatedTaskRunner::run
> (this=0x9552550) at activemq/threads/DedicatedTaskRunner.cpp:111
> #24 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9551ad8) at decaf/lang/Thread.cpp:135
> #25 0x00be2847 in threadWorker (arg=0x9551ad8) at decaf/lang/Thread.cpp:188
> #26 0x0046796e in start_thread (arg=0xb5f96b70) at pthread_create.c:300
> #27 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
>
>
> BACKTRACE 2:
>
> Thread 1:
> #0  0x0079b422 in __kernel_vsyscall ()
> #1  0x008bfb5d in pthread_join (threadid=3061021552,
> thread_return=0xbfbff07c) at pthread_join.c:89
> #2  0x0132a3c0 in decaf::lang::Thread::join (this=0x9be0574) at
> decaf/lang/Thread.cpp:421
> #3  0x0804ca15 in AppTest::joinEndPoins (this=0xbfbff104) at main.cpp:587
> #4  0x0804c69c in AppTest::run (this=0xbfbff104) at main.cpp:556
> #5  0x0804a33c in main (argc=4, argv=0xbfbff224) at main.cpp:634
>
> Thread 2:
> #0  0x0079b422 in __kernel_vsyscall ()
> #1  0x008c3015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x004ed9dd in __pthread_cond_wait (cond=0x9bd6e38,
> mutex=0x9bd6e08) at forward.c:139
> #3  0x012f86a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x9bd6e38) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x01369353 in decaf::util::concurrent::Mutex::wait
> (this=0x9bd6d6c) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x010949c5 in activemq::threads::CompositeTaskRunner::run
> (this=0x9bd6d48) at activemq/threads/CompositeTaskRunner.cpp:118
> #6  0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9bd6e90) at decaf/lang/Thread.cpp:135
> #7  0x0132a847 in threadWorker (arg=0x9bd6e90) at decaf/lang/Thread.cpp:188
> #8  0x008be96e in start_thread (arg=0xb7739b70) at pthread_create.c:300
> #9  0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 3:
> #0  0x0079b422 in __kernel_vsyscall ()
> #1  0x008c5af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
> pthread_mutex_lock.c:61
> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x9be12d8) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
> (this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x0105d770 in
>
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> (this=0x9be11d8)
>     at ./decaf/util/StlQueue.h:253
> #8  activemq::core::MessageDispatchChannel::lock (this=0x9be11d8) at
> activemq/core/MessageDispatchChannel.h:153
> #9  0x01368ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb6f37fe4) at decaf/util/concurrent/Lock.cpp:54
> #10 0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #11 0x01026102 in
> activemq::core::ActiveMQConsumer::clearMessagesInProgress
> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:1112
> #12 0x0104715c in
> activemq::core::ActiveMQSession::clearMessagesInProgress
> (this=0x9be0b90) at activemq/core/ActiveMQSession.cpp:239
> #13 0x01010543 in
> activemq::core::ActiveMQConnection::transportInterrupted
> (this=0x9bd7210) at activemq/core/ActiveMQConnection.cpp:704
> #14 0x0109bb34 in
> activemq::transport::TransportFilter::transportInterrupted
> (this=0x9bd71c0) at activemq/transport/TransportFilter.cpp:67
> #15 0x010b07ae in
> activemq::transport::failover::FailoverTransport::handleTransportFailure
> (this=0x9bd6a38, error=...) at
> activemq/transport/failover/FailoverTransport.cpp:476
> #16 0x010bc06c in
> activemq::transport::failover::FailoverTransportListener::onException
> (this=0x9bd6878, ex=...) at
> activemq/transport/failover/FailoverTransportListener.cpp:97
> #17 0x0109bc0b in activemq::transport::TransportFilter::fire
> (this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:49
> #18 0x0109bc64 in activemq::transport::TransportFilter::onException
> (this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:41
> #19 0x0109bc0b in activemq::transport::TransportFilter::fire
> (this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:49
> #20 0x0109bc64 in activemq::transport::TransportFilter::onException
> (this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:41
> #21 0x01099b13 in activemq::transport::IOTransport::fire
> (this=0x9bd88e8, ex=...) at activemq/transport/IOTransport.cpp:73
> #22 0x0109a3bf in activemq::transport::IOTransport::run
> (this=0x9bd88e8) at activemq/transport/IOTransport.cpp:246
> #23 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9bd8fe8) at decaf/lang/Thread.cpp:135
> #24 0x0132a847 in threadWorker (arg=0x9bd8fe8) at decaf/lang/Thread.cpp:188
> #25 0x008be96e in start_thread (arg=0xb6f38b70) at pthread_create.c:300
> #26 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 4:
> #0  0x0079b422 in __kernel_vsyscall ()
> #1  0x008c5af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
> pthread_mutex_lock.c:61
> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x9be12d8) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
> (this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x01036830 in
>
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> (this=0x9be11e0)
>     at ./decaf/util/StlQueue.h:253
> #8  0x01368ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb6737064) at decaf/util/concurrent/Lock.cpp:54
> #9  0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #10 0x0105c507 in activemq::core::MessageDispatchChannel::clear
> (this=0x9be11d8) at activemq/core/MessageDispatchChannel.cpp:143
> #11 0x0102ec37 in activemq::core::ActiveMQConsumer::doClose
> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:335
> #12 0x0102fb00 in activemq::core::ActiveMQConsumer::close
> (this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:309
> #13 0x01030390 in ~ActiveMQConsumer (this=0x9be11a0, __in_chrg=<value
> optimized out>) at activemq/core/ActiveMQConsumer.cpp:261
> #14 0x0804c276 in Consumer::cleanup (this=0x9be0570) at main.cpp:469
> #15 0x0804bcdd in Consumer::run (this=0x9be0570) at main.cpp:394
> #16 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9be05b0) at decaf/lang/Thread.cpp:135
> #17 0x0132a847 in threadWorker (arg=0x9be05b0) at decaf/lang/Thread.cpp:188
> #18 0x008be96e in start_thread (arg=0xb6737b70) at pthread_create.c:300
> #19 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> Thread 5:
> #0  0x0079b422 in __kernel_vsyscall ()
> #1  0x008c5af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9bd65e0) at
> pthread_mutex_lock.c:61
> #4  0x004edba6 in pthread_mutex_lock (mutex=0x9bd65e0) at forward.c:182
> #5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x9bd65e0) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x01369412 in decaf::util::concurrent::Mutex::lock
> (this=0x9bd6a88) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x01368ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb5f35ea0) at decaf/util/concurrent/Lock.cpp:54
> #8  0x01368c08 in Lock (this=0xfffffe00, object=0x9bd65e0,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #9  0x010b38fd in
> activemq::transport::failover::FailoverTransport::oneway
> (this=0x9bd6a38, command=...) at
> activemq/transport/failover/FailoverTransport.cpp:186
> #10 0x0109ff9f in
> activemq::transport::correlator::ResponseCorrelator::oneway
> (this=0x9bd71c0, command=...) at
> activemq/transport/correlator/ResponseCorrelator.cpp:82
> #11 0x010093be in activemq::core::ActiveMQConnection::oneway
> (this=0x9bd7210, command=...) at
> activemq/core/ActiveMQConnection.cpp:741
> #12 0x01048a4f in activemq::core::ActiveMQSession::oneway
> (this=0x9be0b90, command=...) at activemq/core/ActiveMQSession.cpp:903
> #13 0x01030df4 in
> activemq::core::ActiveMQConsumer::afterMessageIsConsumed
> (this=0x9be11a0, message=..., messageExpired=false) at
> activemq/core/ActiveMQConsumer.cpp:642
> #14 0x01031743 in activemq::core::ActiveMQConsumer::dispatch
> (this=0x9be11a0, dispatch=...) at
> activemq/core/ActiveMQConsumer.cpp:1021
> #15 0x010555c3 in activemq::core::ActiveMQSessionExecutor::dispatch
> (this=0x9be0dc8, dispatch=...) at
> activemq/core/ActiveMQSessionExecutor.cpp:129
> #16 0x01055993 in activemq::core::ActiveMQSessionExecutor::iterate
> (this=0x9be0dc8) at activemq/core/ActiveMQSessionExecutor.cpp:166
> #17 0x010987f3 in activemq::threads::DedicatedTaskRunner::run
> (this=0x9be1480) at activemq/threads/DedicatedTaskRunner.cpp:111
> #18 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x9be0ad8) at decaf/lang/Thread.cpp:135
> #19 0x0132a847 in threadWorker (arg=0x9be0ad8) at decaf/lang/Thread.cpp:188
> #20 0x008be96e in start_thread (arg=0xb5f36b70) at pthread_create.c:300
> #21 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>
> 2011/3/1 Oscar Pernas <os...@pernas.es>
> >
> > Hi,
> >
> > I ran into this bug before. If I am shutting down and starting the broker
> so
> > many times, connection are not reestablished and the consumer gets
> frozen.
> > If you try to close the consumer that is frozen, the closing stucks your
> > thread too? In my case, this problem only happens when I'm debugging the
> > application (without breaks but with visual studio debug).
> > I've partially resolved it manually, if I dont received anything over a
> > topic in a time, I try to close the consumer and reconnect it again.
> >
> > regards
> >
> > --
> > Óscar Pernas Plaza.
>



-- 
Óscar Pernas Plaza.

Re: AMQCPP: unexpected exception

Posted by Henrique Magarotto <ma...@gmail.com>.

Hi Tim and Oscar,
thanks for your attention,

So, in my case i got frozen in a commit attempt.
But really, testing without transaction and MessageListener i got
frozen in consumer destructor too.
Maybe it's the same reason for both cases. I tested with several
configurations to try avoid freezes for both  cases.
I opened the AMQCPP-354 issue with 'unexpected exception' report, but
i think that these cases deserve more one (or two) new issues in Jira
for later investigation.

regards
Henrique

Following how to reproduce the cases:

MESSAGELISTENER WITH TRANSACTION, FREEZES IN COMMIT
MessageListener consumer destructor freezes in commit:
1- Start ActiveMQ
2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
3- Send one message './activemqTest -p1'
4- Wait consumer 'Starting delay'
5- Stop ActiveMQ
6- Consumer will try commit
8- Freezes (see below 'BACKTRACE 1')!!!


MESSAGELISTENER WITHOUT TRANSACTION, FREEZES IN DESTRUCTOR
MessageListener consumer destructor freezes (apply changes in test code):
1- Start ActiveMQ
2- Start 'MessageListener' consumer './activemqTest -l -c1 -d10000'
3- Send one message './activemqTest -p1'
4- Wait consumer 'Starting delay'
5- Stop ActiveMQ
6- Wait consumer delay finishes 'Delay finish'
7- Try kill consumer ( not kill -9 :-) )
8- Freezes (see below 'BACKTRACE 2')!!!

Changes in test:
@@ -440,6 +440,7 @@
             if(_delay) {
                 std::cout<<"Starting delay..."<<std::endl;
                 stopCheck(_delay);
+                std::cout<<"Delay finish..."<<std::endl;
             }

         } catch (CMSException& e) {
@@ -561,7 +562,7 @@

     void loadEndPoins() {
         for(int i=0; i<_consumer;i++) {
-            _endPointList.push_back( new
Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
+            _endPointList.push_back( new
Consumer(&_poolManager,"TEST.FOO",_delay,_useListener,false) );
         }
         for(int i=0; i<_producer;i++) {
             _endPointList.push_back( new
Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );


BACKTRACE 1:

Thread 1:
#0  0x00eed422 in __kernel_vsyscall ()
#1  0x00468b5d in pthread_join (threadid=3061414768,
thread_return=0xbf951e8c) at pthread_join.c:89
#2  0x00be23c0 in decaf::lang::Thread::join (this=0x9551574) at
decaf/lang/Thread.cpp:421
#3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf951f14) at main.cpp:586
#4  0x0804c678 in AppTest::run (this=0xbf951f14) at main.cpp:555
#5  0x0804a33c in main (argc=4, argv=0xbf952034) at main.cpp:633

Thread 2:
#0  0x00eed422 in __kernel_vsyscall ()
#1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
#2  0x003259dd in __pthread_cond_wait (cond=0x9547e38,
mutex=0x9547e08) at forward.c:139
#3  0x00bb06a9 in
decaf::internal::util::concurrent::ConditionImpl::wait
(condition=0x9547e38) at
decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
#4  0x00c21353 in decaf::util::concurrent::Mutex::wait
(this=0x9547d6c) at decaf/util/concurrent/Mutex.cpp:95
#5  0x0094c9c5 in activemq::threads::CompositeTaskRunner::run
(this=0x9547d48) at activemq/threads/CompositeTaskRunner.cpp:118
#6  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9547e90) at decaf/lang/Thread.cpp:135
#7  0x00be2847 in threadWorker (arg=0x9547e90) at decaf/lang/Thread.cpp:188
#8  0x0046796e in start_thread (arg=0xb7799b70) at pthread_create.c:300
#9  0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 3:
#0  0x00eed422 in __kernel_vsyscall ()
#1  0x0046eaf9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x00469f61 in __pthread_mutex_lock (mutex=0x95523a8) at
pthread_mutex_lock.c:61
#4  0x00325ba6 in pthread_mutex_lock (mutex=0x95523a8) at forward.c:182
#5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x95523a8) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x00c21412 in decaf::util::concurrent::Mutex::lock
(this=0x95522bc) at decaf/util/concurrent/Mutex.cpp:75
#7  0x00915770 in
decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
(this=0x95522a8)
    at ./decaf/util/StlQueue.h:253
#8  activemq::core::MessageDispatchChannel::lock (this=0x95522a8) at
activemq/core/MessageDispatchChannel.h:153
#9  0x00c20ab5 in decaf::util::concurrent::Lock::lock
(this=0xb6f97fe4) at decaf/util/concurrent/Lock.cpp:54
#10 0x00c20c08 in Lock (this=0xfffffe00, object=0x95523a8,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#11 0x008de102 in
activemq::core::ActiveMQConsumer::clearMessagesInProgress
(this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:1112
#12 0x008ff15c in
activemq::core::ActiveMQSession::clearMessagesInProgress
(this=0x9551b90) at activemq/core/ActiveMQSession.cpp:239
#13 0x008c8543 in
activemq::core::ActiveMQConnection::transportInterrupted
(this=0x9548210) at activemq/core/ActiveMQConnection.cpp:704
#14 0x00953b34 in
activemq::transport::TransportFilter::transportInterrupted
(this=0x95481c0) at activemq/transport/TransportFilter.cpp:67
#15 0x009687ae in
activemq::transport::failover::FailoverTransport::handleTransportFailure
(this=0x9547a38, error=...) at
activemq/transport/failover/FailoverTransport.cpp:476
#16 0x0097406c in
activemq::transport::failover::FailoverTransportListener::onException
(this=0x9547878, ex=...) at
activemq/transport/failover/FailoverTransportListener.cpp:97
#17 0x00953c0b in activemq::transport::TransportFilter::fire
(this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:49
#18 0x00953c64 in activemq::transport::TransportFilter::onException
(this=0x9549d80, ex=...) at activemq/transport/TransportFilter.cpp:41
#19 0x00953c0b in activemq::transport::TransportFilter::fire
(this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:49
#20 0x00953c64 in activemq::transport::TransportFilter::onException
(this=0x95497f0, ex=...) at activemq/transport/TransportFilter.cpp:41
#21 0x00951b13 in activemq::transport::IOTransport::fire
(this=0x95497a0, ex=...) at activemq/transport/IOTransport.cpp:73
#22 0x009523bf in activemq::transport::IOTransport::run
(this=0x95497a0) at activemq/transport/IOTransport.cpp:246
#23 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9549ea0) at decaf/lang/Thread.cpp:135
#24 0x00be2847 in threadWorker (arg=0x9549ea0) at decaf/lang/Thread.cpp:188
#25 0x0046796e in start_thread (arg=0xb6f98b70) at pthread_create.c:300
#26 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 4:
#0  0x00eed422 in __kernel_vsyscall ()
#1  0x0046c015 in pthread_cond_wait@@GLIBC_2.3.2 () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
#2  0x003259dd in __pthread_cond_wait (cond=0x95516c8,
mutex=0x9551698) at forward.c:139
#3  0x00bb06a9 in
decaf::internal::util::concurrent::ConditionImpl::wait
(condition=0x95516c8) at
decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
#4  0x00c21353 in decaf::util::concurrent::Mutex::wait
(this=0x9551584) at decaf/util/concurrent/Mutex.cpp:95
#5  0x00c2029a in decaf::util::concurrent::CountDownLatch::await
(this=0x955157c) at decaf/util/concurrent/CountDownLatch.cpp:56
#6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x9551574,
timeOut=-1) at main.cpp:226
#7  0x0804bb46 in Consumer::run (this=0x9551570) at main.cpp:379
#8  0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x95515b0) at decaf/lang/Thread.cpp:135
#9  0x00be2847 in threadWorker (arg=0x95515b0) at decaf/lang/Thread.cpp:188
#10 0x0046796e in start_thread (arg=0xb6797b70) at pthread_create.c:300
#11 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 5:
#0  0x00eed422 in __kernel_vsyscall ()
#1  0x0046eaf9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x0046a13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x00469f61 in __pthread_mutex_lock (mutex=0x95475e0) at
pthread_mutex_lock.c:61
#4  0x00325ba6 in pthread_mutex_lock (mutex=0x95475e0) at forward.c:182
#5  0x00bb0aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x95475e0) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x00c21412 in decaf::util::concurrent::Mutex::lock
(this=0x9547a88) at decaf/util/concurrent/Mutex.cpp:75
#7  0x00c20ab5 in decaf::util::concurrent::Lock::lock
(this=0xb5f95ce0) at decaf/util/concurrent/Lock.cpp:54
#8  0x00c20c08 in Lock (this=0xfffffe00, object=0x95475e0,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#9  0x0096b8fd in
activemq::transport::failover::FailoverTransport::oneway
(this=0x9547a38, command=...) at
activemq/transport/failover/FailoverTransport.cpp:186
#10 0x00957f9f in
activemq::transport::correlator::ResponseCorrelator::oneway
(this=0x95481c0, command=...) at
activemq/transport/correlator/ResponseCorrelator.cpp:82
#11 0x008c13be in activemq::core::ActiveMQConnection::oneway
(this=0x9548210, command=...) at
activemq/core/ActiveMQConnection.cpp:741
#12 0x00900a4f in activemq::core::ActiveMQSession::oneway
(this=0x9551b90, command=...) at activemq/core/ActiveMQSession.cpp:903
#13 0x008e5a1c in activemq::core::ActiveMQConsumer::acknowledge
(this=0x9552270) at activemq/core/ActiveMQConsumer.cpp:860
#14 0x008ef885 in
activemq::core::TransactionSynhcronization::beforeEnd (this=0x95529e8)
at activemq/core/ActiveMQConsumer.cpp:85
#15 0x00910588 in
activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x9551dc8)
at activemq/core/ActiveMQTransactionContext.cpp:192
#16 0x00910d63 in activemq::core::ActiveMQTransactionContext::commit
(this=0x9551dc8) at activemq/core/ActiveMQTransactionContext.cpp:127
#17 0x008fe460 in activemq::core::ActiveMQSession::commit
(this=0x9551b90) at activemq/core/ActiveMQSession.cpp:189
#18 0x007ec1a0 in activemq::cmsutil::PooledSession::commit
(this=0x9551f38) at activemq/cmsutil/PooledSession.h:87
#19 0x0804c0fb in Consumer::onMessage (this=0x9551570,
message=0x9552ec0) at main.cpp:455
#20 0x008e972f in activemq::core::ActiveMQConsumer::dispatch
(this=0x9552270, dispatch=...) at
activemq/core/ActiveMQConsumer.cpp:1018
#21 0x0090d5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
(this=0x9551e78, dispatch=...) at
activemq/core/ActiveMQSessionExecutor.cpp:129
#22 0x0090d993 in activemq::core::ActiveMQSessionExecutor::iterate
(this=0x9551e78) at activemq/core/ActiveMQSessionExecutor.cpp:166
#23 0x009507f3 in activemq::threads::DedicatedTaskRunner::run
(this=0x9552550) at activemq/threads/DedicatedTaskRunner.cpp:111
#24 0x00be3cf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9551ad8) at decaf/lang/Thread.cpp:135
#25 0x00be2847 in threadWorker (arg=0x9551ad8) at decaf/lang/Thread.cpp:188
#26 0x0046796e in start_thread (arg=0xb5f96b70) at pthread_create.c:300
#27 0x00318a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130



BACKTRACE 2:

Thread 1:
#0  0x0079b422 in __kernel_vsyscall ()
#1  0x008bfb5d in pthread_join (threadid=3061021552,
thread_return=0xbfbff07c) at pthread_join.c:89
#2  0x0132a3c0 in decaf::lang::Thread::join (this=0x9be0574) at
decaf/lang/Thread.cpp:421
#3  0x0804ca15 in AppTest::joinEndPoins (this=0xbfbff104) at main.cpp:587
#4  0x0804c69c in AppTest::run (this=0xbfbff104) at main.cpp:556
#5  0x0804a33c in main (argc=4, argv=0xbfbff224) at main.cpp:634

Thread 2:
#0  0x0079b422 in __kernel_vsyscall ()
#1  0x008c3015 in pthread_cond_wait@@GLIBC_2.3.2 () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
#2  0x004ed9dd in __pthread_cond_wait (cond=0x9bd6e38,
mutex=0x9bd6e08) at forward.c:139
#3  0x012f86a9 in
decaf::internal::util::concurrent::ConditionImpl::wait
(condition=0x9bd6e38) at
decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
#4  0x01369353 in decaf::util::concurrent::Mutex::wait
(this=0x9bd6d6c) at decaf/util/concurrent/Mutex.cpp:95
#5  0x010949c5 in activemq::threads::CompositeTaskRunner::run
(this=0x9bd6d48) at activemq/threads/CompositeTaskRunner.cpp:118
#6  0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9bd6e90) at decaf/lang/Thread.cpp:135
#7  0x0132a847 in threadWorker (arg=0x9bd6e90) at decaf/lang/Thread.cpp:188
#8  0x008be96e in start_thread (arg=0xb7739b70) at pthread_create.c:300
#9  0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 3:
#0  0x0079b422 in __kernel_vsyscall ()
#1  0x008c5af9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
pthread_mutex_lock.c:61
#4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
#5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x9be12d8) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x01369412 in decaf::util::concurrent::Mutex::lock
(this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
#7  0x0105d770 in
decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
(this=0x9be11d8)
    at ./decaf/util/StlQueue.h:253
#8  activemq::core::MessageDispatchChannel::lock (this=0x9be11d8) at
activemq/core/MessageDispatchChannel.h:153
#9  0x01368ab5 in decaf::util::concurrent::Lock::lock
(this=0xb6f37fe4) at decaf/util/concurrent/Lock.cpp:54
#10 0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#11 0x01026102 in
activemq::core::ActiveMQConsumer::clearMessagesInProgress
(this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:1112
#12 0x0104715c in
activemq::core::ActiveMQSession::clearMessagesInProgress
(this=0x9be0b90) at activemq/core/ActiveMQSession.cpp:239
#13 0x01010543 in
activemq::core::ActiveMQConnection::transportInterrupted
(this=0x9bd7210) at activemq/core/ActiveMQConnection.cpp:704
#14 0x0109bb34 in
activemq::transport::TransportFilter::transportInterrupted
(this=0x9bd71c0) at activemq/transport/TransportFilter.cpp:67
#15 0x010b07ae in
activemq::transport::failover::FailoverTransport::handleTransportFailure
(this=0x9bd6a38, error=...) at
activemq/transport/failover/FailoverTransport.cpp:476
#16 0x010bc06c in
activemq::transport::failover::FailoverTransportListener::onException
(this=0x9bd6878, ex=...) at
activemq/transport/failover/FailoverTransportListener.cpp:97
#17 0x0109bc0b in activemq::transport::TransportFilter::fire
(this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:49
#18 0x0109bc64 in activemq::transport::TransportFilter::onException
(this=0x9bd8ec8, ex=...) at activemq/transport/TransportFilter.cpp:41
#19 0x0109bc0b in activemq::transport::TransportFilter::fire
(this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:49
#20 0x0109bc64 in activemq::transport::TransportFilter::onException
(this=0x9bd8938, ex=...) at activemq/transport/TransportFilter.cpp:41
#21 0x01099b13 in activemq::transport::IOTransport::fire
(this=0x9bd88e8, ex=...) at activemq/transport/IOTransport.cpp:73
#22 0x0109a3bf in activemq::transport::IOTransport::run
(this=0x9bd88e8) at activemq/transport/IOTransport.cpp:246
#23 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9bd8fe8) at decaf/lang/Thread.cpp:135
#24 0x0132a847 in threadWorker (arg=0x9bd8fe8) at decaf/lang/Thread.cpp:188
#25 0x008be96e in start_thread (arg=0xb6f38b70) at pthread_create.c:300
#26 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 4:
#0  0x0079b422 in __kernel_vsyscall ()
#1  0x008c5af9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9be12d8) at
pthread_mutex_lock.c:61
#4  0x004edba6 in pthread_mutex_lock (mutex=0x9be12d8) at forward.c:182
#5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x9be12d8) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x01369412 in decaf::util::concurrent::Mutex::lock
(this=0x9be11ec) at decaf/util/concurrent/Mutex.cpp:75
#7  0x01036830 in
decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
(this=0x9be11e0)
    at ./decaf/util/StlQueue.h:253
#8  0x01368ab5 in decaf::util::concurrent::Lock::lock
(this=0xb6737064) at decaf/util/concurrent/Lock.cpp:54
#9  0x01368c08 in Lock (this=0xfffffe00, object=0x9be12d8,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#10 0x0105c507 in activemq::core::MessageDispatchChannel::clear
(this=0x9be11d8) at activemq/core/MessageDispatchChannel.cpp:143
#11 0x0102ec37 in activemq::core::ActiveMQConsumer::doClose
(this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:335
#12 0x0102fb00 in activemq::core::ActiveMQConsumer::close
(this=0x9be11a0) at activemq/core/ActiveMQConsumer.cpp:309
#13 0x01030390 in ~ActiveMQConsumer (this=0x9be11a0, __in_chrg=<value
optimized out>) at activemq/core/ActiveMQConsumer.cpp:261
#14 0x0804c276 in Consumer::cleanup (this=0x9be0570) at main.cpp:469
#15 0x0804bcdd in Consumer::run (this=0x9be0570) at main.cpp:394
#16 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9be05b0) at decaf/lang/Thread.cpp:135
#17 0x0132a847 in threadWorker (arg=0x9be05b0) at decaf/lang/Thread.cpp:188
#18 0x008be96e in start_thread (arg=0xb6737b70) at pthread_create.c:300
#19 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

Thread 5:
#0  0x0079b422 in __kernel_vsyscall ()
#1  0x008c5af9 in __lll_lock_wait () at
../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
#2  0x008c113b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
#3  0x008c0f61 in __pthread_mutex_lock (mutex=0x9bd65e0) at
pthread_mutex_lock.c:61
#4  0x004edba6 in pthread_mutex_lock (mutex=0x9bd65e0) at forward.c:182
#5  0x012f8aa1 in decaf::internal::util::concurrent::MutexImpl::lock
(handle=0x9bd65e0) at
decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
#6  0x01369412 in decaf::util::concurrent::Mutex::lock
(this=0x9bd6a88) at decaf/util/concurrent/Mutex.cpp:75
#7  0x01368ab5 in decaf::util::concurrent::Lock::lock
(this=0xb5f35ea0) at decaf/util/concurrent/Lock.cpp:54
#8  0x01368c08 in Lock (this=0xfffffe00, object=0x9bd65e0,
intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
#9  0x010b38fd in
activemq::transport::failover::FailoverTransport::oneway
(this=0x9bd6a38, command=...) at
activemq/transport/failover/FailoverTransport.cpp:186
#10 0x0109ff9f in
activemq::transport::correlator::ResponseCorrelator::oneway
(this=0x9bd71c0, command=...) at
activemq/transport/correlator/ResponseCorrelator.cpp:82
#11 0x010093be in activemq::core::ActiveMQConnection::oneway
(this=0x9bd7210, command=...) at
activemq/core/ActiveMQConnection.cpp:741
#12 0x01048a4f in activemq::core::ActiveMQSession::oneway
(this=0x9be0b90, command=...) at activemq/core/ActiveMQSession.cpp:903
#13 0x01030df4 in
activemq::core::ActiveMQConsumer::afterMessageIsConsumed
(this=0x9be11a0, message=..., messageExpired=false) at
activemq/core/ActiveMQConsumer.cpp:642
#14 0x01031743 in activemq::core::ActiveMQConsumer::dispatch
(this=0x9be11a0, dispatch=...) at
activemq/core/ActiveMQConsumer.cpp:1021
#15 0x010555c3 in activemq::core::ActiveMQSessionExecutor::dispatch
(this=0x9be0dc8, dispatch=...) at
activemq/core/ActiveMQSessionExecutor.cpp:129
#16 0x01055993 in activemq::core::ActiveMQSessionExecutor::iterate
(this=0x9be0dc8) at activemq/core/ActiveMQSessionExecutor.cpp:166
#17 0x010987f3 in activemq::threads::DedicatedTaskRunner::run
(this=0x9be1480) at activemq/threads/DedicatedTaskRunner.cpp:111
#18 0x0132bcf1 in decaf::lang::ThreadProperties::runCallback
(properties=0x9be0ad8) at decaf/lang/Thread.cpp:135
#19 0x0132a847 in threadWorker (arg=0x9be0ad8) at decaf/lang/Thread.cpp:188
#20 0x008be96e in start_thread (arg=0xb5f36b70) at pthread_create.c:300
#21 0x004e0a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130

2011/3/1 Oscar Pernas <os...@pernas.es>
>
> Hi,
>
> I ran into this bug before. If I am shutting down and starting the broker so
> many times, connection are not reestablished and the consumer gets frozen.
> If you try to close the consumer that is frozen, the closing stucks your
> thread too? In my case, this problem only happens when I'm debugging the
> application (without breaks but with visual studio debug).
> I've partially resolved it manually, if I dont received anything over a
> topic in a time, I try to close the consumer and reconnect it again.
>
> regards
>
> --
> Óscar Pernas Plaza.

Re: AMQCPP: unexpected exception

Posted by Oscar Pernas <os...@pernas.es>.

Hi,

I ran into this bug before. If I am shutting down and starting the broker so
many times, connection are not reestablished and the consumer gets frozen.
If you try to close the consumer that is frozen, the closing stucks your
thread too? In my case, this problem only happens when I'm debugging the
application (without breaks but with visual studio debug).
I've partially resolved it manually, if I dont received anything over a
topic in a time, I try to close the consumer and reconnect it again.

regards

2011/3/2 Timothy Bish <ta...@gmail.com>

> On Tue, 2011-03-01 at 20:22 -0300, Henrique Magarotto wrote:
> > Hi everybody,
> >
> > First, sorry for long mail and thanks for great job in ActiveMQ and
> > ActiveMQ-CPP.
> >
> > I'm testing some failure cases with ActiveMQ and ActiveMQ-CPP.
> >
> > In my test, an unexpected exception is thrown if broker is stopped
> > while consumer is in transaction.
> >
> > Environment:
> > ActiveMQ 5.4.2
> > ActiveMQ-CPP 3.2.4
> > Ubuntu 10.04
> > JRE 1.6.0_20-b02
> >
>
> > I get the following message:
> >
> > terminate called after throwing an instance of 'cms::CMSException'
> >   what():  Failover timeout of 1000 ms reached.
> > Aborted (core dumped)
> >
>
> Sounds like a bug, can you open a new issue in are Jira bug tracker and
> attach your test case along with instructions for reproducing the error?
>
> https://issues.apache.org/jira/browse/AMQCPP
>
> Regards
> Tim.
>
> > core backtrace:
> > #0  0x001ef422 in __kernel_vsyscall ()
> > #1  0x00c5a651 in *__GI_raise (sig=6) at
> > ../nptl/sysdeps/unix/sysv/linux/raise.c:64
> > #2  0x00c5da82 in *__GI_abort () at abort.c:92
> > #3  0x00bf952f in __gnu_cxx::__verbose_terminate_handler() () from
> > /usr/lib/libstdc++.so.6
> > #4  0x00bf7465 in ?? () from /usr/lib/libstdc++.so.6
> > #5  0x00bf74a2 in std::terminate() () from /usr/lib/libstdc++.so.6
> > #6  0x00bf74c5 in ?? () from /usr/lib/libstdc++.so.6
> > #7  0x00bf6915 in __cxa_call_unexpected () from /usr/lib/libstdc++.so.6
> > #8  0x0052f8ae in
> > activemq::core::TransactionSynhcronization::beforeEnd (this=0x93ac548)
> > at activemq/core/ActiveMQConsumer.cpp:84
> > #9  0x00550588 in
> > activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x93b3e48)
> > at activemq/core/ActiveMQTransactionContext.cpp:192
> > #10 0x00550d63 in activemq::core::ActiveMQTransactionContext::commit
> > (this=0x93b3e48) at activemq/core/ActiveMQTransactionContext.cpp:127
> > #11 0x0053e460 in activemq::core::ActiveMQSession::commit
> > (this=0x93b3c10) at activemq/core/ActiveMQSession.cpp:189
> > #12 0x0042c1a0 in activemq::cmsutil::PooledSession::commit
> > (this=0x93b3fd8) at activemq/cmsutil/PooledSession.h:87
> > #13 0x0804c0fb in Consumer::onMessage (this=0x93a8c30,
> > message=0x93b49b0) at main.cpp:455
> > #14 0x0804bdc8 in Consumer::consumeLoop (this=0x93a8c30) at main.cpp:415
> > #15 0x0804bb56 in Consumer::run (this=0x93a8c30) at main.cpp:381
> > #16 0x00823cf1 in decaf::lang::ThreadProperties::runCallback
> > (properties=0x93abff0) at decaf/lang/Thread.cpp:135
> > #17 0x00822847 in threadWorker (arg=0x93abff0) at
> decaf/lang/Thread.cpp:188
> > #18 0x0015a96e in start_thread (arg=0xb6f8bb70) at pthread_create.c:300
> > #19 0x00cfda4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> >
> >
> > Exception specification for method
> > 'activemq::core::TransactionSynhcronization::beforeEnd' says: 'throw(
> > exceptions::ActiveMQException )' but 'ActiveMQConsumer::acknowledge'
> > throws 'cms::CMSException', so 'std::unexpected' is called and 'abort'
> > occurs. (
> http://www.linuxprogrammingblog.com/cpp-exception-specifications-are-evil)
> >
> >
> > When 'MessageListener' is used, the commit freezes, probability, in
> thread 5:
> >
> > Thread 1:
> > #0  0x00cfd422 in __kernel_vsyscall ()
> > #1  0x00ddcb5d in pthread_join (threadid=3061869424,
> > thread_return=0xbf964bdc) at pthread_join.c:89
> > #2  0x00a923c0 in decaf::lang::Thread::join (this=0x8eba574) at
> > decaf/lang/Thread.cpp:421
> > #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf964c64) at main.cpp:586
> > #4  0x0804c678 in AppTest::run (this=0xbf964c64) at main.cpp:555
> > #5  0x0804a33c in main (argc=4, argv=0xbf964d84) at main.cpp:633
> >
> > Thread 2:
> > #0  0x00cfd422 in __kernel_vsyscall ()
> > #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> > ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> > #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eb0e38,
> > mutex=0x8eb0e08) at forward.c:139
> > #3  0x00a606a9 in
> > decaf::internal::util::concurrent::ConditionImpl::wait
> > (condition=0x8eb0e38) at
> > decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> > #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> > (this=0x8eb0d6c) at decaf/util/concurrent/Mutex.cpp:95
> > #5  0x007fc9c5 in activemq::threads::CompositeTaskRunner::run
> > (this=0x8eb0d48) at activemq/threads/CompositeTaskRunner.cpp:118
> > #6  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> > (properties=0x8eb0e90) at decaf/lang/Thread.cpp:135
> > #7  0x00a92847 in threadWorker (arg=0x8eb0e90) at
> decaf/lang/Thread.cpp:188
> > #8  0x00ddb96e in start_thread (arg=0xb7808b70) at pthread_create.c:300
> > #9  0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> >
> > Thread 3:
> > #0  0x00cfd422 in __kernel_vsyscall ()
> > #1  0x00de2af9 in __lll_lock_wait () at
> > ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> > #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> > #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8ebb3a8) at
> > pthread_mutex_lock.c:61
> > #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8ebb3a8) at forward.c:182
> > #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> > (handle=0x8ebb3a8) at
> > decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> > #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> > (this=0x8ebb2bc) at decaf/util/concurrent/Mutex.cpp:75
> > #7  0x007c5770 in
> >
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> > decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> > (this=0x8ebb2a8)
> >     at ./decaf/util/StlQueue.h:253
> > #8  activemq::core::MessageDispatchChannel::lock (this=0x8ebb2a8) at
> > activemq/core/MessageDispatchChannel.h:153
> > #9  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> > (this=0xb7006fe4) at decaf/util/concurrent/Lock.cpp:54
> > #10 0x00ad0c08 in Lock (this=0xfffffe00, object=0x8ebb3a8,
> > intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> > #11 0x0078e102 in
> > activemq::core::ActiveMQConsumer::clearMessagesInProgress
> > (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:1112
> > #12 0x007af15c in
> > activemq::core::ActiveMQSession::clearMessagesInProgress
> > (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:239
> > #13 0x00778543 in
> > activemq::core::ActiveMQConnection::transportInterrupted
> > (this=0x8eb1210) at activemq/core/ActiveMQConnection.cpp:704
> > #14 0x00803b34 in
> > activemq::transport::TransportFilter::transportInterrupted
> > (this=0x8eb11c0) at activemq/transport/TransportFilter.cpp:67
> > #15 0x008187ae in
> > activemq::transport::failover::FailoverTransport::handleTransportFailure
> > (this=0x8eb0a38, error=...) at
> > activemq/transport/failover/FailoverTransport.cpp:476
> > #16 0x0082406c in
> > activemq::transport::failover::FailoverTransportListener::onException
> > (this=0x8eb0878, ex=...) at
> > activemq/transport/failover/FailoverTransportListener.cpp:97
> > #17 0x00803c0b in activemq::transport::TransportFilter::fire
> > (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:49
> > #18 0x00803c64 in activemq::transport::TransportFilter::onException
> > (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:41
> > #19 0x00803c0b in activemq::transport::TransportFilter::fire
> > (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:49
> > #20 0x00803c64 in activemq::transport::TransportFilter::onException
> > (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:41
> > #21 0x00801b13 in activemq::transport::IOTransport::fire
> > (this=0x8eb27a0, ex=...) at activemq/transport/IOTransport.cpp:73
> > #22 0x008023bf in activemq::transport::IOTransport::run
> > (this=0x8eb27a0) at activemq/transport/IOTransport.cpp:246
> > #23 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> > (properties=0x8eb2ea0) at decaf/lang/Thread.cpp:135
> > #24 0x00a92847 in threadWorker (arg=0x8eb2ea0) at
> decaf/lang/Thread.cpp:188
> > #25 0x00ddb96e in start_thread (arg=0xb7007b70) at pthread_create.c:300
> > #26 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> >
> > Thread 4:
> > #0  0x00cfd422 in __kernel_vsyscall ()
> > #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> > ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> > #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eba6c8,
> > mutex=0x8eba698) at forward.c:139
> > #3  0x00a606a9 in
> > decaf::internal::util::concurrent::ConditionImpl::wait
> > (condition=0x8eba6c8) at
> > decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> > #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> > (this=0x8eba584) at decaf/util/concurrent/Mutex.cpp:95
> > #5  0x00ad029a in decaf::util::concurrent::CountDownLatch::await
> > (this=0x8eba57c) at decaf/util/concurrent/CountDownLatch.cpp:56
> > #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x8eba574,
> > timeOut=-1) at main.cpp:226
> > #7  0x0804bb46 in Consumer::run (this=0x8eba570) at main.cpp:379
> > #8  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> > (properties=0x8eba5b0) at decaf/lang/Thread.cpp:135
> > #9  0x00a92847 in threadWorker (arg=0x8eba5b0) at
> decaf/lang/Thread.cpp:188
> > #10 0x00ddb96e in start_thread (arg=0xb6806b70) at pthread_create.c:300
> > #11 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> >
> > Thread 5:
> > #0  0x00cfd422 in __kernel_vsyscall ()
> > #1  0x00de2af9 in __lll_lock_wait () at
> > ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> > #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> > #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8eb05e0) at
> > pthread_mutex_lock.c:61
> > #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8eb05e0) at forward.c:182
> > #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> > (handle=0x8eb05e0) at
> > decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> > #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> > (this=0x8eb0a88) at decaf/util/concurrent/Mutex.cpp:75
> > #7  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> > (this=0xb6004ce0) at decaf/util/concurrent/Lock.cpp:54
> > #8  0x00ad0c08 in Lock (this=0xfffffe00, object=0x8eb05e0,
> > intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> > #9  0x0081b8fd in
> > activemq::transport::failover::FailoverTransport::oneway
> > (this=0x8eb0a38, command=...) at
> > activemq/transport/failover/FailoverTransport.cpp:186
> > #10 0x00807f9f in
> > activemq::transport::correlator::ResponseCorrelator::oneway
> > (this=0x8eb11c0, command=...) at
> > activemq/transport/correlator/ResponseCorrelator.cpp:82
> > #11 0x007713be in activemq::core::ActiveMQConnection::oneway
> > (this=0x8eb1210, command=...) at
> > activemq/core/ActiveMQConnection.cpp:741
> > #12 0x007b0a4f in activemq::core::ActiveMQSession::oneway
> > (this=0x8ebab90, command=...) at activemq/core/ActiveMQSession.cpp:903
> > #13 0x00795a1c in activemq::core::ActiveMQConsumer::acknowledge
> > (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:860
> > #14 0x0079f885 in
> > activemq::core::TransactionSynhcronization::beforeEnd (this=0x8ebb9e8)
> > at activemq/core/ActiveMQConsumer.cpp:85
> > #15 0x007c0588 in
> > activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x8ebadc8)
> > at activemq/core/ActiveMQTransactionContext.cpp:192
> > #16 0x007c0d63 in activemq::core::ActiveMQTransactionContext::commit
> > (this=0x8ebadc8) at activemq/core/ActiveMQTransactionContext.cpp:127
> > #17 0x007ae460 in activemq::core::ActiveMQSession::commit
> > (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:189
> > #18 0x0069c1a0 in activemq::cmsutil::PooledSession::commit
> > (this=0x8ebaf38) at activemq/cmsutil/PooledSession.h:87
> > #19 0x0804c0fb in Consumer::onMessage (this=0x8eba570,
> > message=0x8ebbec0) at main.cpp:455
> > #20 0x0079972f in activemq::core::ActiveMQConsumer::dispatch
> > (this=0x8ebb270, dispatch=...) at
> > activemq/core/ActiveMQConsumer.cpp:1018
> > #21 0x007bd5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
> > (this=0x8ebae78, dispatch=...) at
> > activemq/core/ActiveMQSessionExecutor.cpp:129
> > #22 0x007bd993 in activemq::core::ActiveMQSessionExecutor::iterate
> > (this=0x8ebae78) at activemq/core/ActiveMQSessionExecutor.cpp:166
> > #23 0x008007f3 in activemq::threads::DedicatedTaskRunner::run
> > (this=0x8ebb550) at activemq/threads/DedicatedTaskRunner.cpp:111
> > #24 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> > (properties=0x8ebaad8) at decaf/lang/Thread.cpp:135
> > #25 0x00a92847 in threadWorker (arg=0x8ebaad8) at
> decaf/lang/Thread.cpp:188
> > #26 0x00ddb96e in start_thread (arg=0xb6005b70) at pthread_create.c:300
> > #27 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> >
> >
> >
> > I tried to resolve this issue changing configurations and code, but i
> > always get an error or client freezes.
> > Is it a bug? Any suggestions?
> >
> > Thanks.
> > Henrique
> >
> > How to reproduce:
> > 1. Start ActiveMQ
> > 2. Send a massage './activemqTest -p1 -n1'
> > 3. Run consumer './activemqTest -c1 -d10000'
> > 4. Stop activemq when 'Starting delay...' message appears.
> > 5. After activemq is down and delay is finish, consumer try commit
> > (message 'Try commit') and unexpected exception occurs.
> >
> > Following test code, based on ActiveMQ-CPP sample
> > 'activemq-cpp-library-3.2.4/src/examples/main.cpp':
> >
> >
> > /*
> >  * Licensed to the Apache Software Foundation (ASF) under one or more
> >  * contributor license agreements.  See the NOTICE file distributed with
> >  * this work for additional information regarding copyright ownership.
> >  * The ASF licenses this file to You under the Apache License, Version
> 2.0
> >  * (the "License"); you may not use this file except in compliance with
> >  * the License.  You may obtain a copy of the License at
> >  *
> >  *     http://www.apache.org/licenses/LICENSE-2.0
> >  *
> >  * Unless required by applicable law or agreed to in writing, software
> >  * distributed under the License is distributed on an "AS IS" BASIS,
> >  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
> implied.
> >  * See the License for the specific language governing permissions and
> >  * limitations under the License.
> >  */
> >
> > // START SNIPPET: demo
> >
> > #include <activemq/library/ActiveMQCPP.h>
> > #include <activemq/cmsutil/SessionPool.h>
> > #include <activemq/cmsutil/PooledSession.h>
> > #include <activemq/cmsutil/ResourceLifecycleManager.h>
> > #include <activemq/exceptions/ActiveMQException.h>
> > #include <decaf/lang/Thread.h>
> > #include <decaf/lang/Runnable.h>
> > #include <decaf/util/concurrent/CountDownLatch.h>
> > #include <decaf/util/concurrent/Mutex.h>
> > #include <decaf/lang/Integer.h>
> > #include <decaf/lang/Long.h>
> > #include <decaf/lang/System.h>
> > #include <activemq/core/ActiveMQConnectionFactory.h>
> > #include <activemq/util/Config.h>
> > #include <cms/Connection.h>
> > #include <cms/Session.h>
> > #include <cms/TextMessage.h>
> > #include <cms/BytesMessage.h>
> > #include <cms/MapMessage.h>
> > #include <cms/ExceptionListener.h>
> > #include <cms/MessageListener.h>
> > #include <iostream>
> > #include <memory>
> > #include <map>
> > #include <vector>
> > #include <unistd.h>
> >
> > using namespace activemq::core;
> > using namespace activemq::cmsutil;
> > using namespace decaf::util::concurrent;
> > using namespace decaf::util;
> > using namespace decaf::lang;
> > using namespace cms;
> >
> >
> > class ConnPoolManager : protected decaf::util::concurrent::Mutex,
> >                         public ExceptionListener {
> > private:
> >     typedef std::map< int, activemq::cmsutil::SessionPool* >
> tSessionPoolMap;
> >
> >     std::string _brokerURI;
> >     ConnectionFactory* _connectionFactory;
> >     Connection* _connection;
> >     ResourceLifecycleManager* _lifecycleManager;
> >     tSessionPoolMap _pools;
> >     bool _started;
> >
> > public:
> >     ConnPoolManager() :
> >             _brokerURI("tcp://localhost:61616"),
> >             _connectionFactory(NULL),
> >             _connection(NULL),
> >             _lifecycleManager(NULL),
> >             _started(false)
> >     {
> >     }
> >
> >     virtual ~ConnPoolManager() {
> >         cleanup();
> >     }
> >
> >     void setBrokerURI(const std::string& uri) {
> >         _brokerURI = uri;
> >     }
> >
> >     bool startPools() {
> >         synchronized(this) {
> >             if(_started) return true;
> >             try {
> >                 createConnection();
> >                 createPools();
> >                 _started=true;
> >                 return true;
> >             } catch(CMSException& e) {
> >                 std::cout<<"(ConnPoolManager::startPools)
> > CMSException: " << e.getMessage() << std::endl;
> >             } catch(...) {
> >                 std::cout<<"(ConnPoolManager::startPools) Unknown
> > exception!!!" << std::endl;
> >             }
> >         }
> >         cleanup();
> >         return false;
> >     }
> >
> >     PooledSession* getSession(cms::Session::AcknowledgeMode ack) {
> >         if(_started) {
> >             return _pools[ack]->takeSession();
> >         } else if(startPools()) {
> >             return _pools[ack]->takeSession();
> >         } else {
> >             return NULL;
> >         }
> >     }
> >
> > private:
> >
> >     void createConnection(){
> >         // Create a ConnectionFactory
> >         _connectionFactory =
> > ConnectionFactory::createCMSConnectionFactory( _brokerURI );
> >
> >         // Create a Connection
> >         _connection = _connectionFactory->createConnection();
> >         _connection->start();
> >         _connection->setExceptionListener(this);
> >     }
> >
> >     void createPools(){
> >         // Create lifecycle manager
> >         _lifecycleManager = new
> activemq::cmsutil::ResourceLifecycleManager();
> >
> >         // Create pools, one for each acknowledge type
> >         for(    int ack = cms::Session::AUTO_ACKNOWLEDGE;
> >                 ack <= cms::Session::INDIVIDUAL_ACKNOWLEDGE;
> >                 ack++)
> >         {
> >             _pools[ack] =  new activemq::cmsutil::SessionPool(
> >                     _connection,
> >                     static_cast<cms::Session::AcknowledgeMode>(ack),
> >                     _lifecycleManager);
> >         }
> >     }
> >
> >     void cleanup() {
> >         synchronized(this) {
> >             releasePools();
> >             if(_lifecycleManager!=NULL) {
> >                 try { delete _lifecycleManager;
> >                 } catch(...) {}
> >             } _lifecycleManager = NULL;
> >
> >             if(_connection!=NULL) {
> >                 try { delete _connection;
> >                 } catch(...) {}
> >             } _connection = NULL;
> >             if(_connectionFactory!=NULL) {
> >                 try { delete _connectionFactory;
> >                 } catch(...) {}
> >             } _connectionFactory = NULL;
> >             _started=false;
> >         }
> >     }
> >
> >     void releasePools() {
> >         for(tSessionPoolMap::iterator it=_pools.begin();
> > it!=_pools.end(); it++) {
> >             try {
> >                 delete it->second;
> >             } catch(...) { }
> >         }
> >         _pools.clear();
> >     }
> >
> >     // If something bad happens you see it here as this class is also
> been
> >     // registered as an ExceptionListener with the connection.
> >     virtual void onException( const CMSException& ex AMQCPP_UNUSED) {
> >         std::cout<<"(ConnPoolManager::onException) CMS Exception
> > occurred.  Shutting down client."<<std::endl;
> >         ex.printStackTrace();
> >         exit(1);
> >     }
> >
> >
> >
> > };
> >
> > class JMSEndPointThread  : public Thread {
> > private:
> >     CountDownLatch _stopNow;
> >     ConnPoolManager* _pools;
> >     std::string _destinationString;
> >     PooledSession* _session;
> >     Destination* _destination;
> >     bool _sessionTransacted;
> >
> > protected:
> >
> >     JMSEndPointThread(  ConnPoolManager* pools,
> >                         const std::string& dest,
> >                         bool sessionTransacted) :
> >             _stopNow(1),
> >             _pools(pools),
> >             _destinationString(dest),
> >             _session(NULL),
> >             _destination(NULL),
> >             _sessionTransacted(sessionTransacted)
> >     {
> >     }
> >
> >     virtual bool prepareSession() {
> >         cleanup();
> >         // Create a Session
> >         if( _sessionTransacted ) {
> >             _session = _pools->getSession(Session::SESSION_TRANSACTED);
> >         } else {
> >             _session = _pools->getSession(Session::AUTO_ACKNOWLEDGE);
> >         }
> >         if(_session==NULL) return false;
> >
> >         // Create the Queue destination
> >         _destination = _session->createQueue( _destinationString );
> >         return true;
> >     }
> >
> >     PooledSession* getSession() { return _session; }
> >     Destination* getDestination() { return _destination; }
> >     bool getTransacted() { return _sessionTransacted; }
> >
> >     bool stopCheck(int timeOut=0) {
> >         if(timeOut>=0) return _stopNow.await(timeOut);
> >         else _stopNow.await();
> >         return true;
> >     }
> >
> >     virtual void cleanup() {
> >         // Destroy resources.
> >         try{
> >             if( _destination != NULL ) delete _destination;
> >         } catch(...) { }
> >         _destination = NULL;
> >
> >         // Back session to pool
> >         try{
> >             if( _session != NULL ) _session->close();
> >         } catch(...) { }
> >         _session = NULL;
> >     }
> >
> > public:
> >
> >     typedef std::vector<JMSEndPointThread*> List;
> >
> >     virtual ~JMSEndPointThread() {
> >         cleanup();
> >     }
> >
> >     void stopNow() {
> >         _stopNow.countDown();
> >     }
> >
> > };
> >
> > class Producer : public JMSEndPointThread {
> > private:
> >
> >     MessageProducer* _producer;
> >     int _delay;
> >     int _numMessages;
> >
> > public:
> >
> >     Producer(   ConnPoolManager* pools,
> >                 const std::string& dest,
> >                 int delay,
> >                 int numMessages,
> >                 bool sessionTransacted = false ) :
> >         JMSEndPointThread(pools,dest,sessionTransacted)
> >     {
> >         _producer = NULL;
> >         _delay = delay;
> >         _numMessages = numMessages;
> >     }
> >
> >     virtual ~Producer() {
> >         cleanup();
> >     }
> >
> >     virtual void run() {
> >         std::cout<<"Producer started!!!! Thread:
> "<<Thread::getId()<<std::endl;
> >         while(!stopCheck() && _numMessages) {
> >             try {
> >                 if(prepareProducer()) {
> >                     sendMessages();
> >                     continue;
> >                 }
> >             } catch(CMSException& e) {
> >                 std::cout<<"(Producer::run) CMSException: " <<
> > e.getMessage() << std::endl;
> >             } catch(...) {
> >                 std::cout<<"(Producer::run) Unknown exception!!!" <<
> std::endl;
> >             }
> >             cleanup();
> >             stopCheck(5000); // reconnect delay
> >         }
> >         std::cout<<"Producer end!!!! Thread:
> "<<Thread::getId()<<std::endl;
> >         cleanup();
> >     }
> >
> > private:
> >
> >     bool prepareProducer() {
> >         // Get Session
> >         if(!prepareSession()) return false;
> >         // Create a MessageProducer from the Session to the Queue
> >         _producer = getSession()->createProducer( getDestination() );
> >         _producer->setDeliveryMode( DeliveryMode::PERSISTENT );
> >         return true;
> >     }
> >     void sendMessages() {
> >         // Create a messages
> >         std::stringstream text;
> >         text<<"Hello world! from thread " << Thread::getId();
> >
> >         while(!stopCheck() && _numMessages) {
> >
> >             TextMessage* message = getSession()->createTextMessage(
> > text.str() );
> >             message->setIntProperty( "Integer", _numMessages );
> >
> >             // Tell the producer to send the message
> >             std::cout<<"Sent message #"<<_numMessages<<" from thread
> > "<<Thread::getId()<<std::endl;
> >             _producer->send( message );
> >             delete message;
> >             --_numMessages;
> >
> >             stopCheck(_delay);
> >             if(getTransacted()) getSession()->commit();
> >         }
> >     }
> >
> >     virtual void cleanup() {
> >         try{
> >             if( _producer != NULL ) delete _producer;
> >         } catch(...) { }
> >         _producer = NULL;
> >
> >         JMSEndPointThread::cleanup();
> >     }
> >
> > };
> >
> >
> > class Consumer :    public MessageListener,
> >                     public JMSEndPointThread {
> >
> > private:
> >
> >     MessageConsumer* _consumer;
> >     int _delay;
> >     bool _useListener;
> >
> > public:
> >
> >     Consumer(   ConnPoolManager* pools,
> >                 const std::string& dest,
> >                 int delay,
> >                 bool useListener = true,
> >                 bool sessionTransacted = true ) :
> >         JMSEndPointThread(pools,dest,sessionTransacted)
> >     {
> >         _consumer = NULL;
> >         _delay = delay;
> >         _useListener = useListener;
> >     }
> >     virtual ~Consumer(){
> >         cleanup();
> >     }
> >
> >     virtual void run() {
> >         std::cout<<"Consumer started!!!! Thread:
> "<<Thread::getId()<<std::endl;
> >         while(!stopCheck()) {
> >             try {
> >                 if(prepareConsumer()) {
> >                     // Wait while asynchronous messages come in.
> >                     if(_useListener) {
> >                         stopCheck(-1);
> >                     } else {
> >                         consumeLoop();
> >                     }
> >                     continue;
> >                 }
> >             } catch (CMSException& e) {
> >                 std::cout<<"(Consumer::run) CMSException: " <<
> > e.getMessage() << std::endl;
> >             } catch(...) {
> >                 std::cout<<"(Consumer::run) Unknown exception!!!" <<
> std::endl;
> >             }
> >             cleanup();
> >             stopCheck(5000); // reconnect delay
> >         }
> >         std::cout<<"Consumer end!!!! Thread:
> "<<Thread::getId()<<std::endl;
> >         cleanup();
> >     }
> >
> >
> > private:
> >
> >     bool prepareConsumer() {
> >         // Get Session
> >         if(!prepareSession()) return false;
> >         // Create a MessageConsumer from the Session to the Queue
> >         _consumer = getSession()->createConsumer( getDestination() );
> >         if(_useListener)
> >             _consumer->setMessageListener( this );
> >
> >         return true;
> >     }
> >
> >     void consumeLoop() {
> >         while(!stopCheck()) {
> >             Message *msg = _consumer->receive(500);
> >             if(msg!=NULL) {
> >                 onMessage(msg);
> >                 delete msg;
> >             }
> >         }
> >     }
> >
> >     // Called from the consumer since this class is a registered
> > MessageListener.
> >     virtual void onMessage( const Message* message ) {
> >
> >         static int count = 0;
> >
> >         try
> >         {
> >             count++;
> >             const TextMessage* textMessage =
> >                 dynamic_cast< const TextMessage* >( message );
> >             std::string text = "";
> >
> >             if( textMessage != NULL ) {
> >                 text = textMessage->getText();
> >             } else {
> >                 text = "NOT A TEXTMESSAGE!";
> >             }
> >
> >             std::cout<<"Message #"<<count<<" Received:
> "<<text<<std::endl;
> >             if(_delay) {
> >                 std::cout<<"Starting delay..."<<std::endl;
> >                 stopCheck(_delay);
> >             }
> >
> >         } catch (CMSException& e) {
> >             std::cout<<"(Consumer::onMessage) CMSException: " <<
> > e.getMessage() << std::endl;
> >         } catch(...) {
> >             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> > std::endl;
> >         }
> >
> >         try
> >         {   // Commit all messages.
> >             if( getTransacted() ) {
> >                 std::cout<<"Try commit....."<<std::endl;
> >                 getSession()->commit();
> >                 std::cout<<"Commit OK!!!"<<std::endl;
> >             }
> >         } catch (CMSException& e) {
> >             std::cout<<"(Consumer::onMessage) CMSException: " <<
> > e.getMessage() << std::endl;
> >         } catch(...) {
> >             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> > std::endl;
> >         }
> >
> >     }
> >
> >     virtual void cleanup() {
> >         try{
> >             if( _consumer != NULL ) delete _consumer;
> >         } catch (...) { }
> >         _consumer = NULL;
> >         JMSEndPointThread::cleanup();
> >     }
> > };
> >
> >
> > class ScopedActiveMQLibrary {
> > public:
> >     ScopedActiveMQLibrary() {
> >         activemq::library::ActiveMQCPP::initializeLibrary();
> >     }
> >     virtual ~ScopedActiveMQLibrary() {
> >         activemq::library::ActiveMQCPP::shutdownLibrary();
> >     }
> > };
> >
> > class AppTest : private ScopedActiveMQLibrary {
> > private:
> >     int _consumer;
> >     int _producer;
> >     int _numMessages;
> >     bool _useListener;
> >     int _delay;
> >     ConnPoolManager _poolManager;
> >     JMSEndPointThread::List _endPointList;
> >
> >     static AppTest* _app;
> >     static CountDownLatch _terminationRequest;
> >
> > public:
> >     AppTest(int argc, char** argv) :
> >             _consumer(0),
> >             _producer(0),
> >             _numMessages(1),
> >             _useListener(false),
> >             _delay(0)
> >     {
> >         _app=this;
> >         int c;
> >         while ((c = getopt (argc, argv, "lc:p:n:d:")) != -1) {
> >             switch(c) {
> >             case 'l': _useListener=true; break;
> >             case 'c': _consumer=atoi(optarg); break;
> >             case 'p': _producer=atoi(optarg); break;
> >             case 'n': _numMessages=atoi(optarg); break;
> >             case 'd': _delay=atoi(optarg); break;
> >             }
> >         }
> >         if(_consumer<0) _consumer=0;
> >         if(_producer<0) _producer=0;
> >         _poolManager.setBrokerURI(
> >                 "failover://("
> >                 "tcp://localhost:61616"
> >                 "?transport.useInactivityMonitor=false"
> >                 ")?timeout=1000"
> >                 "&cms.RedeliveryPolicy.maximumRedeliveries=-1"
> >                 );
> >         _poolManager.startPools();
> >     }
> >
> >     virtual ~AppTest() {
> >     }
> >
> >     int run() {
> >
> >         // Install signal handler
> >         if(!installSigAction())
> >             return 1;
> >
> >         // Create consumer/producer objects
> >         loadEndPoins();
> >
> >         // Start the producer/consumer thread.
> >         startEndPoins();
> >
> >         // start another tasks
> >         // ...
> >
> >         // Wait for termination request
> >         /*if(_consumer) {
> >             _terminationRequest.await();
> >             stopEndPoins();
> >         }*/
> >
> >         // join threads
> >         joinEndPoins();
> >
> >         return 0;
> >     }
> >
> > private:
> >
> >     void loadEndPoins() {
> >         for(int i=0; i<_consumer;i++) {
> >             _endPointList.push_back( new
> > Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
> >         }
> >         for(int i=0; i<_producer;i++) {
> >             _endPointList.push_back( new
> > Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
> >         }
> >     }
> >
> >     void startEndPoins() {
> >         for(JMSEndPointThread::List::iterator it=_endPointList.begin();
> >             it!=_endPointList.end(); it++ )
> >             (**it).start();
> >     }
> >
> >     void stopEndPoins() {
> >         for(JMSEndPointThread::List::reverse_iterator
> it=_endPointList.rbegin();
> >             it!=_endPointList.rend(); it++ )
> >             (**it).stopNow();
> >     }
> >
> >     void joinEndPoins() {
> >         for(JMSEndPointThread::List::reverse_iterator
> it=_endPointList.rbegin();
> >             it!=_endPointList.rend(); it++ ) {
> >             (**it).join();
> >             delete *it;
> >             *it = NULL;
> >         }
> >         _endPointList.clear();
> >     }
> >
> >     static void signalHandler(int sig) {
> >         _app->_terminationRequest.countDown();
> >         _app->stopEndPoins();
> >     }
> >
> >     bool installSigAction() {
> >         struct sigaction action;
> >         memset(&action, 0, sizeof(action));
> >         action.sa_handler = signalHandler;
> >         if( sigaction(SIGTERM, &action, NULL)<0 ||
> >             sigaction(SIGQUIT, &action, NULL)<0 ||
> >             sigaction(SIGINT,  &action, NULL)<0 )
> >         { return false; }
> >         return true;
> >     }
> >
> >     int termWait() {
> >         sigset_t sset;
> >         sigemptyset(&sset);
> >         sigaddset(&sset, SIGINT);
> >         sigaddset(&sset, SIGQUIT);
> >         sigaddset(&sset, SIGTERM);
> >         sigprocmask(SIG_BLOCK, &sset, NULL);
> >         int sig;
> >         sigwait(&sset, &sig);
> >         return sig;
> >     }
> >
> > };
> > AppTest* AppTest::_app = NULL;
> > CountDownLatch AppTest::_terminationRequest(1);
> >
> >
> > void my_unexpected() {
> >     throw activemq::exceptions::ActiveMQException();
> > }
> >
> > int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
> >     //std::set_unexpected(my_unexpected);
> >     AppTest test(argc, argv);
> >     return test.run();
> > }
> >
> > // END SNIPPET: demo
>
> --
> Tim Bish
> ------------
> FuseSource
> Email: tim.bish@fusesource.com
> Web: http://fusesource.com
> Twitter: tabish121
> Blog: http://timbish.blogspot.com/
>
>
>


-- 
Óscar Pernas Plaza.

Re: AMQCPP: unexpected exception

Posted by Henrique Magarotto <ma...@gmail.com>.

Thanks for reply!

issue opened:
https://issues.apache.org/jira/browse/AMQCPP-354

Regards
Henrique

2011/3/1 Timothy Bish <ta...@gmail.com>:
> On Tue, 2011-03-01 at 20:22 -0300, Henrique Magarotto wrote:
>> Hi everybody,
>>
>> First, sorry for long mail and thanks for great job in ActiveMQ and
>> ActiveMQ-CPP.
>>
>> I'm testing some failure cases with ActiveMQ and ActiveMQ-CPP.
>>
>> In my test, an unexpected exception is thrown if broker is stopped
>> while consumer is in transaction.
>>
>> Environment:
>> ActiveMQ 5.4.2
>> ActiveMQ-CPP 3.2.4
>> Ubuntu 10.04
>> JRE 1.6.0_20-b02
>>
>
>> I get the following message:
>>
>> terminate called after throwing an instance of 'cms::CMSException'
>>   what():  Failover timeout of 1000 ms reached.
>> Aborted (core dumped)
>>
>
> Sounds like a bug, can you open a new issue in are Jira bug tracker and
> attach your test case along with instructions for reproducing the error?
>
> https://issues.apache.org/jira/browse/AMQCPP
>
> Regards
> Tim.
>
>> core backtrace:
>> #0  0x001ef422 in __kernel_vsyscall ()
>> #1  0x00c5a651 in *__GI_raise (sig=6) at
>> ../nptl/sysdeps/unix/sysv/linux/raise.c:64
>> #2  0x00c5da82 in *__GI_abort () at abort.c:92
>> #3  0x00bf952f in __gnu_cxx::__verbose_terminate_handler() () from
>> /usr/lib/libstdc++.so.6
>> #4  0x00bf7465 in ?? () from /usr/lib/libstdc++.so.6
>> #5  0x00bf74a2 in std::terminate() () from /usr/lib/libstdc++.so.6
>> #6  0x00bf74c5 in ?? () from /usr/lib/libstdc++.so.6
>> #7  0x00bf6915 in __cxa_call_unexpected () from /usr/lib/libstdc++.so.6
>> #8  0x0052f8ae in
>> activemq::core::TransactionSynhcronization::beforeEnd (this=0x93ac548)
>> at activemq/core/ActiveMQConsumer.cpp:84
>> #9  0x00550588 in
>> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x93b3e48)
>> at activemq/core/ActiveMQTransactionContext.cpp:192
>> #10 0x00550d63 in activemq::core::ActiveMQTransactionContext::commit
>> (this=0x93b3e48) at activemq/core/ActiveMQTransactionContext.cpp:127
>> #11 0x0053e460 in activemq::core::ActiveMQSession::commit
>> (this=0x93b3c10) at activemq/core/ActiveMQSession.cpp:189
>> #12 0x0042c1a0 in activemq::cmsutil::PooledSession::commit
>> (this=0x93b3fd8) at activemq/cmsutil/PooledSession.h:87
>> #13 0x0804c0fb in Consumer::onMessage (this=0x93a8c30,
>> message=0x93b49b0) at main.cpp:455
>> #14 0x0804bdc8 in Consumer::consumeLoop (this=0x93a8c30) at main.cpp:415
>> #15 0x0804bb56 in Consumer::run (this=0x93a8c30) at main.cpp:381
>> #16 0x00823cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x93abff0) at decaf/lang/Thread.cpp:135
>> #17 0x00822847 in threadWorker (arg=0x93abff0) at decaf/lang/Thread.cpp:188
>> #18 0x0015a96e in start_thread (arg=0xb6f8bb70) at pthread_create.c:300
>> #19 0x00cfda4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>>
>> Exception specification for method
>> 'activemq::core::TransactionSynhcronization::beforeEnd' says: 'throw(
>> exceptions::ActiveMQException )' but 'ActiveMQConsumer::acknowledge'
>> throws 'cms::CMSException', so 'std::unexpected' is called and 'abort'
>> occurs. (http://www.linuxprogrammingblog.com/cpp-exception-specifications-are-evil)
>>
>>
>> When 'MessageListener' is used, the commit freezes, probability, in thread 5:
>>
>> Thread 1:
>> #0  0x00cfd422 in __kernel_vsyscall ()
>> #1  0x00ddcb5d in pthread_join (threadid=3061869424,
>> thread_return=0xbf964bdc) at pthread_join.c:89
>> #2  0x00a923c0 in decaf::lang::Thread::join (this=0x8eba574) at
>> decaf/lang/Thread.cpp:421
>> #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf964c64) at main.cpp:586
>> #4  0x0804c678 in AppTest::run (this=0xbf964c64) at main.cpp:555
>> #5  0x0804a33c in main (argc=4, argv=0xbf964d84) at main.cpp:633
>>
>> Thread 2:
>> #0  0x00cfd422 in __kernel_vsyscall ()
>> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
>> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eb0e38,
>> mutex=0x8eb0e08) at forward.c:139
>> #3  0x00a606a9 in
>> decaf::internal::util::concurrent::ConditionImpl::wait
>> (condition=0x8eb0e38) at
>> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
>> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
>> (this=0x8eb0d6c) at decaf/util/concurrent/Mutex.cpp:95
>> #5  0x007fc9c5 in activemq::threads::CompositeTaskRunner::run
>> (this=0x8eb0d48) at activemq/threads/CompositeTaskRunner.cpp:118
>> #6  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x8eb0e90) at decaf/lang/Thread.cpp:135
>> #7  0x00a92847 in threadWorker (arg=0x8eb0e90) at decaf/lang/Thread.cpp:188
>> #8  0x00ddb96e in start_thread (arg=0xb7808b70) at pthread_create.c:300
>> #9  0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 3:
>> #0  0x00cfd422 in __kernel_vsyscall ()
>> #1  0x00de2af9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8ebb3a8) at
>> pthread_mutex_lock.c:61
>> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8ebb3a8) at forward.c:182
>> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x8ebb3a8) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
>> (this=0x8ebb2bc) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x007c5770 in
>> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
>> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
>> (this=0x8ebb2a8)
>>     at ./decaf/util/StlQueue.h:253
>> #8  activemq::core::MessageDispatchChannel::lock (this=0x8ebb2a8) at
>> activemq/core/MessageDispatchChannel.h:153
>> #9  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb7006fe4) at decaf/util/concurrent/Lock.cpp:54
>> #10 0x00ad0c08 in Lock (this=0xfffffe00, object=0x8ebb3a8,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #11 0x0078e102 in
>> activemq::core::ActiveMQConsumer::clearMessagesInProgress
>> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:1112
>> #12 0x007af15c in
>> activemq::core::ActiveMQSession::clearMessagesInProgress
>> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:239
>> #13 0x00778543 in
>> activemq::core::ActiveMQConnection::transportInterrupted
>> (this=0x8eb1210) at activemq/core/ActiveMQConnection.cpp:704
>> #14 0x00803b34 in
>> activemq::transport::TransportFilter::transportInterrupted
>> (this=0x8eb11c0) at activemq/transport/TransportFilter.cpp:67
>> #15 0x008187ae in
>> activemq::transport::failover::FailoverTransport::handleTransportFailure
>> (this=0x8eb0a38, error=...) at
>> activemq/transport/failover/FailoverTransport.cpp:476
>> #16 0x0082406c in
>> activemq::transport::failover::FailoverTransportListener::onException
>> (this=0x8eb0878, ex=...) at
>> activemq/transport/failover/FailoverTransportListener.cpp:97
>> #17 0x00803c0b in activemq::transport::TransportFilter::fire
>> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #18 0x00803c64 in activemq::transport::TransportFilter::onException
>> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #19 0x00803c0b in activemq::transport::TransportFilter::fire
>> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:49
>> #20 0x00803c64 in activemq::transport::TransportFilter::onException
>> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:41
>> #21 0x00801b13 in activemq::transport::IOTransport::fire
>> (this=0x8eb27a0, ex=...) at activemq/transport/IOTransport.cpp:73
>> #22 0x008023bf in activemq::transport::IOTransport::run
>> (this=0x8eb27a0) at activemq/transport/IOTransport.cpp:246
>> #23 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x8eb2ea0) at decaf/lang/Thread.cpp:135
>> #24 0x00a92847 in threadWorker (arg=0x8eb2ea0) at decaf/lang/Thread.cpp:188
>> #25 0x00ddb96e in start_thread (arg=0xb7007b70) at pthread_create.c:300
>> #26 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 4:
>> #0  0x00cfd422 in __kernel_vsyscall ()
>> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
>> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eba6c8,
>> mutex=0x8eba698) at forward.c:139
>> #3  0x00a606a9 in
>> decaf::internal::util::concurrent::ConditionImpl::wait
>> (condition=0x8eba6c8) at
>> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
>> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
>> (this=0x8eba584) at decaf/util/concurrent/Mutex.cpp:95
>> #5  0x00ad029a in decaf::util::concurrent::CountDownLatch::await
>> (this=0x8eba57c) at decaf/util/concurrent/CountDownLatch.cpp:56
>> #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x8eba574,
>> timeOut=-1) at main.cpp:226
>> #7  0x0804bb46 in Consumer::run (this=0x8eba570) at main.cpp:379
>> #8  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x8eba5b0) at decaf/lang/Thread.cpp:135
>> #9  0x00a92847 in threadWorker (arg=0x8eba5b0) at decaf/lang/Thread.cpp:188
>> #10 0x00ddb96e in start_thread (arg=0xb6806b70) at pthread_create.c:300
>> #11 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>> Thread 5:
>> #0  0x00cfd422 in __kernel_vsyscall ()
>> #1  0x00de2af9 in __lll_lock_wait () at
>> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
>> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
>> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8eb05e0) at
>> pthread_mutex_lock.c:61
>> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8eb05e0) at forward.c:182
>> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
>> (handle=0x8eb05e0) at
>> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
>> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
>> (this=0x8eb0a88) at decaf/util/concurrent/Mutex.cpp:75
>> #7  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
>> (this=0xb6004ce0) at decaf/util/concurrent/Lock.cpp:54
>> #8  0x00ad0c08 in Lock (this=0xfffffe00, object=0x8eb05e0,
>> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
>> #9  0x0081b8fd in
>> activemq::transport::failover::FailoverTransport::oneway
>> (this=0x8eb0a38, command=...) at
>> activemq/transport/failover/FailoverTransport.cpp:186
>> #10 0x00807f9f in
>> activemq::transport::correlator::ResponseCorrelator::oneway
>> (this=0x8eb11c0, command=...) at
>> activemq/transport/correlator/ResponseCorrelator.cpp:82
>> #11 0x007713be in activemq::core::ActiveMQConnection::oneway
>> (this=0x8eb1210, command=...) at
>> activemq/core/ActiveMQConnection.cpp:741
>> #12 0x007b0a4f in activemq::core::ActiveMQSession::oneway
>> (this=0x8ebab90, command=...) at activemq/core/ActiveMQSession.cpp:903
>> #13 0x00795a1c in activemq::core::ActiveMQConsumer::acknowledge
>> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:860
>> #14 0x0079f885 in
>> activemq::core::TransactionSynhcronization::beforeEnd (this=0x8ebb9e8)
>> at activemq/core/ActiveMQConsumer.cpp:85
>> #15 0x007c0588 in
>> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x8ebadc8)
>> at activemq/core/ActiveMQTransactionContext.cpp:192
>> #16 0x007c0d63 in activemq::core::ActiveMQTransactionContext::commit
>> (this=0x8ebadc8) at activemq/core/ActiveMQTransactionContext.cpp:127
>> #17 0x007ae460 in activemq::core::ActiveMQSession::commit
>> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:189
>> #18 0x0069c1a0 in activemq::cmsutil::PooledSession::commit
>> (this=0x8ebaf38) at activemq/cmsutil/PooledSession.h:87
>> #19 0x0804c0fb in Consumer::onMessage (this=0x8eba570,
>> message=0x8ebbec0) at main.cpp:455
>> #20 0x0079972f in activemq::core::ActiveMQConsumer::dispatch
>> (this=0x8ebb270, dispatch=...) at
>> activemq/core/ActiveMQConsumer.cpp:1018
>> #21 0x007bd5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
>> (this=0x8ebae78, dispatch=...) at
>> activemq/core/ActiveMQSessionExecutor.cpp:129
>> #22 0x007bd993 in activemq::core::ActiveMQSessionExecutor::iterate
>> (this=0x8ebae78) at activemq/core/ActiveMQSessionExecutor.cpp:166
>> #23 0x008007f3 in activemq::threads::DedicatedTaskRunner::run
>> (this=0x8ebb550) at activemq/threads/DedicatedTaskRunner.cpp:111
>> #24 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
>> (properties=0x8ebaad8) at decaf/lang/Thread.cpp:135
>> #25 0x00a92847 in threadWorker (arg=0x8ebaad8) at decaf/lang/Thread.cpp:188
>> #26 0x00ddb96e in start_thread (arg=0xb6005b70) at pthread_create.c:300
>> #27 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
>>
>>
>>
>> I tried to resolve this issue changing configurations and code, but i
>> always get an error or client freezes.
>> Is it a bug? Any suggestions?
>>
>> Thanks.
>> Henrique
>>
>> How to reproduce:
>> 1. Start ActiveMQ
>> 2. Send a massage './activemqTest -p1 -n1'
>> 3. Run consumer './activemqTest -c1 -d10000'
>> 4. Stop activemq when 'Starting delay...' message appears.
>> 5. After activemq is down and delay is finish, consumer try commit
>> (message 'Try commit') and unexpected exception occurs.
>>
>> Following test code, based on ActiveMQ-CPP sample
>> 'activemq-cpp-library-3.2.4/src/examples/main.cpp':
>>
>>
>> /*
>>  * Licensed to the Apache Software Foundation (ASF) under one or more
>>  * contributor license agreements.  See the NOTICE file distributed with
>>  * this work for additional information regarding copyright ownership.
>>  * The ASF licenses this file to You under the Apache License, Version 2.0
>>  * (the "License"); you may not use this file except in compliance with
>>  * the License.  You may obtain a copy of the License at
>>  *
>>  *     http://www.apache.org/licenses/LICENSE-2.0
>>  *
>>  * Unless required by applicable law or agreed to in writing, software
>>  * distributed under the License is distributed on an "AS IS" BASIS,
>>  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>  * See the License for the specific language governing permissions and
>>  * limitations under the License.
>>  */
>>
>> // START SNIPPET: demo
>>
>> #include <activemq/library/ActiveMQCPP.h>
>> #include <activemq/cmsutil/SessionPool.h>
>> #include <activemq/cmsutil/PooledSession.h>
>> #include <activemq/cmsutil/ResourceLifecycleManager.h>
>> #include <activemq/exceptions/ActiveMQException.h>
>> #include <decaf/lang/Thread.h>
>> #include <decaf/lang/Runnable.h>
>> #include <decaf/util/concurrent/CountDownLatch.h>
>> #include <decaf/util/concurrent/Mutex.h>
>> #include <decaf/lang/Integer.h>
>> #include <decaf/lang/Long.h>
>> #include <decaf/lang/System.h>
>> #include <activemq/core/ActiveMQConnectionFactory.h>
>> #include <activemq/util/Config.h>
>> #include <cms/Connection.h>
>> #include <cms/Session.h>
>> #include <cms/TextMessage.h>
>> #include <cms/BytesMessage.h>
>> #include <cms/MapMessage.h>
>> #include <cms/ExceptionListener.h>
>> #include <cms/MessageListener.h>
>> #include <iostream>
>> #include <memory>
>> #include <map>
>> #include <vector>
>> #include <unistd.h>
>>
>> using namespace activemq::core;
>> using namespace activemq::cmsutil;
>> using namespace decaf::util::concurrent;
>> using namespace decaf::util;
>> using namespace decaf::lang;
>> using namespace cms;
>>
>>
>> class ConnPoolManager : protected decaf::util::concurrent::Mutex,
>>                         public ExceptionListener {
>> private:
>>     typedef std::map< int, activemq::cmsutil::SessionPool* > tSessionPoolMap;
>>
>>     std::string _brokerURI;
>>     ConnectionFactory* _connectionFactory;
>>     Connection* _connection;
>>     ResourceLifecycleManager* _lifecycleManager;
>>     tSessionPoolMap _pools;
>>     bool _started;
>>
>> public:
>>     ConnPoolManager() :
>>             _brokerURI("tcp://localhost:61616"),
>>             _connectionFactory(NULL),
>>             _connection(NULL),
>>             _lifecycleManager(NULL),
>>             _started(false)
>>     {
>>     }
>>
>>     virtual ~ConnPoolManager() {
>>         cleanup();
>>     }
>>
>>     void setBrokerURI(const std::string& uri) {
>>         _brokerURI = uri;
>>     }
>>
>>     bool startPools() {
>>         synchronized(this) {
>>             if(_started) return true;
>>             try {
>>                 createConnection();
>>                 createPools();
>>                 _started=true;
>>                 return true;
>>             } catch(CMSException& e) {
>>                 std::cout<<"(ConnPoolManager::startPools)
>> CMSException: " << e.getMessage() << std::endl;
>>             } catch(...) {
>>                 std::cout<<"(ConnPoolManager::startPools) Unknown
>> exception!!!" << std::endl;
>>             }
>>         }
>>         cleanup();
>>         return false;
>>     }
>>
>>     PooledSession* getSession(cms::Session::AcknowledgeMode ack) {
>>         if(_started) {
>>             return _pools[ack]->takeSession();
>>         } else if(startPools()) {
>>             return _pools[ack]->takeSession();
>>         } else {
>>             return NULL;
>>         }
>>     }
>>
>> private:
>>
>>     void createConnection(){
>>         // Create a ConnectionFactory
>>         _connectionFactory =
>> ConnectionFactory::createCMSConnectionFactory( _brokerURI );
>>
>>         // Create a Connection
>>         _connection = _connectionFactory->createConnection();
>>         _connection->start();
>>         _connection->setExceptionListener(this);
>>     }
>>
>>     void createPools(){
>>         // Create lifecycle manager
>>         _lifecycleManager = new activemq::cmsutil::ResourceLifecycleManager();
>>
>>         // Create pools, one for each acknowledge type
>>         for(    int ack = cms::Session::AUTO_ACKNOWLEDGE;
>>                 ack <= cms::Session::INDIVIDUAL_ACKNOWLEDGE;
>>                 ack++)
>>         {
>>             _pools[ack] =  new activemq::cmsutil::SessionPool(
>>                     _connection,
>>                     static_cast<cms::Session::AcknowledgeMode>(ack),
>>                     _lifecycleManager);
>>         }
>>     }
>>
>>     void cleanup() {
>>         synchronized(this) {
>>             releasePools();
>>             if(_lifecycleManager!=NULL) {
>>                 try { delete _lifecycleManager;
>>                 } catch(...) {}
>>             } _lifecycleManager = NULL;
>>
>>             if(_connection!=NULL) {
>>                 try { delete _connection;
>>                 } catch(...) {}
>>             } _connection = NULL;
>>             if(_connectionFactory!=NULL) {
>>                 try { delete _connectionFactory;
>>                 } catch(...) {}
>>             } _connectionFactory = NULL;
>>             _started=false;
>>         }
>>     }
>>
>>     void releasePools() {
>>         for(tSessionPoolMap::iterator it=_pools.begin();
>> it!=_pools.end(); it++) {
>>             try {
>>                 delete it->second;
>>             } catch(...) { }
>>         }
>>         _pools.clear();
>>     }
>>
>>     // If something bad happens you see it here as this class is also been
>>     // registered as an ExceptionListener with the connection.
>>     virtual void onException( const CMSException& ex AMQCPP_UNUSED) {
>>         std::cout<<"(ConnPoolManager::onException) CMS Exception
>> occurred.  Shutting down client."<<std::endl;
>>         ex.printStackTrace();
>>         exit(1);
>>     }
>>
>>
>>
>> };
>>
>> class JMSEndPointThread  : public Thread {
>> private:
>>     CountDownLatch _stopNow;
>>     ConnPoolManager* _pools;
>>     std::string _destinationString;
>>     PooledSession* _session;
>>     Destination* _destination;
>>     bool _sessionTransacted;
>>
>> protected:
>>
>>     JMSEndPointThread(  ConnPoolManager* pools,
>>                         const std::string& dest,
>>                         bool sessionTransacted) :
>>             _stopNow(1),
>>             _pools(pools),
>>             _destinationString(dest),
>>             _session(NULL),
>>             _destination(NULL),
>>             _sessionTransacted(sessionTransacted)
>>     {
>>     }
>>
>>     virtual bool prepareSession() {
>>         cleanup();
>>         // Create a Session
>>         if( _sessionTransacted ) {
>>             _session = _pools->getSession(Session::SESSION_TRANSACTED);
>>         } else {
>>             _session = _pools->getSession(Session::AUTO_ACKNOWLEDGE);
>>         }
>>         if(_session==NULL) return false;
>>
>>         // Create the Queue destination
>>         _destination = _session->createQueue( _destinationString );
>>         return true;
>>     }
>>
>>     PooledSession* getSession() { return _session; }
>>     Destination* getDestination() { return _destination; }
>>     bool getTransacted() { return _sessionTransacted; }
>>
>>     bool stopCheck(int timeOut=0) {
>>         if(timeOut>=0) return _stopNow.await(timeOut);
>>         else _stopNow.await();
>>         return true;
>>     }
>>
>>     virtual void cleanup() {
>>         // Destroy resources.
>>         try{
>>             if( _destination != NULL ) delete _destination;
>>         } catch(...) { }
>>         _destination = NULL;
>>
>>         // Back session to pool
>>         try{
>>             if( _session != NULL ) _session->close();
>>         } catch(...) { }
>>         _session = NULL;
>>     }
>>
>> public:
>>
>>     typedef std::vector<JMSEndPointThread*> List;
>>
>>     virtual ~JMSEndPointThread() {
>>         cleanup();
>>     }
>>
>>     void stopNow() {
>>         _stopNow.countDown();
>>     }
>>
>> };
>>
>> class Producer : public JMSEndPointThread {
>> private:
>>
>>     MessageProducer* _producer;
>>     int _delay;
>>     int _numMessages;
>>
>> public:
>>
>>     Producer(   ConnPoolManager* pools,
>>                 const std::string& dest,
>>                 int delay,
>>                 int numMessages,
>>                 bool sessionTransacted = false ) :
>>         JMSEndPointThread(pools,dest,sessionTransacted)
>>     {
>>         _producer = NULL;
>>         _delay = delay;
>>         _numMessages = numMessages;
>>     }
>>
>>     virtual ~Producer() {
>>         cleanup();
>>     }
>>
>>     virtual void run() {
>>         std::cout<<"Producer started!!!! Thread: "<<Thread::getId()<<std::endl;
>>         while(!stopCheck() && _numMessages) {
>>             try {
>>                 if(prepareProducer()) {
>>                     sendMessages();
>>                     continue;
>>                 }
>>             } catch(CMSException& e) {
>>                 std::cout<<"(Producer::run) CMSException: " <<
>> e.getMessage() << std::endl;
>>             } catch(...) {
>>                 std::cout<<"(Producer::run) Unknown exception!!!" << std::endl;
>>             }
>>             cleanup();
>>             stopCheck(5000); // reconnect delay
>>         }
>>         std::cout<<"Producer end!!!! Thread: "<<Thread::getId()<<std::endl;
>>         cleanup();
>>     }
>>
>> private:
>>
>>     bool prepareProducer() {
>>         // Get Session
>>         if(!prepareSession()) return false;
>>         // Create a MessageProducer from the Session to the Queue
>>         _producer = getSession()->createProducer( getDestination() );
>>         _producer->setDeliveryMode( DeliveryMode::PERSISTENT );
>>         return true;
>>     }
>>     void sendMessages() {
>>         // Create a messages
>>         std::stringstream text;
>>         text<<"Hello world! from thread " << Thread::getId();
>>
>>         while(!stopCheck() && _numMessages) {
>>
>>             TextMessage* message = getSession()->createTextMessage(
>> text.str() );
>>             message->setIntProperty( "Integer", _numMessages );
>>
>>             // Tell the producer to send the message
>>             std::cout<<"Sent message #"<<_numMessages<<" from thread
>> "<<Thread::getId()<<std::endl;
>>             _producer->send( message );
>>             delete message;
>>             --_numMessages;
>>
>>             stopCheck(_delay);
>>             if(getTransacted()) getSession()->commit();
>>         }
>>     }
>>
>>     virtual void cleanup() {
>>         try{
>>             if( _producer != NULL ) delete _producer;
>>         } catch(...) { }
>>         _producer = NULL;
>>
>>         JMSEndPointThread::cleanup();
>>     }
>>
>> };
>>
>>
>> class Consumer :    public MessageListener,
>>                     public JMSEndPointThread {
>>
>> private:
>>
>>     MessageConsumer* _consumer;
>>     int _delay;
>>     bool _useListener;
>>
>> public:
>>
>>     Consumer(   ConnPoolManager* pools,
>>                 const std::string& dest,
>>                 int delay,
>>                 bool useListener = true,
>>                 bool sessionTransacted = true ) :
>>         JMSEndPointThread(pools,dest,sessionTransacted)
>>     {
>>         _consumer = NULL;
>>         _delay = delay;
>>         _useListener = useListener;
>>     }
>>     virtual ~Consumer(){
>>         cleanup();
>>     }
>>
>>     virtual void run() {
>>         std::cout<<"Consumer started!!!! Thread: "<<Thread::getId()<<std::endl;
>>         while(!stopCheck()) {
>>             try {
>>                 if(prepareConsumer()) {
>>                     // Wait while asynchronous messages come in.
>>                     if(_useListener) {
>>                         stopCheck(-1);
>>                     } else {
>>                         consumeLoop();
>>                     }
>>                     continue;
>>                 }
>>             } catch (CMSException& e) {
>>                 std::cout<<"(Consumer::run) CMSException: " <<
>> e.getMessage() << std::endl;
>>             } catch(...) {
>>                 std::cout<<"(Consumer::run) Unknown exception!!!" << std::endl;
>>             }
>>             cleanup();
>>             stopCheck(5000); // reconnect delay
>>         }
>>         std::cout<<"Consumer end!!!! Thread: "<<Thread::getId()<<std::endl;
>>         cleanup();
>>     }
>>
>>
>> private:
>>
>>     bool prepareConsumer() {
>>         // Get Session
>>         if(!prepareSession()) return false;
>>         // Create a MessageConsumer from the Session to the Queue
>>         _consumer = getSession()->createConsumer( getDestination() );
>>         if(_useListener)
>>             _consumer->setMessageListener( this );
>>
>>         return true;
>>     }
>>
>>     void consumeLoop() {
>>         while(!stopCheck()) {
>>             Message *msg = _consumer->receive(500);
>>             if(msg!=NULL) {
>>                 onMessage(msg);
>>                 delete msg;
>>             }
>>         }
>>     }
>>
>>     // Called from the consumer since this class is a registered
>> MessageListener.
>>     virtual void onMessage( const Message* message ) {
>>
>>         static int count = 0;
>>
>>         try
>>         {
>>             count++;
>>             const TextMessage* textMessage =
>>                 dynamic_cast< const TextMessage* >( message );
>>             std::string text = "";
>>
>>             if( textMessage != NULL ) {
>>                 text = textMessage->getText();
>>             } else {
>>                 text = "NOT A TEXTMESSAGE!";
>>             }
>>
>>             std::cout<<"Message #"<<count<<" Received: "<<text<<std::endl;
>>             if(_delay) {
>>                 std::cout<<"Starting delay..."<<std::endl;
>>                 stopCheck(_delay);
>>             }
>>
>>         } catch (CMSException& e) {
>>             std::cout<<"(Consumer::onMessage) CMSException: " <<
>> e.getMessage() << std::endl;
>>         } catch(...) {
>>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
>> std::endl;
>>         }
>>
>>         try
>>         {   // Commit all messages.
>>             if( getTransacted() ) {
>>                 std::cout<<"Try commit....."<<std::endl;
>>                 getSession()->commit();
>>                 std::cout<<"Commit OK!!!"<<std::endl;
>>             }
>>         } catch (CMSException& e) {
>>             std::cout<<"(Consumer::onMessage) CMSException: " <<
>> e.getMessage() << std::endl;
>>         } catch(...) {
>>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
>> std::endl;
>>         }
>>
>>     }
>>
>>     virtual void cleanup() {
>>         try{
>>             if( _consumer != NULL ) delete _consumer;
>>         } catch (...) { }
>>         _consumer = NULL;
>>         JMSEndPointThread::cleanup();
>>     }
>> };
>>
>>
>> class ScopedActiveMQLibrary {
>> public:
>>     ScopedActiveMQLibrary() {
>>         activemq::library::ActiveMQCPP::initializeLibrary();
>>     }
>>     virtual ~ScopedActiveMQLibrary() {
>>         activemq::library::ActiveMQCPP::shutdownLibrary();
>>     }
>> };
>>
>> class AppTest : private ScopedActiveMQLibrary {
>> private:
>>     int _consumer;
>>     int _producer;
>>     int _numMessages;
>>     bool _useListener;
>>     int _delay;
>>     ConnPoolManager _poolManager;
>>     JMSEndPointThread::List _endPointList;
>>
>>     static AppTest* _app;
>>     static CountDownLatch _terminationRequest;
>>
>> public:
>>     AppTest(int argc, char** argv) :
>>             _consumer(0),
>>             _producer(0),
>>             _numMessages(1),
>>             _useListener(false),
>>             _delay(0)
>>     {
>>         _app=this;
>>         int c;
>>         while ((c = getopt (argc, argv, "lc:p:n:d:")) != -1) {
>>             switch(c) {
>>             case 'l': _useListener=true; break;
>>             case 'c': _consumer=atoi(optarg); break;
>>             case 'p': _producer=atoi(optarg); break;
>>             case 'n': _numMessages=atoi(optarg); break;
>>             case 'd': _delay=atoi(optarg); break;
>>             }
>>         }
>>         if(_consumer<0) _consumer=0;
>>         if(_producer<0) _producer=0;
>>         _poolManager.setBrokerURI(
>>                 "failover://("
>>                 "tcp://localhost:61616"
>>                 "?transport.useInactivityMonitor=false"
>>                 ")?timeout=1000"
>>                 "&cms.RedeliveryPolicy.maximumRedeliveries=-1"
>>                 );
>>         _poolManager.startPools();
>>     }
>>
>>     virtual ~AppTest() {
>>     }
>>
>>     int run() {
>>
>>         // Install signal handler
>>         if(!installSigAction())
>>             return 1;
>>
>>         // Create consumer/producer objects
>>         loadEndPoins();
>>
>>         // Start the producer/consumer thread.
>>         startEndPoins();
>>
>>         // start another tasks
>>         // ...
>>
>>         // Wait for termination request
>>         /*if(_consumer) {
>>             _terminationRequest.await();
>>             stopEndPoins();
>>         }*/
>>
>>         // join threads
>>         joinEndPoins();
>>
>>         return 0;
>>     }
>>
>> private:
>>
>>     void loadEndPoins() {
>>         for(int i=0; i<_consumer;i++) {
>>             _endPointList.push_back( new
>> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
>>         }
>>         for(int i=0; i<_producer;i++) {
>>             _endPointList.push_back( new
>> Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
>>         }
>>     }
>>
>>     void startEndPoins() {
>>         for(JMSEndPointThread::List::iterator it=_endPointList.begin();
>>             it!=_endPointList.end(); it++ )
>>             (**it).start();
>>     }
>>
>>     void stopEndPoins() {
>>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>>             it!=_endPointList.rend(); it++ )
>>             (**it).stopNow();
>>     }
>>
>>     void joinEndPoins() {
>>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>>             it!=_endPointList.rend(); it++ ) {
>>             (**it).join();
>>             delete *it;
>>             *it = NULL;
>>         }
>>         _endPointList.clear();
>>     }
>>
>>     static void signalHandler(int sig) {
>>         _app->_terminationRequest.countDown();
>>         _app->stopEndPoins();
>>     }
>>
>>     bool installSigAction() {
>>         struct sigaction action;
>>         memset(&action, 0, sizeof(action));
>>         action.sa_handler = signalHandler;
>>         if( sigaction(SIGTERM, &action, NULL)<0 ||
>>             sigaction(SIGQUIT, &action, NULL)<0 ||
>>             sigaction(SIGINT,  &action, NULL)<0 )
>>         { return false; }
>>         return true;
>>     }
>>
>>     int termWait() {
>>         sigset_t sset;
>>         sigemptyset(&sset);
>>         sigaddset(&sset, SIGINT);
>>         sigaddset(&sset, SIGQUIT);
>>         sigaddset(&sset, SIGTERM);
>>         sigprocmask(SIG_BLOCK, &sset, NULL);
>>         int sig;
>>         sigwait(&sset, &sig);
>>         return sig;
>>     }
>>
>> };
>> AppTest* AppTest::_app = NULL;
>> CountDownLatch AppTest::_terminationRequest(1);
>>
>>
>> void my_unexpected() {
>>     throw activemq::exceptions::ActiveMQException();
>> }
>>
>> int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
>>     //std::set_unexpected(my_unexpected);
>>     AppTest test(argc, argv);
>>     return test.run();
>> }
>>
>> // END SNIPPET: demo
>
> --
> Tim Bish
> ------------
> FuseSource
> Email: tim.bish@fusesource.com
> Web: http://fusesource.com
> Twitter: tabish121
> Blog: http://timbish.blogspot.com/
>
>
>

Re: AMQCPP: unexpected exception

Posted by Timothy Bish <ta...@gmail.com>.

On Tue, 2011-03-01 at 20:22 -0300, Henrique Magarotto wrote:
> Hi everybody,
> 
> First, sorry for long mail and thanks for great job in ActiveMQ and
> ActiveMQ-CPP.
> 
> I'm testing some failure cases with ActiveMQ and ActiveMQ-CPP.
> 
> In my test, an unexpected exception is thrown if broker is stopped
> while consumer is in transaction.
> 
> Environment:
> ActiveMQ 5.4.2
> ActiveMQ-CPP 3.2.4
> Ubuntu 10.04
> JRE 1.6.0_20-b02
> 

> I get the following message:
> 
> terminate called after throwing an instance of 'cms::CMSException'
>   what():  Failover timeout of 1000 ms reached.
> Aborted (core dumped)
> 

Sounds like a bug, can you open a new issue in are Jira bug tracker and
attach your test case along with instructions for reproducing the error?

https://issues.apache.org/jira/browse/AMQCPP

Regards
Tim.

> core backtrace:
> #0  0x001ef422 in __kernel_vsyscall ()
> #1  0x00c5a651 in *__GI_raise (sig=6) at
> ../nptl/sysdeps/unix/sysv/linux/raise.c:64
> #2  0x00c5da82 in *__GI_abort () at abort.c:92
> #3  0x00bf952f in __gnu_cxx::__verbose_terminate_handler() () from
> /usr/lib/libstdc++.so.6
> #4  0x00bf7465 in ?? () from /usr/lib/libstdc++.so.6
> #5  0x00bf74a2 in std::terminate() () from /usr/lib/libstdc++.so.6
> #6  0x00bf74c5 in ?? () from /usr/lib/libstdc++.so.6
> #7  0x00bf6915 in __cxa_call_unexpected () from /usr/lib/libstdc++.so.6
> #8  0x0052f8ae in
> activemq::core::TransactionSynhcronization::beforeEnd (this=0x93ac548)
> at activemq/core/ActiveMQConsumer.cpp:84
> #9  0x00550588 in
> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x93b3e48)
> at activemq/core/ActiveMQTransactionContext.cpp:192
> #10 0x00550d63 in activemq::core::ActiveMQTransactionContext::commit
> (this=0x93b3e48) at activemq/core/ActiveMQTransactionContext.cpp:127
> #11 0x0053e460 in activemq::core::ActiveMQSession::commit
> (this=0x93b3c10) at activemq/core/ActiveMQSession.cpp:189
> #12 0x0042c1a0 in activemq::cmsutil::PooledSession::commit
> (this=0x93b3fd8) at activemq/cmsutil/PooledSession.h:87
> #13 0x0804c0fb in Consumer::onMessage (this=0x93a8c30,
> message=0x93b49b0) at main.cpp:455
> #14 0x0804bdc8 in Consumer::consumeLoop (this=0x93a8c30) at main.cpp:415
> #15 0x0804bb56 in Consumer::run (this=0x93a8c30) at main.cpp:381
> #16 0x00823cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x93abff0) at decaf/lang/Thread.cpp:135
> #17 0x00822847 in threadWorker (arg=0x93abff0) at decaf/lang/Thread.cpp:188
> #18 0x0015a96e in start_thread (arg=0xb6f8bb70) at pthread_create.c:300
> #19 0x00cfda4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> 
> Exception specification for method
> 'activemq::core::TransactionSynhcronization::beforeEnd' says: 'throw(
> exceptions::ActiveMQException )' but 'ActiveMQConsumer::acknowledge'
> throws 'cms::CMSException', so 'std::unexpected' is called and 'abort'
> occurs. (http://www.linuxprogrammingblog.com/cpp-exception-specifications-are-evil)
> 
> 
> When 'MessageListener' is used, the commit freezes, probability, in thread 5:
> 
> Thread 1:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00ddcb5d in pthread_join (threadid=3061869424,
> thread_return=0xbf964bdc) at pthread_join.c:89
> #2  0x00a923c0 in decaf::lang::Thread::join (this=0x8eba574) at
> decaf/lang/Thread.cpp:421
> #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf964c64) at main.cpp:586
> #4  0x0804c678 in AppTest::run (this=0xbf964c64) at main.cpp:555
> #5  0x0804a33c in main (argc=4, argv=0xbf964d84) at main.cpp:633
> 
> Thread 2:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eb0e38,
> mutex=0x8eb0e08) at forward.c:139
> #3  0x00a606a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x8eb0e38) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> (this=0x8eb0d6c) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x007fc9c5 in activemq::threads::CompositeTaskRunner::run
> (this=0x8eb0d48) at activemq/threads/CompositeTaskRunner.cpp:118
> #6  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eb0e90) at decaf/lang/Thread.cpp:135
> #7  0x00a92847 in threadWorker (arg=0x8eb0e90) at decaf/lang/Thread.cpp:188
> #8  0x00ddb96e in start_thread (arg=0xb7808b70) at pthread_create.c:300
> #9  0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 3:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de2af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8ebb3a8) at
> pthread_mutex_lock.c:61
> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8ebb3a8) at forward.c:182
> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x8ebb3a8) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> (this=0x8ebb2bc) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x007c5770 in
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> (this=0x8ebb2a8)
>     at ./decaf/util/StlQueue.h:253
> #8  activemq::core::MessageDispatchChannel::lock (this=0x8ebb2a8) at
> activemq/core/MessageDispatchChannel.h:153
> #9  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb7006fe4) at decaf/util/concurrent/Lock.cpp:54
> #10 0x00ad0c08 in Lock (this=0xfffffe00, object=0x8ebb3a8,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #11 0x0078e102 in
> activemq::core::ActiveMQConsumer::clearMessagesInProgress
> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:1112
> #12 0x007af15c in
> activemq::core::ActiveMQSession::clearMessagesInProgress
> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:239
> #13 0x00778543 in
> activemq::core::ActiveMQConnection::transportInterrupted
> (this=0x8eb1210) at activemq/core/ActiveMQConnection.cpp:704
> #14 0x00803b34 in
> activemq::transport::TransportFilter::transportInterrupted
> (this=0x8eb11c0) at activemq/transport/TransportFilter.cpp:67
> #15 0x008187ae in
> activemq::transport::failover::FailoverTransport::handleTransportFailure
> (this=0x8eb0a38, error=...) at
> activemq/transport/failover/FailoverTransport.cpp:476
> #16 0x0082406c in
> activemq::transport::failover::FailoverTransportListener::onException
> (this=0x8eb0878, ex=...) at
> activemq/transport/failover/FailoverTransportListener.cpp:97
> #17 0x00803c0b in activemq::transport::TransportFilter::fire
> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:49
> #18 0x00803c64 in activemq::transport::TransportFilter::onException
> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:41
> #19 0x00803c0b in activemq::transport::TransportFilter::fire
> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:49
> #20 0x00803c64 in activemq::transport::TransportFilter::onException
> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:41
> #21 0x00801b13 in activemq::transport::IOTransport::fire
> (this=0x8eb27a0, ex=...) at activemq/transport/IOTransport.cpp:73
> #22 0x008023bf in activemq::transport::IOTransport::run
> (this=0x8eb27a0) at activemq/transport/IOTransport.cpp:246
> #23 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eb2ea0) at decaf/lang/Thread.cpp:135
> #24 0x00a92847 in threadWorker (arg=0x8eb2ea0) at decaf/lang/Thread.cpp:188
> #25 0x00ddb96e in start_thread (arg=0xb7007b70) at pthread_create.c:300
> #26 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 4:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eba6c8,
> mutex=0x8eba698) at forward.c:139
> #3  0x00a606a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x8eba6c8) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> (this=0x8eba584) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x00ad029a in decaf::util::concurrent::CountDownLatch::await
> (this=0x8eba57c) at decaf/util/concurrent/CountDownLatch.cpp:56
> #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x8eba574,
> timeOut=-1) at main.cpp:226
> #7  0x0804bb46 in Consumer::run (this=0x8eba570) at main.cpp:379
> #8  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eba5b0) at decaf/lang/Thread.cpp:135
> #9  0x00a92847 in threadWorker (arg=0x8eba5b0) at decaf/lang/Thread.cpp:188
> #10 0x00ddb96e in start_thread (arg=0xb6806b70) at pthread_create.c:300
> #11 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 5:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de2af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8eb05e0) at
> pthread_mutex_lock.c:61
> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8eb05e0) at forward.c:182
> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x8eb05e0) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> (this=0x8eb0a88) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb6004ce0) at decaf/util/concurrent/Lock.cpp:54
> #8  0x00ad0c08 in Lock (this=0xfffffe00, object=0x8eb05e0,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #9  0x0081b8fd in
> activemq::transport::failover::FailoverTransport::oneway
> (this=0x8eb0a38, command=...) at
> activemq/transport/failover/FailoverTransport.cpp:186
> #10 0x00807f9f in
> activemq::transport::correlator::ResponseCorrelator::oneway
> (this=0x8eb11c0, command=...) at
> activemq/transport/correlator/ResponseCorrelator.cpp:82
> #11 0x007713be in activemq::core::ActiveMQConnection::oneway
> (this=0x8eb1210, command=...) at
> activemq/core/ActiveMQConnection.cpp:741
> #12 0x007b0a4f in activemq::core::ActiveMQSession::oneway
> (this=0x8ebab90, command=...) at activemq/core/ActiveMQSession.cpp:903
> #13 0x00795a1c in activemq::core::ActiveMQConsumer::acknowledge
> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:860
> #14 0x0079f885 in
> activemq::core::TransactionSynhcronization::beforeEnd (this=0x8ebb9e8)
> at activemq/core/ActiveMQConsumer.cpp:85
> #15 0x007c0588 in
> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x8ebadc8)
> at activemq/core/ActiveMQTransactionContext.cpp:192
> #16 0x007c0d63 in activemq::core::ActiveMQTransactionContext::commit
> (this=0x8ebadc8) at activemq/core/ActiveMQTransactionContext.cpp:127
> #17 0x007ae460 in activemq::core::ActiveMQSession::commit
> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:189
> #18 0x0069c1a0 in activemq::cmsutil::PooledSession::commit
> (this=0x8ebaf38) at activemq/cmsutil/PooledSession.h:87
> #19 0x0804c0fb in Consumer::onMessage (this=0x8eba570,
> message=0x8ebbec0) at main.cpp:455
> #20 0x0079972f in activemq::core::ActiveMQConsumer::dispatch
> (this=0x8ebb270, dispatch=...) at
> activemq/core/ActiveMQConsumer.cpp:1018
> #21 0x007bd5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
> (this=0x8ebae78, dispatch=...) at
> activemq/core/ActiveMQSessionExecutor.cpp:129
> #22 0x007bd993 in activemq::core::ActiveMQSessionExecutor::iterate
> (this=0x8ebae78) at activemq/core/ActiveMQSessionExecutor.cpp:166
> #23 0x008007f3 in activemq::threads::DedicatedTaskRunner::run
> (this=0x8ebb550) at activemq/threads/DedicatedTaskRunner.cpp:111
> #24 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8ebaad8) at decaf/lang/Thread.cpp:135
> #25 0x00a92847 in threadWorker (arg=0x8ebaad8) at decaf/lang/Thread.cpp:188
> #26 0x00ddb96e in start_thread (arg=0xb6005b70) at pthread_create.c:300
> #27 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> 
> 
> I tried to resolve this issue changing configurations and code, but i
> always get an error or client freezes.
> Is it a bug? Any suggestions?
> 
> Thanks.
> Henrique
> 
> How to reproduce:
> 1. Start ActiveMQ
> 2. Send a massage './activemqTest -p1 -n1'
> 3. Run consumer './activemqTest -c1 -d10000'
> 4. Stop activemq when 'Starting delay...' message appears.
> 5. After activemq is down and delay is finish, consumer try commit
> (message 'Try commit') and unexpected exception occurs.
> 
> Following test code, based on ActiveMQ-CPP sample
> 'activemq-cpp-library-3.2.4/src/examples/main.cpp':
> 
> 
> /*
>  * Licensed to the Apache Software Foundation (ASF) under one or more
>  * contributor license agreements.  See the NOTICE file distributed with
>  * this work for additional information regarding copyright ownership.
>  * The ASF licenses this file to You under the Apache License, Version 2.0
>  * (the "License"); you may not use this file except in compliance with
>  * the License.  You may obtain a copy of the License at
>  *
>  *     http://www.apache.org/licenses/LICENSE-2.0
>  *
>  * Unless required by applicable law or agreed to in writing, software
>  * distributed under the License is distributed on an "AS IS" BASIS,
>  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>  * See the License for the specific language governing permissions and
>  * limitations under the License.
>  */
> 
> // START SNIPPET: demo
> 
> #include <activemq/library/ActiveMQCPP.h>
> #include <activemq/cmsutil/SessionPool.h>
> #include <activemq/cmsutil/PooledSession.h>
> #include <activemq/cmsutil/ResourceLifecycleManager.h>
> #include <activemq/exceptions/ActiveMQException.h>
> #include <decaf/lang/Thread.h>
> #include <decaf/lang/Runnable.h>
> #include <decaf/util/concurrent/CountDownLatch.h>
> #include <decaf/util/concurrent/Mutex.h>
> #include <decaf/lang/Integer.h>
> #include <decaf/lang/Long.h>
> #include <decaf/lang/System.h>
> #include <activemq/core/ActiveMQConnectionFactory.h>
> #include <activemq/util/Config.h>
> #include <cms/Connection.h>
> #include <cms/Session.h>
> #include <cms/TextMessage.h>
> #include <cms/BytesMessage.h>
> #include <cms/MapMessage.h>
> #include <cms/ExceptionListener.h>
> #include <cms/MessageListener.h>
> #include <iostream>
> #include <memory>
> #include <map>
> #include <vector>
> #include <unistd.h>
> 
> using namespace activemq::core;
> using namespace activemq::cmsutil;
> using namespace decaf::util::concurrent;
> using namespace decaf::util;
> using namespace decaf::lang;
> using namespace cms;
> 
> 
> class ConnPoolManager : protected decaf::util::concurrent::Mutex,
>                         public ExceptionListener {
> private:
>     typedef std::map< int, activemq::cmsutil::SessionPool* > tSessionPoolMap;
> 
>     std::string _brokerURI;
>     ConnectionFactory* _connectionFactory;
>     Connection* _connection;
>     ResourceLifecycleManager* _lifecycleManager;
>     tSessionPoolMap _pools;
>     bool _started;
> 
> public:
>     ConnPoolManager() :
>             _brokerURI("tcp://localhost:61616"),
>             _connectionFactory(NULL),
>             _connection(NULL),
>             _lifecycleManager(NULL),
>             _started(false)
>     {
>     }
> 
>     virtual ~ConnPoolManager() {
>         cleanup();
>     }
> 
>     void setBrokerURI(const std::string& uri) {
>         _brokerURI = uri;
>     }
> 
>     bool startPools() {
>         synchronized(this) {
>             if(_started) return true;
>             try {
>                 createConnection();
>                 createPools();
>                 _started=true;
>                 return true;
>             } catch(CMSException& e) {
>                 std::cout<<"(ConnPoolManager::startPools)
> CMSException: " << e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(ConnPoolManager::startPools) Unknown
> exception!!!" << std::endl;
>             }
>         }
>         cleanup();
>         return false;
>     }
> 
>     PooledSession* getSession(cms::Session::AcknowledgeMode ack) {
>         if(_started) {
>             return _pools[ack]->takeSession();
>         } else if(startPools()) {
>             return _pools[ack]->takeSession();
>         } else {
>             return NULL;
>         }
>     }
> 
> private:
> 
>     void createConnection(){
>         // Create a ConnectionFactory
>         _connectionFactory =
> ConnectionFactory::createCMSConnectionFactory( _brokerURI );
> 
>         // Create a Connection
>         _connection = _connectionFactory->createConnection();
>         _connection->start();
>         _connection->setExceptionListener(this);
>     }
> 
>     void createPools(){
>         // Create lifecycle manager
>         _lifecycleManager = new activemq::cmsutil::ResourceLifecycleManager();
> 
>         // Create pools, one for each acknowledge type
>         for(    int ack = cms::Session::AUTO_ACKNOWLEDGE;
>                 ack <= cms::Session::INDIVIDUAL_ACKNOWLEDGE;
>                 ack++)
>         {
>             _pools[ack] =  new activemq::cmsutil::SessionPool(
>                     _connection,
>                     static_cast<cms::Session::AcknowledgeMode>(ack),
>                     _lifecycleManager);
>         }
>     }
> 
>     void cleanup() {
>         synchronized(this) {
>             releasePools();
>             if(_lifecycleManager!=NULL) {
>                 try { delete _lifecycleManager;
>                 } catch(...) {}
>             } _lifecycleManager = NULL;
> 
>             if(_connection!=NULL) {
>                 try { delete _connection;
>                 } catch(...) {}
>             } _connection = NULL;
>             if(_connectionFactory!=NULL) {
>                 try { delete _connectionFactory;
>                 } catch(...) {}
>             } _connectionFactory = NULL;
>             _started=false;
>         }
>     }
> 
>     void releasePools() {
>         for(tSessionPoolMap::iterator it=_pools.begin();
> it!=_pools.end(); it++) {
>             try {
>                 delete it->second;
>             } catch(...) { }
>         }
>         _pools.clear();
>     }
> 
>     // If something bad happens you see it here as this class is also been
>     // registered as an ExceptionListener with the connection.
>     virtual void onException( const CMSException& ex AMQCPP_UNUSED) {
>         std::cout<<"(ConnPoolManager::onException) CMS Exception
> occurred.  Shutting down client."<<std::endl;
>         ex.printStackTrace();
>         exit(1);
>     }
> 
> 
> 
> };
> 
> class JMSEndPointThread  : public Thread {
> private:
>     CountDownLatch _stopNow;
>     ConnPoolManager* _pools;
>     std::string _destinationString;
>     PooledSession* _session;
>     Destination* _destination;
>     bool _sessionTransacted;
> 
> protected:
> 
>     JMSEndPointThread(  ConnPoolManager* pools,
>                         const std::string& dest,
>                         bool sessionTransacted) :
>             _stopNow(1),
>             _pools(pools),
>             _destinationString(dest),
>             _session(NULL),
>             _destination(NULL),
>             _sessionTransacted(sessionTransacted)
>     {
>     }
> 
>     virtual bool prepareSession() {
>         cleanup();
>         // Create a Session
>         if( _sessionTransacted ) {
>             _session = _pools->getSession(Session::SESSION_TRANSACTED);
>         } else {
>             _session = _pools->getSession(Session::AUTO_ACKNOWLEDGE);
>         }
>         if(_session==NULL) return false;
> 
>         // Create the Queue destination
>         _destination = _session->createQueue( _destinationString );
>         return true;
>     }
> 
>     PooledSession* getSession() { return _session; }
>     Destination* getDestination() { return _destination; }
>     bool getTransacted() { return _sessionTransacted; }
> 
>     bool stopCheck(int timeOut=0) {
>         if(timeOut>=0) return _stopNow.await(timeOut);
>         else _stopNow.await();
>         return true;
>     }
> 
>     virtual void cleanup() {
>         // Destroy resources.
>         try{
>             if( _destination != NULL ) delete _destination;
>         } catch(...) { }
>         _destination = NULL;
> 
>         // Back session to pool
>         try{
>             if( _session != NULL ) _session->close();
>         } catch(...) { }
>         _session = NULL;
>     }
> 
> public:
> 
>     typedef std::vector<JMSEndPointThread*> List;
> 
>     virtual ~JMSEndPointThread() {
>         cleanup();
>     }
> 
>     void stopNow() {
>         _stopNow.countDown();
>     }
> 
> };
> 
> class Producer : public JMSEndPointThread {
> private:
> 
>     MessageProducer* _producer;
>     int _delay;
>     int _numMessages;
> 
> public:
> 
>     Producer(   ConnPoolManager* pools,
>                 const std::string& dest,
>                 int delay,
>                 int numMessages,
>                 bool sessionTransacted = false ) :
>         JMSEndPointThread(pools,dest,sessionTransacted)
>     {
>         _producer = NULL;
>         _delay = delay;
>         _numMessages = numMessages;
>     }
> 
>     virtual ~Producer() {
>         cleanup();
>     }
> 
>     virtual void run() {
>         std::cout<<"Producer started!!!! Thread: "<<Thread::getId()<<std::endl;
>         while(!stopCheck() && _numMessages) {
>             try {
>                 if(prepareProducer()) {
>                     sendMessages();
>                     continue;
>                 }
>             } catch(CMSException& e) {
>                 std::cout<<"(Producer::run) CMSException: " <<
> e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(Producer::run) Unknown exception!!!" << std::endl;
>             }
>             cleanup();
>             stopCheck(5000); // reconnect delay
>         }
>         std::cout<<"Producer end!!!! Thread: "<<Thread::getId()<<std::endl;
>         cleanup();
>     }
> 
> private:
> 
>     bool prepareProducer() {
>         // Get Session
>         if(!prepareSession()) return false;
>         // Create a MessageProducer from the Session to the Queue
>         _producer = getSession()->createProducer( getDestination() );
>         _producer->setDeliveryMode( DeliveryMode::PERSISTENT );
>         return true;
>     }
>     void sendMessages() {
>         // Create a messages
>         std::stringstream text;
>         text<<"Hello world! from thread " << Thread::getId();
> 
>         while(!stopCheck() && _numMessages) {
> 
>             TextMessage* message = getSession()->createTextMessage(
> text.str() );
>             message->setIntProperty( "Integer", _numMessages );
> 
>             // Tell the producer to send the message
>             std::cout<<"Sent message #"<<_numMessages<<" from thread
> "<<Thread::getId()<<std::endl;
>             _producer->send( message );
>             delete message;
>             --_numMessages;
> 
>             stopCheck(_delay);
>             if(getTransacted()) getSession()->commit();
>         }
>     }
> 
>     virtual void cleanup() {
>         try{
>             if( _producer != NULL ) delete _producer;
>         } catch(...) { }
>         _producer = NULL;
> 
>         JMSEndPointThread::cleanup();
>     }
> 
> };
> 
> 
> class Consumer :    public MessageListener,
>                     public JMSEndPointThread {
> 
> private:
> 
>     MessageConsumer* _consumer;
>     int _delay;
>     bool _useListener;
> 
> public:
> 
>     Consumer(   ConnPoolManager* pools,
>                 const std::string& dest,
>                 int delay,
>                 bool useListener = true,
>                 bool sessionTransacted = true ) :
>         JMSEndPointThread(pools,dest,sessionTransacted)
>     {
>         _consumer = NULL;
>         _delay = delay;
>         _useListener = useListener;
>     }
>     virtual ~Consumer(){
>         cleanup();
>     }
> 
>     virtual void run() {
>         std::cout<<"Consumer started!!!! Thread: "<<Thread::getId()<<std::endl;
>         while(!stopCheck()) {
>             try {
>                 if(prepareConsumer()) {
>                     // Wait while asynchronous messages come in.
>                     if(_useListener) {
>                         stopCheck(-1);
>                     } else {
>                         consumeLoop();
>                     }
>                     continue;
>                 }
>             } catch (CMSException& e) {
>                 std::cout<<"(Consumer::run) CMSException: " <<
> e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(Consumer::run) Unknown exception!!!" << std::endl;
>             }
>             cleanup();
>             stopCheck(5000); // reconnect delay
>         }
>         std::cout<<"Consumer end!!!! Thread: "<<Thread::getId()<<std::endl;
>         cleanup();
>     }
> 
> 
> private:
> 
>     bool prepareConsumer() {
>         // Get Session
>         if(!prepareSession()) return false;
>         // Create a MessageConsumer from the Session to the Queue
>         _consumer = getSession()->createConsumer( getDestination() );
>         if(_useListener)
>             _consumer->setMessageListener( this );
> 
>         return true;
>     }
> 
>     void consumeLoop() {
>         while(!stopCheck()) {
>             Message *msg = _consumer->receive(500);
>             if(msg!=NULL) {
>                 onMessage(msg);
>                 delete msg;
>             }
>         }
>     }
> 
>     // Called from the consumer since this class is a registered
> MessageListener.
>     virtual void onMessage( const Message* message ) {
> 
>         static int count = 0;
> 
>         try
>         {
>             count++;
>             const TextMessage* textMessage =
>                 dynamic_cast< const TextMessage* >( message );
>             std::string text = "";
> 
>             if( textMessage != NULL ) {
>                 text = textMessage->getText();
>             } else {
>                 text = "NOT A TEXTMESSAGE!";
>             }
> 
>             std::cout<<"Message #"<<count<<" Received: "<<text<<std::endl;
>             if(_delay) {
>                 std::cout<<"Starting delay..."<<std::endl;
>                 stopCheck(_delay);
>             }
> 
>         } catch (CMSException& e) {
>             std::cout<<"(Consumer::onMessage) CMSException: " <<
> e.getMessage() << std::endl;
>         } catch(...) {
>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> std::endl;
>         }
> 
>         try
>         {   // Commit all messages.
>             if( getTransacted() ) {
>                 std::cout<<"Try commit....."<<std::endl;
>                 getSession()->commit();
>                 std::cout<<"Commit OK!!!"<<std::endl;
>             }
>         } catch (CMSException& e) {
>             std::cout<<"(Consumer::onMessage) CMSException: " <<
> e.getMessage() << std::endl;
>         } catch(...) {
>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> std::endl;
>         }
> 
>     }
> 
>     virtual void cleanup() {
>         try{
>             if( _consumer != NULL ) delete _consumer;
>         } catch (...) { }
>         _consumer = NULL;
>         JMSEndPointThread::cleanup();
>     }
> };
> 
> 
> class ScopedActiveMQLibrary {
> public:
>     ScopedActiveMQLibrary() {
>         activemq::library::ActiveMQCPP::initializeLibrary();
>     }
>     virtual ~ScopedActiveMQLibrary() {
>         activemq::library::ActiveMQCPP::shutdownLibrary();
>     }
> };
> 
> class AppTest : private ScopedActiveMQLibrary {
> private:
>     int _consumer;
>     int _producer;
>     int _numMessages;
>     bool _useListener;
>     int _delay;
>     ConnPoolManager _poolManager;
>     JMSEndPointThread::List _endPointList;
> 
>     static AppTest* _app;
>     static CountDownLatch _terminationRequest;
> 
> public:
>     AppTest(int argc, char** argv) :
>             _consumer(0),
>             _producer(0),
>             _numMessages(1),
>             _useListener(false),
>             _delay(0)
>     {
>         _app=this;
>         int c;
>         while ((c = getopt (argc, argv, "lc:p:n:d:")) != -1) {
>             switch(c) {
>             case 'l': _useListener=true; break;
>             case 'c': _consumer=atoi(optarg); break;
>             case 'p': _producer=atoi(optarg); break;
>             case 'n': _numMessages=atoi(optarg); break;
>             case 'd': _delay=atoi(optarg); break;
>             }
>         }
>         if(_consumer<0) _consumer=0;
>         if(_producer<0) _producer=0;
>         _poolManager.setBrokerURI(
>                 "failover://("
>                 "tcp://localhost:61616"
>                 "?transport.useInactivityMonitor=false"
>                 ")?timeout=1000"
>                 "&cms.RedeliveryPolicy.maximumRedeliveries=-1"
>                 );
>         _poolManager.startPools();
>     }
> 
>     virtual ~AppTest() {
>     }
> 
>     int run() {
> 
>         // Install signal handler
>         if(!installSigAction())
>             return 1;
> 
>         // Create consumer/producer objects
>         loadEndPoins();
> 
>         // Start the producer/consumer thread.
>         startEndPoins();
> 
>         // start another tasks
>         // ...
> 
>         // Wait for termination request
>         /*if(_consumer) {
>             _terminationRequest.await();
>             stopEndPoins();
>         }*/
> 
>         // join threads
>         joinEndPoins();
> 
>         return 0;
>     }
> 
> private:
> 
>     void loadEndPoins() {
>         for(int i=0; i<_consumer;i++) {
>             _endPointList.push_back( new
> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
>         }
>         for(int i=0; i<_producer;i++) {
>             _endPointList.push_back( new
> Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
>         }
>     }
> 
>     void startEndPoins() {
>         for(JMSEndPointThread::List::iterator it=_endPointList.begin();
>             it!=_endPointList.end(); it++ )
>             (**it).start();
>     }
> 
>     void stopEndPoins() {
>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>             it!=_endPointList.rend(); it++ )
>             (**it).stopNow();
>     }
> 
>     void joinEndPoins() {
>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>             it!=_endPointList.rend(); it++ ) {
>             (**it).join();
>             delete *it;
>             *it = NULL;
>         }
>         _endPointList.clear();
>     }
> 
>     static void signalHandler(int sig) {
>         _app->_terminationRequest.countDown();
>         _app->stopEndPoins();
>     }
> 
>     bool installSigAction() {
>         struct sigaction action;
>         memset(&action, 0, sizeof(action));
>         action.sa_handler = signalHandler;
>         if( sigaction(SIGTERM, &action, NULL)<0 ||
>             sigaction(SIGQUIT, &action, NULL)<0 ||
>             sigaction(SIGINT,  &action, NULL)<0 )
>         { return false; }
>         return true;
>     }
> 
>     int termWait() {
>         sigset_t sset;
>         sigemptyset(&sset);
>         sigaddset(&sset, SIGINT);
>         sigaddset(&sset, SIGQUIT);
>         sigaddset(&sset, SIGTERM);
>         sigprocmask(SIG_BLOCK, &sset, NULL);
>         int sig;
>         sigwait(&sset, &sig);
>         return sig;
>     }
> 
> };
> AppTest* AppTest::_app = NULL;
> CountDownLatch AppTest::_terminationRequest(1);
> 
> 
> void my_unexpected() {
>     throw activemq::exceptions::ActiveMQException();
> }
> 
> int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
>     //std::set_unexpected(my_unexpected);
>     AppTest test(argc, argv);
>     return test.run();
> }
> 
> // END SNIPPET: demo

-- 
Tim Bish
------------
FuseSource
Email: tim.bish@fusesource.com
Web: http://fusesource.com
Twitter: tabish121
Blog: http://timbish.blogspot.com/