Commit fd9a200d authored by Dominik Charousset's avatar Dominik Charousset

Deprecate `last_dequeued` and `last_sender`

After the most recent changes for optimizing forwarding and reducing memory
allocation (d72c0ec8), those two member
function now have a slightly different behavior than before. This change in
semantics is reflected by renaming `last_dequeued` to `current_message` and
`last_sender` to `current_sender`. The main change in semantics is that the new
functions have a longer a "nop" effect when called outside callbacks or after
forwarding a message. Instead, these function now cause undefined behavior,
i.e., dereference a null pointer. This is now clearly stated in the manual.
parent c6e72fba
......@@ -147,7 +147,7 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
self->send(buddy, atm, ival);
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
cout << "unexpected: " << to_string(self->current_message()) << endl;
}
};
}
......@@ -165,7 +165,7 @@ behavior server(broker* self, const actor& buddy) {
self->quit();
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
cout << "unexpected: " << to_string(self->current_message()) << endl;
}
};
}
......
......@@ -62,7 +62,7 @@ behavior server(broker* self) {
self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
},
others() >> [=] {
aout(self) << "unexpected: " << to_string(self->last_dequeued()) << endl;
aout(self) << "unexpected: " << to_string(self->current_message()) << endl;
}
};
}
......
......@@ -323,7 +323,7 @@ class curl_master : public base_actor {
m_idle_worker.push_back(spawn<curl_worker, detached+linked>(this));
}
auto worker_finished = [=] {
auto sender = last_sender();
auto sender = current_sender();
auto i = std::find(m_busy_worker.begin(), m_busy_worker.end(), sender);
m_idle_worker.push_back(*i);
m_busy_worker.erase(i);
......
......@@ -26,7 +26,7 @@ void blocking_calculator(blocking_actor* self) {
return a - b;
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
cout << "unexpected: " << to_string(self->current_message()) << endl;
}
);
}
......@@ -41,7 +41,7 @@ behavior calculator(event_based_actor* self) {
return a - b;
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
cout << "unexpected: " << to_string(self->current_message()) << endl;
}
};
}
......
......@@ -40,7 +40,7 @@ chopstick::behavior_type taken_chopstick(chopstick::pointer self, actor_addr);
chopstick::behavior_type available_chopstick(chopstick::pointer self) {
return {
[=](take_atom) {
self->become(taken_chopstick(self, self->last_sender()));
self->become(taken_chopstick(self, self->current_sender()));
return taken_atom::value;
},
[](put_atom) {
......@@ -56,7 +56,7 @@ chopstick::behavior_type taken_chopstick(chopstick::pointer self,
return busy_atom::value;
},
[=](put_atom) {
if (self->last_sender() == user) {
if (self->current_sender() == user) {
self->become(available_chopstick(self));
}
}
......@@ -127,7 +127,7 @@ class philosopher : public event_based_actor {
become(eating);
},
[=](busy_atom) {
send(last_sender() == left ? right : left, put_atom::value);
send(current_sender() == left ? right : left, put_atom::value);
send(this, eat_atom::value);
become(thinking);
}
......@@ -135,7 +135,7 @@ class philosopher : public event_based_actor {
// philosopher was *not* able to obtain the first chopstick
denied.assign(
[=](taken_atom) {
send(last_sender() == left ? left : right, put_atom::value);
send(current_sender() == left ? left : right, put_atom::value);
send(this, eat_atom::value);
become(thinking);
},
......
......@@ -37,7 +37,7 @@ ChatWidget::ChatWidget(QWidget* parent, Qt::WindowFlags f)
},
[=](const string& txt) {
// don't print own messages
if (self != self->last_sender()) {
if (self != self->current_sender()) {
print(QString::fromUtf8(txt.c_str()));
}
},
......
......@@ -78,7 +78,7 @@ void client_bhvr(event_based_actor* self, const string& host,
}
}
auto sync_send_request = [=](int lhs, const char* op, int rhs) {
self->sync_send(server, self->last_dequeued()).then(
self->sync_send(server, self->current_message()).then(
[=](result_atom, int result) {
aout(self) << lhs << " " << op << " " << rhs << " = " << result << endl;
}
......
......@@ -54,13 +54,13 @@ void client(event_based_actor* self, const string& name) {
},
[=](const string& txt) {
// don't print own messages
if (self->last_sender() != self) cout << txt << endl;
if (self->current_sender() != self) cout << txt << endl;
},
[=](const group_down_msg& g) {
cout << "*** chatroom offline: " << to_string(g.source) << endl;
},
others() >> [=]() {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
cout << "unexpected: " << to_string(self->current_message()) << endl;
}
);
}
......
......@@ -143,8 +143,8 @@ void testee(event_based_actor* self, size_t remaining) {
[=](const tree& tmsg) {
// prints the tree in its serialized format:
// @<> ( { tree ( 0, { 10, { 11, { }, 12, { }, 13, { } }, 20, { 21, { }, 22, { } } } ) } )
cout << "to_string(self->last_dequeued()): "
<< to_string(self->last_dequeued())
cout << "to_string(self->current_message()): "
<< to_string(self->current_message())
<< endl;
// prints the tree using the print member function:
// 0 { 10 { 11, 12, 13 } , 20 { 21, 22 } }
......@@ -161,7 +161,7 @@ void testee(event_based_actor* self, size_t remaining) {
// tree ( 0, { 10, { 11, { }, 12, { }, 13, { } }, 20, { 21, { }, 22, { } } } )
// )
// } )
cout << "to_string: " << to_string(self->last_dequeued()) << endl;
cout << "to_string: " << to_string(self->current_message()) << endl;
set_next_behavior();
}
);
......
......@@ -30,8 +30,8 @@
#include <exception>
#include <type_traits>
#include "caf/node_id.hpp"
#include "caf/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/attachable.hpp"
#include "caf/message_id.hpp"
#include "caf/exit_reason.hpp"
......@@ -43,11 +43,6 @@
namespace caf {
class actor_addr;
class serializer;
class deserializer;
class execution_unit;
/**
* A unique actor ID.
* @relates abstract_actor
......@@ -59,21 +54,12 @@ using actor_id = uint32_t;
*/
constexpr actor_id invalid_actor_id = 0;
class actor;
class abstract_actor;
class response_promise;
using abstract_actor_ptr = intrusive_ptr<abstract_actor>;
/**
* Base class for all actor implementations.
*/
class abstract_actor : public abstract_channel {
// needs access to m_host
friend class response_promise;
// java-like access to base class
using super = abstract_channel;
public:
/**
* Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on
......@@ -100,13 +86,6 @@ class abstract_actor : public abstract_channel {
*/
size_t detach(const attachable::token& what);
enum linking_operation {
establish_link_op,
establish_backlink_op,
remove_link_op,
remove_backlink_op
};
/**
* Links this actor to `whom`.
*/
......@@ -174,27 +153,6 @@ class abstract_actor : public abstract_channel {
*/
virtual std::set<std::string> message_types() const;
enum actor_state_flag {
// used by ...
trap_exit_flag = 0x01, // local_actor
has_timeout_flag = 0x02, // mixin::single_timeout
is_registered_flag = 0x04, // no_resume, resumable, and scoped_actor
is_initialized_flag = 0x08, // event-based actors
is_blocking_flag = 0x10 // blocking_actor
};
inline void set_flag(bool enable_flag, actor_state_flag mask) {
if (enable_flag) {
m_flags |= static_cast<int>(mask);
} else {
m_flags &= ~static_cast<int>(mask);
}
}
inline bool get_flag(actor_state_flag mask) const {
return static_cast<bool>(m_flags & static_cast<int>(mask));
}
/**
* Returns the execution unit currently used by this actor.
* @warning not thread safe
......@@ -214,22 +172,12 @@ class abstract_actor : public abstract_channel {
*/
abstract_actor(actor_id aid, node_id nid, size_t initial_ref_count = 0);
virtual bool link_impl(linking_operation op, const actor_addr& other);
/**
* Called by the runtime system to perform cleanup actions for this actor.
* Subtypes should always call this member function when overriding it.
*/
void cleanup(uint32_t reason);
bool establish_link_impl(const actor_addr& other);
bool remove_link_impl(const actor_addr& other);
bool establish_backlink_impl(const actor_addr& other);
bool remove_backlink_impl(const actor_addr& other);
/**
* Returns `exit_reason() != exit_reason::not_exited`.
*/
......@@ -244,6 +192,79 @@ class abstract_actor : public abstract_channel {
m_host = new_host;
}
/****************************************************************************
* here be dragons: end of public interface *
****************************************************************************/
public:
/** @cond PRIVATE */
enum linking_operation {
establish_link_op,
establish_backlink_op,
remove_link_op,
remove_backlink_op
};
enum actor_state_flag {
// used by ...
trap_exit_flag = 0x01, // local_actor
has_timeout_flag = 0x02, // mixin::single_timeout
is_registered_flag = 0x04, // no_resume, resumable, and scoped_actor
is_initialized_flag = 0x08, // event-based actors
is_blocking_flag = 0x10 // blocking_actor
};
inline void set_flag(bool enable_flag, actor_state_flag mask) {
m_flags = enable_flag ? m_flags | static_cast<int>(mask)
: m_flags & ~static_cast<int>(mask);
}
inline bool get_flag(actor_state_flag mask) const {
return static_cast<bool>(m_flags & static_cast<int>(mask));
}
inline bool has_timeout() const {
return get_flag(has_timeout_flag);
}
inline void has_timeout(bool value) {
set_flag(value, has_timeout_flag);
}
inline bool is_registered() const {
return get_flag(is_registered_flag);
}
void is_registered(bool value);
inline bool is_initialized() const {
return get_flag(is_initialized_flag);
}
inline void is_initialized(bool value) {
set_flag(value, is_initialized_flag);
}
inline bool is_blocking() const {
return get_flag(is_blocking_flag);
}
inline void is_blocking(bool value) {
set_flag(value, is_blocking_flag);
}
protected:
virtual bool link_impl(linking_operation op, const actor_addr& other);
bool establish_link_impl(const actor_addr& other);
bool remove_link_impl(const actor_addr& other);
bool establish_backlink_impl(const actor_addr& other);
bool remove_backlink_impl(const actor_addr& other);
inline void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(m_attachables_head);
m_attachables_head.swap(ptr);
......@@ -254,28 +275,25 @@ class abstract_actor : public abstract_channel {
bool stop_on_first_hit = false,
bool dry_run = false);
/** @cond PRIVATE */
/*
* Tries to run a custom exception handler for `eptr`.
*/
// Tries to run a custom exception handler for `eptr`.
optional<uint32_t> handle(const std::exception_ptr& eptr);
/** @endcond */
// cannot be changed after construction
const actor_id m_id;
// initially exit_reason::not_exited
// initially set to exit_reason::not_exited
std::atomic<uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, and m_links
mutable std::mutex m_mtx;
// attached functors that are executed on cleanup (for monitors, links, etc)
// attached functors that are executed on cleanup (monitors, links, etc)
attachable_ptr m_attachables_head;
// identifies the execution unit this actor is currently executed by
execution_unit* m_host;
/** @endcond */
};
} // namespace caf
......
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_BLOCKING_UNTYPED_ACTOR_HPP
#define CAF_BLOCKING_UNTYPED_ACTOR_HPP
#ifndef CAF_BLOCKING_ACTOR_HPP
#define CAF_BLOCKING_ACTOR_HPP
#include "caf/none.hpp"
......@@ -40,7 +40,7 @@ namespace caf {
/**
* A thread-mapped or context-switching actor using a blocking
* receive rather than a behavior-stack based message processing.
* @extends local_actor
* @extends mailbox_based_actor
*/
class blocking_actor
: public extend<mailbox_based_actor, blocking_actor>::
......@@ -55,10 +55,10 @@ class blocking_actor
/**************************************************************************
* utility stuff and receive() member function family *
**************************************************************************/
using timeout_type = std::chrono::high_resolution_clock::time_point;
struct receive_while_helper {
std::function<void(behavior&)> m_dq;
std::function<bool()> m_stmt;
......@@ -69,12 +69,10 @@ class blocking_actor
behavior bhvr{std::forward<Ts>(args)...};
while (m_stmt()) m_dq(bhvr);
}
};
template <class T>
struct receive_for_helper {
std::function<void(behavior&)> m_dq;
T& begin;
T end;
......@@ -84,11 +82,9 @@ class blocking_actor
behavior bhvr{std::forward<Ts>(args)...};
for (; begin != end; ++begin) m_dq(bhvr);
}
};
struct do_receive_helper {
std::function<void(behavior&)> m_dq;
behavior m_bhvr;
......@@ -102,7 +98,6 @@ class blocking_actor
void until(const bool& bvalue) {
until([&] { return bvalue; });
}
};
/**
......@@ -227,20 +222,16 @@ class blocking_actor
/** @endcond */
private:
// helper function to implement receive_(for|while) and do_receive
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
std::map<message_id, behavior> m_sync_handler;
};
class blocking_actor::functor_based : public blocking_actor {
public:
using act_fun = std::function<void(blocking_actor*)>;
template <class F, class... Ts>
......@@ -274,4 +265,4 @@ class blocking_actor::functor_based : public blocking_actor {
} // namespace caf
#endif // CAF_BLOCKING_UNTYPED_ACTOR_HPP
#endif // CAF_BLOCKING_ACTOR_HPP
......@@ -37,11 +37,10 @@
namespace caf {
/**
* A cooperatively scheduled, event-based actor implementation.
* This is the recommended base class for user-defined actors and is used
* implicitly when spawning functor-based actors without the
* `blocking_api` flag.
* @extends local_actor
* A cooperatively scheduled, event-based actor implementation. This is the
* recommended base class for user-defined actors and is used implicitly when
* spawning functor-based actors without the `blocking_api` flag.
* @extends mailbox_based_actor
*/
class event_based_actor
: public extend<mailbox_based_actor, event_based_actor>::
......
This diff is collapsed.
......@@ -224,11 +224,11 @@ class sync_sender_impl : public Base {
std::forward<Vs>(vs)...);
}
// <backward_compatibility version="0.9">
/****************************************************************************
* outdated member functions *
* deprecated member functions *
****************************************************************************/
// <backward_compatibility version="0.9">
response_handle_type sync_send_tuple(message_priority prio, const actor& dest,
message what) CAF_DEPRECATED;
......
......@@ -114,7 +114,7 @@ class invoke_policy {
<< CAF_MARG(node->mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node->mid == awaited_response) {
node.swap(self->current_element());
node.swap(self->current_mailbox_element());
auto res = invoke_fun(self, fun);
if (!res && handle_sync_failure_on_mismatch) {
CAF_LOG_WARNING("sync failure occured in actor "
......@@ -123,15 +123,15 @@ class invoke_policy {
}
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
node.swap(self->current_element());
node.swap(self->current_mailbox_element());
return im_success;
}
return im_skipped;
case msg_type::ordinary:
if (!awaited_response.valid()) {
node.swap(self->current_element());
node.swap(self->current_mailbox_element());
auto res = invoke_fun(self, fun);
node.swap(self->current_element());
node.swap(self->current_mailbox_element());
if (res) {
return im_success;
}
......@@ -164,8 +164,8 @@ class invoke_policy {
auto msg_str = to_string(msg);
# endif
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", msg = " << msg_str);
auto mid = self->current_element()->mid;
auto res = fun(self->current_element()->msg);
auto mid = self->current_mailbox_element()->mid;
auto res = fun(self->current_mailbox_element()->msg);
CAF_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str);
CAF_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str);
if (!res) {
......@@ -173,8 +173,8 @@ class invoke_policy {
}
if (res->empty()) {
// make sure synchronous requests always receive a response;
// note: !current_element() means client has forwarded the request
auto& ptr = self->current_element();
// note: !current_mailbox_element() means client has forwarded the request
auto& ptr = self->current_mailbox_element();
if (ptr) {
mid = ptr->mid;
if (mid.is_request() && !mid.is_answered()) {
......@@ -223,7 +223,7 @@ class invoke_policy {
others() >> [=] {
// inner is const inside this lambda and mutable a C++14 feature
behavior cpy = inner;
auto inner_res = cpy(self->last_dequeued());
auto inner_res = cpy(self->current_message());
if (inner_res && !handle_message_id_res(self, *inner_res, fhdl)) {
fhdl.deliver(*inner_res);
}
......
......@@ -35,7 +35,7 @@ namespace caf {
* checking. This is the recommended base class for user-defined actors and is
* used implicitly when spawning typed, functor-based actors without the
* `blocking_api` flag.
* @extends local_actor
* @extends mailbox_based_actor
*/
template <class... Rs>
class typed_event_based_actor : public
......
......@@ -48,8 +48,8 @@ using guard_type = std::unique_lock<std::mutex>;
// by std::atomic<> constructor
abstract_actor::abstract_actor(actor_id aid, node_id nid, size_t initial_count)
: super(abstract_channel::is_abstract_actor_flag,
std::move(nid), initial_count),
: abstract_channel(abstract_channel::is_abstract_actor_flag,
std::move(nid), initial_count),
m_id(aid),
m_exit_reason(exit_reason::not_exited),
m_host(nullptr) {
......@@ -57,8 +57,8 @@ abstract_actor::abstract_actor(actor_id aid, node_id nid, size_t initial_count)
}
abstract_actor::abstract_actor(size_t initial_count)
: super(abstract_channel::is_abstract_actor_flag,
detail::singletons::get_node_id(), initial_count),
: abstract_channel(abstract_channel::is_abstract_actor_flag,
detail::singletons::get_node_id(), initial_count),
m_id(detail::singletons::get_actor_registry()->next_id()),
m_exit_reason(exit_reason::not_exited),
m_host(nullptr) {
......@@ -110,6 +110,18 @@ size_t abstract_actor::detach(const attachable::token& what) {
return detach_impl(what, m_attachables_head);
}
void abstract_actor::is_registered(bool value) {
if (is_registered() == value) {
return;
}
if (value) {
detail::singletons::get_actor_registry()->inc_running();
} else {
detail::singletons::get_actor_registry()->dec_running();
}
set_flag(value, is_registered_flag);
}
bool abstract_actor::link_impl(linking_operation op, const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(op) << ", " << CAF_TSARG(other));
switch (op) {
......
......@@ -159,7 +159,7 @@ void printer_loop(blocking_actor* self) {
bool running = true;
self->receive_while([&] { return running; })(
on(atom("add"), arg_match) >> [&](std::string& str) {
auto s = self->last_sender();
auto s = self->current_sender();
if (str.empty() || s == invalid_actor_addr) {
return;
}
......@@ -174,7 +174,7 @@ void printer_loop(blocking_actor* self) {
flush_if_needed(i->second);
},
on(atom("flush")) >> [&] {
flush_output(self->last_sender());
flush_output(self->current_sender());
},
[&](const down_msg& dm) {
flush_output(dm.source);
......@@ -184,7 +184,7 @@ void printer_loop(blocking_actor* self) {
running = false;
},
others() >> [&] {
std::cerr << "*** unexpected: " << to_string(self->last_dequeued())
std::cerr << "*** unexpected: " << to_string(self->current_message())
<< std::endl;
}
);
......
......@@ -140,12 +140,12 @@ class local_broker : public event_based_actor {
CAF_LOGC_TRACE("caf::local_broker", "init$FORWARD",
CAF_TARG(what, to_string));
// local forwarding
m_group->send_all_subscribers(last_sender(), what, host());
m_group->send_all_subscribers(current_sender(), what, host());
// forward to all acquaintances
send_to_acquaintances(what);
},
[=](const down_msg&) {
auto sender = last_sender();
auto sender = current_sender();
CAF_LOGC_TRACE("caf::local_broker", "init$DOWN",
CAF_TARG(sender, to_string));
if (sender) {
......@@ -160,7 +160,7 @@ class local_broker : public event_based_actor {
}
},
others() >> [=] {
auto msg = last_dequeued();
auto msg = current_message();
CAF_LOGC_TRACE("caf::local_broker", "init$others",
CAF_TARG(msg, to_string));
send_to_acquaintances(msg);
......@@ -171,7 +171,7 @@ class local_broker : public event_based_actor {
private:
void send_to_acquaintances(const message& what) {
// send to all remote subscribers
auto sender = last_sender();
auto sender = current_sender();
CAF_LOG_DEBUG("forward message to " << m_acquaintances.size()
<< " acquaintances; " << CAF_TSARG(sender) << ", "
<< CAF_TSARG(what));
......@@ -248,11 +248,10 @@ class proxy_broker : public event_based_actor {
}
behavior make_behavior() {
return {
others() >> [=] {
m_group->send_all_subscribers(last_sender(), last_dequeued(), host());
}
};
return {others() >> [=] {
m_group->send_all_subscribers(current_sender(), current_message(),
host());
}};
}
private:
......
......@@ -41,14 +41,6 @@ local_actor::~local_actor() {
// nop
}
message& local_actor::last_dequeued() {
return m_current_element ? m_current_element->msg : m_dummy_message;
}
actor_addr& local_actor::last_sender() {
return m_current_element ? m_current_element->sender : m_dummy_sender;
}
void local_actor::monitor(const actor_addr& whom) {
if (whom == invalid_actor_addr) {
return;
......@@ -212,16 +204,24 @@ message_id local_actor::sync_send_impl(message_priority mp,
return nri.response_id();
}
void local_actor::is_registered(bool value) {
if (is_registered() == value) {
return;
//<backward_compatibility version="0.12">
message& local_actor::last_dequeued() {
if (!m_current_element) {
auto errstr = "last_dequeued called after forward_to or not in a callback";
CAF_LOG_ERROR(errstr);
throw std::logic_error(errstr);
}
if (value) {
detail::singletons::get_actor_registry()->inc_running();
} else {
detail::singletons::get_actor_registry()->dec_running();
return m_current_element->msg;
}
actor_addr& local_actor::last_sender() {
if (!m_current_element) {
auto errstr = "last_sender called after forward_to or not in a callback";
CAF_LOG_ERROR(errstr);
throw std::logic_error(errstr);
}
set_flag(value, is_registered_flag);
return m_current_element->sender;
}
//</backward_compatibility>
} // namespace caf
......@@ -22,8 +22,6 @@
#include "caf/local_actor.hpp"
#include "caf/response_promise.hpp"
using std::move;
namespace caf {
response_promise::response_promise(const actor_addr& from, const actor_addr& to,
......@@ -38,7 +36,7 @@ void response_promise::deliver(message msg) const {
}
auto to = actor_cast<abstract_actor_ptr>(m_to);
auto from = actor_cast<abstract_actor_ptr>(m_from);
to->enqueue(m_from, m_id, move(msg), from->m_host);
to->enqueue(m_from, m_id, std::move(msg), from->host());
}
} // namespace caf
......@@ -209,7 +209,7 @@ behavior basp_broker::make_behavior() {
// catch-all error handler
others() >> [=] {
CAF_LOG_ERROR("received unexpected message: "
<< to_string(last_dequeued()));
<< to_string(current_message()));
}
};
}
......
\section{Actors}
\label{Sec::Actors}
\lib provides several actor implementations, each covering a particular use case.
The class \lstinline^local_actor^ is the base class for all implementations, except for (remote) proxy actors.
......@@ -17,6 +18,7 @@ When dealing with typed actors, the types are \lstinline^typed_event_based_actor
\clearpage
\subsection{Interface}
\label{Sec::Actors::Interfaces}
\begin{lstlisting}
class local_actor;
......@@ -26,23 +28,23 @@ class local_actor;
\begin{tabular*}{\textwidth}{m{0.45\textwidth}m{0.5\textwidth}}
\multicolumn{2}{m{\linewidth}}{\large{\textbf{Member functions}}\vspace{3pt}} \\
\\
\hline
\lstinline^quit(uint32_t reason = normal)^ & Finishes execution of this actor \\
\hline
\\
\multicolumn{2}{l}{\textbf{Observers}\vspace{3pt}} \\
\hline
\lstinline^actor_addr address()^ & Returns the address of this actor \\
\hline
\lstinline^bool trap_exit()^ & Checks whether this actor traps exit messages \\
\hline
\lstinline^message last_dequeued()^ & Returns the last message that was dequeued from the actor's mailbox\newline\textbf{Note}: Only set during callback invocation \\
\lstinline^message& current_message()^ & Returns the currently processed message\newline\textbf{Warning}: Only set during callback invocation; calling this function after forwarding the message or while not in a callback is undefined behavior \\
\hline
\lstinline^actor_addr last_sender()^ & Returns the sender of the last dequeued message\newline\textbf{Note}: Only set during callback invocation \\
\lstinline^actor_addr& current_sender()^ & Returns the sender of the current message\newline\textbf{Warning}: Only set during callback invocation; calling this function after forwarding the message or while not in a callback is undefined behavior \\
\hline
\lstinline^vector<group> joined_groups()^ & Returns all subscribed groups \\
\hline
\\
\multicolumn{2}{l}{\textbf{Modifiers}\vspace{3pt}} \\
\hline
\lstinline^quit(uint32_t reason = normal)^ & Finishes execution of this actor \\
\hline
\lstinline^void trap_exit(bool enabled)^ & Enables or disables trapping of exit messages \\
\hline
\lstinline^void join(const group& g)^ & Subscribes to group \lstinline^g^ \\
......@@ -53,7 +55,7 @@ class local_actor;
\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)^ & Adds a unidirectional monitor to \lstinline^whom^ (see Section \ref{Sec::Management::Monitors}) \\
\lstinline^void monitor(actor whom)^ & Unidirectionally monitors \lstinline^whom^ (see Section \ref{Sec::Management::Monitors}) \\
\hline
\lstinline^void demonitor(actor whom)^ & Removes a monitor from \lstinline^whom^ \\
\hline
......
......@@ -198,3 +198,11 @@ This feature has been removed without substitution.
This release removes the (since 0.9 deprecated) \lstinline^cppa^ headers and deprecates all \lstinline^*_send_tuple^ versions (simply use the function without \lstinline^_tuple^ suffix).
In case you were using the old \lstinline^cppa::options_description^ API, you can migrate to the new API for filtering command line arguments (cf. \ref{Sec::Messages::FilterCLI}).
Most importantly, version 0.13 slightly changes \lstinline^last_dequeued^ and \lstinline^last_sender^.
Both functions will now cause undefined behavior (dereferencing a \lstinline^nullptr^) instead of returning dummy values when accessed from outside a callback or after forwarding the current message.
Besides, these function names were not a good choice in the first place, since ``last'' implies accessing data received in the past.
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}).
No preview for this file type
......@@ -17,22 +17,22 @@ size_t s_pongs = 0;
behavior ping_behavior(local_actor* self, size_t num_pings) {
return {
[=](pong_atom, int value) -> message {
if (!self->last_sender()) {
if (!self->current_sender()) {
CAF_PRINT("last_sender() invalid!");
}
CAF_PRINT("received {'pong', " << value << "}");
// cout << to_string(self->last_dequeued()) << endl;
// cout << to_string(self->current_message()) << endl;
if (++s_pongs >= num_pings) {
CAF_PRINT("reached maximum, send {'EXIT', user_defined} "
<< "to last sender and quit with normal reason");
self->send_exit(self->last_sender(),
self->send_exit(self->current_sender(),
exit_reason::user_shutdown);
self->quit();
}
return make_message(ping_atom::value, value);
},
others() >> [=] {
CAF_LOGF_ERROR("unexpected; " << to_string(self->last_dequeued()));
CAF_LOGF_ERROR("unexpected; " << to_string(self->current_message()));
self->quit(exit_reason::user_shutdown);
}
};
......@@ -45,7 +45,7 @@ behavior pong_behavior(local_actor* self) {
return make_message(pong_atom::value, value + 1);
},
others() >> [=] {
CAF_LOGF_ERROR("unexpected; " << to_string(self->last_dequeued()));
CAF_LOGF_ERROR("unexpected; " << to_string(self->current_sender()));
self->quit(exit_reason::user_shutdown);
}
};
......
......@@ -195,7 +195,7 @@ inline void caf_check_value(V1 v1, V2 v2, const char* fname, size_t line,
#define CAF_UNEXPECTED_TOUT() caf_unexpected_timeout(__FILE__, __LINE__)
#define CAF_UNEXPECTED_MSG(selfptr) \
caf_unexpected_message(__FILE__, __LINE__, selfptr->last_dequeued())
caf_unexpected_message(__FILE__, __LINE__, selfptr->current_message())
// some convenience macros for defining callbacks
#define CAF_CHECKPOINT_CB() \
......
......@@ -39,7 +39,7 @@ void testee::on_exit() {
behavior testee::make_behavior() {
return {
others() >> [=] {
return last_dequeued();
return current_message();
}
};
}
......@@ -61,8 +61,8 @@ behavior tester(event_based_actor* self, const actor& aut) {
// must be still alive at this point
CAF_CHECK_EQUAL(s_testees.load(), 1);
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(self->last_dequeued().vals()->get_reference_count(), 1);
CAF_CHECK(&msg == self->last_dequeued().at(0));
CAF_CHECK_EQUAL(self->current_message().vals()->get_reference_count(), 1);
CAF_CHECK(&msg == self->current_message().at(0));
// testee might be still running its cleanup code in
// another worker thread; by waiting some milliseconds, we make sure
// testee had enough time to return control to the scheduler
......
......@@ -57,7 +57,7 @@ void pong(event_based_actor* self) {
self->become(
[=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_CHECKPOINT();
self->monitor(self->last_sender());
self->monitor(self->current_sender());
// set next behavior
self->become(
[](ping_atom, int val) {
......
......@@ -27,9 +27,9 @@ testee::~testee() {
behavior testee::make_behavior() {
return {
others() >> [=] {
CAF_CHECK_EQUAL(last_dequeued().cvals()->get_reference_count(), 2);
CAF_CHECK_EQUAL(current_message().cvals()->get_reference_count(), 2);
quit();
return std::move(last_dequeued());
return std::move(current_message());
}
};
}
......@@ -59,13 +59,13 @@ behavior tester::make_behavior() {
send(m_aut, m_msg);
return {
on(1, 2, 3) >> [=] {
CAF_CHECK_EQUAL(last_dequeued().cvals()->get_reference_count(), 2);
CAF_CHECK(last_dequeued().cvals().get() == m_msg.cvals().get());
CAF_CHECK_EQUAL(current_message().cvals()->get_reference_count(), 2);
CAF_CHECK(current_message().cvals().get() == m_msg.cvals().get());
},
[=](const down_msg& dm) {
CAF_CHECK(dm.source == m_aut);
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
CAF_CHECK_EQUAL(last_dequeued().cvals()->get_reference_count(), 1);
CAF_CHECK_EQUAL(current_message().cvals()->get_reference_count(), 1);
quit();
},
others() >> CAF_UNEXPECTED_MSG_CB(this)
......@@ -79,8 +79,8 @@ void test_message_lifetime_in_scoped_actor() {
self->receive(
on(1, 2, 3) >> [&] {
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2);
CAF_CHECK_EQUAL(self->last_dequeued().cvals()->get_reference_count(), 2);
CAF_CHECK(self->last_dequeued().cvals().get() == msg.cvals().get());
CAF_CHECK_EQUAL(self->current_message().cvals()->get_reference_count(), 2);
CAF_CHECK(self->current_message().cvals().get() == msg.cvals().get());
}
);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 1);
......@@ -89,8 +89,8 @@ void test_message_lifetime_in_scoped_actor() {
self->receive(
[&](int& value) {
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 1);
CAF_CHECK_EQUAL(self->last_dequeued().cvals()->get_reference_count(), 1);
CAF_CHECK(self->last_dequeued().cvals().get() != msg.cvals().get());
CAF_CHECK_EQUAL(self->current_message().cvals()->get_reference_count(), 1);
CAF_CHECK(self->current_message().cvals().get() != msg.cvals().get());
value = 10;
}
);
......
......@@ -30,7 +30,7 @@ void reflector(event_based_actor* self) {
self->become(others() >> [=] {
CAF_PRINT("reflect and quit");
self->quit();
return self->last_dequeued();
return self->current_message();
});
}
......@@ -221,7 +221,7 @@ class client : public event_based_actor {
CAF_PRINT("test group communication via network (inverted setup)");
become(on(atom("GClient")) >> [=]()->message {
CAF_CHECKPOINT();
auto cptr = last_sender();
auto cptr = current_sender();
auto s5c = spawn<monitored>(spawn5_client);
// set next behavior
await_down(this, s5c, [=] {
......@@ -262,7 +262,7 @@ class server : public event_based_actor {
CAF_PRINT("await {'SpawnPing'}");
return (on(atom("SpawnPing")) >> [=]()->message {
CAF_PRINT("received {'SpawnPing'}");
auto client = last_sender();
auto client = current_sender();
if (!client) {
CAF_PRINT("last_sender() invalid!");
}
......@@ -290,21 +290,23 @@ class server : public event_based_actor {
void await_foobars() {
CAF_PRINT("await foobars");
auto foobars = make_shared<int>(0);
become(on(atom("foo"), atom("bar"), arg_match) >> [=](int i)->message {
++*foobars;
if (i == 99) {
CAF_CHECK_EQUAL(*foobars, 100);
test_group_comm();
become(
on(atom("foo"), atom("bar"), arg_match) >> [=](int i)->message {
++*foobars;
if (i == 99) {
CAF_CHECK_EQUAL(*foobars, 100);
test_group_comm();
}
return std::move(current_message());
}
return last_dequeued();
});
);
}
void test_group_comm() {
CAF_PRINT("test group communication via network");
become(on(atom("GClient")) >> [=]()->message {
CAF_CHECKPOINT();
auto cptr = last_sender();
auto cptr = current_sender();
auto s5c = spawn<monitored>(spawn5_client);
await_down(this, s5c, [=] {
CAF_CHECKPOINT();
......
......@@ -11,8 +11,8 @@ using namespace caf;
void test_serial_reply() {
auto mirror_behavior = [=](event_based_actor* self) {
self->become(others() >> [=]() -> message {
CAF_PRINT("return self->last_dequeued()");
return self->last_dequeued();
CAF_PRINT("return self->current_message()");
return self->current_message();
});
};
auto master = spawn([=](event_based_actor* self) {
......
......@@ -8,9 +8,9 @@ void test_simple_reply_response() {
auto s = spawn([](event_based_actor* self) -> behavior {
return (
others() >> [=]() -> message {
CAF_CHECK(self->last_dequeued() == make_message(ok_atom::value));
CAF_CHECK(self->current_message() == make_message(ok_atom::value));
self->quit();
return self->last_dequeued();
return self->current_message();
}
);
});
......@@ -18,7 +18,7 @@ void test_simple_reply_response() {
self->send(s, ok_atom::value);
self->receive(
others() >> [&] {
CAF_CHECK(self->last_dequeued() == make_message(ok_atom::value));
CAF_CHECK(self->current_message() == make_message(ok_atom::value));
}
);
self->await_all_other_actors_done();
......
......@@ -273,7 +273,7 @@ behavior echo_actor::make_behavior() {
return {
others() >> [=]() -> message {
quit(exit_reason::normal);
return last_dequeued();
return current_message();
}
};
}
......@@ -297,7 +297,7 @@ behavior simple_mirror::make_behavior() {
return {
others() >> [=] {
CAF_CHECKPOINT();
return last_dequeued();
return current_message();
}
};
}
......@@ -499,7 +499,7 @@ void test_spawn() {
on("hi", arg_match) >> [&](actor from) {
s->sync_send(from, "whassup?", s).await(
on_arg_match >> [&](const string& str) -> string {
CAF_CHECK(s->last_sender() != nullptr);
CAF_CHECK(s->current_sender() != nullptr);
CAF_CHECK_EQUAL(str, "nothing");
return "goodbye!";
},
......@@ -597,7 +597,7 @@ void test_spawn() {
return {
others() >> [=] {
// forward message and die
send(m_pal, last_dequeued());
send(m_pal, current_message());
quit();
}
};
......@@ -698,7 +698,7 @@ void test_spawn() {
flags |= 0x08;
},
others() >> [&]() {
CAF_FAILURE("unexpected message: " << to_string(self->last_dequeued()));
CAF_FAILURE("unexpected message: " << to_string(self->current_message()));
},
after(chrono::milliseconds(500)) >> [&]() {
CAF_FAILURE("timeout in file " << __FILE__ << " in line " << __LINE__);
......
......@@ -26,7 +26,7 @@ struct sync_mirror : event_based_actor {
behavior make_behavior() override {
return {
others() >> [=] {
return last_dequeued();
return current_message();
}
};
}
......@@ -158,10 +158,10 @@ class D : public popular_actor {
behavior make_behavior() override {
return {
others() >> [=] {
return sync_send(buddy(), last_dequeued()).then(
return sync_send(buddy(), std::move(current_message())).then(
others() >> [=]() -> message {
quit();
return last_dequeued();
return std::move(current_message());
}
);
}
......@@ -211,7 +211,7 @@ class server : public event_based_actor {
void test_sync_send() {
scoped_actor self;
self->on_sync_failure([&] {
CAF_FAILURE("received: " << to_string(self->last_dequeued()));
CAF_FAILURE("received: " << to_string(self->current_message()));
});
self->spawn<monitored + blocking_api>([](blocking_actor* s) {
CAF_LOGC_TRACE("NONE", "main$sync_failure_test", "id = " << s->id());
......@@ -224,7 +224,7 @@ void test_sync_send() {
}
);
s->on_sync_failure([=] {
CAF_FAILURE("received: " << to_string(s->last_dequeued()));
CAF_FAILURE("received: " << to_string(s->current_message()));
});
s->sync_send(foi, i_atom::value).await(
[&](int i) {
......@@ -369,10 +369,10 @@ void test_sync_send() {
s->sync_send(serv, request_atom::value).await(
[=](response_atom) {
CAF_CHECKPOINT();
CAF_CHECK_EQUAL(s->last_sender(), work);
CAF_CHECK_EQUAL(s->current_sender(), work);
},
others() >> [&] {
CAF_PRINTERR("unexpected message: " << to_string(s->last_dequeued()));
CAF_PRINTERR("unexpected message: " << to_string(s->current_message()));
}
);
// first 'request', then 'idle'
......@@ -381,7 +381,7 @@ void test_sync_send() {
handle.await(
[=](response_atom) {
CAF_CHECKPOINT();
CAF_CHECK_EQUAL(s->last_sender(), work);
CAF_CHECK_EQUAL(s->current_sender(), work);
},
others() >> CAF_UNEXPECTED_MSG_CB(s)
);
......
......@@ -272,7 +272,7 @@ int_actor::behavior_type int_fun2(int_actor::pointer self) {
self->trap_exit(true);
return {
[=](int i) {
self->monitor(self->last_sender());
self->monitor(self->current_sender());
return i * i;
},
[=](const down_msg& dm) {
......
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