Hello Mr boundabout,
I had a go at reproducing this. I decided to use the headers exchange
'cause that's what I tend to use mostly myself.
I'm running with a single producer with a sender capacity of 100 sending
1500 byte messages with headers set thus:
message.setProperty("data-service", utf8("amqp-delivery"));
message.setProperty("data-type", utf8("audio"));
I've got consumers with headers bindings that match either:
data-service: amqp-delivery, data-type: audio
for consumers that I want to actually match the messages I'm sending and:
data-service: amqp-delivery, data-type: video
For consumers that I want to run up, but ignore the messages being sent
by the producer.
I've set link: {reliability: unreliable} in the Address and
{tcp-nodelay: true} for connectionOptions to try to eke out extra
performance.
I hope that's a reasonable facsimile of your test case? I've attached my
test code (caveat it's quite hacky - my C++ is pretty hideously rusty :-()
I've tried using separate Processes (ItemProducer.cpp/ItemConsumer.cpp)
and threads (Threads.cpp) The Threads programme takes command line
arguments Usage: ./Threads [matchCount (default = 1)] [missCount
(default = 0)] [shared]
So doing:
./Threads 3 2 shared
Will start up one producer and 5 consumers with 3 matching the sent
items and 2 not matching in this case the consumers will share the same
Connection (if shared is omitted each consumer will use its own Connection).
Unfortunately I've not been able to reproduce what you're seeing in the
consumers at all, however I don't have the luxury of an M4 server with
24 logical cores and all I've got to run things on is my Core 2 Duo
laptop with two P7450 cores running at 2.13GHz (the vastly different
number of cores *might well* be significant to the different observations).
Here's what I've been seeing:
Using separate processes:
1 Producer 1 Consumer:
5435 messages/second observed on the queue enqueue/dequeue rate, qpidd@87%
1 Producer 2 Consumers (both receiving):
3491 messages/second observed on each queue enqueue/dequeue rate, qpidd@92%
1 Producer 3 Consumers (all receiving):
2679 messages/second observed on each queue enqueue/dequeue rate, qpidd@95%
1 Producer 5 Consumers (3 receiving, 2 not matching):
2387 messages/second observed on each matching queue enqueue/dequeue
rate, qpidd@99%
0 messages/second observed on each non-matching queue enqueue/dequeue rate
Using Threads (with separate connection for each consumer):
1 Producer 1 Consumer:
5667 messages/second observed on the queue enqueue/dequeue rate,
qpidd@94%, Threads@97%
1 Producer 2 Consumers (both receiving):
3721 messages/second observed on each queue enqueue/dequeue rate,
qpidd@100%, Threads@90%
1 Producer 3 Consumers (all receiving):
2735 messages/second observed on each queue enqueue/dequeue rate,
qpidd@103%, Threads@84%
1 Producer 5 Consumers (3 receiving, 2 not matching):
2433 messages/second observed on each matching queue enqueue/dequeue
rate, qpidd@110%, Threads@78%
0 messages/second observed on each non-matching queue enqueue/dequeue rate
Using Threads (with shared connection):
1 Producer 1 Consumer:
5675 messages/second observed on the queue enqueue/dequeue rate,
qpidd@94%, Threads@98%
1 Producer 2 Consumers (both receiving):
3840 messages/second observed on each queue enqueue/dequeue rate,
qpidd@97%, Threads@91%
1 Producer 3 Consumers (all receiving):
2987 messages/second observed on each queue enqueue/dequeue rate,
qpidd@101%, Threads@88%
1 Producer 5 Consumers (3 receiving, 2 not matching):
2637 messages/second observed on each matching queue enqueue/dequeue
rate, qpidd@104%, Threads@76%
0 messages/second observed on each non-matching queue enqueue/dequeue rate
So I'm seeing that using Threads is giving consistently higher figures
than Processes and the shared Connection giving higher figures than
using a separate Connection for each consumer.
I also tried
1 Producer 13 Consumers (3 receiving, 10 not matching):
1768 messages/second observed on each matching queue enqueue/dequeue
rate, qpidd@108%, Threads@56%
0 messages/second observed on each non-matching queue enqueue/dequeue rate
This last figure seems slightly interesting, so as I increase the number
of *non-matching* consumers but keep the number of matching consumers
constant the overall CPU utilisation of the Threads program drops, so
even though they aren't receiving data it looks like there's some form
of thread "contention" (I get similar with shared and non-shared queues).
I don't know that C++ qpid::messaging code at all really, but my guess
is that there's likely an internal thread handling the prefetch into the
client runtime receiver queues (sized by setCapacity) and likely some
mutex or other synchronisation between that and the receiver.get() call
- the more things synchronising to that blocking queue the more wait
cycles. But that's all just a guess I'm afraid.
One other potentially interesting observation, so you say "It is
observed that each listener gets another thread created in the qpid
library "
I don't see that at all. I ran up "top -H" to see the threads being
created and I consistently see a number of Threads equal to
num-receiving-consumers + 1 (for the producer thread) + 2 (probably
internal qpid client threads)
e.g. running ./Threads 5 10 shared
I'm seeing 8 Threads in other words the threads for the non-matching
consumers are essentially idle and not registered by top.
As I say I've only got a two core box and I don't know the
qpid::messaging code at all really, but if you are indeed observing each
listener getting another thread created on a 24 core box *I wonder* if
there's a mechanism in the qpid client run-time that's allocating
internal worker threads up to some limit based on the number of cores?
The broker certainly does that....
Given that I'm observing the overall CPU utilisation dropping as I
increase the number of non-matching threads I wonder if you are seeing
lots of worker threads getting spawned which is creating even more
synchronisation contention than I'm seeing, but that's *really* just a
guess.
Although it's just a guess it's at least vaguely plausible given your
observation for Issue 1 where you say reducing the broker thread pool
size to something smaller that 24 actually increases throughput.
Unfortunately though if the client runtime is actually creating a worker
thread pool based on the number of cores I've no idea how to change this
size. I know with the broker it's a simple case of doing
--worker-threads N but I've no idea at all with the client :-(
I'm not sure that I've really helped any, perhaps you'd be able to try
my test programs to see if they are able to reproduce (or otherwise) the
sort of issues you are seeing, at the very least if my test program
yields the same results that you are seeing on your system it rules out
an application issue.
Perhaps one of the qpid devs with access to a fancier box than me might
be able to try some of this stuff against a few more cores.
My tests were run against a build off trunk (qpid 0.25) I did on 20th
July I don't know how materially different that is to 0.18 in terms of
thread behaviour.
Best regards,
Frase
On 19/07/13 08:29, boundabout wrote:
We have observed two issues, possibly with a common cause, with threading in
qpid 0.18.
Issue 1 - Using the C++ broker on an IBM M4, performance is very poor (and
CPU utilisation very high) unless the size of the thread pool is reduced by
setting worker-threads to a small number. I believe the default is to create
one more thread than the number of logical cores, which for this box is 24.
So the question here is why having a lot of threads could increase CPU
loading. I wondered if it might be that threads were being moved between
cores, causing a lot of cache invalidation, but setting hard affinity
indicates that this is not the case.
Issue 2 - A client process in C++ on Linux RHEL 5.8 creates a number of
listener threads, using the messaging interface, all listening to the same
on-box broker. Each listener has its own connection, session and receiver.
It is observed that each listener gets another thread created in the qpid
library. The listeners are configured, and messages sent to the broker, such
that some will receive data, and some will not. CPU loading for the active
and dormant qpid threads increases with the number of threads. This is not
the case when listeners in separate processes are used.
It seems, in both cases, as if threads somehow have a dependency on each
other, such that for each working thread it has to do some work relating to
each other thread, and these other threads somehow get some work to do due
to the first. This seems pretty weird to me, I'd be grateful if anyone could
shed any light and, even better, suggest a way of avoiding this effect.
--
View this message in context:
http://qpid.2158936.n2.nabble.com/Potentially-related-issues-between-QPID-threads-tp7595682.html
Sent from the Apache Qpid users mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
/*
*
* 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 <qpid/messaging/Address.h>
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Session.h>
#include <qpid/types/Variant.h>
#include <sys/time.h> // For gettimeofday()
#include <iostream>
using namespace std;
using namespace qpid::messaging;
using namespace qpid::types;
int kbhit(void)
{
struct timeval tv; fd_set
read_fd; /* Do not wait at all, not even a microsecond */
tv.tv_sec=0;
tv.tv_usec=0; /* Must be done first to initialize read_fd */
FD_ZERO(&read_fd); /* Makes select() ask if input is ready: 0 is the file descriptor for stdin */
FD_SET(0,&read_fd); /* The first parameter is the number of the largest file descriptor to check + 1. */
if (select(1, &read_fd, NULL, /*No writes*/ NULL, /*No exceptions*/&tv) == -1) return 0; /* An error occured */
/* read_fd now holds a bit map of files that are readable. We test the entry for the standard input (file 0). */
if (FD_ISSET(0,&read_fd)) return 1; /* no characters were pending */
return 0;
}
int main(int argc, char** argv) {
string broker = "localhost:5672";
//string connectionOptions = "{reconnect: true}";
string connectionOptions = "{tcp-nodelay: true}";
string instance = "1";
string dataType = "audio";
if (argc >= 2) {
instance = argv[1];
}
if (argc >= 3) {
dataType = argv[2];
}
cout << "Queue = item" << instance << ", type = " << dataType << endl;
/*string address = "item" + instance + "; {create: receiver, node: {x-declare: {auto-delete: true, arguments: {'qpid.policy_type': ring, 'qpid.max_size': 100000000}}, x-bindings: [{exchange: 'amq.match', queue: 'item" + instance + "', key: 'key" + instance + "', arguments: {x-match: all, data-service: amqp-delivery, data-type: '" + dataType + "'}}]}}";*/
string address = "item" + instance + "; {create: receiver, node: {x-declare: {auto-delete: true, arguments: {'qpid.policy_type': ring, 'qpid.max_size': 100000000}}, x-bindings: [{exchange: 'amq.match', queue: 'item" + instance + "', key: 'key" + instance + "', arguments: {x-match: all, data-service: amqp-delivery, data-type: '" + dataType + "'}}]}, link: {reliability: unreliable}}";
Connection connection(broker, connectionOptions);
try {
connection.open();
Session session = connection.createSession();
Receiver receiver = session.createReceiver(address);
receiver.setCapacity(100); // Enable receiver prefetch
unsigned int count = 0;
unsigned int unackedCount = 0;
Message message;
while (true) {
if (receiver.get(message, Duration::SECOND * 1)) {
count++;
unackedCount++;
//const char* buffer = message.getContentPtr();
//cout << "message count = " << count << ", length = " << message.getContentSize() << endl;
/*if (unackedCount > receiver.getCapacity()/2) {
//cout << "auto acknowledged message #" << count << endl;
session.acknowledge();
unackedCount = 0;
}*/
}
if (kbhit()) break;
}
session.acknowledge();
connection.close();
return 0;
} catch(const exception& error) {
cerr << "ItemConsumer Exception: " << error.what() << endl;
connection.close();
return 1;
}
}
/*
*
* 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 <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Session.h>
#include <qpid/types/Variant.h>
#include <sys/time.h> // For gettimeofday()
#include <iostream>
using namespace std;
using namespace qpid::messaging;
using namespace qpid::types;
unsigned long currentTimeMillis() {
struct timeval curTime;
gettimeofday(&curTime, NULL);
return (curTime.tv_usec + curTime.tv_sec * 1000000ul)/1000;
}
inline Variant utf8(const char* s) {
Variant utf8Value(s);
utf8Value.setEncoding("utf8");
return utf8Value;
}
int main(int argc, char** argv) {
string broker = "localhost:5672";
//string connectionOptions = "{reconnect: true}";
string connectionOptions = "{tcp-nodelay: true}";
string address = "amq.match";
Connection connection(broker, connectionOptions);
try {
connection.open();
Session session = connection.createSession();
Sender sender = session.createSender(address);
sender.setCapacity(100);
int NUMBER_OF_ITERATIONS = 1000000;
int ITEM_SIZE = 1500;
unsigned long startTime = currentTimeMillis();
for (int i = 0; i < NUMBER_OF_ITERATIONS; i++) {
char* buffer = new char[ITEM_SIZE];
Message message(buffer, ITEM_SIZE);
message.setProperty("data-service", utf8("amqp-delivery"));
message.setProperty("data-type", utf8("audio"));
sender.send(message);
delete buffer;
// cout << "Sent message #" << i << endl;
}
session.sync();
unsigned long finishTime = currentTimeMillis();
cout << "Elapsed time = " << (finishTime - startTime) << ", messages/second = " << NUMBER_OF_ITERATIONS*1000.0f/(finishTime - startTime) << endl;
connection.close();
return 0;
} catch(const exception& error) {
cerr << "ItemProducer Exception: " << error.what() << endl;
connection.close();
return 1;
}
}
SLASH = /
REMOVE = -rm
LIBPTHREAD = -lpthread
#----------------------- Compiler Options ---------------------
COMPILER = g++
DEBUG =
#DEBUG = -DDEBUG
#DEBUG = -g
PROFILE =
#PROFILE = -pg
CFLAGS = -O6 -march=native\
-Wall -posix $(DEBUG) $(PROFILE)
CDEFINES = -D_POSIX_C_SOURCE=199506L\
-D_XOPEN_SOURCE\
-D_XOPEN_SOURCE_EXTENDED\
-D_REENTRANT\
-D_THREAD_SAFE
# -------------------------------------------------------------
CINCLUDES = \
-I.\
SRCFILES = \
\
CPPOBJFILES = $(SRCFILES:.cpp=.o)
OBJFILES = $(CPPOBJFILES:.c=.o)
all : ItemProducer ItemConsumer Threads
ItemProducer : ItemProducer.o
$(COMPILER) $(CFLAGS) $(CDEFINES) \
$(OBJFILES) ItemProducer.o $(LIBPTHREAD) -lqpidmessaging -lqpidtypes -o
$@
ItemProducer.o : ItemProducer.cpp
$(COMPILER) $(CFLAGS) $(CDEFINES) $(CINCLUDES) -c $?
ItemConsumer : ItemConsumer.o
$(COMPILER) $(CFLAGS) $(CDEFINES) \
$(OBJFILES) ItemConsumer.o $(LIBPTHREAD) -lqpidmessaging -lqpidtypes -o
$@
ItemConsumer.o : ItemConsumer.cpp
$(COMPILER) $(CFLAGS) $(CDEFINES) $(CINCLUDES) -c $?
Threads : Threads.o
$(COMPILER) $(CFLAGS) $(CDEFINES) \
$(OBJFILES) Threads.o $(LIBPTHREAD) -lqpidmessaging -lqpidcommon
-lqpidtypes -o $@
Threads.o : Threads.cpp
$(COMPILER) $(CFLAGS) $(CDEFINES) $(CINCLUDES) -c $?
# Makes all of the .cpp & .c specified in the SRCFILES macro
.objects : $(SRCFILES)
$(COMPILER) $(CFLAGS) $(CDEFINES) $(CINCLUDES) -c $?
echo touched > .objects
clean :
$(REMOVE) *~
$(REMOVE) *.~*
$(REMOVE) *.o
$(REMOVE) .objects
$(REMOVE) ItemProducer
$(REMOVE) ItemConsumer
$(REMOVE) Threads
/*
*
* 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 <qpid/messaging/Address.h>
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Session.h>
#include <qpid/types/Variant.h>
#include <sys/time.h> // For gettimeofday()
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Thread.h"
#include <iostream>
#include <sstream>
using namespace std;
using namespace qpid::messaging;
using namespace qpid::sys;
using namespace qpid::types;
unsigned long currentTimeMillis() {
struct timeval curTime;
gettimeofday(&curTime, NULL);
return (curTime.tv_usec + curTime.tv_sec * 1000000ul)/1000;
}
inline Variant utf8(const char* s) {
Variant utf8Value(s);
utf8Value.setEncoding("utf8");
return utf8Value;
}
string broker = "localhost:5672";
string connectionOptions = "{tcp-nodelay: true}";
class PublishThread : public Runnable {
private:
string address;
Thread thread;
public:
PublishThread(string address) : address(address) {};
void start() {
thread = Thread(this);
thread.join(); // Block until producer run ends.
};
private:
void run() {
Connection connection(broker, connectionOptions);
try {
connection.open();
Session session = connection.createSession();
Sender sender = session.createSender(address);
sender.setCapacity(100);
int NUMBER_OF_ITERATIONS = 1000000;
int ITEM_SIZE = 1500;
unsigned long startTime = currentTimeMillis();
for (int i = 0; i < NUMBER_OF_ITERATIONS; i++) {
char* buffer = new char[ITEM_SIZE];
Message message(buffer, ITEM_SIZE);
message.setProperty("data-service", utf8("amqp-delivery"));
message.setProperty("data-type", utf8("audio"));
sender.send(message);
delete buffer;
// cout << "Sent message #" << i << endl;
}
session.sync();
unsigned long finishTime = currentTimeMillis();
cout << "Elapsed time = " << (finishTime - startTime)
<< ", producer messages/second = " << NUMBER_OF_ITERATIONS*1000.0f/(finishTime - startTime) << endl;
connection.close();
} catch(const exception& error) {
cerr << "PublishThread Exception: " << error.what() << endl;
connection.close();
}
};
};
class SubscribeThread : public Runnable {
private:
string address;
Connection* connection;
bool ownConnection; // Set to true if each subscriber thread should use its own connection.
bool running;
Thread thread;
public:
SubscribeThread(string instance, string dataType, Connection* conn = 0) :
address("item" + instance + "; {create: receiver, node: {x-declare: {auto-delete: true, arguments: {'qpid.policy_type': ring, 'qpid.max_size': 100000000}}, x-bindings: [{exchange: 'amq.match', queue: 'item" + instance + "', key: 'key" + instance + "', arguments: {x-match: all, data-service: amqp-delivery, data-type: '" + dataType + "'}}]}, link: {reliability: unreliable}}"),
connection(conn),
ownConnection(connection == 0 ? true : false) {
};
void start() {
running = true;
if (ownConnection) {
connection = new Connection(broker, connectionOptions);
connection->open();
}
thread = Thread(this);
};
void stop() {
running = false;
thread.join();
if (ownConnection) {
connection->close();
delete connection;
}
};
private:
void run() {
try {
Session session = connection->createSession();
Receiver receiver = session.createReceiver(address);
receiver.setCapacity(100); // Enable receiver prefetch
unsigned int count = 0;
unsigned int unackedCount = 0;
Message message;
unsigned long startTime = currentTimeMillis();
while (running) {
if (receiver.get(message, qpid::messaging::Duration::SECOND)) {
count++;
unackedCount++;
//const char* buffer = message.getContentPtr();
//cout << "message count = " << count << ", length = " << message.getContentSize() << endl;
// We need this block if we don't enable link: {reliability: unreliable}
/*if (unackedCount > receiver.getCapacity()/2) {
//cout << "auto acknowledged message #" << count << endl;
session.acknowledge();
unackedCount = 0;
}*/
}
}
session.acknowledge();
unsigned long finishTime = currentTimeMillis();
cout << "SubscribeThread stopped" << endl;
cout << "Elapsed time = " << (finishTime - startTime)
<< ", consumer messages/second = " << count*1000.0f/(finishTime - startTime) << endl;
}
catch (const std::exception& e) {
cout << "SubscribeThread exception: " << e.what() << endl;
stop();
}
}
};
int main(int argc, char** argv) {
cout << "Usage: " << argv[0] << " [matchCount (default = 1)] [missCount (default = 0)] [shared]" << endl;
cout << "e.g. " << argv[0] << " 3 2 shared" << endl;
cout << " " << "3 matching subscribers 2 non-matching using a shared Connection" << endl;
cout << "e.g. " << argv[0] << " 5 3" << endl;
cout << " " << "5 matching subscribers 3 non-matching each using their own Connection" << endl;
cout << endl;
int exitCode = 0;
try {
int matchCount = 1;
int missCount = 0;
Connection* connection = 0; // Default behaviour is each consumer having its own connection.
if (argc >= 2) {
matchCount = atoi(argv[1]);
}
if (argc >= 3) {
missCount = atoi(argv[2]);
}
if (argc >= 4) { // A Shared consumer Connection has been specified.
connection = new Connection(broker, connectionOptions);
connection->open();
}
int subscriptionCount = matchCount + missCount;
cout << "Starting " << subscriptionCount << " consumer(s), "
<< matchCount << " matching, "
<< missCount << " not matching" << endl;
if (connection) {
cout << "Consumers are using a shared Connection" << endl;
} else {
cout << "Consumers are using their own Connections" << endl;
}
SubscribeThread* subscriptions[subscriptionCount];
for (int i = 0; i < subscriptionCount; i++) {
string queue = static_cast<ostringstream*>(&(ostringstream() << i))->str();
if (i < matchCount) {
subscriptions[i] = new SubscribeThread(queue, "audio", connection);
} else {
subscriptions[i] = new SubscribeThread(queue, "video", connection);
}
subscriptions[i]->start();
}
PublishThread pub("amq.match");
pub.start();
for (int i = 0; i < subscriptionCount; i++) { // Tidy up consumer threads.
subscriptions[i]->stop();
delete subscriptions[i];
}
if (connection) { // If a shared Connection has been specified then close and delete it before exiting.
connection->close();
delete connection;
}
}
catch (const std::exception& e) {
cout << endl << e.what() << endl;
exitCode = 1;
}
return exitCode;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
