Commit 8a2be934 authored by Dominik Charousset's avatar Dominik Charousset

Clearly separate scheduled and blocking actors

In detail:
- move code from local_actor to specific impl types
- move special-purpose handlers to scheduled_actor
- streamline interface of scheduled_actor
- implement catch-all handlers for blocking actors
- re-implement receive* functions with new catch-all
parent 94f186a5
...@@ -26,7 +26,6 @@ ...@@ -26,7 +26,6 @@
#include <string> #include <string>
#include <vector> #include <vector>
#include <cstdint> #include <cstdint>
#include <exception>
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -128,8 +127,8 @@ public: ...@@ -128,8 +127,8 @@ public:
template <class... Ts> template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender, void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) { execution_unit* ctx, Ts&&... xs) {
enqueue(mailbox_element::make(std::move(sender), mid, enqueue(make_mailbox_element(std::move(sender), mid,
{}, std::forward<Ts>(xs)...), {}, std::forward<Ts>(xs)...),
ctx); ctx);
} }
......
...@@ -108,7 +108,7 @@ protected: ...@@ -108,7 +108,7 @@ protected:
private: private:
bool filter(upgrade_lock<detail::shared_spinlock>&, bool filter(upgrade_lock<detail::shared_spinlock>&,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
const message& content, execution_unit* host); const type_erased_tuple& content, execution_unit* host);
// call without workers_mtx_ held // call without workers_mtx_ held
void quit(execution_unit* host); void quit(execution_unit* host);
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include <memory> #include <memory>
#include <cstdint> #include <cstdint>
#include <typeinfo> #include <typeinfo>
#include <exception>
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
...@@ -68,13 +67,6 @@ public: ...@@ -68,13 +67,6 @@ public:
virtual ~attachable(); virtual ~attachable();
/// Executed if the actor did not handle an exception and must
/// not return `none` if this attachable did handle `eptr`.
/// Note that the first handler to handle `eptr` "wins" and no other
/// handler will be invoked.
/// @returns The exit reason the actor should use.
virtual optional<exit_reason> handle_exception(const std::exception_ptr& eptr);
/// Executed if the actor finished execution with given `reason`. /// Executed if the actor finished execution with given `reason`.
/// The default implementation does nothing. /// The default implementation does nothing.
/// @warning `host` can be `nullptr` /// @warning `host` can be `nullptr`
......
...@@ -101,6 +101,10 @@ public: ...@@ -101,6 +101,10 @@ public:
return impl_ ? impl_->invoke(xs) : none; return impl_ ? impl_->invoke(xs) : none;
} }
inline optional<message> operator()(type_erased_tuple& xs) {
return impl_ ? impl_->invoke(xs) : none;
}
/// Runs this handler with callback. /// Runs this handler with callback.
inline match_case::result operator()(detail::invoke_result_visitor& f, inline match_case::result operator()(detail::invoke_result_visitor& f,
type_erased_tuple& xs) { type_erased_tuple& xs) {
......
...@@ -64,13 +64,18 @@ class blocking_actor ...@@ -64,13 +64,18 @@ class blocking_actor
public dynamically_typed_actor_base { public dynamically_typed_actor_base {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Direct base type.
using super = extend<local_actor, blocking_actor>:: using super = extend<local_actor, blocking_actor>::
with<mixin::requester, mixin::sender>; with<mixin::requester, mixin::sender>;
/// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point; using timeout_type = std::chrono::high_resolution_clock::time_point;
/// Supported behavior type.
using behavior_type = behavior; using behavior_type = behavior;
/// Declared message passing interface.
using signatures = none_t; using signatures = none_t;
// -- nested classes --------------------------------------------------------- // -- nested classes ---------------------------------------------------------
...@@ -176,11 +181,13 @@ public: ...@@ -176,11 +181,13 @@ public:
~blocking_actor(); ~blocking_actor();
// -- overridden modifiers of abstract_actor --------------------------------- // -- overridden functions of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr, execution_unit*) override; void enqueue(mailbox_element_ptr, execution_unit*) override;
// -- overridden modifiers of local_actor ------------------------------------ // -- overridden functions of local_actor ------------------------------------
const char* name() const override;
void launch(execution_unit* eu, bool lazy, bool hide) override; void launch(execution_unit* eu, bool lazy, bool hide) override;
......
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
#ifndef CAF_DETAIL_DISPOSER_HPP #ifndef CAF_DETAIL_DISPOSER_HPP
#define CAF_DETAIL_DISPOSER_HPP #define CAF_DETAIL_DISPOSER_HPP
#include <type_traits>
#include "caf/memory_managed.hpp" #include "caf/memory_managed.hpp"
namespace caf { namespace caf {
...@@ -30,6 +32,12 @@ public: ...@@ -30,6 +32,12 @@ public:
inline void operator()(memory_managed* ptr) const { inline void operator()(memory_managed* ptr) const {
ptr->request_deletion(false); ptr->request_deletion(false);
} }
template <class T>
typename std::enable_if<! std::is_base_of<memory_managed, T>::value>::type
operator()(T* ptr) const {
delete ptr;
}
}; };
} // namespace detail } // namespace detail
......
...@@ -32,6 +32,11 @@ ...@@ -32,6 +32,11 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
/// Describes a partitioned list of elements. The first part
/// of the list is described by the iterator pair `[begin, separator)`.
/// The second part by `[continuation, end)`. Actors use the second half
/// of the list to store previously skipped elements. Priority-aware actors
/// also use the first half of the list to sort messages by priority.
template <class T, class Delete = std::default_delete<T>> template <class T, class Delete = std::default_delete<T>>
class intrusive_partitioned_list { class intrusive_partitioned_list {
public: public:
...@@ -86,6 +91,10 @@ public: ...@@ -86,6 +91,10 @@ public:
bool operator!=(const iterator& other) const { bool operator!=(const iterator& other) const {
return ptr != other.ptr; return ptr != other.ptr;
} }
iterator next() const {
return ptr->next;
}
}; };
intrusive_partitioned_list() { intrusive_partitioned_list() {
...@@ -99,43 +108,57 @@ public: ...@@ -99,43 +108,57 @@ public:
clear(); clear();
} }
void clear() { iterator begin() {
while (! first_empty()) {
erase(first_begin());
}
while (! second_empty()) {
erase(second_begin());
}
}
template <class F>
void clear(F f) {
while (! first_empty()) {
f(first_front());
erase(first_begin());
}
while (! second_empty()) {
f(second_front());
erase(second_begin());
}
}
iterator first_begin() {
return head_.next; return head_.next;
} }
iterator first_end() { iterator separator() {
return &separator_; return &separator_;
} }
iterator second_begin() { iterator continuation() {
return separator_.next; return separator_.next;
} }
iterator second_end() { iterator end() {
return &tail_; return &tail_;
} }
using range = std::pair<iterator, iterator>;
/// Returns the two iterator pairs describing the first and second part
/// of the partitioned list.
std::array<range, 2> ranges() {
return {{range{begin(), separator()}, range{continuation(), end()}}};
}
template <class F>
void clear(F f) {
for (auto& range : ranges()) {
auto i = range.first;
auto e = range.second;
while (i != e) {
auto ptr = i.ptr;
++i;
f(*ptr);
delete_(ptr);
}
}
if (head_.next != &separator_) {
head_.next = &separator_;
separator_.prev = &head_;
}
if (separator_.next != &tail_) {
separator_.next = &tail_;
tail_.prev = &separator_;
}
}
void clear() {
auto nop = [](value_type&) {};
clear(nop);
}
iterator insert(iterator next, pointer val) { iterator insert(iterator next, pointer val) {
auto prev = next->prev; auto prev = next->prev;
val->prev = prev; val->prev = prev;
...@@ -145,16 +168,8 @@ public: ...@@ -145,16 +168,8 @@ public:
return val; return val;
} }
bool first_empty() { bool empty() const {
return first_begin() == first_end(); return head_.next == &separator_ && separator_.next == &tail_;
}
bool second_empty() {
return second_begin() == second_end();
}
bool empty() {
return first_empty() && second_empty();
} }
pointer take(iterator pos) { pointer take(iterator pos) {
...@@ -172,96 +187,15 @@ public: ...@@ -172,96 +187,15 @@ public:
return next; return next;
} }
pointer take_first_front() { size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
return take(first_begin());
}
pointer take_second_front() {
return take(second_begin());
}
value_type& first_front() {
return *first_begin();
}
value_type& second_front() {
return *second_begin();
}
void pop_first_front() {
erase(first_begin());
}
void pop_second_front() {
erase(second_begin());
}
void push_first_back(pointer val) {
insert(first_end(), val);
}
void push_second_back(pointer val) {
insert(second_end(), val);
}
template <class Actor>
bool invoke(Actor* self, iterator first, iterator last,
behavior& bhvr, message_id mid) {
pointer prev = first->prev;
pointer next = first->next;
auto move_on = [&](bool first_valid) {
if (first_valid) {
prev = first.ptr;
}
first = next;
next = first->next;
};
while (first != last) {
std::unique_ptr<value_type, deleter_type> tmp{first.ptr};
// since this function can be called recursively during
// self->invoke_message(tmp, xs...), we have to remove the
// element from the list proactively and put it back in if
// it's safe, i.e., if invoke_message returned im_skipped
prev->next = next;
next->prev = prev;
switch (self->invoke_message(tmp, bhvr, mid)) {
case im_dropped:
move_on(false);
break;
case im_success:
return true;
case im_skipped:
if (tmp) {
// re-integrate tmp and move on
prev->next = tmp.get();
next->prev = tmp.release();
move_on(true);
}
else {
// only happens if the user does something
// really, really stupid; check it nonetheless
move_on(false);
}
break;
}
}
return false;
}
size_t count(iterator first, iterator last, size_t max_count) {
size_t result = 0; size_t result = 0;
while (first != last && result < max_count) { for (auto& range : ranges())
++first; for (auto i = range.first; i != range.second; ++i)
++result; if (++result == max_count)
} return max_count;
return result; return result;
} }
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
auto r1 = count(first_begin(), first_end(), max_count);
return r1 + count(second_begin(), second_end(), max_count - r1);
}
private: private:
value_type head_; value_type head_;
value_type separator_; value_type separator_;
......
...@@ -106,6 +106,12 @@ public: ...@@ -106,6 +106,12 @@ public:
return *get_unshared(); return *get_unshared();
} }
/// Returns the raw pointer. Callers are responsible for unsharing
/// the content if necessary.
inline message_data* raw_ptr() {
return ptr_.get();
}
// -- observers ------------------------------------------------------------ // -- observers ------------------------------------------------------------
inline const message_data* operator->() const noexcept { inline const message_data* operator->() 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_DETAIL_PAIR_STORAGE_HPP
#define CAF_DETAIL_PAIR_STORAGE_HPP
#include "caf/extend.hpp"
#include "caf/ref_counted.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace caf {
namespace detail {
// Reduces memory allocations by placing two independent types on one
// memory block. Typical use case is to combine the content of a message
// (tuple_vals) with its "context" (message ID and sender; mailbox_element).
//
// pair_storage<mailbox_element, tuple_vals<Ts...>>:
//
// +-----------------------------------------------+
// | |
// | +------------+ |
// | | | intrusive_ptr | intrusive_ptr
// v v | |
// +------------+-------------------+---------------------+
// | refcount | mailbox_element | tuple_vals<Ts...> |
// +------------+-------------------+---------------------+
// ^ ^
// | |
// unique_ptr<mailbox_element, |
// detail::disposer> |
// |
// |
// intrusive_ptr<message_data>
template <class FirstType, class SecondType>
class pair_storage {
public:
union { embedded<FirstType> first; };
union { embedded<SecondType> second; };
template <class... Ts>
pair_storage(intrusive_ptr<ref_counted> storage,
std::integral_constant<size_t, 0>, Ts&&... xs)
: first(storage),
second(std::move(storage), std::forward<Ts>(xs)...) {
// nop
}
template <class T, class... Ts>
pair_storage(intrusive_ptr<ref_counted> storage,
std::integral_constant<size_t, 1>, T&& x, Ts&&... xs)
: first(storage, std::forward<T>(x)),
second(std::move(storage), std::forward<Ts>(xs)...) {
// nop
}
template <class T0, class T1, class... Ts>
pair_storage(intrusive_ptr<ref_counted> storage,
std::integral_constant<size_t, 2>, T0&& x0, T1&& x1, Ts&&... xs)
: first(storage, std::forward<T0>(x0), std::forward<T1>(x1)),
second(std::move(storage), std::forward<Ts>(xs)...) {
// 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
}
static constexpr auto memory_cache_flag = provides_embedding;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_PAIR_STORAGE_HPP
...@@ -22,8 +22,9 @@ ...@@ -22,8 +22,9 @@
#include <cstddef> #include <cstddef>
#include "caf/config.hpp"
#include "caf/param.hpp" #include "caf/param.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
......
...@@ -49,15 +49,15 @@ public: ...@@ -49,15 +49,15 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
auto f = [=](local_actor*, const type_erased_tuple* tup) -> result<message> { auto f = [=](scheduled_actor*, const type_erased_tuple& tup) -> result<message> {
auto msg = message::from(tup); auto msg = message::from(&tup);
auto rp = this->make_response_promise(); auto rp = this->make_response_promise();
split_(workset_, msg); split_(workset_, msg);
for (auto& x : workset_) for (auto& x : workset_)
this->send(x.first, std::move(x.second)); this->send(x.first, std::move(x.second));
auto g = [=](local_actor*, const type_erased_tuple* x) mutable auto g = [=](scheduled_actor*, const type_erased_tuple& x) mutable
-> result<message> { -> result<message> {
auto res = message::from(x); auto res = message::from(&x);
join_(value_, res); join_(value_, res);
if (--awaited_results_ == 0) { if (--awaited_results_ == 0) {
rp.deliver(value_); rp.deliver(value_);
......
...@@ -25,9 +25,8 @@ ...@@ -25,9 +25,8 @@
#include <typeinfo> #include <typeinfo>
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/message.hpp"
#include "caf/type_nr.hpp" #include "caf/type_nr.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/pseudo_tuple.hpp" #include "caf/detail/pseudo_tuple.hpp"
......
...@@ -129,8 +129,8 @@ struct tuple_vals_type_helper<T, 0> { ...@@ -129,8 +129,8 @@ struct tuple_vals_type_helper<T, 0> {
} }
}; };
template <class... Ts> template <class Base, class... Ts>
class tuple_vals : public message_data { class tuple_vals_impl : public Base {
public: public:
static_assert(sizeof...(Ts) > 0, "tuple_vals is not allowed to be empty"); static_assert(sizeof...(Ts) > 0, "tuple_vals is not allowed to be empty");
...@@ -140,10 +140,10 @@ public: ...@@ -140,10 +140,10 @@ public:
using data_type = std::tuple<Ts...>; using data_type = std::tuple<Ts...>;
tuple_vals(const tuple_vals&) = default; tuple_vals_impl(const tuple_vals_impl&) = default;
template <class... Us> template <class... Us>
tuple_vals(Us&&... xs) tuple_vals_impl(Us&&... xs)
: data_(std::forward<Us>(xs)...), : data_(std::forward<Us>(xs)...),
types_{{tuple_vals_type_helper<Ts>::get()...}} { types_{{tuple_vals_type_helper<Ts>::get()...}} {
// nop // nop
...@@ -161,10 +161,6 @@ public: ...@@ -161,10 +161,6 @@ public:
return sizeof...(Ts); return sizeof...(Ts);
} }
message_data::cow_ptr copy() const override {
return message_data::cow_ptr(new tuple_vals(*this), false);
}
const void* get(size_t pos) const noexcept override { const void* get(size_t pos) const noexcept override {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
return tup_ptr_access<0, sizeof...(Ts)>::get(pos, data_); return tup_ptr_access<0, sizeof...(Ts)>::get(pos, data_);
...@@ -179,6 +175,8 @@ public: ...@@ -179,6 +175,8 @@ public:
return tup_ptr_access<0, sizeof...(Ts)>::stringify(pos, data_); return tup_ptr_access<0, sizeof...(Ts)>::stringify(pos, data_);
} }
using Base::copy;
type_erased_value_ptr copy(size_t pos) const override { type_erased_value_ptr copy(size_t pos) const override {
return tup_ptr_access<0, sizeof...(Ts)>::copy(pos, data_); return tup_ptr_access<0, sizeof...(Ts)>::copy(pos, data_);
} }
...@@ -207,6 +205,22 @@ private: ...@@ -207,6 +205,22 @@ private:
std::array<rtti_pair, sizeof...(Ts)> types_; std::array<rtti_pair, sizeof...(Ts)> types_;
}; };
template <class... Ts>
class tuple_vals : public tuple_vals_impl<message_data, Ts...> {
public:
static_assert(sizeof...(Ts) > 0, "tuple_vals is not allowed to be empty");
using super = tuple_vals_impl<message_data, Ts...>;
using super::super;
using super::copy;
message_data::cow_ptr copy() const override {
return message_data::cow_ptr(new tuple_vals(*this), false);
}
};
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
This diff is collapsed.
...@@ -26,94 +26,132 @@ ...@@ -26,94 +26,132 @@
#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/type_erased_tuple.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/detail/memory.hpp"
#include "caf/detail/embedded.hpp" #include "caf/detail/embedded.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp" #include "caf/detail/tuple_vals.hpp"
#include "caf/detail/pair_storage.hpp"
#include "caf/detail/message_data.hpp" #include "caf/detail/message_data.hpp"
#include "caf/detail/memory_cache_flag_type.hpp" #include "caf/detail/memory_cache_flag_type.hpp"
namespace caf { namespace caf {
class mailbox_element : public memory_managed { class mailbox_element {
public: public:
static constexpr auto memory_cache_flag = detail::needs_embedding;
using forwarding_stack = std::vector<strong_actor_ptr>; using forwarding_stack = std::vector<strong_actor_ptr>;
// intrusive pointer to the next mailbox element /// Intrusive pointer to the next mailbox element.
mailbox_element* next; mailbox_element* next;
// intrusive pointer to the previous mailbox element
/// Intrusive pointer to the previous mailbox element.
mailbox_element* prev; mailbox_element* prev;
// avoid multi-processing in blocking actors via flagging
/// Avoids multi-processing in blocking actors via flagging.
bool marked; bool marked;
// source of this message and receiver of the final response
/// Source of this message and receiver of the final response.
strong_actor_ptr sender; strong_actor_ptr sender;
// denotes whether this a sync or async message
/// Denotes whether this an asynchronous message or a request.
message_id mid; message_id mid;
// stages.back() is the next actor in the forwarding chain,
// if this is empty then the original sender receives the response /// `stages.back()` is the next actor in the forwarding chain,
/// if this is empty then the original sender receives the response.
forwarding_stack stages; forwarding_stack stages;
// content of this element
message msg;
mailbox_element(); mailbox_element();
mailbox_element(strong_actor_ptr sender, message_id id, mailbox_element(strong_actor_ptr&& sender, message_id id,
forwarding_stack stages); forwarding_stack&& stages);
mailbox_element(strong_actor_ptr sender, message_id id, virtual ~mailbox_element();
forwarding_stack stages, message data);
~mailbox_element(); virtual type_erased_tuple& content();
const type_erased_tuple& content() const;
mailbox_element(mailbox_element&&) = delete; mailbox_element(mailbox_element&&) = delete;
mailbox_element(const mailbox_element&) = delete; mailbox_element(const mailbox_element&) = delete;
mailbox_element& operator=(mailbox_element&&) = delete; mailbox_element& operator=(mailbox_element&&) = delete;
mailbox_element& operator=(const mailbox_element&) = delete; mailbox_element& operator=(const mailbox_element&) = delete;
using unique_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
static unique_ptr make(strong_actor_ptr sender, message_id id,
forwarding_stack stages, message msg);
template <class T, class... Ts>
static typename std::enable_if<
! 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 =
detail::tuple_vals<
typename unbox_message_element<
typename detail::strip_and_convert<T>::type
>::type,
typename unbox_message_element<
typename detail::strip_and_convert<Ts>::type
>::type...
>;
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<T>(x),
std::forward<Ts>(xs)...);
ptr->first.msg.reset(&(ptr->second), false);
return unique_ptr{&(ptr->first)};
}
inline bool is_high_priority() const { inline bool is_high_priority() const {
return mid.is_high_priority(); return mid.is_high_priority();
} }
protected:
empty_type_erased_tuple dummy_;
}; };
/// Encapsulates arbitrary data in a message element.
template <class... Ts>
class mailbox_element_vals
: public mailbox_element,
public detail::tuple_vals_impl<type_erased_tuple, Ts...> {
public:
template <class... Us>
mailbox_element_vals(strong_actor_ptr&& sender, message_id id,
forwarding_stack&& stages, Us&&... xs)
: mailbox_element(std::move(sender), id, std::move(stages)),
detail::tuple_vals_impl<type_erased_tuple, Ts...>(std::forward<Us>(xs)...) {
// nop
}
type_erased_tuple& content() {
return *this;
}
};
/// Provides a view for treating arbitrary data as message element.
template <class... Ts>
class mailbox_element_view : public mailbox_element,
public type_erased_tuple_view<Ts...> {
public:
mailbox_element_view(strong_actor_ptr&& sender, message_id id,
forwarding_stack&& stages, Ts&... xs)
: mailbox_element(std::move(sender), id, std::move(stages)),
type_erased_tuple_view<Ts...>(xs...) {
// nop
}
type_erased_tuple& content() {
return *this;
}
};
/// @relates mailbox_element
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
/// @relates mailbox_element
mailbox_element_ptr
make_mailbox_element(strong_actor_ptr sender, message_id id,
mailbox_element::forwarding_stack stages, message msg);
/// @relates mailbox_element
template <class T, class... Ts>
typename std::enable_if<
! std::is_same<typename std::decay<T>::type, message>::value
|| (sizeof...(Ts) > 0),
mailbox_element_ptr
>::type
make_mailbox_element(strong_actor_ptr sender, message_id id,
mailbox_element::forwarding_stack stages,
T&& x, Ts&&... xs) {
using impl =
mailbox_element_vals<
typename unbox_message_element<
typename detail::strip_and_convert<T>::type
>::type,
typename unbox_message_element<
typename detail::strip_and_convert<Ts>::type
>::type...
>;
auto ptr = new impl(std::move(sender), id, std::move(stages),
std::forward<T>(x), std::forward<Ts>(xs)...);
return mailbox_element_ptr{ptr};
}
std::string to_string(const mailbox_element&); std::string to_string(const mailbox_element&);
} // namespace caf } // namespace caf
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/detail/memory.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
...@@ -34,24 +33,7 @@ namespace caf { ...@@ -34,24 +33,7 @@ namespace caf {
/// @relates ref_counted /// @relates ref_counted
/// @relates intrusive_ptr /// @relates intrusive_ptr
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if< intrusive_ptr<T> make_counted(Ts&&... xs) {
detail::is_memory_cached<T>::value,
intrusive_ptr<T>
>::type
make_counted(Ts&&... xs) {
return intrusive_ptr<T>(detail::memory::create<T>(std::forward<Ts>(xs)...),
false);
}
/// Constructs an object of type `T` in an `intrusive_ptr`.
/// @relates ref_counted
/// @relates intrusive_ptr
template <class T, class... Ts>
typename std::enable_if<
! detail::is_memory_cached<T>::value,
intrusive_ptr<T>
>::type
make_counted(Ts&&... xs) {
return intrusive_ptr<T>(new T(std::forward<Ts>(xs)...), false); return intrusive_ptr<T>(new T(std::forward<Ts>(xs)...), false);
} }
......
...@@ -101,7 +101,7 @@ public: ...@@ -101,7 +101,7 @@ public:
/// Returns the value at position `p` as mutable reference of type `T`. /// Returns the value at position `p` as mutable reference of type `T`.
template <class T> template <class T>
T& get_as_mutable(size_t p) { T& get_mutable_as(size_t p) {
CAF_ASSERT(match_element(p, type_nr<T>::value, &typeid(T))); CAF_ASSERT(match_element(p, type_nr<T>::value, &typeid(T)));
return *reinterpret_cast<T*>(get_mutable(p)); return *reinterpret_cast<T*>(get_mutable(p));
} }
...@@ -115,8 +115,8 @@ public: ...@@ -115,8 +115,8 @@ public:
vals_.unshare(); vals_.unshare();
} }
/// Returns a mutable reference to the content. Causes the message /// Returns a mutable reference to the content. Callers are responsible
/// to unshare its content if necessary. /// for unsharing content if necessary.
inline data_ptr& vals() { inline data_ptr& vals() {
return vals_; return vals_;
} }
......
...@@ -76,7 +76,7 @@ public: ...@@ -76,7 +76,7 @@ public:
auto req_id = dptr()->new_request_id(P); auto req_id = dptr()->new_request_id(P);
dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(), dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
dptr()->request_sync_timeout_msg(timeout, req_id); dptr()->request_response_timeout(timeout, req_id);
return {req_id.response_id(), dptr()}; return {req_id.response_id(), dptr()};
} }
......
...@@ -27,10 +27,10 @@ ...@@ -27,10 +27,10 @@
#include <string> #include <string>
#include <vector> #include <vector>
#include <cstdint> #include <cstdint>
#include <exception>
#include <type_traits> #include <type_traits>
#include <condition_variable> #include <condition_variable>
#include "caf/type_nr.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
...@@ -129,9 +129,6 @@ protected: ...@@ -129,9 +129,6 @@ protected:
bool remove_backlink_impl(abstract_actor* other); bool remove_backlink_impl(abstract_actor* other);
// tries to run a custom exception handler for `eptr`
optional<exit_reason> handle(const std::exception_ptr& eptr);
// precondition: `mtx_` is acquired // precondition: `mtx_` is acquired
inline void attach_impl(attachable_ptr& ptr) { inline void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(attachables_head_); ptr->next.swap(attachables_head_);
...@@ -154,9 +151,14 @@ protected: ...@@ -154,9 +151,14 @@ protected:
template <class F> template <class F>
bool handle_system_message(mailbox_element& x, execution_unit* context, bool handle_system_message(mailbox_element& x, execution_unit* context,
bool trap_exit, F& down_msg_handler) { bool trap_exit, F& down_msg_handler) {
auto& msg = x.msg; auto& content = x.content();
if (msg.size() == 1 && msg.match_element<down_msg>(0)) { if (content.type_token() == make_type_token<down_msg>()) {
down_msg_handler(msg.get_as_mutable<down_msg>(0)); if (content.shared()) {
auto vptr = content.copy(0);
down_msg_handler(vptr->get_mutable_as<down_msg>());
} else {
down_msg_handler(content.get_mutable_as<down_msg>(0));
}
return true; return true;
} }
return handle_system_message(x, context, trap_exit); return handle_system_message(x, context, trap_exit);
......
...@@ -228,6 +228,34 @@ class optional<T&> { ...@@ -228,6 +228,34 @@ class optional<T&> {
T* m_value; T* m_value;
}; };
template <>
class optional<void> {
public:
optional(none_t = none) : m_value(false) {
// nop
}
optional(unit_t) : m_value(true) {
// nop
}
optional(const optional& other) = default;
optional& operator=(const optional& other) = default;
explicit operator bool() const {
return m_value;
}
bool operator!() const {
return ! m_value;
}
private:
bool m_value;
};
/// @relates optional /// @relates optional
template <class T> template <class T>
std::string to_string(const optional<T>& x) { std::string to_string(const optional<T>& x) {
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include <type_traits> #include <type_traits>
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/type_erased_tuple.hpp"
namespace caf { namespace caf {
......
...@@ -88,7 +88,7 @@ private: ...@@ -88,7 +88,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_awaited_response_handler(mid_, message_handler{std::move(f)}); self_->add_awaited_response_handler(mid_, message_handler{std::move(f)});
} }
template <class F, class OnError> template <class F, class OnError>
...@@ -100,7 +100,7 @@ private: ...@@ -100,7 +100,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_awaited_response_handler(mid_, behavior{std::move(f), self_->add_awaited_response_handler(mid_, behavior{std::move(f),
std::move(ef)}); std::move(ef)});
} }
...@@ -113,7 +113,7 @@ private: ...@@ -113,7 +113,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_multiplexed_response_handler(mid_, behavior{std::move(f)}); self_->add_multiplexed_response_handler(mid_, behavior{std::move(f)});
} }
template <class F, class OnError> template <class F, class OnError>
...@@ -125,7 +125,7 @@ private: ...@@ -125,7 +125,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_multiplexed_response_handler(mid_, behavior{std::move(f), self_->add_multiplexed_response_handler(mid_, behavior{std::move(f),
std::move(ef)}); std::move(ef)});
} }
......
This diff is collapsed.
...@@ -32,8 +32,8 @@ namespace caf { ...@@ -32,8 +32,8 @@ namespace caf {
/// skipping to the runtime. /// skipping to the runtime.
class skip_t { class skip_t {
public: public:
using fun = std::function<result<message>(local_actor* self, using fun = std::function<result<message>(scheduled_actor* self,
const type_erased_tuple*)>; const type_erased_tuple&)>;
constexpr skip_t() { constexpr skip_t() {
// nop // nop
...@@ -46,7 +46,8 @@ public: ...@@ -46,7 +46,8 @@ public:
operator fun() const; operator fun() const;
private: private:
static result<message> skip_fun_impl(local_actor*, const type_erased_tuple*); static result<message> skip_fun_impl(scheduled_actor*,
const type_erased_tuple&);
}; };
/// Tells the runtime system to skip a message when used as message /// Tells the runtime system to skip a message when used as message
......
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_TYPE_ERASED_COLLECTION_HPP #ifndef CAF_TYPE_ERASED_TUPLE_HPP
#define CAF_TYPE_ERASED_COLLECTION_HPP #define CAF_TYPE_ERASED_TUPLE_HPP
#include <tuple> #include <tuple>
#include <cstddef> #include <cstddef>
...@@ -27,8 +27,13 @@ ...@@ -27,8 +27,13 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/type_nr.hpp" #include "caf/type_nr.hpp"
#include "caf/optional.hpp"
#include "caf/type_erased_value.hpp" #include "caf/type_erased_value.hpp"
#include "caf/detail/try_match.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/pseudo_tuple.hpp"
namespace caf { namespace caf {
/// Represents a tuple of type-erased values. /// Represents a tuple of type-erased values.
...@@ -63,9 +68,6 @@ public: ...@@ -63,9 +68,6 @@ public:
/// Returns the size of this tuple. /// Returns the size of this tuple.
virtual size_t size() const noexcept = 0; virtual size_t size() const noexcept = 0;
/// Returns whether multiple references to this tuple exist.
virtual bool shared() const noexcept = 0;
/// Returns a type hint for the element types. /// Returns a type hint for the element types.
virtual uint32_t type_token() const noexcept = 0; virtual uint32_t type_token() const noexcept = 0;
...@@ -87,6 +89,11 @@ public: ...@@ -87,6 +89,11 @@ public:
// -- observers -------------------------------------------------------------- // -- observers --------------------------------------------------------------
/// Returns whether multiple references to this tuple exist.
/// The default implementation returns false.
virtual bool shared() const noexcept;
/// Returns `size() == 0`.
bool empty() const; bool empty() const;
/// Returns a string representation of the tuple. /// Returns a string representation of the tuple.
...@@ -99,7 +106,7 @@ public: ...@@ -99,7 +106,7 @@ public:
/// matches type number `n` and run-time type information `p`. /// matches type number `n` and run-time type information `p`.
bool matches(size_t pos, uint16_t n, const std::type_info* p) const noexcept; bool matches(size_t pos, uint16_t n, const std::type_info* p) const noexcept;
// -- inline observers ------------------------------------------------------- // -- convenience functions --------------------------------------------------
/// Returns the type number for the element at position `pos`. /// Returns the type number for the element at position `pos`.
inline uint16_t type_nr(size_t pos) const noexcept { inline uint16_t type_nr(size_t pos) const noexcept {
...@@ -110,6 +117,98 @@ public: ...@@ -110,6 +117,98 @@ public:
inline bool matches(size_t pos, const rtti_pair& rtti) const noexcept { inline bool matches(size_t pos, const rtti_pair& rtti) const noexcept {
return matches(pos, rtti.first, rtti.second); return matches(pos, rtti.first, rtti.second);
} }
/// Convenience function for `*reinterpret_cast<const T*>(get())`.
template <class T>
const T& get_as(size_t pos) const {
return *reinterpret_cast<const T*>(get(pos));
}
/// Convenience function for `*reinterpret_cast<T*>(get_mutable())`.
template <class T>
T& get_mutable_as(size_t pos) {
return *reinterpret_cast<T*>(get_mutable(pos));
}
/// Returns `true` if the element at `pos` matches `T`.
template <class T>
bool match_element(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
auto x = detail::meta_element_factory<T>::create();
return detail::match_element(x, *this, pos);
}
/// Returns `true` if the pattern `Ts...` matches the content of this tuple.
template <class... Ts>
bool match_elements() const noexcept {
if (sizeof...(Ts) != size())
return false;
detail::meta_elements<detail::type_list<Ts...>> xs;
for (size_t i = 0; i < xs.arr.size(); ++i)
if (! detail::match_element(xs.arr[i], *this, i))
return false;
return true;
}
template <class F>
auto apply(F fun)
-> optional<typename detail::get_callable_trait<F>::result_type> {
using trait = typename detail::get_callable_trait<F>::type;
detail::type_list<typename trait::result_type> result_token;
typename trait::arg_types args_token;
return apply(fun, result_token, args_token);
}
private:
template <class F, class R, class... Ts>
optional<R> apply(F& fun, detail::type_list<R>,
detail::type_list<Ts...> tk) {
if (! match_elements<Ts...>())
return none;
detail::pseudo_tuple<typename std::decay<Ts>::type...> xs{shared()};
for (size_t i = 0; i < size(); ++i)
xs[i] = const_cast<void*>(get(i)); // pseud_tuple figures out const-ness
return detail::apply_args(fun, detail::get_indices(tk), xs);
}
template <class F, class... Ts>
optional<void> apply(F& fun, detail::type_list<void>,
detail::type_list<Ts...> tk) {
if (! match_elements<Ts...>())
return none;
detail::pseudo_tuple<typename std::decay<Ts>::type...> xs{shared()};
for (size_t i = 0; i < size(); ++i)
xs[i] = const_cast<void*>(get(i)); // pseud_tuple figures out const-ness
detail::apply_args(fun, detail::get_indices(tk), xs);
return unit;
}
};
/// @relates type_erased_tuple
/// Dummy objects representing empty tuples.
class empty_type_erased_tuple : public type_erased_tuple {
public:
empty_type_erased_tuple() = default;
~empty_type_erased_tuple();
void* get_mutable(size_t pos) override;
void load(size_t pos, deserializer& source) override;
size_t size() const noexcept override;
uint32_t type_token() const noexcept override;
rtti_pair type(size_t pos) const noexcept override;
const void* get(size_t pos) const noexcept override;
std::string stringify(size_t pos) const override;
type_erased_value_ptr copy(size_t pos) const override;
void save(size_t pos, serializer& sink) const override;
}; };
/// @relates type_erased_tuple /// @relates type_erased_tuple
...@@ -166,10 +265,6 @@ public: ...@@ -166,10 +265,6 @@ public:
return sizeof...(Ts); return sizeof...(Ts);
} }
bool shared() const noexcept override {
return false;
}
uint32_t type_token() const noexcept override { uint32_t type_token() const noexcept override {
return make_type_token<Ts...>(); return make_type_token<Ts...>();
} }
...@@ -225,4 +320,4 @@ type_erased_tuple_view<Ts...> make_type_erased_tuple_view(Ts&... xs) { ...@@ -225,4 +320,4 @@ type_erased_tuple_view<Ts...> make_type_erased_tuple_view(Ts&... xs) {
} // namespace caf } // namespace caf
#endif // CAF_TYPE_ERASED_COLLECTION_HPP #endif // CAF_TYPE_ERASED_TUPLE_HPP
...@@ -75,7 +75,7 @@ public: ...@@ -75,7 +75,7 @@ public:
/// the type nr and type info object. /// the type nr and type info object.
bool matches(uint16_t tnr, const std::type_info* tinf) const; bool matches(uint16_t tnr, const std::type_info* tinf) const;
// -- inline observers ------------------------------------------------------- // -- convenience functions --------------------------------------------------
/// Returns the type number for the stored value. /// Returns the type number for the stored value.
inline uint16_t type_nr() const { inline uint16_t type_nr() const {
...@@ -86,6 +86,18 @@ public: ...@@ -86,6 +86,18 @@ public:
inline bool matches(const rtti_pair& rtti) const { inline bool matches(const rtti_pair& rtti) const {
return matches(rtti.first, rtti.second); return matches(rtti.first, rtti.second);
} }
/// Convenience function for `reinterpret_cast<const T*>(get())`.
template <class T>
const T& get_as() const {
return *reinterpret_cast<const T*>(get());
}
/// Convenience function for `reinterpret_cast<T*>(get_mutable())`.
template <class T>
T& get_mutable_as() {
return *reinterpret_cast<T*>(get_mutable());
}
}; };
/// @relates type_erased_value_impl /// @relates type_erased_value_impl
......
...@@ -60,7 +60,7 @@ void abstract_actor::on_destroy() { ...@@ -60,7 +60,7 @@ void abstract_actor::on_destroy() {
void abstract_actor::enqueue(strong_actor_ptr sender, message_id mid, void abstract_actor::enqueue(strong_actor_ptr sender, message_id mid,
message msg, execution_unit* host) { message msg, execution_unit* host) {
enqueue(mailbox_element::make(sender, mid, {}, std::move(msg)), host); enqueue(make_mailbox_element(sender, mid, {}, std::move(msg)), host);
} }
abstract_actor::abstract_actor(actor_config& cfg) abstract_actor::abstract_actor(actor_config& cfg)
......
...@@ -124,14 +124,15 @@ public: ...@@ -124,14 +124,15 @@ public:
msg_ptr = try_dequeue(it->first); msg_ptr = try_dequeue(it->first);
} }
} }
if (msg_ptr->msg.match_element<exit_msg>(0)) { auto& content = msg_ptr->content();
auto& em = msg_ptr->msg.get_as<exit_msg>(0); if (content.type_token() == make_type_token<exit_msg>()) {
auto& em = content.get_as<exit_msg>(0);
if (em.reason) { if (em.reason) {
fail_state(em.reason); fail_state(em.reason);
return; return;
} }
} }
mfun(msg_ptr->msg); mfun(content);
msg_ptr.reset(); msg_ptr.reset();
} }
} }
......
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/decorator/adapter.hpp" #include "caf/decorator/adapter.hpp"
......
...@@ -45,8 +45,7 @@ void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -45,8 +45,7 @@ void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
void actor_companion::enqueue(strong_actor_ptr src, message_id mid, void actor_companion::enqueue(strong_actor_ptr src, message_id mid,
message content, execution_unit* eu) { message content, execution_unit* eu) {
using detail::memory; auto ptr = make_mailbox_element(std::move(src), mid, {}, std::move(content));
auto ptr = mailbox_element::make(std::move(src), mid, {}, std::move(content));
enqueue(std::move(ptr), eu); enqueue(std::move(ptr), eu);
} }
......
...@@ -41,15 +41,15 @@ actor_ostream::actor_ostream(scoped_actor& self) ...@@ -41,15 +41,15 @@ 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(nullptr, message_id::make(), {}, printer_->enqueue(make_mailbox_element(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(nullptr, message_id::make(), {}, printer_->enqueue(make_mailbox_element(nullptr, message_id::make(), {},
flush_atom::value, self_), flush_atom::value, self_),
nullptr); nullptr);
return *this; return *this;
...@@ -59,7 +59,7 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) { ...@@ -59,7 +59,7 @@ 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(nullptr, message_id::make(), {}, pr->enqueue(make_mailbox_element(nullptr, message_id::make(), {},
redirect_atom::value, self->id(), redirect_atom::value, self->id(),
std::move(fn), flags), std::move(fn), flags),
nullptr); nullptr);
...@@ -67,7 +67,7 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) { ...@@ -67,7 +67,7 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) {
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(nullptr, message_id::make(), {}, pr->enqueue(make_mailbox_element(nullptr, message_id::make(), {},
redirect_atom::value, redirect_atom::value,
std::move(fn), flags), std::move(fn), flags),
nullptr); nullptr);
......
...@@ -55,8 +55,9 @@ void broadcast_dispatch(actor_system&, actor_pool::uplock&, ...@@ -55,8 +55,9 @@ void broadcast_dispatch(actor_system&, actor_pool::uplock&,
const actor_pool::actor_vec& vec, const actor_pool::actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) { mailbox_element_ptr& ptr, execution_unit* host) {
CAF_ASSERT(!vec.empty()); CAF_ASSERT(!vec.empty());
auto msg = message::from(&ptr->content());
for (size_t i = 1; i < vec.size(); ++i) for (size_t i = 1; i < vec.size(); ++i)
vec[i]->enqueue(ptr->sender, ptr->mid, ptr->msg, host); vec[i]->enqueue(ptr->sender, ptr->mid, msg, host);
vec.front()->enqueue(std::move(ptr), host); vec.front()->enqueue(std::move(ptr), host);
} }
...@@ -125,7 +126,7 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers, ...@@ -125,7 +126,7 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
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->content(), eu))
return; return;
policy_(home_system(), guard, workers_, what, eu); policy_(home_system(), guard, workers_, what, eu);
} }
...@@ -140,20 +141,21 @@ void actor_pool::on_cleanup() { ...@@ -140,20 +141,21 @@ void actor_pool::on_cleanup() {
bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
const message& msg, execution_unit* eu) { const type_erased_tuple& msg, execution_unit* eu) {
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(msg));
if (msg.match_elements<exit_msg>()) { if (msg.match_elements<exit_msg>()) {
// acquire second mutex as well // acquire second mutex as well
std::vector<actor> workers; std::vector<actor> workers;
auto tmp = msg.get_as<exit_msg>(0).reason; auto tmp = msg.get_as<exit_msg>(0).reason;
if (cleanup(std::move(tmp), eu)) { if (cleanup(std::move(tmp), eu)) {
auto tmp = message::from(&msg);
// send exit messages *always* to all workers and clear vector afterwards // send exit messages *always* to all workers and clear vector afterwards
// but first swap workers_ out of the critical section // but first swap workers_ out of the critical section
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.swap(workers); workers_.swap(workers);
unique_guard.unlock(); unique_guard.unlock();
for (auto& w : workers) for (auto& w : workers)
anon_send(w, msg); anon_send(w, tmp);
is_registered(false); is_registered(false);
} }
return true; return true;
......
...@@ -44,12 +44,12 @@ adapter::adapter(strong_actor_ptr decorated, message msg) ...@@ -44,12 +44,12 @@ adapter::adapter(strong_actor_ptr decorated, message msg)
default_attachable::make_monitor(decorated_->get()->address(), address())); default_attachable::make_monitor(decorated_->get()->address(), address()));
} }
void adapter::enqueue(mailbox_element_ptr what, execution_unit* context) { void adapter::enqueue(mailbox_element_ptr x, execution_unit* context) {
CAF_ASSERT(what); CAF_ASSERT(x);
auto down_msg_handler = [&](down_msg& dm) { auto down_msg_handler = [&](down_msg& dm) {
cleanup(std::move(dm.reason), context); cleanup(std::move(dm.reason), context);
}; };
if (handle_system_message(*what, context, false, down_msg_handler)) if (handle_system_message(*x, context, false, down_msg_handler))
return; return;
strong_actor_ptr decorated; strong_actor_ptr decorated;
message merger; message merger;
...@@ -60,13 +60,14 @@ void adapter::enqueue(mailbox_element_ptr what, execution_unit* context) { ...@@ -60,13 +60,14 @@ void adapter::enqueue(mailbox_element_ptr what, execution_unit* context) {
fail_state = fail_state_; fail_state = fail_state_;
}); });
if (! decorated) { if (! decorated) {
bounce(what, fail_state); bounce(x, fail_state);
return; return;
} }
auto& msg = what->msg; message tmp{detail::merged_tuple::make(merger,
message tmp{detail::merged_tuple::make(merger, std::move(msg))}; message::from(&x->content()))};
msg.swap(tmp); decorated->enqueue(make_mailbox_element(std::move(x->sender), x->mid,
decorated->enqueue(std::move(what), context); std::move(x->stages), std::move(tmp)),
context);
} }
void adapter::on_cleanup() { void adapter::on_cleanup() {
......
...@@ -30,10 +30,6 @@ attachable::token::token(size_t typenr, const void* vptr) ...@@ -30,10 +30,6 @@ attachable::token::token(size_t typenr, const void* vptr)
// nop // nop
} }
optional<exit_reason> attachable::handle_exception(const std::exception_ptr&) {
return none;
}
void attachable::actor_exited(const error&, execution_unit*) { void attachable::actor_exited(const error&, execution_unit*) {
// nop // nop
} }
......
...@@ -27,9 +27,7 @@ namespace caf { ...@@ -27,9 +27,7 @@ namespace caf {
namespace detail { namespace detail {
void behavior_stack::pop_back() { void behavior_stack::pop_back() {
if (elements_.empty()) { CAF_ASSERT(! elements_.empty());
return;
}
erased_elements_.push_back(std::move(elements_.back())); erased_elements_.push_back(std::move(elements_.back()));
elements_.pop_back(); elements_.pop_back();
} }
......
...@@ -70,6 +70,10 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -70,6 +70,10 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
} }
} }
const char* blocking_actor::name() const {
return "blocking_actor";
}
void blocking_actor::launch(execution_unit*, bool, bool hide) { void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_LOG_TRACE(CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(is_blocking()); CAF_ASSERT(is_blocking());
...@@ -81,19 +85,12 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) { ...@@ -81,19 +85,12 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0); CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
auto self = static_cast<blocking_actor*>(this_ptr); auto self = static_cast<blocking_actor*>(this_ptr);
error rsn; error rsn;
std::exception_ptr eptr = nullptr;
try { try {
self->act(); self->act();
rsn = self->fail_state_; rsn = self->fail_state_;
} }
catch (...) { catch (...) {
rsn = exit_reason::unhandled_exception; rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception;
} }
try { try {
self->on_exit(); self->on_exit();
...@@ -130,72 +127,250 @@ void blocking_actor::fail_state(error err) { ...@@ -130,72 +127,250 @@ void blocking_actor::fail_state(error err) {
fail_state_ = std::move(err); fail_state_ = std::move(err);
} }
namespace {
class message_sequence {
public:
virtual bool at_end() = 0;
virtual bool await_value(bool reset_timeout) = 0;
virtual mailbox_element& value() = 0;
virtual void advance() = 0;
virtual void erase_and_advance() = 0;
};
class cached_sequence : public message_sequence {
public:
using cache_type = local_actor::mailbox_type::cache_type;
using iterator = cache_type::iterator;
cached_sequence(cache_type& cache)
: cache_(cache),
i_(cache.continuation()),
e_(cache.end()) {
// iterater to the first un-marked element
i_ = advance_impl(i_);
}
bool at_end() override {
return i_ == e_;
}
void advance() override {
CAF_ASSERT(i_->marked);
i_->marked = false;
i_ = advance_impl(i_.next());
}
void erase_and_advance() override {
CAF_ASSERT(i_->marked);
i_ = advance_impl(cache_.erase(i_));
}
bool await_value(bool) override {
return true;
}
mailbox_element& value() override {
CAF_ASSERT(i_->marked);
return *i_;
}
public:
iterator advance_impl(iterator i) {
while (i != e_) {
if (! i->marked) {
i->marked = true;
return i;
}
++i;
}
return i;
}
cache_type& cache_;
iterator i_;
iterator e_;
};
class mailbox_sequence : public message_sequence {
public:
mailbox_sequence(blocking_actor* self, duration rel_timeout)
: self_(self),
rel_tout_(rel_timeout) {
next_timeout();
}
bool at_end() override {
return false;
}
void advance() override {
if (ptr_)
self_->push_to_cache(std::move(ptr_));
next_timeout();
}
void erase_and_advance() override {
ptr_.reset();
next_timeout();
}
bool await_value(bool reset_timeout) override {
if (! rel_tout_.valid()) {
self_->await_data();
return true;
}
if (reset_timeout)
next_timeout();
return self_->await_data(abs_tout_);
}
mailbox_element& value() override {
ptr_ = self_->next_message();
CAF_ASSERT(ptr_ != nullptr);
return *ptr_;
}
private:
void next_timeout() {
abs_tout_ = std::chrono::high_resolution_clock::now();
abs_tout_ += rel_tout_;
}
blocking_actor* self_;
duration rel_tout_;
std::chrono::high_resolution_clock::time_point abs_tout_;
mailbox_element_ptr ptr_;
};
class message_sequence_combinator : public message_sequence {
public:
using pointer = message_sequence*;
message_sequence_combinator(pointer first, pointer second)
: ptr_(first),
fallback_(second) {
// nop
}
bool at_end() override {
if (ptr_->at_end()) {
if (! fallback_)
return true;
ptr_ = fallback_;
fallback_ = nullptr;
return at_end();
}
return false;
}
void advance() override {
ptr_->advance();
}
void erase_and_advance() override {
ptr_->erase_and_advance();
}
bool await_value(bool reset_timeout) override {
return ptr_->await_value(reset_timeout);
}
mailbox_element& value() override {
return ptr_->value();
}
private:
pointer ptr_;
pointer fallback_;
};
} // namespace <anonymous>
void blocking_actor::receive_impl(receive_cond& rcc, void blocking_actor::receive_impl(receive_cond& rcc,
message_id mid, message_id mid,
detail::blocking_behavior& bhvr) { detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(mid));
// calculate absolute timeout if requested by user // we start iterating the cache and iterating mailbox elements afterwards
auto rel_tout = bhvr.timeout(); cached_sequence seq1{mailbox().cache()};
std::chrono::high_resolution_clock::time_point abs_tout; mailbox_sequence seq2{this, bhvr.timeout()};
if (rel_tout.valid()) { message_sequence_combinator seq{&seq1, &seq2};
abs_tout = std::chrono::high_resolution_clock::now();
abs_tout += rel_tout;
}
// try to dequeue from cache first
if (invoke_from_cache(bhvr.nested, mid))
return;
// read incoming messages until we have a match or a timeout
detail::default_invoke_result_visitor visitor{this}; detail::default_invoke_result_visitor visitor{this};
// read incoming messages until we have a match or a timeout
for (;;) { for (;;) {
// check loop pre-condition
if (! rcc.pre()) if (! rcc.pre())
return; return;
// mailbox sequence is infinite, but at_end triggers the
// transition from seq1 to seq2 if we iterated our cache
if (seq.at_end())
CAF_RAISE_ERROR("reached the end of an infinite sequence");
// reset the timeout each iteration
if (! seq.await_value(true)) {
// short-circuit "loop body"
bhvr.nested.handle_timeout();
if (! rcc.post())
return;
continue;
}
// skip messages in the loop body until we have a match
bool skipped; bool skipped;
bool timed_out;
do { do {
skipped = false; skipped = false;
if (rel_tout.valid()) { timed_out = false;
if (! await_data(abs_tout)) { auto& x = seq.value();
bhvr.handle_timeout();
return;
}
} else {
await_data();
}
auto ptr = next_message();
CAF_ASSERT(ptr != nullptr);
// skip messages that don't match our message ID // skip messages that don't match our message ID
if (ptr->mid != mid) { if (x.mid != mid) {
push_to_cache(std::move(ptr)); skipped = true;
continue; } else {
} // blocking actors can use nested receives => restore current_element_
ptr.swap(current_element_); auto prev_element = current_element_;
switch (bhvr.nested(visitor, current_element_->msg)) { current_element_ = &x;
case match_case::skip: switch (bhvr.nested(visitor, x.content())) {
skipped = true; case match_case::skip:
break; skipped = true;
default: break;
break; default:
case match_case::no_match: { break;
auto sres = bhvr.fallback(current_element_->msg.cvals().get()); case match_case::no_match: {
// when dealing with response messages, there's either a match auto sres = bhvr.fallback(&current_element_->content());
// on the first handler or we produce an error to // when dealing with response messages, there's either a match
// get a match on the second (error) handler // on the first handler or we produce an error to
if (sres.flag != rt_skip) { // get a match on the second (error) handler
visitor.visit(sres); if (sres.flag != rt_skip) {
} else if (mid.valid()) { visitor.visit(sres);
// make new message to replace current_element_->msg } else if (mid.valid()) {
auto x = make_message(make_error(sec::unexpected_response, // invoke again with an unexpected_response error
std::move(current_element_->msg))); auto& old = *current_element_;
current_element_->msg = std::move(x); auto err = make_error(sec::unexpected_response,
bhvr.nested(current_element_->msg); message::from(&old.content()));
} else { mailbox_element_view<error> tmp{std::move(old.sender), old.mid,
skipped = true; std::move(old.stages), err};
current_element_ = &tmp;
bhvr.nested(tmp.content());
} else {
skipped = true;
}
} }
} }
current_element_ = prev_element;
}
if (skipped) {
seq.advance();
if (seq.at_end())
CAF_RAISE_ERROR("reached the end of an infinite sequence");
if (! seq.await_value(false)) {
timed_out = true;
}
} }
ptr.swap(current_element_); } while (skipped && ! timed_out);
if (skipped) if (timed_out)
push_to_cache(std::move(ptr)); bhvr.nested.handle_timeout();
} while (skipped); else
seq.erase_and_advance();
// check loop post condition
if (! rcc.post()) if (! rcc.post())
return; return;
} }
......
...@@ -68,7 +68,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what, ...@@ -68,7 +68,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what); CAF_ASSERT(what);
forward_msg(std::move(what->sender), what->mid, forward_msg(std::move(what->sender), what->mid,
std::move(what->msg), &what->stages); message::from(&what->content()), &what->stages);
} }
......
...@@ -156,8 +156,8 @@ public: ...@@ -156,8 +156,8 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](local_actor*, const type_erased_tuple* ptr) -> result<message> { auto fwd = [=](scheduled_actor*, const type_erased_tuple& x) -> result<message> {
send_to_acquaintances(message::from(ptr)); send_to_acquaintances(message::from(&x));
return message{}; return message{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
...@@ -313,8 +313,8 @@ behavior proxy_broker::make_behavior() { ...@@ -313,8 +313,8 @@ behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](local_actor*, const type_erased_tuple* x) -> result<message> { auto fwd = [=](local_actor*, const type_erased_tuple& x) -> result<message> {
group_->send_all_subscribers(current_element_->sender, message::from(x), group_->send_all_subscribers(current_element_->sender, message::from(&x),
context()); context());
return message{}; return message{};
}; };
......
This diff is collapsed.
...@@ -21,6 +21,31 @@ ...@@ -21,6 +21,31 @@
namespace caf { namespace caf {
namespace {
/// Wraps a `message` into a mailbox element.
class mailbox_element_wrapper : public mailbox_element {
public:
/// Stores the elements for this mailbox element.
message msg;
mailbox_element_wrapper(strong_actor_ptr&& x0, message_id x1,
forwarding_stack&& x2, message&& x3)
: mailbox_element(std::move(x0), x1, std::move(x2)),
msg(std::move(x3)) {
// nop
}
type_erased_tuple& content() override {
auto ptr = msg.vals().raw_ptr();
if (ptr)
return *ptr;
return dummy_;
}
};
} // namespace <anonymous>
mailbox_element::mailbox_element() mailbox_element::mailbox_element()
: next(nullptr), : next(nullptr),
prev(nullptr), prev(nullptr),
...@@ -28,8 +53,8 @@ mailbox_element::mailbox_element() ...@@ -28,8 +53,8 @@ mailbox_element::mailbox_element()
// nop // nop
} }
mailbox_element::mailbox_element(strong_actor_ptr x, message_id y, mailbox_element::mailbox_element(strong_actor_ptr&& x, message_id y,
forwarding_stack z) forwarding_stack&& z)
: next(nullptr), : next(nullptr),
prev(nullptr), prev(nullptr),
marked(false), marked(false),
...@@ -39,39 +64,35 @@ mailbox_element::mailbox_element(strong_actor_ptr x, message_id y, ...@@ -39,39 +64,35 @@ mailbox_element::mailbox_element(strong_actor_ptr x, message_id y,
// nop // nop
} }
mailbox_element::mailbox_element(strong_actor_ptr x, message_id y, mailbox_element::~mailbox_element() {
forwarding_stack z, message m)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(x)),
mid(y),
stages(std::move(z)),
msg(std::move(m)) {
// nop // nop
} }
mailbox_element::~mailbox_element() { type_erased_tuple& mailbox_element::content() {
// nop return dummy_;
}
const type_erased_tuple& mailbox_element::content() const {
return const_cast<mailbox_element*>(this)->content();
} }
mailbox_element_ptr mailbox_element::make(strong_actor_ptr sender, mailbox_element_ptr make_mailbox_element(strong_actor_ptr sender, message_id id,
message_id id, mailbox_element::forwarding_stack stages,
forwarding_stack stages, message msg) {
message msg) { auto ptr = new mailbox_element_wrapper(std::move(sender), id,
auto ptr = detail::memory::create<mailbox_element>(std::move(sender), id, std::move(stages), std::move(msg));
std::move(stages),
std::move(msg));
return mailbox_element_ptr{ptr}; return mailbox_element_ptr{ptr};
} }
std::string to_string(const mailbox_element& x) { std::string to_string(const mailbox_element& x) {
std::string result = "("; std::string result = "mailbox_element(";
result += to_string(x.sender); result += to_string(x.sender);
result += ", "; result += ", ";
result += to_string(x.mid); result += to_string(x.mid);
result += ", "; result += ", ";
result += to_string(x.msg); result += deep_to_string(x.stages);
result += ", ";
result += to_string(x.content());
result += ")"; result += ")";
return result; return result;
} }
......
...@@ -66,8 +66,8 @@ cache_map& get_cache_map() { ...@@ -66,8 +66,8 @@ cache_map& get_cache_map() {
cache = new cache_map; cache = new cache_map;
pthread_setspecific(s_key, cache); pthread_setspecific(s_key, cache);
// insert default types // insert default types
std::unique_ptr<memory_cache> tmp(new basic_memory_cache<mailbox_element>); //std::unique_ptr<memory_cache> tmp(new basic_memory_cache<mailbox_element>);
cache->emplace(&typeid(mailbox_element), std::move(tmp)); //cache->emplace(&typeid(mailbox_element), std::move(tmp));
} }
return *cache; return *cache;
} }
......
...@@ -65,8 +65,7 @@ void* message::get_mutable(size_t p) { ...@@ -65,8 +65,7 @@ void* message::get_mutable(size_t p) {
} }
message message::from(const type_erased_tuple* ptr) { message message::from(const type_erased_tuple* ptr) {
if (! ptr) CAF_ASSERT(ptr != nullptr);
return message{};
auto dptr = dynamic_cast<const detail::message_data*>(ptr); auto dptr = dynamic_cast<const detail::message_data*>(ptr);
if (dptr) { if (dptr) {
data_ptr dp{const_cast<detail::message_data*>(dptr), true}; data_ptr dp{const_cast<detail::message_data*>(dptr), true};
......
...@@ -85,7 +85,7 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) { ...@@ -85,7 +85,7 @@ 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(nullptr, message_id::make(), {}, pr->enqueue(make_mailbox_element(nullptr, message_id::make(), {},
delete_atom::value, id()), delete_atom::value, id()),
nullptr); nullptr);
} }
...@@ -116,21 +116,6 @@ monitorable_actor::monitorable_actor(actor_config& cfg) : abstract_actor(cfg) { ...@@ -116,21 +116,6 @@ monitorable_actor::monitorable_actor(actor_config& cfg) : abstract_actor(cfg) {
// nop // nop
} }
optional<exit_reason> monitorable_actor::handle(const std::exception_ptr& eptr) {
std::unique_lock<std::mutex> guard{mtx_};
for (auto i = attachables_head_.get(); i != nullptr; i = i->next.get()) {
try {
auto result = i->handle_exception(eptr);
if (result)
return *result;
}
catch (...) {
// ignore exceptions
}
}
return none;
}
bool monitorable_actor::link_impl(linking_operation op, abstract_actor* other) { bool monitorable_actor::link_impl(linking_operation op, abstract_actor* other) {
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other));
switch (op) { switch (op) {
...@@ -240,10 +225,10 @@ size_t monitorable_actor::detach_impl(const attachable::token& what, ...@@ -240,10 +225,10 @@ size_t monitorable_actor::detach_impl(const attachable::token& what,
bool monitorable_actor::handle_system_message(mailbox_element& x, bool monitorable_actor::handle_system_message(mailbox_element& x,
execution_unit* ctx, execution_unit* ctx,
bool trap_exit) { bool trap_exit) {
auto& msg = x.msg; auto& msg = x.content();
if (! trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) { if (! trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) {
// exits for non-normal exit reasons // exits for non-normal exit reasons
auto& em = msg.get_as_mutable<exit_msg>(0); auto& em = msg.get_mutable_as<exit_msg>(0);
if (em.reason) if (em.reason)
cleanup(std::move(em.reason), ctx); cleanup(std::move(em.reason), ctx);
return true; return true;
...@@ -252,22 +237,22 @@ bool monitorable_actor::handle_system_message(mailbox_element& x, ...@@ -252,22 +237,22 @@ bool monitorable_actor::handle_system_message(mailbox_element& x,
return true; return true;
error err; error err;
mailbox_element_ptr res; mailbox_element_ptr res;
msg.apply({ msg.apply(
[&](sys_atom, get_atom, std::string& what) { [&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
if (what != "info") { if (what != "info") {
err = sec::unsupported_sys_key; err = sec::unsupported_sys_key;
return; return;
} }
res = mailbox_element::make(ctrl(), x.mid.response_id(), {}, res = make_mailbox_element(ctrl(), x.mid.response_id(), {},
ok_atom::value, std::move(what), ok_atom::value, std::move(what),
strong_actor_ptr{ctrl()}, name()); strong_actor_ptr{ctrl()}, name());
} }
}); );
if (! res && ! err) if (! res && ! err)
err = sec::unsupported_sys_message; err = sec::unsupported_sys_message;
if (err && x.mid.is_request()) if (err && x.mid.is_request())
res = mailbox_element::make(ctrl(), x.mid.response_id(), res = make_mailbox_element(ctrl(), x.mid.response_id(),
{}, std::move(err)); {}, std::move(err));
if (res) { if (res) {
auto s = actor_cast<strong_actor_ptr>(x.sender); auto s = actor_cast<strong_actor_ptr>(x.sender);
......
...@@ -54,7 +54,7 @@ response_promise response_promise::deliver_impl(message msg) { ...@@ -54,7 +54,7 @@ response_promise response_promise::deliver_impl(message msg) {
if (! stages_.empty()) { if (! stages_.empty()) {
auto next = std::move(stages_.back()); auto next = std::move(stages_.back());
stages_.pop_back(); stages_.pop_back();
next->enqueue(mailbox_element::make(std::move(source_), id_, next->enqueue(make_mailbox_element(std::move(source_), id_,
std::move(stages_), std::move(msg)), std::move(stages_), std::move(msg)),
self_->context()); self_->context());
return *this; return *this;
......
This diff is collapsed.
...@@ -24,7 +24,8 @@ ...@@ -24,7 +24,8 @@
namespace caf { namespace caf {
result<message> skip_t::skip_fun_impl(local_actor*, const type_erased_tuple*) { result<message> skip_t::skip_fun_impl(scheduled_actor*,
const type_erased_tuple&) {
return skip(); return skip();
} }
......
...@@ -41,8 +41,8 @@ struct splitter_state { ...@@ -41,8 +41,8 @@ struct splitter_state {
behavior fan_out_fan_in(stateful_actor<splitter_state>* self, behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
const std::vector<strong_actor_ptr>& workers) { const std::vector<strong_actor_ptr>& workers) {
auto f = [=](local_actor*, const type_erased_tuple* x) -> result<message> { auto f = [=](local_actor*, const type_erased_tuple& x) -> result<message> {
auto msg = message::from(x); auto msg = message::from(&x);
self->state.rp = self->make_response_promise(); self->state.rp = self->make_response_promise();
self->state.pending = workers.size(); self->state.pending = workers.size();
// request().await() has LIFO ordering // request().await() has LIFO ordering
......
...@@ -19,6 +19,8 @@ ...@@ -19,6 +19,8 @@
#include "caf/detail/try_match.hpp" #include "caf/detail/try_match.hpp"
#include "caf/type_erased_tuple.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
......
...@@ -34,6 +34,10 @@ void type_erased_tuple::load(deserializer& source) { ...@@ -34,6 +34,10 @@ void type_erased_tuple::load(deserializer& source) {
load(i, source); load(i, source);
} }
bool type_erased_tuple::shared() const noexcept {
return false;
}
bool type_erased_tuple::empty() const { bool type_erased_tuple::empty() const {
return size() == 0; return size() == 0;
} }
...@@ -67,4 +71,44 @@ bool type_erased_tuple::matches(size_t pos, uint16_t nr, ...@@ -67,4 +71,44 @@ bool type_erased_tuple::matches(size_t pos, uint16_t nr,
return true; return true;
} }
empty_type_erased_tuple::~empty_type_erased_tuple() {
// nop
}
void* empty_type_erased_tuple::get_mutable(size_t) {
CAF_RAISE_ERROR("empty_type_erased_tuple::get_mutable");
}
void empty_type_erased_tuple::load(size_t, deserializer&) {
CAF_RAISE_ERROR("empty_type_erased_tuple::get_mutable");
}
size_t empty_type_erased_tuple::size() const noexcept {
return 0;
}
uint32_t empty_type_erased_tuple::type_token() const noexcept {
return make_type_token();
}
auto empty_type_erased_tuple::type(size_t) const noexcept -> rtti_pair {
CAF_RAISE_ERROR("empty_type_erased_tuple::type");
}
const void* empty_type_erased_tuple::get(size_t) const noexcept {
CAF_RAISE_ERROR("empty_type_erased_tuple::get");
}
std::string empty_type_erased_tuple::stringify(size_t) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::stringify");
}
type_erased_value_ptr empty_type_erased_tuple::copy(size_t) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::copy");
}
void empty_type_erased_tuple::save(size_t, serializer&) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::save");
}
} // namespace caf } // namespace caf
...@@ -109,8 +109,10 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -109,8 +109,10 @@ CAF_TEST(round_robin_actor_pool) {
}, },
handle_err handle_err
); );
CAF_MESSAGE("await last worker");
anon_send_exit(workers.back(), exit_reason::user_shutdown); anon_send_exit(workers.back(), exit_reason::user_shutdown);
self->wait_for(workers.back()); self->wait_for(workers.back());
CAF_MESSAGE("last worker shut down");
workers.pop_back(); workers.pop_back();
// poll actor pool up to 10 times or until it removes the failed worker // poll actor pool up to 10 times or until it removes the failed worker
bool success = false; bool success = false;
...@@ -130,7 +132,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -130,7 +132,7 @@ CAF_TEST(round_robin_actor_pool) {
handle_err handle_err
); );
} }
CAF_REQUIRE(success); CAF_REQUIRE_EQUAL(success, true);
CAF_MESSAGE("about to send exit to workers"); CAF_MESSAGE("about to send exit to workers");
self->send_exit(pool, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
self->wait_for(workers); self->wait_for(workers);
...@@ -178,7 +180,7 @@ CAF_TEST(split_join_actor_pool) { ...@@ -178,7 +180,7 @@ CAF_TEST(split_join_actor_pool) {
auto spawn_split_worker = [&] { auto spawn_split_worker = [&] {
return system.spawn<lazy_init>([]() -> behavior { return system.spawn<lazy_init>([]() -> behavior {
return { return {
[](size_t pos, std::vector<int> xs) { [](size_t pos, const std::vector<int>& xs) {
return xs[pos]; return xs[pos];
} }
}; };
...@@ -195,6 +197,7 @@ CAF_TEST(split_join_actor_pool) { ...@@ -195,6 +197,7 @@ CAF_TEST(split_join_actor_pool) {
}); });
}; };
scoped_actor self{system}; scoped_actor self{system};
CAF_MESSAGE("create actor pool");
auto pool = 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(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive( self->request(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive(
......
...@@ -102,6 +102,7 @@ CAF_TEST(multiple_awaited_requests) { ...@@ -102,6 +102,7 @@ CAF_TEST(multiple_awaited_requests) {
for (int i = 0; i < 3; ++i) for (int i = 0; i < 3; ++i)
self->request(server, infinite, 42).await( self->request(server, infinite, 42).await(
[=](int x) { [=](int x) {
CAF_MESSAGE("received response #" << (i + 1));
CAF_REQUIRE_EQUAL(x, 42); CAF_REQUIRE_EQUAL(x, 42);
} }
); );
......
...@@ -278,8 +278,8 @@ CAF_TEST(param_detaching) { ...@@ -278,8 +278,8 @@ CAF_TEST(param_detaching) {
ERROR_HANDLER ERROR_HANDLER
); );
// alter our initial put, this time moving it to the dictionary // alter our initial put, this time moving it to the dictionary
put_msg.get_as_mutable<counting_string>(1) = "neverlord"; put_msg.get_mutable_as<counting_string>(1) = "neverlord";
put_msg.get_as_mutable<counting_string>(2) = "CAF"; put_msg.get_mutable_as<counting_string>(2) = "CAF";
// send put message to dictionary // send put message to dictionary
self->request(dict, infinite, std::move(put_msg)).receive( self->request(dict, infinite, std::move(put_msg)).receive(
[&] { [&] {
......
...@@ -30,10 +30,11 @@ class exception_testee : public event_based_actor { ...@@ -30,10 +30,11 @@ class exception_testee : public event_based_actor {
public: public:
~exception_testee(); ~exception_testee();
exception_testee(actor_config& cfg) : event_based_actor(cfg) { exception_testee(actor_config& cfg) : event_based_actor(cfg) {
set_exception_handler([](const std::exception_ptr&) -> optional<exit_reason> { set_exception_handler([](std::exception_ptr&) -> error {
return exit_reason::remote_link_unreachable; return exit_reason::remote_link_unreachable;
}); });
} }
behavior make_behavior() override { behavior make_behavior() override {
return { return {
[](const std::string&) { [](const std::string&) {
...@@ -49,7 +50,7 @@ exception_testee::~exception_testee() { ...@@ -49,7 +50,7 @@ exception_testee::~exception_testee() {
CAF_TEST(test_custom_exception_handler) { CAF_TEST(test_custom_exception_handler) {
actor_system system; actor_system system;
auto handler = [](const std::exception_ptr& eptr) -> optional<exit_reason> { auto handler = [](std::exception_ptr& eptr) -> error {
try { try {
std::rethrow_exception(eptr); std::rethrow_exception(eptr);
} }
......
...@@ -565,7 +565,7 @@ namespace { ...@@ -565,7 +565,7 @@ namespace {
class exception_testee : public event_based_actor { class exception_testee : public event_based_actor {
public: public:
exception_testee(actor_config& cfg) : event_based_actor(cfg) { exception_testee(actor_config& cfg) : event_based_actor(cfg) {
set_exception_handler([](const std::exception_ptr&) -> optional<exit_reason> { set_exception_handler([](std::exception_ptr&) -> exit_reason {
return exit_reason::unhandled_exception; return exit_reason::unhandled_exception;
}); });
} }
...@@ -581,7 +581,7 @@ public: ...@@ -581,7 +581,7 @@ public:
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST(custom_exception_handler) { CAF_TEST(custom_exception_handler) {
auto handler = [](const std::exception_ptr& eptr) -> optional<exit_reason> { auto handler = [](std::exception_ptr& eptr) -> error {
try { try {
std::rethrow_exception(eptr); std::rethrow_exception(eptr);
} }
......
...@@ -373,7 +373,7 @@ CAF_TEST(invalid_request) { ...@@ -373,7 +373,7 @@ CAF_TEST(invalid_request) {
CAF_ERROR("C did reply to 'HiThere'"); CAF_ERROR("C did reply to 'HiThere'");
}, },
[&](const error& err) { [&](const error& err) {
CAF_REQUIRE(err == sec::unexpected_message); CAF_REQUIRE_EQUAL(err, sec::unexpected_message);
} }
); );
} }
......
...@@ -147,7 +147,7 @@ CAF_TEST(no_name) { ...@@ -147,7 +147,7 @@ CAF_TEST(no_name) {
struct state { struct state {
// empty // empty
}; };
test_name<state>("actor"); test_name<state>("scheduled_actor");
} }
CAF_TEST(char_name) { CAF_TEST(char_name) {
......
...@@ -315,18 +315,21 @@ struct fixture { ...@@ -315,18 +315,21 @@ struct fixture {
CAF_CHECK_EQUAL(value, false); CAF_CHECK_EQUAL(value, false);
} }
); );
CAF_MESSAGE("async send + receive");
self->send(ts, my_request{42, 42}); self->send(ts, my_request{42, 42});
self->receive( self->receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
} }
); );
CAF_MESSAGE("request + receive with result true");
self->request(ts, infinite, my_request{10, 20}).receive( self->request(ts, infinite, my_request{10, 20}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, false); CAF_CHECK_EQUAL(value, false);
}, },
ERROR_HANDLER ERROR_HANDLER
); );
CAF_MESSAGE("request + receive with result false");
self->request(ts, infinite, my_request{0, 0}).receive( self->request(ts, infinite, my_request{0, 0}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
......
...@@ -148,9 +148,10 @@ public: ...@@ -148,9 +148,10 @@ public:
using super = stateful_actor<basp_broker_state, broker>; using super = stateful_actor<basp_broker_state, broker>;
explicit basp_broker(actor_config& cfg); explicit basp_broker(actor_config& cfg);
behavior make_behavior() override; behavior make_behavior() override;
proxy_registry* proxy_registry_ptr() override;
resume_result resume(execution_unit*, size_t) override; resume_result resume(execution_unit*, size_t) override;
void exec_single_event(execution_unit*, mailbox_element_ptr&) override;
}; };
} // namespace io } // namespace io
......
...@@ -34,8 +34,12 @@ namespace io { ...@@ -34,8 +34,12 @@ namespace io {
template <class Base, class Handle, class SysMsgType> template <class Base, class Handle, class SysMsgType>
class broker_servant : public Base { class broker_servant : public Base {
public: public:
broker_servant(abstract_broker* ptr, Handle x) : Base(ptr), hdl_(x) { broker_servant(abstract_broker* ptr, Handle x)
// nop : Base(ptr),
hdl_(x),
value_(strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, SysMsgType{}) {
set_hdl(msg(), x);
} }
Handle hdl() const { Handle hdl() const {
...@@ -48,13 +52,19 @@ protected: ...@@ -48,13 +52,19 @@ protected:
} }
void invoke_mailbox_element(execution_unit* ctx) { void invoke_mailbox_element(execution_unit* ctx) {
this->parent()->exec_single_event(ctx, mailbox_elem_ptr_); auto self = this->parent();
auto pfac = self->proxy_registry_ptr();
if (pfac)
ctx->proxy_registry_ptr(pfac);
auto guard = detail::make_scope_guard([=] {
if (pfac)
ctx->proxy_registry_ptr(nullptr);
});
self->activate(ctx, value_);
} }
SysMsgType& msg() { SysMsgType& msg() {
if (! mailbox_elem_ptr_) return value_.template get_mutable_as<SysMsgType>(0);
reset_mailbox_element();
return mailbox_elem_ptr_->msg.get_as_mutable<SysMsgType>(0);
} }
static void set_hdl(new_connection_msg& lhs, Handle& hdl) { static void set_hdl(new_connection_msg& lhs, Handle& hdl) {
...@@ -65,15 +75,8 @@ protected: ...@@ -65,15 +75,8 @@ protected:
lhs.handle = hdl; lhs.handle = hdl;
} }
void reset_mailbox_element() {
SysMsgType tmp;
set_hdl(tmp, hdl_);
mailbox_elem_ptr_ = mailbox_element::make(nullptr, invalid_message_id,
{}, tmp);
}
Handle hdl_; Handle hdl_;
mailbox_element_ptr mailbox_elem_ptr_; mailbox_element_vals<SysMsgType> value_;
}; };
} // namespace io } // namespace io
......
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/memory_managed.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
......
...@@ -37,7 +37,7 @@ namespace io { ...@@ -37,7 +37,7 @@ namespace io {
void abstract_broker::enqueue(strong_actor_ptr src, message_id mid, void abstract_broker::enqueue(strong_actor_ptr src, message_id mid,
message msg, execution_unit*) { message msg, execution_unit*) {
enqueue(mailbox_element::make(std::move(src), mid, {}, std::move(msg)), enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend()); &backend());
} }
......
...@@ -246,7 +246,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid, ...@@ -246,7 +246,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
return; return;
} }
self->parent().notify<hook::message_received>(src_nid, src, dest, mid, msg); self->parent().notify<hook::message_received>(src_nid, src, dest, mid, msg);
dest->enqueue(mailbox_element::make(std::move(src), mid, std::move(stages), dest->enqueue(make_mailbox_element(std::move(src), mid, std::move(stages),
std::move(msg)), std::move(msg)),
nullptr); nullptr);
} }
...@@ -671,14 +671,8 @@ resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) { ...@@ -671,14 +671,8 @@ resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) {
return super::resume(ctx, mt); return super::resume(ctx, mt);
} }
void basp_broker::exec_single_event(execution_unit* ctx, proxy_registry* basp_broker::proxy_registry_ptr() {
mailbox_element_ptr& ptr) { return &state.instance.proxies();
CAF_ASSERT(ctx != nullptr);
ctx->proxy_registry_ptr(&state.instance.proxies());
auto guard = detail::make_scope_guard([=] {
ctx->proxy_registry_ptr(nullptr);
});
super::exec_single_event(ctx, ptr);
} }
} // namespace io } // namespace io
......
...@@ -44,7 +44,7 @@ abstract_broker* manager::parent() { ...@@ -44,7 +44,7 @@ abstract_broker* manager::parent() {
return parent_ ? static_cast<abstract_broker*>(parent_->get()) : nullptr; return parent_ ? static_cast<abstract_broker*>(parent_->get()) : nullptr;
} }
void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) { void manager::detach(execution_unit*, bool invoke_disconnect_message) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (! detached()) { if (! detached()) {
CAF_LOG_DEBUG("disconnect servant from broker"); CAF_LOG_DEBUG("disconnect servant from broker");
...@@ -54,9 +54,10 @@ void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) { ...@@ -54,9 +54,10 @@ void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) {
ptr.swap(parent_); ptr.swap(parent_);
detach_from(raw_ptr); detach_from(raw_ptr);
if (invoke_disconnect_message) { if (invoke_disconnect_message) {
auto mptr = mailbox_element::make(nullptr, invalid_message_id, auto mptr = make_mailbox_element(nullptr, invalid_message_id,
{}, detach_message()); {}, detach_message());
raw_ptr->exec_single_event(ctx, mptr); if (raw_ptr->consume(*mptr) == im_skipped)
raw_ptr->push_to_cache(std::move(mptr));
} }
} }
} }
......
...@@ -303,7 +303,7 @@ void middleman::stop() { ...@@ -303,7 +303,7 @@ void middleman::stop() {
if (! ptr->is_terminated()) { if (! ptr->is_terminated()) {
ptr->context(&backend()); ptr->context(&backend());
ptr->is_terminated(true); ptr->is_terminated(true);
ptr->finished(); ptr->finalize();
} }
} }
}); });
......
...@@ -57,12 +57,9 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) { ...@@ -57,12 +57,9 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
auto& msg_buf = msg().buf; auto& msg_buf = msg().buf;
msg_buf.swap(buf); msg_buf.swap(buf);
invoke_mailbox_element(ctx); invoke_mailbox_element(ctx);
// `mailbox_elem_ptr_ == nullptr` if the broker moved it to the cache // swap buffer back to stream and implicitly flush wr_buf()
if (mailbox_elem_ptr_) { msg_buf.swap(buf);
// swap buffer back to stream and implicitly flush wr_buf() flush();
msg_buf.swap(buf);
flush();
}
} }
void scribe::data_transferred(execution_unit* ctx, size_t written, void scribe::data_transferred(execution_unit* ctx, size_t written,
...@@ -71,8 +68,9 @@ void scribe::data_transferred(execution_unit* ctx, size_t written, ...@@ -71,8 +68,9 @@ 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(nullptr, invalid_message_id, {}, tmp); auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp);
parent()->exec_single_event(ctx, ptr); parent()->context(ctx);
parent()->consume(std::move(ptr));
} }
void scribe::io_failure(execution_unit* ctx, network::operation op) { void scribe::io_failure(execution_unit* ctx, network::operation op) {
......
...@@ -340,7 +340,7 @@ public: ...@@ -340,7 +340,7 @@ public:
auto src = actor_cast<strong_actor_ptr>(registry_->get(hdr.source_actor)); auto src = actor_cast<strong_actor_ptr>(registry_->get(hdr.source_actor));
auto dest = registry_->get(hdr.dest_actor); auto dest = registry_->get(hdr.dest_actor);
CAF_REQUIRE(dest); CAF_REQUIRE(dest);
dest->enqueue(mailbox_element::make(src, message_id::make(), dest->enqueue(make_mailbox_element(src, message_id::make(),
std::move(stages), std::move(msg)), std::move(stages), std::move(msg)),
nullptr); nullptr);
} }
......
...@@ -170,7 +170,9 @@ void run_server(int argc, char** argv) { ...@@ -170,7 +170,9 @@ void run_server(int argc, char** argv) {
self->request(serv, infinite, publish_atom::value).receive( self->request(serv, infinite, publish_atom::value).receive(
[&](uint16_t port) { [&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port); CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] { run_client(argc, argv, port); }); child = std::thread([=] {
run_client(argc, argv, port);
});
}, },
[&](const error& err) { [&](const error& err) {
CAF_ERROR("Error: " << self->system().render(err)); CAF_ERROR("Error: " << self->system().render(err));
......
...@@ -257,7 +257,7 @@ int main(int argc, char** argv) { ...@@ -257,7 +257,7 @@ int main(int argc, char** argv) {
auto& remainder = res.remainder; auto& remainder = res.remainder;
if (remainder.empty()) if (remainder.empty())
return cerr << "empty command line" << endl, 1; return cerr << "empty command line" << endl, 1;
auto cmd = std::move(remainder.get_as_mutable<std::string>(0)); auto cmd = std::move(remainder.get_mutable_as<std::string>(0));
vector<string> xs; vector<string> xs;
remainder.drop(1).extract([&](string& x) { xs.emplace_back(std::move(x)); }); remainder.drop(1).extract([&](string& x) { xs.emplace_back(std::move(x)); });
auto hosts = read_hostfile(hostfile); auto hosts = read_hostfile(hostfile);
......
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