Commit 2f45e494 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #251 from actor-framework/topic/remove-actor-policies

Remove actor policies and `proper_actor`
parents 280636f6 7ee835ac
......@@ -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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_INVOKE_POLICY_HPP
#define CAF_POLICY_INVOKE_POLICY_HPP
#include <memory>
#include <type_traits>
#include "caf/none.hpp"
#include "caf/on.hpp"
#include "caf/behavior.hpp"
#include "caf/to_string.hpp"
#include "caf/message_id.hpp"
#include "caf/exit_reason.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/response_promise.hpp"
#include "caf/detail/memory.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/scope_guard.hpp"
namespace caf {
namespace policy {
enum invoke_message_result {
im_success,
im_skipped,
im_dropped
};
/**
* Base class for invoke policies.
*/
class invoke_policy {
public:
enum class msg_type {
normal_exit, // an exit message with normal exit reason
non_normal_exit, // an exit message with abnormal exit reason
expired_timeout, // an 'old & obsolete' timeout
inactive_timeout, // a currently inactive timeout
expired_sync_response, // a sync response that already timed out
timeout, // triggers currently active timeout
ordinary, // an asynchronous message or sync. request
sync_response // a synchronous response
};
// the workflow of invoke_message (im) is as follows:
// - should_skip? if yes: return im_skipped
// - msg is ordinary message? if yes:
// - begin(...) -> prepares a self for message handling
// - self could process message?
// - yes: cleanup()
// - no: revert(...) -> set self back to state it had before begin()
template <class Actor>
invoke_message_result invoke_message(Actor* self, mailbox_element_ptr& node,
behavior& fun,
message_id awaited_response) {
CAF_LOG_TRACE("");
switch (this->filter_msg(self, *node)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return im_dropped;
case msg_type::expired_sync_response:
CAF_LOG_DEBUG("dropped expired sync response");
return im_dropped;
case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
case msg_type::inactive_timeout:
CAF_LOG_DEBUG("skipped inactive timeout message");
return im_skipped;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
// by calling self->quit(...)
return im_success;
case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message");
auto& tm = node->msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
}
return im_success;
}
case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(node->msg, to_string) << ", "
<< CAF_MARG(node->mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node->mid == awaited_response) {
bool is_sync_tout = node->msg.match_elements<sync_timeout_msg>();
node.swap(self->current_mailbox_element());
auto res = invoke_fun(self, fun);
node.swap(self->current_mailbox_element());
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
if (!res) {
if (is_sync_tout) {
CAF_LOG_WARNING("sync timeout occured in actor "
<< "with ID " << self->id());
self->handle_sync_timeout();
} else {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << self->id());
self->handle_sync_failure();
}
}
return im_success;
}
return im_skipped;
case msg_type::ordinary:
if (!awaited_response.valid()) {
node.swap(self->current_mailbox_element());
auto res = invoke_fun(self, fun);
node.swap(self->current_mailbox_element());
if (res) {
return im_success;
}
}
CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: "
<< awaited_response.integer_value());
return im_skipped;
}
// should be unreachable
CAF_CRITICAL("invalid message type");
}
template <class Actor>
response_promise fetch_response_promise(Actor* cl, int) {
return cl->make_response_promise();
}
template <class Actor>
response_promise fetch_response_promise(Actor*, response_promise& hdl) {
return std::move(hdl);
}
// - extracts response message from handler
// - returns true if fun was successfully invoked
template <class Actor, class Fun, class MaybeResponseHdl = int>
optional<message> invoke_fun(Actor* self, Fun& fun,
MaybeResponseHdl hdl = MaybeResponseHdl{}) {
# ifdef CAF_LOG_LEVEL
auto msg = to_string(self->current_mailbox_element()->msg);
# endif
auto mid = self->current_mailbox_element()->mid;
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", " << CAF_ARG(msg));
auto res = fun(self->current_mailbox_element()->msg);
CAF_LOG_DEBUG(msg << " => " << to_string(res));
if (!res) {
return none;
}
if (res->empty()) {
// make sure synchronous requests always receive a response;
// note: !current_mailbox_element() means client has forwarded the request
auto& ptr = self->current_mailbox_element();
if (ptr) {
mid = ptr->mid;
if (mid.is_request() && !mid.is_answered()) {
CAF_LOG_WARNING("actor with ID " << self->id()
<< " did not reply to a synchronous request message");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(make_message(unit));
}
}
}
return res;
}
CAF_LOGF_DEBUG("res = " << to_string(*res));
if (handle_message_id_res(self, *res, none)) {
return message{};
}
// respond by using the result of 'fun'
CAF_LOG_DEBUG("respond via response_promise");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(std::move(*res));
// inform caller about success by returning not none
return message{};
}
return res;
}
private:
// enables `return sync_send(...).then(...)`
template <class Actor>
bool handle_message_id_res(Actor* self, message& res,
optional<response_promise> hdl) {
if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) {
CAF_LOG_DEBUG("message handler returned a message id wrapper");
auto id = res.get_as<uint64_t>(1);
auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->sync_handler(msg_id);
// install a behavior that calls the user-defined behavior
// and using the result of its inner behavior as response
if (ref_opt) {
response_promise fhdl = hdl ? *hdl : self->make_response_promise();
behavior inner = *ref_opt;
ref_opt->assign(
others >> [=] {
// inner is const inside this lambda and mutable a C++14 feature
behavior cpy = inner;
auto inner_res = cpy(self->current_message());
if (inner_res && !handle_message_id_res(self, *inner_res, fhdl)) {
fhdl.deliver(*inner_res);
}
}
);
return true;
}
}
return false;
}
// identifies 'special' messages that should not be processed normally:
// - system messages such as exit_msg and timeout_msg
// - expired synchronous response messages
template <class Actor>
msg_type filter_msg(Actor* self, mailbox_element& node) {
const message& msg = node.msg;
auto mid = node.mid;
if (mid.is_response()) {
return self->awaits(mid) ? msg_type::sync_response
: msg_type::expired_sync_response;
}
if (msg.size() != 1) {
return msg_type::ordinary;
}
if (msg.match_element<timeout_msg>(0)) {
auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_REQUIRE(!mid.valid());
if (self->is_active_timeout(tid)) {
return msg_type::timeout;
}
return self->waits_for_timeout(tid) ? msg_type::inactive_timeout
: msg_type::expired_timeout;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_REQUIRE(!mid.valid());
// make sure to get rid of attachables if they're no longer needed
self->unlink_from(em.source);
if (self->trap_exit() == false) {
if (em.reason != exit_reason::normal) {
self->quit(em.reason);
return msg_type::non_normal_exit;
}
return msg_type::normal_exit;
}
}
return msg_type::ordinary;
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_INVOKE_POLICY_HPP
#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
......
......@@ -18,6 +18,7 @@
******************************************************************************/
#include <string>
#include "caf/all.hpp"
#include "caf/atom.hpp"
#include "caf/actor_cast.hpp"
......@@ -25,18 +26,24 @@
#include "caf/default_attachable.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
// local actors are created with a reference count of one that is adjusted
// later on in spawn(); this prevents subtle bugs that lead to segfaults,
// e.g., when calling address() in the ctor of a derived class
local_actor::local_actor() : m_planned_exit_reason(exit_reason::not_exited) {
local_actor::local_actor()
: m_planned_exit_reason(exit_reason::not_exited),
m_timeout_id(0) {
// nop
}
local_actor::~local_actor() {
// nop
if (!m_mailbox.closed()) {
detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
}
}
void local_actor::monitor(const actor_addr& whom) {
......@@ -81,6 +88,10 @@ void local_actor::leave(const group& what) {
}
}
void local_actor::on_exit() {
// nop
}
std::vector<group> local_actor::joined_groups() const {
std::vector<group> result;
result.reserve(20);
......@@ -121,6 +132,671 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
dest->enqueue(std::move(m_current_element), host());
}
uint32_t local_actor::request_timeout(const duration& d) {
if (!d.valid()) {
has_timeout(false);
return 0;
}
has_timeout(true);
auto result = ++m_timeout_id;
auto msg = make_message(timeout_msg{++m_timeout_id});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
enqueue(address(), invalid_message_id, std::move(msg), host());
} else {
delayed_send(this, d, std::move(msg));
}
return result;
}
void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) {
if (!is_active_timeout(timeout_id)) {
return;
}
bhvr.handle_timeout();
if (m_bhvr_stack.empty() || m_bhvr_stack.back() != bhvr) {
return;
}
// auto-remove behavior for blocking actors
if (is_blocking()) {
CAF_REQUIRE(m_bhvr_stack.back() == bhvr);
m_bhvr_stack.pop_back();
return;
}
// request next timeout for non-blocking (i.e. event-based) actors
// if behavior stack was not modified by calling become()/unbecome()
if (m_bhvr_stack.back() == bhvr) {
CAF_REQUIRE(bhvr.timeout().valid());
request_timeout(bhvr.timeout());
}
}
void local_actor::reset_timeout(uint32_t timeout_id) {
if (is_active_timeout(timeout_id)) {
has_timeout(false);
}
}
bool local_actor::is_active_timeout(uint32_t tid) const {
return has_timeout() && m_timeout_id == tid;
}
uint32_t local_actor::active_timeout_id() const {
return m_timeout_id;
}
enum class msg_type {
normal_exit, // an exit message with normal exit reason
non_normal_exit, // an exit message with abnormal exit reason
expired_timeout, // an 'old & obsolete' timeout
expired_sync_response, // a sync response that already timed out
timeout, // triggers currently active timeout
ordinary, // an asynchronous message or sync. request
sync_response // a synchronous response
};
msg_type filter_msg(local_actor* self, mailbox_element& node) {
const message& msg = node.msg;
auto mid = node.mid;
if (mid.is_response()) {
return self->awaits(mid) ? msg_type::sync_response
: msg_type::expired_sync_response;
}
if (msg.size() != 1) {
return msg_type::ordinary;
}
if (msg.match_element<timeout_msg>(0)) {
auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_REQUIRE(!mid.valid());
if (self->is_active_timeout(tid)) {
return msg_type::timeout;
}
return msg_type::expired_timeout;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_REQUIRE(!mid.valid());
// make sure to get rid of attachables if they're no longer needed
self->unlink_from(em.source);
if (self->trap_exit() == false) {
if (em.reason != exit_reason::normal) {
self->quit(em.reason);
return msg_type::non_normal_exit;
}
return msg_type::normal_exit;
}
}
return msg_type::ordinary;
}
response_promise fetch_response_promise(local_actor* self, int) {
return self->make_response_promise();
}
response_promise fetch_response_promise(local_actor*, response_promise& hdl) {
return std::move(hdl);
}
// enables `return sync_send(...).then(...)`
bool handle_message_id_res(local_actor* self, message& res,
optional<response_promise> hdl) {
if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) {
CAF_LOGF_DEBUG("message handler returned a message id wrapper");
auto id = res.get_as<uint64_t>(1);
auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->find_pending_response(msg_id);
// install a behavior that calls the user-defined behavior
// and using the result of its inner behavior as response
if (ref_opt) {
response_promise fhdl = hdl ? *hdl : self->make_response_promise();
behavior& ref = std::get<1>(*ref_opt);
behavior inner = ref;
ref.assign(
others >> [=] {
// inner is const inside this lambda and mutable a C++14 feature
behavior cpy = inner;
auto inner_res = cpy(self->current_message());
if (inner_res && !handle_message_id_res(self, *inner_res, fhdl)) {
fhdl.deliver(*inner_res);
}
}
);
return true;
}
}
return false;
}
// - extracts response message from handler
// - returns true if fun was successfully invoked
template <class Handle = int>
optional<message> post_process_invoke_res(local_actor* self,
message_id mid,
optional<message>&& res,
Handle hdl = Handle{}) {
CAF_LOGF_TRACE(CAF_MARG(mid, integer_value) << ", " << CAF_TSARG(res));
if (!res) {
return none;
}
if (res->empty()) {
// make sure synchronous requests always receive a response;
// note: !current_mailbox_element() means client has forwarded the request
auto& ptr = self->current_mailbox_element();
if (ptr) {
mid = ptr->mid;
if (mid.is_request() && !mid.is_answered()) {
CAF_LOGF_WARNING("actor with ID " << self->id()
<< " did not reply to a synchronous request message");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(make_message(unit));
}
}
}
return res;
}
if (handle_message_id_res(self, *res, none)) {
return message{};
}
// respond by using the result of 'fun'
CAF_LOGF_DEBUG("respond via response_promise");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(std::move(*res));
// inform caller about success by returning not none
return message{};
}
return res;
}
invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
behavior& fun,
message_id awaited_id) {
CAF_REQUIRE(ptr != nullptr);
CAF_LOG_TRACE(CAF_TSARG(*ptr) << ", " << CAF_MARG(awaited_id, integer_value));
switch (filter_msg(this, *ptr)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return im_dropped;
case msg_type::expired_sync_response:
CAF_LOG_DEBUG("dropped expired sync response");
return im_dropped;
case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
// by calling quit(...)
return im_success;
case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message");
auto& tm = ptr->msg.get_as<timeout_msg>(0);
handle_timeout(fun, tm.timeout_id);
if (awaited_id.valid()) {
mark_arrived(awaited_id);
}
return im_success;
}
case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(ptr->msg, to_string) << ", "
<< CAF_MARG(ptr->mid, integer_value) << ", "
<< CAF_MARG(awaited_id, integer_value));
if (awaited_id.valid() && ptr->mid == awaited_id) {
bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>();
ptr.swap(current_mailbox_element());
auto mid = current_mailbox_element()->mid;
auto res = post_process_invoke_res(this, mid,
fun(current_mailbox_element()->msg));
ptr.swap(current_mailbox_element());
mark_arrived(awaited_id);
if (!res) {
if (is_sync_tout) {
CAF_LOG_WARNING("sync timeout occured in actor "
<< "with ID " << id());
handle_sync_timeout();
} else {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << id());
handle_sync_failure();
}
}
return im_success;
}
return im_skipped;
case msg_type::ordinary:
if (!awaited_id.valid()) {
ptr.swap(current_mailbox_element());
auto mid = current_mailbox_element()->mid;
auto res = post_process_invoke_res(this, mid,
fun(current_mailbox_element()->msg));
ptr.swap(current_mailbox_element());
if (res) {
return im_success;
}
}
CAF_LOG_DEBUG_IF(awaited_id.valid(),
"ignored message; await response: "
<< awaited_id.integer_value());
return im_skipped;
}
// should be unreachable
CAF_CRITICAL("invalid message type");
}
struct pending_response_predicate {
public:
pending_response_predicate(message_id mid) : m_mid(mid) {
// nop
}
bool operator()(const local_actor::pending_response& pr) const {
return std::get<0>(pr) == m_mid;
}
private:
message_id m_mid;
};
message_id local_actor::new_request_id() {
auto result = ++m_last_request_id;
m_pending_responses.push_front(std::make_pair(result.response_id(),
behavior{}));
return result;
}
void local_actor::mark_arrived(message_id mid) {
CAF_REQUIRE(mid.is_response());
pending_response_predicate predicate{mid};
m_pending_responses.remove_if(predicate);
}
bool local_actor::awaits_response() const {
return !m_pending_responses.empty();
}
bool local_actor::awaits(message_id mid) const {
CAF_REQUIRE(mid.is_response());
pending_response_predicate predicate{mid};
return std::any_of(m_pending_responses.begin(), m_pending_responses.end(),
predicate);
}
optional<local_actor::pending_response&>
local_actor::find_pending_response(message_id mid) {
pending_response_predicate predicate{mid};
auto last = m_pending_responses.end();
auto i = std::find_if(m_pending_responses.begin(), last, predicate);
if (i == last) {
return none;
}
return *i;
}
void local_actor::set_response_handler(message_id response_id, behavior bhvr) {
auto pr = find_pending_response(response_id);
if (pr) {
pr->second = std::move(bhvr);
}
}
behavior& local_actor::awaited_response_handler() {
return m_pending_responses.front().second;
}
message_id local_actor::awaited_response_id() {
return m_pending_responses.empty()
? message_id::make()
: m_pending_responses.front().first;
}
void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
is_registered(!hide);
if (is_detached()) {
// actor lives in its own thread
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(lazy) << ", " << CAF_ARG(hide));
intrusive_ptr<local_actor> mself{this};
attach_to_scheduler();
std::thread([=] {
CAF_PUSH_AID(id());
CAF_LOG_TRACE("");
auto max_throughput = std::numeric_limits<size_t>::max();
while (resume(nullptr, max_throughput) != resumable::done) {
// await new data before resuming actor
await_data();
CAF_REQUIRE(mailbox().blocked() == false);
}
detach_from_scheduler();
}).detach();
return;
}
// actor is cooperatively scheduled
attach_to_scheduler();
if (lazy) {
mailbox().try_block();
return;
}
if (eu) {
eu->exec_later(this);
} else {
detail::singletons::get_scheduling_coordinator()->enqueue(this);
}
}
void local_actor::enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) {
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
}
void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
if (is_detached()) {
// actor lives in its own thread
auto mid = ptr->mid;
auto sender = ptr->sender;
// returns false if mailbox has been closed
if (!mailbox().synchronized_enqueue(m_mtx,
m_cv,
ptr.release())) {
if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid);
}
}
return;
}
// actor is cooperatively scheduled
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// re-schedule actor
if (eu) {
eu->exec_later(this);
} else {
detail::singletons::get_scheduling_coordinator()->enqueue(this);
}
break;
}
case detail::enqueue_result::queue_closed: {
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do
break;
}
}
void local_actor::attach_to_scheduler() {
ref();
}
void local_actor::detach_from_scheduler() {
deref();
}
resumable::resume_result local_actor::resume(execution_unit* eu,
size_t max_throughput) {
CAF_LOG_TRACE("");
if (is_blocking()) {
// actor lives in its own thread
CAF_REQUIRE(dynamic_cast<blocking_actor*>(this) != 0);
auto self = static_cast<blocking_actor*>(this);
uint32_t rsn = exit_reason::normal;
std::exception_ptr eptr = nullptr;
try {
self->act();
}
catch (actor_exited& e) {
rsn = e.reason();
}
catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = *opt_reason;
}
self->planned_exit_reason(rsn);
try {
self->on_exit();
}
catch (...) {
// simply ignore exception
}
// exit reason might have been changed by on_exit()
self->cleanup(self->planned_exit_reason());
return resumable::done;
}
// actor is cooperatively scheduled
host(eu);
auto done_cb = [&]() -> bool {
CAF_LOG_TRACE("");
bhvr_stack().clear();
bhvr_stack().cleanup();
on_exit();
if (has_behavior()) {
CAF_LOG_DEBUG("on_exit did set a new behavior");
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
auto rsn = planned_exit_reason();
if (rsn == exit_reason::not_exited) {
rsn = exit_reason::normal;
planned_exit_reason(rsn);
}
cleanup(rsn);
return true;
};
auto actor_done = [&]() -> bool {
if (!has_behavior()
|| 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(!is_initialized()
|| (has_behavior()
&& planned_exit_reason() == exit_reason::not_exited));
std::exception_ptr eptr = nullptr;
try {
if (!is_initialized()) {
CAF_LOG_DEBUG("initialize actor");
initialize();
if (actor_done()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return resumable::resume_result::done;
}
}
auto had_tout = has_timeout();
auto tout = active_timeout_id();
int handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (had_tout && handled_msgs > 0 && tout == active_timeout_id()) {
request_timeout(get_behavior().timeout());
}
};
// max_throughput = 0 means infinite
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = next_message();
if (ptr) {
auto& bhvr = awaits_response()
? awaited_response_handler()
: bhvr_stack().back();
auto mid = awaited_response_id();
switch (invoke_message(ptr, bhvr, mid)) {
case im_success:
bhvr_stack().cleanup();
++handled_msgs;
if (actor_done()) {
CAF_LOG_DEBUG("actor exited");
return resumable::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 (invoke_from_cache()) {
if (actor_done()) {
CAF_LOG_DEBUG("actor exited");
return resumable::resume_result::done;
}
}
break;
case im_skipped:
CAF_REQUIRE(ptr != nullptr);
push_to_cache(std::move(ptr));
break;
case im_dropped:
// destroy msg
break;
}
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
if (mailbox().try_block()) {
reset_timeout_if_needed();
return resumable::awaiting_message;
}
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
if (!has_next_message() && 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 (exit_reason() == exit_reason::not_exited) {
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 (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception();
}
catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception");
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = handle(eptr);
if (opt_reason) {
// use exit reason defined by custom handler
planned_exit_reason(*opt_reason);
}
}
if (!actor_done()) {
// actor has been "revived", try running it again later
return resumable::resume_later;
}
return resumable::done;
}
mailbox_element_ptr local_actor::next_message() {
if (!is_priority_aware()) {
return mailbox_element_ptr{mailbox().try_pop()};
}
// we partition the mailbox into four segments in this case:
// <-------- !was_skipped --------> | <-------- was_skipped -------->
// <-- high prio --><-- low prio -->|<-- high prio --><-- low prio -->
auto& cache = 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 = 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 = mailbox().try_pop();
}
}
mailbox_element_ptr result;
if (!cache.first_empty()) {
result.reset(cache.take_first_front());
}
return result;
}
bool local_actor::has_next_message() {
if (!is_priority_aware()) {
return m_mailbox.can_fetch_more();
}
auto& mbox = mailbox();
auto& cache = mbox.cache();
return !cache.first_empty() || mbox.can_fetch_more();
}
void local_actor::push_to_cache(mailbox_element_ptr ptr) {
if (!is_priority_aware()) {
mailbox().cache().push_second_back(ptr.release());
return;
}
auto high_prio = [](const mailbox_element& val) {
return val.is_high_priority();
};
auto& cache = 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());
}
bool local_actor::invoke_from_cache() {
return invoke_from_cache(get_behavior(), awaited_response_id());
}
bool local_actor::invoke_from_cache(behavior& bhvr, message_id mid) {
auto& cache = 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(this, i, e, bhvr, mid);
}
void local_actor::do_become(behavior bhvr, bool discard_old) {
if (discard_old) {
m_bhvr_stack.pop_back();
}
// request_timeout simply resets the timeout when it's invalid
request_timeout(bhvr.timeout());
m_bhvr_stack.push_back(std::move(bhvr));
}
void local_actor::await_data() {
if (has_next_message()) {
return;
}
mailbox().synchronized_await(m_mtx, m_cv);
}
void local_actor::send_impl(message_priority prio, abstract_channel* dest,
message what) {
if (!dest) {
......@@ -156,6 +832,8 @@ response_promise local_actor::make_response_promise() {
void local_actor::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
abstract_actor::cleanup(reason);
// tell registry we're done
is_registered(false);
......
......@@ -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