Commit 63eb3af5 authored by Dominik Charousset's avatar Dominik Charousset

started to re-implement basic libcppa features

parent c91bf3fc
......@@ -88,12 +88,15 @@ endif ()
if (ENABLE_DEBUG)
set(CMAKE_BUILD_TYPE Debug)
add_definitions(-DCPPA_DEBUG_MODE)
message(STATUS "----- configured using --enable-debug")
endif (ENABLE_DEBUG)
if (CPPA_LOG_LEVEL)
add_definitions(-DCPPA_LOG_LEVEL=${CPPA_LOG_LEVEL})
endif(CPPA_LOG_LEVEL)
message(STATUS "BLUBB: ${CMAKE_BUILD_TYPE}")
# set build default build type if not set
if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
set(CMAKE_BUILD_TYPE RelWithDebInfo)
......
......@@ -123,9 +123,9 @@ configure ()
cd $workdir
if [ -n "$5" ]; then
cmake -G "$5" "$CMakeCacheEntries" "$sourcedir"
cmake -G "$5" $CMakeCacheEntries $sourcedir
else
cmake "$CMakeCacheEntries" "$sourcedir"
cmake $CMakeCacheEntries $sourcedir
fi
echo "# This is the command used to configure this build" > config.status
......@@ -136,7 +136,6 @@ configure ()
# Set defaults.
builddir="$sourcedir/build"
CMakeCacheEntries=""
append_cache_entry CMAKE_BUILD_TYPE STRING RelWithDebInfo
append_cache_entry CMAKE_INSTALL_PREFIX PATH /usr/local
append_cache_entry ENABLE_DEBUG BOOL false
append_cache_entry DISABLE_CONTEXT_SWITCHING BOOL false
......
......@@ -43,6 +43,7 @@ class node_id;
class behavior;
class any_tuple;
class self_type;
class actor_addr;
class actor_proxy;
class untyped_actor;
class abstract_actor;
......
......@@ -22,9 +22,8 @@ template<class Base,
class SchedulingPolicy,
class PriorityPolicy,
class ResumePolicy,
class InvokePolicy,
bool OverrideDequeue = std::is_base_of<blocking_untyped_actor, Base>::value>
class proper_actor : public ResumePolicy::template mixin<Base> {
class InvokePolicy>
class proper_actor_base : public ResumePolicy::template mixin<Base> {
friend SchedulingPolicy;
friend PriorityPolicy;
......@@ -36,22 +35,16 @@ class proper_actor : public ResumePolicy::template mixin<Base> {
public:
template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) { }
void enqueue(const message_header& hdr, any_tuple msg) override {
m_scheduling_policy.enqueue(this, hdr, msg);
}
inline void launch() { m_scheduling_policy.launch(this); }
inline mailbox_element* next_message() {
return m_priority_policy.next_message(this);
}
inline void invoke(mailbox_element* msg) {
m_invoke_policy.invoke(this, msg);
}
// grant access to the actor's mailbox
typename Base::mailbox_type& mailbox() { return this->m_mailbox; }
......@@ -67,16 +60,10 @@ class proper_actor : public ResumePolicy::template mixin<Base> {
m_scheduling_policy.push_timeout();
}
inline void current_node(mailbox_element* ptr) {
this->m_current_node = ptr;
}
inline void request_timeout(const util::duration& rel_timeout) {
m_invoke_policy.request_timeout(this, rel_timeout);
}
inline mailbox_element* current_node() { return this->m_current_node; }
detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; }
protected:
......@@ -88,16 +75,49 @@ class proper_actor : public ResumePolicy::template mixin<Base> {
};
template<class Base,
class SchedulingPolicy,
class PriorityPolicy,
class ResumePolicy,
class InvokePolicy,
bool OverrideDequeue = std::is_base_of<blocking_untyped_actor, Base>::value>
class proper_actor : public proper_actor_base<Base, SchedulingPolicy,
PriorityPolicy, ResumePolicy,
InvokePolicy> {
typedef proper_actor_base<Base, SchedulingPolicy, PriorityPolicy,
ResumePolicy, InvokePolicy>
super;
public:
template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; }
inline void launch() {
this->bhvr_stack().push_back(this->make_behavior());
this->m_scheduling_policy.launch(this);
}
bool invoke(mailbox_element* msg) {
return this->m_invoke_policy.invoke(this, msg, bhvr_stack().back());
}
};
// for blocking actors, there's one more member function to implement
template <class Base, class SchedulingPolicy, class PriorityPolicy,
class ResumePolicy, class InvokePolicy>
class proper_actor<
Base, SchedulingPolicy, PriorityPolicy, ResumePolicy, InvokePolicy,
true> final : public proper_actor<Base, SchedulingPolicy, PriorityPolicy,
ResumePolicy, InvokePolicy, false> {
true> final : public proper_actor_base<Base, SchedulingPolicy,
PriorityPolicy, ResumePolicy,
InvokePolicy> {
typedef proper_actor<Base, SchedulingPolicy, PriorityPolicy,
ResumePolicy, InvokePolicy, false>
typedef proper_actor_base<Base, SchedulingPolicy, PriorityPolicy,
ResumePolicy, InvokePolicy>
super;
public:
......@@ -105,6 +125,10 @@ class proper_actor<
template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
inline void launch() {
this->m_scheduling_policy.launch(this);
}
void dequeue(behavior& bhvr) override {
if (bhvr.timeout().valid()) {
auto tout =
......
......@@ -60,16 +60,16 @@ namespace cppa { namespace detail {
// ordered according to demangled type name (see uniform_type_info_map.cpp)
using mapped_type_list = util::type_list<
bool,
actor,
actor_addr,
any_tuple,
atom_value,
actor,
channel,
group_ptr,
node_id_ptr,
io::accept_handle,
io::connection_handle,
message_header,
std::nullptr_t,
unit_t,
util::buffer,
util::duration,
......
......@@ -343,6 +343,14 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
/** @cond PRIVATE */
inline void current_node(mailbox_element* ptr) {
this->m_current_node = ptr;
}
inline mailbox_element* current_node() {
return this->m_current_node;
}
inline message_id new_request_id() {
auto result = ++m_last_request_id;
m_pending_responses.push_back(result.response_id());
......
......@@ -59,7 +59,8 @@ class continue_helper {
template<typename F>
continue_helper& continue_with(F) {
//FIXME
throw std::logic_error("not implemented yet");
}
inline message_id get_message_id() const {
......
......@@ -75,18 +75,14 @@ class context_switching_resume {
for (;;) {
switch (call(&m_fiber, from)) {
case yield_state::done: {
CPPA_REQUIRE(next_job == nullptr);
return resumable::done;
}
case yield_state::ready: { break; }
case yield_state::blocked: {
CPPA_REQUIRE(next_job == nullptr);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (this->cas_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
CPPA_REQUIRE(next_job == nullptr);
break;
}
case actor_state::blocked: {
......
......@@ -106,7 +106,6 @@ class cooperative_scheduling {
switch (state) {
case actor_state::blocked: {
if (set_ready()) {
CPPA_REQUIRE(m_scheduler != nullptr);
//m_scheduler->enqueue(this);
return;
}
......
......@@ -57,6 +57,9 @@ class event_based_resume {
template<typename... Ts>
mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
// implemented in detail::proper_actor
virtual bool invoke(mailbox_element* msg) = 0;
resumable::resume_result resume(util::fiber*) override {
CPPA_LOG_TRACE("id = " << this->id()
<< ", state = " << static_cast<int>(this->state()));
......@@ -81,13 +84,11 @@ class event_based_resume {
this->on_exit();
return true;
};
CPPA_REQUIRE(next_job == nullptr);
try {
//auto e = m_mailbox.try_pop();
for (auto e = this->m_mailbox.try_pop(); ; e = this->m_mailbox.try_pop()) {
//e = m_mailbox.try_pop();
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
this->set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
......@@ -97,9 +98,9 @@ class event_based_resume {
case actor_state::ready:
// interrupted by arriving message
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
CPPA_LOG_DEBUG("switched back to ready: "
"interrupted by "
"arriving message");
break;
case actor_state::blocked:
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked");
......@@ -116,21 +117,19 @@ class event_based_resume {
this->set_state(actor_state::ready);
}
}
/*
else {
if (this->bhvr_stack().invoke(m_recv_policy, this, e)) {
if (this->invoke(e)) {
CPPA_LOG_DEBUG_IF(m_chained_actor,
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (this->bhvr_stack().empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
return resume_result::actor_done;
return resume_result::done;
}
this->bhvr_stack().cleanup();
}
}
*/
}
}
catch (actor_exited& what) {
......
......@@ -56,25 +56,30 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
public:
static inline bool hm_should_skip(pointer) {
return false;
static inline bool hm_should_skip(pointer node) {
return node->marked;
}
template<class Client>
static inline pointer hm_begin(Client* client, pointer node) {
auto previous = client->current_node();
client->current_node(node);
template<class Actor>
inline pointer hm_begin(Actor* self, pointer node) {
auto previous = self->m_current_node;
self->m_current_node = node;
push_timeout();
node->marked = true;
return previous;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer /*previous*/) {
client->current_node(&(client->m_dummy_node));
template<class Actor>
inline void hm_cleanup(Actor* self, pointer previous) {
self->m_current_node->marked = false;
self->m_current_node = previous;
}
template<class Client>
static inline void hm_revert(Client* client, pointer previous) {
client->current_node(previous);
template<class Actor>
inline void hm_revert(Actor* self, pointer previous) {
self->m_current_node->marked = false;
self->m_current_node = previous;
pop_timeout();
}
typedef std::chrono::high_resolution_clock::time_point timeout_type;
......
......@@ -92,15 +92,20 @@ class no_scheduling {
template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
std::cout << "enqueue\n";
auto ptr = self->new_mailbox_element(hdr, std::move(msg));
switch (self->mailbox().enqueue(ptr)) {
default:
std::cout << "enqueue: default case (do nothing)\n";
break;
case intrusive::first_enqueued: {
std::cout << "enqueue: first enqueue -> notify\n";
lock_type guard(m_mtx);
m_cv.notify_one();
break;
}
default: break;
case intrusive::queue_closed:
std::cout << "enqueue: mailbox closed!\n";
if (hdr.id.valid()) {
detail::sync_request_bouncer f{self->exit_reason()};
f(hdr.sender, hdr.id);
......@@ -115,15 +120,18 @@ class no_scheduling {
util::fiber fself;
auto rr = resumable::resume_later;
while (rr != resumable::done) {
std::cout << "before await_data\n";
await_data(self);
self->resume(&fself);
std::cout << "before resume\n";
rr = self->resume(&fself);
std::cout << "after resume\n";
}
}).detach();
}
template<class Actor>
void await_data(Actor* self) {
if (self->mailbox().empty()) {
while (self->mailbox().empty()) {
lock_type guard(m_mtx);
while (self->mailbox().empty()) m_cv.wait(guard);
}
......@@ -132,7 +140,7 @@ class no_scheduling {
template<class Actor>
bool await_data(Actor* self, const timeout_type& abs_time) {
CPPA_REQUIRE(!self->mailbox().closed());
if (self->mailbox().empty()) {
while (self->mailbox().empty()) {
lock_type guard(m_mtx);
while (self->mailbox().empty()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) {
......
......@@ -46,8 +46,11 @@ struct not_prioritizing {
}
template<class Actor, typename F>
bool fetch_messages(Actor*, F) {
//FIXME
bool fetch_messages(Actor* self, F cb) {
auto fetch = [self] { return self->mailbox().try_pop(); };
for (auto e = fetch(); e != nullptr; e = fetch()) {
cb(e);
}
}
};
......
......@@ -49,30 +49,25 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
sequential_invoke() : m_has_pending_tout(false), m_pending_tout(0) { }
static inline bool hm_should_skip(pointer node) {
return node->marked;
static inline bool hm_should_skip(pointer) {
return false;
}
template<class Actor>
inline pointer hm_begin(Actor* self, pointer node) {
auto previous = self->m_current_node;
self->m_current_node = node;
push_timeout();
node->marked = true;
static inline pointer hm_begin(Actor* self, pointer node) {
auto previous = self->current_node();
self->current_node(node);
return previous;
}
template<class Actor>
inline void hm_cleanup(Actor* self, pointer previous) {
self->m_current_node->marked = false;
self->m_current_node = previous;
static inline void hm_cleanup(Actor* self, pointer) {
self->current_node(&(self->m_dummy_node));
}
template<class Actor>
inline void hm_revert(Actor* self, pointer previous) {
self->m_current_node->marked = false;
self->m_current_node = previous;
pop_timeout();
static inline void hm_revert(Actor* self, pointer previous) {
self->current_node(previous);
}
inline void reset_timeout() {
......
......@@ -68,11 +68,14 @@ actor spawn(Ts&&... args) {
"top-level spawns cannot have monitor or link flag");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag");
/*
using scheduling_policy = typename std::conditional<
has_detach_flag(Options),
policy::no_scheduling,
policy::cooperative_scheduling
>::type;
*/
using scheduling_policy = policy::no_scheduling;
using priority_policy = typename std::conditional<
has_priority_aware_flag(Options),
policy::prioritizing,
......@@ -100,19 +103,6 @@ actor spawn(Ts&&... args) {
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
ptr->launch();
return ptr;
/*
scheduled_actor_ptr ptr;
if (has_priority_aware_flag(Options)) {
using derived = typename extend<Impl>::template with<threaded, prioritizing>;
ptr = make_counted<derived>(std::forward<Ts>(args)...);
}
else if (has_detach_flag(Options)) {
using derived = typename extend<Impl>::template with<threaded>;
ptr = make_counted<derived>(std::forward<Ts>(args)...);
}
else ptr = make_counted<Impl>(std::forward<Ts>(args)...);
return get_scheduler()->exec(Options, std::move(ptr));
*/
}
/**
......
......@@ -32,7 +32,7 @@
#include "cppa/config.hpp"
#include "cppa/cppa.hpp"
//#include "cppa/cppa.hpp"
#include "cppa/singletons.hpp"
#include "cppa/util/scope_guard.hpp"
......@@ -287,7 +287,6 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
m_dummy_node.msg = move(msg);
m_dummy_node.mid = hdr.id;
try {
using detail::receive_policy;
auto bhvr = m_bhvr_stack.back();
switch (m_invoke.handle_message(this,
&m_dummy_node,
......
......@@ -214,7 +214,6 @@ class local_group_proxy : public local_group {
: super(false, forward<Ts>(args)...) {
CPPA_REQUIRE(m_broker == nullptr);
CPPA_REQUIRE(remote_broker != nullptr);
CPPA_REQUIRE(remote_broker->is_proxy());
m_broker = move(remote_broker);
m_proxy_broker = spawn<proxy_broker, hidden>(this);
}
......
......@@ -148,12 +148,16 @@ response_handle local_actor::make_response_handle() {
}
*/
void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) {
dest.enqueue({address(), dest, prio}, std::move(what));
}
void local_actor::send_tuple(const channel& dest, any_tuple what) {
//TODO:
send_tuple(message_priority::normal, dest, std::move(what));
}
void local_actor::send_exit(const actor_addr& whom, std::uint32_t reason) {
send(detail::raw_access::get(whom), atom("EXIT"), reason);
}
void local_actor::remove_handler(message_id) {
......@@ -170,16 +174,15 @@ message_future local_actor::timed_sync_send_tuple(const util::duration& rtime,
}
void local_actor::send_tuple(message_priority prio, const channel& dest, any_tuple what) {
}
message_future local_actor::sync_send_tuple(const actor& dest, any_tuple what) {
}
response_handle local_actor::make_response_handle() {
return {};
auto n = m_current_node;
response_handle result{address(), n->sender, n->mid.response_id()};
n->mid.mark_as_answered();
return result;
}
void local_actor::cleanup(std::uint32_t reason) {
......
......@@ -55,9 +55,11 @@ bool response_handle::synchronous() const {
}
void response_handle::apply(any_tuple msg) const {
std::cout << "response_handle::apply\n";
if (valid()) {
auto ptr = detail::actor_addr_cast<abstract_actor>(m_to);
ptr->enqueue({m_from, ptr, m_id}, move(msg));
std::cout << "response_handle::apply: after ptr->enqueue\n";
}
}
......
......@@ -69,7 +69,6 @@ void publish(actor whom, std::unique_ptr<acceptor> aptr) {
CPPA_LOG_TRACE(CPPA_TARG(whom, to_string) << ", " << CPPA_MARG(ptr, get)
<< ", args.size() = " << args.size());
if (!whom) return;
CPPA_REQUIRE(args.size() == 0);
get_actor_registry()->put(whom->id(), detail::actor_addr_cast<abstract_actor>(whom));
auto mm = get_middleman();
mm->register_acceptor(whom, new peer_acceptor(mm, move(aptr), whom));
......
......@@ -58,18 +58,17 @@ namespace cppa { namespace detail {
// maps demangled names to libcppa names
// WARNING: this map is sorted, insert new elements *in sorted order* as well!
/* extern */ const char* mapped_type_names[][2] = {
{ "bool", "bool" },
{ "cppa::actor", "@actor" },
{ "cppa::actor_addr", "@addr" },
{ "bool", "bool" },
{ "cppa::any_tuple", "@tuple" },
{ "cppa::atom_value", "@atom" },
{ "cppa::intrusive_ptr<cppa::channel>", "@channel" },
{ "cppa::channel", "@channel" },
{ "cppa::intrusive_ptr<cppa::group>", "@group" },
{ "cppa::intrusive_ptr<cppa::node_id>", "@proc" },
{ "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" },
{ "cppa::message_header", "@header" },
{ "cppa::nullptr_t", "@null" },
{ "cppa::unit_t", "@0" },
{ "cppa::util::buffer", "@buffer" },
{ "cppa::util::duration", "@duration" },
......
......@@ -458,6 +458,24 @@ void test_continuation() {
void test_spawn() {
scoped_actor self;
auto s = spawn([](untyped_actor* self) -> behavior {
return (
others() >> [=]() -> any_tuple {
cout << "received: " << to_string(self->last_dequeued()) << endl;
return self->last_dequeued();
}
);
});
cout << "spawned actor, waiting for response" << endl;
self->send(s, atom("hello"));
self->receive(
others() >> [&] {
cout << "received: " << to_string(self->last_dequeued()) << endl;
}
);
self->await_all_other_actors_done();
return;
test_serial_reply();
test_or_else();
test_continuation();
......
......@@ -102,7 +102,7 @@ struct C : sb_actor<C> {
struct D : popular_actor {
D(const actor& buddy) : popular_actor(buddy) { }
behavior make_behavior() override {
become (
return (
others() >> [=] {
/*
response_handle handle = make_response_handle();
......@@ -140,7 +140,7 @@ struct server : untyped_actor {
behavior make_behavior() override {
auto die = [=] { quit(exit_reason::user_shutdown); };
become (
return (
on(atom("idle"), arg_match) >> [=](actor worker) {
become (
keep_behavior,
......
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