Hi there,
We have a large, vertical C++ application with ActiveMQ connectivity added
about a year ago, steadily upgrading the CMS interface until current (v3.1.1)
and, beginning before 3.0.1, we noticed after consecutive login failures --
within our app, it takes from 3-10 actual retries -- we find a pure virtual
function call somewhere in the AMQ thread pool, triggering our built-in
crashdump mechanism and shutting things down.
An exception that's at least cosmetically similar may be reproduced with the
SimpleProducer, as I've doctored it and attached to this email (not sure if
attachments make it to the list...), along with the thread stacks/traces
witnessed at the time. Start SimpleProducer and, after an indeterminate number
of retries -- in this example, it could take several hundred -- the
circumstance should arise. The test is performed between two hosts, using AMQ
broker v5.3 and CMS 3.1.1 and ought to manifest in either debug/release build.
SimpleProducer has been altered to:
===================================
- Provide username/password (in our test case, the username "foo" is incorrect)
- Connect to a remote v5.3 AMQ broker
- Loop the example indefinitely with a short delay between retries until (in a
debug build) a pure virtual assert should trip in the standard library
Our application is built unusually in the following ways:
=========================================================
- AMQ CMS and APR are all reconfigured and built __stdcall, since our main
application is that way and name mangling won't otherwise match at link time
- We don't have STL debug iterators enabled, so key AMQ CMS/APR headers are
prepended with (more or less):
#ifdef _HAS_ITERATOR_DEBUGGING
#undef _HAS_ITERATOR_DEBUGGING
#endif
#define _HAS_ITERATOR_DEBUGGING 0
...This may seem like an odd circumstance, but we function in enterprises where
AMQ connectivity is only part of what our application is responsible for and,
occasionally (for example) brokers may be misconfigured and remain so for
notable periods before changes make their way to all of the clients, during
which time we must remain operational in other regards. In practice, as I've
said, our application will experience the pure virtual call within a few
minutes.
Thank you for your time/assistance, in advance, and we otherwise very much
appreciate the functionality AMQ/CMS provides us.
Please let me know if I may provide any additional information. Thanks.
-Regards,
MjK
- - - - -
"Here lies Zeke, second-fastest draw in Cripple Creek."
- Epitaph (Cripple Creek, CO)
Michael J. Kitchin (Senior Software Engineer)
Tideworks Technology
http://www.tideworks.com
/*
* 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.
*/
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/lang/Long.h>
#include <decaf/util/Date.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/util/Config.h>
#include <activemq/library/ActiveMQCPP.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 <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <memory>
using namespace activemq;
using namespace activemq::core;
using namespace decaf;
using namespace decaf::lang;
using namespace decaf::util;
using namespace decaf::util::concurrent;
using namespace cms;
using namespace std;
////////////////////////////////////////////////////////////////////////////////
class SimpleProducer : public Runnable {
private:
Connection* connection;
Session* session;
Destination* destination;
MessageProducer* producer;
bool useTopic;
bool clientAck;
unsigned int numMessages;
std::string brokerURI;
std::string destURI;
std::string userName;
std::string passWord;
public:
SimpleProducer( const std::string& brokerURI,
unsigned int numMessages,
const std::string& destURI,
const std::string& userName,
const std::string& passWord,
bool useTopic = false,
bool clientAck = false ){
this->connection = NULL;
this->session = NULL;
this->destination = NULL;
this->producer = NULL;
this->numMessages = numMessages;
this->useTopic = useTopic;
this->brokerURI = brokerURI;
this->userName = userName;
this->passWord = passWord;
this->destURI = destURI;
this->clientAck = clientAck;
}
virtual ~SimpleProducer(){
cleanup();
}
void close() {
this->cleanup();
}
virtual void run() {
try {
// Create a ConnectionFactory
auto_ptr<ActiveMQConnectionFactory> connectionFactory(
new ActiveMQConnectionFactory( brokerURI ) );
// Create a Connection
try{
if ( !this->userName.empty() ||
!this->passWord.empty() ) {
connection =
connectionFactory->createConnection(userName, passWord);
} else {
connection =
connectionFactory->createConnection();
}
connection->start();
} catch( CMSException& e ) {
throw e;
}
// Create a Session
if( clientAck ) {
session = connection->createSession(
Session::CLIENT_ACKNOWLEDGE );
} else {
session = connection->createSession( Session::AUTO_ACKNOWLEDGE
);
}
// Create the destination (Topic or Queue)
if( useTopic ) {
destination = session->createTopic( destURI );
} else {
destination = session->createQueue( destURI );
}
// Create a MessageProducer from the Session to the Topic or Queue
producer = session->createProducer( destination );
producer->setDeliveryMode( DeliveryMode::NON_PERSISTENT );
// Create the Thread Id String
string threadIdStr = Long::toString( Thread::getId() );
// Create a messages
string text = (string)"Hello world! from thread " + threadIdStr;
for( unsigned int ix=0; ix<numMessages; ++ix ){
TextMessage* message = session->createTextMessage( text );
message->setIntProperty( "Integer", ix );
// Tell the producer to send the message
printf( "Sent message #%d from thread %s\n", ix+1,
threadIdStr.c_str() );
producer->send( message );
delete message;
}
}catch ( CMSException& e ) {
e.printStackTrace();
}
}
private:
void cleanup(){
// Destroy resources.
try{
if( destination != NULL ) delete destination;
}catch ( CMSException& e ) { e.printStackTrace(); }
destination = NULL;
try{
if( producer != NULL ) delete producer;
}catch ( CMSException& e ) { e.printStackTrace(); }
producer = NULL;
// Close open resources.
try{
if( session != NULL ) session->close();
if( connection != NULL ) connection->close();
}catch ( CMSException& e ) { e.printStackTrace(); }
try{
if( session != NULL ) delete session;
}catch ( CMSException& e ) { e.printStackTrace(); }
session = NULL;
try{
if( connection != NULL ) delete connection;
}catch ( CMSException& e ) { e.printStackTrace(); }
connection = NULL;
}
};
////////////////////////////////////////////////////////////////////////////////
int _cdecl main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
activemq::library::ActiveMQCPP::initializeLibrary();
std::cout << "=====================================================\n";
std::cout << "Starting the example:" << std::endl;
std::cout << "-----------------------------------------------------\n";
// Set the URI to point to the IPAddress of your broker.
// add any optional params to the url to enable things like
// tightMarshalling or tcp logging etc. See the CMS web site for
// a full list of configuration options.
//
// http://activemq.apache.org/cms/
//
// Wire Format Options:
// =====================
// Use either stomp or openwire, the default ports are different for each
//
// Examples:
// tcp://127.0.0.1:61616 default to openwire
// tcp://127.0.0.1:61616?wireFormat=openwire same as above
// tcp://127.0.0.1:61613?wireFormat=stomp use stomp instead
//
std::string brokerURI =
"tcp://mkitchinxp:61616";
// "failover:(tcp://127.0.0.1:61616"
// "?wireFormat=openwire"
// "&connection.useAsyncSend=true"
// "&transport.commandTracingEnabled=true"
// "&transport.tcpTracingEnabled=true"
// "&wireFormat.tightEncodingEnabled=true"
// ")";
std::string userName = "foo";
std::string passWord = "manager";
//============================================================
// Total number of messages for this producer to send.
//============================================================
unsigned int numMessages = 2000;
//============================================================
// This is the Destination Name and URI options. Use this to
// customize where the Producer produces, to have the producer
// use a topic or queue set the 'useTopics' flag.
//============================================================
std::string destURI = "TEST.FOO";
//============================================================
// set to true to use topics instead of queues
// Note in the code above that this causes createTopic or
// createQueue to be used in the producer.
//============================================================
bool useTopics = false;
int ctr = 0;
while ( true )
{
std::cout << "Try #" << (ctr+1) << std::endl;
// Create the producer and run it.
SimpleProducer producer( brokerURI, numMessages, destURI,
userName, passWord, useTopics);
// Publish the given number of Messages
producer.run();
// Before exiting we ensure that all CMS resources are closed.
producer.close();
// wait for retry
Thread::sleep(500);
ctr++;
}
std::cout << "-----------------------------------------------------\n";
std::cout << "Finished with the example." << std::endl;
std::cout << "=====================================================\n";
activemq::library::ActiveMQCPP::shutdownLibrary();
}
Main Exception:
===============
*** BEGIN SERVER-SIDE STACK TRACE ***
Message: User name or password is invalid.
Exception Class java.lang.SecurityException
[FILE: SimpleAuthenticationBroker.java, LINE: 52] occurred in:
org.apache.activemq.security.SimpleAuthenticationBroker.addConnection
[FILE: BrokerFilter.java, LINE: 82] occurred in:
org.apache.activemq.broker.BrokerFilter.addConnection
[FILE: MutableBrokerFilter.java, LINE: 89] occurred in:
org.apache.activemq.broker.MutableBrokerFilter.addConnection
[FILE: TransportConnection.java, LINE: 666] occurred in:
org.apache.activemq.broker.TransportConnection.processAddConnection
[FILE: ManagedTransportConnection.java, LINE: 83] occurred in:
org.apache.activemq.broker.jmx.ManagedTransportConnection.processAddConnection
[FILE: ConnectionInfo.java, LINE: 134] occurred in:
org.apache.activemq.command.ConnectionInfo.visit
[FILE: TransportConnection.java, LINE: 297] occurred in:
org.apache.activemq.broker.TransportConnection.service
[FILE: TransportConnection.java, LINE: 175] occurred in:
org.apache.activemq.broker.TransportConnection$1.onCommand
[FILE: TransportFilter.java, LINE: 68] occurred in:
org.apache.activemq.transport.TransportFilter.onCommand
[FILE: WireFormatNegotiator.java, LINE: 113] occurred in:
org.apache.activemq.transport.WireFormatNegotiator.onCommand
[FILE: InactivityMonitor.java, LINE: 210] occurred in:
org.apache.activemq.transport.InactivityMonitor.onCommand
[FILE: TransportSupport.java, LINE: 84] occurred in:
org.apache.activemq.transport.TransportSupport.doConsume
[FILE: TcpTransport.java, LINE: 203] occurred in:
org.apache.activemq.transport.tcp.TcpTransport.doRun
[FILE: TcpTransport.java, LINE: 185] occurred in:
org.apache.activemq.transport.tcp.TcpTransport.run
[FILE: Thread.java, LINE: 619] occurred in: java.lang.Thread.run
*** END SERVER-SIDE STACK TRACE ***
FILE:
D:\Projects\Work\AMQ\activemq-cpp-library\src\main\activemq\core\ActiveMQConnection.cpp,
LINE: 622
FILE:
D:\Projects\Work\AMQ\activemq-cpp-library\src\main\activemq\core\ActiveMQConnection.cpp,
LINE: 628
FILE:
D:\Projects\Work\AMQ\activemq-cpp-library\src\main\activemq\core\ActiveMQConnection.cpp,
LINE: 326
FILE:
D:\Projects\Work\AMQ\activemq-cpp-library\src\main\activemq\core\ActiveMQConnectionFactory.cpp,
LINE: 151
Main Thread:
============
ntdll.dll!770464f4()
[Frames below may be incorrect and/or missing, no symbols loaded for
ntdll.dll]
ntdll.dll!77045e6c()
KernelBase.dll!7540179c()
>
> activemq-cpp.dll!decaf::internal::util::concurrent::ConditionImpl::wait(decaf::util::concurrent::ConditionHandle
> * condition=0x01321b78, __int64 mills=4294967295, __int64 nanos=0) Line 110
> + 0x12 bytes C++
activemq-cpp.dll!decaf::util::concurrent::Mutex::wait(__int64
millisecs=4294967295, int nanos=0) Line 125 C++
activemq-cpp.dll!decaf::lang::Thread::join(__int64
millisecs=4294967295, unsigned int nanos=0) Line 464 + 0x35 bytes C++
activemq-cpp.dll!decaf::lang::Thread::join() Line 421 C++
activemq-cpp.dll!activemq::transport::IOTransport::close() Line 207 +
0x32 bytes C++
activemq-cpp.dll!activemq::transport::TransportFilter::close() Line
101 + 0x1d bytes C++
activemq-cpp.dll!activemq::transport::tcp::TcpTransport::close() Line
85 C++
activemq-cpp.dll!activemq::transport::TransportFilter::close() Line
101 + 0x1d bytes C++
activemq-cpp.dll!activemq::transport::inactivity::InactivityMonitor::close()
Line 262 C++
activemq-cpp.dll!activemq::wireformat::openwire::OpenWireFormatNegotiator::close()
Line 246 + 0x32 bytes C++
activemq-cpp.dll!activemq::transport::correlator::ResponseCorrelator::close()
Line 264 + 0x32 bytes C++
activemq-cpp.dll!activemq::core::ActiveMQConnectionSupport::shutdownTransport()
Line 126 + 0x32 bytes C++
activemq-cpp.dll!activemq::core::ActiveMQConnectionSupport::~ActiveMQConnectionSupport()
Line 93 C++
msvcr90d.dll!_CallSettingFrame(unsigned long funclet=1242356, unsigned
long pRN=259, unsigned long dwInCode=2156408963) Line 73 Asm
msvcr90d.dll!__FrameUnwindToState(EHRegistrationNode * pRN=0x0012f4f4,
void * pDC=0x0012d7d0, const _s_FuncInfo * pFuncInfo=0x106b90f8, int
targetState=-1) Line 1151 C++
msvcr90d.dll!__InternalCxxFrameHandler(EHExceptionRecord *
pExcept=0x0012dcc0, EHRegistrationNode * pRN=0x0012f4f4, _CONTEXT *
pContext=0x0012d7dc, void * pDC=0x0012d7d0, const _s_FuncInfo *
pFuncInfo=0x106b90f8, int CatchDepth=0, EHRegistrationNode *
pMarkerRN=0x00000000, unsigned char recursive=0) Line 479 + 0x13 bytes C++
msvcr90d.dll!__CxxFrameHandler3(EHExceptionRecord * pExcept=0x0012f4f4,
EHRegistrationNode * pRN=0x0012d7dc, void * pContext=0x0012d7d0, void *
pDC=0x0012f4f4) Line 311 + 0x1f bytes C++
ntdll.dll!770465f9()
ntdll.dll!770465cb()
ntdll.dll!77029012()
ntdll.dll!77046457()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
activemq-cpp.dll!_freelibr...@4() + 0x62035 bytes C++
msvcr90d.dll!585d0b6f()
msvcr90d.dll!585f383f()
ntdll.dll!7704660d()
ntdll.dll!770465cb()
ntdll.dll!77028d3d()
ntdll.dll!77046457()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
activemq-cpp.dll!_freelibr...@4() + 0x62035 bytes C++
msvcr90d.dll!585d0b6f()
msvcr90d.dll!585f383f()
ntdll.dll!7704660d()
ntdll.dll!770465cb()
ntdll.dll!77028d3d()
ntdll.dll!77046457()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
KernelBase.dll!75409617()
activemq-cpp.dll!activemq::core::ActiveMQConnection::syncRequest(decaf::lang::Pointer<activemq::commands::Command,decaf::lang::AtomicRefCounter>
command={...}, unsigned int timeout=0) Line 626 C++
activemq-cpp.dll!activemq::core::ActiveMQConnection::connect() Line
325 C++
activemq-cpp.dll!activemq::core::ActiveMQConnection::ActiveMQConnection(const
decaf::lang::Pointer<activemq::transport::Transport,decaf::lang::AtomicRefCounter>
& transport={...}, const
decaf::lang::Pointer<decaf::util::Properties,decaf::lang::AtomicRefCounter> &
properties={...}) Line 81 C++
activemq-cpp.dll!activemq::core::ActiveMQConnectionFactory::createConnection(const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
url="tcp://mkitchinxp:61616", const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
username="foo", const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
password="manager", const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
clientId="454fee98-6bdb-8a46-c4d0-79950aa6fe8e") Line 143 + 0x3f bytes C++
activemq-cpp.dll!activemq::core::ActiveMQConnectionFactory::createConnection(const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
username="foo", const
std::basic_string<char,std::char_traits<char>,std::allocator<char> > &
password="manager") Line 81 + 0x5e bytes C++
vs2005-activemq-producer.exe!SimpleProducer::run() Line 105 + 0x31
bytes C++
vs2005-activemq-producer.exe!main(int argc=1, char * * argv=0x00281bf0)
Line 257 C++
vs2005-activemq-producer.exe!__tmainCRTStartup() Line 586 + 0x19 bytes
C
vs2005-activemq-producer.exe!mainCRTStartup() Line 403 C
kernel32.dll!76381194()
ntdll.dll!7705b3f5()
ntdll.dll!7705b3c8()
Other Thread #1 (ID=11844)
==========================
> ntdll.dll!770464f4()
[Frames below may be incorrect and/or missing, no symbols loaded for
ntdll.dll]
ntdll.dll!770457dc()
ntdll.dll!7702e6c3()
ntdll.dll!7702f187()
ntdll.dll!7702f3e3()
ntdll.dll!7702f443()
ntdll.dll!7702ef6a()
kernel32.dll!76381194()
ntdll.dll!7705b3f5()
ntdll.dll!7705b3c8()
Other Thread #2 (ID=10544)
==========================
> msvcr90d.dll!_NMSG_WRITE(int rterrnum=25) Line 198 C
msvcr90d.dll!_purecall() Line 54 + 0x7 bytes C
activemq-cpp.dll!activemq::transport::TransportFilter::fire(const
decaf::lang::Exception & ex={...}) Line 49 + 0x19 bytes C++
activemq-cpp.dll!activemq::transport::TransportFilter::onException(const
decaf::lang::Exception & ex={...}) Line 42 C++
activemq-cpp.dll!activemq::transport::TransportFilter::fire(const
decaf::lang::Exception & ex={...}) Line 49 + 0x19 bytes C++
activemq-cpp.dll!activemq::transport::TransportFilter::onException(const
decaf::lang::Exception & ex={...}) Line 42 C++
activemq-cpp.dll!activemq::transport::TransportFilter::fire(const
decaf::lang::Exception & ex={...}) Line 49 + 0x19 bytes C++
activemq-cpp.dll!activemq::transport::TransportFilter::onException(const
decaf::lang::Exception & ex={...}) Line 42 C++
activemq-cpp.dll!activemq::transport::inactivity::InactivityMonitor::onException(const
decaf::lang::Exception & ex={...}) Line 275 C++
activemq-cpp.dll!activemq::transport::TransportFilter::fire(const
decaf::lang::Exception & ex={...}) Line 49 + 0x19 bytes C++
activemq-cpp.dll!activemq::transport::TransportFilter::onException(const
decaf::lang::Exception & ex={...}) Line 42 C++
activemq-cpp.dll!activemq::transport::IOTransport::fire(decaf::lang::Exception
& ex={...}) Line 73 + 0x19 bytes C++
activemq-cpp.dll!activemq::transport::IOTransport::run() Line 247
C++
activemq-cpp.dll!decaf::lang::ThreadProperties::runCallback(decaf::lang::ThreadProperties
* properties=0x01317f20) Line 133 + 0x13 bytes C++
activemq-cpp.dll!`anonymous namespace'::threadWorker(void *
arg=0x01317f20) Line 206 C++
msvcr90d.dll!_callthreadstartex() Line 348 + 0xf bytes C
msvcr90d.dll!_threadstartex(void * ptd=0x013164f0) Line 331 C
kernel32.dll!76381194()
[Frames below may be incorrect and/or missing, no symbols loaded for
kernel32.dll]
ntdll.dll!7705b3f5()
ntdll.dll!7705b3c8()
