Commit 0c774e3e authored by Dominik Charousset's avatar Dominik Charousset

added priorities to delayed / timed messages

parent 31bcfc16
......@@ -314,7 +314,7 @@ class local_actor : public extend<actor>::with<memory_cached> {
void reply_message(any_tuple&& what);
void forward_message(const actor_ptr& new_receiver);
void forward_message(const actor_ptr& new_receiver, message_priority prio);
inline const actor_ptr& chained_actor();
......
......@@ -108,35 +108,26 @@ class scheduler {
virtual attachable* register_hidden_context();
template<typename Duration, typename... Data>
void delayed_send(const channel_ptr& to,
void delayed_send(message_header hdr,
const Duration& rel_time,
any_tuple data ) {
auto tup = make_any_tuple(atom("SEND"),
util::duration{rel_time},
to,
std::move(hdr),
std::move(data));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
template<typename Duration, typename... Data>
void delayed_reply(const actor_ptr& to,
void delayed_reply(message_header hdr,
const Duration& rel_time,
message_id id,
any_tuple data ) {
CPPA_REQUIRE(!id.valid() || id.is_response());
if (id.valid()) {
auto tup = make_any_tuple(atom("REPLY"),
util::duration{rel_time},
to,
id,
std::move(data));
auto dsh = delayed_send_helper();
dsh->enqueue({self, dsh}, std::move(tup));
}
else {
this->delayed_send(to, rel_time, std::move(data));
}
CPPA_REQUIRE(hdr.id.valid() && hdr.id.is_response());
auto tup = make_any_tuple(atom("SEND"),
util::duration{rel_time},
std::move(hdr),
std::move(data));
delayed_send_helper()->enqueue(nullptr, std::move(tup));
}
/**
......
......@@ -51,27 +51,30 @@ namespace cppa {
/**
* @brief Stores sender, receiver, and message priority.
*/
template<typename T = channel_ptr>
struct destination_header {
channel_ptr receiver;
T receiver;
message_priority priority;
inline destination_header(const self_type& s)
: receiver(s), priority(message_priority::normal) { }
template<typename T>
inline destination_header(T dest)
: receiver(std::move(dest)), priority(message_priority::normal) { }
inline destination_header(channel_ptr dest, message_priority prio)
: receiver(std::move(dest)), priority(prio) { }
inline destination_header(destination_header&& hdr)
: receiver(std::move(hdr.receiver)), priority(hdr.priority) { }
template<typename U>
destination_header(U&& dest, message_priority mp = message_priority::normal)
: receiver(std::forward<U>(dest)), priority(mp) { }
destination_header(destination_header&&) = default;
destination_header(const destination_header&) = default;
destination_header& operator=(destination_header&&) = default;
destination_header& operator=(const destination_header&) = default;
};
using channel_destination = destination_header<channel_ptr>;
using actor_destination = destination_header<actor_ptr>;
/**
* @brief Sends @p what to the receiver specified in @p hdr.
*/
inline void send_tuple(destination_header hdr, any_tuple what) {
if (hdr.receiver == nullptr) return;
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(hdr.receiver), hdr.priority};
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
......@@ -85,7 +88,7 @@ inline void send_tuple(destination_header hdr, any_tuple what) {
* @brief Sends @p what to the receiver specified in @p hdr.
*/
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
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<
......@@ -97,7 +100,7 @@ void send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
>>::template eval
>::value;
static_assert(input_pos >= 0, "typed actor does not support given input");
send(whom.type_erased(), std::forward<Ts>(what)...);
send(dest./*receiver.*/type_erased(), std::forward<Ts>(what)...);
}
/**
......@@ -105,17 +108,17 @@ void send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
inline void send(destination_header hdr, Ts&&... what) {
inline void send(channel_destination dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
send_tuple(std::move(hdr), make_any_tuple(std::forward<Ts>(what)...));
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_ptr whom, any_tuple what) {
message_header hdr{std::move(from), std::move(whom)};
hdr.deliver(std::move(what));
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));
}
/**
......@@ -127,8 +130,8 @@ inline void send_tuple_as(actor_ptr from, channel_ptr whom, any_tuple what) {
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
inline void send_as(actor_ptr from, channel_ptr whom, Ts&&... what) {
send_tuple_as(std::move(from), std::move(whom),
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)...));
}
......@@ -141,7 +144,7 @@ inline void send_as(actor_ptr from, channel_ptr whom, Ts&&... what) {
* message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
message_future sync_send_tuple(actor_ptr whom, any_tuple what);
message_future sync_send_tuple(actor_destination dest, any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
......@@ -154,9 +157,9 @@ message_future sync_send_tuple(actor_ptr whom, any_tuple what);
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline message_future sync_send(actor_ptr whom, Ts&&... what) {
inline message_future sync_send(actor_destination dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(std::move(whom),
return sync_send_tuple(std::move(dest),
make_any_tuple(std::forward<Ts>(what)...));
}
......@@ -192,26 +195,12 @@ sync_send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
* @param whom Receiver of the message.
* @param rtime Relative time duration to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param what Message content as a tuple.
* @param data Message content as a tuple.
*/
void delayed_send_tuple(const channel_ptr& to,
const util::duration& rel_time,
void delayed_send_tuple(channel_destination dest,
const util::duration& rtime,
any_tuple data);
/**
* @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 content as a tuple.
*/
template<class Rep, class Period, typename... Ts>
inline void delayed_send_tuple(const channel_ptr& whom,
const std::chrono::duration<Rep, Period>& rtime,
any_tuple what) {
delayed_send_tuple(whom, util::duration{rtime}, std::move(what));
}
/**
* @brief Sends a reply message that is delayed by @p rel_time.
* @param rtime Relative time duration to delay the message in
......@@ -223,18 +212,6 @@ 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,
actor_ptr whom,
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
......@@ -244,39 +221,22 @@ void delayed_reply_tuple(const util::duration& rel_time,
*/
void delayed_reply_tuple(const util::duration& rel_time, 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()
*/
template<class Rep, class Period, typename... Ts>
inline void delayed_reply_tuple(const std::chrono::duration<Rep, Period>& rtime,
any_tuple what) {
delayed_reply_tuple(util::duration{rtime}, std::move(what));
}
/**
* @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 whom Receiver of the message.
* @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>
*/
template<class Rep, class Period, typename... Ts>
message_future timed_sync_send_tuple(actor_ptr whom,
const std::chrono::duration<Rep, Period>& rel_time,
any_tuple what) {
auto mf = sync_send_tuple(std::move(whom), std::move(what));
auto tmp = make_any_tuple(atom("TIMEOUT"));
delayed_reply_tuple(util::duration{rel_time}, self, mf.id(), std::move(tmp));
return mf;
}
message_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
......@@ -292,13 +252,13 @@ message_future timed_sync_send_tuple(actor_ptr whom,
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<class Rep, class Period, typename... Ts>
message_future timed_sync_send(actor_ptr whom,
const std::chrono::duration<Rep, Period>& rel_time,
template<typename... Ts>
message_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),
rel_time,
rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
......@@ -341,8 +301,8 @@ inline void reply_tuple_to(const response_handle& handle, any_tuple what) {
/**
* @brief Forwards the last received message to @p whom.
*/
inline void forward_to(const actor_ptr& whom) {
self->forward_message(whom);
inline void forward_to(actor_destination dest) {
self->forward_message(dest.receiver, dest.priority);
}
/**
......@@ -352,13 +312,13 @@ inline void forward_to(const actor_ptr& whom) {
* microseconds, milliseconds, seconds or minutes.
* @param what Message elements.
*/
template<class Rep, class Period, typename... Ts>
inline void delayed_send(const channel_ptr& whom,
const std::chrono::duration<Rep, Period>& rtime,
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 (whom) {
delayed_send_tuple(whom,
if (dest.receiver) {
delayed_send_tuple(std::move(dest),
rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
......@@ -371,9 +331,8 @@ inline void delayed_send(const channel_ptr& whom,
* @param what Message elements.
* @see delayed_send()
*/
template<class Rep, class Period, typename... Ts>
inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
Ts&&... what) {
template<typename... Ts>
inline void delayed_reply(const util::duration& rtime, Ts&&... what) {
delayed_reply_tuple(rtime, make_any_tuple(std::forward<Ts>(what)...));
}
......@@ -384,9 +343,9 @@ inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
* <tt>send(whom, atom("EXIT"), reason)</tt>.
* @pre <tt>reason != exit_reason::normal</tt>
*/
inline void send_exit(actor_ptr whom, std::uint32_t rsn) {
inline void send_exit(channel_destination dest, std::uint32_t rsn) {
CPPA_REQUIRE(rsn != exit_reason::normal);
send(std::move(whom), atom("EXIT"), rsn);
send(std::move(dest), atom("EXIT"), rsn);
}
/**
......
......@@ -82,6 +82,7 @@ class duration {
"only seconds, milliseconds or microseconds allowed");
}
// convert minutes to seconds
template<class Rep>
constexpr duration(std::chrono::duration<Rep, std::ratio<60, 1> > d)
: unit(time_unit::seconds), count(d.count() * 60) { }
......
......@@ -8,7 +8,7 @@ Typed actors use \lstinline^typed_actor_ptr<...>^ instead of \lstinline^actor_pt
For example, an actor responding to two integers with a dobule would use the type \lstinline^typed_actor_ptr<replies_to<int, int>::with<double>>^.
All functions for message passing, linking and monitoring are overloaded to accept both types of actors.
As of version 0.8, strongly typed actors cannot be published (this is a planned feature for future releases).
As of version 0.8, strongly typed actors cannot be published and do not support message priorities (those are planned feature for future releases).
\subsection{Spawning Typed Actors}
\label{sec:strong:spawn}
......
......@@ -158,10 +158,10 @@ void local_actor::reply_message(any_tuple&& what) {
}
}
void local_actor::forward_message(const actor_ptr& new_receiver) {
if (new_receiver == nullptr) return;
void local_actor::forward_message(const actor_ptr& dest, message_priority p) {
if (dest == nullptr) return;
auto& id = m_current_node->mid;
new_receiver->enqueue({last_sender(), new_receiver, id}, m_current_node->msg);
dest->enqueue({last_sender(), dest, id, p}, m_current_node->msg);
// treat this message as asynchronous message from now on
id = message_id{};
}
......
......@@ -66,17 +66,9 @@ class delayed_msg {
public:
delayed_msg(const channel_ptr& arg0,
const actor_ptr& arg1,
message_id,
any_tuple&& arg3)
: ptr_a(arg0), from(arg1), msg(move(arg3)) { }
delayed_msg(const actor_ptr& arg0,
const actor_ptr& arg1,
message_id arg2,
any_tuple&& arg3)
: ptr_b(arg0), from(arg1), id(arg2), msg(move(arg3)) { }
delayed_msg(message_header&& arg1,
any_tuple&& arg2)
: hdr(move(arg1)), msg(move(arg2)) { }
delayed_msg(delayed_msg&&) = default;
delayed_msg(const delayed_msg&) = default;
......@@ -84,18 +76,13 @@ class delayed_msg {
delayed_msg& operator=(const delayed_msg&) = default;
inline void eval() {
CPPA_REQUIRE(ptr_a || ptr_b);
if (ptr_a) ptr_a->enqueue(from, move(msg));
else ptr_b->enqueue({from, id}, move(msg));
hdr.deliver(std::move(msg));
}
private:
channel_ptr ptr_a;
actor_ptr ptr_b;
actor_ptr from;
message_id id;
any_tuple msg;
message_header hdr;
any_tuple msg;
};
......@@ -140,16 +127,14 @@ class scheduler_helper {
};
template<class Map, class T>
template<class Map>
inline void insert_dmsg(Map& storage,
const util::duration& d,
const T& to,
const actor_ptr& sender,
any_tuple&& tup,
message_id id = message_id{}) {
const util::duration& d,
message_header&& hdr,
any_tuple&& tup ) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{to, sender, id, move(tup)};
delayed_msg dmsg{move(hdr), move(tup)};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
......@@ -163,15 +148,9 @@ void scheduler_helper::timer_loop(scheduler_helper::ptr_type m_self) {
// message handling rules
auto mfun = (
on(atom("SEND"), arg_match) >> [&](const util::duration& d,
const channel_ptr& ptr,
message_header& hdr,
any_tuple& tup) {
insert_dmsg(messages, d, ptr, msg_ptr->sender, move(tup));
},
on(atom("REPLY"), arg_match) >> [&](const util::duration& d,
const actor_ptr& ptr,
message_id id,
any_tuple& tup) {
insert_dmsg(messages, d, ptr, msg_ptr->sender, move(tup), id);
insert_dmsg(messages, d, move(hdr), move(tup));
},
on(atom("DIE")) >> [&] {
done = true;
......
......@@ -34,10 +34,10 @@
namespace cppa {
message_future sync_send_tuple(actor_ptr whom, any_tuple what) {
if (!whom) throw std::invalid_argument("whom == nullptr");
message_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(whom), req};
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>());
......@@ -47,27 +47,30 @@ message_future sync_send_tuple(actor_ptr whom, any_tuple what) {
return req.response_id();
}
void delayed_send_tuple(const channel_ptr& to,
const util::duration& rel_time,
void delayed_send_tuple(channel_destination dest,
const util::duration& rtime,
any_tuple data) {
if (to) get_scheduler()->delayed_send(to, rel_time, std::move(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));
}
}
void delayed_reply_tuple(const util::duration& rel_time,
actor_ptr whom,
message_id mid,
any_tuple data) {
if (whom) get_scheduler()->delayed_reply(whom,
rel_time,
mid,
std::move(data));
message_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;
}
void delayed_reply_tuple(const util::duration& rel_time,
void delayed_reply_tuple(const util::duration& rtime,
message_id mid,
any_tuple data) {
delayed_reply_tuple(rel_time, self->last_sender(), mid, std::move(data));
message_header hdr{self, self->last_sender(), mid};
get_scheduler()->delayed_send(std::move(hdr), rtime, std::move(data));
}
void delayed_reply_tuple(const util::duration& rel_time, any_tuple data) {
......
......@@ -66,6 +66,7 @@ class typed_testee : public typed_actor<replies_to<my_request>::with<bool>> {
int main() {
CPPA_TEST(test_typed_spawn);
announce<my_request>(&my_request::a, &my_request::b);
auto sptr = spawn_typed_server();
sync_send(sptr, my_request{2, 2}).await(
......
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