Commit 7ee835ac authored by Dominik Charousset's avatar Dominik Charousset

Remove actor policies and `proper_actor`

The previous approach based on mixins, policies, and the glue-it-all-together
`proper_actor` was ok for fast prototyping. The downside of this approach is
that stack traces are basically useless, since they are flooded with irrelevant
template metaprogramming types. Since CAF has long outgrown the prototyping
stage, it was about time to streamline actor classes. This also shortens build
times.
parent bddecb9c
......@@ -55,7 +55,6 @@ set (LIBCAF_CORE_SRCS
src/match_case.cpp
src/local_actor.cpp
src/logging.cpp
src/mailbox_based_actor.cpp
src/mailbox_element.cpp
src/memory.cpp
src/memory_managed.cpp
......
......@@ -29,6 +29,7 @@
#include <cstdint>
#include <exception>
#include <type_traits>
#include <condition_variable>
#include "caf/fwd.hpp"
#include "caf/node_id.hpp"
......@@ -161,6 +162,13 @@ class abstract_actor : public abstract_channel {
return m_host;
}
/**
* Sets the execution unit for this actor.
*/
inline void host(execution_unit* new_host) {
m_host = new_host;
}
protected:
/**
* Creates a non-proxy instance.
......@@ -185,13 +193,6 @@ class abstract_actor : public abstract_channel {
return exit_reason() != exit_reason::not_exited;
}
/**
* Sets the execution unit for this actor.
*/
inline void host(execution_unit* new_host) {
m_host = new_host;
}
/****************************************************************************
* here be dragons: end of public interface *
****************************************************************************/
......@@ -212,7 +213,9 @@ class abstract_actor : public abstract_channel {
has_timeout_flag = 0x02, // mixin::single_timeout
is_registered_flag = 0x04, // no_resume, resumable, and scoped_actor
is_initialized_flag = 0x08, // event-based actors
is_blocking_flag = 0x10 // blocking_actor
is_blocking_flag = 0x10, // blocking_actor
is_detached_flag = 0x20, // local_actor (set by spawn)
is_priority_aware_flag = 0x40 // local_actor (set by spawn)
};
inline void set_flag(bool enable_flag, actor_state_flag mask) {
......@@ -254,6 +257,25 @@ class abstract_actor : public abstract_channel {
set_flag(value, is_blocking_flag);
}
inline bool is_detached() const {
return get_flag(is_detached_flag);
}
inline void is_detached(bool value) {
set_flag(value, is_detached_flag);
}
inline bool is_priority_aware() const {
return get_flag(is_priority_aware_flag);
}
inline void is_priority_aware(bool value) {
set_flag(value, is_priority_aware_flag);
}
// Tries to run a custom exception handler for `eptr`.
optional<uint32_t> handle(const std::exception_ptr& eptr);
protected:
virtual bool link_impl(linking_operation op, const actor_addr& other);
......@@ -275,18 +297,19 @@ class abstract_actor : public abstract_channel {
bool stop_on_first_hit = false,
bool dry_run = false);
// Tries to run a custom exception handler for `eptr`.
optional<uint32_t> handle(const std::exception_ptr& eptr);
// cannot be changed after construction
const actor_id m_id;
// initially set to exit_reason::not_exited
std::atomic<uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, and m_links
// guards access to m_exit_reason, m_attachables, m_links,
// and enqueue operations if actor is thread-mapped
mutable std::mutex m_mtx;
// only used in blocking and thread-mapped actors
mutable std::condition_variable m_cv;
// attached functors that are executed on cleanup (monitors, links, etc)
attachable_ptr m_attachables_head;
......
......@@ -17,39 +17,81 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MAILBOX_BASED_ACTOR_HPP
#define CAF_MAILBOX_BASED_ACTOR_HPP
#ifndef CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
#define CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
#include <type_traits>
#include "caf/message_id.hpp"
#include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/behavior_policy.hpp"
#include "caf/response_handle.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/single_reader_queue.hpp"
#include "caf/mixin/sync_sender.hpp"
namespace caf {
/**
* Base class for local running actors using a mailbox.
/*
* Base type for typed and untyped event-based actors.
* @tparam BehaviorType Denotes the expected type for become().
* @tparam HasSyncSend Configures whether this base class extends `sync_sender`.
* @extends local_actor
*/
class mailbox_based_actor : public local_actor {
template <class BehaviorType, bool HasSyncSend>
class abstract_event_based_actor : public
std::conditional<
HasSyncSend,
extend<local_actor>
::with<mixin::sync_sender<nonblocking_response_handle_tag>::template impl>,
local_actor
>::type {
public:
using del = detail::disposer;
using mailbox_type = detail::single_reader_queue<mailbox_element, del>;
using behavior_type = BehaviorType;
~mailbox_based_actor();
/****************************************************************************
* become() member function family *
****************************************************************************/
void cleanup(uint32_t reason);
void become(behavior_type bhvr) {
this->do_become(std::move(unbox(bhvr)), true);
}
void become(const keep_behavior_t&, behavior_type bhvr) {
this->do_become(std::move(unbox(bhvr)), false);
}
template <class T, class... Ts>
typename std::enable_if<
!std::is_same<keep_behavior_t, typename std::decay<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
behavior_type bhvr{std::forward<T>(arg), std::forward<Ts>(args)...};
this->do_become(std::move(unbox(bhvr)), true);
}
inline mailbox_type& mailbox() {
return m_mailbox;
template <class... Ts>
void become(const keep_behavior_t&, Ts&&... args) {
behavior_type bhvr{std::forward<Ts>(args)...};
this->do_become(std::move(unbox(bhvr)), false);
}
protected:
mailbox_type m_mailbox;
void unbecome() {
this->m_bhvr_stack.pop_back();
}
private:
template <class... Ts>
static behavior& unbox(typed_behavior<Ts...>& arg) {
return arg.unbox();
}
static inline behavior& unbox(behavior& bhvr) {
return bhvr;
}
};
} // namespace caf
#endif // CAF_MAILBOX_BASED_ACTOR_HPP
#endif // CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
......@@ -25,9 +25,9 @@
#include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/shared_spinlock.hpp"
......@@ -38,11 +38,9 @@ namespace caf {
* An co-existing forwarding all messages through a user-defined
* callback to another object, thus serving as gateway to
* allow any object to interact with other actors.
* @extends local_actor
*/
class actor_companion
: public extend<local_actor, actor_companion>::
with<mixin::behavior_stack_based<behavior>::impl,
mixin::sync_sender<nonblocking_response_handle_tag>::impl> {
class actor_companion : public abstract_event_based_actor<behavior, true> {
public:
using lock_type = detail::shared_spinlock;
using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>;
......@@ -60,6 +58,8 @@ class actor_companion
*/
void on_enqueue(enqueue_handler handler);
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) override;
......
......@@ -20,6 +20,9 @@
#ifndef CAF_BLOCKING_ACTOR_HPP
#define CAF_BLOCKING_ACTOR_HPP
#include <mutex>
#include <condition_variable>
#include "caf/none.hpp"
#include "caf/on.hpp"
......@@ -29,7 +32,6 @@
#include "caf/typed_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/response_handle.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/detail/type_traits.hpp"
......@@ -40,10 +42,10 @@ namespace caf {
/**
* A thread-mapped or context-switching actor using a blocking
* receive rather than a behavior-stack based message processing.
* @extends mailbox_based_actor
* @extends local_actor
*/
class blocking_actor
: public extend<mailbox_based_actor, blocking_actor>::
: public extend<local_actor, blocking_actor>::
with<mixin::sync_sender<blocking_response_handle_tag>::impl> {
public:
class functor_based;
......@@ -184,12 +186,6 @@ class blocking_actor
return {make_dequeue_callback(), behavior{std::forward<Ts>(args)...}};
}
optional<behavior&> sync_handler(message_id msg_id) {
auto i = m_sync_handler.find(msg_id);
if (i != m_sync_handler.end()) return i->second;
return none;
}
/**
* Blocks this actor until all other actors are done.
*/
......@@ -202,34 +198,17 @@ class blocking_actor
/** @cond PRIVATE */
// required from invoke_policy; unused in blocking actors
inline void remove_handler(message_id) {}
// required by receive() member function family
inline void dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
}
// required by receive() member function family
inline void dequeue(behavior& bhvr) {
dequeue_response(bhvr, invalid_message_id);
}
// implemented by detail::proper_actor
virtual void dequeue_response(behavior& bhvr, message_id mid) = 0;
void initialize() override;
void cleanup(uint32_t reason);
void dequeue(behavior& bhvr, message_id mid = invalid_message_id);
/** @endcond */
private:
protected:
// helper function to implement receive_(for|while) and do_receive
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
std::map<message_id, behavior> m_sync_handler;
};
class blocking_actor::functor_based : public blocking_actor {
......
......@@ -38,78 +38,39 @@ namespace detail {
struct behavior_stack_mover;
class behavior_stack {
public:
friend struct behavior_stack_mover;
behavior_stack(const behavior_stack&) = delete;
behavior_stack& operator=(const behavior_stack&) = delete;
using element_type = std::pair<behavior, message_id>;
public:
behavior_stack() = default;
// @pre expected_response.valid()
optional<behavior&> sync_handler(message_id expected_response);
// erases the last asynchronous message handler
void pop_async_back();
// erases the last (asynchronous) behavior
void pop_back();
void clear();
// erases the synchronous response handler associated with `rid`
void erase(message_id rid) {
erase_if([=](const element_type& e) { return e.second == rid; });
inline bool empty() const {
return m_elements.empty();
}
inline bool empty() const { return m_elements.empty(); }
inline behavior& back() {
CAF_REQUIRE(!empty());
return m_elements.back().first;
return m_elements.back();
}
inline message_id back_id() {
CAF_REQUIRE(!empty());
return m_elements.back().second;
inline void push_back(behavior&& what) {
m_elements.emplace_back(std::move(what));
}
inline void push_back(behavior&& what,
message_id response_id = invalid_message_id) {
m_elements.emplace_back(std::move(what), response_id);
inline void cleanup() {
m_erased_elements.clear();
}
inline void cleanup() { m_erased_elements.clear(); }
private:
std::vector<element_type> m_elements;
std::vector<behavior> m_elements;
std::vector<behavior> m_erased_elements;
// note: checks wheter i points to m_elements.end() before calling erase()
inline void erase_at(std::vector<element_type>::iterator i) {
if (i != m_elements.end()) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
}
inline void rerase_at(std::vector<element_type>::reverse_iterator i) {
// base iterator points to the element *after* the correct element
if (i != m_elements.rend()) erase_at(i.base() - 1);
}
template <class UnaryPredicate>
inline void erase_if(UnaryPredicate p) {
erase_at(std::find_if(m_elements.begin(), m_elements.end(), p));
}
template <class UnaryPredicate>
inline void rerase_if(UnaryPredicate p) {
rerase_at(std::find_if(m_elements.rbegin(), m_elements.rend(), p));
}
};
} // namespace detail
......
......@@ -23,7 +23,7 @@
#include <memory>
#include <iterator>
#include "caf/policy/invoke_policy.hpp"
#include "caf/invoke_message_result.hpp"
namespace caf {
namespace detail {
......@@ -200,8 +200,9 @@ class intrusive_partitioned_list {
insert(second_end(), val);
}
template <class Actor, class... Ts>
bool invoke(Actor* self, iterator first, iterator last, Ts&... xs) {
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) {
......@@ -219,13 +220,13 @@ class intrusive_partitioned_list {
// it's safe, i.e., if invoke_message returned im_skipped
prev->next = next;
next->prev = prev;
switch (self->invoke_message(tmp, xs...)) {
case policy::im_dropped:
switch (self->invoke_message(tmp, bhvr, mid)) {
case im_dropped:
move_on(false);
break;
case policy::im_success:
case im_success:
return true;
case policy::im_skipped:
case im_skipped:
if (tmp) {
// re-integrate tmp and move on
prev->next = tmp.get();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_PROPER_ACTOR_HPP
#define CAF_DETAIL_PROPER_ACTOR_HPP
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/duration.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/logging.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/scheduling_policy.hpp"
namespace caf {
namespace detail {
// 'imports' all member functions from policies to the actor,
// the resume mixin also adds the m_hidden member which *must* be
// initialized to `true`
template <class Base, class Derived, class Policies>
class proper_actor_base : public Policies::resume_policy::template
mixin<Base, Derived> {
public:
using super = typename Policies::resume_policy::template mixin<Base, Derived>;
template <class... Ts>
proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) {
CAF_REQUIRE(this->is_registered() == false);
}
// member functions from scheduling policy
using timeout_type = typename Policies::scheduling_policy::timeout_type;
void enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) override {
auto d = dptr();
scheduling_policy().enqueue(d, mailbox_element::make(sender, mid,
std::move(msg)),
eu);
}
void enqueue(mailbox_element_ptr what, execution_unit* eu) override {
scheduling_policy().enqueue(dptr(), std::move(what), eu);
}
void launch(bool hide, bool lazy, execution_unit* eu) {
CAF_LOG_TRACE("");
this->is_registered(!hide);
this->scheduling_policy().launch(this, eu, lazy);
}
template <class F>
bool fetch_messages(F cb) {
return scheduling_policy().fetch_messages(dptr(), cb);
}
template <class F>
bool try_fetch_messages(F cb) {
return scheduling_policy().try_fetch_messages(dptr(), cb);
}
template <class F>
policy::timed_fetch_result fetch_messages(F cb, timeout_type abs_time) {
return scheduling_policy().fetch_messages(dptr(), cb, abs_time);
}
// member functions from priority policy
mailbox_element_ptr next_message() {
return priority_policy().next_message(dptr());
}
bool has_next_message() {
return priority_policy().has_next_message(dptr());
}
void push_to_cache(mailbox_element_ptr ptr) {
priority_policy().push_to_cache(dptr(), std::move(ptr));
}
// member functions from resume policy
// NOTE: resume_policy::resume is implemented in the mixin
// member functions from invoke policy
template <class PartialFunctionOrBehavior>
policy::invoke_message_result invoke_message(mailbox_element_ptr& me,
PartialFunctionOrBehavior& fun,
message_id awaited_response) {
return invoke_policy().invoke_message(dptr(), me, fun, awaited_response);
}
protected:
typename Policies::scheduling_policy& scheduling_policy() {
return m_policies.get_scheduling_policy();
}
typename Policies::priority_policy& priority_policy() {
return m_policies.get_priority_policy();
}
typename Policies::resume_policy& resume_policy() {
return m_policies.get_resume_policy();
}
typename policy::invoke_policy& invoke_policy() {
return m_policies.get_invoke_policy();
}
Derived* dptr() {
return static_cast<Derived*>(this);
}
private:
Policies m_policies;
};
// this is the nonblocking version of proper_actor; it assumes that Base is
// derived from local_actor and uses the behavior_stack_based mixin
template <class Base, class Policies,
bool OverrideDequeue = std::is_base_of<blocking_actor, Base>::value>
class proper_actor
: public proper_actor_base<Base, proper_actor<Base, Policies, false>,
Policies> {
public:
static_assert(std::is_base_of<local_actor, Base>::value,
"Base is not derived from local_actor");
using super = proper_actor_base<Base, proper_actor, Policies>;
template <class... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) {
// nop
}
// required by event_based_resume::mixin::resume
policy::invoke_message_result invoke_message(mailbox_element_ptr& me) {
CAF_LOG_TRACE("");
auto bhvr = this->bhvr_stack().back();
auto mid = this->bhvr_stack().back_id();
return this->invoke_policy().invoke_message(this, me, bhvr, mid);
}
bool invoke_message_from_cache() {
CAF_LOG_TRACE("");
return this->priority_policy().invoke_from_cache(this);
}
};
// for blocking actors, there's one more member function to implement
template <class Base, class Policies>
class proper_actor<Base, Policies, true>
: public proper_actor_base<Base, proper_actor<Base, Policies, true>,
Policies> {
using super = proper_actor_base<Base, proper_actor, Policies>;
public:
template <class... Ts>
proper_actor(Ts&&... args)
: super(std::forward<Ts>(args)...), m_next_timeout_id(0) {}
// 'import' optional blocking member functions from policies
void await_data() {
this->scheduling_policy().await_data(this);
}
void await_ready() {
this->resume_policy().await_ready(this);
}
// implement blocking_actor::dequeue_response
void dequeue_response(behavior& bhvr, message_id mid) override {
// try to dequeue from cache first
if (this->priority_policy().invoke_from_cache(this, bhvr, mid)) {
return;
}
bool timeout_valid = false;
uint32_t timeout_id;
// request timeout if needed
if (bhvr.timeout().valid()) {
timeout_valid = true;
timeout_id = this->request_timeout(bhvr.timeout());
}
// workaround for GCC 4.7 bug (const this when capturing refs)
auto& pending_timeouts = m_pending_timeouts;
auto guard = detail::make_scope_guard([&] {
if (timeout_valid) {
auto e = pending_timeouts.end();
auto i = std::find(pending_timeouts.begin(), e, timeout_id);
if (i != e) {
pending_timeouts.erase(i);
}
}
});
// read incoming messages
for (;;) {
this->await_ready();
auto msg = this->next_message();
switch (this->invoke_message(msg, bhvr, mid)) {
case policy::im_success:
return;
case policy::im_skipped:
if (msg) {
this->push_to_cache(std::move(msg));
}
break;
default:
// delete msg
break;
}
}
}
// timeout handling
uint32_t request_timeout(const duration& d) {
CAF_REQUIRE(d.valid());
auto tid = ++m_next_timeout_id;
auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host());
} else {
this->delayed_send(this, d, std::move(msg));
}
m_pending_timeouts.push_back(tid);
return tid;
}
void handle_timeout(behavior& bhvr, uint32_t timeout_id) {
auto e = m_pending_timeouts.end();
auto i = std::find(m_pending_timeouts.begin(), e, timeout_id);
CAF_LOG_WARNING_IF(i == e, "ignored unexpected timeout");
if (i != e) {
m_pending_timeouts.erase(i);
bhvr.handle_timeout();
}
}
// required by invoke policy
void pop_timeout() {
m_pending_timeouts.pop_back();
}
// required by invoke policy; adds a dummy timeout to the pending
// timeouts to prevent invokes to trigger an inactive timeout
void push_timeout() {
m_pending_timeouts.push_back(++m_next_timeout_id);
}
bool waits_for_timeout(uint32_t timeout_id) const {
auto e = m_pending_timeouts.end();
auto i = std::find(m_pending_timeouts.begin(), e, timeout_id);
return i != e;
}
bool is_active_timeout(uint32_t tid) const {
return !m_pending_timeouts.empty() && m_pending_timeouts.back() == tid;
}
private:
std::vector<uint32_t> m_pending_timeouts;
uint32_t m_next_timeout_id;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_PROPER_ACTOR_HPP
......@@ -26,11 +26,9 @@
#include "caf/extend.hpp"
#include "caf/local_actor.hpp"
#include "caf/response_handle.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
#include "caf/detail/logging.hpp"
......@@ -40,12 +38,9 @@ namespace caf {
* A cooperatively scheduled, event-based actor implementation. This is the
* recommended base class for user-defined actors and is used implicitly when
* spawning functor-based actors without the `blocking_api` flag.
* @extends mailbox_based_actor
* @extends local_actor
*/
class event_based_actor
: public extend<mailbox_based_actor, event_based_actor>::
with<mixin::behavior_stack_based<behavior>::impl,
mixin::sync_sender<nonblocking_response_handle_tag>::impl> {
class event_based_actor : public abstract_event_based_actor<behavior, true> {
public:
/**
* Forwards the last received message to `whom`.
......@@ -59,6 +54,8 @@ class event_based_actor
class functor_based;
void initialize() override;
protected:
/**
* Returns the initial actor behavior.
......@@ -66,6 +63,11 @@ class event_based_actor
virtual behavior make_behavior() = 0;
};
/**
* Implicitly used whenever spawning a cooperatively scheduled actor
* using a function of function object.
* @extends event_based_actor
*/
class event_based_actor::functor_based : public extend<event_based_actor>::
with<mixin::functor_based> {
public:
......
......@@ -54,7 +54,6 @@ class mailbox_element;
class message_handler;
class uniform_type_info;
class event_based_actor;
class mailbox_based_actor;
class forwarding_actor_proxy;
// structs
......
......@@ -17,21 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/mailbox_based_actor.hpp"
#ifndef CAF_INVOKE_MESSAGE_RESULT_HPP
#define CAF_INVOKE_MESSAGE_RESULT_HPP
namespace caf {
mailbox_based_actor::~mailbox_based_actor() {
if (!m_mailbox.closed()) {
detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
}
}
void mailbox_based_actor::cleanup(uint32_t reason) {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
local_actor::cleanup(reason);
}
enum invoke_message_result {
im_success,
im_skipped,
im_dropped
};
} // namespace caf
#endif // CAF_INVOKE_MESSAGE_RESULT_HPP
......@@ -26,6 +26,8 @@
#include <functional>
#include <forward_list>
#include "caf/fwd.hpp"
#include "caf/actor.hpp"
#include "caf/extend.hpp"
#include "caf/message.hpp"
......@@ -33,6 +35,7 @@
#include "caf/duration.hpp"
#include "caf/behavior.hpp"
#include "caf/spawn_fwd.hpp"
#include "caf/resumable.hpp"
#include "caf/message_id.hpp"
#include "caf/exit_reason.hpp"
#include "caf/typed_actor.hpp"
......@@ -44,6 +47,7 @@
#include "caf/response_promise.hpp"
#include "caf/message_priority.hpp"
#include "caf/check_typed_input.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/disposer.hpp"
......@@ -54,14 +58,15 @@
namespace caf {
// forward declarations
class sync_handle_helper;
/**
* Base class for local running actors.
* Base class for actors running on this node, either
* living in an own thread or cooperatively scheduled.
*/
class local_actor : public abstract_actor {
class local_actor : public abstract_actor, public resumable {
public:
using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>;
static constexpr auto memory_cache_flag = detail::needs_embedding;
~local_actor();
......@@ -138,7 +143,7 @@ class local_actor : public abstract_actor {
****************************************************************************/
/**
* Sends `{xs...} to `dest` using the priority `mp`.
* Sends `{xs...}` to `dest` using the priority `mp`.
*/
template <class... Ts>
void send(message_priority mp, const channel& dest, Ts&&... xs) {
......@@ -147,7 +152,7 @@ class local_actor : public abstract_actor {
}
/**
* Sends `{xs...} to `dest` using normal priority.
* Sends `{xs...}` to `dest` using normal priority.
*/
template <class... Ts>
void send(const channel& dest, Ts&&... xs) {
......@@ -155,7 +160,7 @@ class local_actor : public abstract_actor {
}
/**
* Sends `{xs...} to `dest` using the priority `mp`.
* Sends `{xs...}` to `dest` using the priority `mp`.
*/
template <class... Sigs, class... Ts>
void send(message_priority mp, const typed_actor<Sigs...>& dest, Ts&&... xs) {
......@@ -170,7 +175,7 @@ class local_actor : public abstract_actor {
}
/**
* Sends `{xs...} to `dest` using normal priority.
* Sends `{xs...}` to `dest` using normal priority.
*/
template <class... Sigs, class... Ts>
void send(const typed_actor<Sigs...>& dest, Ts&&... xs) {
......@@ -316,9 +321,7 @@ class local_actor : public abstract_actor {
* Can be overridden to perform cleanup code after an actor
* finished execution.
*/
inline void on_exit() {
// nop
}
virtual void on_exit();
/**
* Returns all joined groups.
......@@ -401,6 +404,16 @@ class local_actor : public abstract_actor {
message data) CAF_DEPRECATED;
// </backward_compatibility>
/****************************************************************************
* override pure virtual member functions of resumable *
****************************************************************************/
void attach_to_scheduler() override;
void detach_from_scheduler() override;
resumable::resume_result resume(execution_unit*, size_t) override;
/****************************************************************************
* here be dragons: end of public interface *
****************************************************************************/
......@@ -424,12 +437,6 @@ class local_actor : public abstract_actor {
return m_current_element;
}
inline message_id new_request_id() {
auto result = ++m_last_request_id;
m_pending_responses.push_front(result.response_id());
return result;
}
inline void handle_sync_timeout() {
if (m_sync_timeout_handler) {
m_sync_timeout_handler();
......@@ -464,16 +471,6 @@ class local_actor : public abstract_actor {
void forward_message(const actor& dest, message_priority mp);
inline bool awaits(message_id response_id) {
CAF_REQUIRE(response_id.is_response());
return std::any_of(m_pending_responses.begin(), m_pending_responses.end(),
[=](message_id other) { return response_id == other; });
}
inline void mark_arrived(message_id response_id) {
m_pending_responses.remove(response_id);
}
inline uint32_t planned_exit_reason() const {
return m_planned_exit_reason;
}
......@@ -482,14 +479,96 @@ class local_actor : public abstract_actor {
m_planned_exit_reason = value;
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
inline mailbox_type& mailbox() {
return m_mailbox;
}
inline bool has_behavior() {
return !m_bhvr_stack.empty() || !m_pending_responses.empty();
}
behavior& get_behavior() {
if (!m_pending_responses.empty()) {
return m_pending_responses.front().second;
}
return m_bhvr_stack.back();
}
virtual void initialize() = 0;
void cleanup(uint32_t reason);
// an actor can have multiple pending timeouts, but only
// the latest one is active (i.e. the m_pending_timeouts.back())
uint32_t request_timeout(const duration& d);
void handle_timeout(behavior& bhvr, uint32_t timeout_id);
void reset_timeout(uint32_t timeout_id);
// @pre has_timeout()
bool is_active_timeout(uint32_t tid) const;
// @pre has_timeout()
uint32_t active_timeout_id() const;
invoke_message_result invoke_message(mailbox_element_ptr& node,
behavior& fun,
message_id awaited_response);
using pending_response = std::pair<message_id, behavior>;
message_id new_request_id();
void mark_arrived(message_id response_id);
bool awaits_response() const;
bool awaits(message_id response_id) const;
optional<pending_response&> find_pending_response(message_id mid);
void set_response_handler(message_id response_id, behavior bhvr);
behavior& awaited_response_handler();
message_id awaited_response_id();
// these functions are dispatched via the actor policies table
void launch(execution_unit* eu, bool lazy, bool hide);
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override;
mailbox_element_ptr next_message();
bool has_next_message();
void push_to_cache(mailbox_element_ptr);
bool invoke_from_cache();
bool invoke_from_cache(behavior&, message_id);
protected:
void do_become(behavior bhvr, bool discard_old);
// used only in thread-mapped actors
void await_data();
// identifies the ID of the last sent synchronous request
message_id m_last_request_id;
// identifies all IDs of sync messages waiting for a response
std::forward_list<message_id> m_pending_responses;
std::forward_list<pending_response> m_pending_responses;
// points to m_dummy_node if no callback is currently invoked,
// points to the node under processing otherwise
......@@ -498,6 +577,15 @@ class local_actor : public abstract_actor {
// set by quit
uint32_t m_planned_exit_reason;
// identifies the timeout messages we are currently waiting for
uint32_t m_timeout_id;
// used by both event-based and blocking actors
detail::behavior_stack m_bhvr_stack;
// used by both event-based and blocking actors
mailbox_type m_mailbox;
/** @endcond */
private:
......
......@@ -115,7 +115,8 @@ class message_id : detail::comparable<message_id> {
return result;
}
static inline message_id make(message_priority prio) {
static inline message_id make(message_priority prio
= message_priority::normal) {
message_id res;
return prio == message_priority::high ? res.with_high_priority() : res;
}
......
......@@ -85,7 +85,6 @@ class actor_widget : public Base {
}
private:
policy::invoke_policy m_invoke;
actor_companion_ptr m_companion;
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MIXIN_BEHAVIOR_STACK_BASED_HPP
#define CAF_MIXIN_BEHAVIOR_STACK_BASED_HPP
#include "caf/message_id.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/behavior_policy.hpp"
#include "caf/response_handle.hpp"
#include "caf/detail/behavior_stack.hpp"
namespace caf {
namespace mixin {
template <class Base, class Subtype, class BehaviorType>
class behavior_stack_based_impl : public Base {
public:
// types and constructors
using behavior_type = BehaviorType;
using combined_type = behavior_stack_based_impl;
using response_handle_type = response_handle<behavior_stack_based_impl,
message,
nonblocking_response_handle_tag>;
template <class... Ts>
behavior_stack_based_impl(Ts&&... xs)
: Base(std::forward<Ts>(xs)...),
m_timeout_id(0) {
// nop
}
/****************************************************************************
* become() member function family *
****************************************************************************/
void become(behavior_type bhvr) {
do_become(std::move(bhvr), true);
}
void become(const keep_behavior_t&, behavior_type bhvr) {
do_become(std::move(bhvr), false);
}
template <class T, class... Ts>
typename std::enable_if<
!std::is_same<keep_behavior_t, typename std::decay<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
do_become(behavior_type{std::forward<T>(arg), std::forward<Ts>(args)...},
true);
}
template <class... Ts>
void become(const keep_behavior_t&, Ts&&... args) {
do_become(behavior_type{std::forward<Ts>(args)...}, false);
}
void unbecome() {
m_bhvr_stack.pop_async_back();
}
/****************************************************************************
* convenience member function for stack manipulation *
****************************************************************************/
bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
behavior& get_behavior() {
CAF_REQUIRE(m_bhvr_stack.empty() == false);
return m_bhvr_stack.back();
}
optional<behavior&> sync_handler(message_id msg_id) {
return m_bhvr_stack.sync_handler(msg_id);
}
void remove_handler(message_id mid) {
m_bhvr_stack.erase(mid);
}
detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
/****************************************************************************
* timeout handling *
****************************************************************************/
void request_timeout(const duration& d) {
if (d.valid()) {
this->has_timeout(true);
auto tid = ++m_timeout_id;
auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host());
} else
this->delayed_send(this, d, std::move(msg));
} else
this->has_timeout(false);
}
bool waits_for_timeout(uint32_t timeout_id) const {
return this->has_timeout() && m_timeout_id == timeout_id;
}
bool is_active_timeout(uint32_t tid) const {
return waits_for_timeout(tid);
}
uint32_t active_timeout_id() const {
return m_timeout_id;
}
void reset_timeout() {
this->has_timeout(false);
}
void handle_timeout(behavior& bhvr, uint32_t timeout_id) {
if (this->is_active_timeout(timeout_id)) {
this->reset_timeout();
bhvr.handle_timeout();
// request next timeout if behavior stack is not empty
// and timeout handler did not set a new timeout, e.g.,
// by calling become()
if (!this->has_timeout() && has_behavior()) {
this->request_timeout(get_behavior().timeout());
}
}
}
private:
void do_become(behavior_type bhvr, bool discard_old) {
if (discard_old) this->m_bhvr_stack.pop_async_back();
// request_timeout simply resets the timeout when it's invalid
this->request_timeout(bhvr.timeout());
this->m_bhvr_stack.push_back(std::move(unbox(bhvr)));
}
static behavior& unbox(behavior& arg) {
return arg;
}
template <class... Ts>
static behavior& unbox(typed_behavior<Ts...>& arg) {
return arg.unbox();
}
// utility for getting a pointer-to-derived-type
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
// utility for getting a const pointer-to-derived-type
const Subtype* dptr() const {
return static_cast<const Subtype*>(this);
}
detail::behavior_stack m_bhvr_stack;
uint32_t m_timeout_id;
};
/**
* Mixin for actors using a stack-based message processing.
* @note This mixin implicitly includes {@link single_timeout}.
*/
template <class BehaviorType>
class behavior_stack_based {
public:
template <class Base, class Subtype>
class impl : public behavior_stack_based_impl<Base, Subtype, BehaviorType> {
public:
using super = behavior_stack_based_impl<Base, Subtype, BehaviorType>;
using combined_type = impl;
template <class... Ts>
impl(Ts&&... args)
: super(std::forward<Ts>(args)...) {
// nop
}
};
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_BEHAVIOR_STACK_BASED_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_POLICIES_HPP
#define CAF_POLICY_POLICIES_HPP
#include "caf/policy/invoke_policy.hpp"
namespace caf {
namespace policy {
/**
* A container for actor-related policies.
*/
template <class SchedulingPolicy, class PriorityPolicy, class ResumePolicy>
class actor_policies {
public:
using scheduling_policy = SchedulingPolicy;
using priority_policy = PriorityPolicy;
using resume_policy = ResumePolicy;
inline scheduling_policy& get_scheduling_policy() {
return m_scheduling_policy;
}
inline priority_policy& get_priority_policy() {
return m_priority_policy;
}
inline resume_policy& get_resume_policy() {
return m_resume_policy;
}
inline invoke_policy& get_invoke_policy() {
return m_invoke_policy;
}
private:
scheduling_policy m_scheduling_policy;
priority_policy m_priority_policy;
resume_policy m_resume_policy;
invoke_policy m_invoke_policy;
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_POLICIES_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_COOPERATIVE_SCHEDULING_HPP
#define CAF_POLICY_COOPERATIVE_SCHEDULING_HPP
#include <atomic>
#include "caf/message.hpp"
#include "caf/execution_unit.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/single_reader_queue.hpp"
namespace caf {
namespace policy {
class cooperative_scheduling {
public:
using timeout_type = int;
template <class Actor>
void launch(Actor* self, execution_unit* host, bool lazy) {
// detached in scheduler::worker::run
self->attach_to_scheduler();
if (lazy) {
self->mailbox().try_block();
return;
}
if (host) {
host->exec_later(self);
} else {
detail::singletons::get_scheduling_coordinator()->enqueue(self);
}
}
template <class Actor>
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* eu) {
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (self->mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// re-schedule actor
if (eu) {
eu->exec_later(self);
} else {
detail::singletons::get_scheduling_coordinator()->enqueue(self);
}
break;
}
case detail::enqueue_result::queue_closed: {
if (mid.is_request()) {
detail::sync_request_bouncer f{self->exit_reason()};
f(sender, mid);
}
break;
}
case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do
break;
}
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_COOPERATIVE_SCHEDULING_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_EVENT_BASED_RESUME_HPP
#define CAF_POLICY_EVENT_BASED_RESUME_HPP
#include <tuple>
#include <stack>
#include <memory>
#include <vector>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/extend.hpp"
#include "caf/behavior.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/resume_policy.hpp"
#include "caf/detail/logging.hpp"
namespace caf {
namespace policy {
class event_based_resume {
public:
// Base must be a mailbox-based actor
template <class Base, class Derived>
struct mixin : Base, resumable {
template <class... Ts>
mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) {
// nop
}
void attach_to_scheduler() override {
this->ref();
}
void detach_from_scheduler() override {
this->deref();
}
resumable::resume_result resume(execution_unit* new_host,
size_t max_throughput) override {
CAF_REQUIRE(max_throughput > 0);
auto d = static_cast<Derived*>(this);
CAF_LOG_TRACE("id = " << d->id());
d->host(new_host);
auto done_cb = [&]() -> bool {
CAF_LOG_TRACE("");
d->bhvr_stack().clear();
d->bhvr_stack().cleanup();
d->on_exit();
if (!d->bhvr_stack().empty()) {
CAF_LOG_DEBUG("on_exit did set a new behavior");
d->planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
auto rsn = d->planned_exit_reason();
if (rsn == exit_reason::not_exited) {
rsn = exit_reason::normal;
d->planned_exit_reason(rsn);
}
d->cleanup(rsn);
return true;
};
auto actor_done = [&]() -> bool {
if (d->bhvr_stack().empty()
|| d->planned_exit_reason() != exit_reason::not_exited) {
return done_cb();
}
return false;
};
// actors without behavior or that have already defined
// an exit reason must not be resumed
CAF_REQUIRE(!d->is_initialized()
|| (!d->bhvr_stack().empty()
&& d->planned_exit_reason() == exit_reason::not_exited));
std::exception_ptr eptr = nullptr;
try {
if (!d->is_initialized()) {
CAF_LOG_DEBUG("initialize actor");
d->is_initialized(true);
auto bhvr = d->make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior, "
<< "bhvr_stack().empty() = "
<< std::boolalpha << d->bhvr_stack().empty());
if (bhvr) {
// make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
d->become(std::move(bhvr));
}
if (actor_done()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return resume_result::done;
}
}
auto had_tout = d->has_timeout();
auto tout = d->active_timeout_id();
int handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (had_tout && handled_msgs > 0 && tout == d->active_timeout_id()) {
d->request_timeout(d->get_behavior().timeout());
}
};
// max_throughput = 0 means infinite
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = d->next_message();
if (ptr) {
switch (d->invoke_message(ptr)) {
case policy::im_success:
d->bhvr_stack().cleanup();
++handled_msgs;
if (actor_done()) {
CAF_LOG_DEBUG("actor exited");
return resume_result::done;
}
// continue from cache if current message was
// handled, because the actor might have changed
// its behavior to match 'old' messages now
while (d->invoke_message_from_cache()) {
if (actor_done()) {
CAF_LOG_DEBUG("actor exited");
return resume_result::done;
}
}
break;
case policy::im_skipped:
CAF_REQUIRE(ptr != nullptr);
d->push_to_cache(std::move(ptr));
break;
case policy::im_dropped:
// destroy msg
break;
}
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
if (d->mailbox().try_block()) {
reset_timeout_if_needed();
return resumable::awaiting_message;
}
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
if (!d->has_next_message() && d->mailbox().try_block()) {
reset_timeout_if_needed();
return resumable::awaiting_message;
}
// time's up
return resumable::resume_later;
}
catch (actor_exited& what) {
CAF_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason());
if (d->exit_reason() == exit_reason::not_exited) {
d->quit(what.reason());
}
}
catch (std::exception& e) {
CAF_LOG_INFO("actor died because of an exception, what: " << e.what());
static_cast<void>(e); // keep compiler happy when not logging
if (d->exit_reason() == exit_reason::not_exited) {
d->quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception();
}
catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception");
if (d->exit_reason() == exit_reason::not_exited) {
d->quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = d->handle(eptr);
if (opt_reason) {
// use exit reason defined by custom handler
d->planned_exit_reason(*opt_reason);
}
}
if (!actor_done()) {
// actor has been "revived", try running it again later
return resumable::resume_later;
}
return resumable::done;
}
};
template <class Actor>
void await_data(Actor*) {
static_assert(std::is_same<Actor, Actor>::value == false,
"The event-based resume policy cannot be used "
"to implement blocking actors");
}
template <class Actor>
bool await_data(Actor*, const duration&) {
static_assert(std::is_same<Actor, Actor>::value == false,
"The event-based resume policy cannot be used "
"to implement blocking actors");
return false;
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_EVENT_BASED_RESUME_HPP
This diff is collapsed.
#ifndef NO_RESUME_HPP
#define NO_RESUME_HPP
#include <chrono>
#include <utility>
#include "caf/exception.hpp"
#include "caf/exit_reason.hpp"
#include "caf/detail/logging.hpp"
#include "caf/policy/resume_policy.hpp"
namespace caf {
namespace policy {
class no_resume {
public:
template <class Base, class Derived>
struct mixin : Base {
template <class... Ts>
mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) {
// nop
}
void attach_to_scheduler() {
this->ref();
}
void detach_from_scheduler() {
this->deref();
}
resumable::resume_result resume(execution_unit*, size_t) {
CAF_LOG_TRACE("");
uint32_t rsn = exit_reason::normal;
std::exception_ptr eptr = nullptr;
try {
this->act();
}
catch (actor_exited& e) {
rsn = e.reason();
}
catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = this->handle(eptr);
rsn = *opt_reason;
}
this->planned_exit_reason(rsn);
try {
this->on_exit();
}
catch (...) {
// simply ignore exception
}
// exit reason might have been changed by on_exit()
this->cleanup(this->planned_exit_reason());
return resumable::done;
}
};
template <class Actor>
void await_ready(Actor* self) {
self->await_data();
}
};
} // namespace policy
} // namespace caf
#endif // NO_RESUME_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_NO_SCHEDULING_HPP
#define CAF_NO_SCHEDULING_HPP
#include <mutex>
#include <thread>
#include <chrono>
#include <limits>
#include <condition_variable>
#include "caf/duration.hpp"
#include "caf/exit_reason.hpp"
#include "caf/policy/scheduling_policy.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/single_reader_queue.hpp"
#include "caf/actor_ostream.hpp"
namespace caf {
namespace policy {
class no_scheduling {
using lock_type = std::unique_lock<std::mutex>;
public:
using timeout_type = std::chrono::high_resolution_clock::time_point;
template <class Actor>
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit*) {
auto mid = ptr->mid;
auto sender = ptr->sender;
// returns false if mailbox has been closed
if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr.release())) {
if (mid.is_request()) {
detail::sync_request_bouncer srb{self->exit_reason()};
srb(sender, mid);
}
}
}
template <class Actor>
void launch(Actor* self, execution_unit*, bool) {
CAF_REQUIRE(self != nullptr);
CAF_PUSH_AID(self->id());
CAF_LOG_TRACE(CAF_ARG(self));
intrusive_ptr<Actor> mself{self};
self->attach_to_scheduler();
std::thread([=] {
CAF_PUSH_AID(mself->id());
CAF_LOG_TRACE("");
auto max_throughput = std::numeric_limits<size_t>::max();
while (mself->resume(nullptr, max_throughput) != resumable::done) {
// await new data before resuming actor
await_data(mself.get());
CAF_REQUIRE(self->mailbox().blocked() == false);
}
self->detach_from_scheduler();
}).detach();
}
// await_data is being called from no_resume (only)
template <class Actor>
void await_data(Actor* self) {
if (self->has_next_message()) return;
self->mailbox().synchronized_await(m_mtx, m_cv);
}
// this additional member function is needed to implement
// timer_actor (see scheduler.cpp)
template <class Actor, class TimePoint>
bool await_data(Actor* self, const TimePoint& tp) {
if (self->has_next_message()) return true;
return self->mailbox().synchronized_await(m_mtx, m_cv, tp);
}
private:
std::mutex m_mtx;
std::condition_variable m_cv;
};
} // namespace policy
} // namespace caf
#endif // CAF_NO_SCHEDULING_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_NOT_PRIORITIZING_HPP
#define CAF_POLICY_NOT_PRIORITIZING_HPP
#include <deque>
#include <iterator>
#include "caf/mailbox_element.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/priority_policy.hpp"
namespace caf {
namespace policy {
class not_prioritizing {
public:
template <class Actor>
typename Actor::mailbox_type::unique_pointer next_message(Actor* self) {
typename Actor::mailbox_type::unique_pointer result;
result.reset(self->mailbox().try_pop());
return result;
}
template <class Actor>
bool has_next_message(Actor* self) {
return self->mailbox().can_fetch_more();
}
template <class Actor>
void push_to_cache(Actor* self,
typename Actor::mailbox_type::unique_pointer ptr) {
self->mailbox().cache().push_second_back(ptr.release());
}
template <class Actor, class... Ts>
bool invoke_from_cache(Actor* self, Ts&... args) {
auto& cache = self->mailbox().cache();
auto i = cache.second_begin();
auto e = cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
return cache.invoke(self, i, e, args...);
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_NOT_PRIORITIZING_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_PRIORITIZING_HPP
#define CAF_POLICY_PRIORITIZING_HPP
#include <list>
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/detail/single_reader_queue.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
namespace policy {
// this policy partitions the mailbox into four segments:
// <-------- !was_skipped --------> | <-------- was_skipped -------->
// <-- high prio --><-- low prio -->|<-- high prio --><-- low prio -->
class prioritizing {
public:
using pointer = mailbox_element*;
using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>;
using cache_type = mailbox_type::cache_type;
using iterator = cache_type::iterator;
template <class Actor>
typename Actor::mailbox_type::unique_pointer next_message(Actor* self) {
auto& cache = self->mailbox().cache();
auto i = cache.first_begin();
auto e = cache.first_end();
if (i == e || !i->is_high_priority()) {
// insert points for high priority
auto hp_pos = i;
// read whole mailbox at once
auto tmp = self->mailbox().try_pop();
while (tmp) {
cache.insert(tmp->is_high_priority() ? hp_pos : e, tmp);
// adjust high priority insert point on first low prio element insert
if (hp_pos == e && !tmp->is_high_priority()) {
--hp_pos;
}
tmp = self->mailbox().try_pop();
}
}
typename Actor::mailbox_type::unique_pointer result;
if (!cache.first_empty()) {
result.reset(cache.take_first_front());
}
return result;
}
template <class Actor>
bool has_next_message(Actor* self) {
auto& mbox = self->mailbox();
auto& cache = mbox.cache();
return !cache.first_empty() || mbox.can_fetch_more();
}
template <class Actor>
void push_to_cache(Actor* self,
typename Actor::mailbox_type::unique_pointer ptr) {
auto high_prio = [](const typename Actor::mailbox_type::value_type& val) {
return val.is_high_priority();
};
auto& cache = self->mailbox().cache();
auto e = cache.second_end();
auto i = ptr->is_high_priority()
? std::partition_point(cache.second_begin(), e, high_prio)
: e;
cache.insert(i, ptr.release());
}
template <class Actor, class... Ts>
bool invoke_from_cache(Actor* self, Ts&... args) {
// same as in not prioritzing policy
not_prioritizing np;
return np.invoke_from_cache(self, args...);
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_PRIORITIZING_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_PRIORITY_POLICY_HPP
#define CAF_PRIORITY_POLICY_HPP
#include "caf/mailbox_element.hpp"
namespace caf {
namespace policy {
/**
* The priority_policy *concept* class. Please note that this class is **not**
* implemented. It merely explains all required member functions.
*/
class priority_policy {
public:
/**
* Returns the next message from the mailbox or `nullptr` if it's empty.
*/
template <class Actor>
mailbox_element_ptr next_message(Actor* self);
/**
* Queries whether the mailbox is not empty.
*/
template <class Actor>
bool has_next_message(Actor* self);
/**
* Stores the message in a cache for later retrieval.
*/
template <class Actor>
void push_to_cache(Actor* self, mailbox_element_ptr ptr);
/**
* Removes the first element from the cache matching predicate `p`.
* @returns `true` if an element was removed, `false` otherwise.
*/
template <class Actor, class Predicate>
bool invoke_from_cache(Actor* self, Predicate p);
};
} // namespace policy
} // namespace caf
#endif // CAF_PRIORITY_POLICY_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_RESUME_POLICY_HPP
#define CAF_RESUME_POLICY_HPP
#include "caf/resumable.hpp"
// this header consists all type definitions needed to
// implement the resume_policy trait
namespace caf {
class execution_unit;
class duration;
namespace policy {
/**
* The resume_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all
* required member function and their behavior for any resume policy.
*/
class resume_policy {
public:
/**
* Resumes the actor by reading a new message and then
* calling `self->invoke(msg)`. This process is repeated
* until either no message is left in the actor's mailbox or the
* actor finishes execution.
*/
template <class Actor>
resumable::resume_result resume(Actor* self, execution_unit*);
/**
* Waits unconditionally until the actor is ready to be resumed.
* This member function calls `self->await_data()`.
* @note This member function must raise a compiler error if the resume
* strategy cannot be used to implement blocking actors.
*/
template <class Actor>
bool await_ready(Actor* self);
};
} // namespace policy
} // namespace caf
#endif // CAF_RESUME_POLICY_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_SCHEDULING_POLICY_HPP
#define CAF_SCHEDULING_POLICY_HPP
#include "caf/fwd.hpp"
namespace caf {
namespace policy {
enum class timed_fetch_result {
no_message,
indeterminate,
success
};
/**
* The scheduling_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all
* required member function and their behavior for any scheduling policy.
*/
class scheduling_policy {
public:
/**
* This type can be set freely by any implementation and is
* used by callers to pass the result of `init_timeout` back to
* `fetch_messages`.
*/
using timeout_type = int;
/**
* Fetches new messages from the actor's mailbox and feeds them
* to the given callback. The member function returns `false` if
* no message was read, `true` otherwise.
*
* In case this function returned `false,` the policy also sets the state
* of the actor to blocked. Any caller must evaluate the return value and
* act properly in case of a returned `false,` i.e., it must <b>not</b>
* atttempt to call any further function on the actor, since it might be
* already in the pipe for re-scheduling.
*/
template <class Actor, typename F>
bool fetch_messages(Actor* self, F cb);
/**
* Tries to fetch new messages from the actor's mailbox and to feeds
* them to the given callback. The member function returns `false`
* if no message was read, `true` otherwise.
*
* This member function does not have any side-effect other than removing
* messages from the actor's mailbox.
*/
template <class Actor, typename F>
bool try_fetch_messages(Actor* self, F cb);
/**
* Tries to fetch new messages before a timeout occurs. The message
* can either return `success,` or `no_message,`
* or `indeterminate`. The latter occurs for cooperative scheduled
* operations and means that timeouts are signaled using
* special-purpose messages. In this case, clients have to simply
* wait for the arriving message.
*/
template <class Actor, typename F>
timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time);
/**
* Enqueues given message to the actor's mailbox and takes any
* steps necessary to resume the actor if it's currently blocked.
*/
template <class Actor>
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* host);
/**
* Starts the given actor either by launching a thread or enqueuing
* it to the cooperative scheduler's job queue.
*/
template <class Actor>
void launch(Actor* self);
};
} // namespace policy
} // namespace caf
#endif // CAF_SCHEDULING_POLICY_HPP
......@@ -70,8 +70,7 @@ class response_handle<Self, message, nonblocking_response_handle_tag> {
template <class... Ts>
continue_helper then(Ts&&... xs) const {
behavior bhvr{std::forward<Ts>(xs)...};
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
m_self->set_response_handler(m_mid, behavior{std::forward<Ts>(xs)...});
return {m_mid};
}
......@@ -110,8 +109,7 @@ class response_handle<Self, TypedOutputPair, nonblocking_response_handle_tag> {
>::value,
"match cases are not allowed in this context");
detail::type_checker<TypedOutputPair, Fs...>::check();
behavior tmp{std::move(fs)...};
m_self->bhvr_stack().push_back(std::move(tmp), m_mid);
m_self->set_response_handler(m_mid, behavior{std::move(fs)...});
return {m_mid};
}
......@@ -135,13 +133,13 @@ class response_handle<Self, message, blocking_response_handle_tag> {
}
void await(behavior& bhvr) {
m_self->dequeue_response(bhvr, m_mid);
m_self->dequeue(bhvr, m_mid);
}
template <class... Ts>
void await(Ts&&... xs) const {
behavior bhvr{std::forward<Ts>(xs)...};
m_self->dequeue_response(bhvr, m_mid);
m_self->dequeue(bhvr, m_mid);
}
private:
......@@ -184,7 +182,7 @@ class response_handle<Self, OutputPair, blocking_response_handle_tag> {
"match cases are not allowed in this context");
detail::type_checker<OutputPair, Fs...>::check();
behavior tmp{std::move(fs)...};
m_self->dequeue_response(tmp, m_mid);
m_self->dequeue(tmp, m_mid);
}
private:
......
......@@ -28,17 +28,8 @@
#include "caf/spawn_options.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/policy/no_resume.hpp"
#include "caf/policy/prioritizing.hpp"
#include "caf/policy/no_scheduling.hpp"
#include "caf/policy/actor_policies.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/policy/event_based_resume.hpp"
#include "caf/policy/cooperative_scheduling.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/proper_actor.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/detail/implicit_conversions.hpp"
......@@ -46,12 +37,6 @@ namespace caf {
class execution_unit;
/**
* Marker interface that prevents `spawn` from wrapping the a class
* in a `proper_actor` when spawning new instances.
*/
class spawn_as_is {};
/**
* Returns a newly spawned instance of type `C` using `args...` as constructor
* arguments. The instance will be added to the job list of `host`. However,
......@@ -68,41 +53,17 @@ intrusive_ptr<C> spawn_impl(execution_unit* host,
static_assert(is_unbound(Os),
"top-level spawns cannot have monitor or link flag");
CAF_LOGF_TRACE("");
using scheduling_policy =
typename std::conditional<
has_detach_flag(Os) || has_blocking_api_flag(Os),
policy::no_scheduling,
policy::cooperative_scheduling
>::type;
using priority_policy =
typename std::conditional<
has_priority_aware_flag(Os),
policy::prioritizing,
policy::not_prioritizing
>::type;
using resume_policy =
typename std::conditional<
has_blocking_api_flag(Os),
policy::no_resume,
policy::event_based_resume
>::type;
using policy_token =
policy::actor_policies<
scheduling_policy,
priority_policy,
resume_policy
>;
using actor_impl =
typename std::conditional<
std::is_base_of<spawn_as_is, C>::value,
C,
detail::proper_actor<C, policy_token>
>::type;
auto ptr = make_counted<actor_impl>(std::forward<Ts>(args)...);
auto ptr = make_counted<C>(std::forward<Ts>(args)...);
CAF_LOGF_DEBUG("spawned actor with ID " << ptr->id());
CAF_PUSH_AID(ptr->id());
if (has_priority_aware_flag(Os)) {
ptr->is_priority_aware(true);
}
if (has_detach_flag(Os) || has_blocking_api_flag(Os)) {
ptr->is_detached(true);
}
before_launch_fun(ptr.get());
ptr->launch(has_hide_flag(Os), has_lazy_init_flag(Os), host);
ptr->launch(host, has_lazy_init_flag(Os), has_hide_flag(Os));
return ptr;
}
......
......@@ -72,6 +72,11 @@ std::string to_string(const node_id& what);
*/
std::string to_string(const atom_value& what);
/**
* @relates mailbox_element
*/
std::string to_string(const mailbox_element& what);
/**
* @relates optional
*/
......
......@@ -207,11 +207,17 @@ class typed_behavior {
return m_bhvr.timeout();
}
/** @cond PRIVATE */
behavior& unbox() {
return m_bhvr;
}
/** @endcond */
private:
typed_behavior() = default;
behavior& unbox() { return m_bhvr; }
template <class... Ts>
void set(intrusive_ptr<detail::default_behavior_impl<std::tuple<Ts...>>> bp) {
using mpi =
......
......@@ -23,10 +23,9 @@
#include "caf/replies_to.hpp"
#include "caf/local_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
namespace caf {
......@@ -35,13 +34,11 @@ namespace caf {
* checking. This is the recommended base class for user-defined actors and is
* used implicitly when spawning typed, functor-based actors without the
* `blocking_api` flag.
* @extends mailbox_based_actor
* @extends local_actor
*/
template <class... Sigs>
class typed_event_based_actor : public
extend<mailbox_based_actor, typed_event_based_actor<Sigs...>>::template
with<mixin::behavior_stack_based<typed_behavior<Sigs...>>::template impl,
mixin::sync_sender<nonblocking_response_handle_tag>::template impl> {
class typed_event_based_actor
: public abstract_event_based_actor<typed_behavior<Sigs...>, true> {
public:
using signatures = detail::type_list<Sigs...>;
......@@ -51,6 +48,19 @@ class typed_event_based_actor : public
return {Sigs::static_type_name()...};
}
void initialize() override {
this->is_initialized(true);
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior, "
<< "has_behavior() = "
<< std::boolalpha << this->has_behavior());
if (bhvr) {
// make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
this->do_become(std::move(bhvr.unbox()), true);
}
}
protected:
virtual behavior_type make_behavior() = 0;
};
......
......@@ -39,7 +39,6 @@
#include "caf/policy/work_stealing.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/proper_actor.hpp"
namespace caf {
namespace scheduler {
......@@ -52,10 +51,6 @@ namespace {
using hrc = std::chrono::high_resolution_clock;
using timer_actor_policies = policy::actor_policies<policy::no_scheduling,
policy::not_prioritizing,
policy::no_resume>;
struct delayed_msg {
actor_addr from;
channel to;
......@@ -75,17 +70,22 @@ inline void insert_dmsg(Map& storage, const duration& d, Ts&&... xs) {
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
}
class timer_actor : public detail::proper_actor<blocking_actor,
timer_actor_policies>,
public spawn_as_is {
class timer_actor : public blocking_actor {
public:
inline mailbox_element_ptr dequeue() {
await_data();
blocking_actor::await_data();
return next_message();
}
inline mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (scheduling_policy().await_data(this, tp)) {
bool await_data(const hrc::time_point& tp) {
if (has_next_message()) {
return true;
}
return mailbox().synchronized_await(m_mtx, m_cv, tp);
}
mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (await_data(tp)) {
return next_message();
}
return mailbox_element_ptr{};
......@@ -94,7 +94,7 @@ class timer_actor : public detail::proper_actor<blocking_actor,
void act() override {
trap_exit(true);
// setup & local variables
bool done = false;
bool received_exit = false;
mailbox_element_ptr msg_ptr;
std::multimap<hrc::time_point, delayed_msg> messages;
// message handling rules
......@@ -105,14 +105,14 @@ class timer_actor : public detail::proper_actor<blocking_actor,
std::move(to), mid, std::move(msg));
},
[&](const exit_msg&) {
done = true;
received_exit = true;
},
others >> [&] {
CAF_LOG_ERROR("unexpected: " << to_string(msg_ptr->msg));
}
};
// loop
while (!done) {
while (!received_exit) {
while (!msg_ptr) {
if (messages.empty())
msg_ptr = dequeue();
......
......@@ -36,14 +36,18 @@ void actor_companion::on_enqueue(enqueue_handler handler) {
m_on_enqueue = std::move(handler);
}
void actor_companion::enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit*) {
using detail::memory;
auto ptr = mailbox_element::make(sender, mid, std::move(content));
void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
shared_lock<lock_type> guard(m_lock);
if (m_on_enqueue) {
m_on_enqueue(std::move(ptr));
}
}
void actor_companion::enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* eu) {
using detail::memory;
auto ptr = mailbox_element::make(sender, mid, std::move(content));
enqueue(std::move(ptr), eu);
}
} // namespace caf
......@@ -77,11 +77,13 @@ behavior_impl::behavior_impl(duration tout)
bhvr_invoke_result behavior_impl::invoke(message& msg) {
auto msg_token = msg.type_token();
bhvr_invoke_result res;
std::find_if(m_begin, m_end, [&](const match_case_info& x) {
return (x.has_wildcard || x.type_token == msg_token)
&& x.ptr->invoke(res, msg) != match_case::no_match;
});
for (auto i = m_begin; i != m_end; ++i) {
if ((i->has_wildcard || i->type_token == msg_token)
&& i->ptr->invoke(res, msg) != match_case::no_match) {
return res;
}
}
return none;
}
void behavior_impl::handle_timeout() {
......
......@@ -28,63 +28,18 @@ using namespace std;
namespace caf {
namespace detail {
struct behavior_stack_mover : iterator<output_iterator_tag, void,
void, void, void> {
public:
behavior_stack_mover(behavior_stack* st) : m_stack(st) {}
behavior_stack_mover& operator=(behavior_stack::element_type&& rval) {
m_stack->m_erased_elements.emplace_back(move(rval.first));
return *this;
}
behavior_stack_mover& operator*() {
return *this;
}
behavior_stack_mover& operator++() {
return *this;
}
behavior_stack_mover operator++(int) {
return *this;
}
private:
behavior_stack* m_stack;
};
inline behavior_stack_mover move_iter(behavior_stack* bs) {
return {bs};
}
optional<behavior&> behavior_stack::sync_handler(message_id expected_response) {
if (expected_response.valid()) {
auto e = m_elements.rend();
auto i = find_if(m_elements.rbegin(), e, [=](element_type& val) {
return val.second == expected_response;
});
if (i != e) {
return i->first;
}
}
return none;
}
void behavior_stack::pop_async_back() {
void behavior_stack::pop_back() {
if (m_elements.empty()) {
return;
}
if (!m_elements.back().second.valid()) {
erase_at(m_elements.end() - 1);
} else {
rerase_if([](const element_type& e) { return e.second.valid() == false; });
}
m_erased_elements.push_back(std::move(m_elements.back()));
m_elements.pop_back();
}
void behavior_stack::clear() {
if (m_elements.empty() == false) {
std::move(m_elements.begin(), m_elements.end(), move_iter(this));
if (!m_elements.empty()) {
std::move(m_elements.begin(), m_elements.end(),
std::back_inserter(m_erased_elements));
m_elements.clear();
}
}
......
......@@ -38,11 +38,6 @@ void blocking_actor::await_all_other_actors_done() {
detail::singletons::get_actor_registry()->await_running_count_equal(1);
}
void blocking_actor::cleanup(uint32_t reason) {
m_sync_handler.clear();
mailbox_based_actor::cleanup(reason);
}
void blocking_actor::functor_based::create(blocking_actor*, act_fun fun) {
m_act = fun;
}
......@@ -57,4 +52,35 @@ void blocking_actor::functor_based::cleanup(uint32_t reason) {
blocking_actor::cleanup(reason);
}
void blocking_actor::initialize() {
// nop
}
void blocking_actor::dequeue(behavior& bhvr, message_id mid) {
// try to dequeue from cache first
if (invoke_from_cache(bhvr, mid)) {
return;
}
// requesting an invalid timeout will reset our active timeout
auto timeout_id = request_timeout(bhvr.timeout());
// read incoming messages
for (;;) {
await_data();
auto msg = next_message();
switch (invoke_message(msg, bhvr, mid)) {
case im_success:
reset_timeout(timeout_id);
return;
case im_skipped:
if (msg) {
push_to_cache(std::move(msg));
}
break;
default:
// delete msg
break;
}
}
}
} // namespace caf
......@@ -33,6 +33,19 @@ void event_based_actor::forward_to(const actor& whom,
forward_message(whom, prio);
}
void event_based_actor::initialize() {
is_initialized(true);
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior, "
<< "has_behavior() = "
<< std::boolalpha << this->has_behavior());
if (bhvr) {
// make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
become(std::move(bhvr));
}
}
behavior event_based_actor::functor_based::make_behavior() {
auto res = m_make_behavior(this);
// reset make_behavior_fun to get rid of any
......
This diff is collapsed.
......@@ -136,10 +136,8 @@ class logging_impl : public logging {
class_name.swap(substr);
}
};
char prefix1[] = "caf.detail.proper_actor<";
char prefix2[] = "caf.detail.embedded<";
char prefix1[] = "caf.detail.embedded<";
strip_magic(prefix1, prefix1 + (sizeof(prefix1) - 1));
strip_magic(prefix2, prefix2 + (sizeof(prefix2) - 1));
# else
std::string class_name = c_class_name;
# endif
......
......@@ -19,13 +19,9 @@
#include "caf/scoped_actor.hpp"
#include "caf/policy/no_resume.hpp"
#include "caf/policy/no_scheduling.hpp"
#include "caf/policy/actor_policies.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/spawn_options.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/proper_actor.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
......@@ -33,21 +29,19 @@ namespace caf {
namespace {
struct impl : blocking_actor {
impl() {
is_detached(true);
}
void act() override {
CAF_LOG_ERROR("act() of scoped_actor impl called");
}
};
blocking_actor* alloc() {
using namespace policy;
using policies = actor_policies<no_scheduling, not_prioritizing, no_resume>;
return new detail::proper_actor<impl, policies>;
}
} // namespace <anonymous>
void scoped_actor::init(bool hide_actor) {
m_self.reset(alloc(), false);
m_self.reset(new impl, false);
if (!hide_actor) {
m_prev = CAF_SET_AID(m_self->id());
}
......
......@@ -37,6 +37,7 @@
#include "caf/deserializer.hpp"
#include "caf/skip_message.hpp"
#include "caf/actor_namespace.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/uniform_type_info.hpp"
......@@ -554,6 +555,23 @@ string to_string(const actor& what) {
return to_string(what.address());
}
string to_string(const node_id::host_id_type& node_id) {
std::ostringstream oss;
oss << std::hex;
oss.fill('0');
for (size_t i = 0; i < node_id::host_id_size; ++i) {
oss.width(2);
oss << static_cast<uint32_t>(node_id[i]);
}
return oss.str();
}
string to_string(const node_id& what) {
std::ostringstream oss;
oss << to_string(what.host_id()) << ":" << what.process_id();
return oss.str();
}
string to_string(const atom_value& what) {
auto x = static_cast<uint64_t>(what);
string result;
......@@ -572,21 +590,16 @@ string to_string(const atom_value& what) {
return result;
}
string to_string(const node_id::host_id_type& node_id) {
std::ostringstream oss;
oss << std::hex;
oss.fill('0');
for (size_t i = 0; i < node_id::host_id_size; ++i) {
oss.width(2);
oss << static_cast<uint32_t>(node_id[i]);
}
return oss.str();
}
string to_string(const node_id& what) {
std::ostringstream oss;
oss << to_string(what.host_id()) << ":" << what.process_id();
return oss.str();
std::string to_string(const mailbox_element& what) {
std::string result;
result += "@mailbox_element ( ";
result += to_string(what.sender);
result += ", ";
result += std::to_string(what.mid.integer_value());
result += ", ";
result += to_string(what.msg);
result += " )";
return result;
}
string to_verbose_string(const std::exception& e) {
......
......@@ -28,7 +28,6 @@
#include "caf/local_actor.hpp"
#include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp"
......@@ -53,11 +52,9 @@ using broker_ptr = intrusive_ptr<broker>;
* A broker mediates between actor systems and other components in the network.
* @extends local_actor
*/
class broker : public extend<local_actor>::
with<mixin::behavior_stack_based<behavior>::impl>,
public spawn_as_is {
class broker : public abstract_event_based_actor<behavior, false> {
public:
using super = combined_type;
using super = abstract_event_based_actor<behavior, false>;
using buffer_type = std::vector<char>;
......@@ -225,6 +222,8 @@ class broker : public extend<local_actor>::
/** @cond PRIVATE */
void initialize() override;
template <class F, class... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... xs) {
// provoke compile-time errors early
......@@ -273,10 +272,6 @@ class broker : public extend<local_actor>::
accept_handle add_tcp_doorman(network::native_socket fd);
policy::invoke_message_result invoke_message(mailbox_element_ptr& msg,
behavior& bhvr,
message_id mid);
void invoke_message(mailbox_element_ptr& msg);
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
......@@ -314,7 +309,7 @@ class broker : public extend<local_actor>::
class functor_based;
void launch(bool is_hidden, bool, execution_unit*);
void launch(execution_unit* eu, bool lazy, bool hide);
void cleanup(uint32_t reason);
......@@ -381,8 +376,6 @@ class broker : public extend<local_actor>::
std::map<accept_handle, doorman_pointer> m_doormen;
std::map<connection_handle, scribe_pointer> m_scribes;
policy::invoke_policy m_invoke_policy;
middleman& m_mm;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
};
......
......@@ -69,7 +69,7 @@ class middleman : public detail::abstract_singleton {
}
auto result = make_counted<Impl>(*this);
CAF_REQUIRE(result->unique());
result->launch(true, false, nullptr);
result->launch(nullptr, false, true);
m_named_brokers.insert(std::make_pair(name, result));
return result;
}
......
......@@ -155,18 +155,12 @@ class broker::continuation {
mailbox_element_ptr m_ptr;
};
policy::invoke_message_result broker::invoke_message(mailbox_element_ptr& msg,
behavior& bhvr,
message_id mid) {
return m_invoke_policy.invoke_message(this, msg, bhvr, mid);
}
void broker::invoke_message(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_TARG(ptr->msg, to_string));
if (exit_reason() != exit_reason::not_exited || bhvr_stack().empty()) {
if (exit_reason() != exit_reason::not_exited || !has_behavior()) {
CAF_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = " << planned_exit_reason()
<< ", bhvr_stack().empty() = " << bhvr_stack().empty());
<< ", has_behavior() = " << has_behavior());
if (ptr->mid.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(ptr->sender, ptr->mid);
......@@ -175,22 +169,24 @@ void broker::invoke_message(mailbox_element_ptr& ptr) {
}
// prepare actor for invocation of message handler
try {
auto bhvr = bhvr_stack().back();
auto bid = bhvr_stack().back_id();
switch (invoke_message(ptr, bhvr, bid)) {
case policy::im_success: {
auto& bhvr = this->awaits_response()
? this->awaited_response_handler()
: this->bhvr_stack().back();
auto bid = this->awaited_response_id();
switch (local_actor::invoke_message(ptr, bhvr, bid)) {
case im_success: {
CAF_LOG_DEBUG("handle_message returned hm_msg_handled");
while (!bhvr_stack().empty()
while (has_behavior()
&& planned_exit_reason() == exit_reason::not_exited
&& invoke_message_from_cache()) {
// rinse and repeat
}
break;
}
case policy::im_dropped:
case im_dropped:
CAF_LOG_DEBUG("handle_message returned hm_drop_msg");
break;
case policy::im_skipped: {
case im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
if (ptr) {
m_cache.push_second_back(ptr.release());
......@@ -210,11 +206,13 @@ void broker::invoke_message(mailbox_element_ptr& ptr) {
CAF_LOG_ERROR("broker killed due to an unknown exception");
quit(exit_reason::unhandled_exception);
}
// cleanup if needed
// safe to actually release behaviors now
bhvr_stack().cleanup();
// cleanup actor if needed
if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason());
} else if (bhvr_stack().empty()) {
CAF_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason");
} else if (!has_behavior()) {
CAF_LOG_DEBUG("no behavior set, quit for normal exit reason");
quit(exit_reason::normal);
cleanup(planned_exit_reason());
}
......@@ -231,12 +229,14 @@ void broker::invoke_message(const actor_addr& v0, message_id v1, message& v2) {
bool broker::invoke_message_from_cache() {
CAF_LOG_TRACE("");
auto bhvr = bhvr_stack().back();
auto bid = bhvr_stack().back_id();
auto& bhvr = this->awaits_response()
? this->awaited_response_handler()
: this->bhvr_stack().back();
auto bid = this->awaited_response_id();
auto i = m_cache.second_begin();
auto e = m_cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
return m_cache.invoke(this, i, e, bhvr, bid);
return m_cache.invoke(static_cast<local_actor*>(this), i, e, bhvr, bid);
}
void broker::write(connection_handle hdl, size_t bs, const void* buf) {
......@@ -280,7 +280,7 @@ void broker::on_exit() {
// nop
}
void broker::launch(bool is_hidden, bool, execution_unit*) {
void broker::launch(execution_unit*, bool, bool is_hidden) {
// add implicit reference count held by the middleman
ref();
is_registered(!is_hidden);
......@@ -362,6 +362,10 @@ std::vector<connection_handle> broker::connections() const {
return result;
}
void broker::initialize() {
// nop
}
broker::functor_based::~functor_based() {
// nop
}
......
......@@ -29,6 +29,7 @@
#include "caf/config.hpp"
#include "caf/node_id.hpp"
#include "caf/announce.hpp"
#include "caf/exception.hpp"
#include "caf/to_string.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp"
......
......@@ -20,6 +20,7 @@
#include "caf/io/publish.hpp"
#include "caf/send.hpp"
#include "caf/exception.hpp"
#include "caf/actor_cast.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp"
......
......@@ -28,6 +28,7 @@
#include <algorithm>
#include "caf/send.hpp"
#include "caf/exception.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/binary_deserializer.hpp"
......
......@@ -19,7 +19,7 @@ add_unit_test(ripemd_160)
add_unit_test(variant)
add_unit_test(atom)
add_unit_test(metaprogramming)
add_unit_test(intrusive_containers)
#add_unit_test(intrusive_containers)
add_unit_test(match)
add_unit_test(message)
add_unit_test(serialization)
......
......@@ -29,6 +29,7 @@ using testee = typed_actor<replies_to<abc_atom>::with<int>>;
testee::behavior_type testee_impl(testee::pointer self) {
return {
[=](abc_atom) {
CAF_CHECKPOINT();
self->quit();
return 42;
}
......@@ -39,7 +40,7 @@ void test_typed_atom_interface() {
CAF_CHECKPOINT();
scoped_actor self;
auto tst = spawn_typed(testee_impl);
self->sync_send(tst, abc_atom()).await(
self->sync_send(tst, abc_atom::value).await(
[](int i) {
CAF_CHECK_EQUAL(i, 42);
},
......
......@@ -69,26 +69,26 @@ struct pseudo_actor {
return mbox;
}
static policy::invoke_message_result invoke_message(uptr& ptr, int i) {
static invoke_message_result invoke_message(uptr& ptr, int i) {
if (ptr->value == 1) {
ptr.reset();
return policy::im_dropped;
return im_dropped;
}
if (ptr->value == i) {
// call reset on some of our messages
if (ptr->is_high_priority()) {
ptr.reset();
}
return policy::im_success;
return im_success;
}
return policy::im_skipped;
return im_skipped;
}
template <class Policy>
policy::invoke_message_result invoke_message(uptr& ptr, Policy& policy, int i,
invoke_message_result invoke_message(uptr& ptr, Policy& policy, int i,
std::vector<int>& remaining) {
auto res = invoke_message(ptr, i);
if (res == policy::im_success && !remaining.empty()) {
if (res == im_success && !remaining.empty()) {
auto next = remaining.front();
remaining.erase(remaining.begin());
policy.invoke_from_cache(this, policy, next, remaining);
......
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