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

library cleanup: become_void => quit_normal, future_send => delayed_send

parent 699a371b
......@@ -51,7 +51,7 @@ struct testee : fsm_actor<testee> {
init_state = (
on(atom("spread"), 0) >> [=]() {
send(parent, atom("result"), (uint32_t) 1);
become_void();
quit_normal();
},
on<atom("spread"), int>() >> [=](int x) {
any_tuple msg = make_cow_tuple(atom("spread"), x - 1);
......@@ -62,7 +62,7 @@ struct testee : fsm_actor<testee> {
become (
on<atom("result"), uint32_t>() >> [=](uint32_t r2) {
send(parent, atom("result"), r1 + r2);
become_void();
quit_normal();
}
);
}
......
......@@ -134,7 +134,7 @@ struct ping_actor : fsm_actor<ping_actor> {
become (
on<atom("pong"), uint32_t>().when(_x2 == uint32_t(0)) >> [=]() {
send(parent, atom("done"));
become_void();
quit_normal();
},
on(atom("pong"), arg_match) >> [=](uint32_t value) {
reply(atom("ping"), value - 1);
......@@ -206,7 +206,7 @@ struct server_actor : fsm_actor<server_actor> {
},
on(atom("shutdown")) >> [=]() {
m_pongs.clear();
become_void();
quit_normal();
},
others() >> [=]() {
cout << "unexpected: " << to_string(last_dequeued()) << endl;
......
......@@ -53,7 +53,7 @@ struct fsm_receiver : fsm_actor<fsm_receiver> {
on(atom("msg")) >> [=]() {
++m_value;
if (m_value == max) {
become_void();
quit_normal();
}
}
);
......
......@@ -73,7 +73,7 @@ struct fsm_worker : fsm_actor<fsm_worker> {
send(mc, atom("result"), factorize(what));
},
on(atom("done")) >> [=]() {
become_void();
quit_normal();
}
);
}
......@@ -86,7 +86,7 @@ struct fsm_chain_link : fsm_actor<fsm_chain_link> {
init_state = (
on<atom("token"), int>() >> [=](int v) {
next << std::move(last_dequeued());
if (v == 0) become_void();
if (v == 0) quit_normal();
}
);
}
......@@ -120,7 +120,7 @@ struct fsm_chain_master : fsm_actor<fsm_chain_master> {
else {
send(worker, atom("done"));
send(mc, atom("masterdone"));
become_void();
quit_normal();
}
},
on<atom("token"), int>() >> [=](int v) {
......@@ -138,11 +138,11 @@ struct fsm_supervisor : fsm_actor<fsm_supervisor> {
fsm_supervisor(int num_msgs) : left(num_msgs) {
init_state = (
on(atom("masterdone")) >> [=]() {
if (--left == 0) become_void();
if (--left == 0) quit_normal();
},
on<atom("result"), factors>() >> [=](const factors& vec) {
check_factors(vec);
if (--left == 0) become_void();
if (--left == 0) quit_normal();
}
);
}
......
......@@ -63,7 +63,7 @@ class actor : public channel {
* this actor is an scheduled actor that successfully changed
* its state to @p pending.
*/
virtual bool pending_enqueue(actor* sender, any_tuple msg);
virtual bool chained_enqueue(actor* sender, any_tuple msg);
/**
* @brief Attaches @p ptr to this actor.
......
......@@ -32,6 +32,7 @@
#define CPPA_HPP
#include <tuple>
#include <chrono>
#include <cstdint>
#include <functional>
#include <type_traits>
......@@ -342,12 +343,12 @@
*
* @section FutureSend Send delayed messages
*
* The function @p future_send provides a simple way to delay a message.
* The function @p delayed_send provides a simple way to delay a message.
* This is particularly useful for recurring events, e.g., periodical polling.
* Usage example:
*
* @code
* future_send(self, std::chrono::seconds(1), atom("poll"));
* delayed_send(self, std::chrono::seconds(1), atom("poll"));
* receive_loop
* (
* // ...
......@@ -355,7 +356,7 @@
* {
* // ... poll something ...
* // and do it again after 1sec
* future_send(self, std::chrono::seconds(1), atom("poll"));
* delayed_send(self, std::chrono::seconds(1), atom("poll"));
* }
* );
* @endcode
......@@ -411,7 +412,7 @@
*/
/**
* @brief A simple example for a future_send based application.
* @brief A simple example for a delayed_send based application.
* @example dancing_kirby.cpp
*/
......@@ -665,7 +666,7 @@ const self_type& operator<<(const self_type& s, any_tuple&& what);
* @brief Sends a message to the sender of the last received message.
*/
template<typename Arg0, typename... Args>
void reply(const Arg0& arg0, const Args&... args) {
inline void reply(const Arg0& arg0, const Args&... args) {
send(self->last_sender(), arg0, args...);
}
......@@ -676,19 +677,22 @@ void reply(const Arg0& arg0, const Args&... args) {
* @param rel_time Relative time duration to delay the message.
* @param data Any number of values for the message content.
*/
template<typename Duration, typename... Data>
void future_send(actor_ptr whom, const Duration& rel_time, const Data&... data) {
get_scheduler()->future_send(whom, rel_time, data...);
template<class Rep, class Period, typename... Data>
inline void delayed_send(actor_ptr whom,
const std::chrono::duration<Rep, Period>& rel_time,
const Data&... data) {
get_scheduler()->delayed_send(whom, rel_time, data...);
}
/**
* @ingroup MessageHandling
* @brief Sends a reply message that is delayed by @p rel_time.
* @see future_send()
* @see delayed_send()
*/
template<typename Duration, typename... Data>
void delayed_reply(const Duration& rel_time, Data const... data) {
future_send(self->last_sender(), rel_time, data...);
template<class Rep, class Period, typename... Data>
inline void delayed_reply(const std::chrono::duration<Rep, Period>& rel_time,
const Data&... data) {
delayed_send(self->last_sender(), rel_time, data...);
}
/**
......
......@@ -92,7 +92,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
enqueue(nullptr, make_any_tuple(atom("TIMEOUT"), ++m_active_timeout_id));
}
else {
get_scheduler()->future_send(this, d, atom("TIMEOUT"), ++m_active_timeout_id);
get_scheduler()->delayed_send(this, d, atom("TIMEOUT"), ++m_active_timeout_id);
m_has_pending_timeout_request = true;
}
}
......
......@@ -52,11 +52,6 @@ class event_based_actor : public event_based_actor_base<event_based_actor> {
event_based_actor();
/**
* @brief Terminates this actor with normal exit reason.
*/
void become_void();
void quit(std::uint32_t reason);
};
......
......@@ -34,6 +34,7 @@
#include "cppa/actor.hpp"
#include "cppa/behavior.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
......@@ -50,9 +51,10 @@ class local_actor : public actor {
protected:
bool m_chaining;
bool m_trap_exit;
bool m_is_scheduled;
actor_ptr m_pending;
actor_ptr m_chained;
actor_ptr m_last_sender;
any_tuple m_last_dequeued;
......@@ -71,6 +73,10 @@ class local_actor : public actor {
*/
virtual void quit(std::uint32_t reason) = 0;
inline void quit_normal() {
quit(exit_reason::normal);
}
/**
* @brief
* @param rules
......@@ -86,56 +92,52 @@ class local_actor : public actor {
*/
virtual void dequeue(partial_function& rules) = 0;
inline bool trap_exit() const;
inline void trap_exit(bool new_value);
inline any_tuple& last_dequeued();
inline actor_ptr& last_sender();
inline actor_ptr& pending_actor();
inline void send_message(channel* whom, any_tuple what);
inline bool trap_exit() const {
return m_trap_exit;
}
inline void send_message(actor* whom, any_tuple what);
inline void trap_exit(bool new_value) {
m_trap_exit = new_value;
}
};
inline bool chaining() const {
return m_chaining;
}
inline bool local_actor::trap_exit() const {
return m_trap_exit;
}
inline void chaining(bool new_value) {
if (m_is_scheduled) {
m_chaining = new_value;
}
}
inline void local_actor::trap_exit(bool new_value) {
m_trap_exit = new_value;
}
inline any_tuple& last_dequeued() {
return m_last_dequeued;
}
inline any_tuple& local_actor::last_dequeued() {
return m_last_dequeued;
}
inline actor_ptr& last_sender() {
return m_last_sender;
}
inline actor_ptr& local_actor::last_sender() {
return m_last_sender;
}
inline actor_ptr& chained_actor() {
return m_chained;
}
inline actor_ptr& local_actor::pending_actor() {
return m_pending;
}
inline void send_message(channel* whom, any_tuple what) {
whom->enqueue(this, std::move(what));
}
inline void local_actor::send_message(actor* whom, any_tuple what) {
if (m_is_scheduled && !m_pending) {
if (whom->pending_enqueue(this, std::move(what))) {
m_pending = whom;
inline void send_message(actor* whom, any_tuple what) {
if (m_chaining && !m_chained) {
if (whom->chained_enqueue(this, std::move(what))) {
m_chained = whom;
}
}
else {
whom->enqueue(this, std::move(what));
}
}
else {
whom->enqueue(this, std::move(what));
}
}
inline void local_actor::send_message(channel* whom, any_tuple what) {
whom->enqueue(this, std::move(what));
}
};
typedef intrusive_ptr<local_actor> local_actor_ptr;
......
......@@ -58,7 +58,7 @@ class scheduler {
scheduler_helper* m_helper;
channel* future_send_helper();
channel* delayed_send_helper();
protected:
......@@ -114,12 +114,12 @@ class scheduler {
virtual attachable* register_hidden_context();
template<typename Duration, typename... Data>
void future_send(const actor_ptr& to,
void delayed_send(const actor_ptr& to,
const Duration& rel_time, const Data&... data) {
static_assert(sizeof...(Data) > 0, "no message to send");
any_tuple data_tup = make_cow_tuple(data...);
any_tuple tup = make_cow_tuple(util::duration(rel_time), to, data_tup);
future_send_helper()->enqueue(self, std::move(tup));
delayed_send_helper()->enqueue(self, std::move(tup));
}
};
......
......@@ -56,7 +56,7 @@ class stacked_event_based_actor : public event_based_actor_base<stacked_event_ba
/**
* @brief Terminates this actor with normal exit reason.
*/
void become_void();
void quit_normal();
};
......
......@@ -41,7 +41,7 @@ void dancing_kirby() {
on<atom("Step")>() >> [&]() {
draw_kirby(*i);
// animate next step in 150ms
future_send(self, std::chrono::milliseconds(150), atom("Step"));
delayed_send(self, std::chrono::milliseconds(150), atom("Step"));
}
);
}
......
......@@ -102,7 +102,7 @@ struct philosopher : fsm_actor<philosopher> {
<< " and starts to eat\n";
cout << oss.str();
// eat some time
future_send(this, seconds(5), atom("think"));
delayed_send(this, seconds(5), atom("think"));
become(&eating);
},
on(atom("busy"), what) >> [=]() {
......@@ -152,7 +152,7 @@ struct philosopher : fsm_actor<philosopher> {
on(atom("think")) >> [=]() {
send(left, atom("put"), this);
send(right, atom("put"), this);
future_send(this, seconds(5), atom("eat"));
delayed_send(this, seconds(5), atom("eat"));
cout << ( name
+ " puts down his chopsticks and starts to think\n");
become(&thinking);
......@@ -162,7 +162,7 @@ struct philosopher : fsm_actor<philosopher> {
init_state = (
on(atom("think")) >> [=]() {
cout << (name + " starts to think\n");
future_send(this, seconds(5), atom("eat"));
delayed_send(this, seconds(5), atom("eat"));
become(&thinking);
}
);
......
......@@ -45,7 +45,7 @@ struct math_actor : event_based_actor {
on(atom("quit")) >> [=]() {
// set an empty behavior
// (terminates actor with normal exit reason)
become_void();
quit_normal();
}
);
}
......
......@@ -61,7 +61,7 @@ actor::actor(std::uint32_t aid, const process_information_ptr& pptr)
}
}
bool actor::pending_enqueue(actor* sender, any_tuple msg) {
bool actor::chained_enqueue(actor* sender, any_tuple msg) {
enqueue(sender, std::move(msg));
return false;
}
......
......@@ -36,14 +36,10 @@ event_based_actor::event_based_actor() {
m_loop_stack.reserve(2);
}
void event_based_actor::become_void() {
cleanup(exit_reason::normal);
m_loop_stack.clear();
}
void event_based_actor::quit(std::uint32_t reason) {
if (reason == exit_reason::normal) {
become_void();
cleanup(exit_reason::normal);
m_loop_stack.clear();
}
else {
abstract_scheduled_actor::quit(reason);
......
......@@ -32,7 +32,8 @@
namespace cppa {
local_actor::local_actor(bool sflag) : m_trap_exit(false), m_is_scheduled(sflag) {
local_actor::local_actor(bool sflag)
: m_chaining(sflag), m_trap_exit(false), m_is_scheduled(sflag) {
}
} // namespace cppa
......@@ -170,7 +170,7 @@ scheduler::~scheduler() {
delete m_helper;
}
channel* scheduler::future_send_helper() {
channel* scheduler::delayed_send_helper() {
return m_helper->m_worker.get();
}
......
......@@ -32,7 +32,7 @@
namespace cppa {
void stacked_event_based_actor::become_void() {
void stacked_event_based_actor::quit_normal() {
m_loop_stack.clear();
}
......
......@@ -125,9 +125,9 @@ struct thread_pool_scheduler::worker {
scheduled_actor* pending_job;
handler() : job(nullptr), pending_job(nullptr) { }
void exec_done() {
if (job->pending_actor()) {
pending_job = static_cast<scheduled_actor*>(job->pending_actor().get());
job->pending_actor().reset();
if (job->chained_actor()) {
pending_job = static_cast<scheduled_actor*>(job->chained_actor().get());
job->chained_actor().reset();
}
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
......@@ -153,9 +153,9 @@ struct thread_pool_scheduler::worker {
do {
h.job->resume(&fself, &h);
if (h.job) {
if (h.job->pending_actor()) {
h.pending_job = static_cast<scheduled_actor*>(h.job->pending_actor().get());
h.job->pending_actor().reset();
if (h.job->chained_actor()) {
h.pending_job = static_cast<scheduled_actor*>(h.job->chained_actor().get());
h.job->chained_actor().reset();
}
else {
h.job = nullptr;
......
......@@ -46,7 +46,7 @@ actor_ptr spawn_event_based_ping(size_t num_pings) {
//cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs >= num_pings) {
reply(atom("EXIT"), exit_reason::user_defined);
become_void();
quit_normal();
}
else {
reply(atom("ping"), value);
......
#define CPPA_VERBOSE_CHECK
#include <stack>
#include <chrono>
#include <iostream>
......@@ -24,6 +22,35 @@ using std::endl;
using namespace cppa;
#if (__GNUC__ >= 4) && (__GNUC_MINOR__ >= 7)
struct simple_mirror : fsm_actor<simple_mirror> {
behavior init_state = (
others() >> []() {
self->last_sender() << self->last_dequeued();
}
);
};
#else
struct simple_mirror : event_based_actor {
void init() {
become (
others() >> []() {
self->last_sender() << self->last_dequeued();
}
);
}
};
#endif
// GCC 4.7 supports non-static member initialization
#if 0 //(__GNUC__ >= 4) && (__GNUC_MINOR__ >= 7)
......@@ -113,7 +140,7 @@ abstract_event_based_actor* event_testee2() {
behavior wait4timeout(int remaining) {
return (
after(std::chrono::milliseconds(50)) >> [=]() {
if (remaining == 1) become_void();
if (remaining == 1) quit_normal();
else become(wait4timeout(remaining - 1));
}
);
......@@ -137,7 +164,7 @@ struct chopstick : public fsm_actor<chopstick> {
become(&init_state);
},
on(atom("break")) >> [=]() {
become_void();
quit_normal();
}
);
}
......@@ -151,7 +178,7 @@ struct chopstick : public fsm_actor<chopstick> {
reply(atom("taken"));
},
on(atom("break")) >> [=]() {
become_void();
quit_normal();
},
others() >> [=]() {
}
......@@ -224,8 +251,8 @@ void testee2(actor_ptr other) {
}
void testee3(actor_ptr parent) {
// test a future_send / delayed_reply based loop
future_send(self, std::chrono::milliseconds(50), atom("Poll"));
// test a delayed_send / delayed_reply based loop
delayed_send(self, std::chrono::milliseconds(50), atom("Poll"));
int polls = 0;
receive_for(polls, 5) (
on(atom("Poll")) >> [&]() {
......@@ -387,11 +414,15 @@ size_t test__spawn() {
);
CPPA_IF_VERBOSE(cout << "ok" << endl);
/*
auto mirror = actor_prototype (
others() >> []() {
self->last_sender() << self->last_dequeued();
}
).spawn();
*/
auto mirror = spawn(new simple_mirror);
CPPA_IF_VERBOSE(cout << "test mirror ... " << std::flush);
send(mirror, "hello mirror");
......@@ -422,8 +453,8 @@ size_t test__spawn() {
}
);
CPPA_IF_VERBOSE(cout << "test future_send() ... " << std::flush);
future_send(self, std::chrono::seconds(1), 1, 2, 3);
CPPA_IF_VERBOSE(cout << "test delayed_send() ... " << std::flush);
delayed_send(self, std::chrono::seconds(1), 1, 2, 3);
receive(on(1, 2, 3) >> []() { });
CPPA_IF_VERBOSE(cout << "ok" << endl);
......@@ -497,7 +528,7 @@ size_t test__spawn() {
self->link_to(pong_actor);
int i = 0;
int flags = 0;
future_send(self, std::chrono::seconds(1), atom("FooBar"));
delayed_send(self, std::chrono::seconds(1), atom("FooBar"));
// wait for DOWN and EXIT messages of pong
receive_for(i, 4) (
on<atom("EXIT"), std::uint32_t>() >> [&](std::uint32_t 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