You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@qpid.apache.org by kb...@texnolyze.us on 2011/11/13 11:14:55 UTC
qpid pub/sub running very slow
I am new to qpid AMQP broker.
Based on the examples provided with qipd,I have made a simple cpp
program to test the pub/sub throughput on the broker. I am noticing
that the throughput very slow for some reason.
I think this is something wrong in my test code. When I run my test
code, the broker does not seem to be doing much and my app and the
broker are mostly in a stalled/sleep state.
I am not sure what I am doing wrong here. I would expect that when I
run the test, the broker should become busy processing messages as fast
as it can. But it does not seem to be the case here.
qpid broker is running with all defaults ( qpidd -d )
When I run qpid-preftest everything seems to be running smooth and I
get good throughput.
-------------- Code Segment Start ---------------
//pubsub_preftest.cpp
//test sending/receiving binary messages though a QPID broker via
Pub/Sub method.
//
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Session.h>
#include <qpid/messaging/Address.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Session.h>
#include <qpid/types/Variant.h>
#include <iostream>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
#include <memory.h>
// prototypes
void makeRandomBinaryMessage( unsigned char* buffer, int size );
int timeval_subtract (struct timeval *result, struct timeval *x, struct
timeval *y );
int main(int argc, char** argv) {
// get message size from command line..
int message_size = atoi( argc>1 ? argv[1] : "2048" );
// get number of messages to run..
int message_count = atoi( argc>2 ? argv[2] : "10000" );
std::cout << "Parameters: message_size = " << message_size << ";
message_count = " << message_count << " \n";
// check input parameters
if ( ( message_size < 4 ) || ( message_count < 1 ) ) {
std::cout << "invalid input parameters..\n";
return -1;
}
// time stamps
timeval tv_start, tv_stop, tv_delta;
// create message buffer of size "message_size" bytes
unsigned char *txMessage_buffer = (unsigned char *)malloc(
message_size );
// fill buffer with random stuff
makeRandomBinaryMessage( txMessage_buffer, message_size );
// AMQP Connection parameters
const char* url = "amqp:tcp:127.0.0.1:5672";
const char* addr = "pubsub.topic; {create: always, node: {type:
topic}}";
std::string connectionOptions = "";
// Qpid objects
qpid::messaging::Connection connection(url, connectionOptions);
qpid::messaging::Session session;
qpid::messaging::Receiver receiver;
qpid::messaging::Sender sender;
qpid::messaging::Message pubMessage;
qpid::messaging::Message subMessage;
// loop counter
int lc = 0;
double delta_time;
// Start it up..
try {
// make connection to broker
connection.open();
// create a session on the broker
session = connection.createSession();
// create a sender on this session
sender = session.createSender(addr);
// create a receiver on this session
receiver = session.createReceiver(addr);
// start the do-stuff loop
gettimeofday(&tv_start, NULL);
// print new line
//std::cout << "\n";
for (lc=0; lc < message_count; lc++) {
// binary copy counter as first 4 bytes of buffer
memcpy( (void *)txMessage_buffer, (void *)&lc, 4 );
pubMessage.setContent( (const char *)txMessage_buffer,
message_size );
sender.send( pubMessage, true );
// receive message from broker
subMessage = receiver.fetch();
// print status update every 1%
if ( ( lc % ( message_count / 100 ) ) == 0 ) {
// print
std::cout << "\x1B[1K"; // voodoo to clear line on a
vt100 terminal
std::cout << lc << "/" << message_count << " messages
transfered..\n\x1BM";
}
session.acknowledge( subMessage, false );
}
std::cout << "\n";
gettimeofday(&tv_stop, NULL);
std::cout << "finished sending " << message_count << " messages
of size " << message_size << " bytes each.";
timeval_subtract( &tv_delta, &tv_stop, &tv_start);
delta_time = tv_delta.tv_sec + ( tv_delta.tv_usec / 1e6 );
std::cout << "Opeation completed in " << delta_time << "
seconds\n";
} catch(const std::exception& error) {
std::cout << error.what() << std::endl;
connection.close();
return -1;
}
connection.close();
return 0;
}
// simple method to fill a buffer with some random stuff
void makeRandomBinaryMessage( unsigned char* buffer, int size ) {
/* random seed, just using time */
srand ( (unsigned int)time(NULL) );
/* now generate random value to fill buffer */
/* note: this data is very random, but random enough for this. */
int i=0;
for (i; i<size; i++) {
buffer[i] = (char)( rand() % 255 );
}
}
// figure out the difference between to timeval structures
int timeval_subtract (struct timeval *result, struct timeval *x, struct
timeval *y )
{
/* Perform the carry for the later subtraction by updating y. */
if (x->tv_usec < y->tv_usec) {
int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
y->tv_usec -= 1000000 * nsec;
y->tv_sec += nsec;
}
if (x->tv_usec - y->tv_usec > 1000000) {
int nsec = (x->tv_usec - y->tv_usec) / 1000000;
y->tv_usec += 1000000 * nsec;
y->tv_sec -= nsec;
}
/* Compute the time remaining to wait.
tv_usec is certainly positive. */
result->tv_sec = x->tv_sec - y->tv_sec;
result->tv_usec = x->tv_usec - y->tv_usec;
/* Return 1 if result is negative. */
return x->tv_sec < y->tv_sec;
}
-------------- Code Segment End ---------------
--
Peter Fetterer (kb3gtn)
* Failure is not an option, but a standard feature to be avoided.
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org
Re: qpid pub/sub running very slow
Posted by kb...@texnolyze.us.
Based on some suggests from Fraser, I was able to make a pub/sub
performance test tool that seems to work at the max rate of the broker.
Here is my code, so that other may use it as a reference.
----- Code Start -----
//pubsub_preftest2.cpp
// Test sending binary messages though a QPID broker via Pub/Sub
method.
// This program provides a system for characterizing the performance of
the AMQP broker
//
// Command line arguments:
// ./pubsub_preftest2 <message size> <message count>
//
//Peter Fetterer <kb...@texnolyze.us>
//
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Session.h>
#include <qpid/messaging/Address.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Session.h>
#include <qpid/types/Variant.h>
#include <iostream>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
#include <memory.h>
#include <pthread.h>
#include <unistd.h>
// prototypes
void makeRandomBinaryMessage( unsigned char* buffer, int size );
int timeval_subtract (struct timeval *result, struct timeval *x, struct
timeval *y );
// Arguments for worker threads
struct threadArgs_t {
// command line params
int message_size;
int message_count;
// AMQP Connection parameters
char* url;
char* addr;
// name of the worker thread
std::string threadName;
int threadResult;
int start;
};
// topic producer thread worker
void *publisher( void *args ) {
// pull args
threadArgs_t *myArgs = (threadArgs_t *)args;
// Qpid objects
qpid::messaging::Connection connection( myArgs->url, "");
qpid::messaging::Session session;
qpid::messaging::Sender sender;
qpid::messaging::Message pubMessage;
// time stamps
timeval tv_start, tv_stop, tv_delta;
// create message buffer of size "message_size" bytes
unsigned char *txMessage_buffer = (unsigned char *)malloc(
myArgs->message_size );
// fill buffer with random stuff
makeRandomBinaryMessage( txMessage_buffer, myArgs->message_size );
// Start it up..
try {
// make connection to broker
connection.open();
// create a session on the broker
session = connection.createSession();
// create a sender on this session
sender = session.createSender(myArgs->addr);
sender.setCapacity(1048576/(myArgs->message_size*2)); // no
more than a 0.5 Mb worth of outsanding messages
// start the do-stuff loop
gettimeofday(&tv_start, NULL);
int lc = 0;
double delta_time;
// do a loop waiting for start to be triggered
while ( myArgs->start == 0 ) {
sleep(1); // sleep
}
std::cout << "Sender \"" << myArgs->threadName << "\"
Starting..\n";
for (lc=0; lc < myArgs->message_count; lc++) {
// binary copy counter as first 4 bytes of buffer
memcpy( (void *)txMessage_buffer, (void *)&lc, 4 );
pubMessage.setContent( (const char *)txMessage_buffer,
myArgs->message_size );
// blocks if sender capacity limit is reached.
// -- which means the receiver(s) are not keeping up..
sender.send( pubMessage, false );
}
gettimeofday(&tv_stop, NULL);
std::cout << "Sender \"" << myArgs->threadName << "\" finished;
sent " << myArgs->message_count << " messages of size " <<
myArgs->message_size << " bytes.\n";
timeval_subtract( &tv_delta, &tv_stop, &tv_start);
delta_time = tv_delta.tv_sec + ( tv_delta.tv_usec / 1e6 );
std::cout << "Sender \"" << myArgs->threadName << "\" Opeation
completed in " << delta_time << " seconds\n";
std::cout << "Sender \"" << myArgs->threadName << "\" datarate
" << (double)(myArgs->message_count*myArgs->message_size)/delta_time <<
" Bytes/second -> "
<<
(double)(myArgs->message_count*myArgs->message_size)/delta_time * 8 << "
bits/second \n";
} catch(const std::exception& error) {
std::cout << "Sender \"" << myArgs->threadName << "\"
exception: " << error.what() << std::endl;
connection.close();
myArgs->threadResult = -1;
return &myArgs->threadResult;
}
connection.close();
myArgs->threadResult = 0;
return &myArgs->threadResult;
}
// topic consumer thread worker
void *consumer( void *args ) {
// pull args
threadArgs_t *myArgs = (threadArgs_t *)args;
// Qpid objects
qpid::messaging::Connection connection(myArgs->url, "");
qpid::messaging::Session session;
qpid::messaging::Receiver receiver;
qpid::messaging::Message subMessage;
// time stamps
timeval tv_start, tv_stop, tv_delta;
int rxCount = 0;
int rxCount_rx = 0;
double delta_time;
// Initialise this outside your main loop
unsigned int unackedCount = 0;
// Start it up..
try {
// make connection to broker
connection.open();
// create a session on the broker
session = connection.createSession();
// create a receiver on this session
receiver = session.createReceiver(myArgs->addr);
receiver.setCapacity(1048576/(myArgs->message_size*2));
// start the do-stuff loop
gettimeofday(&tv_start, NULL);
// print new line
// std::cout << "\n";
int running = 1;
std::cout << "Receiver \"" << myArgs->threadName << "\" has
started.. \n";
while( running ) {
// this receiver blocks for 1 sec before giving up..
if ( receiver.fetch( subMessage,
qpid::messaging::Duration::SECOND * 1 ) == false ) {
// no message received, exit
std::cout << "Receiver \"" << myArgs->threadName << "\"
says receiver timed out.. exit.. \n";
running = 0;
}
if ( ( rxCount % ( myArgs->message_count / 10 ) ) == 0 ) {
// print
std::cout << "Receiver \"" << myArgs->threadName << "\"
says " << rxCount << " messages transfered..\n";
}
// verify count, copy first 4 bytes of tx message into an
int
if (subMessage.getContentSize() > 4 ) {
memcpy( (void *)&rxCount_rx, (void *)(char
*)subMessage.getContentPtr(), 4 );
if ( rxCount_rx == (myArgs->message_count)-1 ) {
// last expected message received
running = 0;
}
}
// is rxCount not what we expect..
if ( rxCount != rxCount_rx ) {
std::cout << "Receiver \"" << myArgs->threadName << "\"
Error recieved count:" << rxCount_rx << " expected " <<rxCount <<
std::endl;
rxCount = rxCount_rx;
}
// increment rxCount for next expected count
rxCount++;
// Try replacing your "session.acknowledge( subMessage,
false ); " with this:
// bulk acknowledge speeds stuff up..
unackedCount++;
if (unackedCount > receiver.getCapacity()/2) {
session.acknowledge();
unackedCount = 0;
}
}
std::cout << "\n";
std::cout << "Receiver \"" << myArgs->threadName << "\" RxCount
= " << rxCount << std::endl;
gettimeofday(&tv_stop, NULL);
std::cout << "Receiver \"" << myArgs->threadName << "\"
finished Receiveing " << myArgs->message_count << " messages of size "
<< myArgs->message_size << " bytes each.\n";
timeval_subtract( &tv_delta, &tv_stop, &tv_start);
delta_time = tv_delta.tv_sec + ( tv_delta.tv_usec / 1e6 );
std::cout << "Receiver \"" << myArgs->threadName << "\"
Opeation completed in " << delta_time << " seconds\n";
std::cout << "Receiver \"" << myArgs->threadName << "\"
datarate " <<
((double)(myArgs->message_count*myArgs->message_size))/delta_time << "
Bytes/second -> "
<<
((double)((myArgs->message_count*myArgs->message_size)/delta_time)) * 8
<< " bits/second \n";
} catch(const std::exception& error) {
std::cout << "Receiver \"" << myArgs->threadName << "\"
exception: " << error.what() << std::endl;
connection.close();
myArgs->threadResult = -1;
return &myArgs->threadResult;
}
connection.close();
myArgs->threadResult = 0;
return &myArgs->threadResult;
}
// <program name> <message_size> <message count>
int main(int argc, char** argv) {
// get message size from command line..
int message_size = atoi( argc>1 ? argv[1] : "2048" );
// get number of messages to run..
int message_count = atoi( argc>2 ? argv[2] : "100000" );
std::cout << "Parameters: message_size = " << message_size << ";
message_count = " << message_count << " \n";
// check input parameters
if ( ( message_size < 4 ) || ( message_count < 1 ) ) {
std::cout << "invalid input parameters..\n";
return -1;
}
// AMQP Connection parameters
const char* url = "amqp:tcp:127.0.0.1:5672";
const char* addr = "pubsub.topic; {create: always, node: {durable:
false, type: topic}, link: {reliability: unreliable}}";
// create threadArgs objects for each thread
threadArgs_t sender1_args;
threadArgs_t receiver1_args;
threadArgs_t receiver2_args;
// pthread objects
pthread_t sender1, receiver1, receiver2;
// return value storage for threads
int threadReturn1, threadReturn2, threadReturn3;
sender1_args.message_size = message_size;
sender1_args.message_count = message_count;
sender1_args.url = (char *)url;
sender1_args.addr = (char *)addr;
sender1_args.threadName = "sender";
sender1_args.start = 0; // don't start right away
receiver1_args.message_size = message_size;
receiver1_args.message_count = message_count;
receiver1_args.url = (char *)url;
receiver1_args.addr = (char *)addr;
receiver1_args.threadName = "receiver1";
receiver2_args.message_size = message_size;
receiver2_args.message_count = message_count;
receiver2_args.url = (char *)url;
receiver2_args.addr = (char *)addr;
receiver2_args.threadName = "receiver2";
// create threads -- 1 publisher and 2 comsumers
threadReturn1 = pthread_create( &sender1, NULL, publisher, (void*)
&sender1_args );
threadReturn2 = pthread_create( &receiver1, NULL, consumer, (void*)
&receiver1_args );
threadReturn3 = pthread_create( &receiver2, NULL, consumer, (void*)
&receiver2_args );
// everyone up a and waiting.. start sender
sender1_args.start = 1;
/* Wait till threads are complete before main continues. Unless we
*/
/* wait we run the risk of executing an exit which will terminate
*/
/* the process and all threads before the threads have completed.
*/
pthread_join( sender1, NULL);
pthread_join( receiver1, NULL);
pthread_join( receiver2, NULL);
std::cout << "Exit Status:\n";
std::cout << "sender1 returned: " << threadReturn1 << std::endl;
std::cout << "receiver1 returned: " << threadReturn2 << std::endl;
std::cout << "receiver2 returned: " << threadReturn3 << std::endl;
std::cout << "Main Complete.. \n";
return 0;
}
// simple method to fill a buffer with some random stuff
void makeRandomBinaryMessage( unsigned char* buffer, int size ) {
/* random seed, just using time */
srand ( (unsigned int)time(NULL) );
/* now generate random value to fill buffer */
/* note: this data is very random, but random enough for this. */
int i=0;
for (i; i<size; i++) {
buffer[i] = (char)( rand() % 255 );
}
}
// figure out the difference between to timeval structures
int timeval_subtract (struct timeval *result, struct timeval *x, struct
timeval *y )
{
/* Perform the carry for the later subtraction by updating y. */
if (x->tv_usec < y->tv_usec) {
int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
y->tv_usec -= 1000000 * nsec;
y->tv_sec += nsec;
}
if (x->tv_usec - y->tv_usec > 1000000) {
int nsec = (x->tv_usec - y->tv_usec) / 1000000;
y->tv_usec += 1000000 * nsec;
y->tv_sec -= nsec;
}
/* Compute the time remaining to wait.
tv_usec is certainly positive. */
result->tv_sec = x->tv_sec - y->tv_sec;
result->tv_usec = x->tv_usec - y->tv_usec;
/* Return 1 if result is negative. */
return x->tv_sec < y->tv_sec;
}
----- Code Stop -----
1 liner build instructions, assuming you have qpid in your include path
and such..
g++ pubsub_perftest2.cpp -lqpidmessaging -lpthread -o pubsub_perftest2
Output I got on a Intel(R) Core(TM) i7 CPU 960 @ 3.20GHz on a X58
triple channel memory configuration system:
Sender "sender" finished; sent 1000000 messages of size 2048 bytes.
Sender "sender" Opeation completed in 15.7477 seconds
Sender "sender" datarate 1.30051e+08 Bytes/second -> 1.04041e+09
bits/second
Receiver "receiver1" RxCount = 1000000
Receiver "receiver1" finished Receiveing 1000000 messages of size 2048
bytes each.
Receiver "receiver1" Opeation completed in 15.8164 seconds
Receiver "receiver1" datarate 1.29486e+08 Bytes/second -> 1.03589e+09
bits/second
Receiver "receiver2" RxCount = 1000000
Receiver "receiver2" finished Receiveing 1000000 messages of size 2048
bytes each.
Receiver "receiver2" Opeation completed in 15.826 seconds
Receiver "receiver2" datarate 1.29408e+08 Bytes/second -> 1.03526e+09
bits/second
--
Peter Fetterer <KB...@texnolyze.us>
On Sun, 13 Nov 2011 11:16:03 +0000, Fraser Adams wrote:
> I skimmed through your code and noticed a couple of things.
> 1) You don't have receiver prefetch enabled if you do:
> receiver.setCapacity(500); // Enable receiver prefetch
> after your call:
> receiver = session.createReceiver(addr);
> you should see a big improvement
> 2) Your main loop has a coupled send and receive (fetch), so
> everything will go at the speed of the slowest. Qpid provides
> asynchronous send and receive but your code isn't seeing the benefit
> of that (it's largely behaving like request/response).
> I expect part of the perceived performance issue is also due to
> acking every message. Try something like:
>
> // Initialise this outside your main loop
> unsigned int unackedCount = 0;
>
>
> // Try replacing your "session.acknowledge( subMessage, false ); "
> with this:
> unackedCount++;
> if (unackedCount > receiver.getCapacity()/2) {
> //cout << "auto acknowledged message #" << count << endl;
> session.acknowledge();
> unackedCount = 0;
> }
>
>
> The above code acknowledges when the unackedCount exceeds half the
> receiver capacity.
>
>
> Also note that qpid-perftest actually uses unreliable messaging,
> which means that messages don't need acknowledged at all. You could
> try an address like:
>
> const char* addr = "pubsub.topic; {create: always, node: {type:
> topic}, link: {reliability: unreliable}}";
>
>
> If my syntax is correct then you should be able to avoid
> acknowledgement and get a performance gain (though you risk increased
> message loss).
>
> Hope this helps a bit.
> Frase
>
>
>
>
> kb3gtn@texnolyze.us wrote:
>> I am new to qpid AMQP broker.
>>
>> Based on the examples provided with qipd,I have made a simple cpp
>> program to test the pub/sub throughput on the broker. I am noticing
>> that the throughput very slow for some reason.
>>
>> I think this is something wrong in my test code. When I run my test
>> code, the broker does not seem to be doing much and my app and the
>> broker are mostly in a stalled/sleep state.
>>
>> I am not sure what I am doing wrong here. I would expect that when
>> I run the test, the broker should become busy processing messages as
>> fast as it can. But it does not seem to be the case here.
>>
>> qpid broker is running with all defaults ( qpidd -d )
>> When I run qpid-preftest everything seems to be running smooth and I
>> get good throughput.
>>
>> -------------- Code Segment Start ---------------
>> //pubsub_preftest.cpp
>> //test sending/receiving binary messages though a QPID broker via
>> Pub/Sub method.
>> //
>> #include <qpid/messaging/Connection.h>
>> #include <qpid/messaging/Message.h>
>> #include <qpid/messaging/Receiver.h>
>> #include <qpid/messaging/Session.h>
>> #include <qpid/messaging/Address.h>
>> #include <qpid/messaging/Sender.h>
>> #include <qpid/messaging/Session.h>
>> #include <qpid/types/Variant.h>
>>
>> #include <iostream>
>>
>> #include <stdlib.h>
>> #include <sys/time.h>
>> #include <time.h>
>> #include <memory.h>
>>
>> // prototypes
>> void makeRandomBinaryMessage( unsigned char* buffer, int size );
>> int timeval_subtract (struct timeval *result, struct timeval *x,
>> struct timeval *y );
>>
>> int main(int argc, char** argv) {
>>
>> // get message size from command line..
>> int message_size = atoi( argc>1 ? argv[1] : "2048" );
>> // get number of messages to run..
>> int message_count = atoi( argc>2 ? argv[2] : "10000" );
>>
>> std::cout << "Parameters: message_size = " << message_size << ";
>> message_count = " << message_count << " \n";
>>
>> // check input parameters
>> if ( ( message_size < 4 ) || ( message_count < 1 ) ) {
>> std::cout << "invalid input parameters..\n";
>> return -1;
>> }
>>
>> // time stamps
>> timeval tv_start, tv_stop, tv_delta;
>>
>> // create message buffer of size "message_size" bytes
>> unsigned char *txMessage_buffer = (unsigned char *)malloc(
>> message_size );
>>
>> // fill buffer with random stuff
>> makeRandomBinaryMessage( txMessage_buffer, message_size );
>>
>> // AMQP Connection parameters
>> const char* url = "amqp:tcp:127.0.0.1:5672";
>> const char* addr = "pubsub.topic; {create: always, node: {type:
>> topic}}";
>> std::string connectionOptions = "";
>>
>> // Qpid objects
>> qpid::messaging::Connection connection(url, connectionOptions);
>> qpid::messaging::Session session;
>> qpid::messaging::Receiver receiver;
>> qpid::messaging::Sender sender;
>> qpid::messaging::Message pubMessage;
>> qpid::messaging::Message subMessage;
>>
>> // loop counter
>> int lc = 0;
>> double delta_time;
>>
>> // Start it up..
>> try {
>> // make connection to broker
>> connection.open();
>> // create a session on the broker
>> session = connection.createSession();
>> // create a sender on this session
>> sender = session.createSender(addr);
>> // create a receiver on this session
>> receiver = session.createReceiver(addr);
>>
>> // start the do-stuff loop
>> gettimeofday(&tv_start, NULL);
>>
>> // print new line
>> //std::cout << "\n";
>>
>> for (lc=0; lc < message_count; lc++) {
>> // binary copy counter as first 4 bytes of buffer
>> memcpy( (void *)txMessage_buffer, (void *)&lc, 4 );
>> pubMessage.setContent( (const char *)txMessage_buffer,
>> message_size );
>> sender.send( pubMessage, true );
>> // receive message from broker
>> subMessage = receiver.fetch();
>> // print status update every 1%
>> if ( ( lc % ( message_count / 100 ) ) == 0 ) {
>> // print
>> std::cout << "\x1B[1K"; // voodoo to clear line on
>> a vt100 terminal
>> std::cout << lc << "/" << message_count << "
>> messages transfered..\n\x1BM";
>>
>> }
>> session.acknowledge( subMessage, false );
>> }
>> std::cout << "\n";
>> gettimeofday(&tv_stop, NULL);
>> std::cout << "finished sending " << message_count << "
>> messages of size " << message_size << " bytes each.";
>> timeval_subtract( &tv_delta, &tv_stop, &tv_start);
>> delta_time = tv_delta.tv_sec + ( tv_delta.tv_usec / 1e6 );
>> std::cout << "Opeation completed in " << delta_time << "
>> seconds\n";
>> } catch(const std::exception& error) {
>> std::cout << error.what() << std::endl;
>> connection.close();
>> return -1;
>> }
>>
>> connection.close();
>> return 0;
>> }
>>
>> // simple method to fill a buffer with some random stuff
>> void makeRandomBinaryMessage( unsigned char* buffer, int size ) {
>> /* random seed, just using time */
>> srand ( (unsigned int)time(NULL) );
>>
>> /* now generate random value to fill buffer */
>> /* note: this data is very random, but random enough for this.
>> */
>> int i=0;
>> for (i; i<size; i++) {
>> buffer[i] = (char)( rand() % 255 );
>> }
>> }
>>
>> // figure out the difference between to timeval structures
>> int timeval_subtract (struct timeval *result, struct timeval *x,
>> struct timeval *y )
>> {
>> /* Perform the carry for the later subtraction by updating y. */
>> if (x->tv_usec < y->tv_usec) {
>> int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
>> y->tv_usec -= 1000000 * nsec;
>> y->tv_sec += nsec;
>> }
>> if (x->tv_usec - y->tv_usec > 1000000) {
>> int nsec = (x->tv_usec - y->tv_usec) / 1000000;
>> y->tv_usec += 1000000 * nsec;
>> y->tv_sec -= nsec;
>> }
>>
>> /* Compute the time remaining to wait.
>> tv_usec is certainly positive. */
>> result->tv_sec = x->tv_sec - y->tv_sec;
>> result->tv_usec = x->tv_usec - y->tv_usec;
>>
>> /* Return 1 if result is negative. */
>> return x->tv_sec < y->tv_sec;
>> }
>>
>> -------------- Code Segment End ---------------
>>
>> -- Peter Fetterer (kb3gtn)
>> * Failure is not an option, but a standard feature to be avoided.
>>
>>
>> ---------------------------------------------------------------------
>> Apache Qpid - AMQP Messaging Implementation
>> Project: http://qpid.apache.org
>> Use/Interact: mailto:users-subscribe@qpid.apache.org
>>
>>
>
>
> ---------------------------------------------------------------------
> Apache Qpid - AMQP Messaging Implementation
> Project: http://qpid.apache.org
> Use/Interact: mailto:users-subscribe@qpid.apache.org
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org
Re: qpid pub/sub running very slow
Posted by Fraser Adams <fr...@blueyonder.co.uk>.
I skimmed through your code and noticed a couple of things.
1) You don't have receiver prefetch enabled if you do:
receiver.setCapacity(500); // Enable receiver prefetch
after your call:
receiver = session.createReceiver(addr);
you should see a big improvement
2) Your main loop has a coupled send and receive (fetch), so everything
will go at the speed of the slowest. Qpid provides asynchronous send and
receive but your code isn't seeing the benefit of that (it's largely
behaving like request/response).
I expect part of the perceived performance issue is also due to acking
every message. Try something like:
// Initialise this outside your main loop
unsigned int unackedCount = 0;
// Try replacing your "session.acknowledge( subMessage, false ); " with
this:
unackedCount++;
if (unackedCount > receiver.getCapacity()/2) {
//cout << "auto acknowledged message #" << count << endl;
session.acknowledge();
unackedCount = 0;
}
The above code acknowledges when the unackedCount exceeds half the
receiver capacity.
Also note that qpid-perftest actually uses unreliable messaging, which
means that messages don't need acknowledged at all. You could try an
address like:
const char* addr = "pubsub.topic; {create: always, node: {type: topic},
link: {reliability: unreliable}}";
If my syntax is correct then you should be able to avoid acknowledgement
and get a performance gain (though you risk increased message loss).
Hope this helps a bit.
Frase
kb3gtn@texnolyze.us wrote:
> I am new to qpid AMQP broker.
>
> Based on the examples provided with qipd,I have made a simple cpp
> program to test the pub/sub throughput on the broker. I am noticing
> that the throughput very slow for some reason.
>
> I think this is something wrong in my test code. When I run my test
> code, the broker does not seem to be doing much and my app and the
> broker are mostly in a stalled/sleep state.
>
> I am not sure what I am doing wrong here. I would expect that when I
> run the test, the broker should become busy processing messages as
> fast as it can. But it does not seem to be the case here.
>
> qpid broker is running with all defaults ( qpidd -d )
> When I run qpid-preftest everything seems to be running smooth and I
> get good throughput.
>
> -------------- Code Segment Start ---------------
> //pubsub_preftest.cpp
> //test sending/receiving binary messages though a QPID broker via
> Pub/Sub method.
> //
> #include <qpid/messaging/Connection.h>
> #include <qpid/messaging/Message.h>
> #include <qpid/messaging/Receiver.h>
> #include <qpid/messaging/Session.h>
> #include <qpid/messaging/Address.h>
> #include <qpid/messaging/Sender.h>
> #include <qpid/messaging/Session.h>
> #include <qpid/types/Variant.h>
>
> #include <iostream>
>
> #include <stdlib.h>
> #include <sys/time.h>
> #include <time.h>
> #include <memory.h>
>
> // prototypes
> void makeRandomBinaryMessage( unsigned char* buffer, int size );
> int timeval_subtract (struct timeval *result, struct timeval *x,
> struct timeval *y );
>
> int main(int argc, char** argv) {
>
> // get message size from command line..
> int message_size = atoi( argc>1 ? argv[1] : "2048" );
> // get number of messages to run..
> int message_count = atoi( argc>2 ? argv[2] : "10000" );
>
> std::cout << "Parameters: message_size = " << message_size << ";
> message_count = " << message_count << " \n";
>
> // check input parameters
> if ( ( message_size < 4 ) || ( message_count < 1 ) ) {
> std::cout << "invalid input parameters..\n";
> return -1;
> }
>
> // time stamps
> timeval tv_start, tv_stop, tv_delta;
>
> // create message buffer of size "message_size" bytes
> unsigned char *txMessage_buffer = (unsigned char *)malloc(
> message_size );
>
> // fill buffer with random stuff
> makeRandomBinaryMessage( txMessage_buffer, message_size );
>
> // AMQP Connection parameters
> const char* url = "amqp:tcp:127.0.0.1:5672";
> const char* addr = "pubsub.topic; {create: always, node: {type:
> topic}}";
> std::string connectionOptions = "";
>
> // Qpid objects
> qpid::messaging::Connection connection(url, connectionOptions);
> qpid::messaging::Session session;
> qpid::messaging::Receiver receiver;
> qpid::messaging::Sender sender;
> qpid::messaging::Message pubMessage;
> qpid::messaging::Message subMessage;
>
> // loop counter
> int lc = 0;
> double delta_time;
>
> // Start it up..
> try {
> // make connection to broker
> connection.open();
> // create a session on the broker
> session = connection.createSession();
> // create a sender on this session
> sender = session.createSender(addr);
> // create a receiver on this session
> receiver = session.createReceiver(addr);
>
> // start the do-stuff loop
> gettimeofday(&tv_start, NULL);
>
> // print new line
> //std::cout << "\n";
>
> for (lc=0; lc < message_count; lc++) {
> // binary copy counter as first 4 bytes of buffer
> memcpy( (void *)txMessage_buffer, (void *)&lc, 4 );
> pubMessage.setContent( (const char *)txMessage_buffer,
> message_size );
> sender.send( pubMessage, true );
> // receive message from broker
> subMessage = receiver.fetch();
> // print status update every 1%
> if ( ( lc % ( message_count / 100 ) ) == 0 ) {
> // print
> std::cout << "\x1B[1K"; // voodoo to clear line on a
> vt100 terminal
> std::cout << lc << "/" << message_count << " messages
> transfered..\n\x1BM";
>
> }
> session.acknowledge( subMessage, false );
> }
> std::cout << "\n";
> gettimeofday(&tv_stop, NULL);
> std::cout << "finished sending " << message_count << "
> messages of size " << message_size << " bytes each.";
> timeval_subtract( &tv_delta, &tv_stop, &tv_start);
> delta_time = tv_delta.tv_sec + ( tv_delta.tv_usec / 1e6 );
> std::cout << "Opeation completed in " << delta_time << "
> seconds\n";
> } catch(const std::exception& error) {
> std::cout << error.what() << std::endl;
> connection.close();
> return -1;
> }
>
> connection.close();
> return 0;
> }
>
> // simple method to fill a buffer with some random stuff
> void makeRandomBinaryMessage( unsigned char* buffer, int size ) {
> /* random seed, just using time */
> srand ( (unsigned int)time(NULL) );
>
> /* now generate random value to fill buffer */
> /* note: this data is very random, but random enough for this. */
> int i=0;
> for (i; i<size; i++) {
> buffer[i] = (char)( rand() % 255 );
> }
> }
>
> // figure out the difference between to timeval structures
> int timeval_subtract (struct timeval *result, struct timeval *x,
> struct timeval *y )
> {
> /* Perform the carry for the later subtraction by updating y. */
> if (x->tv_usec < y->tv_usec) {
> int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
> y->tv_usec -= 1000000 * nsec;
> y->tv_sec += nsec;
> }
> if (x->tv_usec - y->tv_usec > 1000000) {
> int nsec = (x->tv_usec - y->tv_usec) / 1000000;
> y->tv_usec += 1000000 * nsec;
> y->tv_sec -= nsec;
> }
>
> /* Compute the time remaining to wait.
> tv_usec is certainly positive. */
> result->tv_sec = x->tv_sec - y->tv_sec;
> result->tv_usec = x->tv_usec - y->tv_usec;
>
> /* Return 1 if result is negative. */
> return x->tv_sec < y->tv_sec;
> }
>
> -------------- Code Segment End ---------------
>
> --
> Peter Fetterer (kb3gtn)
> * Failure is not an option, but a standard feature to be avoided.
>
> ---------------------------------------------------------------------
> Apache Qpid - AMQP Messaging Implementation
> Project: http://qpid.apache.org
> Use/Interact: mailto:users-subscribe@qpid.apache.org
>
>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:users-subscribe@qpid.apache.org