diff options
Diffstat (limited to 'qpid/cpp/src/tests/QueueFlowLimitTest.cpp')
| -rw-r--r-- | qpid/cpp/src/tests/QueueFlowLimitTest.cpp | 463 |
1 files changed, 463 insertions, 0 deletions
diff --git a/qpid/cpp/src/tests/QueueFlowLimitTest.cpp b/qpid/cpp/src/tests/QueueFlowLimitTest.cpp new file mode 100644 index 0000000000..8a6923fb09 --- /dev/null +++ b/qpid/cpp/src/tests/QueueFlowLimitTest.cpp @@ -0,0 +1,463 @@ +/* + * + * 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 <sstream> +#include <deque> +#include "unit_test.h" +#include "test_tools.h" + +#include "qpid/broker/QueuePolicy.h" +#include "qpid/broker/QueueFlowLimit.h" +#include "qpid/sys/Time.h" +#include "qpid/framing/reply_exceptions.h" +#include "MessageUtils.h" +#include "BrokerFixture.h" + +using namespace qpid::broker; +using namespace qpid::framing; + +namespace qpid { +namespace tests { + +QPID_AUTO_TEST_SUITE(QueueFlowLimitTestSuite) + +namespace { + +class TestFlow : public QueueFlowLimit +{ +public: + TestFlow(uint32_t flowStopCount, uint32_t flowResumeCount, + uint64_t flowStopSize, uint64_t flowResumeSize) : + QueueFlowLimit(0, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize) + {} + virtual ~TestFlow() {} + + static TestFlow *createTestFlow(const qpid::framing::FieldTable& settings) + { + FieldTable::ValuePtr v; + + v = settings.get(flowStopCountKey); + uint32_t flowStopCount = (v) ? (uint32_t)v->get<int64_t>() : 0; + v = settings.get(flowResumeCountKey); + uint32_t flowResumeCount = (v) ? (uint32_t)v->get<int64_t>() : 0; + v = settings.get(flowStopSizeKey); + uint64_t flowStopSize = (v) ? (uint64_t)v->get<int64_t>() : 0; + v = settings.get(flowResumeSizeKey); + uint64_t flowResumeSize = (v) ? (uint64_t)v->get<int64_t>() : 0; + + return new TestFlow(flowStopCount, flowResumeCount, flowStopSize, flowResumeSize); + } + + static QueueFlowLimit *getQueueFlowLimit(const qpid::framing::FieldTable& settings) + { + return QueueFlowLimit::createLimit(0, settings); + } +}; + + + +QueuedMessage createMessage(uint32_t size) +{ + static uint32_t seqNum; + QueuedMessage msg; + msg.payload = MessageUtils::createMessage(); + msg.position = ++seqNum; + MessageUtils::addContent(msg.payload, std::string (size, 'x')); + return msg; +} +} + +QPID_AUTO_TEST_CASE(testFlowCount) +{ + FieldTable args; + args.setInt(QueueFlowLimit::flowStopCountKey, 7); + args.setInt(QueueFlowLimit::flowResumeCountKey, 5); + + std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args)); + + BOOST_CHECK_EQUAL((uint32_t) 7, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 5, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); + + std::deque<QueuedMessage> msgs; + for (size_t i = 0; i < 6; i++) { + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(!flow->isFlowControlActive()); + } + BOOST_CHECK(!flow->isFlowControlActive()); // 6 on queue + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(!flow->isFlowControlActive()); // 7 on queue + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(flow->isFlowControlActive()); // 8 on queue, ON + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(flow->isFlowControlActive()); // 9 on queue, no change to flow control + + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 8 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 7 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 6 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 5 on queue, no change + + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); // 4 on queue, OFF +} + + +QPID_AUTO_TEST_CASE(testFlowSize) +{ + FieldTable args; + args.setUInt64(QueueFlowLimit::flowStopSizeKey, 70); + args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 50); + + std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args)); + + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL((uint32_t) 70, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint32_t) 50, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); + + std::deque<QueuedMessage> msgs; + for (size_t i = 0; i < 6; i++) { + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(!flow->isFlowControlActive()); + } + BOOST_CHECK(!flow->isFlowControlActive()); // 60 on queue + BOOST_CHECK_EQUAL(6u, flow->getFlowCount()); + BOOST_CHECK_EQUAL(60u, flow->getFlowSize()); + + QueuedMessage msg_9 = createMessage(9); + flow->enqueued(msg_9); + BOOST_CHECK(!flow->isFlowControlActive()); // 69 on queue + QueuedMessage tinyMsg_1 = createMessage(1); + flow->enqueued(tinyMsg_1); + BOOST_CHECK(!flow->isFlowControlActive()); // 70 on queue + + QueuedMessage tinyMsg_2 = createMessage(1); + flow->enqueued(tinyMsg_2); + BOOST_CHECK(flow->isFlowControlActive()); // 71 on queue, ON + msgs.push_back(createMessage(10)); + flow->enqueued(msgs.back()); + BOOST_CHECK(flow->isFlowControlActive()); // 81 on queue + BOOST_CHECK_EQUAL(10u, flow->getFlowCount()); + BOOST_CHECK_EQUAL(81u, flow->getFlowSize()); + + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 71 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 61 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // 51 on queue + + flow->dequeued(tinyMsg_1); + BOOST_CHECK(flow->isFlowControlActive()); // 50 on queue + flow->dequeued(tinyMsg_2); + BOOST_CHECK(!flow->isFlowControlActive()); // 49 on queue, OFF + + flow->dequeued(msg_9); + BOOST_CHECK(!flow->isFlowControlActive()); // 40 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); // 30 on queue + flow->dequeued(msgs.front()); + msgs.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); // 20 on queue + BOOST_CHECK_EQUAL(2u, flow->getFlowCount()); + BOOST_CHECK_EQUAL(20u, flow->getFlowSize()); +} + +QPID_AUTO_TEST_CASE(testFlowArgs) +{ + FieldTable args; + const uint64_t stop(0x2FFFFFFFFull); + const uint64_t resume(0x1FFFFFFFFull); + args.setInt(QueueFlowLimit::flowStopCountKey, 30); + args.setInt(QueueFlowLimit::flowResumeCountKey, 21); + args.setUInt64(QueueFlowLimit::flowStopSizeKey, stop); + args.setUInt64(QueueFlowLimit::flowResumeSizeKey, resume); + + std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args)); + + BOOST_CHECK_EQUAL((uint32_t) 30, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 21, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL(stop, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL(resume, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); +} + + +QPID_AUTO_TEST_CASE(testFlowCombo) +{ + FieldTable args; + args.setInt(QueueFlowLimit::flowStopCountKey, 10); + args.setInt(QueueFlowLimit::flowResumeCountKey, 5); + args.setUInt64(QueueFlowLimit::flowStopSizeKey, 200); + args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 100); + + std::deque<QueuedMessage> msgs_1; + std::deque<QueuedMessage> msgs_10; + std::deque<QueuedMessage> msgs_50; + std::deque<QueuedMessage> msgs_100; + + QueuedMessage msg; + + std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args)); + BOOST_CHECK(!flow->isFlowControlActive()); // count:0 size:0 + + // verify flow control comes ON when only count passes its stop point. + + for (size_t i = 0; i < 10; i++) { + msgs_10.push_back(createMessage(10)); + flow->enqueued(msgs_10.back()); + BOOST_CHECK(!flow->isFlowControlActive()); + } + // count:10 size:100 + + msgs_1.push_back(createMessage(1)); + flow->enqueued(msgs_1.back()); // count:11 size: 101 ->ON + BOOST_CHECK(flow->isFlowControlActive()); + + for (size_t i = 0; i < 6; i++) { + flow->dequeued(msgs_10.front()); + msgs_10.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); + } + // count:5 size: 41 + + flow->dequeued(msgs_1.front()); // count: 4 size: 40 ->OFF + msgs_1.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); + + for (size_t i = 0; i < 4; i++) { + flow->dequeued(msgs_10.front()); + msgs_10.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); + } + // count:0 size:0 + + // verify flow control comes ON when only size passes its stop point. + + msgs_100.push_back(createMessage(100)); + flow->enqueued(msgs_100.back()); // count:1 size: 100 + BOOST_CHECK(!flow->isFlowControlActive()); + + msgs_50.push_back(createMessage(50)); + flow->enqueued(msgs_50.back()); // count:2 size: 150 + BOOST_CHECK(!flow->isFlowControlActive()); + + msgs_50.push_back(createMessage(50)); + flow->enqueued(msgs_50.back()); // count:3 size: 200 + BOOST_CHECK(!flow->isFlowControlActive()); + + msgs_1.push_back(createMessage(1)); + flow->enqueued(msgs_1.back()); // count:4 size: 201 ->ON + BOOST_CHECK(flow->isFlowControlActive()); + + flow->dequeued(msgs_100.front()); // count:3 size:101 + msgs_100.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); + + flow->dequeued(msgs_1.front()); // count:2 size:100 + msgs_1.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); + + flow->dequeued(msgs_50.front()); // count:1 size:50 ->OFF + msgs_50.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); + + // verify flow control remains ON until both thresholds drop below their + // resume point. + + for (size_t i = 0; i < 8; i++) { + msgs_10.push_back(createMessage(10)); + flow->enqueued(msgs_10.back()); + BOOST_CHECK(!flow->isFlowControlActive()); + } + // count:9 size:130 + + msgs_10.push_back(createMessage(10)); + flow->enqueued(msgs_10.back()); // count:10 size: 140 + BOOST_CHECK(!flow->isFlowControlActive()); + + msgs_1.push_back(createMessage(1)); + flow->enqueued(msgs_1.back()); // count:11 size: 141 ->ON + BOOST_CHECK(flow->isFlowControlActive()); + + msgs_100.push_back(createMessage(100)); + flow->enqueued(msgs_100.back()); // count:12 size: 241 (both thresholds crossed) + BOOST_CHECK(flow->isFlowControlActive()); + + // at this point: 9@10 + 1@50 + 1@100 + 1@1 == 12@241 + + flow->dequeued(msgs_50.front()); // count:11 size:191 + msgs_50.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); + + for (size_t i = 0; i < 9; i++) { + flow->dequeued(msgs_10.front()); + msgs_10.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); + } + // count:2 size:101 + flow->dequeued(msgs_1.front()); // count:1 size:100 + msgs_1.pop_front(); + BOOST_CHECK(flow->isFlowControlActive()); // still active due to size + + flow->dequeued(msgs_100.front()); // count:0 size:0 ->OFF + msgs_100.pop_front(); + BOOST_CHECK(!flow->isFlowControlActive()); +} + + +QPID_AUTO_TEST_CASE(testFlowDefaultArgs) +{ + QueueFlowLimit::setDefaults(2950001, // max queue byte count + 80, // 80% stop threshold + 70); // 70% resume threshold + FieldTable args; + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + + BOOST_CHECK(ptr); + std::auto_ptr<QueueFlowLimit> flow(ptr); + BOOST_CHECK_EQUAL((uint64_t) 2360001, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint64_t) 2065000, flow->getFlowResumeSize()); + BOOST_CHECK_EQUAL( 0u, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL( 0u, flow->getFlowResumeCount()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); +} + + +QPID_AUTO_TEST_CASE(testFlowOverrideArgs) +{ + QueueFlowLimit::setDefaults(2950001, // max queue byte count + 80, // 80% stop threshold + 70); // 70% resume threshold + { + FieldTable args; + args.setInt(QueueFlowLimit::flowStopCountKey, 35000); + args.setInt(QueueFlowLimit::flowResumeCountKey, 30000); + + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(ptr); + std::auto_ptr<QueueFlowLimit> flow(ptr); + + BOOST_CHECK_EQUAL((uint32_t) 35000, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 30000, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL((uint64_t) 0, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint64_t) 0, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); + } + { + FieldTable args; + args.setInt(QueueFlowLimit::flowStopSizeKey, 350000); + args.setInt(QueueFlowLimit::flowResumeSizeKey, 300000); + + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(ptr); + std::auto_ptr<QueueFlowLimit> flow(ptr); + + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL((uint64_t) 350000, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint64_t) 300000, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); + } + { + FieldTable args; + args.setInt(QueueFlowLimit::flowStopCountKey, 35000); + args.setInt(QueueFlowLimit::flowResumeCountKey, 30000); + args.setInt(QueueFlowLimit::flowStopSizeKey, 350000); + args.setInt(QueueFlowLimit::flowResumeSizeKey, 300000); + + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(ptr); + std::auto_ptr<QueueFlowLimit> flow(ptr); + + BOOST_CHECK_EQUAL((uint32_t) 35000, flow->getFlowStopCount()); + BOOST_CHECK_EQUAL((uint32_t) 30000, flow->getFlowResumeCount()); + BOOST_CHECK_EQUAL((uint64_t) 350000, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint64_t) 300000, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); + } +} + + +QPID_AUTO_TEST_CASE(testFlowOverrideDefaults) +{ + QueueFlowLimit::setDefaults(2950001, // max queue byte count + 97, // stop threshold + 73); // resume threshold + FieldTable args; + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(ptr); + std::auto_ptr<QueueFlowLimit> flow(ptr); + + BOOST_CHECK_EQUAL((uint32_t) 2861501, flow->getFlowStopSize()); + BOOST_CHECK_EQUAL((uint32_t) 2153500, flow->getFlowResumeSize()); + BOOST_CHECK(!flow->isFlowControlActive()); + BOOST_CHECK(flow->monitorFlowControl()); +} + + +QPID_AUTO_TEST_CASE(testFlowDisable) +{ + { + FieldTable args; + args.setInt(QueueFlowLimit::flowStopCountKey, 0); + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(!ptr); + } + { + FieldTable args; + args.setInt(QueueFlowLimit::flowStopSizeKey, 0); + QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args); + BOOST_CHECK(!ptr); + } +} + + +QPID_AUTO_TEST_SUITE_END() + +}} // namespace qpid::tests |