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

Simplify lifo_inbox and streamline tests

parent eaa811ef
......@@ -160,6 +160,7 @@ caf_add_component(
caf/hash/fnv.test.cpp
caf/hash/sha1.cpp
caf/init_global_meta_objects.cpp
caf/intrusive/lifo_inbox.test.cpp
caf/intrusive/linked_list.test.cpp
caf/intrusive/stack.test.cpp
caf/ipv4_address.cpp
......@@ -312,7 +313,6 @@ caf_add_component(
function_view
handles
hash.sha1
intrusive.lifo_inbox
intrusive_ptr
ipv4_address
ipv4_endpoint
......
......@@ -19,20 +19,6 @@ namespace caf::detail {
/// messages.
class CAF_CORE_EXPORT default_mailbox : public abstract_mailbox {
public:
struct policy {
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
static size_t task_size(const mapped_type&) noexcept {
return 1;
}
};
default_mailbox() noexcept : ref_count_(1) {
// nop
}
......@@ -85,7 +71,7 @@ private:
intrusive::linked_list<mailbox_element> normal_queue_;
/// Stores incoming messages in LIFO order.
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_;
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<mailbox_element> inbox_;
/// The intrusive reference count.
alignas(CAF_CACHE_LINE_SIZE) std::atomic<size_t> ref_count_;
......
......@@ -9,20 +9,19 @@
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
namespace caf::intrusive {
/// An intrusive, thread-safe LIFO queue implementation for a single reader
/// with any number of writers.
template <class Policy>
template <class T>
class lifo_inbox {
public:
// -- member types -----------------------------------------------------------
using policy_type = Policy;
using value_type = typename policy_type::mapped_type;
using value_type = T;
using pointer = value_type*;
......@@ -30,7 +29,7 @@ public:
using node_pointer = node_type*;
using unique_pointer = typename policy_type::unique_pointer;
using unique_pointer = std::unique_ptr<value_type>;
using deleter_type = typename unique_pointer::deleter_type;
......@@ -146,21 +145,15 @@ public:
/// Closes this queue and deletes all remaining elements.
/// @warning Call only from the reader (owner).
void close() noexcept {
deleter_type d;
// We assume the node destructor to never throw. However, the following
// static assert fails. Unfortunately, std::default_delete's apply operator
// is not noexcept (event for types that have a noexcept destructor).
// static_assert(noexcept(d(std::declval<pointer>())),
// "deleter is not noexcept");
close(d);
void close() {
close(deleter_type{});
}
/// Closes this queue and applies `f` to each pointer. The function object
/// `f` must take ownership of the passed pointer.
/// @warning Call only from the reader (owner).
template <class F>
void close(F&& f) noexcept(noexcept(f(std::declval<pointer>()))) {
void close(F&& f) {
node_pointer ptr = take_head(stack_closed_tag());
while (ptr != nullptr) {
auto next = ptr->next;
......@@ -178,62 +171,6 @@ public:
close();
}
// -- synchronized access ---------------------------------------------------
template <class Mutex, class CondVar>
bool synchronized_push_front(Mutex& mtx, CondVar& cv, pointer ptr) {
switch (push_front(ptr)) {
default:
// enqueued message to a running actor's mailbox
return true;
case inbox_result::unblocked_reader: {
std::unique_lock<Mutex> guard(mtx);
cv.notify_one();
return true;
}
case inbox_result::queue_closed:
// actor no longer alive
return false;
}
}
template <class Mutex, class CondVar>
bool synchronized_push_front(Mutex& mtx, CondVar& cv, unique_pointer ptr) {
return synchronized_push_front(mtx, cv, ptr.release());
}
template <class Mutex, class CondVar, class... Ts>
bool synchronized_emplace_front(Mutex& mtx, CondVar& cv, Ts&&... xs) {
return synchronized_push_front(mtx, cv,
new value_type(std::forward<Ts>(xs)...));
}
template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) {
CAF_ASSERT(!closed());
if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked())
cv.wait(guard);
}
}
template <class Mutex, class CondVar, class TimePoint>
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
CAF_ASSERT(!closed());
if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) {
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) {
// if we're unable to set the queue from blocked to empty,
// than there's a new element in the list
return !try_unblock();
}
}
}
return true;
}
private:
static constexpr pointer stack_empty_tag() {
// We are *never* going to dereference the returned pointer. It is only
......
// 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/lifo_inbox.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include <memory>
#include <numeric>
using namespace caf;
using intrusive::inbox_result;
struct inode : intrusive::singly_linked<inode> {
explicit inode(int x = 0) : value(x) {
// nop
}
int value;
};
using inbox_type = intrusive::lifo_inbox<inode>;
std::string drain(inbox_type& xs) {
std::vector<int> tmp;
std::unique_ptr<inode> ptr{xs.take_head()};
while (ptr != nullptr) {
auto next = ptr->next;
tmp.push_back(ptr->value);
ptr.reset(inbox_type::promote(next));
}
return caf::deep_to_string(tmp);
}
TEST("a default default-constructed inbox is empty") {
inbox_type uut;
check(!uut.closed());
check(!uut.blocked());
check(uut.empty());
check_eq(uut.take_head(), nullptr);
}
TEST("push_front adds elements to the front of the inbox") {
inbox_type uut;
check_eq(uut.emplace_front(1), inbox_result::success);
check_eq(uut.emplace_front(2), inbox_result::success);
check_eq(uut.emplace_front(3), inbox_result::success);
check_eq(drain(uut), "[3, 2, 1]");
}
TEST("push_front discards elements if the inbox is closed") {
inbox_type uut;
uut.close();
check(uut.closed());
auto res = uut.emplace_front(0);
check_eq(res, inbox_result::queue_closed);
}
TEST("push_front unblocks a blocked reader") {
inbox_type uut;
check(uut.try_block());
check_eq(uut.emplace_front(1), inbox_result::unblocked_reader);
check_eq(uut.emplace_front(2), inbox_result::success);
check_eq(drain(uut), "[2, 1]");
}
CAF_TEST_MAIN()
......@@ -12,60 +12,50 @@
using namespace caf;
namespace {
struct inode : intrusive::singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
explicit inode(int x = 0) : value(x) {
// nop
}
int value;
};
[[maybe_unused]] std::string to_string(const inode& x) {
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) {
template <class... Ts>
void fill(list_type& xs, Ts... args) {
(xs.emplace_back(args), ...);
}
TEST("a default default-constructed uut is empty") {
TEST("a default default-constructed list is empty") {
list_type uut;
check_eq(uut.empty(), true);
check_eq(uut.size(), 0u);
check_eq(uut.peek(), nullptr);
check_eq(uut.begin(), uut.end());
}
TEST("uuts are convertible to strings") {
TEST("lists are convertible to strings") {
list_type uut;
check_eq(deep_to_string(uut), "[]");
fill(uut, 1, 2, 3, 4);
check_eq(deep_to_string(uut), "[1, 2, 3, 4]");
}
TEST("push_back adds elements to the back of the uut") {
TEST("push_back adds elements to the back of the list") {
list_type 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") {
TEST("push_front adds elements to the front of the list") {
list_type uut;
uut.emplace_front(1);
uut.push_front(std::make_unique<inode>(2));
uut.push_front(new inode(3));
......@@ -73,6 +63,7 @@ TEST("push_front adds elements to the front of the uut") {
}
TEST("insert_after inserts elements after a given position") {
list_type uut;
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));
......@@ -80,7 +71,8 @@ TEST("insert_after inserts elements after a given position") {
check_eq(deep_to_string(uut), "[0, 1, 2, 3]");
}
TEST("uuts are movable") {
TEST("lists are movable") {
list_type uut;
SECTION("move constructor") {
fill(uut, 1, 2, 3);
list_type q2 = std::move(uut);
......@@ -99,26 +91,30 @@ TEST("uuts are movable") {
}
TEST("peek returns a pointer to the first element without removing it") {
list_type uut;
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") {
TEST("the size of the list is the number of elements") {
list_type uut;
fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u);
fill(uut, 4, 5);
check_eq(uut.size(), 5u);
}
TEST("calling clear removes all elements from a uut") {
TEST("calling clear removes all elements from a list") {
list_type 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") {
TEST("find_if selects an element from the list") {
list_type 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; });
......@@ -131,7 +127,8 @@ TEST("find_if selects an element from the uut") {
}
}
TEST("uut allow iterator-based access") {
TEST("lists allow iterator-based access") {
list_type uut;
fill(uut, 1, 2, 3);
// Mutable access via begin/end.
for (auto& x : uut)
......@@ -146,6 +143,4 @@ TEST("uut allow iterator-based access") {
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.
#define CAF_SUITE intrusive.lifo_inbox
#include "caf/intrusive/lifo_inbox.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
}
};
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 deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
};
using inbox_type = lifo_inbox<inode_policy>;
struct fixture {
inode_policy policy;
inbox_type inbox;
void fill(inbox_type&) {
// nop
}
template <class T, class... Ts>
void fill(inbox_type& i, T x, Ts... xs) {
i.emplace_front(x);
fill(i, xs...);
}
std::string fetch() {
std::string result;
inode_policy::unique_pointer ptr{inbox.take_head()};
while (ptr != nullptr) {
auto next = ptr->next;
result += to_string(*ptr);
ptr.reset(inbox_type::promote(next));
}
return result;
}
std::string close_and_fetch() {
std::string result;
auto f = [&](inode* x) {
result += to_string(*x);
delete x;
};
inbox.close(f);
return result;
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(inbox.empty(), true);
}
CAF_TEST(push_front) {
fill(inbox, 1, 2, 3);
CAF_REQUIRE_EQUAL(close_and_fetch(), "321");
CAF_REQUIRE_EQUAL(inbox.closed(), true);
}
CAF_TEST(push_after_close) {
inbox.close();
auto res = inbox.push_front(new inode(0));
CAF_REQUIRE_EQUAL(res, inbox_result::queue_closed);
}
CAF_TEST(unblock) {
CAF_REQUIRE_EQUAL(inbox.try_block(), true);
auto res = inbox.push_front(new inode(1));
CAF_REQUIRE_EQUAL(res, inbox_result::unblocked_reader);
res = inbox.push_front(new inode(2));
CAF_REQUIRE_EQUAL(res, inbox_result::success);
CAF_REQUIRE_EQUAL(close_and_fetch(), "21");
}
CAF_TEST(await) {
std::mutex mx;
std::condition_variable cv;
std::thread t{[&] { inbox.synchronized_emplace_front(mx, cv, 1); }};
inbox.synchronized_await(mx, cv);
CAF_REQUIRE_EQUAL(close_and_fetch(), "1");
t.join();
}
CAF_TEST(timed_await) {
std::mutex mx;
std::condition_variable cv;
auto tout = std::chrono::system_clock::now();
tout += std::chrono::microseconds(1);
auto res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, false);
fill(inbox, 1);
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CHECK_EQ(fetch(), "1");
tout += std::chrono::hours(1000);
std::thread t{[&] { inbox.synchronized_emplace_front(mx, cv, 2); }};
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CAF_REQUIRE_EQUAL(close_and_fetch(), "2");
t.join();
}
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