Commit d6f6bfa3 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Allow task queues to drop items

parent 036fc630
...@@ -129,25 +129,25 @@ public: ...@@ -129,25 +129,25 @@ public:
/// Appends `ptr` to the queue. /// Appends `ptr` to the queue.
/// @pre `ptr != nullptr` /// @pre `ptr != nullptr`
void push_back(pointer ptr) noexcept { bool push_back(pointer ptr) noexcept {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
tail_.next->next = ptr; tail_.next->next = ptr;
tail_.next = ptr; tail_.next = ptr;
ptr->next = &tail_; ptr->next = &tail_;
inc_total_task_size(*ptr); inc_total_task_size(*ptr);
return true;
} }
/// Appends `ptr` to the queue. /// Appends `ptr` to the queue.
/// @pre `ptr != nullptr` /// @pre `ptr != nullptr`
void push_back(unique_pointer ptr) noexcept { bool push_back(unique_pointer ptr) noexcept {
CAF_ASSERT(ptr != nullptr); return push_back(ptr.release());
push_back(ptr.release());
} }
/// Creates a new element from `xs...` and appends it. /// Creates a new element from `xs...` and appends it.
template <class... Ts> template <class... Ts>
void emplace_back(Ts&&... xs) { bool emplace_back(Ts&&... xs) {
push_back(new value_type(std::forward<Ts>(xs)...)); return push_back(new value_type(std::forward<Ts>(xs)...));
} }
/// @private /// @private
......
...@@ -64,42 +64,63 @@ public: ...@@ -64,42 +64,63 @@ public:
return policy_; return policy_;
} }
void push_back(mapped_type* ptr) noexcept { bool push_back(mapped_type* ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr)); auto i = qs_.find(policy_.id_of(*ptr));
if (i != qs_.end()) { if (i != qs_.end()) {
i->second.push_back(ptr); i->second.push_back(ptr);
return true;
} else { } else {
deleter_type d; deleter_type d;
d(ptr); d(ptr);
return false;
} }
} }
void push_back(unique_pointer ptr) noexcept { bool push_back(unique_pointer ptr) noexcept {
push_back(ptr.release()); return push_back(ptr.release());
} }
template <class... Ts> template <class... Ts>
void emplace_back(Ts&&... xs) { bool emplace_back(Ts&&... xs) {
push_back(new mapped_type(std::forward<Ts>(xs)...)); return push_back(new mapped_type(std::forward<Ts>(xs)...));
} }
/// @private
template <class F>
struct new_round_helper {
const key_type& k;
queue_type& q;
F& f;
template <class... Ts>
auto operator()(Ts&&... xs)
-> decltype((std::declval<F&>())(std::declval<const key_type&>(),
std::declval<queue_type&>(),
std::forward<Ts>(xs)...)) {
return f(k, q, std::forward<Ts>(xs)...);
}
};
/// Run a new round with `quantum`, dispatching all tasks to `consumer`. /// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise. /// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F> template <class F>
bool new_round(long quantum, bool new_round(long quantum, F& f) {
F& f) noexcept(noexcept(f(std::declval<const key_type&>(),
std::declval<queue_type&>(),
std::declval<mapped_type&>()))) {
bool result = false; bool result = false;
for (auto& kvp : qs_) { for (auto& kvp : qs_) {
auto g = [&](mapped_type& x) { new_round_helper<F> g{kvp.first, kvp.second, f};
f(kvp.first, kvp.second, x); result |= g.q.new_round(policy_.quantum(g.q, quantum), g);
}; }
result |= kvp.second.new_round(policy_.quantum(kvp.second, quantum), g); if (!erase_list_.empty()) {
for (auto& k : erase_list_)
qs_.erase(k);
erase_list_.clear();
} }
return result; return result;
} }
void erase_later(key_type k) {
erase_list_.emplace_back(std::move(k));
}
pointer peek() noexcept { pointer peek() noexcept {
for (auto& kvp : qs_) { for (auto& kvp : qs_) {
auto ptr = kvp.second.peek(); auto ptr = kvp.second.peek();
...@@ -159,6 +180,9 @@ private: ...@@ -159,6 +180,9 @@ private:
/// Policy object for adjusting a quantum based on queue weight etc. /// Policy object for adjusting a quantum based on queue weight etc.
Policy policy_; Policy policy_;
/// List of keys that are marked erased.
std::vector<key_type> erase_list_;
}; };
} // namespace intrusive } // namespace intrusive
......
...@@ -68,17 +68,17 @@ public: ...@@ -68,17 +68,17 @@ public:
return policy_; return policy_;
} }
void push_back(mapped_type* ptr) noexcept { bool push_back(mapped_type* ptr) noexcept {
push_back_recursion<0>(policy_.id_of(*ptr), ptr); return push_back_recursion<0>(policy_.id_of(*ptr), ptr);
} }
void push_back(unique_pointer ptr) noexcept { bool push_back(unique_pointer ptr) noexcept {
push_back(ptr.release()); return push_back(ptr.release());
} }
template <class... Ts> template <class... Ts>
void emplace_back(Ts&&... xs) { bool emplace_back(Ts&&... xs) {
push_back(new mapped_type(std::forward<Ts>(xs)...)); return push_back(new mapped_type(std::forward<Ts>(xs)...));
} }
/// Run a new round with `quantum`, dispatching all tasks to `consumer`. /// Run a new round with `quantum`, dispatching all tasks to `consumer`.
...@@ -127,21 +127,34 @@ private: ...@@ -127,21 +127,34 @@ private:
// -- recursive helper functions --------------------------------------------- // -- recursive helper functions ---------------------------------------------
template <size_t I> template <size_t I>
detail::enable_if_t<I == num_queues> detail::enable_if_t<I == num_queues, bool>
push_back_recursion(size_t, mapped_type*) noexcept { push_back_recursion(size_t, mapped_type*) noexcept {
// nop return false;
} }
template <size_t I> template <size_t I>
detail::enable_if_t<I != num_queues> detail::enable_if_t<I != num_queues, bool>
push_back_recursion(size_t pos, mapped_type* ptr) noexcept { push_back_recursion(size_t pos, mapped_type* ptr) noexcept {
if (pos == I) { if (pos == I) {
auto& q = std::get<I>(qs_); auto& q = std::get<I>(qs_);
q.push_back(ptr); return q.push_back(ptr);
} else {
push_back_recursion<I + 1>(pos, ptr);
} }
} return push_back_recursion<I + 1>(pos, ptr);
}
template <size_t I, class Queue, class F>
struct new_round_recursion_helper {
Queue& q;
F& f;
template <class... Ts>
auto operator()(Ts&&... xs)
-> decltype((std::declval<F&>())(std::declval<index<I>>(),
std::declval<Queue&>(),
std::forward<Ts>(xs)...)) {
index<I> id;
return f(id, q, std::forward<Ts>(xs)...);
}
};
template <size_t I, class F> template <size_t I, class F>
detail::enable_if_t<I == num_queues, int> detail::enable_if_t<I == num_queues, int>
...@@ -153,10 +166,8 @@ private: ...@@ -153,10 +166,8 @@ private:
detail::enable_if_t<I != num_queues, int> detail::enable_if_t<I != num_queues, int>
new_round_recursion(deficit_type quantum, F& f) { new_round_recursion(deficit_type quantum, F& f) {
auto& q = std::get<I>(qs_); auto& q = std::get<I>(qs_);
auto g = [&](mapped_type& x) { using q_type = typename std::decay<decltype(q)>::type;
index<I> id; new_round_recursion_helper<I, q_type, F> g{q, f};
return f(id, q, x);
};
if (q.new_round(policy_.quantum(q, quantum), g)) if (q.new_round(policy_.quantum(q, quantum), g))
return 1 + new_round_recursion<I + 1>(quantum, f); return 1 + new_round_recursion<I + 1>(quantum, f);
return 0 + new_round_recursion<I + 1>(quantum, f); return 0 + new_round_recursion<I + 1>(quantum, f);
......
...@@ -85,10 +85,6 @@ struct inode_policy { ...@@ -85,10 +85,6 @@ struct inode_policy {
using queue_map_type = std::map<key_type, queue_type>; 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) { static inline key_type id_of(const inode& x) {
return x.value % 3; return x.value % 3;
} }
...@@ -108,14 +104,13 @@ struct fixture { ...@@ -108,14 +104,13 @@ struct fixture {
inode_policy policy; inode_policy policy;
queue_type queue{policy}; queue_type queue{policy};
void fill(queue_type&) { int fill(queue_type&) {
// nop return 0;
} }
template <class T, class... Ts> template <class T, class... Ts>
void fill(queue_type& q, T x, Ts... xs) { int fill(queue_type& q, T x, Ts... xs) {
q.emplace_back(x); return (q.emplace_back(x) ? 1 : 0) + fill(q, xs...);
fill(q, xs...);
} }
std::string fetch(int quantum) { std::string fetch(int quantum) {
...@@ -148,7 +143,7 @@ CAF_TEST(default_constructed) { ...@@ -148,7 +143,7 @@ CAF_TEST(default_constructed) {
CAF_TEST(dropping) { CAF_TEST(dropping) {
CAF_REQUIRE_EQUAL(queue.empty(), true); CAF_REQUIRE_EQUAL(queue.empty(), true);
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12); CAF_REQUIRE_EQUAL(fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12), 0);
CAF_REQUIRE_EQUAL(queue.empty(), true); CAF_REQUIRE_EQUAL(queue.empty(), true);
} }
......
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