Commit 188be50c authored by Dominik Charousset's avatar Dominik Charousset

Add intrusive::task_result::stop_all

The new enum value allows consumers to stop a queue from calling
`new_round` on succeeding queues.
parent 6c187e09
......@@ -182,10 +182,18 @@ public:
/// @returns `true` if `f` consumed at least one item.
template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
return new_round(0, f).consumed_items;
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer)
noexcept(noexcept(consumer(std::declval<value_type&>()))) {
if (!cache_.empty())
list_.prepend(cache_);
if (list_.empty())
return false;
return {false, false};
deficit_ += quantum;
long consumed = 0;
auto ptr = list_.front();
auto ts = policy().task_size(*ptr);
......@@ -199,22 +207,16 @@ public:
if (next == list_.end().ptr)
list_.end()->next = list_.before_begin().ptr;
CAF_ASSERT(total_task_size() != 0 || list_.begin() == list_.end());
/*
if (total_task_size() == 0) {
CAF_ASSERT(list_.begin() == list_.end());
list_.end()->next = list_.begin().ptr;
}
*/
// Always decrease the deficit_ counter, again because `f` is allowed
// to call consume again.
deficit_ -= ts;
auto res = f(*ptr);
auto res = consumer(*ptr);
if (res == task_result::skip) {
// Push the unconsumed item to the cache.
cache_.push_back(ptr);
if (list_.empty()) {
deficit_ = 0;
return consumed != 0;
return {consumed != 0, false};
}
// Fix deficit counter since we didn't actually use it.
deficit_ += ts;
......@@ -226,27 +228,16 @@ public:
list_.prepend(cache_);
if (list_.empty()) {
deficit_ = 0;
return consumed != 0;
return {consumed != 0, res == task_result::stop_all};
}
if (res == task_result::stop)
return consumed != 0;
if (res != task_result::resume)
return {consumed != 0, res == task_result::stop_all};
}
// Next iteration.
ptr = list_.front();
ts = policy().task_size(*ptr);
}
return consumed != 0;
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
bool new_round(deficit_type quantum, F& consumer)
noexcept(noexcept(consumer(std::declval<value_type&>()))) {
if (!list_.empty()) {
deficit_ += quantum;
return consume(consumer);
}
return false;
return {consumed != 0, false};
}
cache_type& cache() noexcept {
......
......@@ -22,7 +22,9 @@
#include <utility>
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_result.hpp"
namespace caf {
namespace intrusive {
......@@ -110,25 +112,33 @@ public:
/// @returns `true` if `f` consumed at least one item.
template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
auto ptr = take_front();
if (ptr == nullptr)
return false;
do {
f(*ptr);
ptr = take_front();
} while (ptr != nullptr);
return true;
auto res = new_round(0, f);
return res.consumed_items;
}
/// 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(deficit_type quantum, F& consumer) {
new_round_result new_round(deficit_type quantum, F& consumer) {
if (!super::empty()) {
deficit_ += quantum;
return consume(consumer);
auto ptr = take_front();
if (ptr == nullptr)
return {false, false};
do {
switch (consumer(*ptr)) {
default:
break;
case task_result::stop:
return {true, false};
case task_result::stop_all:
return {true, true};
}
ptr = take_front();
} while (ptr != nullptr);
return {true, false};
}
return false;
return {false, false};
}
private:
......
......@@ -34,6 +34,7 @@
#include "caf/intrusive/inbox_result.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/new_round_result.hpp"
namespace caf {
namespace intrusive {
......@@ -152,7 +153,7 @@ public:
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
bool new_round(deficit_type quantum, F& consumer) {
new_round_result new_round(deficit_type quantum, F& consumer) {
fetch_more();
return queue_.new_round(quantum, consumer);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_NEW_ROUND_RESULT_HPP
#define CAF_INTRUSIVE_NEW_ROUND_RESULT_HPP
#include <type_traits>
#include "caf/fwd.hpp"
namespace caf {
namespace intrusive {
/// Returns the state of a consumer from `new_round`.
struct new_round_result {
/// Denotes whether the consumer accepted at least one element.
bool consumed_items : 1;
/// Denotes whether the consumer returned `task_result::stop_all`.
bool stop_all : 1;
};
constexpr bool operator==(new_round_result x, new_round_result y) {
return x.consumed_items == y.consumed_items && x.stop_all == y.stop_all;
}
constexpr bool operator!=(new_round_result x, new_round_result y) {
return !(x == y);
}
constexpr new_round_result make_new_round_result(bool consumed_items,
bool stop_all = false) {
return {consumed_items, stop_all};
}
constexpr new_round_result operator|(new_round_result x, new_round_result y) {
return {x.consumed_items || y.consumed_items, x.stop_all || y.stop_all};
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_NEW_ROUND_RESULT_HPP
......@@ -32,9 +32,14 @@ enum class task_result {
/// The consumer processed the task and is ready to receive the next one.
resume,
/// The consumer skipped the task and is ready to receive the next one.
/// Illegal for consumers of non-cached queues (non-cached queues treat
/// `skip` and `resume` in the same way).
skip,
/// The consumer processed the task but does not accept further tasks.
stop
stop,
/// The consumer processed the task but does not accept further tasks and no
/// subsequent queue shall start a new round.
stop_all,
};
} // namespace intrusive
......
......@@ -22,6 +22,9 @@
#include <utility>
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
......@@ -98,15 +101,33 @@ public:
}
};
void inc_deficit(deficit_type x) noexcept {
for (auto& kvp : qs_) {
auto& q = kvp.second;
q.inc_deficit(policy_.quantum(q, x));
}
}
/// 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) {
new_round_result new_round(deficit_type quantum, F& f) {
bool result = false;
bool stopped = false;
for (auto& kvp : qs_) {
if (policy_.enabled(kvp.second)) {
new_round_helper<F> g{kvp.first, kvp.second, f};
result |= g.q.new_round(policy_.quantum(g.q, quantum), g);
auto& q = kvp.second;
if (!stopped) {
new_round_helper<F> g{kvp.first, q, f};
auto res = q.new_round(policy_.quantum(q, quantum), g);
result = res.consumed_items;
if (res.stop_all)
stopped = true;
} else {
// Always increase deficit, even if a previous queue stopped the
// consumer preemptively.
q.inc_deficit(policy_.quantum(q, quantum));
}
}
}
if (!erase_list_.empty()) {
......@@ -114,7 +135,7 @@ public:
qs_.erase(k);
erase_list_.clear();
}
return result;
return {result, stopped};
}
void erase_later(key_type k) {
......
......@@ -24,6 +24,9 @@
#include <type_traits>
#include <utility>
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
......@@ -81,11 +84,15 @@ public:
return push_back(new value_type(std::forward<Ts>(xs)...));
}
void inc_deficit(deficit_type x) noexcept {
inc_deficit_recursion<0>(x);
}
/// 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) {
return new_round_recursion<0>(quantum, f) != 0;
new_round_result new_round(deficit_type quantum, F& f) {
return new_round_recursion<0>(quantum, f);
}
pointer peek() noexcept {
......@@ -166,21 +173,40 @@ private:
}
};
template <size_t I>
detail::enable_if_t<I == num_queues>
inc_deficit_recursion(deficit_type) noexcept {
// end of recursion
}
template <size_t I>
detail::enable_if_t<I != num_queues>
inc_deficit_recursion(deficit_type quantum) noexcept {
auto& q = std::get<I>(qs_);
q.inc_deficit(policy_.quantum(q, quantum));
inc_deficit_recursion<I + 1>(quantum);
}
template <size_t I, class F>
detail::enable_if_t<I == num_queues, int>
detail::enable_if_t<I == num_queues, new_round_result>
new_round_recursion(deficit_type, F&) noexcept {
return 0;
return {false, false};
}
template <size_t I, class F>
detail::enable_if_t<I != num_queues, int>
detail::enable_if_t<I != num_queues, new_round_result>
new_round_recursion(deficit_type quantum, F& f) {
auto& q = std::get<I>(qs_);
using q_type = typename std::decay<decltype(q)>::type;
new_round_recursion_helper<I, q_type, F> g{q, f};
if (q.new_round(policy_.quantum(q, quantum), g))
return 1 + new_round_recursion<I + 1>(quantum, f);
return 0 + new_round_recursion<I + 1>(quantum, f);
auto res = q.new_round(policy_.quantum(q, quantum), g);
if (res.stop_all) {
// Always increase deficit, even if a previous queue stopped the
// consumer preemptively.
inc_deficit_recursion<I + 1>(quantum);
return res;
}
return res | new_round_recursion<I + 1>(quantum, f);
}
template <size_t I>
......
......@@ -309,7 +309,7 @@ bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce))
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
}
// Dispatch to parent's `cleanup` function.
......
......@@ -208,7 +208,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce))
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
}
// Dispatch to parent's `cleanup` function.
......@@ -252,7 +252,7 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
case activation_result::success:
return ++handled_msgs < max_throughput
? intrusive::task_result::resume
: intrusive::task_result::stop;
: intrusive::task_result::stop_all;
case activation_result::skipped:
return intrusive::task_result::skip;
default:
......@@ -274,7 +274,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
mailbox_visitor f{this, result, handled_msgs, max_throughput};
mailbox_element_ptr ptr;
while (handled_msgs < max_throughput) {
if (!mailbox_.new_round(3, f)) {
if (!mailbox_.new_round(3, f).consumed_items) {
reset_timeout_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
......
......@@ -107,12 +107,12 @@ CAF_TEST(new_round) {
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2, 4, and 6.
auto round_result = queue.new_round(3, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(fseq, "246");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow g to consume 1, 3, 5, and 7.
round_result = queue.new_round(4, g);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(gseq, "1357");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
......@@ -127,21 +127,21 @@ CAF_TEST(skipping) {
return task_result::resume;
};
CAF_MESSAGE("make a round on an empty queue");
CAF_CHECK_EQUAL(queue.new_round(10, f), false);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false));
CAF_MESSAGE("make a round on a queue with only odd numbers (skip all)");
fill(queue, 1, 3, 5);
CAF_CHECK_EQUAL(queue.new_round(10, f), false);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false));
CAF_MESSAGE("make a round on a queue with an even number at the front");
fill(queue, 2);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "2");
CAF_MESSAGE("make a round on a queue with an even number in between");
fill(queue, 7, 9, 4, 11, 13);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "24");
CAF_MESSAGE("make a round on a queue with an even number at the back");
fill(queue, 15, 17, 6);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "246");
}
......@@ -175,7 +175,7 @@ CAF_TEST(alternating_consumer) {
// sequences and no odd value to read after 7 is available.
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
auto round_result = queue.new_round(1000, h);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "21436587");
CAF_CHECK_EQUAL(queue.deficit(), 0);
CAF_CHECK_EQUAL(deep_to_string(queue.cache()), "[9]");
......
......@@ -107,25 +107,26 @@ CAF_TEST(new_round) {
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
// Allow f to consume 1, 2, and 3 with a leftover deficit of 1.
auto round_result = queue.new_round(7, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "123");
CAF_CHECK_EQUAL(queue.deficit(), 1);
// Allow f to consume 4 and 5 with a leftover deficit of 0.
round_result = queue.new_round(8, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "12345");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow f to consume 6 with a leftover deficit of 0 (queue is empty).
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// new_round on an empty queue does nothing.
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, false);
CAF_CHECK_EQUAL(round_result, make_new_round_result(false));
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
......
......@@ -81,6 +81,7 @@ struct fixture {
std::string result;
auto f = [&](inode& x) {
result += to_string(x);
return task_result::resume;
};
inbox.new_round(1000, f);
return result;
......@@ -90,6 +91,7 @@ struct fixture {
std::string result;
auto f = [&](inode& x) {
result += to_string(x);
return task_result::resume;
};
inbox.close();
inbox.queue().new_round(1000, f);
......
......@@ -99,7 +99,7 @@ struct print_with_comma_t {
}
template <class T>
std::ostream& operator()(std::ostream& out, const logger::arg_wrapper<T>& x) {
std::ostream& operator()(std::ostream& out, const detail::arg_wrapper<T>& x) {
if (!first)
out << ", ";
else
......
......@@ -126,6 +126,7 @@ struct fixture {
result += to_string(id);
result += ':';
result += to_string(x);
return task_result::resume;
};
queue.new_round(quantum, f);
return result;
......
......@@ -94,12 +94,14 @@ struct fetch_helper {
std::string result;
template <size_t I, class Queue>
void operator()(std::integral_constant<size_t, I>, const Queue&, inode& x) {
task_result operator()(std::integral_constant<size_t, I>, const Queue&,
inode& x) {
if (!result.empty())
result += ',';
result += to_string(I);
result += ':';
result += to_string(x);
return task_result::resume;
};
};
......@@ -125,6 +127,7 @@ struct fixture {
result += to_string(id);
result += ':';
result += to_string(x);
return task_result::resume;
};
queue.new_round(quantum, f);
return result;
......@@ -144,19 +147,19 @@ CAF_TEST(new_round) {
// Allow f to consume 2 items per nested queue.
fetch_helper f;
auto round_result = queue.new_round(2, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(f.result, "0:3,0:6,1:1,1:4,2:2,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume one more item from each queue.
f.result.clear();
round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(f.result, "0:9,1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume the remainder, i.e., 12.
f.result.clear();
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(f.result, "0:12");
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