Commit e13e5d51 authored by Dominik Charousset's avatar Dominik Charousset

sync_reply

parent b5ebe06b
...@@ -65,10 +65,28 @@ class actor : public channel { ...@@ -65,10 +65,28 @@ class actor : public channel {
/** /**
* @brief Enqueues @p msg to the actor's mailbox and returns true if * @brief Enqueues @p msg to the actor's mailbox and returns true if
* this actor is an scheduled actor that successfully changed * this actor is an scheduled actor that successfully changed
* its state to @p pending. * its state to @p pending in response to the enqueue operation.
*/ */
virtual bool chained_enqueue(actor* sender, any_tuple msg); virtual bool chained_enqueue(actor* sender, any_tuple msg);
/**
* @brief Enqueues @p msg as a reply to @p request_id
* to this actor's mailbox.
*/
virtual void sync_enqueue(actor* sender,
std::uint64_t sequence_id,
any_tuple msg ) = 0;
/**
* @brief Enqueues @p msg as a reply to @p request_id
* to this actor's mailbox and returns true if
* this actor is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_sync_enqueue(actor* sender,
std::uint64_t response_id,
any_tuple msg);
/** /**
* @brief Attaches @p ptr to this actor. * @brief Attaches @p ptr to this actor.
* *
......
...@@ -55,6 +55,8 @@ class actor_proxy : public detail::abstract_actor<actor> { ...@@ -55,6 +55,8 @@ class actor_proxy : public detail::abstract_actor<actor> {
void enqueue(actor* sender, any_tuple msg); void enqueue(actor* sender, any_tuple msg);
void sync_enqueue(actor* sender, std::uint64_t response_id, any_tuple msg);
void link_to(intrusive_ptr<actor>& other); void link_to(intrusive_ptr<actor>& other);
// do not cause to send this actor an "UNLINK" message // do not cause to send this actor an "UNLINK" message
...@@ -71,9 +73,10 @@ class actor_proxy : public detail::abstract_actor<actor> { ...@@ -71,9 +73,10 @@ class actor_proxy : public detail::abstract_actor<actor> {
bool establish_backlink(intrusive_ptr<actor>& to); bool establish_backlink(intrusive_ptr<actor>& to);
public: void forward_message(const process_information_ptr&,
actor*,
void forward_message(const process_information_ptr&, actor*, any_tuple&&); std::uint64_t,
any_tuple&& );
}; };
......
...@@ -461,7 +461,7 @@ send(const intrusive_ptr<C>& whom, Args&&... what) { ...@@ -461,7 +461,7 @@ send(const intrusive_ptr<C>& whom, Args&&... what) {
*/ */
template<typename... Args> template<typename... Args>
inline void reply(Args&&... what) { inline void reply(Args&&... what) {
send(self->last_sender(), std::forward<Args>(what)...); self->reply_message(make_any_tuple(std::forward<Args>(what)...));
} }
/** /**
...@@ -492,7 +492,8 @@ inline void delayed_send(const channel_ptr& whom, ...@@ -492,7 +492,8 @@ inline void delayed_send(const channel_ptr& whom,
template<class Rep, class Period, typename... Args> template<class Rep, class Period, typename... Args>
inline void delayed_reply(const std::chrono::duration<Rep, Period>& rel_time, inline void delayed_reply(const std::chrono::duration<Rep, Period>& rel_time,
Args&&... what) { Args&&... what) {
delayed_send(self->last_sender(), rel_time, std::forward<Args>(what)...); delayed_send(self->last_sender(), rel_time, self->get_response_id(),
std::forward<Args>(what)...);
} }
/** @} */ /** @} */
......
...@@ -39,6 +39,7 @@ ...@@ -39,6 +39,7 @@
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <thread> #include <thread>
#include <cstdint>
#include <algorithm> #include <algorithm>
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
...@@ -188,7 +189,9 @@ class abstract_actor : public abstract_actor_base<Base> { ...@@ -188,7 +189,9 @@ class abstract_actor : public abstract_actor_base<Base> {
typedef std::lock_guard<util::shared_spinlock> lock_type; typedef std::lock_guard<util::shared_spinlock> lock_type;
inline mailbox_element* fetch_node(actor* sender, any_tuple msg) { inline mailbox_element* fetch_node(actor* sender,
any_tuple msg,
std::uint64_t seq_id = 0) {
mailbox_element* result = nullptr; mailbox_element* result = nullptr;
{ // lifetime scope of guard { // lifetime scope of guard
lock_type guard{m_nodes_lock}; lock_type guard{m_nodes_lock};
...@@ -202,9 +205,10 @@ class abstract_actor : public abstract_actor_base<Base> { ...@@ -202,9 +205,10 @@ class abstract_actor : public abstract_actor_base<Base> {
result->marked = false; result->marked = false;
result->sender = sender; result->sender = sender;
result->msg = std::move(msg); result->msg = std::move(msg);
result->seq_id = seq_id;
} }
else { else {
result = new mailbox_element(sender, std::move(msg)); result = new mailbox_element(sender, std::move(msg), seq_id);
} }
return result; return result;
} }
......
...@@ -46,7 +46,6 @@ ...@@ -46,7 +46,6 @@
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/single_reader_queue.hpp" #include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
// A spawned, scheduled Actor. // A spawned, scheduled Actor.
...@@ -58,29 +57,6 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -58,29 +57,6 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
std::atomic<int> m_state; std::atomic<int> m_state;
filter_result filter_msg(const any_tuple& msg) {
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.type_at(1) == arr[1]) {
auto v0 = *reinterpret_cast<const atom_value*>(msg.at(0));
auto v1 = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (v0 == atom("EXIT")) {
if (this->m_trap_exit == false) {
if (v1 != exit_reason::normal) {
quit(v1);
}
return normal_exit_signal;
}
}
else if (v0 == atom("TIMEOUT")) {
return (v1 == m_active_timeout_id) ? timeout_message
: expired_timeout_message;
}
}
return ordinary_message;
}
bool has_pending_timeout() { bool has_pending_timeout() {
return m_has_pending_timeout_request; return m_has_pending_timeout_request;
} }
...@@ -120,6 +96,10 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -120,6 +96,10 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
--m_active_timeout_id; --m_active_timeout_id;
} }
inline bool waits_for_timeout(std::uint32_t timeout_id) {
return m_active_timeout_id == timeout_id;
}
bool m_has_pending_timeout_request; bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id; std::uint32_t m_active_timeout_id;
...@@ -150,6 +130,10 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -150,6 +130,10 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
enqueue_node(super::fetch_node(sender, std::move(msg))); enqueue_node(super::fetch_node(sender, std::move(msg)));
} }
void sync_enqueue(actor* sender, std::uint64_t seq_id, any_tuple msg) {
enqueue_node(super::fetch_node(sender, std::move(msg), seq_id));
}
int compare_exchange_state(int expected, int new_value) { int compare_exchange_state(int expected, int new_value) {
int e = expected; int e = expected;
do { do {
......
...@@ -45,7 +45,8 @@ class addressed_message { ...@@ -45,7 +45,8 @@ class addressed_message {
public: public:
addressed_message(actor_ptr from, channel_ptr to, any_tuple ut); addressed_message(actor_ptr from, channel_ptr to,
std::uint64_t seq_id, any_tuple ut);
addressed_message() = default; addressed_message() = default;
addressed_message(addressed_message&&) = default; addressed_message(addressed_message&&) = default;
...@@ -77,6 +78,10 @@ class addressed_message { ...@@ -77,6 +78,10 @@ class addressed_message {
return m_content; return m_content;
} }
inline std::uint64_t sequence_id() const {
return m_sequence_id;
}
inline bool empty() const { inline bool empty() const {
return m_content.empty(); return m_content.empty();
} }
...@@ -85,6 +90,7 @@ class addressed_message { ...@@ -85,6 +90,7 @@ class addressed_message {
actor_ptr m_sender; actor_ptr m_sender;
channel_ptr m_receiver; channel_ptr m_receiver;
std::uint64_t m_sequence_id;
any_tuple m_content; any_tuple m_content;
}; };
......
...@@ -35,7 +35,9 @@ namespace cppa { namespace detail { ...@@ -35,7 +35,9 @@ namespace cppa { namespace detail {
enum filter_result { enum filter_result {
normal_exit_signal, normal_exit_signal,
non_normal_exit_signal,
expired_timeout_message, expired_timeout_message,
expired_sync_enqueue,
timeout_message, timeout_message,
ordinary_message ordinary_message
}; };
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#include <type_traits> #include <type_traits>
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/partial_function.hpp" #include "cppa/partial_function.hpp"
#include "cppa/detail/filter_result.hpp" #include "cppa/detail/filter_result.hpp"
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
...@@ -175,6 +176,44 @@ class receive_policy { ...@@ -175,6 +176,44 @@ class receive_policy {
CPPA_CRITICAL("handle_timeout(partial_function&)"); CPPA_CRITICAL("handle_timeout(partial_function&)");
} }
template<class Client>
filter_result filter_msg(Client* client, recursive_queue_node* node) {
const any_tuple& msg = node->msg;
bool is_sync_msg = node->seq_id != 0;
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.type_at(1) == arr[1]) {
auto v0 = msg.get_as<atom_value>(0);
auto v1 = msg.get_as<std::uint32_t>(1);
if (v0 == atom("EXIT")) {
CPPA_REQUIRE(is_sync_msg == false);
if (client->m_trap_exit == false) {
if (v1 != exit_reason::normal) {
// TODO: check for possible memory leak here
// ('node' might not get destroyed)
client->quit(v1);
}
return normal_exit_signal;
}
}
else if (v0 == atom("TIMEOUT")) {
CPPA_REQUIRE(is_sync_msg == false);
return client->waits_for_timeout(v1) ? timeout_message
: expired_timeout_message;
}
}
constexpr std::uint64_t is_response_mask = 0x8000000000000000;
constexpr std::uint64_t request_id_mask = 0x7FFFFFFFFFFFFFFF;
// first bit is 1 if this is a response message
if ( is_sync_msg
&& (node->seq_id & is_response_mask) != 0
&& (node->seq_id & request_id_mask) != client->m_sync_request_id) {
return expired_sync_enqueue;
}
return ordinary_message;
}
template<class Client, class FunOrBehavior> template<class Client, class FunOrBehavior>
handle_message_result handle_message(Client* client, handle_message_result handle_message(Client* client,
recursive_queue_node* node, recursive_queue_node* node,
...@@ -183,8 +222,9 @@ class receive_policy { ...@@ -183,8 +222,9 @@ class receive_policy {
if (node->marked) { if (node->marked) {
return hm_skip_msg; return hm_skip_msg;
} }
switch (client->filter_msg(node->msg)) { switch (this->filter_msg(client, node)) {
case normal_exit_signal: case normal_exit_signal:
case expired_sync_enqueue:
case expired_timeout_message: { case expired_timeout_message: {
return hm_drop_msg; return hm_drop_msg;
} }
...@@ -217,8 +257,9 @@ class receive_policy { ...@@ -217,8 +257,9 @@ class receive_policy {
FunOrBehavior& fun, FunOrBehavior& fun,
sequential) { sequential) {
CPPA_REQUIRE(node->marked == false); CPPA_REQUIRE(node->marked == false);
switch (client->filter_msg(node->msg)) { switch (this->filter_msg(client, node)) {
case normal_exit_signal: case normal_exit_signal:
case expired_sync_enqueue:
case expired_timeout_message: { case expired_timeout_message: {
return hm_drop_msg; return hm_drop_msg;
} }
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#ifndef CPPA_RECURSIVE_QUEUE_NODE_HPP #ifndef CPPA_RECURSIVE_QUEUE_NODE_HPP
#define CPPA_RECURSIVE_QUEUE_NODE_HPP #define CPPA_RECURSIVE_QUEUE_NODE_HPP
#include <cstdint>
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
...@@ -38,15 +40,25 @@ namespace cppa { namespace detail { ...@@ -38,15 +40,25 @@ namespace cppa { namespace detail {
struct recursive_queue_node { struct recursive_queue_node {
recursive_queue_node* next; // intrusive next pointer typedef std::uint64_t seq_id_t;
bool marked; // denotes if this node is currently processed typedef recursive_queue_node* pointer;
actor_ptr sender;
any_tuple msg; pointer next; // intrusive next pointer
bool marked; // denotes if this node is currently processed
actor_ptr sender; // points to the sender of msg
any_tuple msg; // 'content field'
seq_id_t seq_id; // sequence id:
// - equals to 0 for asynchronous messages
// - first bit is 1 if this messages is a
// response messages, otherwise 0
// - the trailing 63 bit uniquely identify a
// request/response message pair
inline recursive_queue_node() : next(nullptr), marked(false) { } inline recursive_queue_node() : next(nullptr), marked(false), seq_id(0) { }
inline recursive_queue_node(actor* from, any_tuple&& content) inline recursive_queue_node(actor* ptr, any_tuple&& data, seq_id_t id = 0)
: next(nullptr), marked(false), sender(from), msg(std::move(content)) { } : next(nullptr), marked(false), sender(std::move(ptr))
, msg(std::move(data)), seq_id(id) { }
recursive_queue_node(recursive_queue_node&&) = delete; recursive_queue_node(recursive_queue_node&&) = delete;
recursive_queue_node(const recursive_queue_node&) = delete; recursive_queue_node(const recursive_queue_node&) = delete;
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#ifndef CPPA_CONTEXT_HPP #ifndef CPPA_CONTEXT_HPP
#define CPPA_CONTEXT_HPP #define CPPA_CONTEXT_HPP
#include <cstdint>
#include "cppa/group.hpp" #include "cppa/group.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
...@@ -309,8 +311,38 @@ class local_actor : public actor { ...@@ -309,8 +311,38 @@ class local_actor : public actor {
m_chained_actor = whom; m_chained_actor = whom;
} }
} }
else { else whom->enqueue(this, std::move(what));
whom->enqueue(this, std::move(what)); }
inline std::uint64_t get_response_id() {
constexpr std::uint64_t response_mask = 0x8000000000000000;
auto& whom = last_sender();
if (whom) {
auto seq_id = m_current_node->seq_id;
if (seq_id != 0 && (seq_id & response_mask) == 0) {
return seq_id | response_mask;
}
}
return 0;
}
inline void reply_message(any_tuple&& what) {
auto& whom = last_sender();
if (whom) {
auto response_id = get_response_id();
if (response_id == 0) {
send_message(whom.get(), std::move(what));
}
else {
if (m_chaining && !m_chained_actor) {
if (whom->chained_sync_enqueue(this,
response_id,
std::move(what))) {
m_chained_actor = whom;
}
}
else whom->sync_enqueue(this, response_id, std::move(what));
}
} }
} }
...@@ -320,12 +352,22 @@ class local_actor : public actor { ...@@ -320,12 +352,22 @@ class local_actor : public actor {
protected: protected:
// true if this actor uses the chained_send optimization
bool m_chaining; bool m_chaining;
// true if this actor receives EXIT messages as ordinary messages
bool m_trap_exit; bool m_trap_exit;
// true if this actor takes part in cooperative scheduling
bool m_is_scheduled; bool m_is_scheduled;
// pointer to the actor that is marked as successor due to a chained send
actor_ptr m_chained_actor; actor_ptr m_chained_actor;
// identifies the ID of the last sent synchronous request
std::uint64_t m_sync_request_id;
// "default value" for m_current_node
detail::recursive_queue_node m_dummy_node; detail::recursive_queue_node m_dummy_node;
// points to m_dummy_node if no callback is currently invoked,
// points to the node under processing otherwise
detail::recursive_queue_node* m_current_node; detail::recursive_queue_node* m_current_node;
// {group => subscription} map of all joined groups
std::map<group_ptr, group::subscription> m_subscriptions; std::map<group_ptr, group::subscription> m_subscriptions;
# endif // CPPA_DOCUMENTATION # endif // CPPA_DOCUMENTATION
......
...@@ -116,13 +116,37 @@ class scheduler { ...@@ -116,13 +116,37 @@ class scheduler {
template<typename Duration, typename... Data> template<typename Duration, typename... Data>
void delayed_send(const channel_ptr& to, void delayed_send(const channel_ptr& to,
const Duration& rel_time, const Duration& rel_time,
Data&&... data) { Data&&... data ) {
static_assert(sizeof...(Data) > 0, "no message to send"); static_assert(sizeof...(Data) > 0, "no message to send");
auto sub = make_any_tuple(std::forward<Data>(data)...); auto sub = make_any_tuple(std::forward<Data>(data)...);
auto tup = make_any_tuple(util::duration{rel_time}, to, std::move(sub)); auto tup = make_any_tuple(atom("SEND"),
util::duration{rel_time},
to,
std::move(sub));
delayed_send_helper()->enqueue(self, std::move(tup)); delayed_send_helper()->enqueue(self, std::move(tup));
} }
template<typename Duration, typename... Data>
void delayed_reply(const actor_ptr& to,
const Duration& rel_time,
std::uint64_t sequence_id,
Data&&... data ) {
static_assert(sizeof...(Data) > 0, "no message to send");
if (sequence_id == 0) {
delayed_send(to, rel_time, sequence_id,
std::forward<Data>(data)...);
}
else {
auto sub = make_any_tuple(std::forward<Data>(data)...);
auto tup = make_any_tuple(atom("REPLY"),
util::duration{rel_time},
to,
sequence_id,
std::move(sub));
delayed_send_helper()->enqueue(self, std::move(tup));
}
}
/** /**
* @brief Spawns a new actor that executes <code>fun()</code> * @brief Spawns a new actor that executes <code>fun()</code>
* with the scheduling policy @p hint if possible. * with the scheduling policy @p hint if possible.
......
...@@ -103,7 +103,7 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -103,7 +103,7 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
void enqueue(actor* sender, any_tuple msg); //override void enqueue(actor* sender, any_tuple msg); //override
detail::filter_result filter_msg(const any_tuple& msg); void sync_enqueue(actor* sender, std::uint64_t response_id, any_tuple msg);
inline decltype(m_mailbox)& mailbox() { return m_mailbox; } inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
...@@ -121,6 +121,7 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -121,6 +121,7 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
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() { }
inline void pop_timeout() { } inline void pop_timeout() { }
inline bool waits_for_timeout(std::uint32_t) { return false; }
inline detail::recursive_queue_node* receive_node() { inline detail::recursive_queue_node* receive_node() {
return m_mailbox.pop(); return m_mailbox.pop();
} }
......
...@@ -66,6 +66,13 @@ bool actor::chained_enqueue(actor* sender, any_tuple msg) { ...@@ -66,6 +66,13 @@ bool actor::chained_enqueue(actor* sender, any_tuple msg) {
return false; return false;
} }
bool actor::chained_sync_enqueue(actor* ptr,
std::uint64_t response_id,
any_tuple msg ) {
sync_enqueue(ptr, response_id, std::move(msg));
return false;
}
actor::actor(const process_information_ptr& pptr) actor::actor(const process_information_ptr& pptr)
: m_id(registry().next_id()), m_is_proxy(false), m_parent_process(pptr) { : m_id(registry().next_id()), m_is_proxy(false), m_parent_process(pptr) {
if (!pptr) { if (!pptr) {
......
...@@ -46,8 +46,9 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr) ...@@ -46,8 +46,9 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
void actor_proxy::forward_message(const process_information_ptr& piptr, void actor_proxy::forward_message(const process_information_ptr& piptr,
actor* sender, actor* sender,
any_tuple&& msg) { std::uint64_t sequence_id,
detail::addressed_message amsg{sender, this, std::move(msg)}; any_tuple&& msg ) {
detail::addressed_message amsg{sender, this, sequence_id, std::move(msg)};
detail::singleton_manager::get_network_manager() detail::singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(piptr, std::move(amsg))); ->send_to_mailman(make_any_tuple(piptr, std::move(amsg)));
} }
...@@ -60,7 +61,13 @@ void actor_proxy::enqueue(actor* sender, any_tuple msg) { ...@@ -60,7 +61,13 @@ void actor_proxy::enqueue(actor* sender, any_tuple msg) {
cleanup(msg.get_as<std::uint32_t>(1)); cleanup(msg.get_as<std::uint32_t>(1));
return; return;
} }
forward_message(parent_process_ptr(), sender, std::move(msg)); forward_message(parent_process_ptr(), sender, 0, std::move(msg));
}
void actor_proxy::sync_enqueue(actor* sender,
std::uint64_t response_id,
any_tuple msg ) {
forward_message(parent_process_ptr(), sender, response_id, std::move(msg));
} }
void actor_proxy::link_to(intrusive_ptr<actor>& other) { void actor_proxy::link_to(intrusive_ptr<actor>& other) {
...@@ -68,7 +75,8 @@ void actor_proxy::link_to(intrusive_ptr<actor>& other) { ...@@ -68,7 +75,8 @@ void actor_proxy::link_to(intrusive_ptr<actor>& other) {
// causes remote actor to link to (proxy of) other // causes remote actor to link to (proxy of) other
forward_message(parent_process_ptr(), forward_message(parent_process_ptr(),
other.get(), other.get(),
make_cow_tuple(atom("LINK"), other)); 0,
make_any_tuple(atom("LINK"), other));
} }
} }
...@@ -81,7 +89,8 @@ void actor_proxy::unlink_from(intrusive_ptr<actor>& other) { ...@@ -81,7 +89,8 @@ void actor_proxy::unlink_from(intrusive_ptr<actor>& other) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(), forward_message(parent_process_ptr(),
other.get(), other.get(),
make_cow_tuple(atom("UNLINK"), other)); 0,
make_any_tuple(atom("UNLINK"), other));
} }
} }
...@@ -95,7 +104,8 @@ bool actor_proxy::establish_backlink(intrusive_ptr<actor>& other) { ...@@ -95,7 +104,8 @@ bool actor_proxy::establish_backlink(intrusive_ptr<actor>& other) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(), forward_message(parent_process_ptr(),
other.get(), other.get(),
make_cow_tuple(atom("LINK"), other)); 0,
make_any_tuple(atom("LINK"), other));
} }
return result; return result;
} }
...@@ -105,7 +115,8 @@ bool actor_proxy::remove_backlink(intrusive_ptr<actor>& other) { ...@@ -105,7 +115,8 @@ bool actor_proxy::remove_backlink(intrusive_ptr<actor>& other) {
if (result) { if (result) {
forward_message(parent_process_ptr(), forward_message(parent_process_ptr(),
nullptr, nullptr,
make_cow_tuple(atom("UNLINK"), actor_ptr(this))); 0,
make_any_tuple(atom("UNLINK"), actor_ptr(this)));
} }
return result; return result;
} }
......
...@@ -33,13 +33,17 @@ ...@@ -33,13 +33,17 @@
namespace cppa { namespace detail { namespace cppa { namespace detail {
addressed_message::addressed_message(actor_ptr from, channel_ptr to, any_tuple ut) addressed_message::addressed_message(actor_ptr from,
: m_sender(std::move(from)), m_receiver(std::move(to)), m_content(std::move(ut)) { channel_ptr to,
} std::uint64_t seq_id,
any_tuple ut )
: m_sender(std::move(from)), m_receiver(std::move(to))
, m_sequence_id(seq_id), m_content(std::move(ut)) { }
bool operator==(const addressed_message& lhs, const addressed_message& rhs) { bool operator==(const addressed_message& lhs, const addressed_message& rhs) {
return lhs.sender() == rhs.sender() return lhs.sender() == rhs.sender()
&& lhs.receiver() == rhs.receiver() && lhs.receiver() == rhs.receiver()
&& lhs.sequence_id() == rhs.sequence_id()
&& lhs.content() == rhs.content(); && lhs.content() == rhs.content();
} }
......
...@@ -267,7 +267,7 @@ class po_peer : public po_socket_handler { ...@@ -267,7 +267,7 @@ class po_peer : public po_socket_handler {
else if (receiver->parent_process() == *process_information::get()) { else if (receiver->parent_process() == *process_information::get()) {
// this message was send from a proxy // this message was send from a proxy
receiver->attach_functor([=](std::uint32_t reason) { receiver->attach_functor([=](std::uint32_t reason) {
addressed_message kmsg{receiver, receiver, make_any_tuple(atom("KILL_PROXY"), reason)}; addressed_message kmsg{receiver, receiver, 0, make_any_tuple(atom("KILL_PROXY"), reason)};
singleton_manager::get_network_manager() singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(m_peer, kmsg)); ->send_to_mailman(make_any_tuple(m_peer, kmsg));
}); });
......
...@@ -45,7 +45,7 @@ ...@@ -45,7 +45,7 @@
#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"
namespace { namespace cppa { namespace {
typedef std::uint32_t ui32; typedef std::uint32_t ui32;
...@@ -64,9 +64,55 @@ inline decltype(std::chrono::high_resolution_clock::now()) now() { ...@@ -64,9 +64,55 @@ inline decltype(std::chrono::high_resolution_clock::now()) now() {
return std::chrono::high_resolution_clock::now(); return std::chrono::high_resolution_clock::now();
} }
} // namespace <anonymous> void async_send_impl(void* to, actor* from, std::uint64_t, const any_tuple& msg) {
reinterpret_cast<channel*>(to)->enqueue(from, msg);
}
void sync_reply_impl(void* to, actor* from, std::uint64_t id, const any_tuple& msg) {
reinterpret_cast<actor*>(to)->sync_enqueue(from, id, msg);
}
typedef void (*impl_fun_ptr)(void*, actor*, std::uint64_t, const any_tuple&);
class delayed_msg {
namespace cppa { public:
delayed_msg(impl_fun_ptr arg0,
const channel_ptr& arg1,
const actor_ptr& arg2,
std::uint64_t arg3,
const any_tuple& arg4)
: fun(arg0), ptr_a(arg1), ptr_b(), from(arg2), id(arg3), msg(arg4) { }
delayed_msg(impl_fun_ptr arg0,
const actor_ptr& arg1,
const actor_ptr& arg2,
std::uint64_t arg3,
const any_tuple& arg4)
: fun(arg0), ptr_a(), ptr_b(arg1), from(arg2), id(arg3), msg(arg4) { }
delayed_msg(delayed_msg&&) = default;
delayed_msg(const delayed_msg&) = default;
delayed_msg& operator=(delayed_msg&&) = default;
delayed_msg& operator=(const delayed_msg&) = default;
inline void eval() {
fun((ptr_a) ? ptr_a.get() : ptr_b.get(), from.get(), id, msg);
}
private:
impl_fun_ptr fun;
channel_ptr ptr_a;
actor_ptr ptr_b;
actor_ptr from;
std::uint64_t id;
any_tuple msg;
};
} // namespace <anonymous>
class scheduler_helper { class scheduler_helper {
...@@ -74,15 +120,14 @@ class scheduler_helper { ...@@ -74,15 +120,14 @@ class scheduler_helper {
typedef intrusive_ptr<thread_mapped_actor> ptr_type; typedef intrusive_ptr<thread_mapped_actor> ptr_type;
scheduler_helper() : m_worker(new thread_mapped_actor) { scheduler_helper() : m_worker(new thread_mapped_actor) { }
}
void start() { void start() {
m_thread = std::thread(&scheduler_helper::time_emitter, m_worker); m_thread = std::thread(&scheduler_helper::time_emitter, m_worker);
} }
void stop() { void stop() {
m_worker->enqueue(nullptr, make_cow_tuple(atom(":_DIE"))); m_worker->enqueue(nullptr, make_cow_tuple(atom("DIE")));
m_thread.join(); m_thread.join();
} }
...@@ -95,28 +140,47 @@ class scheduler_helper { ...@@ -95,28 +140,47 @@ class scheduler_helper {
}; };
template<class Map, class T>
void insert_dmsg(Map& storage,
const util::duration& d,
impl_fun_ptr fun_ptr,
const T& to,
const actor_ptr& sender,
std::uint64_t id,
const any_tuple& tup ) {
auto tout = now();
tout += d;
delayed_msg dmsg{fun_ptr, to, sender, id, tup};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) { void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
typedef detail::abstract_actor<local_actor> impl_type; typedef detail::abstract_actor<local_actor> impl_type;
typedef std::unique_ptr<detail::recursive_queue_node> queue_node_ptr; typedef std::unique_ptr<detail::recursive_queue_node> queue_node_ptr;
// setup & local variables // setup & local variables
self.set(m_self.get()); self.set(m_self.get());
auto& queue = m_self->mailbox(); auto& queue = m_self->mailbox();
std::multimap<decltype(now()), queue_node_ptr> messages; std::multimap<decltype(now()), delayed_msg> messages;
queue_node_ptr msg_ptr; queue_node_ptr msg_ptr;
//decltype(queue.pop()) msg_ptr = nullptr; //decltype(queue.pop()) msg_ptr = nullptr;
decltype(now()) tout; decltype(now()) tout;
bool done = false; bool done = false;
// message handling rules // message handling rules
auto mfun = ( auto mfun = (
on<util::duration,channel_ptr,anything>() >> [&](const util::duration& d, on(atom("SEND"), arg_match) >> [&](const util::duration& d,
const channel_ptr&) { const channel_ptr& ptr,
// calculate timeout const any_tuple& tup) {
auto timeout = now(); insert_dmsg(messages, d, async_send_impl,
timeout += d; ptr, msg_ptr->sender, 0, tup);
messages.insert(std::make_pair(std::move(timeout), },
std::move(msg_ptr))); on(atom("REPLY"), arg_match) >> [&](const util::duration& d,
const actor_ptr& ptr,
std::uint64_t id,
const any_tuple& tup) {
insert_dmsg(messages, d, sync_reply_impl,
ptr, msg_ptr->sender, id, tup);
}, },
on<atom(":_DIE")>() >> [&]() { on<atom("DIE")>() >> [&]() {
done = true; done = true;
}, },
others() >> [&]() { others() >> [&]() {
...@@ -125,7 +189,6 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) { ...@@ -125,7 +189,6 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
<< to_string(msg_ptr->msg) << to_string(msg_ptr->msg)
<< std::endl; << std::endl;
# endif # endif
msg_ptr.reset();
} }
); );
// loop // loop
...@@ -139,15 +202,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) { ...@@ -139,15 +202,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
// handle timeouts (send messages) // handle timeouts (send messages)
auto it = messages.begin(); auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) { while (it != messages.end() && (it->first) <= tout) {
queue_node_ptr ptr{std::move(it->second)}; it->second.eval();
CPPA_REQUIRE(ptr->marked == false);
auto whom = const_cast<actor_ptr*>(
reinterpret_cast<const actor_ptr*>(
ptr->msg.at(1)));
if (*whom) {
any_tuple msg = *(reinterpret_cast<const any_tuple*>(
ptr->msg.at(2))); (*whom)->enqueue(ptr->sender.get(), std::move(msg));
}
messages.erase(it); messages.erase(it);
it = messages.begin(); it = messages.begin();
} }
...@@ -158,6 +213,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) { ...@@ -158,6 +213,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self) {
} }
} }
mfun(msg_ptr->msg); mfun(msg_ptr->msg);
msg_ptr.reset();
} }
} }
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#include "cppa/self.hpp" #include "cppa/self.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/thread_mapped_actor.hpp" #include "cppa/thread_mapped_actor.hpp"
#include "cppa/detail/matches.hpp" #include "cppa/detail/matches.hpp"
...@@ -56,34 +57,17 @@ void thread_mapped_actor::quit(std::uint32_t reason) { ...@@ -56,34 +57,17 @@ void thread_mapped_actor::quit(std::uint32_t reason) {
} }
void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) { void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) {
auto node = fetch_node(sender, std::move(msg)); m_mailbox.push_back(fetch_node(sender, std::move(msg)));
CPPA_REQUIRE(node->marked == false);
m_mailbox.push_back(node);
} }
bool thread_mapped_actor::initialized() { void thread_mapped_actor::sync_enqueue(actor* sender,
return m_initialized; std::uint64_t response_id,
any_tuple msg ) {
m_mailbox.push_back(fetch_node(sender, std::move(msg), response_id));
} }
detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) { bool thread_mapped_actor::initialized() {
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr; return m_initialized;
if ( m_trap_exit == false
&& msg.size() == 2
&& msg.type_at(0) == arr[0]
&& msg.type_at(1) == arr[1]) {
auto v0 = msg.get_as<atom_value>(0);
auto v1 = msg.get_as<std::uint32_t>(1);
if (v0 == atom("EXIT")) {
if (v1 != exit_reason::normal) {
quit(v1);
}
return detail::normal_exit_signal;
}
}
return detail::ordinary_message;
} }
} // namespace cppa::detail } // namespace cppa::detail
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