Commit 73d03881 authored by Dominik Charousset's avatar Dominik Charousset

Remove timed_sync_send and fix sync. timeouts

Closes #283.
parent 613499c3
...@@ -45,11 +45,6 @@ static constexpr uint32_t unhandled_exception = 0x00002; ...@@ -45,11 +45,6 @@ static constexpr uint32_t unhandled_exception = 0x00002;
*/ */
static constexpr uint32_t unhandled_sync_failure = 0x00004; static constexpr uint32_t unhandled_sync_failure = 0x00004;
/**
* Indicates that a synchronous message timed out.
*/
static constexpr uint32_t unhandled_sync_timeout = 0x00005;
/** /**
* Indicates that the exit reason for this actor is unknown, i.e., * Indicates that the exit reason for this actor is unknown, i.e.,
* the actor has been terminated and no longer exists. * the actor has been terminated and no longer exists.
......
...@@ -384,13 +384,6 @@ class local_actor : public abstract_actor, public resumable { ...@@ -384,13 +384,6 @@ class local_actor : public abstract_actor, public resumable {
/** /**
* Sets the handler for unexpected synchronous response messages. * Sets the handler for unexpected synchronous response messages.
*/ */
inline void on_sync_timeout(std::function<void()> fun) {
m_sync_timeout_handler = std::move(fun);
}
/**
* Sets the handler for `timed_sync_send` timeout messages.
*/
inline void on_sync_failure(std::function<void()> fun) { inline void on_sync_failure(std::function<void()> fun) {
m_sync_failure_handler = std::move(fun); m_sync_failure_handler = std::move(fun);
} }
...@@ -402,14 +395,6 @@ class local_actor : public abstract_actor, public resumable { ...@@ -402,14 +395,6 @@ class local_actor : public abstract_actor, public resumable {
return static_cast<bool>(m_sync_failure_handler); return static_cast<bool>(m_sync_failure_handler);
} }
/**
* Calls `on_sync_timeout(fun); on_sync_failure(fun);.
*/
inline void on_sync_timeout_or_failure(std::function<void()> fun) {
on_sync_timeout(fun);
on_sync_failure(fun);
}
/** /**
* Sets a custom exception handler for this actor. If multiple handlers are * Sets a custom exception handler for this actor. If multiple handlers are
* defined, only the functor that was added *last* is being executed. * defined, only the functor that was added *last* is being executed.
...@@ -485,14 +470,6 @@ class local_actor : public abstract_actor, public resumable { ...@@ -485,14 +470,6 @@ class local_actor : public abstract_actor, public resumable {
return m_current_element; return m_current_element;
} }
inline void handle_sync_timeout() {
if (m_sync_timeout_handler) {
m_sync_timeout_handler();
} else {
quit(exit_reason::unhandled_sync_timeout);
}
}
inline void handle_sync_failure() { inline void handle_sync_failure() {
if (m_sync_failure_handler) { if (m_sync_failure_handler) {
m_sync_failure_handler(); m_sync_failure_handler();
...@@ -514,15 +491,6 @@ class local_actor : public abstract_actor, public resumable { ...@@ -514,15 +491,6 @@ class local_actor : public abstract_actor, public resumable {
void request_sync_timeout_msg(const duration& dr, message_id mid); void request_sync_timeout_msg(const duration& dr, message_id mid);
// returns the response ID
template <class Handle, class... Ts>
message_id timed_sync_send_impl(message_priority mp, const Handle& dh,
const duration& dr, Ts&&... xs) {
auto result = sync_send_impl(mp, dh, std::forward<Ts>(xs)...);
request_sync_timeout_msg(dr, result);
return result;
}
// returns 0 if last_dequeued() is an asynchronous or sync request message, // returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise // a response id generated from the request id otherwise
inline message_id get_response_id() { inline message_id get_response_id() {
...@@ -673,7 +641,6 @@ class local_actor : public abstract_actor, public resumable { ...@@ -673,7 +641,6 @@ class local_actor : public abstract_actor, public resumable {
const duration& rtime, message data); const duration& rtime, message data);
std::function<void()> m_sync_failure_handler; std::function<void()> m_sync_failure_handler;
std::function<void()> m_sync_timeout_handler;
}; };
/** /**
......
...@@ -29,13 +29,14 @@ constexpr const char* s_names_table[] = { ...@@ -29,13 +29,14 @@ constexpr const char* s_names_table[] = {
"unhandled_exception", "unhandled_exception",
"-invalid-", "-invalid-",
"unhandled_sync_failure", "unhandled_sync_failure",
"unhandled_sync_timeout", "-invalid-",
"unknown" "unknown",
"out_of_workers"
}; };
} // namespace <anonymous> } // namespace <anonymous>
const char* as_string(uint32_t value) { const char* as_string(uint32_t value) {
if (value <= unhandled_sync_timeout) { if (value <= out_of_workers) {
return s_names_table[value]; return s_names_table[value];
} }
switch (value) { switch (value) {
...@@ -46,7 +47,7 @@ const char* as_string(uint32_t value) { ...@@ -46,7 +47,7 @@ const char* as_string(uint32_t value) {
if (value < user_defined) { if (value < user_defined) {
return "-invalid-"; return "-invalid-";
} }
return "-user defined-"; return "user_defined";
} }
} }
......
...@@ -351,21 +351,21 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -351,21 +351,21 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>(); bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>();
ptr.swap(current_mailbox_element()); ptr.swap(current_mailbox_element());
auto mid = current_mailbox_element()->mid; auto mid = current_mailbox_element()->mid;
auto res = post_process_invoke_res(this, mid, if (is_sync_tout) {
fun(current_mailbox_element()->msg)); if (fun.timeout().valid()) {
ptr.swap(current_mailbox_element()); fun.handle_timeout();
mark_arrived(awaited_id); }
if (!res) { } else {
if (is_sync_tout) { auto res = post_process_invoke_res(this, mid,
CAF_LOG_WARNING("sync timeout occured in actor " fun(current_mailbox_element()->msg));
<< "with ID " << id()); ptr.swap(current_mailbox_element());
handle_sync_timeout(); if (!res) {
} else {
CAF_LOG_WARNING("sync failure occured in actor " CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << id()); << "with ID " << id());
handle_sync_failure(); handle_sync_failure();
} }
} }
mark_arrived(awaited_id);
return im_success; return im_success;
} }
return im_skipped; return im_skipped;
...@@ -441,6 +441,9 @@ local_actor::find_pending_response(message_id mid) { ...@@ -441,6 +441,9 @@ local_actor::find_pending_response(message_id mid) {
void local_actor::set_response_handler(message_id response_id, behavior bhvr) { void local_actor::set_response_handler(message_id response_id, behavior bhvr) {
auto pr = find_pending_response(response_id); auto pr = find_pending_response(response_id);
if (pr) { if (pr) {
if (bhvr.timeout().valid()) {
request_sync_timeout_msg(bhvr.timeout(), response_id);
}
pr->second = std::move(bhvr); pr->second = std::move(bhvr);
} }
} }
......
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/none.hpp"
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -365,13 +364,13 @@ CAF_TEST(sync_send) { ...@@ -365,13 +364,13 @@ CAF_TEST(sync_send) {
CAF_MESSAGE("block on `await_all_other_actors_done`"); CAF_MESSAGE("block on `await_all_other_actors_done`");
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_MESSAGE("`await_all_other_actors_done` finished"); CAF_MESSAGE("`await_all_other_actors_done` finished");
self->timed_sync_send(self, milliseconds(50), no_way_atom::value).await( self->sync_send(self, no_way_atom::value).await(
on<sync_timeout_msg>() >> [] {
CAF_MESSAGE("Got timeout");
},
others >> [&] { others >> [&] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
},
after(milliseconds(50)) >> [] {
CAF_MESSAGE("Got timeout");
} }
); );
// we should have received two DOWN messages with normal exit reason // we should have received two DOWN messages with normal exit reason
...@@ -406,17 +405,17 @@ CAF_TEST(sync_send) { ...@@ -406,17 +405,17 @@ CAF_TEST(sync_send) {
auto c = spawn<C>(); // replies only to 'gogo' messages auto c = spawn<C>(); // replies only to 'gogo' messages
// first test: sync error must occur, continuation must not be called // first test: sync error must occur, continuation must not be called
bool timeout_occured = false; bool timeout_occured = false;
self->on_sync_timeout([&] {
CAF_MESSAGE("timeout occured");
timeout_occured = true;
});
self->on_sync_failure([&] { self->on_sync_failure([&] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
}); });
self->timed_sync_send(c, milliseconds(500), hi_there_atom::value).await( self->sync_send(c, milliseconds(500), hi_there_atom::value).await(
on(val<atom_value>) >> [&] { on(val<atom_value>) >> [&] {
cout << "C did reply to 'HiThere'" << endl; CAF_TEST_ERROR("C did reply to 'HiThere'");
},
after(milliseconds(500)) >> [&] {
CAF_MESSAGE("timeout occured");
timeout_occured = true;
} }
); );
CAF_CHECK_EQUAL(timeout_occured, true); CAF_CHECK_EQUAL(timeout_occured, true);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE sync_timeout
#include "caf/test/unit_test.hpp"
#include <thread>
#include <chrono>
#include "caf/all.hpp"
using namespace caf;
namespace {
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
using send_ping_atom = atom_constant<atom("send_ping")>;
behavior pong() {
return {
[=] (ping_atom) {
std::this_thread::sleep_for(std::chrono::seconds(1));
return pong_atom::value;
}
};
}
behavior ping1(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor);
self->send(self, send_ping_atom::value);
return {
[=](send_ping_atom) {
self->sync_send(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("sync timeout: check");
self->quit(exit_reason::user_shutdown);
}
);
}
};
}
behavior ping2(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor);
self->send(self, send_ping_atom::value);
auto received_inner = std::make_shared<bool>(false);
return {
[=](send_ping_atom) {
self->sync_send(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("inner timeout: check");
*received_inner = true;
}
);
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_CHECK_EQUAL(*received_inner, true);
self->quit(exit_reason::user_shutdown);
}
};
}
struct fixture {
~fixture() {
await_all_actors_done();
shutdown();
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(atom_tests, fixture)
CAF_TEST(single_timeout) {
spawn(ping1, spawn(pong));
}
CAF_TEST(scoped_timeout) {
spawn(ping2, spawn(pong));
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -53,8 +53,6 @@ class local_actor; ...@@ -53,8 +53,6 @@ class local_actor;
\hline \hline
\lstinline^void on_sync_failure(auto fun)^ & Sets a handler, i.e., a functor taking no arguments, for unexpected synchronous response messages (default action is to kill the actor for reason \lstinline^unhandled_sync_failure^) \\ \lstinline^void on_sync_failure(auto fun)^ & Sets a handler, i.e., a functor taking no arguments, for unexpected synchronous response messages (default action is to kill the actor for reason \lstinline^unhandled_sync_failure^) \\
\hline \hline
\lstinline^void on_sync_timeout(auto fun)^ & Sets a handler, i.e., a functor taking no arguments, for \lstinline^timed_sync_send^ timeout messages (default action is to kill the actor for reason \lstinline^unhandled_sync_timeout^) \\
\hline
\lstinline^void monitor(actor whom)^ & Unidirectionally monitors \lstinline^whom^ (see Section \ref{Sec::Management::Monitors}) \\ \lstinline^void monitor(actor whom)^ & Unidirectionally monitors \lstinline^whom^ (see Section \ref{Sec::Management::Monitors}) \\
\hline \hline
\lstinline^void demonitor(actor whom)^ & Removes a monitor from \lstinline^whom^ \\ \lstinline^void demonitor(actor whom)^ & Removes a monitor from \lstinline^whom^ \\
......
...@@ -205,4 +205,6 @@ Besides, these function names were not a good choice in the first place, since ` ...@@ -205,4 +205,6 @@ Besides, these function names were not a good choice in the first place, since `
As a result, both functions are now deprecated. As a result, both functions are now deprecated.
Their replacements are named \lstinline^current_message^ and \lstinline^current_sender^ (cf. Section \ref{Sec::Actors::Interfaces}). Their replacements are named \lstinline^current_message^ and \lstinline^current_sender^ (cf. Section \ref{Sec::Actors::Interfaces}).
\subsubsection{0.13 $\Rightarrow$ 0.14}
The function \lstinline^timed_sync_send^ has been removed. It offered an alternative way of defining message handlers, which is inconsistent with the rest of the API.
\section{Synchronous Communication} \section{Synchronous Communication}
\label{Sec::Sync} \label{Sec::Sync}
\lib supports both asynchronous and synchronous communication. \lib supports both asynchronous and synchronous communication. The latter is provided by the member function \lstinline^sync_send^.
The member function \lstinline^sync_send^ sends synchronous request messages.
\begin{lstlisting} \begin{lstlisting}
template<typename... Args> template<typename... Args>
__unspecified__ sync_send(actor whom, Args&&... what); __unspecified__ sync_send(actor whom, Args&&... what);
template<typename Duration, typename... Args>
__unspecified__ timed_sync_send(actor whom,
Duration timeout,
Args&&... what);
\end{lstlisting} \end{lstlisting}
A synchronous message is sent to the receiving actor's mailbox like any other (asynchronous) message. A synchronous message is sent to the receiving actor's mailbox like any other (asynchronous) message. Only the response message is treated separately.
The response message, on the other hand, is treated separately.
The difference between \lstinline^sync_send^ and \lstinline^timed_sync_send^ is how timeouts are handled.
The behavior of \lstinline^sync_send^ is analogous to \lstinline^send^, i.e., timeouts are specified by using \lstinline^after(...)^ statements (see Section \ref{Sec::Receive::Timeouts}).
When using \lstinline^timed_sync_send^ function, \lstinline^after(...)^ statements are ignored and the actor will receive a \lstinline^sync_timeout_msg^ after the given duration instead.
\subsection{Error Messages} \subsection{Additional Error Messages}
When using synchronous messaging, \lib's runtime environment will send ... \begin{lstlisting}
struct sync_exited_msg {
actor_addr source;
uint32_t reason;
};
\end{lstlisting}
\begin{itemize} When using synchronous messaging, \lib's runtime will send a \lstinline^sync_exited_msg^ message if the receiver is not alive. This is in addition to exit and down messages caused by linking or monitoring.
\item if the receiver is not alive:\newline\lstinline^sync_exited_msg { actor_addr source; std::uint32_t reason; };^
%\item \lstinline^{'VOID'}^ if the receiver handled the message but did not respond to it
\item if a message send by \lstinline^timed_sync_send^ timed out: \lstinline^sync_timeout_msg^
\end{itemize}
\clearpage
\subsection{Receive Response Messages} \subsection{Receive Response Messages}
When sending a synchronous message, the response handler can be passed by either using \lstinline^then^ (event-based actors) or \lstinline^await^ (blocking actors). When sending a synchronous message, the response handler can be passed by either using \lstinline^then^ (event-based actors) or \lstinline^await^ (blocking actors).
...@@ -53,29 +42,24 @@ void foo(event_based_actor* self, actor testee) { ...@@ -53,29 +42,24 @@ void foo(event_based_actor* self, actor testee) {
Similar to \lstinline^become^, the \lstinline^then^ function modifies an actor's behavior stack. Similar to \lstinline^become^, the \lstinline^then^ function modifies an actor's behavior stack.
However, it is used as ``one-shot handler'' and automatically returns to the previous behavior afterwards. However, it is used as ``one-shot handler'' and automatically returns to the previous behavior afterwards.
\clearpage
\subsection{Synchronous Failures and Error Handlers} \subsection{Synchronous Failures and Error Handlers}
An unexpected response message, i.e., a message that is not handled by given behavior, will invoke the actor's \lstinline^on_sync_failure^ handler. An unexpected response message, i.e., a message that is not handled by the ``one-shot-handler'', is considered as an error. The runtime will invoke the actor's \lstinline^on_sync_failure^, which kills the actor by calling \lstinline^self->quit(exit_reason::unhandled_sync_failure)^ per default. The error handler can be overridden by calling \lstinline^self->on_sync_failure(...)^ as shown below.
The default handler kills the actor by calling \lstinline^self->quit(exit_reason::unhandled_sync_failure)^.
The handler can be overridden by calling \lstinline^self->on_sync_failure(/*...*/)^.
Unhandled timeout messages trigger the \lstinline^on_sync_timeout^ handler.
The default handler kills the actor for reason \lstinline^exit_reason::unhandled_sync_failure^.
It is possible set both error handlers by calling \lstinline^self->on_sync_timeout_or_failure(/*...*)^.
\begin{lstlisting} \begin{lstlisting}
void foo(event_based_actor* self, actor testee) { void foo(event_based_actor* self, actor testee) {
// testee replies with a string to 'get'
// set handler for unexpected messages // set handler for unexpected messages
self->on_sync_failure = [] { self->on_sync_failure([=] {
aout << "received: " << to_string(self->last_dequeued()) << endl; aout(self) << "received unexpected synchronous response: "
}; << to_string(self->last_dequeued()) << endl;
// set handler for timeouts });
self->on_sync_timeout = [] {
aout << "timeout occured" << endl;
};
// set response handler by using "then" // set response handler by using "then"
timed_sync_send(testee, std::chrono::seconds(30), get_atom::value).then( sync_send(testee, get_atom::value).then(
[=](const std::string& str) { /* handle str */ } [=](const std::string& str) {
/* handle str */
}
// any other result will call the on_sync_failure handler
); );
}
\end{lstlisting} \end{lstlisting}
No preview for this file type
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