Bug 1804134
| Summary: | High memory usage after rerouted messages end in the same queue | ||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Product: | Red Hat Enterprise MRG | Reporter: | Pavel Moravec <pmoravec> | ||||||||||||||
| Component: | qpid-cpp | Assignee: | Kim van der Riet <kim.vdriet> | ||||||||||||||
| Status: | CLOSED UPSTREAM | QA Contact: | Messaging QE <messaging-qe-bugs> | ||||||||||||||
| Severity: | high | Docs Contact: | |||||||||||||||
| Priority: | high | ||||||||||||||||
| Version: | 3.2 | CC: | jross | ||||||||||||||
| Target Milestone: | --- | ||||||||||||||||
| Target Release: | --- | ||||||||||||||||
| Hardware: | Unspecified | ||||||||||||||||
| OS: | Unspecified | ||||||||||||||||
| Whiteboard: | |||||||||||||||||
| Fixed In Version: | Doc Type: | If docs needed, set a value | |||||||||||||||
| Doc Text: | Story Points: | --- | |||||||||||||||
| Clone Of: | Environment: | ||||||||||||||||
| Last Closed: | 2025-02-10 04:00:05 UTC | Type: | Bug | ||||||||||||||
| Regression: | --- | Mount Type: | --- | ||||||||||||||
| Documentation: | --- | CRM: | |||||||||||||||
| Verified Versions: | Category: | --- | |||||||||||||||
| oVirt Team: | --- | RHEL 7.3 requirements from Atomic Host: | |||||||||||||||
| Cloudforms Team: | --- | Target Upstream Version: | |||||||||||||||
| Embargoed: | |||||||||||||||||
| Attachments: |
|
||||||||||||||||
The very first iteration of the scenario adds message annotations "x-qpid.trace:" (empty string). That would explain *some* memory increase, but: - only in the very fisrt iteration - configuring the queue with --argument=qpid.trace.id=AltQueue sets the -qpid.trace annotation from the begining, while memory growth is still the same So, message annotations does not seem to be behind the memory consumption increase. s/redirect/reroute/g The bugzilla topic is message rerouting. NOT about redirecting. See "myQueue.reroute" in the reproducer. The fact we reroute messages to the same queue is NOT the key factor here: the memory increase occurs also when rerouting the messages to another exchange that has a binding to some queue, like:
qpid-config add exchange fanout AltEx1
qpid-config add queue AltQueue1
qpid-config bind AltEx1 AltQueue1
qpid-config add exchange fanout AltEx2
qpid-config add queue AltQueue2
qpid-config bind AltEx2 AltQueue2
qpid-config add exchange fanout AltEx3
qpid-config add queue AltQueue3
qpid-config bind AltEx3 AltQueue3
qpid-send --content-size 0 -m 100000 -a "AltQueue1"
Now, rerouting messages from AltQueue1 to AltEx2, then from Q2 to E3, then from Q3 to E1:
# qpid-stat -q; ps aux | grep -v grep | grep qpidd
Queues
queue dur autoDel excl msg msgIn msgOut bytes bytesIn bytesOut cons bind
==========================================================================================================================
2b34beae-1a60-4108-b02b-d20ce333b0e6:0.0 Y Y 0 0 0 0 0 0 1 2
AltQueue1 100k 100k 0 7.60m 7.60m 0 0 2
AltQueue2 0 0 0 0 0 0 0 2
AltQueue3 0 0 0 0 0 0 0 2
qpidd 341 8.5 0.3 893608 115156 ? Ssl 13:09 0:01 /usr/sbin/qpidd --config /etc/qpid/qpidd.conf --daemon --data-dir=/var/lib/qpidd --close-fd 9 --pidfile /var/run/qpidd.pid
#
# ./reroute_queue AltQueue1 0 AltEx2
{_arguments:{exchange:AltEx2, request:0, useAltExchange:False}, _method_name:reroute, _object_id:{_object_name:org.apache.qpid.broker:queue:AltQueue1}}
Response: OK
# qpid-stat -q; ps aux | grep -v grep | grep qpidd
Queues
queue dur autoDel excl msg msgIn msgOut bytes bytesIn bytesOut cons bind
==========================================================================================================================
AltQueue1 0 100k 100k 0 7.60m 7.60m 0 2
AltQueue2 100k 100k 0 7.60m 7.60m 0 0 2
AltQueue3 0 0 0 0 0 0 0 2
b5e7da49-5085-4d2c-aac5-cd924aef97a8:0.0 Y Y 0 0 0 0 0 0 1 2
qpidd 341 5.1 0.8 1067512 289068 ? Ssl 13:09 0:02 /usr/sbin/qpidd --config /etc/qpid/qpidd.conf --daemon --data-dir=/var/lib/qpidd --close-fd 9 --pidfile /var/run/qpidd.pid
#
# ./reroute_queue AltQueue2 0 AltEx3
{_arguments:{exchange:AltEx3, request:0, useAltExchange:False}, _method_name:reroute, _object_id:{_object_name:org.apache.qpid.broker:queue:AltQueue2}}
Response: OK
#
# qpid-stat -q; ps aux | grep -v grep | grep qpidd
Queues
queue dur autoDel excl msg msgIn msgOut bytes bytesIn bytesOut cons bind
==========================================================================================================================
10ee159e-53b7-45e2-a4eb-8a4ff57d60ab:0.0 Y Y 0 0 0 0 0 0 1 2
AltQueue1 0 100k 100k 0 7.60m 7.60m 0 2
AltQueue2 0 100k 100k 0 7.60m 7.60m 0 2
AltQueue3 100k 100k 0 7.60m 7.60m 0 0 2
qpidd 341 7.8 1.5 1264120 499624 ? Ssl 13:09 0:05 /usr/sbin/qpidd --config /etc/qpid/qpidd.conf --daemon --data-dir=/var/lib/qpidd --close-fd 9 --pidfile /var/run/qpidd.pid
#
# ./reroute_queue AltQueue3 0 AltEx1
{_arguments:{exchange:AltEx1, request:0, useAltExchange:False}, _method_name:reroute, _object_id:{_object_name:org.apache.qpid.broker:queue:AltQueue3}}
Response: OK
#
# qpid-stat -q; ps aux | grep -v grep | grep qpidd
Queues
queue dur autoDel excl msg msgIn msgOut bytes bytesIn bytesOut cons bind
==========================================================================================================================
505bff12-f7fb-4854-9978-5d135e8af400:0.0 Y Y 0 0 0 0 0 0 1 2
AltQueue1 100k 200k 100k 7.60m 15.2m 7.60m 0 2
AltQueue2 0 100k 100k 0 7.60m 7.60m 0 2
AltQueue3 0 100k 100k 0 7.60m 7.60m 0 2
qpidd 341 10.0 2.1 1460728 709864 ? Ssl 13:09 0:07 /usr/sbin/qpidd --config /etc/qpid/qpidd.conf --daemon --data-dir=/var/lib/qpidd --close-fd 9 --pidfile /var/run/qpidd.pid
#
See the memory increase after each reroute :-/
Now, C++ program for invoking the reroute message was used:
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Session.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Address.h>
#include <iostream>
#include <cstdlib>
using namespace std;
using namespace qpid::messaging;
using namespace qpid::types;
int main(int argc, char** argv) {
if (argc < 4) {
cerr << "Invalid number of parameters, expecting: queue_name, quantity, exchange_name" << endl;
return 1;
}
string queue_name = argv[1];
uint32_t qty = atoi(argv[2]);
string exchange_name = argv[3];
Connection connection(argc>4?argv[4]:"localhost:5672");
connection.open();
Session session = connection.createSession();
Sender sender = session.createSender("qmf.default.direct/broker");
Address responseQueue("#reply-queue; {create:always, node:{x-declare:{auto-delete:true}}}");
Receiver receiver = session.createReceiver(responseQueue);
Message message;
Variant::Map content;
Variant::Map OID;
Variant::Map arguments;
string object_name = "org.apache.qpid.broker:queue:" + queue_name;
OID["_object_name"] = object_name;
arguments["request"] = qty;
arguments["useAltExchange"] = false;
arguments["exchange"] = exchange_name;
content["_object_id"] = OID;
content["_method_name"] = "reroute";
content["_arguments"] = arguments;
encode(content, message);
message.setReplyTo(responseQueue);
message.setProperty("x-amqp-0-10.app-id", "qmf2");
message.setProperty("qmf.opcode", "_method_request");
message.setContentType("amqp/map");
Variant::Map map;
decode(message, map);
std::cout << map << std::endl;
sender.send(message, true);
Message response;
if (receiver.fetch(response,qpid::messaging::Duration(30000)) == true) {
qpid::types::Variant::Map recv_props = response.getProperties();
if (recv_props["x-amqp-0-10.app-id"] == "qmf2")
if (recv_props["qmf.opcode"] == "_method_response")
std::cout << "Response: OK" << std::endl;
else if (recv_props["qmf.opcode"] == "_exception")
std::cerr << "Error: " << response.getContent() << std::endl;
else
std::cerr << "Invalid response received!" << std::endl;
else
std::cerr << "Invalid response not of qmf2 type received!" << std::endl;
}
else
std::cout << "Timeout: No response received within 30 seconds!" << std::endl;
receiver.close();
sender.close();
session.close();
connection.close();
return 0;
}
Latest comment: if I split the "reroute all (100k) messages" to 1000 chunks of "reroute 100 messages", I get same memory consumption at the end. Which means, the reproducer is not sensitive to number of QMF reroute methods called or the number of messages rerouted in one batch, but of course it is sensitive to the total number of rerouted messages or their size. FYI re-routing 100k messages from AltQueue1 via AltEx2 to AltQueue2, then to AltQueue3, then to AltQueue1, etc. in a loop: - after first "iteration" (messages back in AltQueue1), memory consumption grew up from 120M to 713M (already too high, no?) - continuing in iterations, memory usage stabilized in interval 2.5G - 3.0G (3-4 times more than after 1st iteration - _so_ much..?) - consuming the messages and restarting the same test but for independent triplet of queues - just a small memory increase (to 3.2G), can be explained by allocating space for messages in yet another queues (and this space might not be fully released back after dequeue) Attached are two potential fixes submitted by the customer. Both show significant improvement. Fix 1 is more blunt. Fix 2 is more specific. Both are related to checking for persistence. Seems the code fails to check this in a few places. Customer would like us to review and make sure the patches do not cause unintended consequences. According to cust, option 1 footprint is reduced to about 300 MB and stays steady. Option 2 footprint is more like 2 GB and steady (proportional to size of message load used), definitely larger than option 1. Description from cust: "As per today call, we are attaching the 2 possible fixes for memory growth issue here. We did run valgrind and found possibly memory lost cases during reroute which matches to our reproducer scenario mentioned in the ticket 02582045. This Fix1 mainly concentrates on the persistent check for Traces. We added a guard code before clearTrace, addTrace to be called only when the persistence is enabled. You can see the code diff in the attachment. With this fix, the test results are very positive and we do not see any linear growth. Valgrind report also does not show any possibly lost cases. Fix2: This is similar to above fix but more surgical inside the Message.cpp where the annotationsChanged gets called only if the persistence is enabled. With this fix, the test results look positive." patch-2 ("if (isPersistent() && persistentContext)") :
- producer flow control (for transient messages) relies on persistentContext, it *should* work but definitely worth to test
- aged queues *might* be affected (havent reviewed code), but imho MCC does not use them
- sending AMQP1.0 transient messages is supposed to be affected, see src/qpid/broker/amqp/Translation.cpp and :
void Translation::write(OutgoingFromQueue& out)
{
const Message* message = dynamic_cast<const Message*>(original.getPersistentContext().get());
//persistent context will contain any newly added annotations
if (!message) message = dynamic_cast<const Message*>(&original.getEncoding());
if (message) {
//write annotations
qpid::amqp::CharSequence deliveryAnnotations = message->getDeliveryAnnotations();
qpid::amqp::CharSequence messageAnnotations = message->getMessageAnnotations();
..
patch-1 ("addTraceId only for persistent messages")
- won't this break x-qpid.trace for transient messages? I.e. same message can be enqueued to the same broker repeatedly?
I agree with Pavel's comment #11: I can see from code inspection that annotations are used for: 1. Message sequence when sequence key set: key = value set with "msg.queue_msg_sequence" - see Queue::push(Message&, bool) 2. Tracing - Adding and clearing traces: key="X_QPID_TRACE" (as Pavel mentioned above) - see Message::addTraceId(const std::string&) 3. Exchange pre-routing: key="qpid.msg_sequence" - see Exchange::PreRoute::PreRoute(Deliverable&, Exchange*) 4. AMQP 1.0: timestamping: key="x-opt-ingress-timestamp" - see DecodingIncoming::deliver(boost::intrusive_ptr<qpid::broker::amqp::Message>, pn_delivery_t*) 5. Proton disposition annotations - see Queue::mergeMessageAnnotations(const QueueCursor&, const qpid::types::Variant::Map&), and which is only called by OutgoingFromQueue::mergeMessageAnnotationsIfRequired(const Record &r), and which merges the annotations obtained from pn_disposition_annotations(pn_delivery_remote(r.delivery)) with the current message. All these would likely be broken for transient messages for patch #1. (In reply to Kim van der Riet from comment #12) > All these would likely be broken for transient messages for patch #1. Sorry, this was misleading. Patch #1 only affects tracing. However, any patch that depends on the message being persistent could potentially break any or all of the items listed above in comment #12. Test of Patch #2 using Pavel's reproducer from comment #1 bare-metal on machine with 16G ram: As reported by "ps aux" Itr# No-patch-#2 With-patch-#2 ----+------------+------------- 14300 14272 <- Broker only, no msgs 0 144912 145128 <- 100k messages w/ no bodies sent, no rerouting 1 345712 177024 <- First reroute 3 566148 239064 5 774444 270480 10 976404 240072 20 1365696 299468 30 1664568 330880 40 2001204 363212 50 1582064 394628 60 1784272 426044 70 1696492 426044 80 1699484 425988 90 1697756 425988 100 1599940 425988 and shows that patch #2 is effective at keeping the memory to around 425M, about 1/3 to 1/4 of the 1.5G to 2G in use without the patch. I have been testing attachment 1668808 [details] (patch #2) and a second patch (below) which tests for the presence of any trace annotations on a message before clearing them. Using Valgrind's massif tool to analyze the memory consumption, it is clear that clearing the message annotations unconditionally is unleashing a cascade of deep-copy operations of the message and it's frames which is responsible for much of the memory churn being observed.
The patches are as follows:
diff --git a/src/qpid/broker/Message.cpp b/src/qpid/broker/Message.cpp
index 8f39fbd1e..f6f31ea96 100644
--- a/src/qpid/broker/Message.cpp
+++ b/src/qpid/broker/Message.cpp
@@ -127,7 +127,9 @@ void Message::addTraceId(const std::string& id)
void Message::clearTrace()
{
- addAnnotation(X_QPID_TRACE, std::string());
+ if (annotations.get().find(X_QPID_TRACE)!= annotations.get().end()) {
+ addAnnotation(X_QPID_TRACE, std::string());
+ }
}
uint64_t Message::getTimestamp() const
@@ -170,7 +172,7 @@ void Message::addAnnotation(const std::string& key, const qpid::types::Variant&
void Message::annotationsChanged()
{
- if (persistentContext) {
+ if (isPersistent() && persistentContext) {
uint64_t id = persistentContext->getPersistenceId();
persistentContext = persistentContext->merge(getAnnotations());
persistentContext->setIngressCompletion(sharedState);
The upper patch is known as "TraceTest", the lower patch as "patch #2" as suggested by the client as a potential fix. These patches are not applied simultaneously for the test, but either none ("baseline"), one or the other. The test also varies the message size as follows: either empty messages (as used by Pavel in his reproducer in the description above, fixed 1kB messages or variable message sizes set randomly to between 0 and 1kB in size. The results are as follows:
-------- EMPTY MSGS -------- ------ MSGS FIXED 1k ------- MSGS VAR 0-1k
Itr baseline Patch#2 TraceTest baseline Patch#2 TraceTest TraceTest
--------------------------------- ---------------------------- -------------
Bkr 15032 14856 15784 14720 14112 15276 14176 <-- Broker only, no msgs
0 145140 144468 144868 250724 251304 251008 218152 <-- Initial upload of msgs, no rerouting
1 345348 176180 156444 557600 283040 284760 229380 <-- 1st reroute iteration
3 566596 237956 176772 884164 344832 306408 239148 etc.
5 777004 268576 187860 1508524 374132 283260 260268
10 1243980 237004 188652 1472648 346344 294272 256888
20 1180924 298252 188844 2034044 437696 294040 261000
30 1846176 393472 178180 2640696 471156 316216 246484
40 1847016 424940 201008 2675540 471692 305848 267388
50 2047684 456356 209612 2450912 471284 316584 267156
60 2043704 456356 231784 2677896 531852 304164 267600
70 2043188 456620 220996 2853956 531852 326996 288652
80 2010536 456620 231808 2902792 531852 327352 288736
90 1937796 425344 221368 3038796 531852 327352 268004
100 1825900 425608 221080 3089588 563268 327352 288856
125 348912 289580
150 338384 288624
175 360296 299416
200 338596 289024
225 338612 310864
250 360524 281848
275 349720 282088
300 338792 288528
350 349812 292292
400 349808 299420
450 349728 289248
500 349992 281640
550 349772 287676
600 350040 281656
650 349896 299992
700 349956 300008
750 350288 299932
800 361108 288976
850 361192 292692
900 337492 310736
950 349796 310796
1000 337296 270780
The two long-running tests (1000 iterations) show that the memory does not keep increasing, but tends to "surf" at around 360MB for fixed 1kB messages and at around 290MB for variable-sized messages.
Created attachment 1671903 [details] Proposed fix #3 This patch supersedes the "TraceTest" patch in comment #15, but provides the same memory use profile. The Message.annotations map is never populated by the broker, and is only used to *add* additional annotations to a message. This, the patch in comment #15 does not detect the presence of a trace annotation, and fails to clear it when it is present. This new patch uses the API's framework to decode AMQP 0-10 and 1.0 message formats to read the message properties and obtain the annotation. This patch passes all broker tests (make test), and with 1000 iterations of rerouting 100k messages with 1k bodies to the same queue, the RSS memory "surfs" at around the 350MB (whereas without the patch, it is usually around 10x that amount at 3.5GB), see comment #15. I have tested the patch against messages created using AMQP 1.0 and AMQP 0-10. Created attachment 1671906 [details] Proposed fix #3 See comment #15. I accidentally attached a patch for another bug, I have obsoleted that patch, use this one. Created attachment 1671936 [details]
"test.sh": Test script based on Pavel's reproducer in the description above.
The test calls a Python file "alt_to_fanout.py" also attached and provided py Pavel in the description above.
Created attachment 1671937 [details]
"alt_to_fanout.py": Script needed by test.sh to reroute messages.
Python script to reroute messages needed by "test.sh"
I have tested brew build 1.36.0-22+hf4.el6_10 on RHEL 6.9. I compared the previous build to the current build with the basic reproducer from above (50,000 1k messages rerouted 1000 times) hf3 build: stabilized at approximately 1.3 GB RSS hf4 build: stablilzed at approximately 168 MB RSS So it's clear that hf4 is way better than hf3 and seems stable, as the memory footprint stayed consistent throughout the run. With hf4, I also experimented with another scenario suggested by the customer: use messages with ttl, and have the broker purge regularly (every minute). I ran various scenarios including: 1) set ttl to 1 min, purge at 1 min, send a big set of messages every 2 min (i.e. let the message count go back to 0 before new messages are sent) 2) ttl = 1 min, purge = 1 min, send messages every 30 sec, let message sets overlap and accumulate before being purged I ran each of these for about 10 minutes, then let the broker continue rerouting for another 5 minutes to settle the memory. Case 1), the RSS memory got as high as 350, before settling at the end at 140 MB. Case 2), the RSS got as high as 520, before settling at 280 MB. (I also ran case 1) on build hf3 and memory use stabilized at around 780 MB RSS.) This raised a question as to whether the increased memory seen was an issue when using ttl and resending new messages. So I ran a longer test (18 hours, 20,000 reroute iterations) with both scenarios above mixed in, sometimes allowing the message count to get back to 0, then at the end letting the broker continue running for an extra 15 min with empty queues to let the memory float down to its baseline. In all tests, it was clear that the memory size was proportional to the size of the messages in the queue. In the long-running test, I maxed out at 100 MB of messages. At this point, the RSS memory could get up as high as 830 MB. However, once the broker was idle, the RSS would go down to around 240 MB, even after 18 hours. I take this as more definite than the short-term results from tests 1) and 2). I also noted that the VSZ tended to stabilize around 920 MB when idle, after going up higher if many messages were in the queue. The max VSZ I saw was around 1.4 GB (when RSS was 830). However, in test 1) above it always stayed around 910 and in test 2) would go a bit higher (1100) when multiple messages sets were in the queue at once, but always reverted back to the 920 range when idle. So in summary, the memory does seem to fluctuate up and down, but over an extended time, it appears to be stable for this single-queue scenario, even with overlapping sets of ttl messages. Comparing release 22 and hf3 and hf4 on a test bed "send 100k messages without a payload, with no or given x-qpid.trace; 20times do reroute and measure mem.increase". qpid-send can set trace via -P "x-qpid.trace=testTrace". format of dataset names: bz1804134.stats.<exchange>.<release>.<trace>.txt, so e.g. "bz1804134.stats.topicEx.22+hf3.el6_10..txt" used topic exchange on hf3 and _no_ trace / no -P used. bz1804134.stats.fanoutEx.22.el6.testTrace.txt 115344 294816 504260 533308 533380 677300 677380 821160 1030212 886280 1030176 1032808 825784 1034844 1174016 967772 967772 821576 886288 886288 886288 bz1804134.stats.fanoutEx.22.el6..txt 117420 296860 472192 680376 649380 679040 822948 825536 825548 825620 1034548 1178864 1181180 1181176 1113680 1266424 1410716 1413004 1415004 1272276 1414432 bz1804134.stats.fanoutEx.22+hf3.el6_10.testTrace.txt 117484 296624 472364 680768 680768 680768 681736 683332 684432 685484 895320 1095556 1095744 1095916 893472 898160 912728 912732 912732 915460 915492 bz1804134.stats.fanoutEx.22+hf3.el6_10..txt 117412 296904 503484 533328 677696 677880 887576 1031716 1034012 849268 678016 822124 824324 826352 1035784 1246156 1390440 1381772 1525708 1528316 1530392 bz1804134.stats.fanoutEx.22+hf4.el6_10.testTrace.txt 117420 360576 567720 712508 714856 717020 718084 928268 1137040 1137112 1132792 1276572 1279440 1281840 1281848 1425760 1425832 1279796 1279796 1279796 1280452 bz1804134.stats.fanoutEx.22+hf4.el6_10..txt 117376 187276 187448 196036 196036 195772 205104 197164 197184 197184 205724 205736 205780 205732 196960 205636 205292 205156 205476 197128 205696 bz1804134.stats.topicEx.22.el6.testTrace.txt 117236 294688 470452 681304 472572 680496 680496 887448 887560 887560 889568 891872 892768 893884 885324 1036048 1038640 1038704 851152 853900 853952 bz1804134.stats.topicEx.22.el6..txt 117528 295744 471532 681576 889732 1034020 1036336 1038328 1038524 1182140 1182260 1180240 1180240 1180240 1180912 1182212 707436 887400 1031468 1033756 1035984 bz1804134.stats.topicEx.22+hf3.el6_10.testTrace.txt 117548 296136 505924 535200 743304 743376 743496 743152 886936 889724 683672 891940 890272 1034056 890324 890324 1031340 824704 826752 1035032 1097072 bz1804134.stats.topicEx.22+hf3.el6_10..txt 117452 295184 470820 471660 679576 889212 889288 712524 856196 886780 886780 886780 832896 887016 1031392 1033920 1035964 1037156 830940 1038216 1246380 bz1804134.stats.topicEx.22+hf4.el6_10.testTrace.txt 117468 359828 567968 601032 746156 747304 957660 956692 1100680 1132808 1132880 1133012 1133012 781120 988320 988320 1132096 1134680 1136984 1137056 1276024 bz1804134.stats.topicEx.22+hf4.el6_10..txt 117284 186684 196592 202896 193436 196956 197728 188068 188068 194184 194156 188040 197680 197536 197700 188060 194024 194052 187924 187924 194068 diffs from mem.utilization before 1st reroute and after 20th: bz1804134.stats.fanoutEx.22.el6.testTrace.txt 770944 bz1804134.stats.fanoutEx.22.el6..txt 1297012 bz1804134.stats.fanoutEx.22+hf3.el6_10.testTrace.txt 798008 bz1804134.stats.fanoutEx.22+hf3.el6_10..txt 1412980 bz1804134.stats.fanoutEx.22+hf4.el6_10.testTrace.txt 1163032 bz1804134.stats.fanoutEx.22+hf4.el6_10..txt 88320 bz1804134.stats.topicEx.22.el6.testTrace.txt 736716 bz1804134.stats.topicEx.22.el6..txt 918456 bz1804134.stats.topicEx.22+hf3.el6_10.testTrace.txt 979524 bz1804134.stats.topicEx.22+hf3.el6_10..txt 1128928 bz1804134.stats.topicEx.22+hf4.el6_10.testTrace.txt 1158556 bz1804134.stats.topicEx.22+hf4.el6_10..txt 76784 Observations: - sometimes, there is an unexpected drop in memory consumption; no idea why - type of exchange does not matter. Sometimes fanout better, sometimes topic better, often very similar - 22.el6 and also hf3 adds _less_ memory when x-qpid.trace _was_ contained in 1st messages - hf4 adds a *lot* _more_ memory when x-qpid.trace _was_ contained in 1st messages Purely from the comparison, hf4 performs the best *but* it would need an improvement when x-qpid.trace is used in original messages. Reproducer script (run with arguments "20 ''" and "20 'testTrace'"): iterations=${1:-20} trace=${2} messages=${3:-100000} typeEx=fanout ################### change to topic if needed service qpidd restart qpid-config add exchange fanout AltEx qpid-config add exchange fanout FanoutEx --alternate-exchange=AltEx qpid-config add queue AltQueue qpid-config bind AltEx AltQueue ################ add '#' as binding key for topic exchange type if [ -n $trace ]; then trace="-P \"x-qpid.trace=$trace\"" fi qpid-send --content-size 0 -m $messages -a "FanoutEx" $trace qpid-receive -a "AltQueue; { mode:browse}" -m1 --print-headers=yes | grep Properties: echo ps aux | grep -v -e vim -e grep | grep qpidd for i in $(seq 1 $iterations); do python bz1804134_reroute_msgs.py ############### this is the .py script from #c0 sleep 2 ps aux | grep -v -e vim -e grep | grep qpidd done echo qpid-receive -a "AltQueue; { mode:browse}" -m1 --print-headers=yes | grep Properties: Created attachment 1675092 [details] Proposed fix #4 Patch #3 above (comment #17 above) works well if there are no trace attachments on the messages. However, if traces (ie x-qpid.trace is present in the header) *are* present, then an approach similar to Patch #2 above is required which will prevent the deep copy of the message needed for persistability, even when the message is not persistent. The problem with patch #2 is that it breaks AMQP 1.0 messages which requires the Message::merge() to be called, even for non-persistent messages. To solve this, if we introduce the following abstract method into PersistableMessage.h: virtual bool isMergeRequired() const = 0; and then implement these in amqp/Message.h/cpp and amqp_0_10/MessageTransfer.h/cpp (for AMQP 1.0 and AMQP 0-10 respectively). For AMQP 1.0 messages, this will always return true, and for AMQP 0-10 messages, this will return true only if the message is persistent. This is implemented in *addition* to patch #3 above, which bypasses all this for messages without trace headers. RSS for 100k msgs with 1k body, no ttl (to ensure messages to not time out during testing): ts=`date +%s` Messages with following headers: user-id="" reply-to="reply-to-string" subject="subject-string" priority=1 correlation-id=0 property qpid.subject="subject-string" property Timestamp=${ts} property x-amqp-0-10.routing-key="amqp-0-10-routing-key" property TransportUsed="blahblah" property specialHandling=0 property messageTtl=180 property MessageClass=0 property x-qpid.trace="trace-string" (for tests with trace, see table below) --------------- baseline -------------- ---- Patch#4 ----- no trace trace trace amqp0-10 amqp1.0 amqp0-10 amqp1.0 amqp0-10 amqp1.0 ---- ------------------ ------------------ ------------------ Bkr 13964 14476 14548 14308 14744 14736 <-- Broker only, no msgs 0 270208 224168 273232 227528 272660 227580 <-- Initial upload of msgs, no rerouting 1 423700 229252 862624 462120 476160 462012 <-- 1st reroute iteration 3 427392 236376 1749684 930716 504672 929816 etc. 5 428196 237428 1779248 930980 536132 929816 10 440604 245612 2203296 1163780 598696 1144268 20 429784 245808 3483368 1163912 598764 1163536 30 437208 237604 4042104 2081248 661332 1632212 40 438040 243084 4219792 2249196 725500 2043756 50 439080 250704 4098248 2249196 725500 1999792 60 443812 255540 3921832 2249196 694420 2234220 70 446072 251628 4043788 2483364 694420 2234284 80 452112 255728 4346504 2276256 694396 2334604 90 448032 255848 4596912 2276232 694396 2480936 100 448980 260600 4493080 2276232 694396 2428224 200 448452 251812 4286696 2426624 757364 2202408 300 447412 260284 4724520 2504836 757504 2428440 400 449904 255832 4561412 2250956 757504 2480956 500 457144 250300 4704024 2240420 757768 2478932 600 457164 256540 4762948 2278280 757708 2208752 700 447652 261056 4614368 2447800 757512 2481080 800 447304 261396 4722460 2485556 757880 2284204 900 447420 251680 4762812 2249520 757720 2259392 1000 447896 259420 4747544 2250884 757720 2493540 This shows that Patch #4 makes a big difference to AMQP 0-10 messages with trace annotations, but (as expected) makes no difference whatsoever for AMQP 1.0 messages. Created attachment 1675093 [details]
Proposed fix #4
I accidentally omitted the change to Message.cpp, please use this patch instead.
This product has been discontinued or is no longer tracked in Red Hat Bugzilla. |
Description of problem: qpidd consumes a lot of memory (but without an evident memory leak) when redirecting messages from a queue to an exchange ends up in enqueueing the messages in the same queue. Doing the redirect in a loop increases memory footprint during several iterations until the high memory usage gets stabilized - i.e. not an evident mem.leak, "just" very high memory usage. Example: - sending 100k messages with no payload, qpidd consumes 115M of RSS. - redirecting them once, qpidd mem.consumption grew to 288M (!) - after 10ish redirections, mem.consumption varies (a lot) around 2GB (!!) During the whole test, just the same 100k messages were enqueued. Version-Release number of selected component (if applicable): qpid-cpp-server-1.36.0-22+hf2.el6_10.x86_64 How reproducible: 100% Steps to Reproduce: 1. Have fanout exchanges FanoutEx and AltEx and AltQueue, where AltEx is the alternate exchange for FanoutEx service qpidd restart qpid-config add exchange fanout AltEx qpid-config add exchange fanout FanoutEx --alternate-exchange=AltEx qpid-config add queue AltQueue qpid-config bind AltEx AltQueue 2. send 100k message to FanoutEx that will reach AltQueue (since FanoutEx has no binding but alt.exchange): qpid-send --content-size 0 -m 100000 -a "FanoutEx" 3. check queues stats and memory usage: qpid-stat -q ps aux | grep -v grep | grep qpidd 4. redirect all messages from AltQueue to FanoutEx, i.e. via python script: #!/usr/bin/python import sys from qpid.messaging import * from qpidtoollibs import BrokerAgent import time broker = "localhost:5672" connection = Connection(broker) try: connection.open() session = connection.session() br = BrokerAgent(connection) queues = br.getAllQueues() myQueue = None for q in queues: if q.name == "AltQueue": myQueue = q myQueue.reroute(0,False,"FanoutEx") time.sleep(1) # just to get a response but silently ignore it except MessagingError,m: print m finally: connection.close() 5. repeat steps 3 and 4 to see impact of the redirect to mem.usage Actual results: 5. RSS grows from 115M to 2GB, despite messages keep the same inside the broker Expected results: 5. some mem.increase is expected, but 18 times more..? Additional info: I played with parameters of the test (# messages, their size, type of exchange, ring queue,..) but nothing seems to have an evident impact to the growth. Increasing # messages or their size does have an increase in the mem.growth - but only linear, that is expected.