Commit 7836c7b3 authored by Dominik Charousset's avatar Dominik Charousset

Implement bind and "dot composition" for actors

parent 8a5e37c5
......@@ -33,7 +33,9 @@ set (LIBCAF_CORE_SRCS
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp
src/bound_actor.cpp
src/channel.cpp
src/composed_actor.cpp
src/concatenated_tuple.cpp
src/continue_helper.cpp
src/decorated_tuple.cpp
......@@ -55,6 +57,7 @@ set (LIBCAF_CORE_SRCS
src/group.cpp
src/group_manager.cpp
src/match_case.cpp
src/merged_tuple.cpp
src/monitorable_actor.cpp
src/local_actor.cpp
src/logger.cpp
......
......@@ -57,6 +57,13 @@ using abstract_actor_ptr = intrusive_ptr<abstract_actor>;
/// Base class for all actor implementations.
class abstract_actor : public abstract_channel {
public:
void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) override;
/// Enqueues a new message wrapped in a `mailbox_element` to the actor.
/// This `enqueue` variant allows to define forwarding chains.
virtual void enqueue(mailbox_element_ptr what, execution_unit* host) = 0;
/// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on
/// exit, or immediately if it already finished execution.
virtual void attach(attachable_ptr ptr) = 0;
......@@ -123,18 +130,10 @@ public:
return *home_system_;
}
protected:
/// Creates a new actor instance.
explicit abstract_actor(actor_config& cfg);
/// Creates a new actor instance.
abstract_actor(actor_id aid, node_id nid);
/****************************************************************************
* here be dragons: end of public interface *
****************************************************************************/
public:
/// @cond PRIVATE
enum linking_operation {
......@@ -226,7 +225,6 @@ public:
void is_registered(bool value);
protected:
virtual bool link_impl(linking_operation op, const actor_addr& other) = 0;
// cannot be changed after construction
......@@ -236,6 +234,14 @@ protected:
actor_system* home_system_;
/// @endcond
protected:
/// Creates a new actor instance.
explicit abstract_actor(actor_config& cfg);
/// Creates a new actor instance.
abstract_actor(actor_system* sys, actor_id aid, node_id nid,
int flags = abstract_channel::is_abstract_actor_flag);
};
std::string to_string(abstract_actor::linking_operation op);
......
......@@ -39,25 +39,24 @@ public:
virtual ~abstract_channel();
/// Enqueues a new message to the channel.
/// Enqueues a new message without forwarding stack to the channel.
virtual void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) = 0;
/// Enqueues a new message wrapped in a `mailbox_element` to the channel.
/// This variant is used by actors whenever it is possible to allocate
/// mailbox element and message on the same memory block and is thus
/// more efficient. Non-actors use the default implementation which simply
/// calls the pure virtual version.
virtual void enqueue(mailbox_element_ptr what, execution_unit* host);
/// Returns the ID of the node this actor is running on.
inline node_id node() const {
return node_;
}
static constexpr int is_abstract_actor_flag = 0x100000;
static constexpr int is_abstract_actor_flag = 0x100000;
static constexpr int is_abstract_group_flag = 0x200000;
static constexpr int is_actor_bind_decorator_flag = 0x400000;
static constexpr int is_abstract_group_flag = 0x200000;
static constexpr int is_actor_dot_decorator_flag = 0x800000;
static constexpr int is_actor_decorator_mask = 0xC00000;
inline bool is_abstract_actor() const {
return static_cast<bool>(flags() & is_abstract_actor_flag);
......@@ -67,6 +66,10 @@ public:
return static_cast<bool>(flags() & is_abstract_group_flag);
}
inline bool is_actor_decorator() const {
return static_cast<bool>(flags() & is_actor_decorator_mask);
}
protected:
// note: *both* operations use relaxed memory order, this is because
// only the actor itself is granted write access while all access
......
......@@ -26,10 +26,10 @@
#include <utility>
#include <type_traits>
#include "caf/intrusive_ptr.hpp"
#include "caf/fwd.hpp"
#include "caf/actor_marker.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/detail/comparable.hpp"
......@@ -137,6 +137,12 @@ public:
/// Exchange content of `*this` and `other`.
void swap(actor& other) noexcept;
/// Create a new actor decorator that presets or reorders inputs.
template <class... Ts>
actor bind(Ts&&... xs) const {
return bind_impl(make_message(std::forward<Ts>(xs)...));
}
/// @cond PRIVATE
inline abstract_actor* operator->() const noexcept {
......@@ -158,6 +164,8 @@ public:
/// @endcond
private:
actor bind_impl(message msg) const;
inline abstract_actor* get() const noexcept {
return ptr_.get();
}
......@@ -167,6 +175,9 @@ private:
abstract_actor_ptr ptr_;
};
/// Combine `f` and `g` so that `(f*g)(x) = f(g(x))`.
actor operator*(actor f, actor g);
/// @relates actor
void serialize(serializer&, actor&, const unsigned int);
......
......@@ -91,6 +91,8 @@ public:
return ptr_.get();
}
static intptr_t compare(const abstract_actor* lhs, const abstract_actor* rhs);
intptr_t compare(const actor_addr& other) const noexcept;
intptr_t compare(const abstract_actor* other) const noexcept;
......
......@@ -34,7 +34,10 @@ public:
input_range<const group>* groups = nullptr;
std::function<behavior (local_actor*)> init_fun;
explicit actor_config(execution_unit* ptr = nullptr) : host(ptr), flags(0) {
explicit actor_config(execution_unit* ptr = nullptr,
int preset_flags = 0)
: host(ptr),
flags(preset_flags) {
// nop
}
};
......
......@@ -88,7 +88,7 @@ public:
}
protected:
actor_proxy(actor_id aid, node_id nid);
actor_proxy(actor_system* sys, actor_id aid, node_id nid);
anchor_ptr anchor_;
};
......
......@@ -47,6 +47,7 @@
#include "caf/replies_to.hpp"
#include "caf/serializer.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/bound_actor.hpp"
#include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp"
#include "caf/ref_counted.hpp"
......@@ -56,10 +57,12 @@
#include "caf/scoped_actor.hpp"
#include "caf/skip_message.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/index_mapping.hpp"
#include "caf/spawn_options.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/abstract_group.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/composed_actor.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/execution_unit.hpp"
#include "caf/memory_managed.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_BOUND_ACTOR_HPP
#define CAF_BOUND_ACTOR_HPP
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/abstract_actor.hpp"
namespace caf {
/// An actor decorator implementing `std::bind`-like compositions.
class bound_actor : public abstract_actor {
public:
bound_actor(actor_system* sys, actor_addr decorated, message msg);
void attach(attachable_ptr ptr) override;
size_t detach(const attachable::token& what) override;
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
protected:
bool link_impl(linking_operation op, const actor_addr& other) override;
private:
actor_addr decorated_;
message merger_;
};
} // namespace caf
#endif // CAF_BOUND_ACTOR_HPP
......@@ -99,9 +99,6 @@ public:
intptr_t compare(const abstract_channel* other) const noexcept;
static intptr_t compare(const abstract_channel* lhs,
const abstract_channel* rhs) noexcept;
/// @relates channel
friend void serialize(serializer& sink, channel& x, const unsigned int);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_COMPOSED_ACTOR_HPP
#define CAF_COMPOSED_ACTOR_HPP
#include "caf/local_actor.hpp"
namespace caf {
/// An actor decorator implementing "dot operator"-like compositions,
/// i.e., `f.g(x) = f(g(x))`.
class composed_actor : public local_actor {
public:
composed_actor(actor_system* sys, actor_addr first, actor_addr second);
void initialize() override;
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
private:
bool is_system_message(const message& msg);
actor_addr first_;
actor_addr second_;
};
} // namespace caf
#endif // CAF_COMPOSED_ACTOR_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_DETAIL_MERGED_TUPLE_HPP
#define CAF_DETAIL_MERGED_TUPLE_HPP
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/abstract_actor.hpp"
namespace caf {
namespace detail {
class merged_tuple : public message_data {
public:
merged_tuple& operator=(const merged_tuple&) = delete;
using mapping_type = std::vector<std::pair<size_t, size_t>>;
using message_data::cow_ptr;
using data_type = std::vector<cow_ptr>;
merged_tuple(data_type xs, mapping_type ys);
// creates a typed subtuple from `d` with mapping `v`
static cow_ptr make(message x, message y);
void* mutable_at(size_t pos) override;
void serialize_at(deserializer& source, size_t pos) override;
size_t size() const override;
cow_ptr copy() const override;
const void* at(size_t pos) const override;
bool compare_at(size_t pos, const element_rtti& rtti,
const void* x) const override;
bool match_element(size_t pos, uint16_t typenr,
const std::type_info* rtti) const override;
uint32_t type_token() const override;
element_rtti type_at(size_t pos) const override;
void serialize_at(serializer& sink, size_t pos) const override;
std::string stringify_at(size_t pos) const override;
const mapping_type& mapping() const;
private:
merged_tuple(const merged_tuple&) = default;
data_type data_;
uint32_t type_token_;
mapping_type mapping_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_MERGED_TUPLE_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_DETAIL_MPI_BIND_HPP
#define CAF_DETAIL_MPI_BIND_HPP
#include <functional>
#include "caf/none.hpp"
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
namespace detail {
template <class T, size_t Pos>
class mpi_bind_sig_arg_t {
// nop
};
template <class T,
int PlaceholderValue,
size_t Pos,
size_t Size,
bool InRange = PlaceholderValue <= Size>
struct mpi_bind_sig_single {
using type =
mpi_bind_sig_arg_t<
typename detail::tl_at<T, Pos>::type,
PlaceholderValue - 1
>;
};
template <class T, size_t Pos, size_t Size>
struct mpi_bind_sig_single<T, 0, Pos, Size, true> {
using type = void;
};
template <class T, int PlaceholderValue, size_t Pos, size_t Size>
struct mpi_bind_sig_single<T, PlaceholderValue, Pos, Size, false> {
using type = none_t;
};
template <size_t I, size_t Size, class Args, class In, class... Ts>
struct mpi_bind_sig_impl {
using sub =
typename mpi_bind_sig_single<
In,
std::is_placeholder<typename detail::tl_at<Args, I>::type>::value,
I,
Size
>::type;
using type =
typename std::conditional<
std::is_same<sub, none_t>::value,
void,
typename std::conditional<
std::is_same<sub, void>::value,
typename mpi_bind_sig_impl<I + 1, Size, Args, In, Ts...>::type,
typename mpi_bind_sig_impl<I + 1, Size, Args, In, Ts..., sub>::type
>::type
>::type;
};
template <size_t Size, class Args, class In>
struct mpi_bind_sig_impl<Size, Size, Args, In> {
using type = void;
};
template <size_t Size, class Args, class In, class T, class... Ts>
struct mpi_bind_sig_impl<Size, Size, Args, In, T, Ts...> {
using type = detail::type_list<T, Ts...>;
};
template <size_t I, class In, class Out, class... Ts>
struct mpi_bind_sort;
template <size_t I, class Out, class... Ts>
struct mpi_bind_sort<I, detail::type_list<>, Out, Ts...> {
using type = typed_mpi<detail::type_list<Ts...>, Out>;
};
template <size_t I, size_t X, class Arg, class... Args, class Out, class... Ts>
struct mpi_bind_sort<I, detail::type_list<mpi_bind_sig_arg_t<Arg, X>, Args...>,
Out, Ts...> {
using type =
typename mpi_bind_sort<
I,
detail::type_list<Args..., mpi_bind_sig_arg_t<Arg, X>>,
Out,
Ts...
>::type;
};
template <size_t I, class Arg, class... Args, class Out, class... Ts>
struct mpi_bind_sort<I, detail::type_list<mpi_bind_sig_arg_t<Arg, I>, Args...>,
Out, Ts...> {
using type =
typename mpi_bind_sort<
I + 1,
detail::type_list<Args...>,
Out,
Ts...,
Arg
>::type;
};
template <class Sig, class Args>
struct mpi_bind_sig {
static constexpr size_t num_args = detail::tl_size<Args>::value;
using type =
typename mpi_bind_sort<
0,
typename mpi_bind_sig_impl<
0,
num_args,
Args,
typename Sig::input_types
>::type,
typename Sig::output_types
>::type;
};
template <template <class...> class Target, class Sigs, class... Args>
struct mpi_bind;
template <template <class...> class Target, class... Sigs, class... Ts>
struct mpi_bind<Target, detail::type_list<Sigs...>, Ts...> {
using args = detail::type_list<Ts...>;
using bound_sigs = detail::type_list<typename mpi_bind_sig<Sigs, args>::type...>;
// drop any mismatch (void) and rebuild typed actor handle
using type =
typename detail::tl_apply<
typename detail::tl_filter_not_type<bound_sigs, void>::type,
Target
>::type;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_MPI_BIND_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_DETAIL_MPI_COMPOSITION_HPP
#define CAF_DETAIL_MPI_COMPOSITION_HPP
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
namespace detail {
template <class X, class Y>
struct mpi_composition_one {
using type = void;
};
template <class... Xs, class... Ys, class... Zs>
struct mpi_composition_one<typed_mpi<type_list<Xs...>, type_list<Ys...>>,
typed_mpi<type_list<Ys...>, type_list<Zs...>>> {
using type = typed_mpi<type_list<Xs...>, type_list<Zs...>>;
};
template <class X, class Y>
struct mpi_composition_all;
template <class X, class... Ys>
struct mpi_composition_all<X, type_list<Ys...>> {
using type = type_list<typename mpi_composition_one<X, Ys>::type...>;
};
template <template <class...> class Target, class Ys, class... Xs>
struct mpi_composition {
// combine each X with all Ys
using all =
typename tl_concat<
typename mpi_composition_all<Xs, Ys>::type...
>::type;
// drop all mismatches (void results)
using filtered = typename tl_filter_not_type<all, void>::type;
// throw error if we don't have a single match
static_assert(tl_size<filtered>::value > 0,
"Left-hand actor type does not produce a single result which "
"is valid as input to the right-hand actor type.");
// compute final actor type
using type = typename tl_apply<filtered, Target>::type;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_MPI_COMPOSITION_HPP
......@@ -81,6 +81,16 @@ public:
// nop
}
template <class T0, class T1, class T2, class... Ts>
pair_storage(intrusive_ptr<ref_counted> storage,
std::integral_constant<size_t, 3>,
T0&& x0, T1&& x1, T2&& x2, Ts&&... xs)
: first(storage, std::forward<T0>(x0),
std::forward<T1>(x1), std::forward<T2>(x2)),
second(std::move(storage), std::forward<Ts>(xs)...) {
// nop
}
~pair_storage() {
// nop
}
......
......@@ -30,12 +30,12 @@ namespace caf {
/// Implements a simple proxy forwarding all operations to a manager.
class forwarding_actor_proxy : public actor_proxy {
public:
forwarding_actor_proxy(actor_id mid, node_id pinfo, actor parent);
forwarding_actor_proxy(actor_system* sys, actor_id mid,
node_id pinfo, actor parent);
~forwarding_actor_proxy();
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
bool link_impl(linking_operation op, const actor_addr& other) override;
......@@ -50,7 +50,8 @@ public:
void manager(actor new_manager);
private:
void forward_msg(const actor_addr& sender, message_id mid, message msg);
void forward_msg(const actor_addr& sender, message_id mid, message msg,
const std::vector<actor_addr>* fwd_stack = nullptr);
mutable detail::shared_spinlock manager_mtx_;
actor manager_;
......
......@@ -26,10 +26,10 @@
namespace caf {
template <class>
class intrusive_ptr;
class maybe;
template <class>
class maybe;
class intrusive_ptr;
// classes
class actor;
......
......@@ -83,6 +83,8 @@ public:
return *get();
}
static intptr_t compare(const abstract_group* lhs, const abstract_group* rhs);
intptr_t compare(const group& other) const noexcept;
inline intptr_t compare(const invalid_group_t&) const noexcept {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_INDEX_MAPPING_HPP
#define CAF_INDEX_MAPPING_HPP
#include <tuple>
#include <string>
#include <functional>
#include "caf/deep_to_string.hpp"
namespace caf {
/// Marker for representing placeholders at runtime.
struct index_mapping {
int value;
explicit index_mapping(int x) : value(x) {
// nop
}
template <class T,
class E = typename std::enable_if<
std::is_placeholder<T>::value != 0
>::type>
index_mapping(T) : value(std::is_placeholder<T>::value) {
// nop
}
};
inline bool operator==(const index_mapping& x, const index_mapping& y) {
return x.value == y.value;
}
template <class T>
void serialize(T& in_or_out, index_mapping& x, const unsigned int) {
in_or_out & x.value;
}
inline std::string to_string(const index_mapping& x) {
return "idx" + deep_to_string(std::forward_as_tuple(x.value));
}
} // namespace caf
#endif // CAF_INDEX_MAPPING_HPP
......@@ -160,7 +160,7 @@ public:
>::type...>;
token tk;
check_typed_input(dest, tk);
send_impl(message_id::make(mp), actor_cast<abstract_channel*>(dest),
send_impl(message_id::make(mp), actor_cast<abstract_actor*>(dest),
std::forward<Ts>(xs)...);
}
......@@ -174,7 +174,7 @@ public:
>::type...>;
token tk;
check_typed_input(dest, tk);
send_impl(message_id::make(), actor_cast<abstract_channel*>(dest),
send_impl(message_id::make(), actor_cast<abstract_actor*>(dest),
std::forward<Ts>(xs)...);
}
......@@ -396,6 +396,10 @@ public:
local_actor(actor_config& sys);
local_actor(actor_system* sys, actor_id aid, node_id nid);
local_actor(actor_system* sys, actor_id aid, node_id nid, int init_flags);
template <class ActorHandle>
inline ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts)) {
......@@ -413,12 +417,10 @@ public:
template <class Handle, class... Ts>
message_id request_impl(message_priority mp, const Handle& dh, Ts&&... xs) {
if (! dh) {
if (! dh)
throw std::invalid_argument("cannot request to invalid_actor");
}
auto req_id = new_request_id(mp);
send_impl(req_id, actor_cast<abstract_channel*>(dh),
std::forward<Ts>(xs)...);
send_impl(req_id, actor_cast<abstract_actor*>(dh), std::forward<Ts>(xs)...);
return req_id.response_id();
}
......@@ -569,8 +571,7 @@ public:
void launch(execution_unit* eu, bool lazy, bool hide);
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr, execution_unit*) override;
......@@ -629,11 +630,22 @@ private:
! std::is_same<typename std::decay<T>::type, message>::value
>::type
send_impl(message_id mid, abstract_channel* dest, T&& x, Ts&&... xs) const {
if (! dest) {
if (! dest)
return;
dest->enqueue(address(), mid,
make_message(std::forward<T>(x), std::forward<Ts>(xs)...),
context());
}
template <class T, class... Ts>
typename std::enable_if<
! std::is_same<typename std::decay<T>::type, message>::value
>::type
send_impl(message_id mid, abstract_actor* dest, T&& x, Ts&&... xs) const {
if (! dest)
return;
}
dest->enqueue(mailbox_element::make_joint(address(),
mid,
mid, {},
std::forward<T>(x),
std::forward<Ts>(xs)...),
context());
......
......@@ -42,16 +42,28 @@ class mailbox_element : public memory_managed {
public:
static constexpr auto memory_cache_flag = detail::needs_embedding;
mailbox_element* next; // intrusive next pointer
mailbox_element* prev; // intrusive previous pointer
bool marked; // denotes if this node is currently processed
using forwarding_stack = std::vector<actor_addr>;
// intrusive pointer to the next mailbox element
mailbox_element* next;
// intrusive pointer to the previous mailbox element
mailbox_element* prev;
// avoid multi-processing in blocking actors via flagging
bool marked;
// source of this message and receiver of the final response
actor_addr sender;
// denotes whether this a sync or async message
message_id mid;
message msg; // 'content field'
// stages.back() is the next actor in the forwarding chain,
// if this is empty then the original sender receives the response
forwarding_stack stages;
// content of this element
message msg;
mailbox_element();
mailbox_element(actor_addr sender, message_id id);
mailbox_element(actor_addr sender, message_id id, message data);
mailbox_element(actor_addr sender, message_id id, forwarding_stack stages);
mailbox_element(actor_addr sender, message_id id,
forwarding_stack stages, message data);
~mailbox_element();
......@@ -62,19 +74,22 @@ public:
using unique_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
static unique_ptr make(actor_addr sender, message_id id, message msg);
static unique_ptr make(actor_addr sender, message_id id,
forwarding_stack stages, message msg);
template <class... Ts>
static unique_ptr make_joint(actor_addr sender, message_id id, Ts&&... xs) {
static unique_ptr make_joint(actor_addr sender, message_id id,
forwarding_stack stages, Ts&&... xs) {
using value_storage =
detail::tuple_vals<
typename unbox_message_element<
typename detail::strip_and_convert<Ts>::type
>::type...
>;
std::integral_constant<size_t, 2> tk;
std::integral_constant<size_t, 3> tk;
using storage = detail::pair_storage<mailbox_element, value_storage>;
auto ptr = detail::memory::create<storage>(tk, std::move(sender), id,
std::move(stages),
std::forward<Ts>(xs)...);
ptr->first.msg.reset(&(ptr->second), false);
return unique_ptr{&(ptr->first)};
......
......@@ -29,6 +29,7 @@
#include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/skip_message.hpp"
#include "caf/index_mapping.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp"
......@@ -367,13 +368,18 @@ inline message operator+(const message& lhs, const message& rhs) {
/// Unboxes atom constants, i.e., converts `atom_constant<V>` to `V`.
/// @relates message
template <class T>
template <class T, int IsPlaceholderRes = std::is_placeholder<T>::value>
struct unbox_message_element {
using type = index_mapping;
};
template <class T>
struct unbox_message_element<T, 0> {
using type = T;
};
template <atom_value V>
struct unbox_message_element<atom_constant<V>> {
struct unbox_message_element<atom_constant<V>, 0> {
using type = atom_value;
};
......
......@@ -59,7 +59,10 @@ protected:
explicit monitorable_actor(actor_config& cfg);
/// Creates a new actor instance.
monitorable_actor(actor_id aid, node_id nid);
monitorable_actor(actor_system* sys, actor_id aid, node_id nid);
/// Creates a new actor instance.
monitorable_actor(actor_system* sys, actor_id aid, node_id nid, int flags);
/// Called by the runtime system to perform cleanup actions for this actor.
/// Subtypes should always call this member function when overriding it.
......
......@@ -20,6 +20,8 @@
#ifndef CAF_RESPONSE_PROMISE_HPP
#define CAF_RESPONSE_PROMISE_HPP
#include <vector>
#include "caf/actor.hpp"
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
......@@ -38,17 +40,26 @@ public:
response_promise& operator=(response_promise&&) = default;
response_promise& operator=(const response_promise&) = default;
response_promise(local_actor* self, const actor_addr& to,
const message_id& response_id);
using forwarding_stack = std::vector<actor_addr>;
//response_promise(local_actor* self, actor_addr source,
// forwarding_stack stages,
// message_id response_id);
response_promise(local_actor* self, mailbox_element& src);
/// Queries whether this promise is still valid, i.e., no response
/// was yet delivered to the client.
inline bool valid() const {
// handle is valid if it has a receiver or a next stage
return source_ || ! stages_.empty();
}
inline explicit operator bool() const {
// handle is valid if it has a receiver
return static_cast<bool>(to_);
return valid();
}
/// Sends the response_message.
/// Sends the response_message and invalidate this promise.
template <class T, class... Ts>
typename std::enable_if<
! std::is_convertible<T, error>::value
......@@ -57,13 +68,15 @@ public:
deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
}
/// Sends an error as response unless the sender used asynchronous messaging.
/// Sends an error as response unless the sender used asynchronous messaging
/// and invalidate this promise.
void deliver(error x) const;
private:
void deliver_impl(message response_message) const;
local_actor* self_;
actor_addr to_;
mutable actor_addr source_;
mutable forwarding_stack stages_;
message_id id_;
};
......
......@@ -22,20 +22,19 @@
#include "caf/intrusive_ptr.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/replies_to.hpp"
#include "caf/bound_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_response_promise.hpp"
namespace caf {
class actor_addr;
class local_actor;
#include "caf/detail/mpi_bind.hpp"
#include "caf/detail/mpi_composition.hpp"
struct invalid_actor_addr_t;
namespace caf {
template <class... Sigs>
class typed_event_based_actor;
......@@ -167,6 +166,15 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
// nop
}
typed_actor(const invalid_actor_t&) {
// nop
}
typed_actor& operator=(const invalid_actor_t&) {
ptr_.reset();
}
/// Queries the address of the stored actor.
actor_addr address() const noexcept {
return ptr_ ? ptr_->address() : actor_addr();
......@@ -197,6 +205,20 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
ptr_.swap(other.ptr_);
}
template <class... Ts>
typename detail::mpi_bind<
caf::typed_actor,
detail::type_list<Sigs...>,
typename std::decay<Ts>::type...
>::type
bind(Ts&&... xs) const {
if (! ptr_)
return invalid_actor;
auto ptr = make_counted<bound_actor>(&ptr_->home_system(), ptr_->address(),
make_message(xs...));
return {ptr.release(), false};
}
/// @cond PRIVATE
abstract_actor* operator->() const noexcept {
......@@ -207,19 +229,19 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
return *ptr_.get();
}
intptr_t compare(const actor_addr& rhs) const noexcept {
return address().compare(rhs);
intptr_t compare(const typed_actor& x) const noexcept {
return actor_addr::compare(get(), x.get());
}
intptr_t compare(const typed_actor& other) const noexcept {
return compare(other.address());
intptr_t compare(const actor_addr& x) const noexcept {
return actor_addr::compare(get(), actor_cast<abstract_actor*>(x));
}
intptr_t compare(const invalid_actor_addr_t&) const noexcept {
intptr_t compare(const invalid_actor_t&) const noexcept {
return ptr_ ? 1 : 0;
}
intptr_t compare(const invalid_actor_t&) const noexcept {
intptr_t compare(const invalid_actor_addr_t&) const noexcept {
return ptr_ ? 1 : 0;
}
......@@ -234,22 +256,58 @@ private:
// nop
}
typed_actor(abstract_actor* ptr, bool add_ref) : ptr_(ptr, add_ref) {
// nop
}
abstract_actor_ptr ptr_;
};
template <class T, class... Ts>
typename std::enable_if<
T::is_saving::value
/// @relates typed_actor
template <class... Xs, class... Ys>
bool operator==(const typed_actor<Xs...>& x,
const typed_actor<Ys...>& y) noexcept {
return actor_addr::compare(actor_cast<abstract_actor*>(x),
actor_cast<abstract_actor*>(y)) == 0;
}
/// @relates typed_actor
template <class... Xs, class... Ys>
bool operator!=(const typed_actor<Xs...>& x,
const typed_actor<Ys...>& y) noexcept {
return ! (x == y);
}
/// Returns a new actor that implements the composition `f.g(x) = f(g(x))`.
/// @relates typed_actor
template <class... Xs, class... Ys>
typename detail::mpi_composition<
typed_actor,
detail::type_list<Ys...>,
Xs...
>::type
operator*(typed_actor<Xs...> f, typed_actor<Ys...> g) {
using result =
typename detail::mpi_composition<
typed_actor,
detail::type_list<Ys...>,
Xs...
>::type;
return actor_cast<result>(actor_cast<actor>(std::move(f))
* actor_cast<actor>(std::move(g)));
}
/// @relates typed_actor
template <class T, class... Ts>
typename std::enable_if<T::is_saving::value>::type
serialize(T& sink, typed_actor<Ts...>& hdl, const unsigned int) {
auto addr = hdl.address();
sink << addr;
}
/// @relates typed_actor
template <class T, class... Ts>
typename std::enable_if<
T::is_loading::value
>::type
typename std::enable_if<T::is_loading::value>::type
serialize(T& sink, typed_actor<Ts...>& hdl, const unsigned int) {
actor_addr addr;
sink >> addr;
......
......@@ -26,17 +26,18 @@
#include "caf/atom.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_registry.hpp"
#include "caf/execution_unit.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp"
#include "caf/logger.hpp"
#include "caf/actor_registry.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
......@@ -44,6 +45,11 @@ namespace caf {
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
void abstract_actor::enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* host) {
enqueue(mailbox_element::make(sender, mid, {}, std::move(msg)), host);
}
abstract_actor::abstract_actor(actor_config& cfg)
: abstract_channel(cfg.flags | abstract_channel::is_abstract_actor_flag,
cfg.host->system().node()),
......@@ -52,10 +58,11 @@ abstract_actor::abstract_actor(actor_config& cfg)
// nop
}
abstract_actor::abstract_actor(actor_id aid, node_id nid)
: abstract_channel(abstract_channel::is_abstract_actor_flag,
std::move(nid)),
id_(aid) {
abstract_actor::abstract_actor(actor_system* sys, actor_id aid,
node_id nid, int flags)
: abstract_channel(flags, std::move(nid)),
id_(aid),
home_system_(sys) {
// nop
}
......
......@@ -34,8 +34,4 @@ abstract_channel::~abstract_channel() {
// nop
}
void abstract_channel::enqueue(mailbox_element_ptr what, execution_unit* host) {
enqueue(what->sender, what->mid, std::move(what->msg), host);
}
} // namespace caf
......@@ -25,9 +25,11 @@
#include "caf/actor_addr.hpp"
#include "caf/serializer.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/bound_actor.hpp"
#include "caf/local_actor.hpp"
#include "caf/deserializer.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/composed_actor.hpp"
#include "caf/event_based_actor.hpp"
namespace caf {
......@@ -46,11 +48,11 @@ actor& actor::operator=(const invalid_actor_t&) {
}
intptr_t actor::compare(const actor& other) const noexcept {
return channel::compare(ptr_.get(), other.ptr_.get());
return actor_addr::compare(ptr_.get(), other.ptr_.get());
}
intptr_t actor::compare(const actor_addr& other) const noexcept {
return static_cast<ptrdiff_t>(ptr_.get() - other.ptr_.get());
return actor_addr::compare(ptr_.get(), other.ptr_.get());
}
void actor::swap(actor& other) noexcept {
......@@ -69,6 +71,22 @@ actor_id actor::id() const noexcept {
return ptr_ ? ptr_->id() : invalid_actor_id;
}
actor actor::bind_impl(message msg) const {
if (! ptr_)
return invalid_actor;
return actor_cast<actor>(make_counted<bound_actor>(&ptr_->home_system(),
address(),
std::move(msg)));
}
actor operator*(actor f, actor g) {
if (! f || ! g)
return invalid_actor;
auto ptr = make_counted<composed_actor>(&f->home_system(),
f.address(), g.address());
return actor_cast<actor>(std::move(ptr));
}
void serialize(serializer& sink, actor& x, const unsigned int) {
sink << x.address();
}
......
......@@ -28,14 +28,6 @@
namespace caf {
namespace {
intptr_t compare_impl(const abstract_actor* lhs, const abstract_actor* rhs) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
} // namespace <anonymous>
actor_addr::actor_addr(const invalid_actor_addr_t&) : ptr_(nullptr) {
// nop
}
......@@ -48,12 +40,31 @@ actor_addr actor_addr::operator=(const invalid_actor_addr_t&) {
ptr_.reset();
return *this;
}
intptr_t actor_addr::compare(const abstract_actor* lhs,
const abstract_actor* rhs) {
// invalid actors are always "less" than valid actors
if (! lhs)
return rhs ? -1 : 0;
if (! rhs)
return 1;
// check for identity
if (lhs == rhs)
return 0;
// check for equality (a decorator is equal to the actor it represents)
auto x = lhs->id();
auto y = rhs->id();
if (x == y)
return lhs->node().compare(rhs->node());
return static_cast<intptr_t>(x) - static_cast<intptr_t>(y);
}
intptr_t actor_addr::compare(const actor_addr& other) const noexcept {
return compare_impl(ptr_.get(), other.ptr_.get());
return compare(ptr_.get(), other.ptr_.get());
}
intptr_t actor_addr::compare(const abstract_actor* other) const noexcept {
return compare_impl(ptr_.get(), other);
return compare(ptr_.get(), other);
}
......
......@@ -46,7 +46,7 @@ void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
void actor_companion::enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* eu) {
using detail::memory;
auto ptr = mailbox_element::make(sender, mid, std::move(content));
auto ptr = mailbox_element::make(sender, mid, {}, std::move(content));
enqueue(std::move(ptr), eu);
}
......
......@@ -120,7 +120,7 @@ void actor_pool::enqueue(const actor_addr& sender, message_id mid,
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));
auto ptr = mailbox_element::make(sender, mid, {}, std::move(content));
policy_(home_system(), guard, workers_, ptr, eu);
}
......
......@@ -66,8 +66,8 @@ actor_proxy::~actor_proxy() {
// nop
}
actor_proxy::actor_proxy(actor_id aid, node_id nid)
: monitorable_actor(aid, nid),
actor_proxy::actor_proxy(actor_system* sys, actor_id aid, node_id nid)
: monitorable_actor(sys, aid, nid),
anchor_(make_counted<anchor>(this)) {
// nop
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/bound_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/merged_tuple.hpp"
namespace caf {
bound_actor::bound_actor(actor_system* sys, actor_addr decorated, message msg)
: abstract_actor(sys, decorated->id(), decorated->node(),
is_abstract_actor_flag | is_actor_bind_decorator_flag),
decorated_(std::move(decorated)),
merger_(std::move(msg)) {
// nop
}
void bound_actor::attach(attachable_ptr ptr) {
decorated_->attach(std::move(ptr));
}
size_t bound_actor::detach(const attachable::token& what) {
return decorated_->detach(what);
}
void bound_actor::enqueue(mailbox_element_ptr what, execution_unit* host) {
// ignore system messages
auto& msg = what->msg;
if (! ((msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0)))) {
// merge non-system messages
message tmp{detail::merged_tuple::make(merger_, std::move(msg))};
msg.swap(tmp);
}
decorated_->enqueue(std::move(what), host);
}
bool bound_actor::link_impl(linking_operation op, const actor_addr& other) {
if (actor_cast<abstract_actor*>(other) == this || other == decorated_)
return false;
return decorated_->link_impl(op, other);
}
} // namespace caf
......@@ -22,6 +22,7 @@
#include "caf/actor.hpp"
#include "caf/group.hpp"
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_system.hpp"
......@@ -43,11 +44,6 @@ channel::channel(const invalid_channel_t&) {
// nop
}
intptr_t channel::compare(const abstract_channel* lhs,
const abstract_channel* rhs) noexcept {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
channel::channel(abstract_channel* ptr) : ptr_(ptr) {
// nop
}
......@@ -72,15 +68,34 @@ channel& channel::operator=(const invalid_channel_t&) {
}
intptr_t channel::compare(const channel& other) const noexcept {
return compare(ptr_.get(), other.ptr_.get());
return compare(other.ptr_.get());
}
intptr_t channel::compare(const actor& other) const noexcept {
return compare(ptr_.get(), actor_cast<abstract_actor_ptr>(other).get());
return compare(actor_cast<abstract_channel*>(other));
}
intptr_t channel::compare(const abstract_channel* other) const noexcept {
return compare(ptr_.get(), other);
intptr_t channel::compare(const abstract_channel* rhs) const noexcept {
// invalid handles are always "less" than valid handles
if (! ptr_)
return rhs ? -1 : 0;
if (! rhs)
return 1;
// check for identity
if (ptr_ == rhs)
return 0;
// groups are sorted before actors
if (ptr_->is_abstract_group()) {
if (! rhs->is_abstract_group())
return -1;
using gptr = const abstract_group*;
return group::compare(static_cast<gptr>(get()), static_cast<gptr>(rhs));
}
CAF_ASSERT(ptr_->is_abstract_actor());
if (! rhs->is_abstract_actor())
return 1;
using aptr = const abstract_actor*;
return actor_addr::compare(static_cast<aptr>(get()), static_cast<aptr>(rhs));
}
void serialize(serializer& sink, channel& x, const unsigned int) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/composed_actor.hpp"
namespace caf {
composed_actor::composed_actor(actor_system* sys,
actor_addr first, actor_addr second)
: local_actor(sys, first->id(), first->node(),
is_abstract_actor_flag | is_actor_dot_decorator_flag),
first_(first),
second_(second) {
// nop
}
void composed_actor::initialize() {
trap_exit(true);
link_to(first_);
link_to(second_);
do_become(
{
[=](const exit_msg& em) {
if (em.source == first_ || em.source == second_)
quit(em.reason);
else if (em.reason != exit_reason::normal)
quit(em.reason);
},
others >> [] {
// drop
}
},
true
);
}
void composed_actor::enqueue(mailbox_element_ptr what, execution_unit* host) {
if (! what)
return;
if (is_system_message(what->msg)) {
local_actor::enqueue(std::move(what), host);
return;
}
// store second_ as the next stage in the forwarding chain
what->stages.push_back(second_);
// forward modified message to first_
first_->enqueue(std::move(what), host);
}
bool composed_actor::is_system_message(const message& msg) {
return (msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0));
}
} // namespace caf
......@@ -21,14 +21,14 @@
#include "caf/send.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_id aid,
forwarding_actor_proxy::forwarding_actor_proxy(actor_system* sys, actor_id aid,
node_id nid, actor mgr)
: actor_proxy(aid, nid),
: actor_proxy(sys, aid, nid),
manager_(mgr) {
CAF_ASSERT(mgr != invalid_actor);
CAF_PUSH_AID(0);
......@@ -54,22 +54,29 @@ void forwarding_actor_proxy::manager(actor new_manager) {
}
void forwarding_actor_proxy::forward_msg(const actor_addr& sender,
message_id mid, message msg) {
message_id mid, message msg,
const std::vector<actor_addr>* fwd) {
CAF_LOG_TRACE(CAF_ARG(id()) << CAF_ARG(sender)
<< CAF_ARG(mid) << CAF_ARG(msg));
std::vector<actor_addr> tmp;
shared_lock<detail::shared_spinlock> guard_(manager_mtx_);
if (manager_) manager_->enqueue(invalid_actor_addr, invalid_message_id,
make_message(forward_atom::value, sender,
address(), mid, std::move(msg)),
nullptr);
if (manager_)
manager_->enqueue(invalid_actor_addr, invalid_message_id,
make_message(forward_atom::value, sender,
fwd ? *fwd : tmp, address(),
mid, std::move(msg)),
nullptr);
}
void forwarding_actor_proxy::enqueue(const actor_addr& sender, message_id mid,
message m, execution_unit*) {
void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
execution_unit*) {
CAF_PUSH_AID(0);
forward_msg(sender, mid, std::move(m));
if (! what)
return;
forward_msg(what->sender, what->mid, std::move(what->msg), &what->stages);
}
bool forwarding_actor_proxy::link_impl(linking_operation op,
const actor_addr& other) {
switch (op) {
......
......@@ -43,8 +43,12 @@ group& group::operator=(const invalid_group_t&) {
return *this;
}
intptr_t group::compare(const abstract_group* lhs, const abstract_group* rhs) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
}
intptr_t group::compare(const group& other) const noexcept {
return channel::compare(ptr_.get(), other.ptr_.get());
return compare(ptr_.get(), other.ptr_.get());
}
void serialize(serializer& sink, const group& x, const unsigned int) {
......
......@@ -52,6 +52,21 @@ local_actor::local_actor(actor_config& cfg)
join(grp);
}
local_actor::local_actor(actor_system* sys, actor_id aid, node_id nid)
: monitorable_actor(sys, aid, std::move(nid)),
planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0) {
// nop
}
local_actor::local_actor(actor_system* sys, actor_id aid,
node_id nid, int init_flags)
: monitorable_actor(sys, aid, std::move(nid), init_flags),
planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0) {
// nop
}
local_actor::~local_actor() {
if (! mailbox_.closed()) {
detail::sync_request_bouncer f{exit_reason_};
......@@ -199,7 +214,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
if (! self->is_serializable()) {
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
sec::state_not_serializable),
{}, sec::state_not_serializable),
self->context());
return;
}
......@@ -215,7 +230,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
[=](ok_atom, const actor_addr& dest) {
// respond to original message with {'OK', dest}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(),
mid.response_id(), {},
ok_atom::value, dest),
self->context());
// "decay" into a proxy for `dest`
......@@ -230,7 +245,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
[=](error& err) {
// respond to original message with {'ERROR', errmsg}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(),
mid.response_id(), {},
std::move(err)),
self->context());
}
......@@ -240,7 +255,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
// "replace" this actor with the content of `buf`
if (! self->is_serializable()) {
node.sender->enqueue(mailbox_element::make_joint(
self->address(), node.mid.response_id(),
self->address(), node.mid.response_id(), {},
sec::state_not_serializable),
self->context());
return;
......@@ -254,7 +269,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
self->load_state(bd, 0);
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
ok_atom::value, self->address()),
{}, ok_atom::value, self->address()),
self->context());
},
[&](sys_atom, get_atom, std::string& what) {
......@@ -263,14 +278,14 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
CAF_LOG_DEBUG("reply to 'info' message");
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
ok_atom::value, std::move(what),
{}, ok_atom::value, std::move(what),
self->address(), self->name()),
self->context());
return;
}
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
sec::invalid_sys_key),
{}, sec::invalid_sys_key),
self->context());
},
others >> [&] {
......@@ -567,11 +582,6 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
eu->exec_later(this);
}
void local_actor::enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) {
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
}
void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_ASSERT(ptr != nullptr);
CAF_PUSH_AID(id());
......@@ -898,12 +908,12 @@ void local_actor::await_data() {
void local_actor::send_impl(message_id mid, abstract_channel* dest,
message what) const {
if (! dest) {
if (! dest)
return;
}
dest->enqueue(address(), mid, std::move(what), context());
}
void local_actor::send_exit(const actor_addr& whom, exit_reason reason) {
send(message_priority::high, actor_cast<actor>(whom),
exit_msg{address(), reason});
......@@ -922,8 +932,7 @@ response_promise local_actor::make_response_promise() {
auto& mid = ptr->mid;
if (mid.is_answered())
return response_promise{};
response_promise result{this, ptr->sender, mid.response_id()};
mid.mark_as_answered();
response_promise result{this, *ptr};
return result;
}
......
......@@ -28,22 +28,25 @@ mailbox_element::mailbox_element()
// nop
}
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1)
mailbox_element::mailbox_element(actor_addr x, message_id y, forwarding_stack z)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(arg0)),
mid(arg1) {
sender(std::move(x)),
mid(y),
stages(std::move(z)) {
// nop
}
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2)
mailbox_element::mailbox_element(actor_addr x, message_id y,
forwarding_stack z, message m)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(arg0)),
mid(arg1),
msg(std::move(arg2)) {
sender(std::move(x)),
mid(y),
stages(std::move(z)),
msg(std::move(m)) {
// nop
}
......@@ -52,8 +55,10 @@ mailbox_element::~mailbox_element() {
}
mailbox_element_ptr mailbox_element::make(actor_addr sender, message_id id,
message msg) {
forwarding_stack stages,
message msg) {
auto ptr = detail::memory::create<mailbox_element>(std::move(sender), id,
std::move(stages),
std::move(msg));
return mailbox_element_ptr{ptr};
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/detail/merged_tuple.hpp"
#include "caf/index_mapping.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/disposer.hpp"
namespace caf {
namespace detail {
merged_tuple::merged_tuple(data_type xs, mapping_type ys)
: data_(std::move(xs)),
type_token_(0xFFFFFFFF),
mapping_(std::move(ys)) {
CAF_ASSERT(! data_.empty());
CAF_ASSERT(! mapping_.empty());
// calculate type token
for (size_t i = 0; i < mapping_.size(); ++i) {
type_token_ <<= 6;
auto& p = mapping_[i];
type_token_ |= data_[p.first]->type_nr_at(p.second);
}
}
// creates a typed subtuple from `d` with mapping `v`
merged_tuple::cow_ptr merged_tuple::make(message x, message y) {
data_type data{x.vals(), y.vals()};
mapping_type mapping;
auto s = x.size();
for (size_t i = 0; i < s; ++i) {
if (x.match_element<index_mapping>(i))
mapping.emplace_back(1, x.get_as<index_mapping>(i).value - 1);
else
mapping.emplace_back(0, i);
}
return cow_ptr{make_counted<merged_tuple>(std::move(data),
std::move(mapping))};
}
void* merged_tuple::mutable_at(size_t pos) {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->mutable_at(p.second);
}
void merged_tuple::serialize_at(deserializer& source, size_t pos) {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
data_[p.first]->serialize_at(source, p.second);
}
size_t merged_tuple::size() const {
return mapping_.size();
}
merged_tuple::cow_ptr merged_tuple::copy() const {
return cow_ptr{make_counted<merged_tuple>(data_, mapping_)};
}
const void* merged_tuple::at(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->at(p.second);
}
bool merged_tuple::compare_at(size_t pos, const element_rtti& rtti,
const void* x) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->compare_at(p.second, rtti, x);
}
bool merged_tuple::match_element(size_t pos, uint16_t typenr,
const std::type_info* rtti) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->match_element(p.second, typenr, rtti);
}
uint32_t merged_tuple::type_token() const {
return type_token_;
}
merged_tuple::element_rtti merged_tuple::type_at(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->type_at(p.second);
}
void merged_tuple::serialize_at(serializer& sink, size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
data_[p.first]->serialize_at(sink, p.second);
}
std::string merged_tuple::stringify_at(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->stringify_at(p.second);
}
const merged_tuple::mapping_type& merged_tuple::mapping() const {
return mapping_;
}
} // namespace detail
} // namespace caf
......@@ -56,9 +56,17 @@ monitorable_actor::monitorable_actor(actor_config& cfg)
// nop
}
/// Creates a new actor instance.
monitorable_actor::monitorable_actor(actor_id aid, node_id nid)
: abstract_actor(aid, nid),
monitorable_actor::monitorable_actor(actor_system* sys,
actor_id aid,
node_id nid)
: abstract_actor(sys, aid, nid),
exit_reason_(exit_reason::not_exited) {
// nop
}
monitorable_actor::monitorable_actor(actor_system* sys, actor_id aid,
node_id nid, int init_flags)
: abstract_actor(sys, aid, nid, init_flags),
exit_reason_(exit_reason::not_exited) {
// nop
}
......
......@@ -24,16 +24,39 @@
namespace caf {
response_promise::response_promise(local_actor* self, const actor_addr& to,
const message_id& id)
: self_(self), to_(to), id_(id) {
/*
response_promise::response_promise(local_actor* self, actor_addr source,
forwarding_stack stages,
message_id id)
: self_(self),
source_(std::move(source)),
stages_(std::move(stages)),
id_(id) {
CAF_ASSERT(id.is_response() || ! id.valid());
}
*/
response_promise::response_promise(local_actor* self, mailbox_element& src)
: self_(self),
source_(std::move(src.sender)),
stages_(std::move(src.stages)),
id_(src.mid) {
src.mid.mark_as_answered();
}
void response_promise::deliver_impl(message msg) const {
if (! to_)
if (! valid())
return;
if (stages_.empty()) {
source_->enqueue(self_->address(), id_.response_id(),
std::move(msg), self_->context());
return;
to_->enqueue(self_->address(), id_, std::move(msg), self_->context());
}
auto next = std::move(stages_.back());
stages_.pop_back();
next->enqueue(mailbox_element::make(std::move(source_), id_,
std::move(stages_), std::move(msg)),
self_->context());
}
void response_promise::deliver(error x) const {
......
......@@ -376,7 +376,7 @@ CAF_TEST(typed_spawns) {
}
}
CAF_TEST(test_event_testee) {
CAF_TEST(event_testee_series) {
// run test series with event_testee
scoped_actor self{system};
auto et = self->spawn<event_testee>();
......@@ -477,7 +477,7 @@ CAF_TEST(maybe_string_delegator_chain) {
anon_send_exit(aut, exit_reason::user_shutdown);
}
CAF_TEST(test_sending_typed_actors) {
CAF_TEST(sending_typed_actors) {
scoped_actor self{system};
auto aut = system.spawn(int_fun);
self->send(self->spawn(foo), 10, aut);
......@@ -489,7 +489,7 @@ CAF_TEST(test_sending_typed_actors) {
self->send_exit(aut, exit_reason::user_shutdown);
}
CAF_TEST(test_sending_typed_actors_and_down_msg) {
CAF_TEST(sending_typed_actors_and_down_msg) {
scoped_actor self{system};
auto aut = system.spawn(int_fun2);
self->send(self->spawn(foo2), 10, aut);
......@@ -528,117 +528,169 @@ CAF_TEST(check_signature) {
);
}
using caf::detail::type_list;
namespace {
template <class X, class Y>
struct typed_actor_combine_one {
using type = void;
};
using first_stage = typed_actor<replies_to<int>::with<double, double>>;
using second_stage = typed_actor<replies_to<double, double>::with<double>>;
template <class... Xs, class... Ys, class... Zs>
struct typed_actor_combine_one<typed_mpi<type_list<Xs...>, type_list<Ys...>>,
typed_mpi<type_list<Ys...>, type_list<Zs...>>> {
using type = typed_mpi<type_list<Xs...>, type_list<Zs...>>;
first_stage::behavior_type first_stage_impl() {
return [](int i) {
return std::make_tuple(i * 2.0, i * 4.0);
};
};
template <class X, class Y>
struct typed_actor_combine_all;
second_stage::behavior_type second_stage_impl() {
return [](double x, double y) {
return x * y;
};
}
template <class X, class... Ys>
struct typed_actor_combine_all<X, type_list<Ys...>> {
using type = type_list<typename typed_actor_combine_one<X, Ys>::type...>;
};
} // namespace <anonymous>
template <class X, class Y>
struct type_actor_combine;
template <class... Xs, class... Ys>
struct type_actor_combine<typed_actor<Xs...>, typed_actor<Ys...>> {
// store Ys in a packed list
using ys = type_list<Ys...>;
// combine each X with all Ys
using all =
typename detail::tl_concat<
typename typed_actor_combine_all<Xs, ys>::type...
>::type;
// drop all mismatches (void results)
using filtered = typename detail::tl_filter_not_type<all, void>::type;
// throw error if we don't have a single match
static_assert(detail::tl_size<filtered>::value > 0,
"Left-hand actor type does not produce a single result which "
"is valid as input to the right-hand actor type.");
// compute final actor type
using type = typename detail::tl_apply<filtered, typed_actor>::type;
};
CAF_TEST(dot_composition) {
actor_system system;
auto first = system.spawn(first_stage_impl);
auto second = system.spawn(second_stage_impl);
auto first_then_second = first * second;
scoped_actor self{system};
self->request(first_then_second, 42).await(
[](double res) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
}
);
}
template <class... Xs, class... Ys>
typename type_actor_combine<typed_actor<Xs...>, typed_actor<Ys...>>::type
operator*(const typed_actor<Xs...>& x, const typed_actor<Ys...>& y) {
using res_type = typename type_actor_combine<typed_actor<Xs...>,
typed_actor<Ys...>>::type;
if (! x || ! y)
return {};
auto f = [=](event_based_actor* self) -> behavior {
self->link_to(x);
self->link_to(y);
self->trap_exit(true);
auto x_ = actor_cast<actor>(x);
auto y_ = actor_cast<actor>(y);
CAF_TEST(currying) {
using namespace std::placeholders;
auto impl = []() -> behavior {
return {
[=](const exit_msg& msg) {
// also terminate for normal exit reasons
if (msg.source == x || msg.source == y)
self->quit(msg.reason);
[](ok_atom, int x) {
return x;
},
others >> [=] {
auto rp = self->make_response_promise();
self->request(x_, self->current_message()).generic_then(
[=](message& msg) {
self->request(y_, std::move(msg)).generic_then(
[=](message& msg) {
rp.deliver(std::move(msg));
},
[=](error& err) {
rp.deliver(std::move(err));
}
);
},
[=](error& err) {
rp.deliver(std::move(err));
}
);
[](ok_atom, double x) {
return x;
}
};
};
return actor_cast<res_type>(x->home_system().spawn(f));
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut.id() == bound.id());
CAF_CHECK(aut.node() == bound.node());
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).await(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).await(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
using first_stage = typed_actor<replies_to<int>::with<double>>;
using second_stage = typed_actor<replies_to<double>::with<string>>;
first_stage::behavior_type first_stage_impl() {
return [](int i) {
return static_cast<double>(i) * 2;
CAF_TEST(type_safe_currying) {
using namespace std::placeholders;
using testee = typed_actor<replies_to<ok_atom, int>::with<int>,
replies_to<ok_atom, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](ok_atom, int x) {
return x;
},
[](ok_atom, double x) {
return x;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using curried_signature = typed_actor<replies_to<int>::with<int>,
replies_to<double>::with<double>>;
//auto bound = actor_bind(aut, ok_atom::value, _1);
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), curried_signature>::value,
"bind returned wrong actor handle");
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).await(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).await(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
second_stage::behavior_type second_stage_impl() {
return [](double x) {
return std::to_string(x);
CAF_TEST(reordering) {
auto impl = []() -> behavior {
return {
[](int x, double y) {
return x * y;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).await(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST(composition) {
CAF_TEST(type_safe_reordering) {
using testee = typed_actor<replies_to<int, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](int x, double y) {
return x * y;
}
};
};
actor_system system;
auto first = system.spawn(first_stage_impl);
auto second = system.spawn(second_stage_impl);
auto first_then_second = first * second;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
using swapped_signature = typed_actor<replies_to<double, int>::with<double>>;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), swapped_signature>::value,
"bind returned wrong actor handle");
scoped_actor self{system};
self->request(first_then_second, 42).await(
[](const string& str) {
CAF_MESSAGE("received: " << str);
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).await(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST_FIXTURE_SCOPE_END()
......
......@@ -401,7 +401,9 @@ public:
/// Called whenever a `dispatch_message` arrived for a local actor.
virtual void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) = 0;
message_id mid,
std::vector<actor_addr>& forwarding_stack,
message& msg) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
......@@ -473,6 +475,7 @@ public:
/// Returns `true` if a path to destination existed, `false` otherwise.
bool dispatch(execution_unit* ctx, const actor_addr& sender,
const std::vector<actor_addr>& forwarding_stack,
const actor_addr& receiver, message_id mid, const message& msg);
/// Returns the actor namespace associated to this BASP protocol instance.
......
......@@ -67,7 +67,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) override;
message_id mid, std::vector<actor_addr>& stages,
message& msg) override;
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
......
......@@ -69,7 +69,8 @@ protected:
SysMsgType tmp;
set_hdl(tmp, hdl_);
mailbox_elem_ptr_ = mailbox_element::make_joint(invalid_actor_addr,
invalid_message_id, tmp);
invalid_message_id,
{}, tmp);
}
Handle hdl_;
......
......@@ -64,7 +64,7 @@ void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
void abstract_broker::enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) {
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
enqueue(mailbox_element::make(sender, mid, {}, std::move(msg)), eu);
}
void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
......
......@@ -436,11 +436,13 @@ connection_state instance::handle(execution_unit* ctx,
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, payload->data(), payload->size()};
std::vector<actor_addr> forwarding_stack;
message msg;
bd & msg;
bd >> forwarding_stack >> msg;
callee_.deliver(hdr.source_node, hdr.source_actor,
hdr.dest_node, hdr.dest_actor, msg,
message_id::from_integer_value(hdr.operation_data));
hdr.dest_node, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
break;
}
case message_type::announce_proxy_instance:
......@@ -539,6 +541,7 @@ size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port,
}
bool instance::dispatch(execution_unit* ctx, const actor_addr& sender,
const std::vector<actor_addr>& forwarding_stack,
const actor_addr& receiver, message_id mid,
const message& msg) {
CAF_LOG_TRACE("");
......@@ -549,7 +552,7 @@ bool instance::dispatch(execution_unit* ctx, const actor_addr& sender,
return false;
}
auto writer = make_callback([&](serializer& sink) {
sink & msg;
sink << forwarding_stack << msg;
});
header hdr{message_type::dispatch_message, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
......
......@@ -83,7 +83,8 @@ actor_proxy_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// receive a kill_proxy_instance message
intrusive_ptr<basp_broker> ptr = static_cast<basp_broker*>(self);
auto mm = &system().middleman();
auto res = make_counted<forwarding_actor_proxy>(aid, nid, ptr);
auto res = make_counted<forwarding_actor_proxy>(self->home_system_,
aid, nid, ptr);
res->attach_functor([=](exit_reason rsn) {
mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive
......@@ -201,8 +202,9 @@ void basp_broker_state::deliver(const node_id& source_node,
actor_id source_actor,
const node_id& dest_node,
actor_id dest_actor,
message& msg,
message_id mid) {
message_id mid,
std::vector<actor_addr>& stages,
message& msg) {
CAF_LOG_TRACE(CAF_ARG(source_node)
<< CAF_ARG(source_actor) << CAF_ARG(dest_node)
<< CAF_ARG(dest_actor) << CAF_ARG(msg) << CAF_ARG(mid));
......@@ -247,7 +249,9 @@ void basp_broker_state::deliver(const node_id& source_node,
}
self->parent().notify<hook::message_received>(source_node, src,
dest->address(), mid, msg);
dest->enqueue(src, mid, std::move(msg), nullptr);
dest->enqueue(mailbox_element::make(src, mid, std::move(stages),
std::move(msg)),
nullptr);
}
void basp_broker_state::learned_new_node(const node_id& nid) {
......@@ -288,8 +292,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
using namespace detail;
system().registry().put(tmp.id(), tmp.address());
auto writer = make_callback([](serializer& sink) {
std::vector<actor_id> stages;
auto msg = make_message(sys_atom::value, get_atom::value, "info");
sink << msg;
sink << stages << msg;
});
auto path = instance.tbl().lookup(nid);
if (! path) {
......@@ -385,8 +390,9 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
auto tmp = system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), tmp.address());
auto writer = make_callback([](serializer& sink) {
auto msg = make_message(get_atom::value, "basp.default-connectivity");
sink << msg;
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, "basp.default-connectivity");
sink << stages << msg;
});
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
......@@ -468,7 +474,8 @@ behavior basp_broker::make_behavior() {
}
},
// received from proxy instances
[=](forward_atom, const actor_addr& sender, const actor_addr& receiver,
[=](forward_atom, const actor_addr& sender,
const std::vector<actor_addr>& fwd_stack, const actor_addr& receiver,
message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver)
<< CAF_ARG(mid) << CAF_ARG(msg));
......@@ -479,7 +486,8 @@ behavior basp_broker::make_behavior() {
}
if (sender && system().node() == sender.node())
system().registry().put(sender->id(), sender);
if (! state.instance.dispatch(context(), sender, receiver, mid, msg)
if (! state.instance.dispatch(context(), sender, fwd_stack,
receiver, mid, msg)
&& mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(sender, mid);
......@@ -498,7 +506,8 @@ behavior basp_broker::make_behavior() {
if (system().node() == sender.node())
system().registry().put(sender->id(), sender);
auto writer = make_callback([&](serializer& sink) {
sink << msg;
std::vector<actor_addr> stages;
sink << stages << msg;
});
auto path = this->state.instance.tbl().lookup(receiving_node);
if (! path) {
......
......@@ -49,7 +49,7 @@ void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) {
detach_from(ptr);
if (invoke_disconnect_message) {
auto mptr = mailbox_element::make(invalid_actor_addr, invalid_message_id,
detach_message());
{}, detach_message());
ptr->exec_single_event(ctx, mptr);
}
}
......
......@@ -300,6 +300,7 @@ public:
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(i),
any_vals, std::numeric_limits<actor_id>::max(),
std::vector<actor_addr>{},
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
......@@ -325,14 +326,18 @@ public:
std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::message_type::dispatch_message);
message msg;
binary_deserializer source{mpx_, buf.data(), buf.size()};
source & msg;
std::vector<actor_addr> stages;
message msg;
source >> stages >> msg;
auto src = registry_->get(hdr.source_actor).first;
auto dest = registry_->get(hdr.dest_actor).first;
CAF_REQUIRE(dest != nullptr);
dest->enqueue(src ? src->address() : invalid_actor_addr,
message_id::make(), std::move(msg), nullptr);
dest->enqueue(mailbox_element::make(src ? src->address()
: invalid_actor_addr,
message_id::make(), std::move(stages),
std::move(msg)),
nullptr);
}
class mock_t {
......@@ -474,6 +479,7 @@ CAF_TEST(client_handshake_and_dispatch) {
mock(remote_hdl(0),
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(), pseudo_remote(0)->id(), self()->id()},
std::vector<actor_addr>{},
make_message(1, 2, 3))
.expect(remote_hdl(0),
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
......@@ -562,12 +568,13 @@ CAF_TEST(remote_actor_and_send) {
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
any_vals, std::numeric_limits<actor_id>::max(),
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.expect(remote_hdl(0),
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id());
// basp broker should've send the proxy
CAF_MESSAGE("BASP broker should've send the proxy");
f.await(
[&](ok_atom, node_id nid, actor_addr res, std::set<std::string> ifs) {
auto aptr = actor_cast<abstract_actor_ptr>(res);
......@@ -583,10 +590,6 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(proxy->address() == res);
result = actor_cast<actor>(res);
}
/* TODO: update error handling ,
[&](error_atom, std::string& msg) {
throw logic_error(std::move(msg));
} */
);
CAF_MESSAGE("send message to proxy");
anon_send(actor_cast<actor>(result), 42);
......@@ -596,6 +599,7 @@ CAF_TEST(remote_actor_and_send) {
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id(),
std::vector<actor_id>{},
make_message(42));
auto msg = make_message("hi there!");
CAF_MESSAGE("send message via BASP (from proxy)");
......@@ -603,6 +607,7 @@ CAF_TEST(remote_actor_and_send) {
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
std::vector<actor_id>{},
make_message("hi there!"));
self()->receive(
[&](const string& str) {
......@@ -640,6 +645,7 @@ CAF_TEST(actor_serialize_and_deserialize) {
{basp::message_type::dispatch_message, 0, 0,
prx->node(), this_node(),
prx->id(), testee->id()},
std::vector<actor_id>{},
msg);
// testee must've responded (process forwarded message in BASP broker)
CAF_MESSAGE("wait until BASP broker writes to its output buffer");
......@@ -651,6 +657,7 @@ CAF_TEST(actor_serialize_and_deserialize) {
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), prx->node(),
testee->id(), prx->id(),
std::vector<actor_id>{},
msg);
}
......@@ -669,6 +676,7 @@ CAF_TEST(indirect_connections) {
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
std::vector<actor_id>{},
make_message("hello from jupiter!"))
// this asks Jupiter if it has a 'SpawnServ'
.expect(remote_hdl(1),
......@@ -676,6 +684,7 @@ CAF_TEST(indirect_connections) {
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
any_vals, std::numeric_limits<actor_id>::max(),
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
// this tells Jupiter that Earth learned the address of one its actors
.expect(remote_hdl(1),
......@@ -696,6 +705,7 @@ CAF_TEST(indirect_connections) {
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
}
......@@ -717,17 +727,20 @@ CAF_TEST(automatic_connection) {
// connect to mars
connect_node(1, ax, self()->id());
CAF_CHECK_EQUAL(tbl().lookup_direct(remote_node(1)).id(), remote_hdl(1).id());
// now, an actor from jupiter sends a message to us via mars
CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars");
mock(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
std::vector<actor_id>{},
make_message("hello from jupiter!"))
.expect(remote_hdl(1),
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
any_vals, std::numeric_limits<actor_id>::max(),
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.expect(remote_hdl(1),
basp::message_type::dispatch_message, any_vals,
......@@ -735,6 +748,7 @@ CAF_TEST(automatic_connection) {
this_node(), remote_node(0),
any_vals, // actor ID of an actor spawned by the BASP broker
std::numeric_limits<actor_id>::max(),
std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity"))
.expect(remote_hdl(1),
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
......@@ -753,6 +767,7 @@ CAF_TEST(automatic_connection) {
{basp::message_type::dispatch_message, 0, 0,
this_node(), this_node(),
invalid_actor_id, connection_helper},
std::vector<actor_id>{},
make_message(ok_atom::value, "basp.default-connectivity",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
......@@ -790,6 +805,7 @@ CAF_TEST(automatic_connection) {
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
CAF_CHECK(mpx()->output_buffer(remote_hdl(1)).size() == 0);
}
......
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