Commit e177410c authored by Dominik Charousset's avatar Dominik Charousset

Streamline actor impls using mixins

parent f72ff6a5
...@@ -55,9 +55,9 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -55,9 +55,9 @@ behavior ping(event_based_actor* self, size_t num_pings) {
[=](kickoff_atom, const actor& pong) { [=](kickoff_atom, const actor& pong) {
self->send(pong, ping_atom::value, int32_t(1)); self->send(pong, ping_atom::value, int32_t(1));
self->become ( self->become (
[=](pong_atom, int32_t value) -> message { [=](pong_atom, int32_t value) -> result<ping_atom, int32_t> {
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit();
return make_message(ping_atom::value, value + 1); return {ping_atom::value, value + 1};
} }
); );
} }
...@@ -66,8 +66,8 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -66,8 +66,8 @@ behavior ping(event_based_actor* self, size_t num_pings) {
behavior pong() { behavior pong() {
return { return {
[](ping_atom, int32_t value) { [](ping_atom, int32_t value) -> result<pong_atom, int32_t> {
return make_message(pong_atom::value, value); return {pong_atom::value, value};
} }
}; };
} }
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -124,6 +125,14 @@ public: ...@@ -124,6 +125,14 @@ public:
/// @cond PRIVATE /// @cond PRIVATE
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) {
enqueue(mailbox_element::make(std::move(sender), mid,
{}, std::forward<Ts>(xs)...),
ctx);
}
enum linking_operation { enum linking_operation {
establish_link_op, establish_link_op,
establish_backlink_op, establish_backlink_op,
......
...@@ -94,6 +94,18 @@ public: ...@@ -94,6 +94,18 @@ public:
return system_; return system_;
} }
/// @cond PRIVATE
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) {
CAF_ASSERT(! mid.is_request());
enqueue(std::move(sender), mid,
make_message(std::forward<Ts>(xs)...), ctx);
}
/// @endcond
protected: protected:
abstract_group(actor_system& sys, module_ptr module, abstract_group(actor_system& sys, module_ptr module,
std::string group_id, const node_id& nid); std::string group_id, const node_id& nid);
......
...@@ -71,6 +71,8 @@ public: ...@@ -71,6 +71,8 @@ public:
// grant access to private ctor // grant access to private ctor
friend class local_actor; friend class local_actor;
using signatures = none_t;
// allow conversion via actor_cast // allow conversion via actor_cast
template <class, class, int> template <class, class, int>
friend class actor_cast_access; friend class actor_cast_access;
......
...@@ -23,11 +23,12 @@ ...@@ -23,11 +23,12 @@
#include <memory> #include <memory>
#include <functional> #include <functional>
#include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -38,7 +39,8 @@ namespace caf { ...@@ -38,7 +39,8 @@ namespace caf {
/// callback to another object, thus serving as gateway to /// callback to another object, thus serving as gateway to
/// allow any object to interact with other actors. /// allow any object to interact with other actors.
/// @extends local_actor /// @extends local_actor
class actor_companion : public abstract_event_based_actor<behavior, true> { class actor_companion : public extend<local_actor, actor_companion>::
with<mixin::sender> {
public: public:
using lock_type = detail::shared_spinlock; using lock_type = detail::shared_spinlock;
using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>; using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>;
......
...@@ -98,9 +98,6 @@ public: ...@@ -98,9 +98,6 @@ public:
/// function `fac` using the dispatch policy `pol`. /// function `fac` using the dispatch policy `pol`.
static actor make(execution_unit* ptr, size_t n, factory fac, policy pol); static actor make(execution_unit* ptr, size_t n, factory fac, policy pol);
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
void enqueue(mailbox_element_ptr what, execution_unit* host) override; void enqueue(mailbox_element_ptr what, execution_unit* host) override;
actor_pool(actor_config& cfg); actor_pool(actor_config& cfg);
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
strong_actor_ptr get(actor_id key) const; strong_actor_ptr get(actor_id key) const;
/// Associates a local actor with its ID. /// Associates a local actor with its ID.
void put(actor_id key, const strong_actor_ptr& value); void put(actor_id key, strong_actor_ptr value);
/// Removes an actor from this registry, /// Removes an actor from this registry,
/// leaving `reason` for future reference. /// leaving `reason` for future reference.
...@@ -81,12 +81,6 @@ public: ...@@ -81,12 +81,6 @@ public:
/// Removes a name mapping. /// Removes a name mapping.
void erase(atom_value key); void erase(atom_value key);
/// @cond PRIVATE
void put(actor_id, actor_control_block*);
/// @endcond
using name_map = std::unordered_map<atom_value, strong_actor_ptr>; using name_map = std::unordered_map<atom_value, strong_actor_ptr>;
name_map named_actors() const; name_map named_actors() const;
......
...@@ -32,10 +32,10 @@ ...@@ -32,10 +32,10 @@
#include "caf/actor_config.hpp" #include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/response_handle.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
namespace caf { namespace caf {
...@@ -45,13 +45,15 @@ namespace caf { ...@@ -45,13 +45,15 @@ namespace caf {
/// @extends local_actor /// @extends local_actor
class blocking_actor class blocking_actor
: public extend<local_actor, blocking_actor>:: : public extend<local_actor, blocking_actor>::
with<mixin::requester<blocking_response_handle_tag>::impl>, with<mixin::requester, mixin::sender>,
public dynamically_typed_actor_base { public dynamically_typed_actor_base {
public: public:
using super = extend<local_actor, blocking_actor>::
with<mixin::requester, mixin::sender>;
using behavior_type = behavior; using behavior_type = behavior;
using super = extend<local_actor, blocking_actor>:: using signatures = none_t;
with<mixin::requester<blocking_response_handle_tag>::impl>;
blocking_actor(actor_config& sys); blocking_actor(actor_config& sys);
......
...@@ -28,32 +28,69 @@ ...@@ -28,32 +28,69 @@
namespace caf { namespace caf {
/// Checks whether `R` does support an input of type `{Ts...}` template <class T>
/// via static assertion. struct output_types_of {
template <class... Sigs, class... Ts> // nop
void check_typed_input(const typed_actor<Sigs...>&, };
const detail::type_list<Ts...>&) {
static_assert(detail::tl_index_of< template <class In, class Out>
detail::type_list<Ts...>, struct output_types_of<typed_mpi<In, Out>> {
atom_value using type = Out;
>::value == -1, };
"atom(...) notation is not sufficient for static type "
"checking, please use atom_constant instead in this context"); template <class T>
static_assert(detail::tl_exists< constexpr typename std::remove_pointer<T>::type::signatures signatures_of() {
detail::type_list<Sigs...>, return {};
detail::input_is<detail::type_list<Ts...>>::template eval
>::value >= 0,
"typed actor does not support given input");
} }
template <class... Ts> template <class T>
void check_typed_input(const actor&, const detail::type_list<Ts...>&) { constexpr bool statically_typed() {
// nop return ! std::is_same<
none_t,
typename std::remove_pointer<T>::type::signatures
>::value;
}
template <class Signatures, class MessageTypes>
constexpr bool actor_accepts_message(Signatures, MessageTypes) {
return detail::tl_exists<
Signatures,
detail::input_is<MessageTypes>::template eval
>::value;
}
template <class MessageTypes>
constexpr bool actor_accepts_message(none_t, MessageTypes) {
return true;
}
template <class Signatures, class MessageTypes>
constexpr typename output_types_of<
typename detail::tl_find<
Signatures,
detail::input_is<MessageTypes>::template eval
>::type
>::type
response_to(Signatures, MessageTypes) {
return {};
}
template <class MessageTypes>
constexpr none_t response_to(none_t, MessageTypes) {
return {};
} }
template <class... Ts> template <class... Ts>
void check_typed_input(const group&, const detail::type_list<Ts...>&) { constexpr bool is_void_response(detail::type_list<Ts...>) {
// nop return false;
}
constexpr bool is_void_response(detail::type_list<void>) {
return true;
}
constexpr bool is_void_response(none_t) {
return true; // true for the purpose of type checking performed by send()
} }
} // namespace caf } // namespace caf
......
...@@ -62,7 +62,7 @@ public: ...@@ -62,7 +62,7 @@ public:
template <class T> template <class T>
typename std::enable_if< typename std::enable_if<
! detail::is_iterable<T>::value, ! detail::is_iterable<T>::value && ! std::is_pointer<T>::value,
std::string std::string
>::type >::type
operator()(const T& x) const { operator()(const T& x) const {
...@@ -145,6 +145,15 @@ public: ...@@ -145,6 +145,15 @@ public:
return decompose_array(xs, S); return decompose_array(xs, S);
} }
template <class T>
typename std::enable_if<
std::is_pointer<T>::value,
std::string
>::type
operator()(const T ptr) const {
return ptr ? "*" + (*this)(*ptr) : "<nullptr>";
}
private: private:
template <class T> template <class T>
auto impl(const T* x) const -> decltype(to_string(*x)) { auto impl(const T* x) const -> decltype(to_string(*x)) {
......
...@@ -73,8 +73,11 @@ template <class Arguments> ...@@ -73,8 +73,11 @@ template <class Arguments>
struct input_is { struct input_is {
template <class Signature> template <class Signature>
struct eval { struct eval {
static constexpr bool value = static constexpr bool value = false;
std::is_same<Arguments, typename Signature::input_types>::value; };
template <class Out>
struct eval<typed_mpi<Arguments, Out>> {
static constexpr bool value = true;
}; };
}; };
...@@ -166,7 +169,7 @@ struct deduce_lifted_output_type<type_list<typed_continue_helper<R>>> { ...@@ -166,7 +169,7 @@ struct deduce_lifted_output_type<type_list<typed_continue_helper<R>>> {
}; };
template <class Signatures, class InputTypes> template <class Signatures, class InputTypes>
struct deduce_output_type { struct deduce_output_type_impl {
using signature = using signature =
typename tl_find< typename tl_find<
Signatures, Signatures,
...@@ -181,6 +184,25 @@ struct deduce_output_type { ...@@ -181,6 +184,25 @@ struct deduce_output_type {
using tuple_type = typename detail::tl_apply<type, std::tuple>::type; using tuple_type = typename detail::tl_apply<type, std::tuple>::type;
}; };
template <class InputTypes>
struct deduce_output_type_impl<none_t, InputTypes> {
using type = message;
using delegated_type = delegated<message>;
using tuple_type = std::tuple<message>;
};
template <class Handle, class InputTypes>
struct deduce_output_type
: deduce_output_type_impl<typename Handle::signatures, InputTypes> {
// nop
};
template <class T, class InputTypes>
struct deduce_output_type<T*, InputTypes>
: deduce_output_type_impl<typename T::signatures, InputTypes> {
// nop
};
template <class... Ts> template <class... Ts>
struct common_result_type; struct common_result_type;
......
...@@ -22,21 +22,40 @@ ...@@ -22,21 +22,40 @@
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
namespace caf { namespace caf {
template <>
class behavior_type_of<event_based_actor> {
public:
using type = behavior;
};
/// A cooperatively scheduled, event-based actor implementation. This is the /// A cooperatively scheduled, event-based actor implementation. This is the
/// recommended base class for user-defined actors. /// recommended base class for user-defined actors.
/// @extends local_actor /// @extends local_actor
class event_based_actor : public abstract_event_based_actor<behavior, true> { class event_based_actor : public extend<local_actor, event_based_actor>::
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base {
public: public:
using super = abstract_event_based_actor<behavior, true>; using super = extend<local_actor, event_based_actor>::
with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t;
using behavior_type = behavior;
explicit event_based_actor(actor_config& cfg); explicit event_based_actor(actor_config& cfg);
......
...@@ -126,7 +126,7 @@ public: ...@@ -126,7 +126,7 @@ public:
class R = class R =
typename function_view_flattened_result< typename function_view_flattened_result<
typename detail::deduce_output_type< typename detail::deduce_output_type<
typename type::signatures, type,
detail::type_list< detail::type_list<
typename detail::implicit_conversions< typename detail::implicit_conversions<
typename std::decay<Ts>::type typename std::decay<Ts>::type
......
...@@ -29,6 +29,7 @@ namespace caf { ...@@ -29,6 +29,7 @@ namespace caf {
template <class> class optional; template <class> class optional;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class behavior_type_of;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> class typed_continue_helper; template <class> class typed_continue_helper;
...@@ -42,6 +43,7 @@ template <class...> class result; ...@@ -42,6 +43,7 @@ template <class...> class result;
template <class...> class delegated; template <class...> class delegated;
template <class...> class typed_actor; template <class...> class typed_actor;
template <class...> class typed_response_promise; template <class...> class typed_response_promise;
template <class...> class typed_event_based_actor;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <string> #include <string>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
...@@ -45,6 +46,8 @@ public: ...@@ -45,6 +46,8 @@ public:
template <class, class, int> template <class, class, int>
friend class actor_cast_access; friend class actor_cast_access;
using signatures = none_t;
group() = default; group() = default;
group(group&&) = default; group(group&&) = default;
......
...@@ -28,6 +28,11 @@ enum invoke_message_result { ...@@ -28,6 +28,11 @@ enum invoke_message_result {
im_dropped im_dropped
}; };
constexpr const char* to_string(invoke_message_result x) {
return x == im_success ? "im_success"
: (x == im_skipped ? "im_skipped" : "im_dropped" );
}
} // namespace caf } // namespace caf
#endif // CAF_INVOKE_MESSAGE_RESULT_HPP #endif // CAF_INVOKE_MESSAGE_RESULT_HPP
......
This diff is collapsed.
...@@ -22,11 +22,11 @@ ...@@ -22,11 +22,11 @@
#include <cstddef> #include <cstddef>
#include "caf/actor.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/memory.hpp"
#include "caf/detail/embedded.hpp" #include "caf/detail/embedded.hpp"
...@@ -80,11 +80,19 @@ public: ...@@ -80,11 +80,19 @@ public:
static unique_ptr make(strong_actor_ptr sender, message_id id, static unique_ptr make(strong_actor_ptr sender, message_id id,
forwarding_stack stages, message msg); forwarding_stack stages, message msg);
template <class... Ts> template <class T, class... Ts>
static unique_ptr make_joint(strong_actor_ptr sender, message_id id, static typename std::enable_if<
forwarding_stack stages, Ts&&... xs) { ! std::is_same<typename std::decay<T>::type, message>::value
|| (sizeof...(Ts) > 0),
unique_ptr
>::type
make(strong_actor_ptr sender, message_id id,
forwarding_stack stages, T&& x, Ts&&... xs) {
using value_storage = using value_storage =
detail::tuple_vals< detail::tuple_vals<
typename unbox_message_element<
typename detail::strip_and_convert<T>::type
>::type,
typename unbox_message_element< typename unbox_message_element<
typename detail::strip_and_convert<Ts>::type typename detail::strip_and_convert<Ts>::type
>::type... >::type...
...@@ -93,6 +101,7 @@ public: ...@@ -93,6 +101,7 @@ public:
using storage = detail::pair_storage<mailbox_element, value_storage>; using storage = detail::pair_storage<mailbox_element, value_storage>;
auto ptr = detail::memory::create<storage>(tk, std::move(sender), id, auto ptr = detail::memory::create<storage>(tk, std::move(sender), id,
std::move(stages), std::move(stages),
std::forward<T>(x),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
ptr->first.msg.reset(&(ptr->second), false); ptr->first.msg.reset(&(ptr->second), false);
return unique_ptr{&(ptr->first)}; return unique_ptr{&(ptr->first)};
......
...@@ -119,10 +119,12 @@ public: ...@@ -119,10 +119,12 @@ public:
return result; return result;
} }
static inline message_id make(message_priority prio static constexpr message_id make() {
= message_priority::normal) { return message_id{};
message_id res; }
return prio == message_priority::high ? res.with_high_priority() : res;
static constexpr message_id make(message_priority prio) {
return prio == message_priority::high ? high_prioity_flag_mask : 0;
} }
long compare(const message_id& other) const { long compare(const message_id& other) const {
...@@ -136,9 +138,10 @@ public: ...@@ -136,9 +138,10 @@ public:
} }
private: private:
explicit inline message_id(uint64_t value) : value_(value) { constexpr message_id(uint64_t value) : value_(value) {
// nop // nop
} }
uint64_t value_; uint64_t value_;
}; };
......
...@@ -17,11 +17,12 @@ ...@@ -17,11 +17,12 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #ifndef CAF_MIXIN_BEHAVIOR_CHANGER_HPP
#define CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #define CAF_MIXIN_BEHAVIOR_CHANGER_HPP
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
...@@ -29,29 +30,24 @@ ...@@ -29,29 +30,24 @@
#include "caf/behavior_policy.hpp" #include "caf/behavior_policy.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
namespace caf { namespace caf {
namespace mixin {
/// Base type for statically and dynamically typed event-based actors. /// A `behavior_changer` is an actor that supports
/// @tparam BehaviorType Denotes the expected type for become(). /// `self->become(...)` and `self->unbecome()`.
/// @tparam HasSyncSend Configures whether this base extends `requester`. template <class Base, class Derived>
/// @tparam Base Either `local_actor` (default) or a subtype thereof. class behavior_changer : public Base {
template <class BehaviorType, bool HasSyncSend, class Base = local_actor>
class abstract_event_based_actor : public Base,
public actor_marker<BehaviorType>::type {
public: public:
using behavior_type = BehaviorType; using behavior_type = typename behavior_type_of<Derived>::type;
template <class... Ts> template <class... Ts>
abstract_event_based_actor(Ts&&... xs) : Base(std::forward<Ts>(xs)...) { behavior_changer(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop // nop
} }
/****************************************************************************
* become() member function family *
****************************************************************************/
void become(behavior_type bhvr) { void become(behavior_type bhvr) {
this->do_become(std::move(bhvr.unbox()), true); this->do_become(std::move(bhvr.unbox()), true);
} }
...@@ -84,23 +80,7 @@ public: ...@@ -84,23 +80,7 @@ public:
} }
}; };
template <class BehaviorType, class Base> } // namespace mixin
class abstract_event_based_actor<BehaviorType, true, Base>
: public extend<abstract_event_based_actor<BehaviorType, false, Base>>
::template with<mixin::requester<nonblocking_response_handle_tag>
::template impl> {
public:
using super
= typename extend<abstract_event_based_actor<BehaviorType, false, Base>>
::template with<mixin::requester<nonblocking_response_handle_tag>
::template impl>;
template <class... Ts>
abstract_event_based_actor(Ts&&... xs) : super(std::forward<Ts>(xs)...) {
// nop
}
};
} // namespace caf } // namespace caf
#endif // CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #endif // CAF_MIXIN_BEHAVIOR_CHANGER_HPP
...@@ -17,12 +17,13 @@ ...@@ -17,12 +17,13 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_SYNC_SENDER_HPP #ifndef CAF_MIXIN_REQUESTER_HPP
#define CAF_MIXIN_SYNC_SENDER_HPP #define CAF_MIXIN_REQUESTER_HPP
#include <tuple> #include <tuple>
#include <chrono> #include <chrono>
#include "caf/fwd.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
...@@ -33,102 +34,45 @@ ...@@ -33,102 +34,45 @@
namespace caf { namespace caf {
namespace mixin { namespace mixin {
template <class Base, class Subtype, class HandleTag> /// A `requester` is an actor that supports
class requester_impl : public Base { /// `self->request(...).{then|await|receive}`.
template <class Base, class Subtype>
class requester : public Base {
public: public:
using response_handle_type = response_handle<Subtype, message, HandleTag>;
template <class... Ts> template <class... Ts>
requester_impl(Ts&&... xs) : Base(std::forward<Ts>(xs)...) { requester(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop // nop
} }
/****************************************************************************
* request(...) *
****************************************************************************/
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @returns A handle identifying a future-like handle to the response.
/// @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 `dest == invalid_actor`
template <class... Ts>
response_handle_type request(message_priority mp, const actor& dest,
const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return {dptr()->request_impl(mp, dest, timeout, std::forward<Ts>(xs)...),
dptr()};
}
/// Sends `{xs...}` as a synchronous message to `dest`.
/// @returns A handle identifying a future-like handle to the response.
/// @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 `dest == invalid_actor`
template <class... Ts>
response_handle_type request(const actor& dest,
const duration& timeout, Ts&&... xs) {
return request(message_priority::normal, dest,
timeout, std::forward<Ts>(xs)...);
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`. /// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @returns A handle identifying a future-like handle to the response. /// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the /// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor. /// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor` template <message_priority P = message_priority::normal,
template <class... Sigs, class... Ts> class Handle = actor, class... Ts>
response_handle<Subtype,
typename detail::deduce_output_type<
detail::type_list<Sigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>
>::type,
HandleTag>
request(message_priority mp, const typed_actor<Sigs...>& dest,
const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
token tk;
check_typed_input(dest, tk);
return {dptr()->request_impl(mp, dest, timeout, std::forward<Ts>(xs)...),
dptr()};
}
/// Sends `{xs...}` as a synchronous message to `dest`.
/// @returns A handle identifying a future-like handle to the response.
/// @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 `dest == invalid_actor`
template <class... Sigs, class... Ts>
response_handle<Subtype, response_handle<Subtype,
typename detail::deduce_output_type< typename detail::deduce_output_type<
detail::type_list<Sigs...>, Handle,
detail::type_list< detail::type_list<
typename detail::implicit_conversions< typename detail::implicit_conversions<
typename std::decay<Ts>::type typename std::decay<Ts>::type
>::type...> >::type...>
>::type, >::type>
HandleTag> request(const Handle& dest, const duration& timeout, Ts&&... xs) {
request(const typed_actor<Sigs...>& dest,
const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
using token = using token =
detail::type_list< detail::type_list<
typename detail::implicit_conversions< typename detail::implicit_conversions<
typename std::decay<Ts>::type typename std::decay<Ts>::type
>::type...>; >::type...>;
token tk; static_assert(actor_accepts_message(signatures_of<Handle>(), token{}),
check_typed_input(dest, tk); "receiver does not accept given message");
return {dptr()->request_impl(message_priority::normal, dest, auto req_id = dptr()->new_request_id(P);
timeout, std::forward<Ts>(xs)...), if (dest)
dptr()}; dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(),
std::forward<Ts>(xs)...);
dptr()->request_sync_timeout_msg(timeout, req_id);
return {req_id.response_id(), dptr()};
} }
private: private:
...@@ -137,14 +81,7 @@ private: ...@@ -137,14 +81,7 @@ private:
} }
}; };
template <class ResponseHandleTag>
class requester {
public:
template <class Base, class Subtype>
using impl = requester_impl<Base, Subtype, ResponseHandleTag>;
};
} // namespace mixin } // namespace mixin
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_SYNC_SENDER_HPP #endif // CAF_MIXIN_REQUESTER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_MIXIN_SENDER_HPP
#define CAF_MIXIN_SENDER_HPP
#include <tuple>
#include <chrono>
#include "caf/fwd.hpp"
#include "caf/actor.hpp"
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/response_handle.hpp"
#include "caf/message_priority.hpp"
#include "caf/check_typed_input.hpp"
namespace caf {
namespace mixin {
/// A `sender` is an actor that supports `self->send(...)`.
template <class Base, class Subtype>
class sender : public Base {
public:
template <class... Ts>
sender(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @returns A handle identifying a future-like handle to the response.
/// @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 `dest == invalid_actor`
template <message_priority P = message_priority::normal,
class Dest = actor, class... Ts>
void send(const Dest& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
static_assert(! statically_typed<Subtype>() || statically_typed<Dest>(),
"statically typed actors can only send() to other "
"statically typed actors; use anon_send() or request() when "
"communicating with dynamically typed actors");
static_assert(actor_accepts_message(signatures_of<Dest>(), token{}),
"receiver does not accept given message");
// TODO: this only checks one way, we should check for loops
static_assert(is_void_response(response_to(signatures_of<Dest>(),
token{}))
|| actor_accepts_message(signatures_of<Subtype>(),
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return;
dest->eq_impl(message_id::make(P), this->ctrl(),
this->context(), std::forward<Ts>(xs)...);
}
template <message_priority P = message_priority::normal,
class Dest = actor, class... Ts>
void delayed_send(const Dest& dest, const duration& rtime, Ts&&... xs) {
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
static_assert(! statically_typed<Subtype>() || statically_typed<Dest>(),
"statically typed actors are only allowed to send() to other "
"statically typed actors; use anon_send() or request() when "
"communicating with dynamically typed actors");
static_assert(actor_accepts_message(signatures_of<Dest>(), token{}),
"receiver does not accept given message");
// TODO: this only checks one way, we should check for loops
static_assert(is_void_response(response_to(signatures_of<Dest>(),
token{}))
|| actor_accepts_message(signatures_of<Subtype>(),
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return;
this->system().scheduler().delayed_send(
rtime, this->ctrl(), actor_cast<strong_actor_ptr>(dest),
message_id::make(P), make_message(std::forward<Ts>(xs)...));
}
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_SENDER_HPP
...@@ -46,14 +46,15 @@ struct blocking_response_handle_tag {}; ...@@ -46,14 +46,15 @@ struct blocking_response_handle_tag {};
/// This helper class identifies an expected response message /// This helper class identifies an expected response message
/// and enables `request(...).then(...)`. /// and enables `request(...).then(...)`.
template <class Self, class Output, class Tag> template <class Self, class Output,
bool IsBlocking = std::is_same<Self, blocking_actor>::value>
class response_handle; class response_handle;
/****************************************************************************** /******************************************************************************
* nonblocking * * nonblocking *
******************************************************************************/ ******************************************************************************/
template <class Self, class Output> template <class Self, class Output>
class response_handle<Self, Output, nonblocking_response_handle_tag> { class response_handle<Self, Output, false> {
public: public:
response_handle() = delete; response_handle() = delete;
response_handle(const response_handle&) = default; response_handle(const response_handle&) = default;
...@@ -146,7 +147,7 @@ private: ...@@ -146,7 +147,7 @@ private:
* blocking * * blocking *
******************************************************************************/ ******************************************************************************/
template <class Self, class Output> template <class Self, class Output>
class response_handle<Self, Output, blocking_response_handle_tag> { class response_handle<Self, Output, true> {
public: public:
response_handle() = delete; response_handle() = delete;
response_handle(const response_handle&) = default; response_handle(const response_handle&) = default;
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#ifndef CAF_SCOPED_ACTOR_HPP #ifndef CAF_SCOPED_ACTOR_HPP
#define CAF_SCOPED_ACTOR_HPP #define CAF_SCOPED_ACTOR_HPP
#include "caf/none.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_storage.hpp" #include "caf/actor_storage.hpp"
...@@ -36,6 +37,8 @@ public: ...@@ -36,6 +37,8 @@ public:
template <class, class, int> template <class, class, int>
friend class actor_cast_access; friend class actor_cast_access;
using signatures = none_t;
// tell actor_cast which semantic this type uses // tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = false; static constexpr bool has_weak_ptr_semantics = false;
...@@ -62,6 +65,10 @@ public: ...@@ -62,6 +65,10 @@ public:
blocking_actor* ptr() const; blocking_actor* ptr() const;
constexpr explicit operator bool() const {
return true;
}
private: private:
inline actor_control_block* get() const { inline actor_control_block* get() const {
......
...@@ -34,53 +34,48 @@ ...@@ -34,53 +34,48 @@
namespace caf { namespace caf {
/// Sends `to` a message under the identity of `from` with priority `prio`. /// Sends `to` a message under the identity of `from` with priority `prio`.
template <class Dest, class... Ts> template <message_priority P = message_priority::normal,
typename std::enable_if<is_message_sink<Dest>::value>::type class Source = actor, class Dest = actor, class... Ts>
send_as(actor from, message_priority prio, const Dest& to, Ts&&... xs) { void send_as(const Source& src, const Dest& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token = using token =
detail::type_list< detail::type_list<
typename detail::implicit_conversions< typename detail::implicit_conversions<
typename std::decay<Ts>::type typename std::decay<Ts>::type
>::type...>; >::type...>;
token tk; static_assert(! statically_typed<Source>() || statically_typed<Dest>(),
check_typed_input(to, tk); "statically typed actors can only send() to other "
if (! to) "statically typed actors; use anon_send() or request() when "
"communicating with dynamically typed actors");
static_assert(actor_accepts_message(signatures_of<Dest>(), token{}),
"receiver does not accept given message");
// TODO: this only checks one way, we should check for loops
static_assert(is_void_response(response_to(signatures_of<Dest>(),
token{}))
|| actor_accepts_message(signatures_of<Source>(),
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return; return;
message_id mid; dest->eq_impl(message_id::make(P), actor_cast<strong_actor_ptr>(src),
to->enqueue(actor_cast<strong_actor_ptr>(std::move(from)), nullptr, std::forward<Ts>(xs)...);
prio == message_priority::high ? mid.with_high_priority() : mid,
make_message(std::forward<Ts>(xs)...), nullptr);
} }
/// Sends `to` a message under the identity of `from`. /// Anonymously sends `dest` a message.
template <class Dest, class... Ts> template <message_priority P = message_priority::normal,
typename std::enable_if<is_message_sink<Dest>::value>::type class Dest = actor, class... Ts>
send_as(actor from, const Dest& to, Ts&&... xs) { void anon_send(const Dest& dest, Ts&&... xs) {
send_as(std::move(from), message_priority::normal, send_as<P>(actor{}, dest, std::forward<Ts>(xs)...);
to, std::forward<Ts>(xs)...);
} }
/// Anonymously sends `to` a message with priority `prio`. /// Anonymously sends `dest` an exit message.
template <class Dest, class... Ts> template <class Dest>
typename std::enable_if<is_message_sink<Dest>::value>::type void anon_send_exit(const Dest& dest, exit_reason reason) {
anon_send(message_priority prio, const Dest& to, Ts&&... xs) { if (! dest)
send_as(invalid_actor, prio, to, std::forward<Ts>(xs)...);
}
/// Anonymously sends `to` a message.
template <class Dest, class... Ts>
typename std::enable_if<is_message_sink<Dest>::value>::type
anon_send(const Dest& to, Ts&&... xs) {
send_as(invalid_actor, message_priority::normal, to, std::forward<Ts>(xs)...);
}
/// Anonymously sends `to` an exit message.
template <class ActorHandle>
void anon_send_exit(const ActorHandle& to, exit_reason reason) {
if (! to)
return; return;
to->enqueue(nullptr, message_id{}.with_high_priority(), dest->enqueue(nullptr, message_id::make(),
make_message(exit_msg{invalid_actor_addr, reason}), nullptr); make_message(exit_msg{invalid_actor_addr, reason}), nullptr);
} }
/// Anonymously sends `to` an exit message. /// Anonymously sends `to` an exit message.
......
...@@ -44,7 +44,7 @@ struct exit_msg { ...@@ -44,7 +44,7 @@ struct exit_msg {
}; };
inline std::string to_string(const exit_msg& x) { inline std::string to_string(const exit_msg& x) {
return "exit" + deep_to_string(std::tie(x.source, x.reason)); return "exit_msg" + deep_to_string(std::tie(x.source, x.reason));
} }
template <class Processor> template <class Processor>
...@@ -62,7 +62,7 @@ struct down_msg { ...@@ -62,7 +62,7 @@ struct down_msg {
}; };
inline std::string to_string(const down_msg& x) { inline std::string to_string(const down_msg& x) {
return "down" + deep_to_string(std::tie(x.source, x.reason)); return "down_msg" + deep_to_string(std::tie(x.source, x.reason));
} }
template <class Processor> template <class Processor>
......
...@@ -22,10 +22,17 @@ ...@@ -22,10 +22,17 @@
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
namespace caf { namespace caf {
struct typed_actor_view_base { };
template <class... Sigs> template <class... Sigs>
class typed_actor_view { class typed_actor_view : public extend<typed_actor_view_base,
typed_actor_view<Sigs...>>::template
with<mixin::sender, mixin::requester> {
public: public:
typed_actor_view(local_actor* selfptr) : self_(selfptr) { typed_actor_view(local_actor* selfptr) : self_(selfptr) {
// nop // nop
...@@ -47,90 +54,6 @@ public: ...@@ -47,90 +54,6 @@ public:
return self_->spawn<Os>(std::move(fun), std::forward<Ts>(xs)...); return self_->spawn<Os>(std::move(fun), std::forward<Ts>(xs)...);
} }
/****************************************************************************
* send asynchronous messages *
****************************************************************************/
template <class... Ts>
void send(message_priority mp, const actor& dest, Ts&&... xs) {
self_->send(mp, dest, std::forward<Ts>(xs)...);
}
template <class... Ts>
void send(const actor& dest, Ts&&... xs) {
self_->send(dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void send(message_priority mp, const typed_actor<DestSigs...>& dest,
Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
self_->send(mp, dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void send(const typed_actor<DestSigs...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
self_->send(dest, std::forward<Ts>(xs)...);
}
template <class... Ts>
void delayed_send(message_priority mp, const actor& dest,
const duration& rtime, Ts&&... xs) {
self_->delayed_send(mp, dest, rtime, std::forward<Ts>(xs)...);
}
template <class... Ts>
void delayed_send(const actor& dest, const duration& rtime, Ts&&... xs) {
self_->delayed_send(dest, rtime, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(message_priority mp, const typed_actor<DestSigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
self_->delayed_send(mp, dest, rtime, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(const typed_actor<DestSigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
self_->delayed_send(dest, rtime, std::forward<Ts>(xs)...);
}
/**************************************************************************** /****************************************************************************
* miscellaneous actor operations * * miscellaneous actor operations *
****************************************************************************/ ****************************************************************************/
......
...@@ -25,23 +25,34 @@ ...@@ -25,23 +25,34 @@
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
namespace caf { namespace caf {
template <class... Sigs>
class behavior_type_of<typed_event_based_actor<Sigs...>> {
public:
using type = typed_behavior<Sigs...>;
};
/// A cooperatively scheduled, event-based actor /// A cooperatively scheduled, event-based actor
/// implementation with static type-checking. /// implementation with static type-checking.
/// @extends local_actor /// @extends local_actor
template <class... Sigs> template <class... Sigs>
class typed_event_based_actor class typed_event_based_actor : public extend<local_actor,
: public abstract_event_based_actor<typed_behavior<Sigs...>, true> { typed_event_based_actor<Sigs...>
>::template
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public statically_typed_actor_base {
public: public:
using base_type = using super = typename
abstract_event_based_actor<typed_behavior<Sigs...>, true>; extend<local_actor, typed_event_based_actor<Sigs...>>::template
with<mixin::sender, mixin::requester, mixin::behavior_changer>;
explicit typed_event_based_actor(actor_config& cfg) : base_type(cfg) { explicit typed_event_based_actor(actor_config& cfg) : super(cfg) {
// nop // nop
} }
...@@ -66,67 +77,6 @@ public: ...@@ -66,67 +77,6 @@ public:
} }
} }
using base_type::send;
using base_type::delayed_send;
template <class... DestSigs, class... Ts>
void send(message_priority mp, const typed_actor<DestSigs...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
base_type::send(mp, dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void send(const typed_actor<DestSigs...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
base_type::send(dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(message_priority mp, const typed_actor<DestSigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
base_type::delayed_send(mp, dest, rtime, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(const typed_actor<DestSigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
base_type::delayed_send(dest, rtime, std::forward<Ts>(xs)...);
}
protected: protected:
virtual behavior_type make_behavior() { virtual behavior_type make_behavior() {
if (this->initial_behavior_fac_) { if (this->initial_behavior_fac_) {
......
...@@ -35,7 +35,8 @@ namespace caf { ...@@ -35,7 +35,8 @@ namespace caf {
template <class T> template <class T>
class weak_intrusive_ptr : detail::comparable<weak_intrusive_ptr<T>>, class weak_intrusive_ptr : detail::comparable<weak_intrusive_ptr<T>>,
detail::comparable<weak_intrusive_ptr<T>, const T*>, detail::comparable<weak_intrusive_ptr<T>, const T*>,
detail::comparable<weak_intrusive_ptr<T>, std::nullptr_t> { detail::comparable<weak_intrusive_ptr<T>,
std::nullptr_t> {
public: public:
using pointer = T*; using pointer = T*;
using const_pointer = const T*; using const_pointer = const T*;
...@@ -170,13 +171,15 @@ private: ...@@ -170,13 +171,15 @@ private:
/// @relates weak_intrusive_ptr /// @relates weak_intrusive_ptr
template <class X, typename Y> template <class X, typename Y>
bool operator==(const weak_intrusive_ptr<X>& lhs, const weak_intrusive_ptr<Y>& rhs) { bool operator==(const weak_intrusive_ptr<X>& lhs,
const weak_intrusive_ptr<Y>& rhs) {
return lhs.get() == rhs.get(); return lhs.get() == rhs.get();
} }
/// @relates weak_intrusive_ptr /// @relates weak_intrusive_ptr
template <class X, typename Y> template <class X, typename Y>
bool operator!=(const weak_intrusive_ptr<X>& lhs, const weak_intrusive_ptr<Y>& rhs) { bool operator!=(const weak_intrusive_ptr<X>& lhs,
const weak_intrusive_ptr<Y>& rhs) {
return !(lhs == rhs); return !(lhs == rhs);
} }
......
...@@ -78,7 +78,7 @@ void safe_actor(serializer& sink, T& storage) { ...@@ -78,7 +78,7 @@ void safe_actor(serializer& sink, T& storage) {
nid = ptr->nid; nid = ptr->nid;
// register locally running actors to be able to deserialize them later // register locally running actors to be able to deserialize them later
if (nid == sys.node()) if (nid == sys.node())
sys.registry().put(aid, ptr); sys.registry().put(aid, actor_cast<strong_actor_ptr>(storage));
} }
sink << aid << nid; sink << aid << nid;
} }
......
...@@ -41,16 +41,16 @@ actor_ostream::actor_ostream(scoped_actor& self) ...@@ -41,16 +41,16 @@ actor_ostream::actor_ostream(scoped_actor& self)
} }
actor_ostream& actor_ostream::write(std::string arg) { actor_ostream& actor_ostream::write(std::string arg) {
printer_->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {}, printer_->enqueue(mailbox_element::make(nullptr, message_id::make(), {},
add_atom::value, self_, add_atom::value, self_,
std::move(arg)), std::move(arg)),
nullptr); nullptr);
return *this; return *this;
} }
actor_ostream& actor_ostream::flush() { actor_ostream& actor_ostream::flush() {
printer_->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {}, printer_->enqueue(mailbox_element::make(nullptr, message_id::make(), {},
flush_atom::value, self_), flush_atom::value, self_),
nullptr); nullptr);
return *this; return *this;
} }
...@@ -59,17 +59,17 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) { ...@@ -59,17 +59,17 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) {
if (! self) if (! self)
return; return;
auto pr = self->home_system().scheduler().printer(); auto pr = self->home_system().scheduler().printer();
pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {}, pr->enqueue(mailbox_element::make(nullptr, message_id::make(), {},
redirect_atom::value, self->id(), redirect_atom::value, self->id(),
std::move(fn), flags), std::move(fn), flags),
nullptr); nullptr);
} }
void actor_ostream::redirect_all(actor_system& sys, std::string fn, int flags) { void actor_ostream::redirect_all(actor_system& sys, std::string fn, int flags) {
auto pr = sys.scheduler().printer(); auto pr = sys.scheduler().printer();
pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {}, pr->enqueue(mailbox_element::make(nullptr, message_id::make(), {},
redirect_atom::value, redirect_atom::value,
std::move(fn), flags), std::move(fn), flags),
nullptr); nullptr);
} }
......
...@@ -123,15 +123,6 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers, ...@@ -123,15 +123,6 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
return res; return res;
} }
void actor_pool::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, sender, mid, content, eu))
return;
auto ptr = mailbox_element::make(sender, mid, {}, std::move(content));
policy_(home_system(), guard, workers_, ptr, eu);
}
void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) { void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_}; upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, what->sender, what->mid, what->msg, eu)) if (filter(guard, what->sender, what->mid, what->msg, eu))
...@@ -163,7 +154,6 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -163,7 +154,6 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
unique_guard.unlock(); unique_guard.unlock();
for (auto& w : workers) for (auto& w : workers)
anon_send(w, msg); anon_send(w, msg);
quit(eu);
is_registered(false); is_registered(false);
} }
return true; return true;
...@@ -226,8 +216,8 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -226,8 +216,8 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
void actor_pool::quit(execution_unit* host) { void actor_pool::quit(execution_unit* host) {
// we can safely run our cleanup code here without holding // we can safely run our cleanup code here without holding
// workers_mtx_ because abstract_actor has its own lock // workers_mtx_ because abstract_actor has its own lock
cleanup(planned_reason_, host); if (cleanup(planned_reason_, host))
is_registered(false); is_registered(false);
} }
} // namespace caf } // namespace caf
...@@ -68,29 +68,23 @@ strong_actor_ptr actor_registry::get(actor_id key) const { ...@@ -68,29 +68,23 @@ strong_actor_ptr actor_registry::get(actor_id key) const {
return nullptr; return nullptr;
} }
void actor_registry::put(actor_id key, actor_control_block* val) { void actor_registry::put(actor_id key, strong_actor_ptr val) {
CAF_LOG_TRACE(CAF_ARG(key));
if (! val) if (! val)
return; return;
auto strong_val = actor_cast<actor>(val);
if (! strong_val)
return;
{ // lifetime scope of guard { // lifetime scope of guard
exclusive_guard guard(instances_mtx_); exclusive_guard guard(instances_mtx_);
if (! entries_.emplace(key, actor_cast<strong_actor_ptr>(val)).second) if (! entries_.emplace(key, val).second)
return; return;
} }
// attach functor without lock // attach functor without lock
CAF_LOG_INFO("added actor:" << CAF_ARG(key)); CAF_LOG_INFO("added actor:" << CAF_ARG(key));
actor_registry* reg = this; actor_registry* reg = this;
strong_val->attach_functor([key, reg]() { val->get()->attach_functor([key, reg]() {
reg->erase(key); reg->erase(key);
}); });
} }
void actor_registry::put(actor_id key, const strong_actor_ptr& val) {
put(key, actor_cast<actor_control_block*>(val));
}
void actor_registry::erase(actor_id key) { void actor_registry::erase(actor_id key) {
exclusive_guard guard(instances_mtx_); exclusive_guard guard(instances_mtx_);
entries_.erase(key); entries_.erase(key);
...@@ -101,7 +95,7 @@ void actor_registry::inc_running() { ...@@ -101,7 +95,7 @@ void actor_registry::inc_running() {
auto value = ++running_; auto value = ++running_;
CAF_LOG_DEBUG(CAF_ARG(value)); CAF_LOG_DEBUG(CAF_ARG(value));
# else # else
++running_; ++running_;
# endif # endif
} }
...@@ -283,11 +277,9 @@ void actor_registry::stop() { ...@@ -283,11 +277,9 @@ void actor_registry::stop() {
// the scheduler is already stopped -> invoke exit messages manually // the scheduler is already stopped -> invoke exit messages manually
dropping_execution_unit dummy{&system_}; dropping_execution_unit dummy{&system_};
for (auto& kvp : named_entries_) { for (auto& kvp : named_entries_) {
auto mp = mailbox_element::make_joint(nullptr, auto mp = mailbox_element::make(nullptr,invalid_message_id, {},
invalid_message_id, exit_msg{invalid_actor_addr,
{}, exit_reason::kill});
exit_msg{invalid_actor_addr,
exit_reason::kill});
auto ptr = static_cast<local_actor*>(actor_cast<abstract_actor*>(kvp.second)); auto ptr = static_cast<local_actor*>(actor_cast<abstract_actor*>(kvp.second));
ptr->exec_single_event(&dummy, mp); ptr->exec_single_event(&dummy, mp);
} }
......
...@@ -41,7 +41,7 @@ void default_attachable::actor_exited(const error& rsn, execution_unit* host) { ...@@ -41,7 +41,7 @@ void default_attachable::actor_exited(const error& rsn, execution_unit* host) {
auto observer = actor_cast<strong_actor_ptr>(observer_); auto observer = actor_cast<strong_actor_ptr>(observer_);
auto observed = actor_cast<strong_actor_ptr>(observed_); auto observed = actor_cast<strong_actor_ptr>(observed_);
if (observer) if (observer)
observer->enqueue(std::move(observed), message_id{}.with_high_priority(), observer->enqueue(std::move(observed), message_id::make(),
factory(actor_cast<abstract_actor*>(observed_), rsn), factory(actor_cast<abstract_actor*>(observed_), rsn),
host); host);
} }
......
...@@ -42,7 +42,7 @@ namespace caf { ...@@ -42,7 +42,7 @@ namespace caf {
class local_actor::private_thread { class local_actor::private_thread {
public: public:
private_thread(local_actor* self) : self_(self) { private_thread(local_actor* self) : self_destroyed_(false), self_(self) {
intrusive_ptr_add_ref(self->ctrl()); intrusive_ptr_add_ref(self->ctrl());
scheduler::inc_detached_threads(); scheduler::inc_detached_threads();
} }
...@@ -54,6 +54,9 @@ public: ...@@ -54,6 +54,9 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_execution_unit ctx{&job->system()}; scoped_execution_unit ctx{&job->system()};
auto max_throughput = std::numeric_limits<size_t>::max(); auto max_throughput = std::numeric_limits<size_t>::max();
auto wait_pred = [&] {
return ! job->mailbox().empty();
};
for (;;) { for (;;) {
bool resume_later; bool resume_later;
do { do {
...@@ -74,7 +77,7 @@ public: ...@@ -74,7 +77,7 @@ public:
} while (resume_later); } while (resume_later);
// wait until actor becomes ready again or was destroyed // wait until actor becomes ready again or was destroyed
std::unique_lock<std::mutex> guard(mtx_); std::unique_lock<std::mutex> guard(mtx_);
cv_.wait(guard); cv_.wait(guard, wait_pred);
job = const_cast<local_actor*>(self_); job = const_cast<local_actor*>(self_);
if (! job) if (! job)
return; return;
...@@ -94,11 +97,31 @@ public: ...@@ -94,11 +97,31 @@ public:
static void exec(private_thread* this_ptr) { static void exec(private_thread* this_ptr) {
this_ptr->run(); this_ptr->run();
delete this_ptr;
scheduler::dec_detached_threads(); scheduler::dec_detached_threads();
// make sure to not destroy the private thread object before the
// detached actor is destroyed and this object is unreachable
this_ptr->await_self_destroyed();
delete this_ptr;
}
void notify_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
self_destroyed_ = true;
cv_.notify_one();
}
void await_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
while (! self_destroyed_)
cv_.wait(guard);
}
void start() {
std::thread{exec, this}.detach();
} }
private: private:
volatile bool self_destroyed_;
std::mutex mtx_; std::mutex mtx_;
std::condition_variable cv_; std::condition_variable cv_;
volatile local_actor* self_; volatile local_actor* self_;
...@@ -159,7 +182,11 @@ local_actor::local_actor(actor_config& cfg) ...@@ -159,7 +182,11 @@ local_actor::local_actor(actor_config& cfg)
} }
local_actor::~local_actor() { local_actor::~local_actor() {
// nop CAF_LOG_TRACE("");
// signalize to the private thread object that it is
// unrachable and can be destroyed as well
if (private_thread_)
private_thread_->notify_self_destroyed();
} }
void local_actor::destroy() { void local_actor::destroy() {
...@@ -180,7 +207,8 @@ void local_actor::intrusive_ptr_release_impl() { ...@@ -180,7 +207,8 @@ void local_actor::intrusive_ptr_release_impl() {
} }
void local_actor::monitor(abstract_actor* ptr) { void local_actor::monitor(abstract_actor* ptr) {
CAF_ASSERT(ptr != nullptr); if (! ptr)
return;
ptr->attach(default_attachable::make_monitor(ptr->address(), address())); ptr->attach(default_attachable::make_monitor(ptr->address(), address()));
} }
...@@ -227,13 +255,13 @@ uint32_t local_actor::request_timeout(const duration& d) { ...@@ -227,13 +255,13 @@ uint32_t local_actor::request_timeout(const duration& d) {
auto result = ++timeout_id_; auto result = ++timeout_id_;
auto msg = make_message(timeout_msg{++timeout_id_}); auto msg = make_message(timeout_msg{++timeout_id_});
CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_)); CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_));
if (d.is_zero()) { if (d.is_zero())
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
enqueue(ctrl(), invalid_message_id, enqueue(ctrl(), invalid_message_id, std::move(msg), context());
std::move(msg), context()); else
} else { system().scheduler().delayed_send(d, ctrl(),
delayed_send(actor_cast<actor>(this), d, std::move(msg)); actor_cast<strong_actor_ptr>(this),
} message_id::make(), std::move(msg));
return result; return result;
} }
...@@ -241,8 +269,10 @@ void local_actor::request_sync_timeout_msg(const duration& d, message_id mid) { ...@@ -241,8 +269,10 @@ void local_actor::request_sync_timeout_msg(const duration& d, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(d) << CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(d) << CAF_ARG(mid));
if (! d.valid()) if (! d.valid())
return; return;
delayed_send_impl(mid.response_id(), ctrl(), d, system().scheduler().delayed_send(d, ctrl(),
make_message(sec::request_timeout)); ctrl(),
mid.response_id(),
make_message(sec::request_timeout));
} }
void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) { void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) {
...@@ -286,16 +316,14 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& node) { ...@@ -286,16 +316,14 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& node) {
if (what == "info") { if (what == "info") {
CAF_LOG_DEBUG("reply to 'info' message"); CAF_LOG_DEBUG("reply to 'info' message");
node.sender->enqueue( node.sender->enqueue(
mailbox_element::make_joint(ctrl(), mailbox_element::make(ctrl(), node.mid.response_id(),
node.mid.response_id(), {}, ok_atom::value, std::move(what),
{}, ok_atom::value, std::move(what), address(), name()),
address(), name()),
context()); context());
} else { } else {
node.sender->enqueue( node.sender->enqueue(
mailbox_element::make_joint(ctrl(), mailbox_element::make(ctrl(), node.mid.response_id(),
node.mid.response_id(), {}, sec::unsupported_sys_key),
{}, sec::unsupported_sys_key),
context()); context());
} }
return msg_type::sys_message; return msg_type::sys_message;
...@@ -385,19 +413,16 @@ public: ...@@ -385,19 +413,16 @@ public:
void operator()(error& x) override { void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("report error back to requesting actor");
delegate(x); delegate(x);
} }
void operator()(message& x) override { void operator()(message& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("respond via response_promise");
delegate(x); delegate(x);
} }
void operator()(const none_t& x) override { void operator()(const none_t& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("message handler returned none");
delegate(x); delegate(x);
} }
...@@ -636,6 +661,7 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -636,6 +661,7 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
std::exception_ptr eptr = nullptr; std::exception_ptr eptr = nullptr;
try { try {
self->act(); self->act();
rsn = self->fail_state_;
} }
catch (actor_exited& e) { catch (actor_exited& e) {
rsn = e.reason(); rsn = e.reason();
...@@ -660,7 +686,7 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -660,7 +686,7 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
return; return;
} }
private_thread_ = new private_thread(this); private_thread_ = new private_thread(this);
std::thread(private_thread::exec, private_thread_).detach(); private_thread_->start();
return; return;
} }
CAF_ASSERT(eu != nullptr); CAF_ASSERT(eu != nullptr);
...@@ -803,6 +829,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -803,6 +829,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
std::pair<resumable::resume_result, invoke_message_result> std::pair<resumable::resume_result, invoke_message_result>
local_actor::exec_event(mailbox_element_ptr& ptr) { local_actor::exec_event(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_ARG(*ptr));
behavior empty_bhvr; behavior empty_bhvr;
auto& bhvr = auto& bhvr =
awaits_response() ? awaited_response_handler() awaits_response() ? awaited_response_handler()
...@@ -810,6 +837,7 @@ local_actor::exec_event(mailbox_element_ptr& ptr) { ...@@ -810,6 +837,7 @@ local_actor::exec_event(mailbox_element_ptr& ptr) {
: bhvr_stack().back(); : bhvr_stack().back();
auto mid = awaited_response_id(); auto mid = awaited_response_id();
auto res = invoke_message(ptr, bhvr, mid); auto res = invoke_message(ptr, bhvr, mid);
CAF_LOG_DEBUG(CAF_ARG(mid) << CAF_ARG(res));
switch (res) { switch (res) {
case im_success: case im_success:
bhvr_stack().cleanup(); bhvr_stack().cleanup();
...@@ -877,9 +905,8 @@ void local_actor::exec_single_event(execution_unit* ctx, ...@@ -877,9 +905,8 @@ void local_actor::exec_single_event(execution_unit* ctx,
} }
mailbox_element_ptr local_actor::next_message() { mailbox_element_ptr local_actor::next_message() {
if (! is_priority_aware()) { if (! is_priority_aware())
return mailbox_element_ptr{mailbox().try_pop()}; return mailbox_element_ptr{mailbox().try_pop()};
}
// we partition the mailbox into four segments in this case: // we partition the mailbox into four segments in this case:
// <-------- ! was_skipped --------> | <-------- was_skipped --------> // <-------- ! was_skipped --------> | <-------- was_skipped -------->
// <-- high prio --><-- low prio -->|<-- high prio --><-- low prio --> // <-- high prio --><-- low prio -->|<-- high prio --><-- low prio -->
...@@ -901,9 +928,8 @@ mailbox_element_ptr local_actor::next_message() { ...@@ -901,9 +928,8 @@ mailbox_element_ptr local_actor::next_message() {
} }
} }
mailbox_element_ptr result; mailbox_element_ptr result;
if (! cache.first_empty()) { if (! cache.first_empty())
result.reset(cache.take_first_front()); result.reset(cache.take_first_front());
}
return result; return result;
} }
...@@ -958,22 +984,12 @@ void local_actor::do_become(behavior bhvr, bool discard_old) { ...@@ -958,22 +984,12 @@ void local_actor::do_become(behavior bhvr, bool discard_old) {
bhvr_stack_.push_back(std::move(bhvr)); bhvr_stack_.push_back(std::move(bhvr));
} }
void local_actor::send_impl(message_id mid, abstract_channel* dest,
message what) const {
if (! dest)
return;
dest->enqueue(ctrl(), mid, std::move(what), context());
}
void local_actor::send_exit(const actor_addr& whom, error reason) { void local_actor::send_exit(const actor_addr& whom, error reason) {
send(message_priority::high, actor_cast<actor>(whom), auto ptr = actor_cast<actor>(whom);
exit_msg{address(), reason}); if (! ptr)
} return;
ptr->eq_impl(message_id::make(), nullptr, context(),
void local_actor::delayed_send_impl(message_id mid, strong_actor_ptr dest, exit_msg{address(), std::move(reason)});
const duration& rel_time, message msg) {
system().scheduler().delayed_send(rel_time, ctrl(), std::move(dest),
mid, std::move(msg));
} }
const char* local_actor::name() const { const char* local_actor::name() const {
......
...@@ -85,8 +85,8 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) { ...@@ -85,8 +85,8 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) {
// tell printer to purge its state for us if we ever used aout() // tell printer to purge its state for us if we ever used aout()
if (get_flag(abstract_actor::has_used_aout_flag)) { if (get_flag(abstract_actor::has_used_aout_flag)) {
auto pr = home_system().scheduler().printer(); auto pr = home_system().scheduler().printer();
pr->enqueue(mailbox_element::make_joint(nullptr, message_id::make(), {}, pr->enqueue(mailbox_element::make(nullptr, message_id::make(), {},
delete_atom::value, id()), delete_atom::value, id()),
nullptr); nullptr);
} }
return true; return true;
...@@ -259,17 +259,16 @@ bool monitorable_actor::handle_system_message(mailbox_element& node, ...@@ -259,17 +259,16 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
err = sec::unsupported_sys_key; err = sec::unsupported_sys_key;
return; return;
} }
res = mailbox_element::make_joint(ctrl(), res = mailbox_element::make(ctrl(), node.mid.response_id(), {},
node.mid.response_id(), {}, ok_atom::value, std::move(what),
ok_atom::value, std::move(what), address(), name());
address(), name());
} }
}); });
if (! res && ! err) if (! res && ! err)
err = sec::unsupported_sys_message; err = sec::unsupported_sys_message;
if (err && node.mid.is_request()) if (err && node.mid.is_request())
res = mailbox_element::make_joint(ctrl(), node.mid.response_id(), res = mailbox_element::make(ctrl(), node.mid.response_id(),
{}, std::move(err)); {}, std::move(err));
if (res) { if (res) {
auto s = actor_cast<actor>(node.sender); auto s = actor_cast<actor>(node.sender);
if (s) if (s)
......
...@@ -79,16 +79,15 @@ CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture) ...@@ -79,16 +79,15 @@ CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture)
CAF_TEST(round_robin_actor_pool) { CAF_TEST(round_robin_actor_pool) {
scoped_actor self{system}; scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_worker, actor_pool::round_robin()); auto pool = actor_pool::make(&context, 5, spawn_worker,
self->monitor(w); actor_pool::round_robin());
self->send(w, sys_atom::value, put_atom::value, spawn_worker()); self->send(pool, sys_atom::value, put_atom::value, spawn_worker());
std::vector<actor> workers; std::vector<actor> workers;
for (int i = 0; i < 6; ++i) { for (int i = 0; i < 6; ++i) {
self->request(w, infinite, i, i).receive( self->request(pool, infinite, i, i).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, i + i); CAF_CHECK_EQUAL(res, i + i);
auto sender = self->current_sender(); auto sender = self->current_sender();
self->monitor(sender);
workers.push_back(actor_cast<actor>(sender)); workers.push_back(actor_cast<actor>(sender));
} }
); );
...@@ -99,7 +98,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -99,7 +98,7 @@ CAF_TEST(round_robin_actor_pool) {
return x == invalid_actor; return x == invalid_actor;
}; };
CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid)); CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid));
self->request(w, infinite, sys_atom::value, get_atom::value).receive( self->request(pool, infinite, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) { [&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end()); std::sort(workers.begin(), workers.end());
std::sort(ws.begin(), ws.end()); std::sort(ws.begin(), ws.end());
...@@ -114,7 +113,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -114,7 +113,7 @@ CAF_TEST(round_robin_actor_pool) {
bool success = false; bool success = false;
size_t i = 0; size_t i = 0;
while (! success && ++i <= 10) { while (! success && ++i <= 10) {
self->request(w, infinite, sys_atom::value, get_atom::value).receive( self->request(pool, infinite, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) { [&](std::vector<actor>& ws) {
success = workers.size() == ws.size(); success = workers.size() == ws.size();
if (success) { if (success) {
...@@ -129,7 +128,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -129,7 +128,7 @@ CAF_TEST(round_robin_actor_pool) {
} }
CAF_REQUIRE(success); CAF_REQUIRE(success);
CAF_MESSAGE("about to send exit to workers"); CAF_MESSAGE("about to send exit to workers");
self->send_exit(w, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
self->wait_for(workers); self->wait_for(workers);
} }
...@@ -139,8 +138,8 @@ CAF_TEST(broadcast_actor_pool) { ...@@ -139,8 +138,8 @@ CAF_TEST(broadcast_actor_pool) {
return actor_pool::make(&context, 5, fixture::spawn_worker, return actor_pool::make(&context, 5, fixture::spawn_worker,
actor_pool::broadcast()); actor_pool::broadcast());
}; };
auto w = actor_pool::make(&context, 5, spawn5, actor_pool::broadcast()); auto pool = actor_pool::make(&context, 5, spawn5, actor_pool::broadcast());
self->send(w, 1, 2); self->send(pool, 1, 2);
std::vector<int> results; std::vector<int> results;
int i = 0; int i = 0;
self->receive_for(i, 25)( self->receive_for(i, 25)(
...@@ -154,20 +153,20 @@ CAF_TEST(broadcast_actor_pool) { ...@@ -154,20 +153,20 @@ CAF_TEST(broadcast_actor_pool) {
CAF_CHECK_EQUAL(results.size(), 25u); CAF_CHECK_EQUAL(results.size(), 25u);
CAF_CHECK(std::all_of(results.begin(), results.end(), CAF_CHECK(std::all_of(results.begin(), results.end(),
[](int res) { return res == 3; })); [](int res) { return res == 3; }));
self->send_exit(w, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
} }
CAF_TEST(random_actor_pool) { CAF_TEST(random_actor_pool) {
scoped_actor self{system}; scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_worker, actor_pool::random()); auto pool = actor_pool::make(&context, 5, spawn_worker, actor_pool::random());
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 5; ++i) {
self->request(w, std::chrono::milliseconds(250), 1, 2).receive( self->request(pool, std::chrono::milliseconds(250), 1, 2).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 3); CAF_CHECK_EQUAL(res, 3);
} }
); );
} }
self->send_exit(w, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
} }
CAF_TEST(split_join_actor_pool) { CAF_TEST(split_join_actor_pool) {
...@@ -191,19 +190,19 @@ CAF_TEST(split_join_actor_pool) { ...@@ -191,19 +190,19 @@ CAF_TEST(split_join_actor_pool) {
}); });
}; };
scoped_actor self{system}; scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_split_worker, auto pool = actor_pool::make(&context, 5, spawn_split_worker,
actor_pool::split_join<int>(join_fun, split_fun)); actor_pool::split_join<int>(join_fun, split_fun));
self->request(w, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive( self->request(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 15); CAF_CHECK_EQUAL(res, 15);
} }
); );
self->request(w, infinite, std::vector<int>{6, 7, 8, 9, 10}).receive( self->request(pool, infinite, std::vector<int>{6, 7, 8, 9, 10}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 40); CAF_CHECK_EQUAL(res, 40);
} }
); );
self->send_exit(w, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -242,7 +242,7 @@ public: ...@@ -242,7 +242,7 @@ public:
behavior high_priority_testee(event_based_actor* self) { behavior high_priority_testee(event_based_actor* self) {
self->send(self, b_atom::value); self->send(self, b_atom::value);
self->send(message_priority::high, self, a_atom::value); self->send<message_priority::high>(self, a_atom::value);
// 'a' must be self->received before 'b' // 'a' must be self->received before 'b'
return { return {
[=](b_atom) { [=](b_atom) {
......
...@@ -43,6 +43,9 @@ using foo_promise = typed_response_promise<int>; ...@@ -43,6 +43,9 @@ using foo_promise = typed_response_promise<int>;
using foo2_promise = typed_response_promise<int, int>; using foo2_promise = typed_response_promise<int, int>;
using foo3_promise = typed_response_promise<double>; using foo3_promise = typed_response_promise<double>;
using get1_helper = typed_actor<replies_to<int, int>::with<put_atom, int, int>>;
using get2_helper = typed_actor<replies_to<int, int, int>::with<put_atom, int, int, int>>;
class foo_actor_impl : public foo_actor::base { class foo_actor_impl : public foo_actor::base {
public: public:
foo_actor_impl(actor_config& cfg) : foo_actor::base(cfg) { foo_actor_impl(actor_config& cfg) : foo_actor::base(cfg) {
...@@ -57,11 +60,10 @@ public: ...@@ -57,11 +60,10 @@ public:
return resp.deliver(x * 4); // has no effect return resp.deliver(x * 4); // has no effect
}, },
[=](get_atom, int x) -> foo_promise { [=](get_atom, int x) -> foo_promise {
auto calculator = spawn([](event_based_actor* self) -> behavior { auto calculator = spawn([]() -> get1_helper::behavior_type {
return { return {
[self](int promise_id, int value) -> message { [](int promise_id, int value) -> result<put_atom, int, int> {
self->quit(); return {put_atom::value, promise_id, value * 2};
return make_message(put_atom::value, promise_id, value * 2);
} }
}; };
}); });
...@@ -71,11 +73,10 @@ public: ...@@ -71,11 +73,10 @@ public:
return entry; return entry;
}, },
[=](get_atom, int x, int y) -> foo2_promise { [=](get_atom, int x, int y) -> foo2_promise {
auto calculator = spawn([](event_based_actor* self) -> behavior { auto calculator = spawn([]() -> get2_helper::behavior_type {
return { return {
[self](int promise_id, int v0, int v1) -> message { [](int promise_id, int v0, int v1) -> result<put_atom, int, int, int> {
self->quit(); return {put_atom::value, promise_id, v0 * 2, v1 * 2};
return make_message(put_atom::value, promise_id, v0 * 2, v1 * 2);
} }
}; };
}); });
......
...@@ -105,7 +105,7 @@ class typed_server3 : public server_type::base { ...@@ -105,7 +105,7 @@ class typed_server3 : public server_type::base {
public: public:
typed_server3(actor_config& cfg, const string& line, actor buddy) typed_server3(actor_config& cfg, const string& line, actor buddy)
: server_type::base(cfg) { : server_type::base(cfg) {
send(buddy, line); anon_send(buddy, line);
} }
behavior_type make_behavior() override { behavior_type make_behavior() override {
......
...@@ -23,25 +23,41 @@ ...@@ -23,25 +23,41 @@
#include <map> #include <map>
#include <vector> #include <vector>
#include "caf/fwd.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp" #include "caf/io/doorman.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
namespace caf { namespace caf {
template <>
class behavior_type_of<io::broker> {
public:
using type = behavior;
};
namespace io { namespace io {
/// Describes a dynamically typed broker. /// Describes a dynamically typed broker.
/// @extends abstract_broker /// @extends abstract_broker
/// @ingroup Broker /// @ingroup Broker
class broker : public abstract_event_based_actor<behavior, false, class broker : public extend<abstract_broker, broker>::
abstract_broker> { with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base {
public: public:
using super = abstract_event_based_actor<behavior, false, abstract_broker>; using super = extend<abstract_broker, broker>::
with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t;
template <class F, class... Ts> template <class F, class... Ts>
typename infer_handle_from_fun<F>::type typename infer_handle_from_fun<F>::type
......
...@@ -68,8 +68,8 @@ protected: ...@@ -68,8 +68,8 @@ protected:
void reset_mailbox_element() { void reset_mailbox_element() {
SysMsgType tmp; SysMsgType tmp;
set_hdl(tmp, hdl_); set_hdl(tmp, hdl_);
mailbox_elem_ptr_ = mailbox_element::make_joint(nullptr, invalid_message_id, mailbox_elem_ptr_ = mailbox_element::make(nullptr, invalid_message_id,
{}, tmp); {}, tmp);
} }
Handle hdl_; Handle hdl_;
......
...@@ -31,6 +31,9 @@ class basp_broker; ...@@ -31,6 +31,9 @@ class basp_broker;
class receive_policy; class receive_policy;
class abstract_broker; class abstract_broker;
template <class... Sigs>
class typed_broker;
namespace network { namespace network {
class multiplexer; class multiplexer;
......
...@@ -36,14 +36,23 @@ ...@@ -36,14 +36,23 @@
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
namespace caf { namespace caf {
namespace io {
template <class... Sigs> template <class... Sigs>
class typed_broker; class behavior_type_of<io::typed_broker<Sigs...>> {
public:
using type = typed_behavior<Sigs...>;
};
namespace io {
/// Denotes a minimal "client" broker managing one or more connection /// Denotes a minimal "client" broker managing one or more connection
/// handles by reacting to `new_data_msg` and `connection_closed_msg`. /// handles by reacting to `new_data_msg` and `connection_closed_msg`.
...@@ -63,8 +72,11 @@ using accept_handler = typed_actor<reacts_to<new_connection_msg>, ...@@ -63,8 +72,11 @@ using accept_handler = typed_actor<reacts_to<new_connection_msg>,
/// components in the network. /// components in the network.
/// @extends local_actor /// @extends local_actor
template <class... Sigs> template <class... Sigs>
class typed_broker : public abstract_event_based_actor<typed_behavior<Sigs...>, class typed_broker : public extend<abstract_broker,
false, abstract_broker> { typed_broker<Sigs...>>::template
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public statically_typed_actor_base {
public: public:
using signatures = detail::type_list<Sigs...>; using signatures = detail::type_list<Sigs...>;
...@@ -72,69 +84,8 @@ public: ...@@ -72,69 +84,8 @@ public:
using behavior_type = typed_behavior<Sigs...>; using behavior_type = typed_behavior<Sigs...>;
using super = abstract_event_based_actor<behavior_type, false, using super = typename extend<abstract_broker, typed_broker<Sigs...>>::
abstract_broker>; template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using super::send;
using super::delayed_send;
template <class... Dest, class... Ts>
void send(message_priority mp, const typed_actor<Dest...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<Dest...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::send(mp, dest, std::forward<Ts>(xs)...);
}
template <class... Dest, class... Ts>
void send(const typed_actor<Dest...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<Dest...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::send(dest, std::forward<Ts>(xs)...);
}
template <class... Dest, class... Ts>
void delayed_send(message_priority mp, const typed_actor<Dest...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<Dest...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::delayed_send(mp, dest, rtime, std::forward<Ts>(xs)...);
}
template <class... Dest, class... Ts>
void delayed_send(const typed_actor<Dest...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<Dest...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::delayed_send(dest, rtime, std::forward<Ts>(xs)...);
}
/// @cond PRIVATE /// @cond PRIVATE
std::set<std::string> message_types() const override { std::set<std::string> message_types() const override {
......
...@@ -223,7 +223,7 @@ void basp_broker_state::deliver(const node_id& src_nid, ...@@ -223,7 +223,7 @@ void basp_broker_state::deliver(const node_id& src_nid,
<< CAF_ARG(dest_aid) << CAF_ARG(msg) << CAF_ARG(mid)); << CAF_ARG(dest_aid) << CAF_ARG(msg) << CAF_ARG(mid));
strong_actor_ptr src; strong_actor_ptr src;
if (src_nid == this_node()) if (src_nid == this_node())
src = actor_cast<strong_actor_ptr>(system().registry().get(src_aid)); src = system().registry().get(src_aid);
else else
src = proxies().get_or_put(src_nid, src_aid); src = proxies().get_or_put(src_nid, src_aid);
strong_actor_ptr dest; strong_actor_ptr dest;
...@@ -231,6 +231,7 @@ void basp_broker_state::deliver(const node_id& src_nid, ...@@ -231,6 +231,7 @@ void basp_broker_state::deliver(const node_id& src_nid,
if (dest_nid != this_node()) { if (dest_nid != this_node()) {
dest = proxies().get_or_put(dest_nid, dest_aid); dest = proxies().get_or_put(dest_nid, dest_aid);
} else { } else {
// TODO: replace hack with clean solution
if (dest_aid == std::numeric_limits<actor_id>::max()) { if (dest_aid == std::numeric_limits<actor_id>::max()) {
// this hack allows CAF to talk to older CAF versions; CAF <= 0.14 will // this hack allows CAF to talk to older CAF versions; CAF <= 0.14 will
// discard this message silently as the receiver is not found, while // discard this message silently as the receiver is not found, while
...@@ -238,7 +239,7 @@ void basp_broker_state::deliver(const node_id& src_nid, ...@@ -238,7 +239,7 @@ void basp_broker_state::deliver(const node_id& src_nid,
auto dest_name = static_cast<atom_value>(mid.integer_value()); auto dest_name = static_cast<atom_value>(mid.integer_value());
CAF_LOG_DEBUG(CAF_ARG(dest_name)); CAF_LOG_DEBUG(CAF_ARG(dest_name));
mid = message_id::make(); // override this since the message is async mid = message_id::make(); // override this since the message is async
dest = actor_cast<strong_actor_ptr>(system().registry().get(dest_name)); dest = system().registry().get(dest_name);
} else { } else {
dest = system().registry().get(dest_aid); dest = system().registry().get(dest_aid);
} }
......
...@@ -48,7 +48,7 @@ instance::instance(abstract_broker* parent, callee& lstnr) ...@@ -48,7 +48,7 @@ instance::instance(abstract_broker* parent, callee& lstnr)
connection_state instance::handle(execution_unit* ctx, connection_state instance::handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, new_data_msg& dm, header& hdr,
bool is_payload) { bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(hdr) << CAF_ARG(is_payload)); CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors // function object providing cleanup code on errors
auto err = [&]() -> connection_state { auto err = [&]() -> connection_state {
auto cb = make_callback([&](const node_id& nid){ auto cb = make_callback([&](const node_id& nid){
...@@ -67,18 +67,19 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -67,18 +67,19 @@ connection_state instance::handle(execution_unit* ctx,
} else { } else {
binary_deserializer bd{ctx, dm.buf}; binary_deserializer bd{ctx, dm.buf};
bd >> hdr; bd >> hdr;
CAF_LOG_DEBUG(CAF_ARG(hdr));
if (! valid(hdr)) { if (! valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr.operation)); CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
return err(); return err();
} }
if (hdr.payload_len > 0) if (hdr.payload_len > 0) {
CAF_LOG_DEBUG("await payload before processing further");
return await_payload; return await_payload;
}
} }
CAF_LOG_DEBUG(CAF_ARG(hdr));
// needs forwarding? // needs forwarding?
if (! is_handshake(hdr) if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
&& ! is_heartbeat(hdr) CAF_LOG_DEBUG("forward message");
&& hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node); auto path = lookup(hdr.dest_node);
if (path) { if (path) {
binary_serializer bs{ctx, path->wr_buf}; binary_serializer bs{ctx, path->wr_buf};
...@@ -175,6 +176,7 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -175,6 +176,7 @@ connection_state instance::handle(execution_unit* ctx,
std::vector<strong_actor_ptr> forwarding_stack; std::vector<strong_actor_ptr> forwarding_stack;
message msg; message msg;
bd >> forwarding_stack >> msg; bd >> forwarding_stack >> msg;
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
callee_.deliver(hdr.source_node, hdr.source_actor, callee_.deliver(hdr.source_node, hdr.source_actor,
hdr.dest_node, hdr.dest_actor, hdr.dest_node, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data), message_id::from_integer_value(hdr.operation_data),
...@@ -328,11 +330,11 @@ void instance::write(execution_unit* ctx, ...@@ -328,11 +330,11 @@ void instance::write(execution_unit* ctx,
actor_id source_actor, actor_id source_actor,
actor_id dest_actor, actor_id dest_actor,
payload_writer* pw) { payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(operation) CAF_LOG_TRACE(CAF_ARG(operation) << CAF_ARG(operation_data)
<< CAF_ARG(payload_len) << CAF_ARG(operation_data)
<< CAF_ARG(source_node) << CAF_ARG(dest_node) << CAF_ARG(source_node) << CAF_ARG(dest_node)
<< CAF_ARG(source_actor) << CAF_ARG(dest_actor)); << CAF_ARG(source_actor) << CAF_ARG(dest_actor));
if (!pw) { if (! pw) {
CAF_LOG_DEBUG("send message without payload");
uint32_t zero = 0; uint32_t zero = 0;
binary_serializer bs{ctx, buf}; binary_serializer bs{ctx, buf};
bs << source_node bs << source_node
...@@ -342,29 +344,34 @@ void instance::write(execution_unit* ctx, ...@@ -342,29 +344,34 @@ void instance::write(execution_unit* ctx,
<< zero << zero
<< operation << operation
<< operation_data; << operation_data;
} else { return;
// reserve space in the buffer to write the payload later on }
auto wr_pos = buf.size(); // reserve space in the buffer to write the payload later on
char placeholder[basp::header_size]; auto wr_pos = buf.size();
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder)); char placeholder[basp::header_size];
auto pl_pos = buf.size(); buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
{ // lifetime scope of first serializer (write payload) auto pl_pos = buf.size();
binary_serializer bs{ctx, buf}; try { // lifetime scope of first serializer (write payload)
(*pw)(bs); binary_serializer bs{ctx, buf};
} (*pw)(bs);
// write broker message to the reserved space }
stream_serializer<charbuf> bs2{ctx, buf.data() + wr_pos, pl_pos - wr_pos}; catch (std::exception& e) {
auto plen = static_cast<uint32_t>(buf.size() - pl_pos); printf("EXCEPTION: %s\n", e.what());
bs2 << source_node CAF_LOG_ERROR(CAF_ARG(e.what()));
<< dest_node
<< source_actor
<< dest_actor
<< plen
<< operation
<< operation_data;
if (payload_len)
*payload_len = plen;
} }
// write broker message to the reserved space
stream_serializer<charbuf> bs2{ctx, buf.data() + wr_pos, pl_pos - wr_pos};
auto plen = static_cast<uint32_t>(buf.size() - pl_pos);
bs2 << source_node
<< dest_node
<< source_actor
<< dest_actor
<< plen
<< operation
<< operation_data;
CAF_LOG_DEBUG("wrote payload" << CAF_ARG(plen));
if (payload_len)
*payload_len = plen;
} }
void instance::write(execution_unit* ctx, buffer_type& buf, void instance::write(execution_unit* ctx, buffer_type& buf,
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <stdexcept> #include <stdexcept>
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
...@@ -187,8 +188,8 @@ private: ...@@ -187,8 +188,8 @@ private:
reuse_addr); reuse_addr);
hdl = res.first; hdl = res.first;
actual_port = res.second; actual_port = res.second;
send(broker_, publish_atom::value, hdl, actual_port, anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs)); std::move(whom), std::move(sigs));
return {ok_atom::value, actual_port}; return {ok_atom::value, actual_port};
} }
catch (std::exception&) { catch (std::exception&) {
......
...@@ -71,7 +71,7 @@ void scribe::data_transferred(execution_unit* ctx, size_t written, ...@@ -71,7 +71,7 @@ void scribe::data_transferred(execution_unit* ctx, size_t written,
if (detached()) if (detached())
return; return;
data_transferred_msg tmp{hdl(), written, remaining}; data_transferred_msg tmp{hdl(), written, remaining};
auto ptr = mailbox_element::make_joint(nullptr, invalid_message_id, {}, tmp); auto ptr = mailbox_element::make(nullptr, invalid_message_id, {}, tmp);
parent()->exec_single_event(ctx, ptr); parent()->exec_single_event(ctx, ptr);
} }
......
...@@ -42,10 +42,13 @@ using pong_atom = caf::atom_constant<atom("pong")>; ...@@ -42,10 +42,13 @@ using pong_atom = caf::atom_constant<atom("pong")>;
using kickoff_atom = caf::atom_constant<atom("kickoff")>; using kickoff_atom = caf::atom_constant<atom("kickoff")>;
using peer = connection_handler::extend<reacts_to<ping_atom, int>, using peer = connection_handler::extend<reacts_to<ping_atom, int>,
reacts_to<pong_atom, int>>; reacts_to<pong_atom, int>>;
using acceptor = using acceptor = accept_handler::extend<replies_to<publish_atom>::with<uint16_t>>;
accept_handler::extend<replies_to<publish_atom>::with<uint16_t>>;
using ping_actor = typed_actor<replies_to<pong_atom, int>::with<ping_atom, int>>;
using pong_actor = typed_actor<replies_to<ping_atom, int>::with<pong_atom, int>>;
behavior ping(event_based_actor* self, size_t num_pings) { behavior ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings); CAF_MESSAGE("num_pings: " << num_pings);
...@@ -71,16 +74,17 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -71,16 +74,17 @@ behavior ping(event_based_actor* self, size_t num_pings) {
behavior pong(event_based_actor* self) { behavior pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started"); CAF_MESSAGE("pong actor started");
self->set_down_handler([=](down_msg& dm) { self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}"); CAF_MESSAGE("received: " << to_string(dm.reason));
self->quit(dm.reason); self->quit(dm.reason);
}); });
return { return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> { [=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom`"); CAF_MESSAGE("received: 'ping', " << value);
self->monitor(self->current_sender()); self->monitor(self->current_sender());
// set next behavior // set next behavior
self->become( self->become(
[](ping_atom, int val) { [](ping_atom, int val) {
//CAF_MESSAGE("received: 'ping', " << val);
return std::make_tuple(pong_atom::value, val); return std::make_tuple(pong_atom::value, val);
} }
); );
...@@ -97,19 +101,13 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl, ...@@ -97,19 +101,13 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
CAF_CHECK(buddy != invalid_actor); CAF_CHECK(buddy != invalid_actor);
self->monitor(buddy); self->monitor(buddy);
// assume exactly one connection // assume exactly one connection
auto cons = self->connections(); CAF_REQUIRE_EQUAL(self->connections().size(), 1u);
if (cons.size() != 1) {
cerr << "expected 1 connection, found " << cons.size() << endl;
throw std::logic_error("num_connections() != 1");
}
self->configure_read( self->configure_read(
hdl, receive_policy::exactly(sizeof(atom_value) + sizeof(int))); hdl, receive_policy::exactly(sizeof(atom_value) + sizeof(int)));
auto write = [=](atom_value type, int value) { auto write = [=](atom_value x, int y) {
auto& buf = self->wr_buf(hdl); auto& buf = self->wr_buf(hdl);
auto first = reinterpret_cast<char*>(&type); binary_serializer sink{self->system(), buf};
buf.insert(buf.end(), first, first + sizeof(atom_value)); sink << x << y;
first = reinterpret_cast<char*>(&value);
buf.insert(buf.end(), first, first + sizeof(int));
self->flush(hdl); self->flush(hdl);
}; };
self->set_down_handler([=](down_msg& dm) { self->set_down_handler([=](down_msg& dm) {
...@@ -124,18 +122,21 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl, ...@@ -124,18 +122,21 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
}, },
[=](const new_data_msg& msg) { [=](const new_data_msg& msg) {
CAF_MESSAGE("received new_data_msg"); CAF_MESSAGE("received new_data_msg");
atom_value type; atom_value x;
int value; int y;
memcpy(&type, msg.buf.data(), sizeof(atom_value)); binary_deserializer source{self->system(), msg.buf};
memcpy(&value, msg.buf.data() + sizeof(atom_value), sizeof(int)); source >> x >> y;
self->send(buddy, type, value); if (x == pong_atom::value)
self->send(actor_cast<ping_actor>(buddy), pong_atom::value, y);
else
self->send(actor_cast<pong_actor>(buddy), ping_atom::value, y);
}, },
[=](ping_atom, int value) { [=](ping_atom, int value) {
CAF_MESSAGE("received ping{" << value << "}"); CAF_MESSAGE("received: 'ping', " << value);
write(ping_atom::value, value); write(ping_atom::value, value);
}, },
[=](pong_atom, int value) { [=](pong_atom, int value) {
CAF_MESSAGE("received pong{" << value << "}"); CAF_MESSAGE("received: 'pong', " << value);
write(pong_atom::value, value); write(pong_atom::value, value);
} }
}; };
......
...@@ -115,8 +115,8 @@ CAF_TEST(unpublishing) { ...@@ -115,8 +115,8 @@ CAF_TEST(unpublishing) {
auto x1 = remote_actor("127.0.0.1", port); auto x1 = remote_actor("127.0.0.1", port);
CAF_CHECK_EQUAL(x1, testee); CAF_CHECK_EQUAL(x1, testee);
CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable"); CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable");
anon_send(system.middleman().actor_handle(), down_msg{testee.address(), anon_send(actor_cast<actor>(system.middleman().actor_handle()),
exit_reason::normal}); down_msg{testee.address(), exit_reason::normal});
// must fail now // must fail now
auto x2 = remote_actor("127.0.0.1", port, true); auto x2 = remote_actor("127.0.0.1", port, true);
CAF_CHECK(! x2); CAF_CHECK(! x2);
......
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