From 1f473a33b3e5b5becd77b14508b06389c2ecd70a Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Tue, 3 May 2011 22:04:51 +0000 Subject: QPID-3243: correctly use --max-queue-count value to compute flow limit. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1099278 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp | 12 +++++++++--- qpid/cpp/src/tests/QueuePolicyTest.cpp | 1 + qpid/cpp/src/tests/cluster_tests.py | 2 +- qpid/cpp/src/tests/queue_flow_limit_tests.py | 21 +++++++++++++++++++++ 4 files changed, 32 insertions(+), 4 deletions(-) diff --git a/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp b/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp index 5abd7fe666..9f1d3b65f8 100644 --- a/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp +++ b/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp @@ -282,7 +282,9 @@ QueueFlowLimit *QueueFlowLimit::createLimit(Queue *queue, const qpid::framing::F return 0; } - if (settings.get(flowStopCountKey) || settings.get(flowStopSizeKey)) { + if (settings.get(flowStopCountKey) || settings.get(flowStopSizeKey) || + settings.get(flowResumeCountKey) || settings.get(flowResumeSizeKey)) { + // user provided (some) flow settings manually... uint32_t flowStopCount = getCapacity(settings, flowStopCountKey, 0); uint32_t flowResumeCount = getCapacity(settings, flowResumeCountKey, 0); uint64_t flowStopSize = getCapacity(settings, flowStopSizeKey, 0); @@ -293,11 +295,15 @@ QueueFlowLimit *QueueFlowLimit::createLimit(Queue *queue, const qpid::framing::F return new QueueFlowLimit(queue, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize); } - if (defaultFlowStopRatio) { + if (defaultFlowStopRatio) { // broker has a default ratio setup... uint64_t maxByteCount = getCapacity(settings, QueuePolicy::maxSizeKey, defaultMaxSize); uint64_t flowStopSize = (uint64_t)(maxByteCount * (defaultFlowStopRatio/100.0) + 0.5); uint64_t flowResumeSize = (uint64_t)(maxByteCount * (defaultFlowResumeRatio/100.0)); - return new QueueFlowLimit(queue, 0, 0, flowStopSize, flowResumeSize); + uint32_t maxMsgCount = getCapacity(settings, QueuePolicy::maxCountKey, 0); // no size by default + uint32_t flowStopCount = (uint32_t)(maxMsgCount * (defaultFlowStopRatio/100.0) + 0.5); + uint32_t flowResumeCount = (uint32_t)(maxMsgCount * (defaultFlowResumeRatio/100.0)); + + return new QueueFlowLimit(queue, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize); } return 0; } diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index f9c058c771..5455105078 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -386,6 +386,7 @@ QPID_AUTO_TEST_CASE(testCapacityConversion) { FieldTable args; args.setString("qpid.max_count", "5"); + args.setString("qpid.flow_stop_count", "0"); ProxySessionFixture f; std::string q("q"); diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py index 727934ae26..85ace1008a 100755 --- a/qpid/cpp/src/tests/cluster_tests.py +++ b/qpid/cpp/src/tests/cluster_tests.py @@ -567,7 +567,7 @@ acl allow all all s0 = c0.session() # Declare multiple queues bound to same key on amq.topic def declare(q,max=0): - if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d}}'%max + if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d, "qpid.flow_stop_count":0}}'%max else: declare = 'x-declare:{}' bind='x-bindings:[{queue:%s,key:key,exchange:"amq.topic"}]'%(q) s0.sender("%s;{create:always,node:{%s,%s}}" % (q,declare,bind)) diff --git a/qpid/cpp/src/tests/queue_flow_limit_tests.py b/qpid/cpp/src/tests/queue_flow_limit_tests.py index 6639cdc393..51f91647fb 100644 --- a/qpid/cpp/src/tests/queue_flow_limit_tests.py +++ b/qpid/cpp/src/tests/queue_flow_limit_tests.py @@ -129,6 +129,27 @@ class QueueFlowLimitTests(TestBase010): self.assertEqual(i.name, "test01") self._delete_queue("test01") + # now verify that the default ratios are applied if max sizing is specified: + command = tool + \ + " --broker-addr=%s:%s " % (self.broker.host, self.broker.port) \ + + "add queue test02 --max-queue-count=10000 --max-queue-size=1000000" + cmd = popen(command) + rc = cmd.close() + self.assertEqual(rc, None) + + # now verify the settings + qs = self.qmf.getObjects(_class="queue") + for i in qs: + if i.name == "test02": + ## @todo KAG: can't get the flow size from qmf! Arrgh! + # no way to verify... + #self.assertEqual(i.arguments.get("qpid.flow_resume_count"), 55) + #self.assertEqual(i.arguments.get("qpid.flow_resume_count"), 55) + self.failIf(i.flowStopped) + break; + self.assertEqual(i.name, "test02") + self._delete_queue("test02") + def test_flow_count(self): """ Create a queue with count-based flow limit. Spawn several -- cgit v1.2.1