Commit 9af2a67d authored by Dominik Charousset's avatar Dominik Charousset

Rename {task_queue => linked_list} and simplify

Since we no longer need the `task_queue` as base class for the DRR
queues, we can simply rename it to the more appropriate name
`linked_list` and simplify the interface.
parent 2e0be448
...@@ -58,7 +58,6 @@ caf_add_component( ...@@ -58,7 +58,6 @@ caf_add_component(
exit_reason exit_reason
flow.op.state flow.op.state
intrusive.inbox_result intrusive.inbox_result
intrusive.task_result
invoke_message_result invoke_message_result
message_priority message_priority
pec pec
...@@ -159,6 +158,7 @@ caf_add_component( ...@@ -159,6 +158,7 @@ caf_add_component(
caf/hash/fnv.test.cpp caf/hash/fnv.test.cpp
caf/hash/sha1.cpp caf/hash/sha1.cpp
caf/init_global_meta_objects.cpp caf/init_global_meta_objects.cpp
caf/intrusive/linked_list.test.cpp
caf/intrusive/stack.test.cpp caf/intrusive/stack.test.cpp
caf/ipv4_address.cpp caf/ipv4_address.cpp
caf/ipv4_endpoint.cpp caf/ipv4_endpoint.cpp
...@@ -311,7 +311,6 @@ caf_add_component( ...@@ -311,7 +311,6 @@ caf_add_component(
handles handles
hash.sha1 hash.sha1
intrusive.lifo_inbox intrusive.lifo_inbox
intrusive.task_queue
intrusive_ptr intrusive_ptr
ipv4_address ipv4_address
ipv4_endpoint ipv4_endpoint
......
...@@ -93,17 +93,17 @@ bool default_mailbox::fetch_more() { ...@@ -93,17 +93,17 @@ bool default_mailbox::fetch_more() {
auto* head = static_cast<node_type*>(inbox_.take_head()); auto* head = static_cast<node_type*>(inbox_.take_head());
if (head == nullptr) if (head == nullptr)
return false; return false;
auto urgent_insertion_point = urgent_queue_.before_end();
auto normal_insertion_point = normal_queue_.before_end();
do { do {
auto next = head->next; auto next = head->next;
auto phead = promote(head); auto phead = promote(head);
if (phead->mid.is_urgent_message()) if (phead->mid.is_urgent_message())
urgent_queue_.lifo_append(phead); urgent_queue_.insert_after(urgent_insertion_point, phead);
else else
normal_queue_.lifo_append(phead); normal_queue_.insert_after(normal_insertion_point, phead);
head = next; head = next;
} while (head != nullptr); } while (head != nullptr);
urgent_queue_.stop_lifo_append();
normal_queue_.stop_lifo_append();
return true; return true;
} }
......
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
#include "caf/abstract_mailbox.hpp" #include "caf/abstract_mailbox.hpp"
#include "caf/intrusive/lifo_inbox.hpp" #include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/task_queue.hpp" #include "caf/intrusive/linked_list.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -55,7 +55,7 @@ public: ...@@ -55,7 +55,7 @@ public:
private: private:
/// Returns the total number of elements stored in the queues. /// Returns the total number of elements stored in the queues.
size_t cached() const noexcept { size_t cached() const noexcept {
return urgent_queue_.total_task_size() + normal_queue_.total_task_size(); return urgent_queue_.size() + normal_queue_.size();
} }
/// Tries to fetch more messages from the LIFO inbox. /// Tries to fetch more messages from the LIFO inbox.
...@@ -65,10 +65,10 @@ private: ...@@ -65,10 +65,10 @@ private:
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_; alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_;
/// Stores urgent messages in FIFO order. /// Stores urgent messages in FIFO order.
intrusive::task_queue<policy> urgent_queue_; intrusive::linked_list<mailbox_element> urgent_queue_;
/// Stores normal messages in FIFO order. /// Stores normal messages in FIFO order.
intrusive::task_queue<policy> normal_queue_; intrusive::linked_list<mailbox_element> normal_queue_;
}; };
} // namespace caf::detail } // namespace caf::detail
...@@ -15,10 +15,6 @@ ...@@ -15,10 +15,6 @@
namespace caf::detail { namespace caf::detail {
sync_request_bouncer::sync_request_bouncer(error r) : rsn(std::move(r)) {
// nop
}
void sync_request_bouncer::operator()(const strong_actor_ptr& sender, void sync_request_bouncer::operator()(const strong_actor_ptr& sender,
const message_id& mid) const { const message_id& mid) const {
if (sender && mid.is_request()) if (sender && mid.is_request())
...@@ -27,10 +23,8 @@ void sync_request_bouncer::operator()(const strong_actor_ptr& sender, ...@@ -27,10 +23,8 @@ void sync_request_bouncer::operator()(const strong_actor_ptr& sender,
nullptr); nullptr);
} }
intrusive::task_result void sync_request_bouncer::operator()(const mailbox_element& e) const {
sync_request_bouncer::operator()(const mailbox_element& e) const {
(*this)(e.sender, e.mid); (*this)(e.sender, e.mid);
return intrusive::task_result::resume;
} }
} // namespace caf::detail } // namespace caf::detail
...@@ -7,28 +7,26 @@ ...@@ -7,28 +7,26 @@
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/task_result.hpp"
#include <cstdint>
namespace caf::detail { namespace caf::detail {
/// Drains a mailbox and sends an error message to each unhandled request. /// Consumes mailbox elements and sends an error message for each request.
struct CAF_CORE_EXPORT sync_request_bouncer { struct CAF_CORE_EXPORT sync_request_bouncer {
error rsn; // -- constructors, destructors, and assignment operators --------------------
explicit sync_request_bouncer(error r);
explicit sync_request_bouncer(error r) : rsn(std::move(r)) {
// nop
}
// -- apply ------------------------------------------------------------------
void operator()(const strong_actor_ptr& sender, const message_id& mid) const; void operator()(const strong_actor_ptr& sender, const message_id& mid) const;
intrusive::task_result operator()(const mailbox_element& e) const; void operator()(const mailbox_element& e) const;
/// Unwrap WDRR queues. Nesting WDRR queues results in a Key/Queue prefix for // -- member variables -------------------------------------------------------
/// each layer of nesting.
template <class Key, class Queue, class... Ts> error rsn;
intrusive::task_result
operator()(const Key&, const Queue&, const Ts&... xs) const {
(*this)(xs...);
return intrusive::task_result::resume;
}
}; };
} // namespace caf::detail } // namespace caf::detail
...@@ -436,19 +436,6 @@ public: ...@@ -436,19 +436,6 @@ public:
static constexpr bool value = false; static constexpr bool value = false;
}; };
template <class T>
class has_peek_all {
private:
template <class U>
static int
fun(const U*, decltype(std::declval<U&>().peek_all(unit))* = nullptr);
static char fun(const void*);
public:
static constexpr bool value = sizeof(fun(static_cast<T*>(nullptr))) > 1;
};
CAF_HAS_MEMBER_TRAIT(clear); CAF_HAS_MEMBER_TRAIT(clear);
CAF_HAS_MEMBER_TRAIT(data); CAF_HAS_MEMBER_TRAIT(data);
CAF_HAS_MEMBER_TRAIT(make_behavior); CAF_HAS_MEMBER_TRAIT(make_behavior);
......
...@@ -215,14 +215,6 @@ class fnv; ...@@ -215,14 +215,6 @@ class fnv;
} // namespace hash } // namespace hash
// -- intrusive containers -----------------------------------------------------
namespace intrusive {
enum class task_result;
} // namespace intrusive
// -- marker classes for mixins ------------------------------------------------ // -- marker classes for mixins ------------------------------------------------
namespace mixin { namespace mixin {
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/intrusive/linked_list.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include <memory>
#include <numeric>
using namespace caf;
namespace {
struct inode : intrusive::singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
[[maybe_unused]] std::string to_string(const inode& x) {
return std::to_string(x.value);
}
using list_type = intrusive::linked_list<inode>;
struct fixture {
list_type uut;
void fill(list_type&) {
// nop
}
template <class T, class... Ts>
void fill(list_type& q, T x, Ts... xs) {
q.emplace_back(x);
fill(q, xs...);
}
};
} // namespace
WITH_FIXTURE(fixture) {
TEST("a default default-constructed uut is empty") {
check_eq(uut.empty(), true);
check_eq(uut.size(), 0u);
check_eq(uut.peek(), nullptr);
check_eq(uut.begin(), uut.end());
}
TEST("push_back adds elements to the back of the uut") {
uut.emplace_back(1);
uut.push_back(std::make_unique<inode>(2));
uut.push_back(new inode(3));
check_eq(deep_to_string(uut), "[1, 2, 3]");
}
TEST("push_front adds elements to the front of the uut") {
uut.emplace_front(1);
uut.push_front(std::make_unique<inode>(2));
uut.push_front(new inode(3));
check_eq(deep_to_string(uut), "[3, 2, 1]");
}
TEST("insert_after inserts elements after a given position") {
uut.insert_after(uut.before_end(), new inode(1));
uut.insert_after(uut.before_end(), new inode(3));
uut.insert_after(uut.begin(), new inode(2));
uut.insert_after(uut.before_begin(), new inode(0));
check_eq(deep_to_string(uut), "[0, 1, 2, 3]");
}
TEST("uuts are movable") {
SECTION("move constructor") {
fill(uut, 1, 2, 3);
list_type q2 = std::move(uut);
check_eq(uut.empty(), true);
check_eq(q2.empty(), false);
check_eq(deep_to_string(q2), "[1, 2, 3]");
}
SECTION("move assignment operator") {
list_type q2;
fill(q2, 1, 2, 3);
uut = std::move(q2);
check_eq(q2.empty(), true);
check_eq(uut.empty(), false);
check_eq(deep_to_string(uut), "[1, 2, 3]");
}
}
TEST("peek returns a pointer to the first element without removing it") {
check_eq(uut.peek(), nullptr);
fill(uut, 1, 2, 3);
check_eq(uut.peek()->value, 1);
}
TEST("the size of the uut is the number of elements") {
fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u);
fill(uut, 4, 5);
check_eq(uut.size(), 5u);
}
TEST("uuts are convertible to strings") {
check_eq(deep_to_string(uut), "[]");
fill(uut, 1, 2, 3, 4);
check_eq(deep_to_string(uut), "[1, 2, 3, 4]");
}
TEST("calling clear removes all elements from a uut") {
fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u);
uut.clear();
check_eq(uut.size(), 0u);
}
TEST("find_if selects an element from the uut") {
fill(uut, 1, 2, 3);
SECTION("find_if returns a pointer to the first matching element") {
auto ptr = uut.find_if([](const inode& x) { return x.value == 2; });
if (check(ptr != nullptr))
check_eq(ptr->value, 2);
}
SECTION("find_if returns nullptr if nothing matches") {
auto ptr = uut.find_if([](const inode& x) { return x.value == 4; });
check(ptr == nullptr);
}
}
TEST("uut allow iterator-based access") {
fill(uut, 1, 2, 3);
// Mutable access via begin/end.
for (auto& x : uut)
x.value *= 2;
check_eq(uut.front()->value, 2);
check_eq(uut.back()->value, 6);
// Immutable access via cbegin/cend.
check_eq(std::accumulate(uut.cbegin(), uut.cend(), 0,
[](int acc, const inode& x) {
return acc + x.value;
}),
12);
}
} // WITH_FIXTURE(fixture)
CAF_TEST_MAIN()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include <string>
#include <type_traits>
namespace caf::intrusive {
/// Communicates the state of a consumer to a task queue.
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,
/// The consumer processed the task but does not accept further tasks and no
/// subsequent queue shall start a new round.
stop_all,
};
std::string to_string(task_result);
} // namespace caf::intrusive
...@@ -13,7 +13,6 @@ ...@@ -13,7 +13,6 @@
#include "caf/detail/meta_object.hpp" #include "caf/detail/meta_object.hpp"
#include "caf/detail/pretty_type_name.hpp" #include "caf/detail/pretty_type_name.hpp"
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
......
...@@ -286,21 +286,21 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx, ...@@ -286,21 +286,21 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
continue; // Interrupted by a new message, try again. continue; // Interrupted by a new message, try again.
} }
auto res = run_with_metrics(*ptr, [this, &ptr, &consumed] { auto res = run_with_metrics(*ptr, [this, &ptr, &consumed] {
switch (reactivate(*ptr)) { auto res = reactivate(*ptr);
switch (res) {
case activation_result::success: case activation_result::success:
++consumed; ++consumed;
unstash(); unstash();
return intrusive::task_result::resume; break;
case activation_result::terminated:
return intrusive::task_result::stop;
case activation_result::skipped: case activation_result::skipped:
stash_.push(ptr.release()); stash_.push(ptr.release());
return intrusive::task_result::skip; break;
default: // drop default: // drop
return intrusive::task_result::resume; break;
} }
return res;
}); });
if (res == intrusive::task_result::stop) if (res == activation_result::terminated)
return resumable::done; return resumable::done;
} }
reset_timeouts_if_needed(); reset_timeouts_if_needed();
......
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/stack.hpp" #include "caf/intrusive/stack.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -729,12 +728,12 @@ private: ...@@ -729,12 +728,12 @@ private:
void unstash(); void unstash();
template <class F> template <class F>
intrusive::task_result run_with_metrics(mailbox_element& x, F body) { activation_result run_with_metrics(mailbox_element& x, F body) {
if (metrics_.mailbox_time) { if (metrics_.mailbox_time) {
auto t0 = std::chrono::steady_clock::now(); auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0); auto mbox_time = x.seconds_until(t0);
auto res = body(); auto res = body();
if (res != intrusive::task_result::skip) { if (res != activation_result::skipped) {
telemetry::timer::observe(metrics_.processing_time, t0); telemetry::timer::observe(metrics_.processing_time, t0);
metrics_.mailbox_time->observe(mbox_time); metrics_.mailbox_time->observe(mbox_time);
metrics_.mailbox_size->dec(); metrics_.mailbox_size->dec();
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.task_queue
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
[[maybe_unused]] std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct inode_policy {
using mapped_type = inode;
using task_size_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& x) {
return x.value;
}
};
using queue_type = task_queue<inode_policy>;
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...);
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
}
CAF_TEST(push_back) {
queue.emplace_back(1);
queue.push_back(inode_policy::unique_pointer{new inode(2)});
queue.push_back(new inode(3));
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(lifo_conversion) {
queue.lifo_append(new inode(3));
queue.lifo_append(new inode(2));
queue.lifo_append(new inode(1));
queue.stop_lifo_append();
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(move_construct) {
fill(queue, 1, 2, 3);
queue_type q2 = std::move(queue);
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(q2.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(q2), "[1, 2, 3]");
}
CAF_TEST(move_assign) {
queue_type q2{policy};
fill(q2, 1, 2, 3);
queue = std::move(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(append) {
queue_type q2{policy};
fill(queue, 1, 2, 3);
fill(q2, 4, 5, 6);
queue.append(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3, 4, 5, 6]");
}
CAF_TEST(prepend) {
queue_type q2{policy};
fill(queue, 1, 2, 3);
fill(q2, 4, 5, 6);
queue.prepend(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[4, 5, 6, 1, 2, 3]");
}
CAF_TEST(peek) {
CHECK_EQ(queue.peek(), nullptr);
fill(queue, 1, 2, 3);
CHECK_EQ(queue.peek()->value, 1);
}
CAF_TEST(task_size) {
fill(queue, 1, 2, 3);
CHECK_EQ(queue.total_task_size(), 6);
fill(queue, 4, 5);
CHECK_EQ(queue.total_task_size(), 15);
queue.clear();
CHECK_EQ(queue.total_task_size(), 0);
}
CAF_TEST(to_string) {
CHECK_EQ(deep_to_string(queue), "[]");
fill(queue, 1, 2, 3, 4);
CHECK_EQ(deep_to_string(queue), "[1, 2, 3, 4]");
}
END_FIXTURE_SCOPE()
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