Commit 04ddeb67 authored by neverlord's avatar neverlord

implemented reply()

parent 7832e8c1
......@@ -14,16 +14,16 @@ class actor_proxy : public detail::abstract_actor<actor>
typedef detail::abstract_actor<actor> super;
// implemented in unicast_network.cpp
static void forward_message(const process_information_ptr&, const any_tuple&);
void forward_message(const process_information_ptr&,
actor*, const any_tuple&);
public:
actor_proxy(std::uint32_t mid, const process_information_ptr& parent);
void enqueue(any_tuple&& msg);
void enqueue(actor* sender, any_tuple&& msg);
void enqueue(const any_tuple& msg);
void enqueue(actor* sender, const any_tuple& msg);
void link_to(intrusive_ptr<actor>& other);
......
......@@ -34,9 +34,9 @@ class channel : public ref_counted
/**
* @brief Enqueues @p msg to the list of received messages.
*/
virtual void enqueue(const any_tuple& msg) = 0;
virtual void enqueue(actor* sender, const any_tuple& msg) = 0;
virtual void enqueue(any_tuple&& msg) = 0;
virtual void enqueue(actor* sender, any_tuple&& msg) = 0;
};
......
......@@ -351,6 +351,15 @@ inline any_tuple& last_received()
return self()->mailbox().last_dequeued();
}
/**
* @brief Gets the sender of the last received message.
* @returns An {@link actor_ptr} to the sender of the last received message.
*/
inline actor_ptr& last_sender()
{
return self()->mailbox().last_sender();
}
#ifdef CPPA_DOCUMENTATION
/**
......@@ -384,7 +393,7 @@ template<class C, typename Arg0, typename... Args>
typename util::enable_if<std::is_base_of<channel, C>, void>::type
send(intrusive_ptr<C>& whom, const Arg0& arg0, const Args&... args)
{
if (whom) whom->enqueue(make_tuple(arg0, args...));
if (whom) whom->enqueue(self(), make_tuple(arg0, args...));
}
template<class C, typename Arg0, typename... Args>
......@@ -392,21 +401,42 @@ typename util::enable_if<std::is_base_of<channel, C>, void>::type
send(intrusive_ptr<C>&& whom, const Arg0& arg0, const Args&... args)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(make_tuple(arg0, args...));
if (tmp) tmp->enqueue(self(), make_tuple(arg0, args...));
}
// matches send(self(), ...);
template<typename Arg0, typename... Args>
void send(local_actor* whom, const Arg0& arg0, const Args&... args)
{
if (whom) whom->enqueue(make_tuple(arg0, args...));
if (whom) whom->enqueue(self(), make_tuple(arg0, args...));
}
template<class C>
typename util::enable_if<std::is_base_of<channel, C>, void>::type
send_tuple(intrusive_ptr<C>& whom, const any_tuple& what)
{
if (whom) whom->enqueue(self(), what);
}
template<class C>
typename util::enable_if<std::is_base_of<channel, C>, void>::type
send_tuple(intrusive_ptr<C>&& whom, const any_tuple& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(self(), what);
}
// matches send(self(), ...);
inline void send_tuple(local_actor* whom, const any_tuple& what)
{
if (whom) whom->enqueue(self(), what);
}
template<class C>
typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>&>::type
operator<<(intrusive_ptr<C>& whom, const any_tuple& what)
{
if (whom) whom->enqueue(what);
if (whom) whom->enqueue(self(), what);
return whom;
}
......@@ -415,7 +445,7 @@ typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>>::type
operator<<(intrusive_ptr<C>&& whom, const any_tuple& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(what);
if (tmp) tmp->enqueue(self(), what);
return std::move(tmp);
}
......@@ -423,7 +453,7 @@ template<class C>
typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>&>::type
operator<<(intrusive_ptr<C>& whom, any_tuple&& what)
{
if (whom) whom->enqueue(std::move(what));
if (whom) whom->enqueue(self(), std::move(what));
return whom;
}
......@@ -432,7 +462,7 @@ typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>>::type
operator<<(intrusive_ptr<C>&& whom, any_tuple&& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(std::move(what));
if (tmp) tmp->enqueue(self(), std::move(what));
return std::move(tmp);
}
......@@ -444,6 +474,17 @@ local_actor* operator<<(local_actor* whom, any_tuple&& what);
#endif
template<class C, typename Arg0, typename... Args>
void reply(const Arg0& arg0, const Args&... args)
{
send(self()->mailbox().last_sender(), arg0, args...);
}
inline void reply_tuple(const any_tuple& what)
{
send_tuple(self()->mailbox().last_sender(), what);
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
......
......@@ -107,11 +107,10 @@ class abstract_actor : public Base
}
if (!mlinks.empty())
{
actor_ptr mself = this;
// send exit messages
for (actor_ptr& aptr : mlinks)
{
aptr->enqueue(make_tuple(atom(":Exit"), mself, reason));
aptr->enqueue(this, make_tuple(atom(":Exit"), reason));
}
}
for (attachable_ptr& ptr : mattachables)
......@@ -228,7 +227,7 @@ class abstract_actor : public Base
}
if (reason != exit_reason::not_exited)
{
other->enqueue(make_tuple(atom(":Exit"), actor_ptr(this), reason));
other->enqueue(this, make_tuple(atom(":Exit"), reason));
}
return result;
}
......
......@@ -12,8 +12,6 @@ template<class Super>
class abstract_message_queue : public Super
{
any_tuple m_last_dequeued;
public:
template<typename... Args>
......@@ -25,9 +23,9 @@ class abstract_message_queue : public Super
{
for (;;)
{
if (Super::dequeue_impl(m_last_dequeued))
if (Super::dequeue_impl(this->m_last_dequeued))
{
return m_last_dequeued;
return this->m_last_dequeued;
}
}
}
......@@ -64,9 +62,8 @@ class abstract_message_queue : public Super
{
return false;
}
else if (Super::dequeue_impl(m_last_dequeued))
else if (Super::dequeue_impl(msg))
{
msg = m_last_dequeued;
return true;
}
}
......@@ -89,11 +86,6 @@ class abstract_message_queue : public Super
}
}
const any_tuple& last_dequeued() /*override*/
{
return m_last_dequeued;
}
};
} } // namespace cppa::detail
......
#ifndef ADDRESSED_MESSAGE_HPP
#define ADDRESSED_MESSAGE_HPP
#include <map>
#include <vector>
#include "cppa/actor.hpp"
#include "cppa/tuple.hpp"
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/process_information.hpp"
#include "cppa/tuple_view.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { class serializer; class deserializer; }
#include "cppa/detail/channel.hpp"
namespace cppa { namespace detail {
class addressed_message
{
friend struct addr_msg_tinfo;
public:
addressed_message() = default;
addressed_message(const actor_ptr& from,
const channel_ptr& to,
const any_tuple& ut);
addressed_message(addressed_message&&) = default;
addressed_message(const actor_ptr& from,
const channel_ptr& to,
any_tuple&& ut);
addressed_message() = default;
addressed_message(addressed_message&&) = default;
addressed_message(const addressed_message&) = default;
addressed_message& operator=(addressed_message&&) = default;
addressed_message& operator=(const addressed_message&) = default;
struct process_ptr_less
inline actor_ptr& sender()
{
bool operator()(const process_information_ptr& lhs,
const process_information_ptr& rhs);
};
return m_sender;
}
typedef std::map<process_information_ptr,
std::vector<actor_id>,
process_ptr_less> receiver_map;
inline const receiver_map& receivers() const;
inline const actor_ptr& sender() const
{
return m_sender;
}
inline receiver_map& receivers();
inline channel_ptr& receiver()
{
return m_receiver;
}
inline const any_tuple& content() const;
inline const channel_ptr& receiver() const
{
return m_receiver;
}
inline any_tuple& content();
inline any_tuple& content()
{
return m_content;
}
private:
inline const any_tuple& content() const
{
return m_content;
}
void serialize_to(serializer* sink) const;
inline bool empty() const
{
return m_content.empty();
}
void deserialize_from(deserializer* source);
private:
actor_ptr m_sender;
channel_ptr m_receiver;
any_tuple m_content;
receiver_map m_receivers;
};
bool operator==(const addressed_message& lhs, const addressed_message& rhs);
inline bool operator!=(const addressed_message& lhs,
const addressed_message& rhs)
inline bool operator!=(const addressed_message& lhs, const addressed_message& rhs)
{
return !(lhs == rhs);
}
inline const any_tuple& addressed_message::content() const
{
return m_content;
}
inline any_tuple& addressed_message::content()
{
return m_content;
}
inline const addressed_message::receiver_map&
addressed_message::receivers() const
{
return m_receivers;
}
inline addressed_message::receiver_map& addressed_message::receivers()
{
return m_receivers;
}
} } // namespace cppa::detail
#endif // ADDRESSED_MESSAGE_HPP
......@@ -21,9 +21,10 @@ class blocking_message_queue_impl : public message_queue
struct queue_node
{
queue_node* next;
actor_ptr sender;
any_tuple msg;
queue_node(any_tuple&& content);
queue_node(const any_tuple& content);
queue_node(actor* sender, any_tuple&& content);
queue_node(actor* sender, const any_tuple& content);
};
typedef util::single_reader_queue<queue_node> queue_type;
......@@ -38,9 +39,9 @@ class blocking_message_queue_impl : public message_queue
return m_queue;
}
virtual void enqueue(any_tuple&& msg) /*override*/;
void enqueue(actor* sender, any_tuple&& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
void enqueue(actor* sender, const any_tuple& msg) /*override*/;
protected:
......
......@@ -6,22 +6,32 @@
#include "cppa/process_information.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/util/single_reader_queue.hpp"
#include "cppa/detail/addressed_message.hpp"
namespace cppa { namespace detail {
struct mailman_send_job
{
process_information_ptr target_peer;
any_tuple original_message;
mailman_send_job(actor_proxy_ptr apptr, any_tuple msg);
mailman_send_job(process_information_ptr peer, any_tuple msg);
addressed_message msg;
inline mailman_send_job(process_information_ptr piptr,
const actor_ptr& from,
const channel_ptr& to,
const any_tuple& content)
: target_peer(piptr), msg(from, to, content)
{
}
};
struct mailman_add_peer
{
native_socket_t sockfd;
process_information_ptr pinfo;
mailman_add_peer(native_socket_t sfd, const process_information_ptr& pinf);
inline mailman_add_peer(native_socket_t fd,
const process_information_ptr& piptr)
: sockfd(fd), pinfo(piptr)
{
}
};
class mailman_job
......@@ -38,9 +48,10 @@ class mailman_job
kill_type
};
mailman_job(process_information_ptr piptr, const any_tuple& omsg);
mailman_job(actor_proxy_ptr apptr, const any_tuple& omsg);
mailman_job(process_information_ptr piptr,
const actor_ptr& from,
const channel_ptr& to,
const any_tuple& omsg);
mailman_job(native_socket_t sockfd, const process_information_ptr& pinfo);
......@@ -89,7 +100,7 @@ class mailman_job
mailman_add_peer m_add_socket;
};
mailman_job(job_type jt);
constexpr mailman_job(job_type jt) : next(nullptr), m_type(jt) { }
};
......
......@@ -34,9 +34,10 @@ class yielding_message_queue_impl : public message_queue
struct queue_node
{
queue_node* next;
actor_ptr sender;
any_tuple msg;
queue_node(any_tuple&& content);
queue_node(const any_tuple& content);
queue_node(actor* from, any_tuple&& content);
queue_node(actor* from, const any_tuple& content);
};
bool m_has_pending_timeout_request;
......@@ -101,9 +102,9 @@ class yielding_message_queue_impl : public message_queue
~yielding_message_queue_impl();
virtual void enqueue(any_tuple&& msg) /*override*/;
void enqueue(actor* sender, any_tuple&& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
void enqueue(actor* sender, const any_tuple& msg) /*override*/;
};
......
......@@ -41,9 +41,9 @@ class local_actor : public actor
*
* Calls <code>mailbox().enqueue(msg)</code>.
*/
virtual void enqueue(const any_tuple& msg) /*override*/;
void enqueue(actor* sender, const any_tuple& msg) /*override*/;
virtual void enqueue(any_tuple&& msg);
void enqueue(actor* sender, any_tuple&& msg) /*override*/;
inline bool trap_exit() const;
......
......@@ -19,6 +19,7 @@ class message_queue : public ref_counted
protected:
bool m_trap_exit;
actor_ptr m_last_sender;
any_tuple m_last_dequeued;
public:
......@@ -28,13 +29,13 @@ class message_queue : public ref_counted
*/
message_queue();
virtual void enqueue(any_tuple&& msg) = 0;
virtual void enqueue(actor* sender, any_tuple&& msg) = 0;
/**
* @brief Enqueues a new element to the message queue.
* @param msg The new message.
*/
virtual void enqueue(const any_tuple& msg) = 0;
virtual void enqueue(actor* sender, const any_tuple& msg) = 0;
/**
* @brief Dequeues the oldest message (FIFO order) from the message queue.
......@@ -90,6 +91,13 @@ class message_queue : public ref_counted
*/
inline any_tuple& last_dequeued();
/**
* @brief Gets the sender of the last dequeued message.
* @returns An {@link actor_ptr} to the sender of the last dequeued message.
* @warning Call only from the owner of the queue.
*/
inline actor_ptr& last_sender();
};
/******************************************************************************
......@@ -111,6 +119,11 @@ inline any_tuple& message_queue::last_dequeued()
return m_last_dequeued;
}
inline actor_ptr& message_queue::last_sender()
{
return m_last_sender;
}
} // namespace cppa
#endif // MESSAGE_QUEUE_HPP
......@@ -8,6 +8,7 @@
#include "cppa/atom.hpp"
#include "cppa/tuple.hpp"
#include "cppa/actor.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/attachable.hpp"
#include "cppa/scheduling_hint.hpp"
......@@ -17,11 +18,11 @@ namespace cppa {
// forward declarations
class local_actor;
//class local_actor;
class actor_behavior;
class scheduler_helper;
local_actor* self();
//local_actor* self();
/**
* @brief
......@@ -91,7 +92,7 @@ class scheduler
{
static_assert(sizeof...(Data) > 0, "no message to send");
any_tuple tup = make_tuple(util::duration(rel_time), to, data...);
future_send_helper()->enqueue(std::move(tup));
future_send_helper()->enqueue(self(), std::move(tup));
}
};
......
......@@ -23,18 +23,20 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
}
void actor_proxy::forward_message(const process_information_ptr& piptr,
actor* sender,
const any_tuple& msg)
{
detail::mailman_queue().push_back(new detail::mailman_job(piptr, msg));
auto mailman_msg = new detail::mailman_job(piptr, sender, this, msg);
detail::mailman_queue().push_back(mailman_msg);
}
void actor_proxy::enqueue(any_tuple&& msg)
void actor_proxy::enqueue(actor* sender, any_tuple&& msg)
{
any_tuple tmp(std::move(msg));
enqueue(tmp);
enqueue(sender, tmp);
}
void actor_proxy::enqueue(const any_tuple& msg)
void actor_proxy::enqueue(actor* sender, const any_tuple& msg)
{
if (msg.size() > 0 && msg.utype_info_at(0) == typeid(atom_value))
{
......@@ -65,7 +67,7 @@ void actor_proxy::enqueue(const any_tuple& msg)
return;
}
}
forward_message(parent_process_ptr(), msg);
forward_message(parent_process_ptr(), sender, msg);
}
void actor_proxy::link_to(intrusive_ptr<actor>& other)
......@@ -74,6 +76,7 @@ void actor_proxy::link_to(intrusive_ptr<actor>& other)
{
// causes remote actor to link to (proxy of) other
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Link"), actor_ptr(this)));
}
}
......@@ -84,6 +87,7 @@ void actor_proxy::unlink_from(intrusive_ptr<actor>& other)
{
// causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Unlink"), actor_ptr(this)));
}
}
......@@ -94,6 +98,7 @@ bool actor_proxy::establish_backlink(intrusive_ptr<actor>& other)
if (result)
{
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Link"), actor_ptr(this)));
}
return result;
......@@ -105,6 +110,7 @@ bool actor_proxy::remove_backlink(intrusive_ptr<actor>& other)
if (result)
{
forward_message(parent_process_ptr(),
nullptr,
make_tuple(atom(":Unlink"), actor_ptr(this)));
}
return result;
......
#include <algorithm>
#include "cppa/to_string.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/util/abstract_uniform_type_info.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa { namespace detail {
bool addressed_message::process_ptr_less::operator()
(const process_information_ptr& lhs, const process_information_ptr& rhs)
{
if (lhs && rhs)
{
return lhs.compare(rhs) < 0;
}
else if (!lhs && rhs)
{
return true;
}
return false;
}
bool operator==(const addressed_message& lhs, const addressed_message& rhs)
addressed_message::addressed_message(const actor_ptr& from,
const channel_ptr& to,
const any_tuple& ut)
: m_sender(from), m_receiver(to), m_content(ut)
{
return lhs.content() == rhs.content()
&& std::equal(lhs.receivers().begin(),
lhs.receivers().end(),
rhs.receivers().begin());
}
void addressed_message::serialize_to(serializer* sink) const
addressed_message::addressed_message(const actor_ptr& from,
const channel_ptr& to,
any_tuple&& ut)
: m_sender(from), m_receiver(to), m_content(std::move(ut))
{
primitive_variant ptup[2];
primitive_variant pval = static_cast<std::uint32_t>(m_receivers.size());
// size of the map
sink->write_value(pval);
for (auto i = m_receivers.begin(); i != m_receivers.end(); ++i)
{
if (i->first)
{
ptup[0] = i->first->process_id();
ptup[1] = to_string(i->first->node_id());
// process id as { process_id, node_id } tuple
sink->write_tuple(2, ptup);
// size of the vector that holds the actor ids
pval = static_cast<std::uint32_t>(i->second.size());
sink->write_value(pval);
for (auto j = i->second.begin(); j != i->second.end(); ++j)
{
pval = static_cast<std::uint32_t>(*j);
// actor id of one of the receivers
sink->write_value(pval);
}
}
}
}
void addressed_message::deserialize_from(deserializer* source)
bool operator==(const addressed_message& lhs, const addressed_message& rhs)
{
m_receivers.clear();
primitive_variant ptup[2];
primitive_type ptypes[] = { pt_uint32, pt_u8string };
// size of the map
auto map_size = get<std::uint32_t>(source->read_value(pt_uint32));
for (std::uint32_t i = 0; i < map_size; ++i)
{
// process id as { process_id, node_id } tuple
source->read_tuple(2, ptypes, ptup);
process_information_ptr pi;
pi.reset(new process_information(get<std::uint32_t>(ptup[0]),
get<std::string>(ptup[1])));
std::vector<actor_id>& vec = m_receivers[pi];
// size of vec
auto vec_size = get<std::uint32_t>(source->read_value(pt_uint32));
vec.reserve(vec_size);
// fill vector with the IDs of all receivers
for (decltype(vec_size) j = 0; j < vec_size; ++j)
{
vec.push_back(get<std::uint32_t>(source->read_value(pt_uint32)));
}
}
return lhs.sender() == rhs.sender()
&& lhs.receiver() == rhs.receiver()
&& lhs.content() == rhs.content();
}
} } // namespace cppa::detail
} } // namespace cppa
......@@ -43,13 +43,15 @@ throw_on_exit_result throw_on_exit(Pattern& exit_pattern, const any_tuple& msg)
namespace cppa { namespace detail {
blocking_message_queue_impl::queue_node::queue_node(any_tuple&& content)
: next(nullptr), msg(std::move(content))
blocking_message_queue_impl::queue_node::queue_node(actor* from,
any_tuple&& content)
: next(nullptr), sender(from), msg(std::move(content))
{
}
blocking_message_queue_impl::queue_node::queue_node(const any_tuple& content)
: next(nullptr), msg(content)
blocking_message_queue_impl::queue_node::queue_node(actor* from,
const any_tuple& content)
: next(nullptr), sender(from), msg(content)
{
}
......@@ -58,14 +60,14 @@ blocking_message_queue_impl::blocking_message_queue_impl()
{
}
void blocking_message_queue_impl::enqueue(any_tuple&& msg)
void blocking_message_queue_impl::enqueue(actor* sender, any_tuple&& msg)
{
m_queue.push_back(new queue_node(std::move(msg)));
m_queue.push_back(new queue_node(sender, std::move(msg)));
}
void blocking_message_queue_impl::enqueue(const any_tuple& msg)
void blocking_message_queue_impl::enqueue(actor* sender, const any_tuple& msg)
{
m_queue.push_back(new queue_node(msg));
m_queue.push_back(new queue_node(sender, msg));
}
bool blocking_message_queue_impl::dequeue_impl(any_tuple& storage)
......@@ -80,7 +82,9 @@ bool blocking_message_queue_impl::dequeue_impl(any_tuple& storage)
return false;
}
}
storage = std::move(node->msg);
m_last_dequeued = std::move(node->msg);
m_last_sender = std::move(node->sender);
storage = m_last_dequeued;
return true;
}
......@@ -96,7 +100,8 @@ bool blocking_message_queue_impl::dq(std::unique_ptr<queue_node>& node,
auto imd = rules.get_intermediate(node->msg);
if (imd)
{
m_last_dequeued = node->msg;
m_last_dequeued = std::move(node->msg);
m_last_sender = std::move(node->sender);
// restore mailbox before invoking imd
if (!buffer.empty()) m_queue.push_front(std::move(buffer));
imd->invoke();
......
......@@ -36,14 +36,14 @@ namespace cppa {
local_actor* operator<<(local_actor* whom, const any_tuple& what)
{
if (whom) whom->enqueue(what);
if (whom) whom->enqueue(self(), what);
return whom;
}
// matches self() << make_tuple(...)
local_actor* operator<<(local_actor* whom, any_tuple&& what)
{
if (whom) whom->enqueue(std::move(what));
if (whom) whom->enqueue(self(), std::move(what));
return whom;
}
......
......@@ -36,19 +36,19 @@ class local_group : public group
public:
virtual void enqueue(const any_tuple& msg)
void enqueue(actor* sender, const any_tuple& msg) /*override*/
{
shared_guard guard(m_shared_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
{
const_cast<channel_ptr&>(*i)->enqueue(msg);
const_cast<channel_ptr&>(*i)->enqueue(sender, msg);
}
}
virtual void enqueue(any_tuple&& msg)
void enqueue(actor* sender, any_tuple&& msg) /*override*/
{
any_tuple tmp(std::move(msg));
enqueue(tmp);
enqueue(sender, tmp);
}
virtual group::subscription subscribe(const channel_ptr& who)
......
......@@ -29,14 +29,14 @@ boost::thread_specific_ptr<cppa::local_actor> t_this_context(cleanup_fun);
namespace cppa {
void local_actor::enqueue(any_tuple&& msg)
void local_actor::enqueue(actor* sender, any_tuple&& msg)
{
mailbox().enqueue(std::move(msg));
mailbox().enqueue(sender, std::move(msg));
}
void local_actor::enqueue(const any_tuple& msg)
void local_actor::enqueue(actor* sender, const any_tuple& msg)
{
mailbox().enqueue(msg);
mailbox().enqueue(sender, msg);
}
local_actor* unchecked_self()
......
......@@ -12,42 +12,19 @@
// implementation of mailman.hpp
namespace cppa { namespace detail {
mailman_send_job::mailman_send_job(actor_proxy_ptr apptr, any_tuple msg)
: target_peer(apptr->parent_process_ptr()), original_message(msg)
mailman_job::mailman_job(process_information_ptr piptr,
const actor_ptr& from,
const channel_ptr& to,
const any_tuple& content)
: next(nullptr), m_type(send_job_type)
{
}
mailman_send_job::mailman_send_job(process_information_ptr peer, any_tuple msg)
: target_peer(peer), original_message(msg)
{
}
mailman_add_peer::mailman_add_peer(native_socket_t sfd,
const process_information_ptr& pinf)
: sockfd(sfd), pinfo(pinf)
{
}
mailman_job::mailman_job(job_type jt) : next(0), m_type(jt)
{
}
mailman_job::mailman_job(process_information_ptr piptr, const any_tuple& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(piptr, omsg);
}
mailman_job::mailman_job(actor_proxy_ptr apptr, const any_tuple& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(apptr, omsg);
new (&m_send_job) mailman_send_job (piptr, from, to, content);
}
mailman_job::mailman_job(native_socket_t sockfd, const process_information_ptr& pinfo)
: next(0), m_type(add_peer_type)
{
new (&m_add_socket) mailman_add_peer(sockfd, pinfo);
new (&m_add_socket) mailman_add_peer (sockfd, pinfo);
}
mailman_job* mailman_job::kill_job()
......@@ -103,7 +80,6 @@ void mailman_loop()
if (job->is_send_job())
{
mailman_send_job& sjob = job->send_job();
const any_tuple& out_msg = sjob.original_message;
// forward message to receiver peer
auto peer_element = peers.find(*(sjob.target_peer));
if (peer_element != peers.end())
......@@ -112,9 +88,9 @@ void mailman_loop()
auto peer = peer_element->second;
try
{
bs << out_msg;
bs << sjob.msg;
auto size32 = static_cast<std::uint32_t>(bs.size());
DEBUG("--> " << to_string(out_msg));
DEBUG("--> " << to_string(sjob.msg));
// write size of serialized message
auto sent = ::send(peer, &size32, sizeof(std::uint32_t), 0);
if (sent > 0)
......
......@@ -229,7 +229,8 @@ class po_peer : public post_office_worker
{
for (actor_proxy_ptr& pptr : m_children)
{
pptr->enqueue(make_tuple(atom(":KillProxy"),
pptr->enqueue(self(),
make_tuple(atom(":KillProxy"),
exit_reason::remote_link_unreachable));
}
}
......@@ -319,6 +320,7 @@ class po_peer : public post_office_worker
&& content.get_as<atom_value>(0) == atom(":Monitor")
&& content.utype_info_at(1) == typeid(actor_ptr))
{
/*
actor_ptr sender = content.get_as<actor_ptr>(1);
if (sender->parent_process() == *process_information::get())
{
......@@ -338,32 +340,15 @@ class po_peer : public post_office_worker
{
DEBUG(":Monitor received for an remote actor");
}
*/
cerr << "NOT IMPLEMENTED YET; post_office line "
<< __LINE__ << endl;
}
else
{
DEBUG("<-- " << to_string(content));
auto& rmap = msg.receivers();
for (auto i = rmap.begin(); i != rmap.end(); ++i)
{
if (*(i->first) == *process_information::get())
{
auto& vec = i->second;
for (auto j = vec.begin(); j != vec.end(); ++j)
{
auto registry = singleton_manager::get_actor_registry();
auto receiver = registry->find(*j);
if (receiver)
{
receiver->enqueue(msg.content());
}
}
}
else
{
cerr << "NOT IMPLEMENTED YET; post_office line "
<< __LINE__ << endl;
}
}
msg.receiver()->enqueue(msg.sender().get(),
std::move(msg.content()));
}
m_rdbuf.reset();
m_state = wait_for_msg_size;
......@@ -496,7 +481,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
// initialize proxy cache
get_actor_proxy_cache().set_callback([&](actor_proxy_ptr& pptr)
{
pptr->enqueue(make_tuple(atom(":Monitor"), pptr));
pptr->enqueue(nullptr, make_tuple(atom(":Monitor"), pptr));
if (selected_peer == nullptr)
{
throw std::logic_error("selected_peer == nullptr");
......
......@@ -46,7 +46,7 @@ struct scheduler_helper
void stop()
{
m_worker->enqueue(make_tuple(atom(":_DIE")));
m_worker->enqueue(nullptr, make_tuple(atom(":_DIE")));
m_thread.join();
}
......@@ -64,8 +64,8 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
// setup & local variables
set_self(m_self.get());
auto& queue = m_self->m_mailbox.queue();
typedef std::pair<cppa::actor_ptr, cppa::any_tuple> future_msg;
std::multimap<decltype(detail::now()), future_msg> messages;
//typedef std::pair<cppa::actor_ptr, decltype(queue.pop())> future_msg;
std::multimap<decltype(detail::now()), decltype(queue.pop())> messages;
decltype(queue.pop()) msg_ptr = nullptr;
decltype(detail::now()) now;
bool done = false;
......@@ -73,23 +73,21 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
auto rules =
(
on<util::duration,actor_ptr,anything>() >> [&](const util::duration& d,
const actor_ptr& whom)
const actor_ptr&)
// on<util::duration,anything>() >> [&](const util::duration& d)
{
any_tuple msg = msg_ptr->msg.tail(2);
if (!msg.empty())
{
// calculate timeout
auto timeout = detail::now();
timeout += d;
future_msg fmsg(whom, msg);
messages.insert(std::make_pair(std::move(timeout),
std::move(fmsg)));
}
//any_tuple msg = msg_ptr->msg.tail();
// calculate timeout
auto timeout = detail::now();
timeout += d;
messages.insert(std::make_pair(std::move(timeout),
std::move(msg_ptr)));
},
on<atom(":_DIE")>() >> [&]()
{
done = true;
}
},
others() >> []() { }
);
// loop
while (!done)
......@@ -107,11 +105,16 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
auto it = messages.begin();
while (it != messages.end() && (it->first) <= now)
{
auto& whom = (it->second).first;
auto& what = (it->second).second;
whom->enqueue(what);
auto ptr = it->second;
auto whom = const_cast<actor_ptr*>(reinterpret_cast<const actor_ptr*>(ptr->msg.at(1)));
if (*whom)
{
auto msg = ptr->msg.tail(2);
(*whom)->enqueue(ptr->sender.get(), std::move(msg));
}
messages.erase(it);
it = messages.begin();
delete ptr;
}
// wait for next message or next timeout
if (it != messages.end())
......@@ -121,7 +124,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
}
}
rules(msg_ptr->msg);
delete msg_ptr;
//delete msg_ptr;
msg_ptr = nullptr;
}
}
......
......@@ -438,32 +438,28 @@ class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple>
};
/*
class message_tinfo : public util::abstract_uniform_type_info<any_tuple>
class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message>
{
std::string any_tuple_name;
std::string actor_ptr_name;
std::string group_ptr_name;
std::string ch_ptr_name;
std::string channel_ptr_name;
public:
virtual void serialize(const void* instance, serializer* sink) const
{
const any_tuple& msg = *reinterpret_cast<const any_tuple*>(instance);
const addressed_message& msg = *reinterpret_cast<const addressed_message*>(instance);
const any_tuple& data = msg.content();
sink->begin_object(name());
actor_ptr_tinfo::s_serialize(msg.sender().get(), sink, actor_ptr_name);
channel_ptr_tinfo::s_serialize(msg.receiver().get(),
sink,
ch_ptr_name,
channel_ptr_name,
actor_ptr_name,
group_ptr_name);
any_tuple_tinfo::s_serialize(data, sink, any_tuple_name);
//uniform_typeid<actor_ptr>()->serialize(&(msg.sender()), sink);
//uniform_typeid<channel_ptr>()->serialize(&(msg.receiver()), sink);
//uniform_typeid<any_tuple>()->serialize(&data, sink);
sink->end_object();
}
......@@ -472,34 +468,31 @@ class message_tinfo : public util::abstract_uniform_type_info<any_tuple>
auto tname = source->seek_object();
if (tname != name()) throw 42;
source->begin_object(tname);
actor_ptr sender;
channel_ptr receiver;
any_tuple content;
actor_ptr_tinfo::s_deserialize(sender, source, actor_ptr_name);
channel_ptr_tinfo::s_deserialize(receiver,
addressed_message& msg = *reinterpret_cast<addressed_message*>(instance);
//actor_ptr sender;
//channel_ptr receiver;
//any_tuple content;
actor_ptr_tinfo::s_deserialize(msg.sender(), source, actor_ptr_name);
channel_ptr_tinfo::s_deserialize(msg.receiver(),
source,
ch_ptr_name,
channel_ptr_name,
actor_ptr_name,
group_ptr_name);
any_tuple_tinfo::s_deserialize(content, source, any_tuple_name);
//uniform_typeid<actor_ptr>()->deserialize(&sender, source);
//uniform_typeid<channel_ptr>()->deserialize(&receiver, source);
//uniform_typeid<any_tuple>()->deserialize(&content, source);
any_tuple_tinfo::s_deserialize(msg.content(), source, any_tuple_name);
source->end_object();
*reinterpret_cast<any_tuple*>(instance) = any_tuple(sender,
receiver,
content);
//*reinterpret_cast<any_tuple*>(instance) = any_tuple(sender,
// receiver,
// content);
}
message_tinfo() : any_tuple_name(to_uniform_name(typeid(any_tuple)))
, actor_ptr_name(to_uniform_name(typeid(actor_ptr)))
, group_ptr_name(to_uniform_name(typeid(group_ptr)))
, ch_ptr_name(to_uniform_name(typeid(channel_ptr)))
addr_msg_tinfo() : any_tuple_name(to_uniform_name(typeid(any_tuple)))
, actor_ptr_name(to_uniform_name(typeid(actor_ptr)))
, group_ptr_name(to_uniform_name(typeid(group_ptr)))
, channel_ptr_name(to_uniform_name(typeid(channel_ptr)))
{
}
};
*/
class atom_value_tinfo : public util::abstract_uniform_type_info<atom_value>
{
......@@ -593,32 +586,6 @@ class int_tinfo : public detail::default_uniform_type_info_impl<T>
};
struct addr_msg_tinfo : util::abstract_uniform_type_info<addressed_message>
{
protected:
void serialize(const void* ptr, serializer* sink) const
{
sink->begin_object(name());
reinterpret_cast<const addressed_message*>(ptr)->serialize_to(sink);
sink->end_object();
}
void deserialize(void* ptr, deserializer* source) const
{
std::string cname = source->seek_object();
if (cname != name())
{
throw std::logic_error("wrong type name found");
}
source->begin_object(name());
reinterpret_cast<addressed_message*>(ptr)->deserialize_from(source);
source->end_object();
}
};
//} } } // namespace cppa::detail::<anonymous>
//namespace cppa { namespace detail {
......
......@@ -28,13 +28,14 @@ typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard;
namespace cppa { namespace detail {
yielding_message_queue_impl::queue_node::queue_node(any_tuple&& content)
: next(0), msg(std::move(content))
yielding_message_queue_impl::queue_node::queue_node(actor* ptr, any_tuple&& tup)
: next(0), sender(ptr), msg(std::move(tup))
{
}
yielding_message_queue_impl::queue_node::queue_node(const any_tuple& content)
: next(0), msg(content)
yielding_message_queue_impl::queue_node::queue_node(actor* ptr,
const any_tuple& tup)
: next(0), sender(ptr), msg(tup)
{
}
......@@ -63,13 +64,12 @@ yielding_message_queue_impl::filter_result
yielding_message_queue_impl::filter_msg(const any_tuple& msg)
{
if ( m_trap_exit == false
&& msg.size() == 3
&& msg.size() == 2
&& m_atom_value_uti == &(msg.utype_info_at(0)))
{
if (*reinterpret_cast<const atom_value*>(msg.at(0)) == atom(":Exit"))
{
if ( m_actor_ptr_uti == &(msg.utype_info_at(1))
&& m_ui32_uti == &(msg.utype_info_at(2)))
if (m_ui32_uti == &(msg.utype_info_at(1)))
{
auto why = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (why != exit_reason::normal)
......@@ -91,29 +91,6 @@ yielding_message_queue_impl::filter_msg(const any_tuple& msg)
: expired_timeout_message;
}
}
/*
if ( m_trap_exit == false
&& match<atom_value, actor_ptr, std::uint32_t>(msg,
nullptr,
atom(":Exit")))
{
auto why = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (why != cppa::exit_reason::normal)
{
// yields
cppa::self()->quit(why);
}
return normal_exit_signal;
}
if (match<atom_value, std::uint32_t>(msg,
nullptr,
atom(":Timeout")))
{
auto id = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
return (id == m_active_timeout_id) ? timeout_message
: expired_timeout_message;
}
*/
return ordinary_message;
}
......@@ -147,14 +124,14 @@ void yielding_message_queue_impl::enqueue_node(queue_node* node)
}
}
void yielding_message_queue_impl::enqueue(any_tuple&& msg)
void yielding_message_queue_impl::enqueue(actor* sender, any_tuple&& msg)
{
enqueue_node(new queue_node(std::move(msg)));
enqueue_node(new queue_node(sender, std::move(msg)));
}
void yielding_message_queue_impl::enqueue(const any_tuple& msg)
void yielding_message_queue_impl::enqueue(actor* sender, const any_tuple& msg)
{
enqueue_node(new queue_node(msg));
enqueue_node(new queue_node(sender, msg));
}
void yielding_message_queue_impl::yield_until_not_empty()
......
......@@ -17,21 +17,15 @@ void ping()
{
s_pongs = 0;
actor_ptr pong_actor;
auto response = make_tuple(atom("Ping"), 0);
// invoke rule
receive_loop
(
on<atom("Pong"), std::int32_t>() >> [&](std::int32_t value)
{
++s_pongs;
auto msg = tuple_cast<atom_value, std::int32_t>(std::move(last_received()));
get_ref<0>(msg) = atom("Ping");
get_ref<1>(msg) = value + 1;
pong_actor->enqueue(std::move(msg));
//send(pong_actor, atom("Ping"), value + 1);
},
on<atom("Hello"), actor_ptr>() >> [&](actor_ptr sender)
{
pong_actor = sender;
get_ref<1>(response) = value + 1;
reply_tuple(response);
}
);
}
......@@ -39,26 +33,22 @@ void ping()
void pong(actor_ptr ping_actor)
{
link(ping_actor);
// tell ping who we are
send(ping_actor, atom("Hello"), self());
auto pong_tuple = make_tuple(atom("Pong"), 0);
// kickoff
send(ping_actor, atom("Pong"), static_cast<std::int32_t>(0));
send_tuple(ping_actor, pong_tuple);
// invoke rules
receive_loop
(
on(atom("Ping"), std::int32_t(9)) >> []()
on(atom("Ping"), 9) >> []()
{
// terminate with non-normal exit reason
// to force ping actor to quit
quit(exit_reason::user_defined);
},
on<atom("Ping"), std::int32_t>() >> [&](std::int32_t value)
on<atom("Ping"), int>() >> [&](int value)
{
auto msg = tuple_cast<atom_value, std::int32_t>(std::move(last_received()));
get_ref<0>(msg) = atom("Pong");
get_ref<1>(msg) = value + 1;
ping_actor->enqueue(std::move(msg));
//send(ping_actor, atom("Pong"), value + 1);
get_ref<1>(pong_tuple) = value + 1;
reply_tuple(pong_tuple);
}
);
}
......
......@@ -92,11 +92,10 @@ size_t test__spawn()
// wait for :Down and :Exit messages of pong
receive_while([&i]() { return ++i <= 3 /*4*/; })
(
on<atom(":Exit"), actor_ptr, std::uint32_t>() >> [&](const actor_ptr& who,
std::uint32_t reason)
on<atom(":Exit"), std::uint32_t>() >> [&](std::uint32_t reason)
{
CPPA_CHECK_EQUAL(reason, exit_reason::user_defined);
CPPA_CHECK_EQUAL(who, pong_actor);
//CPPA_CHECK_EQUAL(who, pong_actor);
flags |= 0x01;
},
on<atom(":Down"), actor_ptr, std::uint32_t>() >> [&](const actor_ptr& who,
......
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