Commit 4b0a9051 authored by Dominik Charousset's avatar Dominik Charousset

added new mixins for thread- and stackless actors

parent 984b08fd
......@@ -137,8 +137,10 @@ cppa/serializer.hpp
cppa/singletons.hpp
cppa/spawn_options.hpp
cppa/stacked.hpp
cppa/stackless.hpp
cppa/thread_mapped_actor.hpp
cppa/threaded.hpp
cppa/threadless.hpp
cppa/timeout_definition.hpp
cppa/to_string.hpp
cppa/tpartial_function.hpp
......@@ -268,6 +270,7 @@ src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp
src/self.cpp
src/send.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
......
......@@ -39,6 +39,7 @@
#include "cppa/config.hpp"
#include "cppa/extend.hpp"
#include "cppa/behavior.hpp"
#include "cppa/stackless.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
......@@ -47,11 +48,11 @@ namespace cppa {
/**
* @brief Base class for all event-based actor implementations.
*/
class event_based_actor : public scheduled_actor {
class event_based_actor : public extend<scheduled_actor>::with<stackless> {
friend class detail::receive_policy;
typedef scheduled_actor super;
typedef combined_type super;
public:
......@@ -64,8 +65,6 @@ class event_based_actor : public scheduled_actor {
virtual void quit(std::uint32_t reason = exit_reason::normal);
bool has_behavior();
scheduled_actor_type impl_type();
static intrusive_ptr<event_based_actor> from(std::function<void()> fun);
......@@ -74,72 +73,6 @@ class event_based_actor : public scheduled_actor {
event_based_actor(actor_state st = actor_state::blocked);
// provoke compiler errors for usage of receive() and related functions
/**
* @brief Provokes a compiler error to ensure that an event-based actor
* does not accidently uses receive() instead of become().
*/
template<typename... Ts>
void receive(Ts&&...) {
// this asssertion always fails
static_assert((sizeof...(Ts) + 1) < 1,
"You shall not use receive in an event-based actor. "
"Use become() instead.");
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_while(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void do_receive(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
void do_become(behavior&& bhvr, bool discard_old);
void become_waiting_for(behavior bhvr, message_id mf);
detail::receive_policy m_recv_policy;
private:
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(m_bhvr_stack.empty() == false);
return m_bhvr_stack.back();
}
// required by detail::nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_sequential;
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
reset_timeout();
bhvr.handle_timeout();
if (m_bhvr_stack.empty() == false) {
request_timeout(get_behavior().timeout());
}
}
};
} // namespace cppa
......
......@@ -35,6 +35,7 @@
#include "cppa/config.hpp"
#include "cppa/extend.hpp"
#include "cppa/threadless.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
......@@ -68,14 +69,12 @@ class scheduled_actor;
* @brief A base class for cooperatively scheduled actors.
* @extends local_actor
*/
class scheduled_actor : public extend<local_actor>::with<mailbox_based> {
class scheduled_actor : public extend<local_actor>::with<mailbox_based, threadless> {
typedef combined_type super;
public:
static constexpr bool has_blocking_receive = false;
~scheduled_actor();
/**
......@@ -121,37 +120,6 @@ class scheduled_actor : public extend<local_actor>::with<mailbox_based> {
bool chained_enqueue(const message_header&, any_tuple) override;
void request_timeout(const util::duration& d);
inline bool has_pending_timeout() const {
return m_has_pending_tout;
}
inline void reset_timeout() {
if (m_has_pending_tout) {
++m_pending_tout;
m_has_pending_tout = false;
}
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void push_timeout() {
++m_pending_tout;
}
inline void pop_timeout() {
CPPA_REQUIRE(m_pending_tout > 0);
--m_pending_tout;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) {
return m_has_pending_tout && m_pending_tout == timeout_id;
}
protected:
scheduled_actor(actor_state init_state, bool enable_chained_send);
......@@ -177,15 +145,12 @@ class scheduled_actor : public extend<local_actor>::with<mailbox_based> {
bool enqueue_impl(actor_state, const message_header&, any_tuple&&);
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
std::atomic<actor_state> m_state;
protected:
scheduler* m_scheduler;
bool m_hidden;
mailbox_type m_mailbox;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_STACKLESS_HPP
#define CPPA_STACKLESS_HPP
#include "cppa/detail/receive_policy.hpp"
namespace cppa {
/**
* @brief An actor that uses the non-blocking API of @p libcppa and
* does not has its own stack.
*/
template<class Base, class Subtype>
class stackless : public Base {
friend class detail::receive_policy;
friend class detail::behavior_stack;
protected:
typedef stackless combined_type;
public:
template<typename... Ts>
stackless(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
static constexpr auto receive_flag = detail::rp_sequential;
bool has_behavior() {
return this->m_bhvr_stack.empty() == false;
}
protected:
void do_become(behavior&& bhvr, bool discard_old) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
}
void become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr), mf);
}
inline bool has_behavior() const {
return this->m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(this->m_bhvr_stack.empty() == false);
return this->m_bhvr_stack.back();
}
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
this->reset_timeout();
bhvr.handle_timeout();
if (this->m_bhvr_stack.empty() == false) {
this->request_timeout(get_behavior().timeout());
}
}
// provoke compiler errors for usage of receive() and related functions
/**
* @brief Provokes a compiler error to ensure that a stackless actor
* does not accidently uses receive() instead of become().
*/
template<typename... Ts>
void receive(Ts&&...) {
// this asssertion always fails
static_assert((sizeof...(Ts) + 1) < 1,
"You shall not use receive in an event-based actor. "
"Use become() instead.");
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void receive_while(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
/**
* @brief Provokes a compiler error.
*/
template<typename... Ts>
void do_receive(Ts&&... args) {
receive(std::forward<Ts>(args)...);
}
detail::receive_policy m_recv_policy;
};
} // namespace cppa
#endif // CPPA_STACKLESS_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_THREADLESS_HPP
#define CPPA_THREADLESS_HPP
#include "cppa/send.hpp"
namespace cppa {
/**
* @brief An actor that is scheduled or otherwise managed.
*/
template<class Base, class Subtype>
class threadless : public Base {
protected:
typedef threadless combined_type;
public:
static constexpr bool has_blocking_receive = false;
template<typename... Ts>
threadless(Ts&&... args) : Base(std::forward<Ts>(args)...)
, m_has_pending_tout(false)
, m_pending_tout(0) { }
inline void reset_timeout() {
if (m_has_pending_tout) {
++m_pending_tout;
m_has_pending_tout = false;
}
}
void request_timeout(const util::duration& d) {
if (!d.valid()) m_has_pending_tout = false;
else {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else delayed_send_tuple(this, d, std::move(msg));
m_has_pending_tout = true;
}
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void pop_timeout() {
CPPA_REQUIRE(m_pending_tout > 0);
--m_pending_tout;
}
inline void push_timeout() {
++m_pending_tout;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) const {
return m_has_pending_tout && m_pending_tout == timeout_id;
}
inline bool has_pending_timeout() const {
return m_has_pending_tout;
}
private:
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
};
} // namespace cppa
#endif // CPPA_THREADLESS_HPP
......@@ -185,24 +185,9 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
return resume_result::actor_done;
}
bool event_based_actor::has_behavior() {
return m_bhvr_stack.empty() == false;
}
void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
reset_timeout();
request_timeout(bhvr.timeout());
if (discard_old) m_bhvr_stack.pop_async_back();
m_bhvr_stack.push_back(std::move(bhvr));
}
void event_based_actor::become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
reset_timeout();
request_timeout(bhvr.timeout());
}
m_bhvr_stack.push_back(std::move(bhvr), mf);
}
void event_based_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
......
......@@ -37,9 +37,8 @@
namespace cppa {
scheduled_actor::scheduled_actor(actor_state init_state, bool chained_send)
: super(chained_send), next(nullptr), m_has_pending_tout(false)
, m_pending_tout(0), m_state(init_state), m_scheduler(nullptr)
, m_hidden(false) { }
: super(chained_send), next(nullptr), m_state(init_state)
, m_scheduler(nullptr), m_hidden(false) { }
void scheduled_actor::attach_to_scheduler(scheduler* sched, bool hidden) {
CPPA_REQUIRE(sched != nullptr);
......@@ -59,20 +58,6 @@ bool scheduled_actor::initialized() const {
return m_scheduler != nullptr;
}
void scheduled_actor::request_timeout(const util::duration& d) {
if (!d.valid()) m_has_pending_tout = false;
else {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
auto e = this->new_mailbox_element(this, std::move(msg));
this->m_mailbox.enqueue(e);
}
else get_scheduler()->delayed_send(this, d, std::move(msg));
m_has_pending_tout = true;
}
}
scheduled_actor::~scheduled_actor() {
if (!m_mailbox.closed()) {
detail::sync_request_bouncer f{exit_reason()};
......
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