diff options
Diffstat (limited to 'qpid/cpp/src/tests/QueueFlowLimitTest.cpp')
-rw-r--r-- | qpid/cpp/src/tests/QueueFlowLimitTest.cpp | 457 |
1 files changed, 0 insertions, 457 deletions
diff --git a/qpid/cpp/src/tests/QueueFlowLimitTest.cpp b/qpid/cpp/src/tests/QueueFlowLimitTest.cpp deleted file mode 100644 index b35294922d..0000000000 --- a/qpid/cpp/src/tests/QueueFlowLimitTest.cpp +++ /dev/null @@ -1,457 +0,0 @@ -/* - * - * 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/QueueFlowLimit.h" -#include "qpid/broker/QueueSettings.h" -#include "qpid/sys/Time.h" -#include "qpid/framing/reply_exceptions.h" -#include "qpid/framing/FieldValue.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("", 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 boost::shared_ptr<qpid::broker::QueueFlowLimit> getQueueFlowLimit(const qpid::framing::FieldTable& arguments) - { - QueueSettings settings; - settings.populate(arguments, settings.storeSettings); - return QueueFlowLimit::createLimit("", settings); - } -}; - -Message createMessage(uint32_t size) -{ - static uint32_t seqNum; - //Need to compute what data size is required to make a given - //overall size (use one byte of content in test message to ensure - //content frame is added) - Message test = MessageUtils::createMessage(qpid::types::Variant::Map(), std::string("x")); - size_t min = test.getMessageSize() - 1; - if (min > size) throw qpid::Exception("Can't create message that small!"); - Message msg = MessageUtils::createMessage(qpid::types::Variant::Map(), std::string (size - min, 'x')); - msg.setSequence(++seqNum);//this doesn't affect message size - 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<Message> msgs; - for (size_t i = 0; i < 6; i++) { - msgs.push_back(createMessage(100)); - flow->enqueued(msgs.back()); - BOOST_CHECK(!flow->isFlowControlActive()); - } - BOOST_CHECK(!flow->isFlowControlActive()); // 6 on queue - msgs.push_back(createMessage(100)); - flow->enqueued(msgs.back()); - BOOST_CHECK(!flow->isFlowControlActive()); // 7 on queue - msgs.push_back(createMessage(100)); - flow->enqueued(msgs.back()); - BOOST_CHECK(flow->isFlowControlActive()); // 8 on queue, ON - msgs.push_back(createMessage(100)); - 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, 700); - args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 460); - - 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) 700, flow->getFlowStopSize()); - BOOST_CHECK_EQUAL((uint32_t) 460, flow->getFlowResumeSize()); - BOOST_CHECK(!flow->isFlowControlActive()); - BOOST_CHECK(flow->monitorFlowControl()); - - std::deque<Message> msgs; - for (size_t i = 0; i < 6; i++) { - msgs.push_back(createMessage(100)); - flow->enqueued(msgs.back()); - BOOST_CHECK(!flow->isFlowControlActive()); - } - BOOST_CHECK(!flow->isFlowControlActive()); // 600 on queue - BOOST_CHECK_EQUAL(6u, flow->getFlowCount()); - BOOST_CHECK_EQUAL(600u, flow->getFlowSize()); - - Message msg_50 = createMessage(50); - flow->enqueued(msg_50); - BOOST_CHECK(!flow->isFlowControlActive()); // 650 on queue - Message tinyMsg_1 = createMessage(40); - flow->enqueued(tinyMsg_1); - BOOST_CHECK(!flow->isFlowControlActive()); // 690 on queue - - Message tinyMsg_2 = createMessage(40); - flow->enqueued(tinyMsg_2); - BOOST_CHECK(flow->isFlowControlActive()); // 730 on queue, ON - msgs.push_back(createMessage(100)); - flow->enqueued(msgs.back()); - BOOST_CHECK(flow->isFlowControlActive()); // 830 on queue - BOOST_CHECK_EQUAL(10u, flow->getFlowCount()); - BOOST_CHECK_EQUAL(830u, flow->getFlowSize()); - - flow->dequeued(msgs.front()); - msgs.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); // 730 on queue - flow->dequeued(msgs.front()); - msgs.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); // 630 on queue - flow->dequeued(msgs.front()); - msgs.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); // 530 on queue - - flow->dequeued(tinyMsg_1); - BOOST_CHECK(flow->isFlowControlActive()); // 490 on queue - flow->dequeued(tinyMsg_2); - BOOST_CHECK(!flow->isFlowControlActive()); // 450 on queue, OFF - - flow->dequeued(msg_50); - BOOST_CHECK(!flow->isFlowControlActive()); // 400 on queue - flow->dequeued(msgs.front()); - msgs.pop_front(); - BOOST_CHECK(!flow->isFlowControlActive()); // 300 on queue - flow->dequeued(msgs.front()); - msgs.pop_front(); - BOOST_CHECK(!flow->isFlowControlActive()); // 200 on queue - BOOST_CHECK_EQUAL(2u, flow->getFlowCount()); - BOOST_CHECK_EQUAL(200u, 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, 2000); - args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 1000); - - std::deque<Message> msgs_50; - std::deque<Message> msgs_100; - std::deque<Message> msgs_500; - std::deque<Message> msgs_1000; - - Message 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_100.push_back(createMessage(100)); - flow->enqueued(msgs_100.back()); - BOOST_CHECK(!flow->isFlowControlActive()); - } - // count:10 size:1000 - - msgs_50.push_back(createMessage(50)); - flow->enqueued(msgs_50.back()); // count:11 size: 1050 ->ON - BOOST_CHECK(flow->isFlowControlActive()); - - for (size_t i = 0; i < 6; i++) { - flow->dequeued(msgs_100.front()); - msgs_100.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); - } - // count:5 size: 450 - - flow->dequeued(msgs_50.front()); // count: 4 size: 400 ->OFF - msgs_50.pop_front(); - BOOST_CHECK(!flow->isFlowControlActive()); - - for (size_t i = 0; i < 4; i++) { - flow->dequeued(msgs_100.front()); - msgs_100.pop_front(); - BOOST_CHECK(!flow->isFlowControlActive()); - } - // count:0 size:0 - - // verify flow control comes ON when only size passes its stop point. - - msgs_1000.push_back(createMessage(1000)); - flow->enqueued(msgs_1000.back()); // count:1 size: 1000 - BOOST_CHECK(!flow->isFlowControlActive()); - - msgs_500.push_back(createMessage(500)); - flow->enqueued(msgs_500.back()); // count:2 size: 1500 - BOOST_CHECK(!flow->isFlowControlActive()); - - msgs_500.push_back(createMessage(500)); - flow->enqueued(msgs_500.back()); // count:3 size: 2000 - BOOST_CHECK(!flow->isFlowControlActive()); - - msgs_50.push_back(createMessage(50)); - flow->enqueued(msgs_50.back()); // count:4 size: 2050 ->ON - BOOST_CHECK(flow->isFlowControlActive()); - - flow->dequeued(msgs_1000.front()); // count:3 size:1050 - msgs_1000.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); - - flow->dequeued(msgs_50.front()); // count:2 size:1000 - msgs_50.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); - - flow->dequeued(msgs_500.front()); // count:1 size:500 ->OFF - msgs_500.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_100.push_back(createMessage(100)); - flow->enqueued(msgs_100.back()); - BOOST_CHECK(!flow->isFlowControlActive()); - } - // count:9 size:1300 - - msgs_100.push_back(createMessage(100)); - flow->enqueued(msgs_100.back()); // count:10 size: 1400 - BOOST_CHECK(!flow->isFlowControlActive()); - - msgs_50.push_back(createMessage(50)); - flow->enqueued(msgs_50.back()); // count:11 size: 1450 ->ON - BOOST_CHECK(flow->isFlowControlActive()); - - msgs_1000.push_back(createMessage(1000)); - flow->enqueued(msgs_1000.back()); // count:12 size: 2450 (both thresholds crossed) - BOOST_CHECK(flow->isFlowControlActive()); - - // at this point: 9@100 + 1@500 + 1@1000 + 1@50 == 12@2450 - - flow->dequeued(msgs_500.front()); // count:11 size:1950 - msgs_500.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); - - for (size_t i = 0; i < 9; i++) { - flow->dequeued(msgs_100.front()); - msgs_100.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); - } - // count:2 size:1050 - flow->dequeued(msgs_50.front()); // count:1 size:1000 - msgs_50.pop_front(); - BOOST_CHECK(flow->isFlowControlActive()); // still active due to size - - flow->dequeued(msgs_1000.front()); // count:0 size:0 ->OFF - msgs_1000.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; - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(flow); - - 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(0, // max queue byte count - 80, // 80% stop threshold - 70); // 70% resume threshold - { - FieldTable args; - args.setInt(QueueFlowLimit::flowStopCountKey, 35000); - args.setInt(QueueFlowLimit::flowResumeCountKey, 30000); -// args.setInt(QueueFlowLimit::flowStopSizeKey, 0); - - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(flow); - - 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); - - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(flow); - - 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); - - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(flow); - - 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; - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(flow); - - 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); - args.setInt(QueueFlowLimit::flowStopSizeKey, 0); - boost::shared_ptr<QueueFlowLimit> flow = TestFlow::getQueueFlowLimit(args); - BOOST_CHECK(!flow); - } -} - -QPID_AUTO_TEST_SUITE_END() - -}} // namespace qpid::tests |