Commit 7e757380 authored by Dominik Charousset's avatar Dominik Charousset

maintenance

parent fd309798
...@@ -59,15 +59,15 @@ class context_switching_actor : public scheduled_actor { ...@@ -59,15 +59,15 @@ class context_switching_actor : public scheduled_actor {
/** /**
* @brief Implements the actor's behavior. * @brief Implements the actor's behavior.
* Reimplemented this function for a class-based context-switching * Reimplemented this function for a class-based actor.
* actor. Returning from this member function will end the * Returning from this member function will end the
* execution of the actor. * execution of the actor.
*/ */
virtual void run(); virtual void run();
}; };
#else #else // CPPA_DOCUMENTATION
class context_switching_actor : public detail::stacked_actor_mixin< class context_switching_actor : public detail::stacked_actor_mixin<
context_switching_actor, context_switching_actor,
...@@ -75,6 +75,10 @@ class context_switching_actor : public detail::stacked_actor_mixin< ...@@ -75,6 +75,10 @@ class context_switching_actor : public detail::stacked_actor_mixin<
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::stacked_actor_mixin<
context_switching_actor,
detail::abstract_scheduled_actor> super;
public: public:
context_switching_actor(std::function<void()> fun); context_switching_actor(std::function<void()> fun);
...@@ -85,8 +89,6 @@ class context_switching_actor : public detail::stacked_actor_mixin< ...@@ -85,8 +89,6 @@ class context_switching_actor : public detail::stacked_actor_mixin<
context_switching_actor(); context_switching_actor();
virtual void run();
private: private:
// required by detail::nestable_receive_policy // required by detail::nestable_receive_policy
...@@ -110,9 +112,6 @@ class context_switching_actor : public detail::stacked_actor_mixin< ...@@ -110,9 +112,6 @@ class context_switching_actor : public detail::stacked_actor_mixin<
// members // members
util::fiber m_fiber; util::fiber m_fiber;
std::function<void()> m_behavior;
detail::receive_policy m_recv_policy;
std::unique_ptr<detail::behavior_stack> m_bhvr_stack_ptr;
}; };
......
...@@ -32,9 +32,11 @@ ...@@ -32,9 +32,11 @@
#define CPPA_STACKED_ACTOR_MIXIN_HPP #define CPPA_STACKED_ACTOR_MIXIN_HPP
#include <memory> #include <memory>
#include <functional>
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
...@@ -43,19 +45,12 @@ namespace cppa { namespace detail { ...@@ -43,19 +45,12 @@ namespace cppa { namespace detail {
template<class Derived, class Base> template<class Derived, class Base>
class stacked_actor_mixin : public Base { class stacked_actor_mixin : public Base {
inline Derived* dthis() {
return static_cast<Derived*>(this);
}
public: public:
virtual void unbecome() { virtual void unbecome() {
if (m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back(); m_bhvr_stack_ptr->pop_back();
} }
else if (this->initialized()) {
this->quit();
}
} }
virtual void dequeue(partial_function& fun) { virtual void dequeue(partial_function& fun) {
...@@ -66,25 +61,47 @@ class stacked_actor_mixin : public Base { ...@@ -66,25 +61,47 @@ class stacked_actor_mixin : public Base {
m_recv_policy.receive(dthis(), bhvr); m_recv_policy.receive(dthis(), bhvr);
} }
virtual void run() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->exec();
m_bhvr_stack_ptr.reset();
}
if (m_behavior) {
m_behavior();
}
}
protected: protected:
receive_policy m_recv_policy; stacked_actor_mixin() = default;
std::unique_ptr<behavior_stack> m_bhvr_stack_ptr;
stacked_actor_mixin(std::function<void()> f) : m_behavior(std::move(f)) { }
virtual void do_become(behavior* bhvr, bool owns_bhvr, bool discard_old) { virtual void do_become(behavior* bhvr, bool owns_bhvr, bool discard_old) {
if (this->m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
if (discard_old) this->m_bhvr_stack_ptr->pop_back(); if (discard_old) m_bhvr_stack_ptr->pop_back();
this->m_bhvr_stack_ptr->push_back(bhvr, owns_bhvr); m_bhvr_stack_ptr->push_back(bhvr, owns_bhvr);
} }
else { else {
this->m_bhvr_stack_ptr.reset(new behavior_stack); m_bhvr_stack_ptr.reset(new behavior_stack);
m_bhvr_stack_ptr->push_back(bhvr, owns_bhvr);
if (this->initialized()) { if (this->initialized()) {
this->m_bhvr_stack_ptr->exec(); m_bhvr_stack_ptr->exec();
this->quit(); m_bhvr_stack_ptr.reset();
} }
} }
} }
private:
std::function<void()> m_behavior;
receive_policy m_recv_policy;
std::unique_ptr<behavior_stack> m_bhvr_stack_ptr;
inline Derived* dthis() {
return static_cast<Derived*>(this);
}
}; };
} } // namespace cppa::detail } } // namespace cppa::detail
......
...@@ -248,6 +248,21 @@ class local_actor : public actor { ...@@ -248,6 +248,21 @@ class local_actor : public actor {
*/ */
virtual void unbecome() = 0; virtual void unbecome() = 0;
/**
* @brief Can be overridden to initialize an actor before any
* message is handled.
* @warning Must not call blocking functions.
*/
virtual void init();
/**
* @brief Can be overridden to perform cleanup code after an actor
* finished execution.
* @warning Must not call any function manipulating the actor's state such
* as join, leave, link, or monitor.
*/
virtual void on_exit();
// library-internal members and member functions that shall // library-internal members and member functions that shall
// not appear in the documentation // not appear in the documentation
......
...@@ -58,20 +58,6 @@ class scheduled_actor : public local_actor { ...@@ -58,20 +58,6 @@ class scheduled_actor : public local_actor {
*/ */
scheduled_actor* next; scheduled_actor* next;
/**
* @brief Can be overridden to perform cleanup code after an actor
* finished execution.
* @warning Must not call any function manipulating the actor's state such
* as join, leave, link, or monitor.
*/
virtual void on_exit();
/**
* @brief Can be overridden to initialize and actor before any
* message is handled.
*/
virtual void init();
// called from worker thread // called from worker thread
virtual resume_result resume(util::fiber* from) = 0; virtual resume_result resume(util::fiber* from) = 0;
......
...@@ -63,12 +63,21 @@ namespace cppa { ...@@ -63,12 +63,21 @@ namespace cppa {
/** /**
* @brief An actor running in its own thread. * @brief An actor running in its own thread.
*/ */
class thread_mapped_actor : public local_actor { }; class thread_mapped_actor : public local_actor {
protected:
#else // CPPA_DOCUMENTATION /**
* @brief Implements the actor's behavior.
* Reimplemented this function for a class-based actor.
* Returning from this member function will end the
* execution of the actor.
*/
virtual void run();
class self_type; };
#else // CPPA_DOCUMENTATION
class thread_mapped_actor : public detail::stacked_actor_mixin< class thread_mapped_actor : public detail::stacked_actor_mixin<
thread_mapped_actor, thread_mapped_actor,
...@@ -76,8 +85,16 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -76,8 +85,16 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::stacked_actor_mixin<
thread_mapped_actor,
detail::abstract_actor<local_actor> > super;
public: public:
thread_mapped_actor();
thread_mapped_actor(std::function<void()> fun);
void quit(std::uint32_t reason = exit_reason::normal); //override void quit(std::uint32_t reason = exit_reason::normal); //override
void enqueue(actor* sender, any_tuple msg); //override void enqueue(actor* sender, any_tuple msg); //override
...@@ -86,12 +103,16 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -86,12 +103,16 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
inline decltype(m_mailbox)& mailbox() { return m_mailbox; } inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
inline void initialized(bool value) { m_initialized = value; }
protected: protected:
bool initialized(); bool initialized();
private: private:
bool m_initialized;
// required by nestable_receive_policy // required by nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_nestable; static const detail::receive_policy_flag receive_flag = detail::rp_nestable;
inline void push_timeout() { } inline void push_timeout() { }
......
...@@ -43,16 +43,7 @@ context_switching_actor::context_switching_actor() ...@@ -43,16 +43,7 @@ context_switching_actor::context_switching_actor()
} }
context_switching_actor::context_switching_actor(std::function<void()> fun) context_switching_actor::context_switching_actor(std::function<void()> fun)
: m_fiber(&context_switching_actor::trampoline, this), m_behavior(fun) { : super(std::move(fun)), m_fiber(&context_switching_actor::trampoline, this) {
}
void context_switching_actor::run() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->exec();
}
else if (m_behavior) {
m_behavior();
}
} }
void context_switching_actor::trampoline(void* this_ptr) { void context_switching_actor::trampoline(void* this_ptr) {
......
...@@ -80,4 +80,8 @@ void local_actor::demonitor(actor_ptr whom) { ...@@ -80,4 +80,8 @@ void local_actor::demonitor(actor_ptr whom) {
if (whom) whom->detach(mtoken); if (whom) whom->detach(mtoken);
} }
void local_actor::on_exit() { }
void local_actor::init() { }
} // namespace cppa } // namespace cppa
...@@ -51,38 +51,46 @@ using std::cout; ...@@ -51,38 +51,46 @@ using std::cout;
using std::cerr; using std::cerr;
using std::endl; using std::endl;
namespace cppa { namespace detail {
typedef intrusive_ptr<thread_mapped_actor> thread_mapped_actor_ptr;
namespace { namespace {
void run_actor(cppa::intrusive_ptr<cppa::local_actor> m_self, void run_actor(thread_mapped_actor_ptr m_self) {
std::function<void()> what) { self.set(m_self.get());
cppa::self.set(m_self.get()); try {
try { what(); } m_self->init();
m_self->initialized(true);
m_self->run();
m_self->on_exit();
}
catch (...) { } catch (...) { }
cppa::self.set(nullptr); self.set(nullptr);
cppa::detail::dec_actor_count(); std::atomic_thread_fence(std::memory_order_seq_cst);
dec_actor_count();
} }
void run_hidden_actor(cppa::intrusive_ptr<cppa::local_actor> m_self, void run_hidden_actor(local_actor_ptr m_self,
std::function<void()> what) { std::function<void()> what) {
cppa::self.set(m_self.get()); self.set(m_self.get());
try { what(); } try { what(); }
catch (...) { } catch (...) { }
cppa::self.set(nullptr); self.set(nullptr);
} }
} // namespace <anonymous> } // namespace <anonymous>
namespace cppa { namespace detail { std::thread mock_scheduler::spawn_hidden_impl(std::function<void()> what,
local_actor_ptr ctx) {
std::thread mock_scheduler::spawn_hidden_impl(std::function<void()> what, local_actor_ptr ctx) { return std::thread{run_hidden_actor, std::move(ctx), std::move(what)};
return std::thread{run_hidden_actor, ctx, std::move(what)};
} }
actor_ptr mock_scheduler::spawn_impl(std::function<void()> what) { actor_ptr mock_scheduler::spawn_impl(std::function<void()> what) {
inc_actor_count(); inc_actor_count();
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
intrusive_ptr<local_actor> ctx{new thread_mapped_actor}; thread_mapped_actor_ptr ctx{new thread_mapped_actor(std::move(what))};
std::thread{run_actor, ctx, std::move(what)}.detach(); std::thread{run_actor, ctx}.detach();
return std::move(ctx); return std::move(ctx);
} }
......
...@@ -36,10 +36,6 @@ namespace cppa { ...@@ -36,10 +36,6 @@ namespace cppa {
scheduled_actor::scheduled_actor(bool enable_chained_send) scheduled_actor::scheduled_actor(bool enable_chained_send)
: local_actor(enable_chained_send), next(nullptr), m_scheduler(nullptr) { } : local_actor(enable_chained_send), next(nullptr), m_scheduler(nullptr) { }
void scheduled_actor::on_exit() { }
void scheduled_actor::init() { }
scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) { scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) {
CPPA_REQUIRE(sched != nullptr); CPPA_REQUIRE(sched != nullptr);
// init is called by the spawning actor, manipulate self to // init is called by the spawning actor, manipulate self to
......
...@@ -42,6 +42,11 @@ ...@@ -42,6 +42,11 @@
namespace cppa { namespace cppa {
thread_mapped_actor::thread_mapped_actor() : m_initialized(false) { }
thread_mapped_actor::thread_mapped_actor(std::function<void()> fun)
: super(std::move(fun)), m_initialized(false) { }
void thread_mapped_actor::quit(std::uint32_t reason) { void thread_mapped_actor::quit(std::uint32_t reason) {
cleanup(reason); cleanup(reason);
// actor_exited should not be catched, but if anyone does, // actor_exited should not be catched, but if anyone does,
...@@ -57,7 +62,7 @@ void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) { ...@@ -57,7 +62,7 @@ void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) {
} }
bool thread_mapped_actor::initialized() { bool thread_mapped_actor::initialized() {
return true; return m_initialized;
} }
detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) { detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) {
......
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