Commit 1a6cbb9a authored by Dominik Charousset's avatar Dominik Charousset

re-organized policy handling

parent 8a9b23eb
...@@ -38,7 +38,7 @@ if (CMAKE_CXX_FLAGS) ...@@ -38,7 +38,7 @@ if (CMAKE_CXX_FLAGS)
set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "") set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "")
else (CMAKE_CXX_FLAGS) else (CMAKE_CXX_FLAGS)
set(CXXFLAGS_PROVIDED false) set(CXXFLAGS_PROVIDED false)
set(CMAKE_CXX_FLAGS "-std=c++11 -Wextra -Wall -pedantic") set(CMAKE_CXX_FLAGS "-std=c++11 -Wextra -Wall -pedantic -fno-omit-frame-pointer -fsanitize=address")
set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g") set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g")
set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os") set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os")
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG") set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG")
...@@ -77,7 +77,7 @@ elseif ("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang") ...@@ -77,7 +77,7 @@ elseif ("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang")
message(STATUS "NOTE: Automatically added -stdlib=libc++ flag, " message(STATUS "NOTE: Automatically added -stdlib=libc++ flag, "
"you can override this by defining CMAKE_CXX_FLAGS " "you can override this by defining CMAKE_CXX_FLAGS "
"(see 'configure --help')") "(see 'configure --help')")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -stdlib=libc++") #set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -stdlib=libc++")
endif () endif ()
else () else ()
message(FATAL_ERROR "Your C++ compiler does not support C++11 " message(FATAL_ERROR "Your C++ compiler does not support C++11 "
......
...@@ -337,3 +337,4 @@ cppa/policy/middleman_scheduling.hpp ...@@ -337,3 +337,4 @@ cppa/policy/middleman_scheduling.hpp
src/untyped_actor.cpp src/untyped_actor.cpp
src/functor_based_blocking_actor.cpp src/functor_based_blocking_actor.cpp
src/blocking_untyped_actor.cpp src/blocking_untyped_actor.cpp
cppa/policy/policies.hpp
...@@ -104,7 +104,7 @@ class behavior_stack_based : public Base { ...@@ -104,7 +104,7 @@ class behavior_stack_based : public Base {
virtual void become_waiting_for(behavior bhvr, message_id mf) = 0; virtual void become_waiting_for(behavior bhvr, message_id mf) = 0;
virtual void do_become(behavior&& bhvr, bool discard_old) = 0; virtual void do_become(behavior bhvr, bool discard_old) = 0;
inline bool has_behavior() const { inline bool has_behavior() const {
return m_bhvr_stack.empty() == false; return m_bhvr_stack.empty() == false;
......
...@@ -321,12 +321,6 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> { ...@@ -321,12 +321,6 @@ class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
virtual void dequeue_response(behavior& bhvr, message_id mid) = 0; virtual void dequeue_response(behavior& bhvr, message_id mid) = 0;
virtual mailbox_element* dequeue() = 0;
virtual mailbox_element* try_dequeue() = 0;
virtual mailbox_element* try_dequeue(const timeout_type&) = 0;
void await_all_other_actors_done(); void await_all_other_actors_done();
virtual void act() = 0; virtual void act() = 0;
......
...@@ -529,11 +529,9 @@ actor spawn_io(io::input_stream_ptr in, ...@@ -529,11 +529,9 @@ actor spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out, io::output_stream_ptr out,
Ts&&... args) { Ts&&... args) {
using namespace policy; using namespace policy;
using proper_impl = detail::proper_actor<Impl, using ps = policies<middleman_scheduling, not_prioritizing,
middleman_scheduling, no_resume, cooperative_scheduling>;
not_prioritizing, using proper_impl = detail::proper_actor<Impl, ps>;
no_resume,
cooperative_scheduling>;
auto ptr = make_counted<proper_impl>(std::move(in), std::move(out), auto ptr = make_counted<proper_impl>(std::move(in), std::move(out),
std::forward<Ts>(args)...); std::forward<Ts>(args)...);
ptr->launch(); ptr->launch();
......
...@@ -94,44 +94,6 @@ class behavior_stack ...@@ -94,44 +94,6 @@ class behavior_stack
m_erased_elements.clear(); m_erased_elements.clear();
} }
template<class Policy, class Client>
bool invoke(Policy& policy, Client* client, mailbox_element* node) {
CPPA_REQUIRE(!empty());
CPPA_REQUIRE(client != nullptr);
CPPA_REQUIRE(node != nullptr);
// use a copy of bhvr, because invoked behavior might change m_elements
auto id = m_elements.back().second;
auto bhvr = m_elements.back().first;
if (policy.invoke(client, node, bhvr, id)) {
bool repeat;
// try to match cached messages
do {
// remove synchronous response handler if needed
if (id.valid()) {
erase_if([id](const element_type& value) {
return id == value.second;
});
}
if (!empty()) {
id = m_elements.back().second;
bhvr = m_elements.back().first;
repeat = policy.invoke_from_cache(client, bhvr, id);
}
else repeat = false;
} while (repeat);
return true;
}
return false;
}
template<class Policy, class Client>
void exec(Policy& policy, Client* client) {
while (!empty()) {
invoke(policy, client, policy.fetch_message(client));
cleanup();
}
}
private: private:
std::vector<element_type> m_elements; std::vector<element_type> m_elements;
......
This diff is collapsed.
...@@ -367,6 +367,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -367,6 +367,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
return m_state; return m_state;
} }
void cleanup(std::uint32_t reason);
protected: protected:
template<typename... Ts> template<typename... Ts>
...@@ -374,8 +376,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -374,8 +376,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
return mailbox_element::create(std::forward<Ts>(args)...); return mailbox_element::create(std::forward<Ts>(args)...);
} }
void cleanup(std::uint32_t reason);
// true if this actor receives EXIT messages as ordinary messages // true if this actor receives EXIT messages as ordinary messages
bool m_trap_exit; bool m_trap_exit;
......
...@@ -201,7 +201,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -201,7 +201,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg) CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg)
#if CPPA_LOG_LEVEL < 4 #if CPPA_LOG_LEVEL < 4
# define CPPA_PRINT4(arg0, arg1, arg2, arg3, arg4) # define CPPA_PRINT4(arg0, arg1, arg2, arg3)
# else # else
# define CPPA_PRINT4(lvlname, classname, funname, msg) \ # define CPPA_PRINT4(lvlname, classname, funname, msg) \
::cppa::logging::trace_helper cppa_trace_helper_ { \ ::cppa::logging::trace_helper cppa_trace_helper_ { \
...@@ -211,8 +211,8 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -211,8 +211,8 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#endif #endif
#if CPPA_LOG_LEVEL < 3 #if CPPA_LOG_LEVEL < 3
# define CPPA_PRINT3(arg0, arg1, arg2, arg3, arg4) # define CPPA_PRINT3(arg0, arg1, arg2, arg3)
# define CPPA_PRINT_IF3(arg0, arg1, arg2, arg3, arg4, arg5) # define CPPA_PRINT_IF3(arg0, arg1, arg2, arg3, arg4)
# else # else
# define CPPA_PRINT3(lvlname, classname, funname, msg) \ # define CPPA_PRINT3(lvlname, classname, funname, msg) \
CPPA_PRINT0(lvlname, classname, funname, msg) CPPA_PRINT0(lvlname, classname, funname, msg)
...@@ -221,8 +221,8 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -221,8 +221,8 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#endif #endif
#if CPPA_LOG_LEVEL < 2 #if CPPA_LOG_LEVEL < 2
# define CPPA_PRINT2(arg0, arg1, arg2, arg3, arg4) # define CPPA_PRINT2(arg0, arg1, arg2, arg3)
# define CPPA_PRINT_IF2(arg0, arg1, arg2, arg3, arg4, arg5) # define CPPA_PRINT_IF2(arg0, arg1, arg2, arg3, arg4)
# else # else
# define CPPA_PRINT2(lvlname, classname, funname, msg) \ # define CPPA_PRINT2(lvlname, classname, funname, msg) \
CPPA_PRINT0(lvlname, classname, funname, msg) CPPA_PRINT0(lvlname, classname, funname, msg)
......
...@@ -58,6 +58,12 @@ class mailbox_based : public Base { ...@@ -58,6 +58,12 @@ class mailbox_based : public Base {
return mailbox_element::create(std::forward<Ts>(args)...); return mailbox_element::create(std::forward<Ts>(args)...);
} }
void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
}
protected: protected:
typedef mailbox_based combined_type; typedef mailbox_based combined_type;
...@@ -67,12 +73,6 @@ class mailbox_based : public Base { ...@@ -67,12 +73,6 @@ class mailbox_based : public Base {
template<typename... Ts> template<typename... Ts>
mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) { } mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
}
mailbox_type m_mailbox; mailbox_type m_mailbox;
}; };
......
...@@ -53,13 +53,11 @@ class mailbox_element : public extend<memory_managed>::with<memory_cached> { ...@@ -53,13 +53,11 @@ class mailbox_element : public extend<memory_managed>::with<memory_cached> {
public: public:
typedef mailbox_element* pointer; mailbox_element* next; // intrusive next pointer
bool marked; // denotes if this node is currently processed
pointer next; // intrusive next pointer actor_addr sender;
bool marked; // denotes if this node is currently processed any_tuple msg; // 'content field'
actor_addr sender; message_id mid;
any_tuple msg; // 'content field'
message_id mid;
mailbox_element(mailbox_element&&) = delete; mailbox_element(mailbox_element&&) = delete;
mailbox_element(const mailbox_element&) = delete; mailbox_element(const mailbox_element&) = delete;
...@@ -79,6 +77,9 @@ class mailbox_element : public extend<memory_managed>::with<memory_cached> { ...@@ -79,6 +77,9 @@ class mailbox_element : public extend<memory_managed>::with<memory_cached> {
}; };
typedef std::unique_ptr<mailbox_element, detail::disposer>
unique_mailbox_element_pointer;
} // namespace cppa } // namespace cppa
#endif // CPPA_RECURSIVE_QUEUE_NODE_HPP #endif // CPPA_RECURSIVE_QUEUE_NODE_HPP
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#ifndef CPPA_POLICY_HPP #ifndef CPPA_POLICY_HPP
#define CPPA_POLICY_HPP #define CPPA_POLICY_HPP
#include "cppa/policy/policies.hpp"
#include "cppa/policy/no_resume.hpp" #include "cppa/policy/no_resume.hpp"
#include "cppa/policy/prioritizing.hpp" #include "cppa/policy/prioritizing.hpp"
#include "cppa/policy/no_scheduling.hpp" #include "cppa/policy/no_scheduling.hpp"
......
...@@ -60,7 +60,7 @@ class context_switching_resume { ...@@ -60,7 +60,7 @@ class context_switching_resume {
static void trampoline(void* _this); static void trampoline(void* _this);
// Base must be a mailbox-based actor // Base must be a mailbox-based actor
template<class Base> template<class Base, class Derived>
struct mixin : Base, resumable { struct mixin : Base, resumable {
template<typename... Ts> template<typename... Ts>
...@@ -103,6 +103,7 @@ class context_switching_resume { ...@@ -103,6 +103,7 @@ class context_switching_resume {
}; };
/*
template <class Actor, typename F> template <class Actor, typename F>
void fetch_messages(Actor* self, F cb) { void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop(); auto e = self->m_mailbox.try_pop();
...@@ -134,13 +135,14 @@ class context_switching_resume { ...@@ -134,13 +135,14 @@ class context_switching_resume {
e = self->m_mailbox.try_pop(); e = self->m_mailbox.try_pop();
} }
} }
*/
template<class Actor> template<class Actor>
void await_data(Actor* self) { void await_ready(Actor* self) {
while (self->m_mailbox.can_fetch_more() == false) { while (!self->has_next_message()) {
self->set_state(actor_state::about_to_block); self->set_state(actor_state::about_to_block);
// make sure mailbox is empty // double-check before going to block
if (self->m_mailbox.can_fetch_more()) { if (self->has_next_message()) {
// someone preempt us => continue // someone preempt us => continue
self->set_state(actor_state::ready); self->set_state(actor_state::ready);
} }
...@@ -149,12 +151,6 @@ class context_switching_resume { ...@@ -149,12 +151,6 @@ class context_switching_resume {
} }
} }
template<class Actor, typename AbsTimeout>
bool await_data(Actor* self, const AbsTimeout&) {
await_data(self);
return true;
}
private: private:
// members // members
......
...@@ -50,13 +50,6 @@ class cooperative_scheduling { ...@@ -50,13 +50,6 @@ class cooperative_scheduling {
using timeout_type = int; using timeout_type = int;
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
// this does return nullptr // this does return nullptr
template<class Actor, typename F> template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) { void fetch_messages(Actor* self, F cb) {
......
...@@ -51,50 +51,63 @@ class event_based_resume { ...@@ -51,50 +51,63 @@ class event_based_resume {
public: public:
// Base must be a mailbox-based actor // Base must be a mailbox-based actor
template<class Base> template<class Base, class Derived>
struct mixin : Base, resumable { struct mixin : Base, resumable {
template<typename... Ts> template<typename... Ts>
mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) { } mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
// implemented in detail::proper_actor inline Derived* dptr() {
virtual bool invoke(mailbox_element* msg) = 0; return static_cast<Derived*>(this);
}
resumable::resume_result resume(util::fiber*) override { resumable::resume_result resume(util::fiber*) override {
CPPA_LOG_TRACE("id = " << this->id() auto d = dptr();
<< ", state = " << static_cast<int>(this->state())); CPPA_LOG_TRACE("id = " << d->id()
CPPA_REQUIRE( this->state() == actor_state::ready << ", state = " << static_cast<int>(d->state()));
|| this->state() == actor_state::pending); CPPA_REQUIRE( d->state() == actor_state::ready
CPPA_PUSH_AID(this->id()); || d->state() == actor_state::pending);
CPPA_PUSH_AID(d->id());
auto done_cb = [&]() -> bool { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
if (this->exit_reason() == exit_reason::not_exited) { if ( d->exit_reason() == exit_reason::not_exited
if (this->planned_exit_reason() == exit_reason::not_exited) { && d->planned_exit_reason() == exit_reason::not_exited) {
this->planned_exit_reason(exit_reason::normal); d->planned_exit_reason(exit_reason::normal);
} }
this->on_exit(); d->on_exit();
if (!this->bhvr_stack().empty()) { if (!d->bhvr_stack().empty()) {
this->planned_exit_reason(exit_reason::not_exited); d->planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior return false; // on_exit did set a new behavior
}
this->cleanup(this->planned_exit_reason());
} }
this->set_state(actor_state::done); d->set_state(actor_state::done);
this->bhvr_stack().clear(); d->bhvr_stack().clear();
this->bhvr_stack().cleanup(); d->bhvr_stack().cleanup();
this->on_exit(); d->cleanup(d->planned_exit_reason());
return true; return true;
}; };
try { try {
//auto e = m_mailbox.try_pop(); for (;;) {
for (auto e = this->m_mailbox.try_pop(); ; e = this->m_mailbox.try_pop()) { auto ptr = dptr()->next_message();
//e = m_mailbox.try_pop(); if (ptr) {
if (e == nullptr) { CPPA_REQUIRE(!dptr()->bhvr_stack().empty());
auto bhvr = dptr()->bhvr_stack().back();
auto mid = dptr()->bhvr_stack().back_id();
if (dptr()->invoke_message(ptr, bhvr, mid)) {
if (dptr()->bhvr_stack().empty() && done_cb()) {
CPPA_LOG_DEBUG("behavior stack empty");
return resume_result::done;
}
}
// add ptr to cache if invoke_message did not
// reset it
if (ptr) dptr()->push_to_cache(std::move(ptr));
}
else {
CPPA_LOG_DEBUG("no more element in mailbox; going to block"); CPPA_LOG_DEBUG("no more element in mailbox; going to block");
this->set_state(actor_state::about_to_block); d->set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) { if (!dptr()->has_next_message()) {
switch (this->cas_state(actor_state::about_to_block, switch (d->cas_state(actor_state::about_to_block,
actor_state::blocked)) { actor_state::blocked)) {
case actor_state::ready: case actor_state::ready:
// interrupted by arriving message // interrupted by arriving message
...@@ -115,16 +128,7 @@ class event_based_resume { ...@@ -115,16 +128,7 @@ class event_based_resume {
else { else {
CPPA_LOG_DEBUG("switched back to ready: " CPPA_LOG_DEBUG("switched back to ready: "
"mailbox can fetch more"); "mailbox can fetch more");
this->set_state(actor_state::ready); d->set_state(actor_state::ready);
}
}
else {
if (this->invoke(e)) {
if (this->bhvr_stack().empty() && done_cb()) {
CPPA_LOG_DEBUG("behavior stack empty");
return resume_result::done;
}
this->bhvr_stack().cleanup();
} }
} }
} }
...@@ -132,22 +136,22 @@ class event_based_resume { ...@@ -132,22 +136,22 @@ class event_based_resume {
catch (actor_exited& what) { catch (actor_exited& what) {
CPPA_LOG_INFO("actor died because of exception: actor_exited, " CPPA_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason()); "reason = " << what.reason());
if (this->exit_reason() == exit_reason::not_exited) { if (d->exit_reason() == exit_reason::not_exited) {
this->quit(what.reason()); d->quit(what.reason());
} }
} }
catch (std::exception& e) { catch (std::exception& e) {
CPPA_LOG_WARNING("actor died because of exception: " CPPA_LOG_WARNING("actor died because of exception: "
<< detail::demangle(typeid(e)) << detail::demangle(typeid(e))
<< ", what() = " << e.what()); << ", what() = " << e.what());
if (this->exit_reason() == exit_reason::not_exited) { if (d->exit_reason() == exit_reason::not_exited) {
this->quit(exit_reason::unhandled_exception); d->quit(exit_reason::unhandled_exception);
} }
} }
catch (...) { catch (...) {
CPPA_LOG_WARNING("actor died because of an unknown exception"); CPPA_LOG_WARNING("actor died because of an unknown exception");
if (this->exit_reason() == exit_reason::not_exited) { if (d->exit_reason() == exit_reason::not_exited) {
this->quit(exit_reason::unhandled_exception); d->quit(exit_reason::unhandled_exception);
} }
} }
done_cb(); done_cb();
...@@ -168,6 +172,7 @@ class event_based_resume { ...@@ -168,6 +172,7 @@ class event_based_resume {
static_assert(std::is_same<Actor, Actor>::value == false, static_assert(std::is_same<Actor, Actor>::value == false,
"The event-based resume policy cannot be used " "The event-based resume policy cannot be used "
"to implement blocking actors"); "to implement blocking actors");
return false;
} }
}; };
......
...@@ -82,137 +82,45 @@ class invoke_policy { ...@@ -82,137 +82,45 @@ class invoke_policy {
public: public:
typedef mailbox_element* pointer; /**
typedef std::unique_ptr<mailbox_element, detail::disposer> smart_pointer; * @note @p node_ptr.release() is called whenever a message was
* handled or dropped.
*/
template<class Actor, class Fun> template<class Actor, class Fun>
bool invoke_from_cache(Actor* self, bool invoke_message(Actor* self,
Fun& fun, unique_mailbox_element_pointer& node_ptr,
message_id awaited_response = message_id{}) { Fun& fun,
auto i = m_cache.begin(); message_id awaited_response) {
auto e = m_cache.end(); if (!node_ptr) return false;
while (i != e) { bool result = false;
switch (handle_message(self, i->get(), fun, awaited_response)) { bool reset_pointer = true;
case hm_msg_handled: { switch (handle_message(self, node_ptr.get(), fun, awaited_response)) {
m_cache.erase(i);
return true;
}
case hm_drop_msg: {
i = m_cache.erase(i);
break;
}
case hm_skip_msg:
case hm_cache_msg: {
++i;
break;
}
default: {
CPPA_CRITICAL("illegal result of handle_message");
}
}
}
return false;
}
inline void add_to_cache(pointer node_ptr) {
m_cache.emplace_back(std::move(node_ptr));
}
template<class Actor, class Fun>
bool invoke(Actor* self,
pointer node_ptr,
Fun& fun,
message_id awaited_response = message_id()) {
smart_pointer node(node_ptr);
switch (handle_message(self, node.get(), fun, awaited_response)) {
case hm_msg_handled: { case hm_msg_handled: {
return true; result = true;
break;
} }
case hm_drop_msg: { case hm_drop_msg: {
break; break;
} }
case hm_cache_msg: { case hm_cache_msg: {
m_cache.emplace_back(std::move(node)); reset_pointer = false;
break; break;
} }
case hm_skip_msg: { case hm_skip_msg: {
CPPA_CRITICAL("received a marked node"); // "received" a marked node
reset_pointer = false;
break;
} }
default: { default: {
CPPA_CRITICAL("illegal result of handle_message"); CPPA_CRITICAL("illegal result of handle_message");
} }
} }
return false; if (reset_pointer) node_ptr.reset();
} return result;
template<class Actor, class FunOrBehavior>
inline void receive_wo_timeout(Actor* self, FunOrBehavior& fun) {
if (!invoke_from_cache(self, fun)) {
while (!invoke(self, self->await_message(), fun)) { }
}
}
template<class Actor>
void receive(Actor* self, partial_function& fun) {
receive_wo_timeout(self, fun);
}
template<class Actor>
void receive(Actor* self, behavior& bhvr) {
if (!bhvr.timeout().valid()) {
receive_wo_timeout(self, bhvr);
}
else if (!invoke_from_cache(self, bhvr)) {
if (bhvr.timeout().is_zero()) {
pointer e = nullptr;
while ((e = self->try_pop()) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(self, e, bhvr)) {
return; // done
}
}
dptr()->handle_timeout(self, bhvr);
}
else {
auto timeout = self->init_timeout(bhvr.timeout());
pointer e = nullptr;
while ((e = self->await_message(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(self, e, bhvr)) {
return; // done
}
}
dptr()->handle_timeout(self, bhvr);
}
}
}
template<class Actor>
void receive(Actor* self, behavior& bhvr, message_id mid) {
CPPA_REQUIRE(mid.is_response());
if (!invoke_from_cache(self, bhvr, mid)) {
if (bhvr.timeout().valid()) {
CPPA_REQUIRE(bhvr.timeout().is_zero() == false);
auto timeout = self->init_timeout(bhvr.timeout());
pointer e = nullptr;
while ((e = self->await_message(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(self, e, bhvr, mid)) {
return; // done
}
}
dptr()->handle_timeout(self, bhvr);
}
else while (!invoke(self, self->await_message(), bhvr, mid)) { }
}
}
template<class Actor>
mailbox_element* fetch_message(Actor* self) {
return self->await_message();
} }
typedef typename rp_flag<rp_nestable>::type nestable; typedef typename rp_flag<rp_nestable>::type nestable;
typedef typename rp_flag<rp_sequential>::type sequential; typedef typename rp_flag<rp_sequential>::type sequential;
private: private:
...@@ -241,7 +149,7 @@ class invoke_policy { ...@@ -241,7 +149,7 @@ class invoke_policy {
// - expired synchronous response messages // - expired synchronous response messages
template<class Actor> template<class Actor>
filter_result filter_msg(Actor* self, pointer node) { filter_result filter_msg(Actor* self, mailbox_element* node) {
const any_tuple& msg = node->msg; const any_tuple& msg = node->msg;
auto mid = node->mid; auto mid = node->mid;
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr; auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
...@@ -328,10 +236,8 @@ class invoke_policy { ...@@ -328,10 +236,8 @@ class invoke_policy {
// original request message // original request message
auto fhdl = fetch_response_promise(self, hdl); auto fhdl = fetch_response_promise(self, hdl);
if (ref_opt) { if (ref_opt) {
auto& ref = *ref_opt; behavior cpy = *ref_opt;
// copy original behavior *ref_opt = cpy.add_continuation(
behavior cpy = ref;
ref = cpy.add_continuation(
[=](any_tuple& intermediate) -> optional<any_tuple> { [=](any_tuple& intermediate) -> optional<any_tuple> {
if (!intermediate.empty()) { if (!intermediate.empty()) {
// do no use lamba expresion type to // do no use lamba expresion type to
...@@ -379,7 +285,7 @@ class invoke_policy { ...@@ -379,7 +285,7 @@ class invoke_policy {
template<class Actor, class Fun> template<class Actor, class Fun>
handle_message_result handle_message(Actor* self, handle_message_result handle_message(Actor* self,
pointer node, mailbox_element* node,
Fun& fun, Fun& fun,
message_id awaited_response) { message_id awaited_response) {
bool handle_sync_failure_on_mismatch = true; bool handle_sync_failure_on_mismatch = true;
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#ifndef CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP #ifndef CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP
#define CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP #define CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP
#include <utility>
#include "cppa/singletons.hpp" #include "cppa/singletons.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
...@@ -57,7 +59,7 @@ class middleman_scheduling { ...@@ -57,7 +59,7 @@ class middleman_scheduling {
: m_self(std::move(ptr)), m_hdr(hdr), m_data(std::move(msg)) { } : m_self(std::move(ptr)), m_hdr(hdr), m_data(std::move(msg)) { }
inline void operator()() const { inline void operator()() const {
m_self->invoke_message(m_hdr, move(m_data)); m_self->invoke_message(m_hdr, std::move(m_data));
} }
private: private:
...@@ -70,13 +72,6 @@ class middleman_scheduling { ...@@ -70,13 +72,6 @@ class middleman_scheduling {
using timeout_type = int; using timeout_type = int;
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
// this does return nullptr // this does return nullptr
template<class Actor, typename F> template<class Actor, typename F>
inline void fetch_messages(Actor*, F) { inline void fetch_messages(Actor*, F) {
......
...@@ -56,29 +56,29 @@ class nestable_invoke : public invoke_policy<nestable_invoke> { ...@@ -56,29 +56,29 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
public: public:
static inline bool hm_should_skip(pointer node) { static inline bool hm_should_skip(mailbox_element* node) {
return node->marked; return node->marked;
} }
template<class Actor> template<class Actor>
inline pointer hm_begin(Actor* self, pointer node) { inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
auto previous = self->m_current_node; auto previous = self->current_node();
self->m_current_node = node; self->current_node(node);
push_timeout(); push_timeout();
node->marked = true; node->marked = true;
return previous; return previous;
} }
template<class Actor> template<class Actor>
inline void hm_cleanup(Actor* self, pointer previous) { inline void hm_cleanup(Actor* self, mailbox_element* previous) {
self->m_current_node->marked = false; self->current_node()->marked = false;
self->m_current_node = previous; self->current_node(previous);
} }
template<class Actor> template<class Actor>
inline void hm_revert(Actor* self, pointer previous) { inline void hm_revert(Actor* self, mailbox_element* previous) {
self->m_current_node->marked = false; self->current_node()->marked = false;
self->m_current_node = previous; self->current_node(previous);
pop_timeout(); pop_timeout();
} }
...@@ -100,19 +100,6 @@ class nestable_invoke : public invoke_policy<nestable_invoke> { ...@@ -100,19 +100,6 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
inline bool waits_for_timeout(std::uint32_t) { return false; } inline bool waits_for_timeout(std::uint32_t) { return false; }
/*
void run_detached() {
auto dthis = util::dptr<Subtype>(this);
dthis->init();
if (dthis->planned_exit_reason() != exit_reason::not_exited) {
// init() did indeed call quit() for some reason
dthis->on_exit();
}
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
*/
}; };
} } // namespace cppa::policy } } // namespace cppa::policy
......
...@@ -2,6 +2,7 @@ ...@@ -2,6 +2,7 @@
#define NO_RESUME_HPP #define NO_RESUME_HPP
#include <chrono> #include <chrono>
#include <utility>
#include "cppa/policy/resume_policy.hpp" #include "cppa/policy/resume_policy.hpp"
...@@ -20,7 +21,7 @@ class no_resume { ...@@ -20,7 +21,7 @@ class no_resume {
public: public:
template<class Base> template<class Base, class Derived>
struct mixin : Base { struct mixin : Base {
template<typename... Ts> template<typename... Ts>
...@@ -33,13 +34,8 @@ class no_resume { ...@@ -33,13 +34,8 @@ class no_resume {
}; };
template<class Actor> template<class Actor>
void await_data(Actor* self) { void await_ready(Actor* self) {
self->scheduling_policy().await_data(self); self->await_data();
}
template<class Actor, typename AbsTimeout>
bool await_data(Actor* self, const AbsTimeout& abs_time) {
return self->scheduling_policy().await_data(self, abs_time);
} }
}; };
......
...@@ -62,30 +62,15 @@ class no_scheduling { ...@@ -62,30 +62,15 @@ class no_scheduling {
typedef std::chrono::high_resolution_clock::time_point timeout_type; typedef std::chrono::high_resolution_clock::time_point timeout_type;
template<class Actor>
inline timeout_type init_timeout(Actor*, const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
return result;
}
template<class Actor, typename F> template<class Actor, typename F>
bool fetch_messages(Actor* self, F cb) { bool fetch_messages(Actor* self, F cb) {
await_data(self); await_data(self);
fetch_messages_impl(self, cb); return fetch_messages_impl(self, cb);
} }
template<class Actor, typename F> template<class Actor, typename F>
bool try_fetch_messages(Actor* self, F cb) { bool try_fetch_messages(Actor* self, F cb) {
auto next = [&] { return self->mailbox().try_pop(); }; return fetch_messages_impl(self, cb);
auto e = next();
if (!e) return false;
do {
cb(e);
e = next();
}
while (e);
return true;
} }
template<class Actor, typename F> template<class Actor, typename F>
...@@ -93,8 +78,8 @@ class no_scheduling { ...@@ -93,8 +78,8 @@ class no_scheduling {
if (!await_data(self, abs_time)) { if (!await_data(self, abs_time)) {
return timed_fetch_result::no_message; return timed_fetch_result::no_message;
} }
fetch_messages_impl(self, cb); if (fetch_messages_impl(self, cb)) return timed_fetch_result::success;
return timed_fetch_result::success; return timed_fetch_result::no_message;
} }
template<class Actor> template<class Actor>
...@@ -130,13 +115,8 @@ class no_scheduling { ...@@ -130,13 +115,8 @@ class no_scheduling {
}); });
util::fiber fself; util::fiber fself;
for (;;) { for (;;) {
try { await_data(self); } await_data(self);
catch (std::exception& e) { self->set_state(actor_state::ready);
std::cerr << detail::demangle(typeid(e)) << ", what: "
<< e.what() << std::endl;
throw;
}
if (self->resume(&fself) == resumable::done) { if (self->resume(&fself) == resumable::done) {
CPPA_LOG_DEBUG("resume returned resumable::done"); CPPA_LOG_DEBUG("resume returned resumable::done");
self->planned_exit_reason(exit_reason::normal); self->planned_exit_reason(exit_reason::normal);
...@@ -153,19 +133,20 @@ class no_scheduling { ...@@ -153,19 +133,20 @@ class no_scheduling {
template<class Actor> template<class Actor>
void await_data(Actor* self) { void await_data(Actor* self) {
while (self->mailbox().empty()) { if (!self->has_next_message()) {
lock_type guard(m_mtx); lock_type guard(m_mtx);
while (self->mailbox().empty()) m_cv.wait(guard); while (!self->has_next_message()) m_cv.wait(guard);
} }
} }
template<class Actor> // this additional member function is needed to implement
bool await_data(Actor* self, const timeout_type& abs_time) { // timer_actor (see scheduler.cpp)
CPPA_REQUIRE(!self->mailbox().closed()); template<class Actor, class TimePoint>
while (self->mailbox().empty()) { bool await_data(Actor* self, const TimePoint& tp) {
if (!self->has_next_message()) {
lock_type guard(m_mtx); lock_type guard(m_mtx);
while (self->mailbox().empty()) { while (!self->has_next_message()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) { if (m_cv.wait_until(guard, tp) == std::cv_status::timeout) {
return false; return false;
} }
} }
...@@ -176,12 +157,16 @@ class no_scheduling { ...@@ -176,12 +157,16 @@ class no_scheduling {
private: private:
template<class Actor, typename F> template<class Actor, typename F>
void fetch_messages_impl(Actor* self, F cb) { bool fetch_messages_impl(Actor* self, F cb) {
auto next = [&] { return self->mailbox().try_pop(); }; auto next = [&] { return self->mailbox().try_pop(); };
for (auto e = next(); e != nullptr; e = next()) { auto e = next();
cb(e); if (e) {
for (; e != nullptr; e = next()) {
cb(e);
}
return true;
} }
return false;
} }
std::mutex m_mtx; std::mutex m_mtx;
......
...@@ -38,21 +38,44 @@ ...@@ -38,21 +38,44 @@
namespace cppa { namespace cppa {
namespace policy { namespace policy {
struct not_prioritizing { class not_prioritizing {
public:
typedef std::vector<unique_mailbox_element_pointer> cache_type;
typedef cache_type::iterator cache_iterator;
template<class Actor> template<class Actor>
mailbox_element* next_message(Actor* self) { unique_mailbox_element_pointer next_message(Actor* self) {
return self->mailbox().try_pop(); return unique_mailbox_element_pointer{self->mailbox().try_pop()};
}
template<class Actor>
inline bool has_next_message(Actor* self) {
return self->mailbox().can_fetch_more();
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
m_cache.push_back(std::move(ptr));
}
inline cache_iterator cache_begin() {
return m_cache.begin();
} }
template<class Actor, typename F> inline cache_iterator cache_end() {
bool fetch_messages(Actor* self, F cb) { return m_cache.begin();
auto fetch = [self] { return self->mailbox().try_pop(); };
for (auto e = fetch(); e != nullptr; e = fetch()) {
cb(e);
}
} }
inline void cache_erase(cache_iterator iter) {
m_cache.erase(iter);
}
private:
cache_type m_cache;
}; };
} // namespace policy } // namespace policy
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_POLICY_POLICIES_HPP
#define CPPA_POLICY_POLICIES_HPP
namespace cppa {
namespace policy {
template<class SchedulingPolicy, class PriorityPolicy,
class ResumePolicy, class InvokePolicy>
class policies {
public:
typedef SchedulingPolicy scheduling_policy;
typedef PriorityPolicy priority_policy;
typedef ResumePolicy resume_policy;
typedef InvokePolicy invoke_policy;
scheduling_policy& get_scheduling_policy() {
return m_scheduling_policy;
}
priority_policy& get_priority_policy() {
return m_priority_policy;
}
resume_policy& get_resume_policy() {
return m_resume_policy;
}
invoke_policy& get_invoke_policy() {
return m_invoke_policy;
}
private:
scheduling_policy m_scheduling_policy;
priority_policy m_priority_policy;
resume_policy m_resume_policy;
invoke_policy m_invoke_policy;
};
} // namespace policy
} // namespace cppa
#endif // CPPA_POLICY_POLICIES_HPP
...@@ -43,11 +43,40 @@ class prioritizing { ...@@ -43,11 +43,40 @@ class prioritizing {
public: public:
template<typename Actor> typedef std::vector<unique_mailbox_element_pointer> cache_type;
mailbox_element* next_message(Actor* self) {
return self->m_mailbox.try_pop(); typedef cache_type::iterator cache_iterator;
template<class Actor>
unique_mailbox_element_pointer next_message(Actor* self) {
return unique_mailbox_element_pointer{self->mailbox().try_pop()};
}
template<class Actor>
inline bool has_next_message(Actor* self) {
return self->mailbox().can_fetch_more();
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
m_cache.push_back(std::move(ptr));
}
inline cache_iterator cache_begin() {
return m_cache.begin();
}
inline cache_iterator cache_end() {
return m_cache.begin();
} }
inline void cache_erase(cache_iterator iter) {
m_cache.erase(iter);
}
private:
cache_type m_cache;
/* /*
mailbox_element* try_pop() { mailbox_element* try_pop() {
auto result = m_high_priority_mailbox.try_pop(); auto result = m_high_priority_mailbox.try_pop();
......
...@@ -50,20 +50,22 @@ class priority_policy { ...@@ -50,20 +50,22 @@ class priority_policy {
* left in the cache. * left in the cache.
*/ */
template<class Actor> template<class Actor>
mailbox_element* next_message(Actor* self); unique_mailbox_element_pointer next_message(Actor* self);
/** template<class Actor>
* @brief Fetches new messages from the actor's mailbox. The member bool has_next_message(Actor* self);
* function returns @p false if no message was read,
* @p true otherwise. void push_to_cache(unique_mailbox_element_pointer ptr);
*
* This member function calls {@link scheduling_policy::fetch_messages}, typedef std::vector<unique_mailbox_element_pointer> cache_type;
* so a returned @p false value indicates that the client must prepare
* for re-scheduling. typedef cache_type::iterator cache_iterator;
*/
template<class Actor, typename F> cache_iterator cache_begin();
bool fetch_messages(Actor* self, F cb);
cache_iterator cache_end();
void cache_erase(cache_iterator iter);
}; };
......
...@@ -38,7 +38,8 @@ ...@@ -38,7 +38,8 @@
namespace cppa { namespace util { class duration; struct fiber; } } namespace cppa { namespace util { class duration; struct fiber; } }
namespace cppa { namespace policy { namespace cppa {
namespace policy {
/** /**
* @brief The resume_policy <b>concept</b> class. Please note that this * @brief The resume_policy <b>concept</b> class. Please note that this
...@@ -59,26 +60,18 @@ class resume_policy { ...@@ -59,26 +60,18 @@ class resume_policy {
resumable::resume_result resume(Actor* self, util::fiber* from); resumable::resume_result resume(Actor* self, util::fiber* from);
/** /**
* @brief Waits unconditionally until a new message arrives. * @brief Waits unconditionally until the actor is ready to resume.
* @note This member function must raise a compiler error if the resume * @note This member function must raise a compiler error if the resume
* strategy cannot be used to implement blocking actors. * strategy cannot be used to implement blocking actors.
*
* This member function calls {@link scheduling_policy::await_data}
*/ */
template<class Actor> template<class Actor>
bool await_data(Actor* self); bool await_ready(Actor* self);
/**
* @brief Waits until either a new message arrives, or a timeout occurs.
* The @p abs_time argument is the return value of
* {@link scheduling_policy::init_timeout}. Returns true if a
* message arrived in time, otherwise false.
* @note This member function must raise a compiler error if the resume
* strategy cannot be used to implement blocking actors.
*/
template<class Actor, typename AbsTimeout>
bool await_data(Actor* self, const AbsTimeout& abs_time);
}; };
} } // namespace cppa::policy } // namespace policy
} // namespace cppa
#endif // CPPA_RESUME_POLICY_HPP #endif // CPPA_RESUME_POLICY_HPP
...@@ -59,12 +59,6 @@ class scheduling_policy { ...@@ -59,12 +59,6 @@ class scheduling_policy {
*/ */
using timeout_type = int; using timeout_type = int;
/**
* @brief Requests a timeout for the next call to @p fetch_message.
*/
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time);
/** /**
* @brief Fetches new messages from the actor's mailbox and feeds them * @brief Fetches new messages from the actor's mailbox and feeds them
* to the given callback. The member function returns @p false if * to the given callback. The member function returns @p false if
......
...@@ -49,24 +49,24 @@ class sequential_invoke : public invoke_policy<sequential_invoke> { ...@@ -49,24 +49,24 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
sequential_invoke() : m_has_pending_tout(false), m_pending_tout(0) { } sequential_invoke() : m_has_pending_tout(false), m_pending_tout(0) { }
static inline bool hm_should_skip(pointer) { static inline bool hm_should_skip(mailbox_element*) {
return false; return false;
} }
template<class Actor> template<class Actor>
static inline pointer hm_begin(Actor* self, pointer node) { static inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
auto previous = self->current_node(); auto previous = self->current_node();
self->current_node(node); self->current_node(node);
return previous; return previous;
} }
template<class Actor> template<class Actor>
static inline void hm_cleanup(Actor* self, pointer) { static inline void hm_cleanup(Actor* self, mailbox_element*) {
self->current_node(&(self->m_dummy_node)); self->current_node(self->dummy_node());
} }
template<class Actor> template<class Actor>
static inline void hm_revert(Actor* self, pointer previous) { static inline void hm_revert(Actor* self, mailbox_element* previous) {
self->current_node(previous); self->current_node(previous);
} }
......
...@@ -54,7 +54,6 @@ namespace cppa { ...@@ -54,7 +54,6 @@ namespace cppa {
class untyped_actor; class untyped_actor;
class scheduled_actor; class scheduled_actor;
class scheduler_helper; class scheduler_helper;
class untyped_actor;
typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr; typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr;
namespace detail { class singleton_manager; } // namespace detail namespace detail { class singleton_manager; } // namespace detail
......
...@@ -95,11 +95,11 @@ actor spawn(Ts&&... args) { ...@@ -95,11 +95,11 @@ actor spawn(Ts&&... args) {
policy::nestable_invoke, policy::nestable_invoke,
policy::sequential_invoke policy::sequential_invoke
>::type; >::type;
using proper_impl = detail::proper_actor<Impl, using policies = policy::policies<scheduling_policy,
scheduling_policy, priority_policy,
priority_policy, resume_policy,
resume_policy, invoke_policy>;
invoke_policy>; using proper_impl = detail::proper_actor<Impl, policies>;
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...); auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
ptr->launch(); ptr->launch();
return ptr; return ptr;
......
...@@ -70,7 +70,7 @@ optional<behavior&> behavior_stack::sync_handler(message_id expected_response) { ...@@ -70,7 +70,7 @@ optional<behavior&> behavior_stack::sync_handler(message_id expected_response) {
}); });
if (i != e) return i->first; if (i != e) return i->first;
} }
return {}; return none;
} }
void behavior_stack::pop_async_back() { void behavior_stack::pop_async_back() {
......
...@@ -287,6 +287,7 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) { ...@@ -287,6 +287,7 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
m_dummy_node.msg = move(msg); m_dummy_node.msg = move(msg);
m_dummy_node.mid = hdr.id; m_dummy_node.mid = hdr.id;
try { try {
/*TODO:
auto bhvr = m_bhvr_stack.back(); auto bhvr = m_bhvr_stack.back();
switch (m_invoke.handle_message(this, switch (m_invoke.handle_message(this,
&m_dummy_node, &m_dummy_node,
...@@ -295,7 +296,7 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) { ...@@ -295,7 +296,7 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
case policy::hm_msg_handled: { case policy::hm_msg_handled: {
if ( not m_bhvr_stack.empty() if ( not m_bhvr_stack.empty()
&& bhvr.as_behavior_impl() == m_bhvr_stack.back().as_behavior_impl()) { && bhvr.as_behavior_impl() == m_bhvr_stack.back().as_behavior_impl()) {
m_invoke.invoke_from_cache(this, bhvr, m_bhvr_stack.back_id()); //TODO: m_invoke.invoke_from_cache(this, bhvr, m_bhvr_stack.back_id());
} }
break; break;
} }
...@@ -304,11 +305,12 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) { ...@@ -304,11 +305,12 @@ void broker::invoke_message(const message_header& hdr, any_tuple msg) {
case policy::hm_skip_msg: case policy::hm_skip_msg:
case policy::hm_cache_msg: { case policy::hm_cache_msg: {
auto e = mailbox_element::create(hdr, move(m_dummy_node.msg)); auto e = mailbox_element::create(hdr, move(m_dummy_node.msg));
m_invoke.add_to_cache(e); //TODO: m_invoke.add_to_cache(e);
break; break;
} }
default: CPPA_CRITICAL("illegal result of handle_message"); default: CPPA_CRITICAL("illegal result of handle_message");
} }
*/
} }
catch (std::exception& e) { catch (std::exception& e) {
CPPA_LOG_ERROR("broker killed due to an unhandled exception: " CPPA_LOG_ERROR("broker killed due to an unhandled exception: "
......
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <string.h> #include <string.h>
#include <sys/epoll.h> #include <sys/epoll.h>
#include "cppa/logging.hpp"
#include "cppa/io/middleman_event_handler.hpp" #include "cppa/io/middleman_event_handler.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
...@@ -139,18 +140,18 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -139,18 +140,18 @@ class middleman_event_handler_impl : public middleman_event_handler {
switch (errno) { switch (errno) {
// supplied file descriptor is already registered // supplied file descriptor is already registered
case EEXIST: { case EEXIST: {
CPPA_LOGMF(CPPA_ERROR, self, "file descriptor registered twice"); CPPA_LOG_ERROR("file descriptor registered twice");
break; break;
} }
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL, // op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance. // and fd is not registered with this epoll instance.
case ENOENT: { case ENOENT: {
CPPA_LOGMF(CPPA_ERROR, self, "cannot delete file descriptor " CPPA_LOG_ERROR("cannot delete file descriptor "
"because it isn't registered"); "because it isn't registered");
break; break;
} }
default: { default: {
CPPA_LOGMF(CPPA_ERROR, self, strerror(errno)); CPPA_LOG_ERROR(strerror(errno));
perror("epoll_ctl() failed"); perror("epoll_ctl() failed");
CPPA_CRITICAL("epoll_ctl() failed"); CPPA_CRITICAL("epoll_ctl() failed");
} }
......
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <iostream> #include <iostream>
#include "cppa/on.hpp" #include "cppa/on.hpp"
#include "cppa/policy.hpp"
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/anything.hpp" #include "cppa/anything.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
...@@ -41,6 +42,7 @@ ...@@ -41,6 +42,7 @@
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/scoped_actor.hpp" #include "cppa/scoped_actor.hpp"
#include "cppa/detail/proper_actor.hpp"
#include "cppa/detail/actor_registry.hpp" #include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp" #include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp" #include "cppa/detail/thread_pool_scheduler.hpp"
...@@ -53,12 +55,18 @@ typedef std::uint32_t ui32; ...@@ -53,12 +55,18 @@ typedef std::uint32_t ui32;
typedef std::chrono::high_resolution_clock hrc; typedef std::chrono::high_resolution_clock hrc;
typedef hrc::time_point time_point;
struct exit_observer : cppa::attachable { struct exit_observer : cppa::attachable {
~exit_observer() { get_actor_registry()->dec_running(); } ~exit_observer() { get_actor_registry()->dec_running(); }
void actor_exited(std::uint32_t) { } void actor_exited(std::uint32_t) { }
bool matches(const token&) { return false; } bool matches(const token&) { return false; }
}; };
typedef policy::policies<policy::no_scheduling, policy::not_prioritizing,
policy::no_resume, policy::nestable_invoke>
timer_actor_policies;
class delayed_msg { class delayed_msg {
public: public:
...@@ -83,13 +91,89 @@ class delayed_msg { ...@@ -83,13 +91,89 @@ class delayed_msg {
}; };
template<class Map>
inline void insert_dmsg(Map& storage, const util::duration& d,
message_header&& hdr, any_tuple&& tup) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{move(hdr), move(tup)};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
class timer_actor final : public detail::proper_actor<blocking_untyped_actor,
timer_actor_policies> {
public:
inline unique_mailbox_element_pointer dequeue() {
await_data();
return next_message();
}
inline unique_mailbox_element_pointer try_dequeue(const time_point& tp) {
if (scheduling_policy().await_data(this, tp)) {
return next_message();
}
return unique_mailbox_element_pointer{};
}
void act() override {
// setup & local variables
bool done = false;
unique_mailbox_element_pointer msg_ptr;
auto tout = hrc::now();
std::multimap<decltype(tout), delayed_msg> messages;
// message handling rules
auto mfun = (
on(atom("SEND"), arg_match) >> [&](const util::duration& d,
message_header& hdr,
any_tuple& tup) {
insert_dmsg(messages, d, move(hdr), move(tup));
},
on(atom("DIE")) >> [&] {
done = true;
},
others() >> [&]() {
# ifdef CPPA_DEBUG_MODE
std::cerr << "scheduler_helper::timer_loop: UNKNOWN MESSAGE: "
<< to_string(msg_ptr->msg)
<< std::endl;
# endif
}
);
// loop
while (!done) {
while (!msg_ptr) {
if (messages.empty()) msg_ptr = dequeue();
else {
tout = hrc::now();
// handle timeouts (send messages)
auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) {
it->second.eval();
messages.erase(it);
it = messages.begin();
}
// wait for next message or next timeout
if (it != messages.end()) {
msg_ptr = try_dequeue(it->first);
}
}
}
mfun(msg_ptr->msg);
msg_ptr.reset();
}
}
};
} // namespace <anonymous> } // namespace <anonymous>
class scheduler_helper { class scheduler_helper {
public: public:
scheduler_helper() : m_timer(true), m_printer(true) { } scheduler_helper() : m_timer(new timer_actor), m_printer(true) { }
void start() { void start() {
// launch threads // launch threads
...@@ -105,7 +189,7 @@ class scheduler_helper { ...@@ -105,7 +189,7 @@ class scheduler_helper {
m_printer_thread.join(); m_printer_thread.join();
} }
scoped_actor m_timer; intrusive_ptr<timer_actor> m_timer;
std::thread m_timer_thread; std::thread m_timer_thread;
scoped_actor m_printer; scoped_actor m_printer;
...@@ -113,69 +197,14 @@ class scheduler_helper { ...@@ -113,69 +197,14 @@ class scheduler_helper {
private: private:
static void timer_loop(blocking_untyped_actor* self); static void timer_loop(timer_actor* self);
static void printer_loop(blocking_untyped_actor* self); static void printer_loop(blocking_untyped_actor* self);
}; };
template<class Map> void scheduler_helper::timer_loop(timer_actor* self) {
inline void insert_dmsg(Map& storage, self->act();
const util::duration& d,
message_header&& hdr,
any_tuple&& tup ) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{move(hdr), move(tup)};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
void scheduler_helper::timer_loop(blocking_untyped_actor* self) {
// setup & local variables
bool done = false;
std::unique_ptr<mailbox_element, detail::disposer> msg_ptr;
auto tout = hrc::now();
std::multimap<decltype(tout), delayed_msg> messages;
// message handling rules
auto mfun = (
on(atom("SEND"), arg_match) >> [&](const util::duration& d,
message_header& hdr,
any_tuple& tup) {
insert_dmsg(messages, d, move(hdr), move(tup));
},
on(atom("DIE")) >> [&] {
done = true;
},
others() >> [&]() {
# ifdef CPPA_DEBUG_MODE
std::cerr << "scheduler_helper::timer_loop: UNKNOWN MESSAGE: "
<< to_string(msg_ptr->msg)
<< std::endl;
# endif
}
);
// loop
while (!done) {
while (!msg_ptr) {
if (messages.empty()) msg_ptr.reset(self->dequeue());
else {
tout = hrc::now();
// handle timeouts (send messages)
auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) {
it->second.eval();
messages.erase(it);
it = messages.begin();
}
// wait for next message or next timeout
if (it != messages.end()) {
msg_ptr.reset(self->try_dequeue(it->first));
}
}
}
mfun(msg_ptr->msg);
msg_ptr.reset();
}
} }
void scheduler_helper::printer_loop(blocking_untyped_actor* self) { void scheduler_helper::printer_loop(blocking_untyped_actor* self) {
......
...@@ -46,10 +46,10 @@ struct impl : blocking_untyped_actor { ...@@ -46,10 +46,10 @@ struct impl : blocking_untyped_actor {
blocking_untyped_actor* alloc() { blocking_untyped_actor* alloc() {
using namespace policy; using namespace policy;
return new detail::proper_actor<impl, return new detail::proper_actor<impl,
no_scheduling, policies<no_scheduling,
not_prioritizing, not_prioritizing,
no_resume, no_resume,
nestable_invoke>; nestable_invoke>>;
} }
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -38,10 +38,8 @@ namespace cppa { ...@@ -38,10 +38,8 @@ namespace cppa {
continue_helper& continue_helper::continue_with(behavior::continuation_fun fun) { continue_helper& continue_helper::continue_with(behavior::continuation_fun fun) {
auto ref_opt = m_self->bhvr_stack().sync_handler(m_mid); auto ref_opt = m_self->bhvr_stack().sync_handler(m_mid);
if (ref_opt) { if (ref_opt) {
auto& ref = *ref_opt; behavior cpy = *ref_opt;
// copy original behavior *ref_opt = cpy.add_continuation(std::move(fun));
behavior cpy = ref;
ref = cpy.add_continuation(std::move(fun));
} }
else CPPA_LOG_ERROR("failed to add continuation"); else CPPA_LOG_ERROR("failed to add continuation");
return *this; return *this;
......
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