Commit 8d1b0361 authored by Dominik Charousset's avatar Dominik Charousset

replaced 'EXIT' and 'DOWN' messages w/ msg types

parent d2afafff
......@@ -140,7 +140,6 @@ cppa/response_promise.hpp
cppa/sb_actor.hpp
cppa/scheduler.hpp
cppa/scoped_actor.hpp
cppa/send.hpp
cppa/serializer.hpp
cppa/singletons.hpp
cppa/spawn.hpp
......@@ -279,7 +278,6 @@ src/response_promise.cpp
src/ripemd_160.cpp
src/scheduler.cpp
src/scoped_actor.cpp
src/send.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
......@@ -339,3 +337,4 @@ src/functor_based_blocking_actor.cpp
src/blocking_untyped_actor.cpp
cppa/policy/policies.hpp
cppa/detail/response_future_util.hpp
cppa/system_messages.hpp
......@@ -32,6 +32,7 @@
#define CPPA_BEHAVIOR_STACK_BASED_HPP
#include "cppa/message_id.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/detail/behavior_stack.hpp"
......@@ -163,7 +164,7 @@ class behavior_stack_based : public Base {
if (d.valid()) {
m_has_timeout = true;
auto tid = ++m_timeout_id;
auto msg = make_any_tuple(atom("SYNC_TOUT"), tid);
auto msg = make_any_tuple(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue({this->address(), this}, std::move(msg));
......
......@@ -59,6 +59,7 @@
#include "cppa/spawn_options.hpp"
#include "cppa/untyped_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/response_promise.hpp"
#include "cppa/blocking_untyped_actor.hpp"
......
......@@ -298,7 +298,7 @@ class proper_actor<Base, Policies,true> : public proper_actor_base<Base,
std::uint32_t request_timeout(const util::duration& d) {
CPPA_REQUIRE(d.valid());
auto tid = ++m_next_timeout_id;
auto msg = make_any_tuple(atom("SYNC_TOUT"), tid);
auto msg = make_any_tuple(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue({this->address(), this}, std::move(msg));
......
......@@ -33,6 +33,7 @@
#include "cppa/on.hpp"
#include "cppa/match_hint.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/util/type_traits.hpp"
......@@ -46,7 +47,7 @@ behavior fs2bhvr(Actor* self, Fs... fs) {
return match_hint::skip;
};
return behavior{
on(atom("TIMEOUT")) >> handle_sync_timeout,
on<sync_timeout_msg>() >> handle_sync_timeout,
on(atom("VOID")) >> skip_message,
on(atom("EXITED")) >> skip_message,
(on(any_vals, arg_match) >> std::move(fs))...
......
......@@ -43,6 +43,7 @@
#include "cppa/unit.hpp"
#include "cppa/none.hpp"
#include "cppa/node_id.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/util/buffer.hpp"
#include "cppa/util/duration.hpp"
......@@ -65,11 +66,15 @@ using mapped_type_list = util::type_list<
any_tuple,
atom_value,
channel,
down_msg,
exit_msg,
group_ptr,
node_id_ptr,
io::accept_handle,
io::connection_handle,
message_header,
sync_timeout_msg,
timeout_msg,
unit_t,
util::buffer,
util::duration,
......
......@@ -43,6 +43,7 @@
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/response_promise.hpp"
......@@ -152,34 +153,32 @@ class invoke_policy {
filter_result filter_msg(Actor* self, mailbox_element* node) {
const any_tuple& msg = node->msg;
auto mid = node->mid;
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")) {
auto& arr = detail::static_types_array<exit_msg,
timeout_msg,
sync_timeout_msg>::arr;
if (msg.size() == 1) {
if (msg.type_at(0) == arr[0]) {
auto& em = msg.get_as<exit_msg>(0);
CPPA_REQUIRE(!mid.valid());
if (self->trap_exit() == false) {
if (v1 != exit_reason::normal) {
self->quit(v1);
if (em.reason != exit_reason::normal) {
self->quit(em.reason);
return non_normal_exit_signal;
}
return normal_exit_signal;
}
}
else if (v0 == atom("SYNC_TOUT")) {
else if (msg.type_at(0) == arr[1]) {
auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CPPA_REQUIRE(!mid.valid());
if (self->is_active_timeout(v1)) return timeout_message;
return self->waits_for_timeout(v1) ? inactive_timeout_message
: expired_timeout_message;
if (self->is_active_timeout(tid)) return timeout_message;
return self->waits_for_timeout(tid) ? inactive_timeout_message
: expired_timeout_message;
}
else if (msg.type_at(0) == arr[2] && mid.is_response()) {
return timeout_response_message;
}
}
else if ( msg.size() == 1
&& msg.type_at(0) == arr[0]
&& msg.get_as<atom_value>(0) == atom("TIMEOUT")
&& mid.is_response()) {
return timeout_response_message;
}
if (mid.is_response()) {
return (self->awaits(mid)) ? sync_response
......@@ -322,8 +321,8 @@ class invoke_policy {
}
case timeout_message: {
CPPA_LOG_DEBUG("handle timeout message");
auto tid = node->msg.get_as<std::uint32_t>(1);
self->handle_timeout(fun, tid);
auto& tm = node->msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_SEND_HPP
#define CPPA_SEND_HPP
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/message_header.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
/**
* @ingroup MessageHandling
* @{
*/
using channel_destination = destination_header<channel>;
using actor_destination = destination_header<abstract_actor_ptr>;
/**
* @brief Sends @p what to the receiver specified in @p hdr.
*/
inline void send_tuple(channel_destination dest, any_tuple what) {
if (dest.receiver == nullptr) return;
auto s = self.get();
message_header fhdr{s, std::move(dest.receiver), dest.priority};
if (fhdr.receiver != s && s->chaining_enabled()) {
if (fhdr.receiver->chained_enqueue(fhdr, std::move(what))) {
// only actors implement chained_enqueue to return true
s->chained_actor(fhdr.receiver.downcast<actor>());
}
}
else fhdr.deliver(std::move(what));
}
/**
* @brief Sends @p what to the receiver specified in @p hdr.
*/
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>& dest, Ts&&... what) {
static constexpr int input_pos = util::tl_find_if<
util::type_list<Signatures...>,
detail::input_is<util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>>::template eval
>::value;
static_assert(input_pos >= 0, "typed actor does not support given input");
send(dest./*receiver.*/type_erased(), std::forward<Ts>(what)...);
}
/**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
inline void send(channel_destination dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
send_tuple(std::move(dest), make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends @p what to @p whom, but sets the sender information to @p from.
*/
inline void send_tuple_as(actor_ptr from, channel_destination dest, any_tuple what) {
message_header mhdr{std::move(from), std::move(dest.receiver), dest.priority};
mhdr.deliver(std::move(what));
}
/**
* @brief Sends <tt>{what...}</tt> as a message to @p whom, but sets
* the sender information to @p from.
* @param from Sender as seen by @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
inline void send_as(actor_ptr from, channel_destination dest, Ts&&... what) {
send_tuple_as(std::move(from), std::move(dest),
make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future sync_send_tuple(actor_destination dest, any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_future sync_send(actor_destination dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(std::move(dest),
make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Signatures, typename... Ts>
typed_response_future<
typename detail::deduce_output_type<
util::type_list<Signatures...>,
util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>
>::type
>
sync_send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
return sync_send(whom.type_erased(), std::forward<Ts>(what)...);
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param data Message content as a tuple.
*/
void delayed_send_tuple(channel_destination dest,
const util::duration& rtime,
any_tuple data);
/**
* @brief Sends a reply message that is delayed by @p rel_time.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param what Message content as a tuple.
* @see delayed_send()
*/
void delayed_reply_tuple(const util::duration& rel_time,
message_id mid,
any_tuple data);
/**
* @brief Sends a reply message that is delayed by @p rel_time.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param what Message content as a tuple.
* @see delayed_send()
*/
void delayed_reply_tuple(const util::duration& rel_time, any_tuple data);
/**
* @brief Sends @p what as a synchronous message to @p whom with a timeout.
*
* The calling actor receives a 'TIMEOUT' message as response after
* given timeout exceeded and no response messages was received.
* @param dest Receiver and optional priority flag.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future timed_sync_send_tuple(actor_destination dest,
const util::duration& rtime,
any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom
* with a timeout.
*
* The calling actor receives a 'TIMEOUT' message as response after
* given timeout exceeded and no response messages was received.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
response_future timed_sync_send(actor_destination whom,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(std::move(whom),
rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple.
*/
CPPA_DEPRECATED inline void reply_tuple(any_tuple what) {
self->reply_message(std::move(what));
}
/**
* @brief Sends a message to the sender of the last received message.
* @param what Message elements.
*/
template<typename... Ts>
CPPA_DEPRECATED inline void reply(Ts&&... what) {
self->reply_message(make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends a message as reply to @p handle.
*/
template<typename... Ts>
inline void reply_to(const response_promise& handle, Ts&&... what) {
if (handle.valid()) {
handle.apply(make_any_tuple(std::forward<Ts>(what)...));
}
}
/**
* @brief Replies with @p what to @p handle.
* @param handle Identifies a previously received request.
* @param what Response message.
*/
inline void reply_tuple_to(const response_promise& handle, any_tuple what) {
handle.apply(std::move(what));
}
/**
* @brief Forwards the last received message to @p whom.
*/
inline void forward_to(actor_destination dest) {
self->forward_message(dest.receiver, dest.priority);
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param what Message elements.
*/
template<typename... Ts>
inline void delayed_send(channel_destination dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
if (dest.receiver) {
delayed_send_tuple(std::move(dest),
rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
}
/**
* @brief Sends a reply message that is delayed by @p rel_time.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param what Message elements.
* @see delayed_send()
*/
template<typename... Ts>
inline void delayed_reply(const util::duration& rtime, Ts&&... what) {
delayed_reply_tuple(rtime, make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends an exit message to @p whom with @p reason.
*
* This function is syntactic sugar for
* <tt>send(whom, atom("EXIT"), reason)</tt>.
* @pre <tt>reason != exit_reason::normal</tt>
*/
inline void send_exit(channel_destination dest, std::uint32_t rsn) {
CPPA_REQUIRE(rsn != exit_reason::normal);
send(std::move(dest), atom("EXIT"), rsn);
}
/**
* @brief Sends an exit message to @p whom with @p reason.
* @pre <tt>reason != exit_reason::normal</tt>
*/
template<typename... Signatures>
void send_exit(const typed_actor_ptr<Signatures...>& whom, std::uint32_t rsn) {
send_exit(whom.type_erased(), rsn);
}
/**
* @}
*/
} // namespace cppa
#endif // CPPA_SEND_HPP
......@@ -28,52 +28,65 @@
\******************************************************************************/
#include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp"
#ifndef CPPA_SYSTEM_MESSAGES_HPP
#define CPPA_SYSTEM_MESSAGES_HPP
namespace cppa {
#include <cstdint>
#include "cppa/actor_addr.hpp"
response_future sync_send_tuple(actor_destination dest, any_tuple what) {
if (!dest.receiver) throw std::invalid_argument("whom == nullptr");
auto req = self->new_request_id();
message_header hdr{self, std::move(dest.receiver), req, dest.priority};
if (self->chaining_enabled()) {
if (hdr.receiver->chained_enqueue(hdr, std::move(what))) {
self->chained_actor(hdr.receiver.downcast<actor>());
}
}
else hdr.deliver(std::move(what));
return req.response_id();
}
namespace cppa {
void delayed_send_tuple(channel_destination dest,
const util::duration& rtime,
any_tuple data) {
if (dest.receiver) {
message_header hdr{self, std::move(dest.receiver), dest.priority};
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(data));
}
}
/**
* @brief Sent to all links when an actor is terminated.
* @note This message can be handled manually by calling
* {@link local_actor::trap_exit(true) local_actor::trap_exit(bool)}
* and is otherwise handled implicitly by the runtime system.
*/
struct exit_msg {
/**
* @brief The source of this message, i.e., the terminated actor.
*/
actor_addr source;
/**
* @brief The exit reason of the terminated actor.
*/
std::uint32_t reason;
};
response_future timed_sync_send_tuple(actor_destination dest,
const util::duration& rtime,
any_tuple what) {
auto mf = sync_send_tuple(std::move(dest), std::move(what));
message_header hdr{self, self, mf.id()};
auto tmp = make_any_tuple(atom("TIMEOUT"));
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(tmp));
return mf;
}
/**
* @brief Sent to all actors monitoring an actor when it is terminated.
*/
struct down_msg {
/**
* @brief The source of this message, i.e., the terminated actor.
*/
actor_addr source;
/**
* @brief The exit reason of the terminated actor.
*/
std::uint32_t reason;
};
void delayed_reply_tuple(const util::duration& rtime,
message_id mid,
any_tuple data) {
message_header hdr{self, self->last_sender(), mid};
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(data));
}
/**
* @brief Sent whenever a timeout occurs during a synchronous send.
*
* This system message does not have any fields, because the message ID
* sent alongside this message identifies the matching request that timed out.
*/
struct sync_timeout_msg { };
void delayed_reply_tuple(const util::duration& rel_time, any_tuple data) {
delayed_reply_tuple(rel_time, self->get_response_id(), std::move(data));
}
/**
* @brief Signalizes a timeout event.
* @note This message is handled implicitly by the runtime system.
*/
struct timeout_msg {
/**
* @brief Actor-specific timeout ID.
*/
std::uint32_t timeout_id;
};
} // namespace cppa
#endif // CPPA_SYSTEM_MESSAGES_HPP
......@@ -83,7 +83,7 @@ void client_bhvr(untyped_actor* self, const string& host, uint16_t port, const a
}
);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t) {
on_arg_match >> [=](const down_msg&) {
aout << "*** server down, try to reconnect ..." << endl;
client_bhvr(self, host, port, nullptr);
},
......
......@@ -104,8 +104,8 @@ void protobuf_io(broker* thisptr, connection_handle hdl, const actor_ptr& buddy)
p.mutable_pong()->set_id(i);
write(p);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t rsn) {
if (self->last_sender() == buddy) self->quit(rsn);
on_arg_match >> [=](const down_msg& dm) {
if (dm.source == buddy) self->quit(dm.reason);
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
......
No preview for this file type
......@@ -43,6 +43,7 @@
#include "cppa/actor_addr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/message_header.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......@@ -71,7 +72,8 @@ bool abstract_actor::link_to_impl(const actor_addr& other) {
auto ptr = detail::actor_addr_cast<abstract_actor>(other);
// send exit message if already exited
if (exited()) {
ptr->enqueue({address(), ptr}, make_any_tuple(atom("EXIT"), exit_reason()));
ptr->enqueue({address(), ptr},
make_any_tuple(exit_msg{address(), exit_reason()}));
}
// add link if not already linked to other
// (checked by establish_backlink)
......@@ -152,7 +154,9 @@ bool abstract_actor::establish_backlink(const actor_addr& other) {
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
//send_as(this, other, make_any_tuple(atom("EXIT"), reason));
auto ptr = detail::raw_access::unsafe_cast(other);
ptr.enqueue({address(), ptr},
make_any_tuple(exit_msg{address(), exit_reason()}));
}
return false;
}
......@@ -202,7 +206,7 @@ void abstract_actor::cleanup(std::uint32_t reason) {
<< " attached functors; exit reason = " << reason
<< ", class = " << detail::demangle(typeid(*this)));
// send exit messages
auto msg = make_any_tuple(atom("EXIT"), reason);
auto msg = make_any_tuple(exit_msg{address(), reason});
CPPA_LOGM_DEBUG("cppa::actor", "send EXIT to " << mlinks.size() << " links");
for (auto& aptr : mlinks) {
aptr->enqueue({address(), aptr, message_id{}.with_high_priority()}, msg);
......
......@@ -163,7 +163,7 @@ class local_broker : public untyped_actor {
// forward to all acquaintances
send_to_acquaintances(what);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t) {
on_arg_match >> [=](const down_msg&) {
auto sender = last_sender();
CPPA_LOGC_TRACE("cppa::local_broker", "init$DOWN",
CPPA_TARG(sender, to_string));
......@@ -519,13 +519,13 @@ class remote_group_module : public group::module {
sm->put(key, nullptr);
}
},
on(atom("TIMEOUT")) >> [sm, &key] {
on<sync_timeout_msg>() >> [sm, &key] {
sm->put(key, nullptr);
}
);
}
},
on<atom("DOWN"), std::uint32_t>() >> [&] {
on_arg_match >> [&](const down_msg&) {
auto who = self->last_sender();
auto find_peer = [&] {
return find_if(begin(*peers), end(*peers), [&](const peer_map::value_type& kvp) {
......
......@@ -58,7 +58,7 @@ class down_observer : public attachable {
if (m_observer) {
auto ptr = detail::actor_addr_cast<abstract_actor>(m_observer);
message_header hdr{m_observed, ptr, message_id{}.with_high_priority()};
hdr.deliver(make_any_tuple(atom("DOWN"), reason));
hdr.deliver(make_any_tuple(down_msg{m_observed, reason}));
}
}
......@@ -154,7 +154,7 @@ void local_actor::send_tuple(const channel& dest, any_tuple what) {
}
void local_actor::send_exit(const actor_addr& whom, std::uint32_t reason) {
send(detail::raw_access::get(whom), atom("EXIT"), reason);
send(detail::raw_access::get(whom), exit_msg{address(), reason});
}
void local_actor::delayed_send_tuple(const channel& dest,
......@@ -203,7 +203,7 @@ message_id local_actor::timed_sync_send_tuple_impl(message_priority mp,
dest.enqueue({address(), dest, nri}, std::move(what));
auto rri = nri.response_id();
get_scheduler()->delayed_send({address(), this, rri}, rtime,
make_any_tuple(atom("TIMEOUT")));
make_any_tuple(sync_timeout_msg{}));
return rri;
}
......
......@@ -41,6 +41,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scoped_actor.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/detail/proper_actor.hpp"
#include "cppa/detail/actor_registry.hpp"
......@@ -241,7 +242,7 @@ void scheduler_helper::printer_loop(blocking_untyped_actor* self) {
on(atom("flush")) >> [&] {
flush_output(self->last_sender());
},
on(atom("DOWN"), any_vals) >> [&] {
on_arg_match >> [&](const down_msg&) {
auto s = self->last_sender();
flush_output(s);
out.erase(s);
......
......@@ -59,41 +59,46 @@ namespace cppa { namespace detail {
// maps demangled names to libcppa names
// WARNING: this map is sorted, insert new elements *in sorted order* as well!
/* extern */ const char* mapped_type_names[][2] = {
{ "bool", "bool" },
{ "cppa::actor", "@actor" },
{ "cppa::actor_addr", "@addr" },
{ "cppa::any_tuple", "@tuple" },
{ "cppa::atom_value", "@atom" },
{ "cppa::channel", "@channel" },
{ "cppa::intrusive_ptr<cppa::group>", "@group" },
{ "cppa::intrusive_ptr<cppa::node_id>", "@proc" },
{ "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" },
{ "cppa::message_header", "@header" },
{ "cppa::unit_t", "@0" },
{ "cppa::util::buffer", "@buffer" },
{ "cppa::util::duration", "@duration" },
{ "double", "double" },
{ "float", "float" },
{ "long double", "@ldouble" },
{ "bool", "bool" },
{ "cppa::actor", "@actor" },
{ "cppa::actor_addr", "@addr" },
{ "cppa::any_tuple", "@tuple" },
{ "cppa::atom_value", "@atom" },
{ "cppa::channel", "@channel" },
{ "cppa::down_msg", "@down" },
{ "cppa::exit_msg", "@exit" },
{ "cppa::intrusive_ptr<cppa::group>", "@group" },
{ "cppa::intrusive_ptr<cppa::node_id>", "@proc" },
{ "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" },
{ "cppa::message_header", "@header" },
{ "cppa::sync_timeout_msg", "@sync_timeout" },
{ "cppa::timeout_msg", "@timeout" },
{ "cppa::unit_t", "@0" },
{ "cppa::util::buffer", "@buffer" },
{ "cppa::util::duration", "@duration" },
{ "double", "double" },
{ "float", "float" },
{ "long double", "@ldouble" },
// std::u16string
{ "std::basic_string<@i16,std::char_traits<@i16>,std::allocator<@i16>>",
"@u16str" },
"@u16str" },
// std::u32string
{ "std::basic_string<@i32,std::char_traits<@i32>,std::allocator<@i32>>",
"@u32str" },
"@u32str" },
// std::string
{ "std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>",
"@str" },
"@str" },
// std::u16string (again, using unsigned char type)
{ "std::basic_string<@u16,std::char_traits<@u16>,std::allocator<@u16>>",
"@u16str" },
"@u16str" },
// std::u32string (again, using unsigned char type)
{ "std::basic_string<@u32,std::char_traits<@u32>,std::allocator<@u32>>",
"@u32str" },
"@u32str" },
// std::map<std::string,std::string>
{ "std::map<@str,@str,std::less<@str>,std::allocator<std::pair<const @str,@str>>>",
"@strmap" }
{ "std::map<@str,@str,std::less<@str>,"
"std::allocator<std::pair<const @str,@str>>>",
"@strmap" }
};
// maps sizeof(T) => {unsigned name, signed name}
......@@ -372,6 +377,38 @@ inline void deserialize_impl(bool& val, deserializer* source) {
val = source->read<uint8_t>() != 0;
}
// exit_msg & down_msg have the same fields
template<typename T>
typename std::enable_if<
std::is_same<T, down_msg>::value || std::is_same<T, exit_msg>::value
>::type
serialize_impl(const T& dm, serializer* sink) {
serialize_impl(dm.source, sink);
sink->write_value(dm.reason);
}
// exit_msg & down_msg have the same fields
template<typename T>
typename std::enable_if<
std::is_same<T, down_msg>::value || std::is_same<T, exit_msg>::value
>::type
deserialize_impl(T& dm, deserializer* source) {
deserialize_impl(dm.source, source);
dm.reason = source->read<std::uint32_t>();
}
inline void serialize_impl(const timeout_msg& tm, serializer* sink) {
sink->write_value(tm.timeout_id);
}
inline void deserialize_impl(timeout_msg& tm, deserializer* source) {
tm.timeout_id = source->read<std::uint32_t>();
}
inline void serialize_impl(const sync_timeout_msg&, serializer*) { }
inline void deserialize_impl(const sync_timeout_msg&, deserializer*) { }
bool types_equal(const std::type_info* lhs, const std::type_info* rhs) {
// in some cases (when dealing with dynamic libraries),
// address can be different although types are equal
......@@ -711,37 +748,41 @@ class utim_impl : public uniform_type_info_map {
size_t, ptrdiff_t,
intptr_t >(mapping);
// fill builtin types *in sorted order* (by uniform name)
size_t i = 0;
m_builtin_types[i++] = &m_type_unit; // @0
m_builtin_types[i++] = &m_ac_hdl; // @ac_hdl
m_builtin_types[i++] = &m_type_actor; // @actor
m_builtin_types[i++] = &m_type_actor_addr; // @actor_addr
m_builtin_types[i++] = &m_type_atom; // @atom
m_builtin_types[i++] = &m_type_buffer; // @buffer
m_builtin_types[i++] = &m_type_channel; // @channel
m_builtin_types[i++] = &m_cn_hdl; // @cn_hdl
m_builtin_types[i++] = &m_type_duration; // @duration
m_builtin_types[i++] = &m_type_group; // @group
m_builtin_types[i++] = &m_type_header; // @header
m_builtin_types[i++] = &m_type_i16; // @i16
m_builtin_types[i++] = &m_type_i32; // @i32
m_builtin_types[i++] = &m_type_i64; // @i64
m_builtin_types[i++] = &m_type_i8; // @i8
m_builtin_types[i++] = &m_type_long_double; // @ldouble
m_builtin_types[i++] = &m_type_proc; // @proc
m_builtin_types[i++] = &m_type_str; // @str
m_builtin_types[i++] = &m_type_strmap; // @strmap
m_builtin_types[i++] = &m_type_tuple; // @tuple
m_builtin_types[i++] = &m_type_u16; // @u16
m_builtin_types[i++] = &m_type_u16str; // @u16str
m_builtin_types[i++] = &m_type_u32; // @u32
m_builtin_types[i++] = &m_type_u32str; // @u32str
m_builtin_types[i++] = &m_type_u64; // @u64
m_builtin_types[i++] = &m_type_u8; // @u8
m_builtin_types[i++] = &m_type_bool; // bool
m_builtin_types[i++] = &m_type_double; // double
m_builtin_types[i++] = &m_type_float; // float
CPPA_REQUIRE(i == m_builtin_types.size());
auto i = m_builtin_types.begin();
*i++ = &m_type_unit; // @0
*i++ = &m_ac_hdl; // @ac_hdl
*i++ = &m_type_actor; // @actor
*i++ = &m_type_actor_addr; // @actor_addr
*i++ = &m_type_atom; // @atom
*i++ = &m_type_buffer; // @buffer
*i++ = &m_type_channel; // @channel
*i++ = &m_cn_hdl; // @cn_hdl
*i++ = &m_type_down_msg; // @down
*i++ = &m_type_duration; // @duration
*i++ = &m_type_exit_msg; // @exit
*i++ = &m_type_group; // @group
*i++ = &m_type_header; // @header
*i++ = &m_type_i16; // @i16
*i++ = &m_type_i32; // @i32
*i++ = &m_type_i64; // @i64
*i++ = &m_type_i8; // @i8
*i++ = &m_type_long_double; // @ldouble
*i++ = &m_type_proc; // @proc
*i++ = &m_type_str; // @str
*i++ = &m_type_strmap; // @strmap
*i++ = &m_type_sync_timeout; // @sync_timeout
*i++ = &m_type_timeout; // @timeout
*i++ = &m_type_tuple; // @tuple
*i++ = &m_type_u16; // @u16
*i++ = &m_type_u16str; // @u16str
*i++ = &m_type_u32; // @u32
*i++ = &m_type_u32str; // @u32str
*i++ = &m_type_u64; // @u64
*i++ = &m_type_u8; // @u8
*i++ = &m_type_bool; // bool
*i++ = &m_type_double; // double
*i++ = &m_type_float; // float
CPPA_REQUIRE(i == m_builtin_types.end());
# if CPPA_DEBUG_MODE
auto cmp = [](pointer lhs, pointer rhs) {
return strcmp(lhs->name(), rhs->name()) < 0;
......@@ -834,22 +875,31 @@ class utim_impl : public uniform_type_info_map {
typedef std::map<std::string, std::string> strmap;
// 0-9
uti_impl<node_id_ptr> m_type_proc;
uti_impl<io::accept_handle> m_ac_hdl;
uti_impl<io::connection_handle> m_cn_hdl;
uti_impl<channel> m_type_channel;
uti_impl<channel> m_type_channel;
uti_impl<down_msg> m_type_down_msg;
uti_impl<exit_msg> m_type_exit_msg;
buffer_type_info_impl m_type_buffer;
uti_impl<actor> m_type_actor;
uti_impl<actor_addr> m_type_actor_addr;
uti_impl<group_ptr> m_type_group;
// 10-19
uti_impl<any_tuple> m_type_tuple;
uti_impl<util::duration> m_type_duration;
uti_impl<sync_timeout_msg> m_type_sync_timeout;
uti_impl<timeout_msg> m_type_timeout;
uti_impl<message_header> m_type_header;
uti_impl<unit_t> m_type_unit;
uti_impl<atom_value> m_type_atom;
uti_impl<std::string> m_type_str;
uti_impl<std::u16string> m_type_u16str;
uti_impl<std::u32string> m_type_u32str;
// 20-29
default_uniform_type_info_impl<strmap> m_type_strmap;
uti_impl<bool> m_type_bool;
uti_impl<float> m_type_float;
......@@ -860,12 +910,14 @@ class utim_impl : public uniform_type_info_map {
int_tinfo<std::int16_t> m_type_i16;
int_tinfo<std::uint16_t> m_type_u16;
int_tinfo<std::int32_t> m_type_i32;
// 30-32
int_tinfo<std::uint32_t> m_type_u32;
int_tinfo<std::int64_t> m_type_i64;
int_tinfo<std::uint64_t> m_type_u64;
// both containers are sorted by uniform name
std::array<pointer, 29> m_builtin_types;
std::array<pointer, 33> m_builtin_types;
std::vector<uniform_type_info*> m_user_types;
mutable util::shared_spinlock m_lock;
......
......@@ -71,8 +71,8 @@ void pong(cppa::untyped_actor* self) {
on(atom("ping"), arg_match) >> [](int value) {
return make_cow_tuple(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
self->quit(reason);
on_arg_match >> [=](const down_msg& dm) {
self->quit(dm.reason);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
......@@ -113,8 +113,8 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
on(atom("pong"), arg_match) >> [=](int value) {
write(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (self->last_sender() == buddy) self->quit(reason);
on_arg_match >> [=](const down_msg& dm) {
if (dm.source == buddy) self->quit(dm.reason);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
......
......@@ -69,8 +69,8 @@ void spawn5_server_impl(untyped_actor* self, actor client, group_ptr grp) {
CPPA_PRINT("wait for DOWN messages");
auto downs = std::make_shared<int>(0);
self->become (
on(atom("DOWN"), arg_match) >> [=](std::uint32_t reason) {
if (reason != exit_reason::normal) {
on_arg_match >> [=](const down_msg& dm) {
if (dm.reason != exit_reason::normal) {
CPPA_PRINTERR("reflector exited for non-normal exit reason!");
}
if (++*downs == 5) {
......@@ -141,8 +141,8 @@ void spawn5_client(untyped_actor* self) {
template<typename F>
void await_down(untyped_actor* self, actor ptr, F continuation) {
self->become (
on(atom("DOWN"), arg_match) >> [=](uint32_t) -> bool {
if (self->last_sender() == ptr) {
on_arg_match >> [=](const down_msg& dm) -> bool {
if (dm.source == ptr) {
continuation();
return true;
}
......@@ -351,9 +351,9 @@ int main(int argc, char** argv) {
}
auto c = self->spawn<client, monitored>(serv);
self->receive (
on(atom("DOWN"), arg_match) >> [&](uint32_t rsn) {
CPPA_CHECK_EQUAL(self->last_sender(), c);
CPPA_CHECK_EQUAL(rsn, exit_reason::normal);
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.source, c);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
});
......@@ -397,9 +397,9 @@ int main(int argc, char** argv) {
else { CPPA_PRINT("actor published at port " << port); }
CPPA_CHECKPOINT();
self->receive (
on(atom("DOWN"), arg_match) >> [&](uint32_t rsn) {
CPPA_CHECK_EQUAL(self->last_sender(), serv);
CPPA_CHECK_EQUAL(rsn, exit_reason::normal);
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.source, serv);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
// wait until separate process (in sep. thread) finished execution
......
......@@ -529,7 +529,12 @@ void test_spawn() {
);
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) {
CPPA_CHECKPOINT();
}
else { CPPA_UNEXPECTED_MSG(); }
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
self->await_all_other_actors_done();
......@@ -545,7 +550,12 @@ void test_spawn() {
);
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) {
CPPA_CHECKPOINT();
}
else { CPPA_UNEXPECTED_MSG(); }
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
self->await_all_other_actors_done();
......@@ -562,7 +572,12 @@ void test_spawn() {
);
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) {
CPPA_CHECKPOINT();
}
else { CPPA_UNEXPECTED_MSG(); }
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
self->await_all_other_actors_done();
......@@ -714,8 +729,9 @@ void test_spawn() {
after(std::chrono::seconds(5)) >> CPPA_UNEXPECTED_TOUT_CB()
);
self->receive (
on(atom("DOWN"), exit_reason::normal) >> [&] {
CPPA_CHECK_EQUAL(self->last_sender(), sync_testee);
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
CPPA_CHECK_EQUAL(dm.source, sync_testee);
}
);
self->await_all_other_actors_done();
......@@ -830,20 +846,19 @@ void test_spawn() {
self->delayed_send(self, chrono::seconds(1), atom("FooBar"));
// wait for DOWN and EXIT messages of pong
self->receive_for(i, 4) (
on(atom("EXIT"), arg_match) >> [&](uint32_t reason) {
CPPA_CHECK_EQUAL(exit_reason::user_shutdown, reason);
CPPA_CHECK(self->last_sender() == pong_actor);
on_arg_match >> [&](const exit_msg& em) {
CPPA_CHECK_EQUAL(em.source, pong_actor);
CPPA_CHECK_EQUAL(em.reason, exit_reason::user_shutdown);
flags |= 0x01;
},
on(atom("DOWN"), arg_match) >> [&](uint32_t reason) {
auto who = self->last_sender();
if (who == pong_actor) {
on_arg_match >> [&](const down_msg& dm) {
if (dm.source == pong_actor) {
flags |= 0x02;
CPPA_CHECK_EQUAL(reason, exit_reason::user_shutdown);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
}
else if (who == ping_actor) {
else if (dm.source == ping_actor) {
flags |= 0x04;
CPPA_CHECK_EQUAL(reason, exit_reason::normal);
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
},
on_arg_match >> [&](const atom_value& val) {
......
......@@ -201,7 +201,9 @@ void test_sync_send() {
self->quit(exit_reason::user_shutdown);
});
self->receive (
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
auto mirror = spawn<sync_mirror>();
......@@ -219,9 +221,12 @@ void test_sync_send() {
self->receive (
on(atom("success")) >> CPPA_CHECKPOINT_CB(),
on(atom("failure")) >> CPPA_FAILURE_CB("A didn't receive sync response"),
on(atom("DOWN"), arg_match).when(_x2 != exit_reason::normal)
>> [&](uint32_t err) {
CPPA_FAILURE("A exited for reason " << err);
on_arg_match >> [&](const down_msg& dm) -> match_hint {
if (dm.reason != exit_reason::normal) {
CPPA_FAILURE("A exited for reason " << dm.reason);
return match_hint::handle;
}
return match_hint::skip;
}
);
};
......@@ -235,14 +240,16 @@ void test_sync_send() {
self->await_all_other_actors_done();
CPPA_CHECKPOINT();
self->timed_sync_send(self, std::chrono::milliseconds(50), atom("NoWay")).await(
on(atom("TIMEOUT")) >> CPPA_CHECKPOINT_CB(),
on<sync_timeout_msg>() >> CPPA_CHECKPOINT_CB(),
others() >> CPPA_UNEXPECTED_MSG_CB()
);
// we should have received two DOWN messages with normal exit reason
// plus 'NoWay'
int i = 0;
self->receive_for(i, 3) (
on(atom("DOWN"), exit_reason::normal) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.reason, exit_reason::normal);
},
on(atom("NoWay")) >> [] {
CPPA_CHECKPOINT();
CPPA_PRINT("trigger \"actor did not reply to a "
......@@ -305,7 +312,9 @@ void test_sync_send() {
self->receive_loop(others() >> CPPA_UNEXPECTED_MSG_CB());
});
self->receive (
on(atom("DOWN"), exit_reason::user_shutdown) >> CPPA_CHECKPOINT_CB(),
on_arg_match >> [&](const down_msg& dm) {
CPPA_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
},
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
......
......@@ -93,7 +93,11 @@ int main() {
"@proc", // intrusive_ptr<node_id>
"@duration", // util::duration
"@buffer", // util::buffer
// default announced cpap tuples
"@down", // down_msg
"@exit", // exit_msg
"@timeout", // timeout_msg
"@sync_timeout", // sync_timeout_msg
// default announced cppa tuples
"@<>+@atom", // {atom_value}
"@<>+@atom+@actor", // {atom_value, actor_ptr}
"@<>+@atom+@proc", // {atom_value, node_id}
......
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