You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@qpid.apache.org by Nithesh Shetty <kn...@yahoo.com> on 2009/07/17 12:08:41 UTC
Async get
Hi
I have created three thread two threads read asynchronoulsy from the queue and one thread writes to the queue..
below is the code.
class listener : public MesageListener
{
// constructor
// received function;
};
unsigned int __stdcall ReadThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
SubsriptionManager subscriptions(session);
Listener listener(subscriptions);
subscription.subscribe(listener, "RMS");
subscription.start();
Sleep(10000);
subscriptions.stop();
session.close();
connection.close();
}
catch(const exception &error )
{
cout << error.wha() << endl;
}
return 0;
}
unsigned int __stdcall WriteThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
string buffer("Hello from 0676 ");
string lastdata(256, Q);
string target("QM");
string routeKey("QM_ROUTE_KEY");
Message message'
message.getDeliveryProperties().setRouteKey(routeKey);
for(int i = 0; i < 100; ++i)
{
message.setData(buffer);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
}
message.setData(lastdata);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
session.sync();
session.close();
connection.close()
}
catch(const exception & error)
{
const << error.what();
}
return 0;
}
int main()
{
thread1(readThreadproc)
thread2(writethreadproc);
thread3(readThreadProc);
}
Now the problem is it give exception if u check the call stack then
it show ptr = 0x00000000 line 664
qpid::sys::SystemInfo::getProcessName()
qpid::client::ConnectionHandler::ConnectionHandle()
qpid::client::ConnectionImpl::ConnectionImp()
qpid::client:Connection::open() ( open connection is from read thread)
tested the above code on windows
Awaiting your reply.
Regards,
Nithesh
RE: Async get
Posted by Steve Huston <sh...@riverace.com>.
FYI, the getProcessName() problem has been fixed; see QPID-1998.
-Steve
> -----Original Message-----
> From: Alan Conway [mailto:aconway@redhat.com]
> Sent: Monday, July 20, 2009 2:03 PM
> To: users@qpid.apache.org; Steve Huston
> Subject: Re: Async get
>
>
> Nithesh Shetty wrote:
> [snip]
> >
> > Now the problem is it give exception if u check the call
> stack then
> > it show ptr = 0x00000000 line 664
> > qpid::sys::SystemInfo::getProcessName()
> > qpid::client::ConnectionHandler::ConnectionHandle()
> > qpid::client::ConnectionImpl::ConnectionImp()
> > qpid::client:Connection::open() ( open connection is from
> read thread)
> >
> > tested the above code on windows
> >
>
> There's a bug in the windows code for getProcessName which
> might be the cause of
> your woes.
>
> std::string SystemInfo::getProcessName()
> {
> // Only want info for the current process, so ask for
> something specific.
> // The module info won't be used here but it keeps the
> snapshot limited to
> // the current process so a search through all processes
> is not needed.
> HANDLE snap = CreateToolhelp32Snapshot(TH32CS_SNAPMODULE, 0);
> if (snap == INVALID_HANDLE_VALUE)
> return 0;
> ^^^^^^^^
>
> This return statment is constructing a string from a 0
> pointer which is going to
> lead to a null pointer crash. I don't know if its likely to
> be the cause of your
> crash since I don't know under what circumstances you'd get the
> INVALID_HANDLE_VALUE.
>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org
Re: Async get
Posted by Alan Conway <ac...@redhat.com>.
Nithesh Shetty wrote:
[snip]
>
> Now the problem is it give exception if u check the call stack then
> it show ptr = 0x00000000 line 664
> qpid::sys::SystemInfo::getProcessName()
> qpid::client::ConnectionHandler::ConnectionHandle()
> qpid::client::ConnectionImpl::ConnectionImp()
> qpid::client:Connection::open() ( open connection is from read thread)
>
> tested the above code on windows
>
There's a bug in the windows code for getProcessName which might be the cause of
your woes.
std::string SystemInfo::getProcessName()
{
// Only want info for the current process, so ask for something specific.
// The module info won't be used here but it keeps the snapshot limited to
// the current process so a search through all processes is not needed.
HANDLE snap = CreateToolhelp32Snapshot(TH32CS_SNAPMODULE, 0);
if (snap == INVALID_HANDLE_VALUE)
return 0;
^^^^^^^^
This return statment is constructing a string from a 0 pointer which is going to
lead to a null pointer crash. I don't know if its likely to be the cause of your
crash since I don't know under what circumstances you'd get the
INVALID_HANDLE_VALUE.
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org
Re: Async get
Posted by Nithesh Shetty <kn...@yahoo.com>.
HI,
I have added wait before close in the below mentioned code.
subscriptions.stop();
subscriptions.wait();
session.close();
But in windows it gives the same error
access violation reading location 0x00000
with below mentioned call stack.
on linux it works fine.
Awaiting your reply.
regards,
Nithesh
--- On Fri, 7/17/09, Nithesh Shetty <kn...@yahoo.com> wrote:
From: Nithesh Shetty <kn...@yahoo.com>
Subject: Async get
To: users@qpid.apache.org
Date: Friday, July 17, 2009, 10:08 AM
Hi
I have created three thread two threads read asynchronoulsy from the queue and one thread writes to the queue..
below is the code.
class listener : public MesageListener
{
// constructor
// received function;
};
unsigned int __stdcall ReadThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
SubsriptionManager subscriptions(session);
Listener listener(subscriptions);
subscription.subscribe(listener, "RMS");
subscription.start();
Sleep(10000);
subscriptions.stop();
session.close();
connection.close();
}
catch(const exception &error )
{
cout << error.wha() << endl;
}
return 0;
}
unsigned int __stdcall WriteThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
string buffer("Hello from 0676 ");
string lastdata(256, Q);
string target("QM");
string routeKey("QM_ROUTE_KEY");
Message message'
message.getDeliveryProperties().setRouteKey(routeKey);
for(int i = 0; i < 100; ++i)
{
message.setData(buffer);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
}
message.setData(lastdata);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
session.sync();
session.close();
connection.close()
}
catch(const exception & error)
{
const << error.what();
}
return 0;
}
int main()
{
thread1(readThreadproc)
thread2(writethreadproc);
thread3(readThreadProc);
}
Now the problem is it give exception if u check the call stack then
it show ptr = 0x00000000 line 664
qpid::sys::SystemInfo::getProcessName()
qpid::client::ConnectionHandler::ConnectionHandle()
qpid::client::ConnectionImpl::ConnectionImp()
qpid::client:Connection::open() ( open connection is from read thread)
tested the above code on windows
Awaiting your reply.
Regards,
Nithesh
Re: Async get
Posted by Gordon Sim <gs...@redhat.com>.
Nithesh Shetty wrote:
> SubsriptionManager subscriptions(session);
> Listener listener(subscriptions);
> subscription.subscribe(listener, "RMS");
> subscription.start();
>
> Sleep(10000);
> subscriptions.stop();
You need to wait for the dispatcher thread to exit before killing the
session. Add the following line here:
subscriptions.wait();
> session.close();
> connection.close();
>
> }
> catch(const exception &error )
> {
> cout << error.wha() << endl;
> }
> return 0;
>
> }
>
>
> unsigned int __stdcall WriteThreadProc(void *)
> {
> const char* host = "172.20.112.111"
> int port = 5004;
> try
> {
> Connection connection;
> ConnectionSettings lcConnSettings;
> lcConnSettings.host = host;
> lcConnSettings.port = port;
> lcConnSettings.maxFrameSize = 2048;
> lcConnSettings.username = "guest";
> lcConnSettings.password = "guest";
> connection.open(lcConnSettings);
> Session session = connection.newSession();
>
> QueueOrderingPolicy cQOP = FIFO;
> QueueOptions lcQueueOptions;
> lcQueueOptions.setOrdering(cQOP);
> session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
> session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
> session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
>
> string buffer("Hello from 0676 ");
> string lastdata(256, Q);
> string target("QM");
> string routeKey("QM_ROUTE_KEY");
> Message message'
> message.getDeliveryProperties().setRouteKey(routeKey);
> for(int i = 0; i < 100; ++i)
> {
> message.setData(buffer);
> async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
> ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
>
> }
> message.setData(lastdata);
> async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
> ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
>
> session.sync();
> session.close();
> connection.close()
>
> }
> catch(const exception & error)
> {
> const << error.what();
> }
>
> return 0;
> }
>
> int main()
> {
> thread1(readThreadproc)
> thread2(writethreadproc);
> thread3(readThreadProc);
> }
>
> Now the problem is it give exception if u check the call stack then
> it show ptr = 0x00000000 line 664
> qpid::sys::SystemInfo::getProcessName()
> qpid::client::ConnectionHandler::ConnectionHandle()
> qpid::client::ConnectionImpl::ConnectionImp()
> qpid::client:Connection::open() ( open connection is from read thread)
>
> tested the above code on windows
>
> Awaiting your reply.
>
> Regards,
> Nithesh
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org
Re: Async get
Posted by Nithesh Shetty <kn...@yahoo.com>.
when i try the same code on linux it says
guest.4578c63b-0089-48bf-8c26-76d0afd78f8b error: not-attached: Session
guest.4578c63b-0089-48bf-8c26-76d0afd78f8b isn't attached (qpid/client/SessionImpl.cpp:736)
terminate called after throwing an instance of 'qpid::framing::NotAttachedException'
what() : not attached: Session guest.4578c63b-0089-48bf-8c26-76d0afd78f8b isn;t attached (qpid/client/SessionImpl.cpp:736)
Aborted (core dumped)
Awaiting your reply.
regards,
Nithesh
--- On Fri, 7/17/09, Nithesh Shetty <kn...@yahoo.com> wrote:
From: Nithesh Shetty <kn...@yahoo.com>
Subject: Async get
To: users@qpid.apache.org
Date: Friday, July 17, 2009, 10:08 AM
Hi
I have created three thread two threads read asynchronoulsy from the queue and one thread writes to the queue..
below is the code.
class listener : public MesageListener
{
// constructor
// received function;
};
unsigned int __stdcall ReadThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
SubsriptionManager subscriptions(session);
Listener listener(subscriptions);
subscription.subscribe(listener, "RMS");
subscription.start();
Sleep(10000);
subscriptions.stop();
session.close();
connection.close();
}
catch(const exception &error )
{
cout << error.wha() << endl;
}
return 0;
}
unsigned int __stdcall WriteThreadProc(void *)
{
const char* host = "172.20.112.111"
int port = 5004;
try
{
Connection connection;
ConnectionSettings lcConnSettings;
lcConnSettings.host = host;
lcConnSettings.port = port;
lcConnSettings.maxFrameSize = 2048;
lcConnSettings.username = "guest";
lcConnSettings.password = "guest";
connection.open(lcConnSettings);
Session session = connection.newSession();
QueueOrderingPolicy cQOP = FIFO;
QueueOptions lcQueueOptions;
lcQueueOptions.setOrdering(cQOP);
session.exchangeDeclare(arg::exchange = "QM", arg::exclusive = false; arg:autoDelete = false; arg::type = "direct", );
session.queueDeclare(arg::queue = "RMS", arg::exclusive = false, arg::qutoDelete = false, arg::arguments = lcQueueOptions);
session.exchangeBind(arg::exchange = "QM", arg::queue = "RMS", arg::bindingKey = "QM_ROUTE_KEY");
string buffer("Hello from 0676 ");
string lastdata(256, Q);
string target("QM");
string routeKey("QM_ROUTE_KEY");
Message message'
message.getDeliveryProperties().setRouteKey(routeKey);
for(int i = 0; i < 100; ++i)
{
message.setData(buffer);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
}
message.setData(lastdata);
async(session).messageTransfer(arg::content=message, arg::destination=target, arg::acceptMode=message::ACCEPT_MODE_NONE, arg::acquireMode=message::
ACQUIRE_MODE_PRE_ACQUIRED, arg::sync = false);
session.sync();
session.close();
connection.close()
}
catch(const exception & error)
{
const << error.what();
}
return 0;
}
int main()
{
thread1(readThreadproc)
thread2(writethreadproc);
thread3(readThreadProc);
}
Now the problem is it give exception if u check the call stack then
it show ptr = 0x00000000 line 664
qpid::sys::SystemInfo::getProcessName()
qpid::client::ConnectionHandler::ConnectionHandle()
qpid::client::ConnectionImpl::ConnectionImp()
qpid::client:Connection::open() ( open connection is from read thread)
tested the above code on windows
Awaiting your reply.
Regards,
Nithesh