Commit 8139a6ed authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Move mailbox to blocking and scheduled actor

parent fb863af3
......@@ -23,26 +23,27 @@
#include <mutex>
#include <condition_variable>
#include "caf/fwd.hpp"
#include "caf/send.hpp"
#include "caf/none.hpp"
#include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp"
#include "caf/after.hpp"
#include "caf/extend.hpp"
#include "caf/behavior.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/local_actor.hpp"
#include "caf/typed_actor.hpp"
#include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/mailbox_policy.hpp"
#include "caf/none.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp"
namespace caf {
......@@ -68,6 +69,12 @@ class blocking_actor
public:
// -- member types -----------------------------------------------------------
/// Base type.
using super = extended_base;
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point;
......@@ -331,6 +338,10 @@ public:
/// The default implementation simply returns `next_message()`.
virtual mailbox_element_ptr dequeue();
/// Returns the queue for storing incoming messages.
inline mailbox_type& mailbox() {
return mailbox_;
}
/// @cond PRIVATE
/// Receives messages until either a pre- or postcheck of `rcc` fails.
......@@ -371,6 +382,8 @@ public:
void receive_impl(receive_cond& rcc, message_id mid,
detail::blocking_behavior& bhvr);
bool cleanup(error&& fail_state, execution_unit* host) override;
/// @endcond
private:
......@@ -392,6 +405,11 @@ private:
res += attach_functor(x);
return res;
}
// -- member variables -------------------------------------------------------
// used by both event-based and blocking actors
mailbox_type mailbox_;
};
} // namespace caf
......
......@@ -23,7 +23,6 @@
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp"
namespace caf {
......
......@@ -23,7 +23,9 @@
#include "caf/delegated.hpp"
#include "caf/replies_to.hpp"
#include "caf/response_promise.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/detail/type_list.hpp"
......@@ -49,6 +51,20 @@ struct input_is {
struct eval : input_is_eval_impl<Arguments, Signature> { };
};
template <class... Ts>
struct make_response_promise_helper {
using type = typed_response_promise<Ts...>;
};
template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>>
: make_response_promise_helper<Ts...> {};
template <>
struct make_response_promise_helper<response_promise> {
using type = response_promise;
};
template <class Output, class F>
struct type_checker {
static void check() {
......
......@@ -19,6 +19,8 @@
#ifndef CAF_INVOKE_MESSAGE_RESULT_HPP
#define CAF_INVOKE_MESSAGE_RESULT_HPP
#include <string>
namespace caf {
enum invoke_message_result {
......
......@@ -21,12 +21,10 @@
#include <atomic>
#include <cstdint>
#include <utility>
#include <exception>
#include <functional>
#include <type_traits>
#include <forward_list>
#include <unordered_map>
#include <utility>
#include "caf/abstract_actor.hpp"
#include "caf/abstract_group.hpp"
......@@ -39,13 +37,7 @@
#include "caf/delegated.hpp"
#include "caf/duration.hpp"
#include "caf/error.hpp"
#include "caf/execution_unit.hpp"
#include "caf/exit_reason.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_policy.hpp"
#include "caf/message.hpp"
#include "caf/message_handler.hpp"
#include "caf/message_id.hpp"
......@@ -58,44 +50,14 @@
#include "caf/typed_actor.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/typed_actor_util.hpp"
namespace caf {
namespace detail {
template <class... Ts>
struct make_response_promise_helper {
using type = typed_response_promise<Ts...>;
};
template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>>
: make_response_promise_helper<Ts...> {};
template <>
struct make_response_promise_helper<response_promise> {
using type = response_promise;
};
} // namespace detail
/// Base class for actors running on this node, either
/// living in an own thread or cooperatively scheduled.
class local_actor : public monitorable_actor {
public:
// -- member types -----------------------------------------------------------
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
// -- constructors, destructors, and assignment operators --------------------
local_actor(actor_config& cfg);
......@@ -119,7 +81,8 @@ public:
template <class T, spawn_options Os = no_spawn_options, class... Ts>
infer_handle_from_class_t<T> spawn(Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class<T, make_unbound(Os)>(cfg, std::forward<Ts>(xs)...));
return eval_opts(Os, system().spawn_class<T, make_unbound(Os)>(
cfg, std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options>
......@@ -132,46 +95,57 @@ public:
template <spawn_options Os = no_spawn_options, class F, class... Ts>
infer_handle_from_fun_t<F> spawn(F fun, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_functor<make_unbound(Os)>(cfg, fun, std::forward<Ts>(xs)...));
return eval_opts(Os, system().spawn_functor<make_unbound(Os)>(
cfg, fun, std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options, class Groups,
class... Ts>
actor spawn_in_groups(const Groups& gs, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options, class... Ts>
actor spawn_in_groups(std::initializer_list<group> gs, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options, class... Ts>
actor spawn_in_group(const group& grp, Ts&&... xs) {
actor_config cfg{context()};
auto first = &grp;
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(cfg, first, first + 1, std::forward<Ts>(xs)...));
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, first, first + 1, std::forward<Ts>(xs)...));
}
template <spawn_options Os = no_spawn_options, class Groups, class F, class... Ts>
template <spawn_options Os = no_spawn_options, class Groups, class F,
class... Ts>
actor spawn_in_groups(const Groups& gs, F fun, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_fun_in_groups<make_unbound(Os)>(cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
return eval_opts(
Os, system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
}
template <spawn_options Os = no_spawn_options, class F, class... Ts>
actor spawn_in_groups(std::initializer_list<group> gs, F fun, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_fun_in_groups<make_unbound(Os)>(cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
return eval_opts(
Os, system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
}
template <spawn_options Os = no_spawn_options, class F, class... Ts>
actor spawn_in_group(const group& grp, F fun, Ts&&... xs) {
actor_config cfg{context()};
auto first = &grp;
return eval_opts(Os, system().spawn_fun_in_groups<make_unbound(Os)>(cfg, first, first + 1, fun, std::forward<Ts>(xs)...));
return eval_opts(Os,
system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, first, first + 1, fun, std::forward<Ts>(xs)...));
}
// -- sending asynchronous messages ------------------------------------------
......@@ -404,27 +378,15 @@ public:
return {};
}
inline mailbox_type& mailbox() {
return mailbox_;
}
virtual void initialize();
bool cleanup(error&& fail_state, execution_unit* host) override;
message_id new_request_id(message_priority mp);
// -- message processing -----------------------------------------------------
/// Appends `x` to the cache for later consumption.
void push_to_cache(mailbox_element_ptr ptr);
protected:
// -- member variables -------------------------------------------------------
// used by both event-based and blocking actors
mailbox_type mailbox_;
// identifies the execution unit this actor is currently executed by
execution_unit* context_;
......
......@@ -25,29 +25,35 @@
#include <exception>
#endif // CAF_NO_EXCEPTIONS
#include <forward_list>
#include <type_traits>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/sec.hpp"
#include "caf/actor_marker.hpp"
#include "caf/error.hpp"
#include "caf/extend.hpp"
#include "caf/no_stages.hpp"
#include "caf/fwd.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp"
#include "caf/stream_result.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_policy.hpp"
#include "caf/no_stages.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
#include "caf/stream_result.hpp"
#include "caf/stream_result_trait.hpp"
#include "caf/to_string.hpp"
#include "caf/policy/arg.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/logger.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
namespace caf {
......@@ -78,6 +84,12 @@ class scheduled_actor : public local_actor, public resumable {
public:
// -- member types -----------------------------------------------------------
// Base type.
using super = local_actor;
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// A reference-counting pointer to a `stream_manager`.
using stream_manager_ptr = intrusive_ptr<stream_manager>;
......@@ -813,6 +825,11 @@ public:
/// @endcond
/// Returns the queue for storing incoming messages.
inline mailbox_type& mailbox() {
return mailbox_;
}
protected:
/// @cond PRIVATE
......@@ -848,8 +865,13 @@ protected:
bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
void push_to_cache(mailbox_element_ptr ptr);
// -- Member Variables -------------------------------------------------------
// used by both event-based and blocking actors
mailbox_type mailbox_;
/// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_;
......
......@@ -30,6 +30,12 @@
namespace caf {
namespace {
constexpr auto mpol = mailbox_policy{};
} // namespace <anonymous>
blocking_actor::receive_cond::~receive_cond() {
// nop
}
......@@ -51,7 +57,8 @@ bool blocking_actor::accept_one_cond::post() {
}
blocking_actor::blocking_actor(actor_config& cfg)
: extended_base(cfg.add_flag(local_actor::is_blocking_flag)) {
: super(cfg.add_flag(local_actor::is_blocking_flag)),
mailbox_(mpol, mpol, mpol, mpol, mpol) {
// nop
}
......@@ -303,4 +310,16 @@ size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) {
return 1;
}
bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
if (!mailbox_.closed()) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce))
; // nop
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
}
} // namespace caf
......@@ -16,6 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/local_actor.hpp"
#include <string>
#include <condition_variable>
......@@ -26,27 +28,16 @@
#include "caf/resumable.hpp"
#include "caf/actor_cast.hpp"
#include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/default_attachable.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
namespace {
constexpr auto mpol = mailbox_policy{};
} // namespace <anonymous>
local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg),
mailbox_(mpol, mpol, mpol, mpol, mpol),
context_(cfg.host),
initial_behavior_fac_(std::move(cfg.init_fun)) {
// nop
......@@ -126,13 +117,6 @@ void local_actor::initialize() {
bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state));
if (!mailbox_.closed()) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce))
; // nop
}
// tell registry we're done
unregister_from_system();
monitorable_actor::cleanup(std::move(fail_state), host);
......@@ -141,18 +125,4 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
return true;
}
void local_actor::push_to_cache(mailbox_element_ptr ptr) {
using namespace intrusive;
auto& p = mailbox_.queue().policy();
auto ts = p.task_size(*ptr);
auto& qs = mailbox_.queue().queues();
drr_cached_queue<mailbox_policy>* q;
if (p.id_of(*ptr) == mailbox_policy::default_queue_index)
q = &std::get<mailbox_policy::default_queue_index>(qs);
else
q = &std::get<mailbox_policy::high_priority_queue_index>(qs);
q->inc_total_task_size(ts);
q->cache().push_back(ptr.release());
}
} // namespace caf
......@@ -18,16 +18,24 @@
#include "caf/scheduled_actor.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/config.hpp"
#include "caf/to_string.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
namespace caf {
namespace {
constexpr auto mpol = mailbox_policy{};
} // namespace <anonymous>
// -- related free functions ---------------------------------------------------
result<message> reflect(scheduled_actor*, message_view& x) {
......@@ -94,7 +102,8 @@ error scheduled_actor::default_exception_handler(pointer ptr,
// -- constructors and destructors ---------------------------------------------
scheduled_actor::scheduled_actor(actor_config& cfg)
: local_actor(cfg),
: super(cfg),
mailbox_(mpol, mpol, mpol, mpol, mpol),
timeout_id_(0),
default_handler_(print_and_drop),
error_handler_(default_error_handler),
......@@ -200,8 +209,15 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
kvp.second->close();
streams_.clear();
*/
if (!mailbox_.closed()) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce))
; // nop
}
// Dispatch to parent's `cleanup` function.
return local_actor::cleanup(std::move(fail_state), host);
return super::cleanup(std::move(fail_state), host);
}
// -- overridden functions of resumable ----------------------------------------
......@@ -641,6 +657,20 @@ bool scheduled_actor::finalize() {
return true;
}
void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
using namespace intrusive;
auto& p = mailbox_.queue().policy();
auto ts = p.task_size(*ptr);
auto& qs = mailbox_.queue().queues();
drr_cached_queue<mailbox_policy>* q;
if (p.id_of(*ptr) == mailbox_policy::default_queue_index)
q = &std::get<mailbox_policy::default_queue_index>(qs);
else
q = &std::get<mailbox_policy::high_priority_queue_index>(qs);
q->inc_total_task_size(ts);
q->cache().push_back(ptr.release());
}
/*
bool scheduled_actor::handle_stream_msg(mailbox_element& x,
behavior* active_behavior) {
......
......@@ -91,7 +91,7 @@ using send_order_t = std::vector<msg_t>;
using sequence_t = std::vector<std::pair<msg_t, bool>>;
using check_order_t = std::pair<send_order_t, sequence_t>;
behavior check_order_behavior_factory(local_actor* self,
behavior check_order_behavior_factory(local_actor*,
sequence_t::const_iterator* seq_it_ptr) {
return {
[=](int i) -> result<void> {
......@@ -99,13 +99,11 @@ behavior check_order_behavior_factory(local_actor* self,
CAF_CHECK_EQUAL(i, seq_it->first);
if (seq_it->second) {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< "; inbox size: " << self->mailbox().size()
<< " SKIPPED");
++seq_it;
return skip();
} else {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< "; inbox size: " << self->mailbox().size()
<< " OK");
++seq_it;
return unit;
......
......@@ -149,6 +149,15 @@ private:
const Tup& xs_;
};
caf::mailbox_element* peek_mailbox(caf::local_actor* ptr) {
auto sptr = dynamic_cast<caf::scheduled_actor*>(ptr);
if (sptr != nullptr)
return sptr->mailbox().peek();
auto bptr = dynamic_cast<caf::blocking_actor*>(ptr);
CAF_REQUIRE(bptr != nullptr);
return bptr->mailbox().peek();
}
template <class Derived>
class expect_clause_base {
public:
......@@ -179,7 +188,7 @@ public:
Derived& to(const Handle& whom) {
CAF_REQUIRE(sched_.prioritize(whom));
dest_ = &sched_.next_job<caf::scheduled_actor>();
auto ptr = dest_->mailbox().peek();
auto ptr = peek_mailbox(dest_);
CAF_REQUIRE(ptr != nullptr);
if (src_)
CAF_REQUIRE_EQUAL(ptr->sender, src_);
......@@ -201,12 +210,12 @@ public:
template <class... Ts>
std::tuple<const Ts&...> peek() {
CAF_REQUIRE(dest_ != nullptr);
auto ptr = dest_->mailbox().peek();
if (!ptr->content().match_elements<Ts...>()) {
CAF_FAIL("Message does not match expected pattern: " << to_string(ptr->content()));
auto ptr = peek_mailbox(dest_);
if (!ptr->content().template match_elements<Ts...>()) {
CAF_FAIL("Message does not match expected pattern: "
<< to_string(ptr->content()));
}
//CAF_REQUIRE(ptr->content().match_elements<Ts...>());
return ptr->content().get_as_tuple<Ts...>();
return ptr->content().template get_as_tuple<Ts...>();
}
protected:
......@@ -300,7 +309,7 @@ public:
void with() {
CAF_REQUIRE(dest_ != nullptr);
auto ptr = dest_->mailbox().peek();
auto ptr = peek_mailbox(dest_);
CAF_CHECK(ptr->content().empty());
this->run_once();
}
......@@ -355,10 +364,10 @@ public:
template <class... Ts>
caf::optional<std::tuple<const Ts&...>> peek() {
CAF_REQUIRE(dest_ != nullptr);
auto ptr = dest_->mailbox().peek();
if (!ptr->content().match_elements<Ts...>())
auto ptr = peek(dest_);
if (!ptr->content().template match_elements<Ts...>())
return caf::none;
return ptr->content().get_as_tuple<Ts...>();
return ptr->content().template get_as_tuple<Ts...>();
}
protected:
......@@ -451,7 +460,7 @@ public:
void with() {
if (dest_ == nullptr)
return;
auto ptr = dest_->mailbox().peek();
auto ptr = peek_mailbox(dest_);
CAF_REQUIRE(!ptr->content().empty());
}
};
......
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