summaryrefslogtreecommitdiff
path: root/cpp/src/tests/perftest.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/tests/perftest.cpp')
-rw-r--r--cpp/src/tests/perftest.cpp286
1 files changed, 182 insertions, 104 deletions
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index 9096854a6d..88d9fd15cb 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -7,9 +7,9 @@
* 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
@@ -28,6 +28,7 @@
#include "qpid/client/Message.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/sys/Time.h"
+#include "qpid/sys/Thread.h"
#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>
@@ -38,7 +39,6 @@
#include <sstream>
#include <numeric>
#include <algorithm>
-#include <unistd.h>
#include <math.h>
@@ -49,6 +49,9 @@ using namespace sys;
using boost::lexical_cast;
using boost::bind;
+namespace qpid {
+namespace tests {
+
enum Mode { SHARED, FANOUT, TOPIC };
const char* modeNames[] = { "shared", "fanout", "topic" };
@@ -75,6 +78,7 @@ struct Opts : public TestOptions {
// Queue policy
uint32_t queueMaxCount;
uint64_t queueMaxSize;
+ std::string baseName;
bool queueDurable;
// Publisher
@@ -92,22 +96,26 @@ struct Opts : public TestOptions {
// General
size_t qt;
+ bool singleConnect;
size_t iterations;
Mode mode;
bool summary;
uint32_t intervalSub;
uint32_t intervalPub;
size_t tx;
+ size_t txPub;
+ size_t txSub;
+ bool commitAsync;
static const std::string helpText;
-
+
Opts() :
TestOptions(helpText),
- setup(false), control(false), publish(false), subscribe(false),
+ setup(false), control(false), publish(false), subscribe(false), baseName("perftest"),
pubs(1), count(500000), size(1024), confirm(true), durable(false), uniqueData(false), syncPub(false),
subs(1), ack(0),
- qt(1), iterations(1), mode(SHARED), summary(false),
- intervalSub(0), intervalPub(0), tx(0)
+ qt(1),singleConnect(false), iterations(1), mode(SHARED), summary(false),
+ intervalSub(0), intervalPub(0), tx(0), txPub(0), txSub(0), commitAsync(false)
{
addOptions()
("setup", optValue(setup), "Create shared queues.")
@@ -131,19 +139,25 @@ struct Opts : public TestOptions {
("nsubs", optValue(subs, "N"), "Create N subscribers.")
("sub-ack", optValue(ack, "N"), "N>0: Subscriber acks batches of N.\n"
"N==0: Subscriber uses unconfirmed mode")
-
+
("qt", optValue(qt, "N"), "Create N queues or topics.")
+ ("single-connection", optValue(singleConnect, "yes|no"), "Use one connection for multiple sessions.")
+
("iterations", optValue(iterations, "N"), "Desired number of iterations of the test.")
("summary,s", optValue(summary), "Summary output: pubs/sec subs/sec transfers/sec Mbytes/sec")
("queue-max-count", optValue(queueMaxCount, "N"), "queue policy: count to trigger 'flow to disk'")
("queue-max-size", optValue(queueMaxSize, "N"), "queue policy: accumulated size to trigger 'flow to disk'")
+ ("base-name", optValue(baseName, "NAME"), "base name used for queues or topics")
("queue-durable", optValue(queueDurable, "N"), "Make queue durable (implied if durable set)")
("interval_sub", optValue(intervalSub, "ms"), ">=0 delay between msg consume")
("interval_pub", optValue(intervalPub, "ms"), ">=0 delay between msg publish")
- ("tx", optValue(tx, "N"), "if non-zero, the transaction batch size");
+ ("tx", optValue(tx, "N"), "if non-zero, the transaction batch size for publishing and consuming")
+ ("pub-tx", optValue(txPub, "N"), "if non-zero, the transaction batch size for publishing")
+ ("async-commit", optValue(commitAsync, "yes|no"), "Don't wait for completion of commit")
+ ("sub-tx", optValue(txSub, "N"), "if non-zero, the transaction batch size for consuming");
}
// Computed values
@@ -160,7 +174,7 @@ struct Opts : public TestOptions {
count += subs - (count % subs);
cout << "WARNING: Adjusted --count to " << count
<< " the nearest multiple of --nsubs" << endl;
- }
+ }
totalPubs = pubs*qt;
totalSubs = subs*qt;
subQuota = (pubs*count)/subs;
@@ -180,6 +194,18 @@ struct Opts : public TestOptions {
break;
}
transfers=(totalPubs*count) + (totalSubs*subQuota);
+ if (tx) {
+ if (txPub) {
+ cerr << "WARNING: Using overriden tx value for publishers: " << txPub << std::endl;
+ } else {
+ txPub = tx;
+ }
+ if (txSub) {
+ cerr << "WARNING: Using overriden tx value for subscribers: " << txSub << std::endl;
+ } else {
+ txSub = tx;
+ }
+ }
}
};
@@ -196,25 +222,46 @@ const std::string Opts::helpText=
"Note the <other options> must be identical for all processes.\n";
Opts opts;
+Connection globalConnection;
+
+std::string fqn(const std::string& name)
+{
+ ostringstream fqn;
+ fqn << opts.baseName << "_" << name;
+ return fqn.str();
+}
struct Client : public Runnable {
- Connection connection;
+ Connection* connection;
+ Connection localConnection;
AsyncSession session;
Thread thread;
Client() {
- opts.open(connection);
- session = connection.newSession();
+ if (opts.singleConnect){
+ connection = &globalConnection;
+ if (!globalConnection.isOpen()) opts.open(globalConnection);
+ }else{
+ connection = &localConnection;
+ opts.open(localConnection);
+ }
+ session = connection->newSession();
}
~Client() {
- session.close();
- connection.close();
+ try {
+ if (connection->isOpen()) {
+ session.close();
+ connection->close();
+ }
+ } catch (const std::exception& e) {
+ std::cerr << "Error in shutdown: " << e.what() << std::endl;
+ }
}
};
struct Setup : public Client {
-
+
void queueInit(string name, bool durable=false, const framing::FieldTable& settings=framing::FieldTable()) {
session.queueDeclare(arg::queue=name, arg::durable=durable, arg::arguments=settings);
session.queuePurge(arg::queue=name);
@@ -222,18 +269,19 @@ struct Setup : public Client {
}
void run() {
- queueInit("pub_start");
- queueInit("pub_done");
- queueInit("sub_ready");
- queueInit("sub_done");
+ queueInit(fqn("pub_start"));
+ queueInit(fqn("pub_done"));
+ queueInit(fqn("sub_ready"));
+ queueInit(fqn("sub_done"));
+ if (opts.iterations > 1) queueInit(fqn("sub_iteration"));
if (opts.mode==SHARED) {
framing::FieldTable settings;//queue policy settings
settings.setInt("qpid.max_count", opts.queueMaxCount);
settings.setInt("qpid.max_size", opts.queueMaxSize);
for (size_t i = 0; i < opts.qt; ++i) {
ostringstream qname;
- qname << "perftest" << i;
- queueInit(qname.str(), opts.durable || opts.queueDurable, settings);
+ qname << opts.baseName << i;
+ queueInit(qname.str(), opts.durable || opts.queueDurable, settings);
}
}
}
@@ -258,7 +306,7 @@ class Stats {
public:
Stats() : sum(0) {}
-
+
// Functor to collect rates.
void operator()(const string& data) {
try {
@@ -269,7 +317,7 @@ class Stats {
throw Exception("Bad report: "+data);
}
}
-
+
double mean() const {
return sum/values.size();
}
@@ -286,7 +334,7 @@ class Stats {
}
return sqrt(ssq/(values.size()-1));
}
-
+
ostream& print(ostream& out) {
ostream_iterator<double> o(out, "\n");
copy(values.begin(), values.end(), o);
@@ -296,11 +344,11 @@ class Stats {
return out << endl;
}
};
-
+
// Manage control queues, collect and print reports.
struct Controller : public Client {
-
+
SubscriptionManager subs;
Controller() : subs(session) {}
@@ -309,7 +357,7 @@ struct Controller : public Client {
void process(size_t n, string queue,
boost::function<void (const string&)> msgFn)
{
- if (!opts.summary)
+ if (!opts.summary)
cout << "Processing " << n << " messages from "
<< queue << " " << flush;
LocalQueue lq;
@@ -325,8 +373,8 @@ struct Controller : public Client {
void process(size_t n, LocalQueue lq, string queue,
boost::function<void (const string&)> msgFn)
{
- session.messageFlow(queue, 0, n);
- if (!opts.summary)
+ session.messageFlow(queue, 0, n);
+ if (!opts.summary)
cout << "Processing " << n << " messages from "
<< queue << " " << flush;
for (size_t i = 0; i < n; ++i) {
@@ -341,20 +389,20 @@ struct Controller : public Client {
cout << "Sending " << data << " " << n << " times to " << queue
<< endl;
Message msg(data, queue);
- for (size_t i = 0; i < n; ++i)
+ for (size_t i = 0; i < n; ++i)
session.messageTransfer(arg::content=msg, arg::acceptMode=1);
}
void run() { // Controller
try {
// Wait for subscribers to be ready.
- process(opts.totalSubs, "sub_ready", bind(expect, _1, "ready"));
+ process(opts.totalSubs, fqn("sub_ready"), bind(expect, _1, "ready"));
LocalQueue pubDone;
LocalQueue subDone;
subs.setFlowControl(0, SubscriptionManager::UNLIMITED, false);
- subs.subscribe(pubDone, "pub_done");
- subs.subscribe(subDone, "sub_done");
+ subs.subscribe(pubDone, fqn("pub_done"));
+ subs.subscribe(subDone, fqn("sub_done"));
double txrateTotal(0);
double mbytesTotal(0);
@@ -363,15 +411,18 @@ struct Controller : public Client {
for (size_t j = 0; j < opts.iterations; ++j) {
AbsTime start=now();
- send(opts.totalPubs, "pub_start", "start"); // Start publishers
+ send(opts.totalPubs, fqn("pub_start"), "start"); // Start publishers
+ if (j) {
+ send(opts.totalPubs, fqn("sub_iteration"), "next"); // Start subscribers on next iteration
+ }
Stats pubRates;
Stats subRates;
- process(opts.totalPubs, pubDone, "pub_done", boost::ref(pubRates));
- process(opts.totalSubs, subDone, "sub_done", boost::ref(subRates));
+ process(opts.totalPubs, pubDone, fqn("pub_done"), boost::ref(pubRates));
+ process(opts.totalSubs, subDone, fqn("sub_done"), boost::ref(subRates));
- AbsTime end=now();
+ AbsTime end=now();
double time=secs(start, end);
double txrate=opts.transfers/time;
@@ -411,7 +462,6 @@ struct Controller : public Client {
}
catch (const std::exception& e) {
cout << "Controller exception: " << e.what() << endl;
- exit(1);
}
}
};
@@ -422,12 +472,12 @@ struct PublishThread : public Client {
string routingKey;
PublishThread() {};
-
+
PublishThread(string key, string dest=string()) {
destination=dest;
routingKey=key;
}
-
+
void run() { // Publisher
try {
string data;
@@ -445,7 +495,7 @@ struct PublishThread : public Client {
}
} else {
size_t msgSize=max(opts.size, sizeof(size_t));
- data = string(msgSize, 'X');
+ data = string(msgSize, 'X');
}
Message msg(data, routingKey);
@@ -453,19 +503,21 @@ struct PublishThread : public Client {
msg.getDeliveryProperties().setDeliveryMode(framing::PERSISTENT);
- if (opts.tx) sync(session).txSelect();
+ if (opts.txPub){
+ session.txSelect();
+ }
SubscriptionManager subs(session);
LocalQueue lq;
- subs.setFlowControl(1, SubscriptionManager::UNLIMITED, true);
- subs.subscribe(lq, "pub_start");
-
+ subs.setFlowControl(1, SubscriptionManager::UNLIMITED, true);
+ subs.subscribe(lq, fqn("pub_start"));
+
for (size_t j = 0; j < opts.iterations; ++j) {
expect(lq.pop().getData(), "start");
AbsTime start=now();
for (size_t i=0; i<opts.count; i++) {
// Stamp the iteration into the message data, avoid
// any heap allocation.
- const_cast<std::string&>(msg.getData()).replace(offset, sizeof(size_t),
+ const_cast<std::string&>(msg.getData()).replace(offset, sizeof(size_t),
reinterpret_cast<const char*>(&i), sizeof(size_t));
if (opts.syncPub) {
sync(session).messageTransfer(
@@ -478,23 +530,31 @@ struct PublishThread : public Client {
arg::content=msg,
arg::acceptMode=1);
}
- if (opts.tx && ((i+1) % opts.tx == 0)) sync(session).txCommit();
- if (opts.intervalPub) ::usleep(opts.intervalPub*1000);
+ if (opts.txPub && ((i+1) % opts.txPub == 0)){
+ if (opts.commitAsync){
+ session.txCommit();
+ } else {
+ sync(session).txCommit();
+ }
+ }
+ if (opts.intervalPub)
+ qpid::sys::usleep(opts.intervalPub*1000);
}
if (opts.confirm) session.sync();
AbsTime end=now();
double time=secs(start,end);
-
+
// Send result to controller.
- Message report(lexical_cast<string>(opts.count/time), "pub_done");
+ Message report(lexical_cast<string>(opts.count/time), fqn("pub_done"));
session.messageTransfer(arg::content=report, arg::acceptMode=1);
- if (opts.tx) sync(session).txCommit();
+ if (opts.txPub){
+ sync(session).txCommit();
+ }
}
session.close();
}
catch (const std::exception& e) {
cout << "PublishThread exception: " << e.what() << endl;
- exit(1);
}
}
};
@@ -504,7 +564,7 @@ struct SubscribeThread : public Client {
string queue;
SubscribeThread() {}
-
+
SubscribeThread(string q) { queue = q; }
SubscribeThread(string key, string ex) {
@@ -529,31 +589,48 @@ struct SubscribeThread : public Client {
}
void run() { // Subscribe
- try {
- if (opts.tx) sync(session).txSelect();
+ try {
+ if (opts.txSub) sync(session).txSelect();
SubscriptionManager subs(session);
- LocalQueue lq(AckPolicy(opts.tx ? opts.tx : opts.ack));
- subs.setAcceptMode(opts.tx || opts.ack ? 0 : 1);
- subs.setFlowControl(opts.subQuota, SubscriptionManager::UNLIMITED,
- false);
- subs.subscribe(lq, queue);
+ SubscriptionSettings settings;
+ settings.autoAck = opts.txSub ? opts.txSub : opts.ack;
+ settings.acceptMode = (opts.txSub || opts.ack ? ACCEPT_MODE_EXPLICIT : ACCEPT_MODE_NONE);
+ settings.flowControl = FlowControl::messageCredit(opts.subQuota);
+ LocalQueue lq;
+ Subscription subscription = subs.subscribe(lq, queue, settings);
// Notify controller we are ready.
- session.messageTransfer(arg::content=Message("ready", "sub_ready"), arg::acceptMode=1);
- if (opts.tx) sync(session).txCommit();
-
+ session.messageTransfer(arg::content=Message("ready", fqn("sub_ready")), arg::acceptMode=1);
+ if (opts.txSub) {
+ if (opts.commitAsync) session.txCommit();
+ else sync(session).txCommit();
+ }
+
+ LocalQueue iterationControl;
+ if (opts.iterations > 1) {
+ subs.subscribe(iterationControl, fqn("sub_iteration"), SubscriptionSettings(FlowControl::messageCredit(0)));
+ }
+
for (size_t j = 0; j < opts.iterations; ++j) {
if (j > 0) {
- //need to allocate some more credit
- session.messageFlow(queue, 0, opts.subQuota);
+ //need to wait here until all subs are done
+ session.messageFlow(fqn("sub_iteration"), 0, 1);
+ iterationControl.pop();
+
+ //need to allocate some more credit for subscription
+ session.messageFlow(queue, 0, opts.subQuota);
}
Message msg;
AbsTime start=now();
size_t expect=0;
for (size_t i = 0; i < opts.subQuota; ++i) {
msg=lq.pop();
- if (opts.tx && ((i+1) % opts.tx == 0)) sync(session).txCommit();
- if (opts.intervalSub) ::usleep(opts.intervalSub*1000);
- // TODO aconway 2007-11-23: check message order for.
+ if (opts.txSub && ((i+1) % opts.txSub == 0)) {
+ if (opts.commitAsync) session.txCommit();
+ else sync(session).txCommit();
+ }
+ if (opts.intervalSub)
+ qpid::sys::usleep(opts.intervalSub*1000);
+ // TODO aconway 2007-11-23: check message order for.
// multiple publishers. Need an array of counters,
// one per publisher and a publisher ID in the
// message. Careful not to introduce a lot of overhead
@@ -568,29 +645,37 @@ struct SubscribeThread : public Client {
expect = n+1;
}
}
- if (opts.tx || opts.ack)
- lq.getAckPolicy().ackOutstanding(session); // Cumulative ack for final batch.
- if (opts.tx)
- sync(session).txCommit();
+ if (opts.txSub || opts.ack)
+ subscription.accept(subscription.getUnaccepted());
+ if (opts.txSub) {
+ if (opts.commitAsync) session.txCommit();
+ else sync(session).txCommit();
+ }
AbsTime end=now();
// Report to publisher.
Message result(lexical_cast<string>(opts.subQuota/secs(start,end)),
- "sub_done");
+ fqn("sub_done"));
session.messageTransfer(arg::content=result, arg::acceptMode=1);
- if (opts.tx) sync(session).txCommit();
+ if (opts.txSub) sync(session).txCommit();
}
session.close();
}
catch (const std::exception& e) {
cout << "SubscribeThread exception: " << e.what() << endl;
- exit(1);
}
}
};
+}} // namespace qpid::tests
+
+using namespace qpid::tests;
+
int main(int argc, char** argv) {
-
+ int exitCode = 0;
+ boost::ptr_vector<Client> subs(opts.subs);
+ boost::ptr_vector<Client> pubs(opts.pubs);
+
try {
opts.parse(argc, argv);
@@ -600,7 +685,7 @@ int main(int argc, char** argv) {
case TOPIC: exchange="amq.topic"; break;
case SHARED: break;
}
-
+
bool singleProcess=
(!opts.setup && !opts.control && !opts.publish && !opts.subscribe);
if (singleProcess)
@@ -608,13 +693,10 @@ int main(int argc, char** argv) {
if (opts.setup) Setup().run(); // Set up queues
- boost::ptr_vector<Client> subs(opts.subs);
- boost::ptr_vector<Client> pubs(opts.pubs);
-
// Start pubs/subs for each queue/topic.
for (size_t i = 0; i < opts.qt; ++i) {
ostringstream key;
- key << "perftest" << i; // Queue or topic name.
+ key << opts.baseName << i; // Queue or topic name.
if (opts.publish) {
size_t n = singleProcess ? opts.pubs : 1;
for (size_t j = 0; j < n; ++j) {
@@ -635,29 +717,25 @@ int main(int argc, char** argv) {
}
if (opts.control) Controller().run();
-
-
- // Wait for started threads.
- if (opts.publish) {
- for (boost::ptr_vector<Client>::iterator i=pubs.begin();
- i != pubs.end();
- ++i)
- i->thread.join();
- }
-
-
- if (opts.subscribe) {
- for (boost::ptr_vector<Client>::iterator i=subs.begin();
- i != subs.end();
- ++i)
- i->thread.join();
- }
- return 0;
}
catch (const std::exception& e) {
- cout << endl << e.what() << endl;
+ cout << endl << e.what() << endl;
+ exitCode = 1;
}
- return 1;
-}
-
+ // Wait for started threads.
+ if (opts.publish) {
+ for (boost::ptr_vector<Client>::iterator i=pubs.begin();
+ i != pubs.end();
+ ++i)
+ i->thread.join();
+ }
+
+ if (opts.subscribe) {
+ for (boost::ptr_vector<Client>::iterator i=subs.begin();
+ i != subs.end();
+ ++i)
+ i->thread.join();
+ }
+ return exitCode;
+}
View Lorry Controller Status