Commit 7fec52d0 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Add test scaffold for new streaming infrastructure

parent d6f6bfa3
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DETAIL_OVERLOAD_HPP
#define CAF_DETAIL_OVERLOAD_HPP
namespace caf {
namespace detail {
template <class... Fs>
struct overload;
template <class F>
struct overload<F> : F {
using F::operator();
overload(F f) : F(f) {
// nop
}
};
template <class F, class... Fs>
struct overload<F, Fs...> : F, overload<Fs...> {
using F::operator();
using overload<Fs...>::operator();
overload(F f, Fs... fs) : F(f), overload<Fs...>(fs...) {
// nop
}
};
template <class... Fs>
overload<Fs...> make_overload(Fs... fs) {
return {fs...};
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_OVERLOAD_HPP
...@@ -20,11 +20,50 @@ ...@@ -20,11 +20,50 @@
#define CAF_STREAM_SLOT_HPP #define CAF_STREAM_SLOT_HPP
#include <cstdint> #include <cstdint>
#include <tuple>
#include "caf/detail/comparable.hpp"
namespace caf { namespace caf {
/// Actor-specific identifier for stream traffic. /// Identifies a single stream path in the same way a TCP port identifies a
using stream_slot = uint64_t; /// connection over IP.
using stream_slot = uint16_t;
/// Maps two `stream_slot` values into a pair for storing sender and receiver
/// slot information.
struct stream_slots : detail::comparable<stream_slots>{
stream_slot sender;
stream_slot receiver;
// -- constructors, destructors, and assignment operators --------------------
constexpr stream_slots(stream_slot sender_slot, stream_slot receiver_slot)
: sender(sender_slot),
receiver(receiver_slot) {
// nop
}
// -- observers --------------------------------------------------------------
/// Returns an inverted pair, i.e., swaps sender and receiver slot.
constexpr stream_slots invert() const {
return {receiver, sender};
}
inline int_fast32_t compare(stream_slots other) const noexcept {
static_assert(sizeof(stream_slots) == sizeof(int32_t),
"sizeof(stream_slots) != sizeof(int32_t)");
return reinterpret_cast<const int32_t&>(*this)
- reinterpret_cast<int32_t&>(other);
}
};
/// @relates stream_slots
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_slots& x) {
return f(std::tie(x.sender, x.receiver));
}
} // namespace caf } // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
// This test simulates a complex multiplexing over multiple layers of WDRR
// scheduled queues. The goal is to reduce the complex mailbox management of
// CAF to its bare bones in order to test whether the multiplexing of stream
// traffic and asynchronous messages works as intended.
//
// The setup is a fixed WDRR queue with three nestes queues. The first nested
// queue stores asynchronous messages, the second one upstream messages, and
// the last queue is a dynamic WDRR queue storing downstream messages.
#define CAF_SUITE mock_streaming_classes
#include <memory>
#include "caf/variant.hpp"
#include "caf/stream_slot.hpp"
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/detail/overload.hpp"
using namespace caf;
using namespace caf::intrusive;
namespace {
// -- utility ------------------------------------------------------------------
struct print_with_comma_t {
bool first = true;
template <class T>
std::ostream& operator()(std::ostream& out, const T& x) {
if (!first)
out << ", ";
else
first = false;
return out << x;
}
template <class T>
std::ostream& operator()(std::ostream& out, const logger::arg_wrapper<T>& x) {
if (!first)
out << ", ";
else
first = false;
return out << x.name << " = " << deep_to_string(x.value);
}
};
template <class T, class... Ts>
std::string collapse_args(const T& x, const Ts&... xs) {
std::ostringstream out;
print_with_comma_t f;
f(out, x);
unit(f(out, xs)...);
return out.str();
}
#define TRACE(name, type, ...) \
CAF_MESSAGE(name << " received a " << #type << ": " \
<< collapse_args(__VA_ARGS__));
// -- forward declarations -----------------------------------------------------
// Mimics an actor.
struct entity;
// Mimics a stream_manager.
struct manager;
// Mimics a message.
struct msg;
// Mimics an inbound_path.
struct in;
// Mimics an outbound_path.
struct out;
// Mimics stream_handshake_msg.
struct handshake;
// Mimics downstream_msg.
struct dmsg;
// Mimics upstream_msg.
struct umsg;
// -- message types ------------------------------------------------------------
struct handshake {
stream_slot sender_slot;
};
struct dmsg {
struct batch {
std::vector<int> xs;
};
struct close {
// nop
};
stream_slots slots;
variant<batch, close> content;
};
struct umsg {
struct ack_handshake {
long credit;
};
struct ack_batch {
long credit;
};
struct drop {
// nop
};
stream_slots slots;
variant<ack_batch, ack_handshake, drop> content;
};
struct msg : intrusive::singly_linked<msg> {
entity* sender;
variant<handshake, umsg, dmsg> content;
template <class T>
msg(entity* from ,T&& x) : sender(from), content(std::forward<T>(x)) {
// nop
}
};
// -- manager and path handlers ------------------------------------------------
struct manager {
entity* self;
int x;
int num_messages;
int input_paths = 0;
int output_paths = 0;
manager(entity* parent, int num = 0) : self(parent), x(0), num_messages(num) {
// nop
}
bool done() const {
return (input_paths | output_paths) == 0;
}
void push(entity* to, stream_slots slots, int num);
void operator()(entity* sender, stream_slots slots, in* path, dmsg::batch& x);
};
using manager_ptr = std::shared_ptr<manager>;
struct in {
manager_ptr mgr;
in(manager_ptr ptr) : mgr(std::move(ptr)) {
// nop
}
void operator()(msg& x);
};
struct out {
};
// -- policies and queues ------------------------------------------------------
struct policy_base {
using mapped_type = msg;
using task_size_type = size_t;
using deficit_type = size_t;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<msg, deleter_type>;
};
struct handshake_queue_policy : policy_base {
task_size_type task_size(const msg&) {
return 1;
}
};
using handshake_queue = drr_queue<handshake_queue_policy>;
struct umsg_queue_policy : policy_base {
manager* mgr;
umsg_queue_policy(manager* ptr) : mgr(ptr) {
// nop
}
task_size_type task_size(const msg&) {
return 1;
}
};
using umsg_queue = drr_queue<umsg_queue_policy>;
struct inner_dmsg_queue_policy : policy_base {
using key_type = stream_slot;
task_size_type task_size(const msg& x) {
return visit(*this, get<dmsg>(x.content).content);
}
task_size_type operator()(const dmsg::batch& x) const {
CAF_REQUIRE_NOT_EQUAL(x.xs.size(), 0u);
return x.xs.size();
}
task_size_type operator()(const dmsg::close&) const {
return 1;
}
inner_dmsg_queue_policy(std::unique_ptr<in> ptr) : handler(std::move(ptr)) {
// nop
}
std::unique_ptr<in> handler;
};
using inner_dmsg_queue = drr_queue<inner_dmsg_queue_policy>;
struct dmsg_queue_policy : policy_base {
using key_type = stream_slot;
using queue_map_type = std::map<stream_slot, inner_dmsg_queue>;
key_type id_of(const msg& x) {
return get<dmsg>(x.content).slots.receiver;
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) {
return x;
}
};
using dmsg_queue = wdrr_dynamic_multiplexed_queue<dmsg_queue_policy>;
struct mbox_policy : policy_base {
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) {
return x;
}
size_t id_of(const msg& x) {
return x.content.index();
}
};
using mbox_queue = wdrr_fixed_multiplexed_queue<mbox_policy, handshake_queue,
umsg_queue, dmsg_queue>;
// -- manager and entity -------------------------------------------------------
template <class Target>
struct dispatcher {
Target& f;
entity* sender;
stream_slots slots;
template <class T>
void operator()(T&& x) {
f(sender, slots, std::forward<T>(x));
}
};
struct entity {
mbox_queue mbox;
const char* name;
entity(const char* cstr_name)
: mbox(mbox_policy{}, handshake_queue_policy{}, nullptr,
dmsg_queue_policy{}),
name(cstr_name) {
// nop
}
void start_streaming(entity& to, int num_messages) {
CAF_REQUIRE_NOT_EQUAL(num_messages, 0);
auto slot = next_slot++;
CAF_MESSAGE(name << " starts streaming to " << to.name
<< " on slot " << slot);
to.enqueue<handshake>(this, slot);
auto ptr = std::make_shared<manager>(this, num_messages);
ptr->output_paths += 1;
pending_managers_.emplace(slot, std::move(ptr));
}
template <class T, class... Ts>
bool enqueue(entity* sender, Ts&&... xs) {
return mbox.emplace_back(sender, T{std::forward<Ts>(xs)...});
}
void operator()(entity* sender, handshake& hs) {
TRACE(name, handshake, CAF_ARG2("sender", sender->name));
auto slot = next_slot++;
//stream_slots id{slot, hs.sender_slot};
stream_slots id{hs.sender_slot, slot};
// Create required state.
auto mgr = std::make_shared<manager>(this, 0);
managers_.emplace(id, mgr);
mgr->input_paths += 1;
// Create a new queue in the mailbox for incoming traffic.
get<2>(mbox.queues())
.queues()
.emplace(slot, std::unique_ptr<in>{new in(mgr)});
// Acknowledge stream.
sender->enqueue<umsg>(this, id.invert(), umsg::ack_handshake{10});
}
void operator()(entity* sender, stream_slots slots, umsg::ack_handshake& x) {
TRACE(name, ack_handshake, CAF_ARG(slots),
CAF_ARG2("sender", sender->name));
// Get the manager for that stream.
auto i = pending_managers_.find(slots.receiver);
CAF_REQUIRE_NOT_EQUAL(i, pending_managers_.end());
// Create a new queue in the mailbox for incoming traffic.
// Swap the sender/receiver perspective to generate the ID we are using.
managers_.emplace(slots, i->second);
i->second->push(sender, slots.invert(), x.credit);
pending_managers_.erase(i);
}
void operator()(entity* sender, stream_slots input_slots,
umsg::ack_batch& x) {
TRACE(name, ack_batch, CAF_ARG(input_slots),
CAF_ARG2("sender", sender->name));
// Get the manager for that stream.
auto slots = input_slots.invert();
auto i = managers_.find(input_slots);
CAF_REQUIRE_NOT_EQUAL(i, managers_.end());
i->second->push(sender, slots, x.credit);
if (i->second->done())
managers_.erase(i);
}
void operator()(entity* sender, stream_slots slots, in*, dmsg::close&) {
TRACE(name, close, CAF_ARG(slots), CAF_ARG2("sender", sender->name));
auto i = managers_.find(slots);
CAF_REQUIRE_NOT_EQUAL(i, managers_.end());
i->second->input_paths -= 1;
get<2>(mbox.queues()).erase_later(slots.receiver);
if (i->second->done()) {
CAF_MESSAGE(name << " cleans up path " << deep_to_string(slots));
managers_.erase(i);
}
}
stream_slot next_slot = 1;
std::map<stream_slot, int> generators;
std::map<stream_slot, std::vector<int>> caches;
struct source {
stream_slot slot;
entity* ptr;
};
std::map<stream_slot, manager_ptr> pending_managers_;
std::map<stream_slots, manager_ptr> managers_;
};
void manager::push(entity* to, stream_slots slots, int num) {
CAF_REQUIRE_NOT_EQUAL(num, 0);
std::vector<int> xs;
if (x + num > num_messages)
num = num_messages - x;
if (num == 0) {
CAF_MESSAGE(self->name << " is done sending batches");
to->enqueue<dmsg>(self, slots, dmsg::close{});
output_paths -= 1;
return;
}
CAF_MESSAGE(self->name << " pushes "
<< num << " new items to " << to->name
<< " slots = " << deep_to_string(slots));
for (int i = 0; i < num; ++i)
xs.emplace_back(x++);
CAF_REQUIRE_NOT_EQUAL(xs.size(), 0u);
auto emplace_res = to->enqueue<dmsg>(self, slots,
dmsg::batch{std::move(xs)});
CAF_CHECK_EQUAL(emplace_res, true);
}
void manager::operator()(entity* sender, stream_slots slots, in*,
dmsg::batch& batch) {
TRACE(self->name, batch, CAF_ARG(slots), CAF_ARG2("sender", sender->name),
CAF_ARG(batch.xs));
sender->enqueue<umsg>(self, slots.invert(), umsg::ack_batch{10});
}
struct msg_visitor {
entity* self;
using is_handshake = std::integral_constant<size_t, 0>;
using is_umsg = std::integral_constant<size_t, 1>;
using is_dmsg = std::integral_constant<size_t, 2>;
using result_type = intrusive::task_result;
result_type operator()(is_handshake, handshake_queue&, msg& x) {
(*self)(x.sender, get<handshake>(x.content));
return intrusive::task_result::resume;
}
result_type operator()(is_umsg, umsg_queue&, msg& x) {
auto sender = x.sender;
auto& um = get<umsg>(x.content);
auto f = detail::make_overload(
[&](umsg::ack_handshake& y) {
(*self)(sender, um.slots, y);
},
[&](umsg::ack_batch& y) {
(*self)(sender, um.slots, y);
},
[&](umsg::drop&) {
//(*self)(sender, um.slots, y);
}
);
visit(f, um.content);
return intrusive::task_result::resume;
}
result_type operator()(is_dmsg, dmsg_queue&, stream_slot, inner_dmsg_queue& q,
msg& x) {
auto inptr = q.policy().handler.get();
auto dm = get<dmsg>(x.content);
auto f = detail::make_overload(
[&](dmsg::batch& y) {
(*inptr->mgr)(x.sender, dm.slots, inptr, y);
},
[&](dmsg::close& y) {
auto self = inptr->mgr->self;
(*self)(x.sender, dm.slots, inptr, y);
});
visit(f, dm.content);
return intrusive::task_result::resume;
}
};
// -- fixture ------------------------------------------------------------------
struct fixture {
entity alice{"alice"};
entity bob{"bob"};
fixture() {
/// Make sure to test whether the slot IDs are properly handled.
alice.next_slot = 123;
bob.next_slot = 321;
}
};
} // namespace <anonymous>
// -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(mock_streaming_classes_tests, fixture)
CAF_TEST(simple_handshake) {
bob.start_streaming(alice, 30);
msg_visitor f{&alice};
msg_visitor g{&bob};
while (!alice.mbox.empty() || !bob.mbox.empty()) {
alice.mbox.new_round(1, f);
bob.mbox.new_round(1, g);
}
// Check whether alice and bob cleaned up their state properly.
CAF_CHECK(get<2>(alice.mbox.queues()).queues().empty());
CAF_CHECK(get<2>(bob.mbox.queues()).queues().empty());
CAF_CHECK(alice.pending_managers_.empty());
CAF_CHECK(bob.pending_managers_.empty());
CAF_CHECK(alice.managers_.empty());
CAF_CHECK(bob.managers_.empty());
}
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