Commit 036fc630 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Add WDRR queue with dynamic number of queues

parent 3e1a34ff
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_INTRUSIVE_WDRR_DYNAMIC_MULTIPLEXED_QUEUE_HPP
#define CAF_INTRUSIVE_WDRR_DYNAMIC_MULTIPLEXED_QUEUE_HPP
#include <utility>
#include "caf/detail/type_traits.hpp"
namespace caf {
namespace intrusive {
/// A work queue that internally multiplexes any number of DRR queues.
template <class Policy>
class wdrr_dynamic_multiplexed_queue {
public:
using policy_type = Policy;
using deleter_type = typename policy_type::deleter_type;
using deficit_type = typename policy_type::deficit_type;
using mapped_type = typename policy_type::mapped_type;
using key_type = typename policy_type::key_type;
using pointer = mapped_type*;
using unique_pointer = typename policy_type::unique_pointer;
using task_size_type = typename policy_type::task_size_type;
using queue_map_type = typename policy_type::queue_map_type;
using queue_type = typename queue_map_type::mapped_type;
template <class... Ps>
wdrr_dynamic_multiplexed_queue(policy_type p) : policy_(p) {
// nop
}
policy_type& policy() noexcept {
return policy_;
}
const policy_type& policy() const noexcept {
return policy_;
}
void push_back(mapped_type* ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr));
if (i != qs_.end()) {
i->second.push_back(ptr);
} else {
deleter_type d;
d(ptr);
}
}
void push_back(unique_pointer ptr) noexcept {
push_back(ptr.release());
}
template <class... Ts>
void emplace_back(Ts&&... xs) {
push_back(new mapped_type(std::forward<Ts>(xs)...));
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
bool new_round(long quantum,
F& f) noexcept(noexcept(f(std::declval<const key_type&>(),
std::declval<queue_type&>(),
std::declval<mapped_type&>()))) {
bool result = false;
for (auto& kvp : qs_) {
auto g = [&](mapped_type& x) {
f(kvp.first, kvp.second, x);
};
result |= kvp.second.new_round(policy_.quantum(kvp.second, quantum), g);
}
return result;
}
pointer peek() noexcept {
for (auto& kvp : qs_) {
auto ptr = kvp.second.peek();
if (ptr != nullptr)
return ptr;
}
return nullptr;
}
/// Returns `true` if all queues are empty, `false` otherwise.
bool empty() const noexcept {
return total_task_size() == 0;
}
void flush_cache() noexcept {
for (auto& kvp : qs_)
kvp.second.flush_cache();
}
task_size_type total_task_size() const noexcept {
task_size_type result = 0;
for (auto& kvp : qs_)
result += kvp.second.total_task_size();
return result;
}
/// Returns the tuple containing all nested queues.
queue_map_type& queues() noexcept {
return qs_;
}
/// Returns the tuple containing all nested queues.
const queue_map_type& queues() const noexcept {
return qs_;
}
void lifo_append(pointer ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr));
if (i != qs_.end()) {
i->second.lifo_append(ptr);
} else {
deleter_type d;
d(ptr);
}
}
void stop_lifo_append() noexcept {
for (auto& kvp : qs_)
kvp.second.stop_lifo_append();
}
private:
// -- member variables -------------------------------------------------------
/// All queues.
queue_map_type qs_;
/// Policy object for adjusting a quantum based on queue weight etc.
Policy policy_;
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_WDRR_DYNAMIC_MULTIPLEXED_QUEUE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE wdrr_dynamic_multiplexed_queue
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct nested_inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static inline task_size_type task_size(const mapped_type&) {
return 1;
}
std::unique_ptr<int> queue_id;
nested_inode_policy(int i) : queue_id(new int(i)) {
// nop
}
nested_inode_policy(nested_inode_policy&&) = default;
nested_inode_policy& operator=(nested_inode_policy&&) = default;
};
struct inode_policy {
using mapped_type = inode;
using key_type = int;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
using queue_type = drr_queue<nested_inode_policy>;
using queue_map_type = std::map<key_type, queue_type>;
static inline task_size_type task_size(const mapped_type&) {
return 1;
}
static inline key_type id_of(const inode& x) {
return x.value % 3;
}
deficit_type quantum(const queue_type& q, deficit_type x) {
return enable_priorities && *q.policy().queue_id == 0 ? 2 * x : x;
}
bool enable_priorities = false;
};
using queue_type = wdrr_dynamic_multiplexed_queue<inode_policy>;
using nested_queue_type = inode_policy::queue_type;
struct fixture {
inode_policy policy;
queue_type queue{policy};
void fill(queue_type&) {
// nop
}
template <class T, class... Ts>
void fill(queue_type& q, T x, Ts... xs) {
q.emplace_back(x);
fill(q, xs...);
}
std::string fetch(int quantum) {
std::string result;
auto f = [&](int id, nested_queue_type& q, inode& x) {
CAF_CHECK_EQUAL(id, *q.policy().queue_id);
if (!result.empty())
result += ',';
result += to_string(id);
result += ':';
result += to_string(x);
};
queue.new_round(quantum, f);
return result;
}
void make_queues() {
for (int i = 0; i < 3; ++i)
queue.queues().emplace(i, nested_inode_policy{i});
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(wdrr_dynamic_multiplexed_queue_tests, fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(dropping) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12);
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(new_round) {
make_queues();
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_CHECK_EQUAL(fetch(1), "0:3,1:1,2:2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_CHECK_EQUAL(fetch(9), "0:6,0:9,0:12,1:4,1:7,2:5,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(priorities) {
make_queues();
queue.policy().enable_priorities = true;
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2 items from the high priority and 1 item otherwise.
CAF_CHECK_EQUAL(fetch(1), "0:3,0:6,1:1,2:2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain the high-priority queue with one item left per other queue.
CAF_CHECK_EQUAL(fetch(1), "0:9,1:4,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain queue.
CAF_CHECK_EQUAL(fetch(1000), "1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST_FIXTURE_SCOPE_END()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment