Commit ef0b744f authored by Dominik Charousset's avatar Dominik Charousset

restructured actor class hierarchy

this patch removes several classes and mixins in the actor class hierarchy and
introduces a new mixin facility (`extend<...>::with<...>`);
this patch also refactores several function signatures to take `actor_ptr`
arguments rather than `actor*`
parent c3beb4d6
......@@ -85,11 +85,9 @@ if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
endif ("${CMAKE_BUILD_TYPE}" STREQUAL "")
set(LIBCPPA_SRC
src/abstract_scheduled_actor.cpp
src/abstract_tuple.cpp
src/actor.cpp
src/actor_addressing.cpp
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/any_tuple.cpp
......
......@@ -18,10 +18,7 @@ cppa/cow_tuple.hpp
cppa/cppa.hpp
cppa/cppa_fwd.hpp
cppa/deserializer.hpp
cppa/detail/abstract_actor.hpp
cppa/detail/abstract_scheduled_actor.hpp
cppa/detail/abstract_tuple.hpp
cppa/detail/actor_count.hpp
cppa/detail/actor_registry.hpp
cppa/detail/atom_val.hpp
cppa/detail/behavior_impl.hpp
......@@ -56,7 +53,7 @@ cppa/detail/scheduled_actor_dummy.hpp
cppa/detail/serialize_tuple.hpp
cppa/detail/singleton_manager.hpp
cppa/detail/singleton_mixin.hpp
cppa/detail/stacked_actor_mixin.hpp
cppa/stacked.hpp
cppa/detail/swap_bytes.hpp
cppa/detail/tdata.hpp
cppa/detail/thread_pool_scheduler.hpp
......@@ -72,7 +69,7 @@ cppa/detail/uniform_type_info_map.hpp
cppa/detail/value_guard.hpp
cppa/detail/yield_interface.hpp
cppa/either.hpp
cppa/enable_weak_ptr_mixin.hpp
cppa/enable_weak_ptr.hpp
cppa/event_based_actor.hpp
cppa/exception.hpp
cppa/exit_reason.hpp
......@@ -197,11 +194,9 @@ examples/type_system/announce_2.cpp
examples/type_system/announce_3.cpp
examples/type_system/announce_4.cpp
examples/type_system/announce_5.cpp
src/abstract_scheduled_actor.cpp
src/abstract_tuple.cpp
src/actor.cpp
src/actor_addressing.cpp
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/any_tuple.cpp
......@@ -288,8 +283,11 @@ unit_testing/test_tuple.cpp
unit_testing/test_metaprogramming.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
cppa/memory_cached_mixin.hpp
unit_testing/test.cpp
src/exit_reason.cpp
src/on.cpp
cppa/spawn_options.hpp
cppa/extend.hpp
cppa/detail/sync_request_bouncer.hpp
cppa/memory_cached.hpp
cppa/singletons.hpp
......@@ -31,7 +31,10 @@
#ifndef CPPA_ACTOR_HPP
#define CPPA_ACTOR_HPP
#include <mutex>
#include <atomic>
#include <memory>
#include <vector>
#include <cstdint>
#include <type_traits>
......@@ -40,11 +43,15 @@
#include "cppa/cppa_fwd.hpp"
#include "cppa/attachable.hpp"
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/util/rm_ref.hpp"
namespace cppa {
class actor;
class self_type;
class serializer;
class deserializer;
......@@ -54,6 +61,12 @@ class deserializer;
*/
typedef std::uint32_t actor_id;
/**
* @brief A smart pointer type that manages instances of {@link actor}.
* @relates actor
*/
typedef intrusive_ptr<actor> actor_ptr;
/**
* @brief Base class for all actor implementations.
*/
......@@ -66,13 +79,15 @@ class actor : public channel {
* this actor is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_enqueue(actor* sender, any_tuple msg);
virtual bool chained_enqueue(const actor_ptr& sender, any_tuple msg);
/**
* @brief Enqueues @p msg as a synchronous message to this actor's mailbox.
* @pre <tt>id.valid()</tt>
*/
virtual void sync_enqueue(actor* sender, message_id_t id, any_tuple msg) = 0;
virtual void sync_enqueue(const actor_ptr& sender,
message_id_t id,
any_tuple msg) = 0;
/**
* @brief Enqueues @p msg as a reply to @p request_id
......@@ -80,7 +95,7 @@ class actor : public channel {
* this actor is an scheduled actor that successfully changed
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_sync_enqueue(actor* sender,
virtual bool chained_sync_enqueue(const actor_ptr& sender,
message_id_t id,
any_tuple msg);
......@@ -95,46 +110,40 @@ class actor : public channel {
* otherwise (actor already exited) @p false.
* @warning The actor takes ownership of @p ptr.
*/
virtual bool attach(attachable* ptr) = 0;
virtual bool attach(attachable_ptr ptr);
/**
* @brief Convenience function that attaches the functor
* or function @p ftor to this actor.
* or function @p f to this actor.
*
* The actor executes <tt>ftor()</tt> on exit, or immediatley
* The actor executes <tt>f()</tt> on exit, or immediatley
* if it already finished execution.
* @param ftor A functor, function or lambda expression that's executed
* @param f A functor, function or lambda expression that's executed
* if the actor finishes execution.
* @returns @c true if @p ftor was successfully attached to the actor;
* @returns @c true if @p f was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*/
template<typename F>
bool attach_functor(F&& ftor);
bool attach_functor(F&& f);
/**
* @brief Detaches the first attached object that matches @p what.
*/
virtual void detach(const attachable::token& what) = 0;
template<typename T>
bool attach(std::unique_ptr<T>&& ptr,
typename std::enable_if<
std::is_base_of<attachable,T>::value
>::type* = 0);
virtual void detach(const attachable::token& what);
/**
* @brief Links this actor to @p other.
* @param other Actor instance that whose execution is coupled to the
* execution of this Actor.
*/
virtual void link_to(const intrusive_ptr<actor>& other) = 0;
virtual void link_to(const actor_ptr& other);
/**
* @brief Unlinks this actor from @p other.
* @param other Linked Actor.
* @note Links are automatically removed if the Actor finishes execution.
* @note Links are automatically removed if the actor finishes execution.
*/
virtual void unlink_from(const intrusive_ptr<actor>& other) = 0;
virtual void unlink_from(const actor_ptr& other);
/**
* @brief Establishes a link relation between this actor and @p other.
......@@ -142,7 +151,7 @@ class actor : public channel {
* @returns @c true if this actor is running and added @p other to its
* list of linked actors; otherwise @c false.
*/
virtual bool establish_backlink(const intrusive_ptr<actor>& other) = 0;
virtual bool establish_backlink(const actor_ptr& other);
/**
* @brief Removes a link relation between this actor and @p other.
......@@ -150,7 +159,7 @@ class actor : public channel {
* @returns @c true if this actor is running and removed @p other
* from its list of linked actors; otherwise @c false.
*/
virtual bool remove_backlink(const intrusive_ptr<actor>& other) = 0;
virtual bool remove_backlink(const actor_ptr& other);
/**
* @brief Gets an integer value that uniquely identifies this Actor in
......@@ -173,23 +182,59 @@ class actor : public channel {
actor(actor_id aid);
/**
* @brief Should be overridden by subtypes and called upon termination.
* @note Default implementation sets 'exit_reason' accordingly.
* @note Overridden functions should always call super::cleanup().
*/
virtual void cleanup(std::uint32_t reason);
/**
* @brief The default implementation for {@link link_to()}.
*/
bool link_to_impl(const actor_ptr& other);
/**
* @brief The default implementation for {@link unlink_from()}.
*/
bool unlink_from_impl(const actor_ptr& other);
/**
* @brief Returns the actor's exit reason of
* <tt>exit_reason::not_exited</tt> if it's still alive.
*/
inline std::uint32_t exit_reason() const;
/**
* @brief Returns <tt>exit_reason() != exit_reason::not_exited</tt>.
*/
inline bool exited() const;
private:
actor_id m_id;
bool m_is_proxy;
// cannot be changed after construction
const actor_id m_id;
};
// you're either a proxy or you're not
const bool m_is_proxy;
/**
* @brief A smart pointer type that manages instances of {@link actor}.
* @relates actor
*/
typedef intrusive_ptr<actor> actor_ptr;
// initially exit_reason::not_exited
std::atomic<std::uint32_t> m_exit_reason;
class self_type;
// guards access to m_exit_reason, m_attachables, and m_links
std::mutex m_mtx;
// links to other actors
std::vector<actor_ptr> m_links;
// attached functors that are executed on cleanup
std::vector<attachable_ptr> m_attachables;
};
bool operator==(const actor_ptr& lhs, const self_type& rhs);
bool operator!=(const self_type& lhs, const actor_ptr& rhs);
// undocumented, because self_type is hidden in documentation
bool operator==(const actor_ptr&, const self_type&);
bool operator!=(const self_type&, const actor_ptr&);
/******************************************************************************
* inline and template member function implementations *
......@@ -203,39 +248,33 @@ inline bool actor::is_proxy() const {
return m_is_proxy;
}
template<typename T>
bool actor::attach(std::unique_ptr<T>&& ptr,
typename std::enable_if<
std::is_base_of<attachable,T>::value
>::type*) {
return attach(static_cast<attachable*>(ptr.release()));
inline std::uint32_t actor::exit_reason() const {
return m_exit_reason;
}
inline bool actor::exited() const {
return exit_reason() != exit_reason::not_exited;
}
template<class F>
class functor_attachable : public attachable {
struct functor_attachable : attachable {
F m_functor;
public:
template<class FArg>
functor_attachable(FArg&& arg) : m_functor(std::forward<FArg>(arg)) {
}
template<typename T>
inline functor_attachable(T&& arg) : m_functor(std::forward<T>(arg)) { }
void actor_exited(std::uint32_t reason) {
m_functor(reason);
}
void actor_exited(std::uint32_t reason) { m_functor(reason); }
bool matches(const attachable::token&) {
return false;
}
bool matches(const attachable::token&) { return false; }
};
template<typename F>
bool actor::attach_functor(F&& ftor) {
bool actor::attach_functor(F&& f) {
typedef typename util::rm_ref<F>::type f_type;
return attach(new functor_attachable<f_type>(std::forward<F>(ftor)));
typedef functor_attachable<f_type> impl;
return attach(attachable_ptr{new impl(std::forward<F>(f))});
}
} // namespace cppa
......
......@@ -36,14 +36,15 @@
#include "cppa/self.hpp"
#include "cppa/actor.hpp"
#include "cppa/extend.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/abstract_actor.hpp"
namespace cppa {
......@@ -90,7 +91,7 @@ class actor_companion_mixin : public Base {
*/
template<typename... Args>
void set_message_handler(Args&&... matchExpressions) {
m_message_handler = match_expr_concat(std::forward<Args>(matchExpressions)...);
m_message_handler = match_expr_convert(std::forward<Args>(matchExpressions)...);
}
/**
......@@ -108,7 +109,7 @@ class actor_companion_mixin : public Base {
private:
class companion : public detail::abstract_actor<local_actor> {
class companion : public local_actor {
friend class actor_companion_mixin;
typedef util::shared_spinlock lock_type;
......@@ -123,24 +124,26 @@ class actor_companion_mixin : public Base {
std::lock_guard<lock_type> guard(m_lock);
m_parent = nullptr;
}
cleanup(cppa::exit_reason::normal);
cleanup(exit_reason::normal);
}
void enqueue(cppa::actor* sender, cppa::any_tuple msg) {
cppa::util::shared_lock_guard<lock_type> guard(m_lock);
void enqueue(const actor_ptr& sender, any_tuple msg) {
util::shared_lock_guard<lock_type> guard(m_lock);
if (m_parent) {
m_parent->new_message(new_node_ptr(sender, std::move(msg)));
}
}
void sync_enqueue(cppa::actor* sender, cppa::message_id_t id, cppa::any_tuple msg) {
cppa::util::shared_lock_guard<lock_type> guard(m_lock);
void sync_enqueue(const actor_ptr& sender,
message_id_t id,
any_tuple msg) {
util::shared_lock_guard<lock_type> guard(m_lock);
if (m_parent) {
m_parent->new_message(new_node_ptr(sender, std::move(msg), id));
}
}
bool initialized() { return true; }
bool initialized() const { return true; }
void quit(std::uint32_t) {
throw std::runtime_error("ActorWidgetMixin::Gateway::exit "
......@@ -148,21 +151,21 @@ class actor_companion_mixin : public Base {
"management instead!");
}
void dequeue(cppa::behavior&) { throw_no_recv(); }
void dequeue(behavior&) { throw_no_recv(); }
void dequeue(cppa::partial_function&) { throw_no_recv(); }
void dequeue(partial_function&) { throw_no_recv(); }
void dequeue_response(cppa::behavior&, cppa::message_id_t) {
void dequeue_response(behavior&, message_id_t) {
throw_no_recv();
}
void become_waiting_for(cppa::behavior&&, cppa::message_id_t) {
void become_waiting_for(behavior&&, message_id_t) {
throw_no_become();
}
protected:
void do_become(cppa::behavior&&, bool) { throw_no_become(); }
void do_become(behavior&&, bool) { throw_no_become(); }
private:
......@@ -172,13 +175,13 @@ class actor_companion_mixin : public Base {
}
void throw_no_become() {
throw std::runtime_error("actor_companion_mixin::companion "
throw std::runtime_error("actor_companion_companion "
"does not support libcppa's "
"become() API");
}
void throw_no_recv() {
throw std::runtime_error("actor_companion_mixin::companion "
throw std::runtime_error("actor_companion_companion "
"does not support libcppa's "
"receive() API");
}
......
......@@ -32,8 +32,9 @@
#define CPPA_ACTOR_PROXY_HPP
#include "cppa/actor.hpp"
#include "cppa/extend.hpp"
#include "cppa/weak_intrusive_ptr.hpp"
#include "cppa/enable_weak_ptr_mixin.hpp"
#include "cppa/enable_weak_ptr.hpp"
#include "cppa/network/message_header.hpp"
......@@ -43,10 +44,11 @@ class actor_proxy_cache;
/**
* @brief Represents a remote actor.
* @extends actor
*/
class actor_proxy : public enable_weak_ptr_mixin<actor> {
class actor_proxy : public extend<actor,actor_proxy>::with<enable_weak_ptr> {
typedef enable_weak_ptr_mixin<actor> super;
typedef combined_type super;
public:
......
......@@ -28,21 +28,20 @@
\******************************************************************************/
#ifndef CPPA_ACTOR_COUNT_HPP
#define CPPA_ACTOR_COUNT_HPP
#ifndef CPPA_ACTOR_STATE_HPP
#define CPPA_ACTOR_STATE_HPP
#include <cstddef>
namespace cppa {
namespace cppa { namespace detail {
enum class actor_state : int {
ready,
done,
blocked,
pending,
about_to_block
};
void inc_actor_count();
void dec_actor_count();
} // namespace cppa
/*
* @pre @p expected <= 1
*/
void actor_count_wait_until(size_t expected);
#endif // CPPA_ACTOR_STATE_HPP
} } // namespace cppa::detail
#endif // CPPA_ACTOR_COUNT_HPP
......@@ -53,7 +53,7 @@ namespace cppa {
* foo(1,2)<br>
* foo_pair(3,4)
* </tt>
* @example announce_example_1.cpp
* @example announce_1.cpp
*/
/**
......@@ -62,7 +62,7 @@ namespace cppa {
* The output of this example program is:
*
* <tt>foo(1,2)</tt>
* @example announce_example_2.cpp
* @example announce_2.cpp
*/
/**
......@@ -72,7 +72,7 @@ namespace cppa {
* The output of this example program is:
*
* <tt>foo(1,2)</tt>
* @example announce_example_3.cpp
* @example announce_3.cpp
*/
/**
......@@ -81,13 +81,13 @@ namespace cppa {
* The output of this example program is:
*
* <tt>bar(foo(1,2),3)</tt>
* @example announce_example_4.cpp
* @example announce_4.cpp
*/
/**
* @brief An advanced example for @c announce implementing serialization
* for a user-defined tree data type.
* @example announce_example_5.cpp
* @example announce_5.cpp
*/
/**
......@@ -105,7 +105,7 @@ bool announce(const std::type_info& tinfo, uniform_type_info* utype);
* @brief Creates meta informations for a non-trivial member @p C.
* @param c_ptr Pointer to the non-trivial member.
* @param args "Sub-members" of @p c_ptr
* @see {@link announce_example_4.cpp announce example 4}
* @see {@link announce_4.cpp announce example 4}
* @returns A pair of @p c_ptr and the created meta informations.
*/
template<class C, class Parent, typename... Args>
......@@ -120,7 +120,7 @@ compound_member(C Parent::*c_ptr, const Args&... args) {
* via a getter returning a mutable reference.
* @param getter Member function pointer to the getter.
* @param args "Sub-members" of @p c_ptr
* @see {@link announce_example_4.cpp announce example 4}
* @see {@link announce_4.cpp announce example 4}
* @returns A pair of @p c_ptr and the created meta informations.
*/
template<class C, class Parent, typename... Args>
......@@ -136,7 +136,7 @@ compound_member(C& (Parent::*getter)(), const Args&... args) {
* @param gspair A pair of two member function pointers representing
* getter and setter.
* @param args "Sub-members" of @p c_ptr
* @see {@link announce_example_4.cpp announce example 4}
* @see {@link announce_4.cpp announce example 4}
* @returns A pair of @p c_ptr and the created meta informations.
*/
template<class Parent, typename GRes,
......
This diff is collapsed.
......@@ -31,6 +31,7 @@
#ifndef CPPA_ATTACHABLE_HPP
#define CPPA_ATTACHABLE_HPP
#include <memory>
#include <cstdint>
#include <typeinfo>
......@@ -54,17 +55,19 @@ class attachable {
* @brief Represents a pointer to a value with its RTTI.
*/
struct token {
/**
* @brief Denotes the type of @c ptr.
*/
const std::type_info& subtype;
/**
* @brief Any value, used to identify @c attachable instances.
*/
const void* ptr;
inline token(const std::type_info& msubtype, const void* mptr)
: subtype(msubtype), ptr(mptr) {
}
token(const std::type_info& subtype, const void* ptr);
};
virtual ~attachable();
......@@ -86,6 +89,12 @@ class attachable {
};
/**
* @brief A managed {@link attachable} pointer.
* @relates attachable
*/
typedef std::unique_ptr<attachable> attachable_ptr;
} // namespace cppa
#endif // CPPA_ATTACHABLE_HPP
......@@ -42,7 +42,6 @@
#include "cppa/util/if_else.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/util/disjunction.hpp"
namespace cppa {
......@@ -59,6 +58,8 @@ class behavior {
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
static constexpr bool may_have_timeout = true;
behavior() = default;
behavior(behavior&&) = default;
behavior(const behavior&) = default;
......@@ -85,7 +86,7 @@ class behavior {
template<typename... Cases, typename Arg1, typename... Args>
behavior(const match_expr<Cases...>& arg0, const Arg1& arg1, const Args&... args)
: m_impl(match_expr_concat(arg0, arg1, args...)) { }
: m_impl(detail::match_expr_concat(arg0, arg1, args...)) { }
inline void handle_timeout() {
m_impl->handle_timeout();
......@@ -109,6 +110,8 @@ class behavior {
return {new detail::continuation_decorator<F>(std::move(fun), m_impl)};
}
const impl_ptr& as_behavior_impl() const { return m_impl; }
private:
impl_ptr m_impl;
......@@ -121,7 +124,6 @@ inline behavior operator,(const match_expr<Lhs...>& lhs,
return match_expr_convert(lhs, rhs);
}
} // namespace cppa
#endif // CPPA_BEHAVIOR_HPP
......@@ -31,6 +31,8 @@
#ifndef CPPA_CHANNEL_HPP
#define CPPA_CHANNEL_HPP
#include <type_traits>
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
......@@ -41,6 +43,8 @@ class actor;
class group;
class any_tuple;
typedef intrusive_ptr<actor> actor_ptr;
/**
* @brief Interface for all message receivers.
*
......@@ -57,7 +61,7 @@ class channel : public ref_counted {
/**
* @brief Enqueues @p msg to the list of received messages.
*/
virtual void enqueue(actor* sender, any_tuple msg) = 0;
virtual void enqueue(const actor_ptr& sender, any_tuple msg) = 0;
protected:
......@@ -77,6 +81,12 @@ class channel : public ref_counted {
*/
typedef intrusive_ptr<channel> channel_ptr;
/**
* @brief Convenience alias.
*/
template<typename T>
using enable_if_channel = std::enable_if<std::is_base_of<channel,T>::value>;
} // namespace cppa
#endif // CPPA_CHANNEL_HPP
......@@ -35,49 +35,33 @@
#include "cppa/config.hpp"
#include "cppa/either.hpp"
#include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/stacked_actor_mixin.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @brief Context-switching actor implementation.
* @extends scheduled_actor
*/
class context_switching_actor : public scheduled_actor {
protected:
/**
* @brief Implements the actor's behavior.
* Reimplemented this function for a class-based actor.
* Returning from this member function will end the
* execution of the actor.
*/
virtual void run();
};
#else // CPPA_DOCUMENTATION
class context_switching_actor : public detail::stacked_actor_mixin<
context_switching_actor,
detail::abstract_scheduled_actor> {
class context_switching_actor : public extend<scheduled_actor,context_switching_actor>::with<stacked> {
friend class detail::behavior_stack;
friend class detail::receive_policy;
typedef detail::stacked_actor_mixin<
context_switching_actor,
detail::abstract_scheduled_actor> super;
typedef combined_type super;
public:
/**
* @brief Creates a context-switching actor running @p fun.
*/
context_switching_actor(std::function<void()> fun);
resume_result resume(util::fiber* from, actor_ptr& next_job); //override
......@@ -86,25 +70,15 @@ class context_switching_actor : public detail::stacked_actor_mixin<
protected:
context_switching_actor();
timeout_type init_timeout(const util::duration& rel_time);
detail::recursive_queue_node* await_message();
detail::recursive_queue_node* await_message(const timeout_type& abs_time);
private:
// required by detail::nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_nestable;
detail::recursive_queue_node* receive_node();
inline int init_timeout(const util::duration& timeout) {
// request timeout message
request_timeout(timeout);
return 0;
}
inline detail::recursive_queue_node* try_receive_node() {
return m_mailbox.try_pop();
}
inline detail::recursive_queue_node* try_receive_node(int) {
// timeout is triggered from an explicit timeout message
return receive_node();
}
// required by util::fiber
static void trampoline(void* _this);
......@@ -114,8 +88,6 @@ class context_switching_actor : public detail::stacked_actor_mixin<
};
#endif // CPPA_DOCUMENTATION
} // namespace cppa
#endif // CPPA_CONTEXT_SWITCHING_ACTOR_HPP
......@@ -170,8 +170,6 @@ struct cow_tuple_from_type_list< util::type_list<Ts...> > {
typedef cow_tuple<Ts...> type;
};
#ifdef CPPA_DOCUMENTATION
/**
* @ingroup CopyOnWrite
* @brief Gets a const-reference to the <tt>N</tt>th element of @p tup.
......@@ -180,8 +178,11 @@ struct cow_tuple_from_type_list< util::type_list<Ts...> > {
* <tt>N</tt>th element of @p tup.
* @relates cow_tuple
*/
template<size_t N, typename T>
const T& get(const cow_tuple<...>& tup);
template<size_t N, typename... Ts>
const typename util::at<N, Ts...>::type& get(const cow_tuple<Ts...>& tup) {
typedef typename util::at<N, Ts...>::type result_type;
return *reinterpret_cast<const result_type*>(tup.at(N));
}
/**
* @ingroup CopyOnWrite
......@@ -192,8 +193,11 @@ const T& get(const cow_tuple<...>& tup);
* @note Detaches @p tup if there are two or more references to the cow_tuple data.
* @relates cow_tuple
*/
template<size_t N, typename T>
T& get_ref(cow_tuple<...>& tup);
template<size_t N, typename... Ts>
typename util::at<N, Ts...>::type& get_ref(cow_tuple<Ts...>& tup) {
typedef typename util::at<N, Ts...>::type result_type;
return *reinterpret_cast<result_type*>(tup.mutable_at(N));
}
/**
* @ingroup ImplicitConversion
......@@ -202,31 +206,12 @@ T& get_ref(cow_tuple<...>& tup);
* @returns A cow_tuple object containing the values @p args.
* @relates cow_tuple
*/
template<typename... Args>
cow_tuple<Args...> make_cow_tuple(Args&&... args);
#else
template<size_t N, typename... Ts>
const typename util::at<N, Ts...>::type& get(const cow_tuple<Ts...>& tup) {
typedef typename util::at<N, Ts...>::type result_type;
return *reinterpret_cast<const result_type*>(tup.at(N));
}
template<size_t N, typename... Ts>
typename util::at<N, Ts...>::type& get_ref(cow_tuple<Ts...>& tup) {
typedef typename util::at<N, Ts...>::type result_type;
return *reinterpret_cast<result_type*>(tup.mutable_at(N));
}
template<typename... Args>
cow_tuple<typename detail::strip_and_convert<Args>::type...>
make_cow_tuple(Args&&... args) {
return {std::forward<Args>(args)...};
}
#endif
/**
* @brief Compares two cow_tuples.
* @param lhs First cow_tuple object.
......
......@@ -53,6 +53,7 @@
#include "cppa/any_tuple.hpp"
#include "cppa/cow_tuple.hpp"
#include "cppa/tuple_cast.hpp"
#include "cppa/singletons.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/spawn_options.hpp"
......@@ -65,8 +66,8 @@
#include "cppa/network/acceptor.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/get_behavior.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/receive_loop_helper.hpp"
/**
......@@ -119,11 +120,11 @@
*
* @section IntroHelloWorld Hello World Example
*
* @include hello_world_example.cpp
* @include hello_world.cpp
*
* @section IntroMoreExamples More Examples
*
* The {@link math_actor_example.cpp Math Actor Example} shows the usage
* The {@link math_actor.cpp Math Actor Example} shows the usage
* of {@link receive_loop} and {@link cppa::arg_match arg_match}.
* The {@link dining_philosophers.cpp Dining Philosophers Example}
* introduces event-based actors and includes a lot of <tt>libcppa</tt>
......@@ -389,13 +390,13 @@
/**
* @brief A trivial example program.
* @example hello_world_example.cpp
* @example hello_world.cpp
*/
/**
* @brief Shows the usage of {@link cppa::atom atoms}
* and {@link cppa::arg_match arg_match}.
* @example math_actor_example.cpp
* @example math_actor.cpp
*/
/**
......@@ -605,8 +606,9 @@ inline void reply_to(const response_handle& handle, Ts&&... what) {
}
/**
* @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple.
* @brief Replies with @p what to @p handle.
* @param handle Identifies a previously received request.
* @param what Response message.
*/
inline void reply_tuple_to(const response_handle& handle, any_tuple what) {
handle.apply(std::move(what));
......@@ -683,7 +685,6 @@ inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
/** @} */
#ifndef CPPA_DOCUMENTATION
// matches "send(this, ...)" and "send(self, ...)"
inline void send_tuple(local_actor* whom, any_tuple what) {
detail::send_impl(whom, std::move(what));
......@@ -701,7 +702,6 @@ inline actor_ptr eval_sopts(spawn_options opts, actor_ptr ptr) {
if (has_link_flag(opts)) self->link_to(ptr);
return std::move(ptr);
}
#endif // CPPA_DOCUMENTATION
/**
* @ingroup ActorCreation
......@@ -780,7 +780,8 @@ actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @warning Do not call this function in cooperatively scheduled actors.
*/
inline void await_all_others_done() {
detail::actor_count_wait_until((self.unchecked() == nullptr) ? 0 : 1);
auto value = (self.unchecked() == nullptr) ? 0 : 1;
get_actor_registry()->await_running_count_equal(value);
}
/**
......
This diff is collapsed.
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_SCHEDULED_ACTOR_HPP
#define CPPA_SCHEDULED_ACTOR_HPP
#include <iostream>
#include <atomic>
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
// A spawned, scheduled Actor.
class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
typedef abstract_actor<scheduled_actor> super;
protected:
std::atomic<int> m_state;
inline bool has_pending_timeout() const {
return m_has_pending_timeout_request;
}
void request_timeout(const util::duration& d);
inline void reset_timeout() {
if (m_has_pending_timeout_request) {
++m_active_timeout_id;
m_has_pending_timeout_request = false;
}
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void push_timeout() {
++m_active_timeout_id;
}
inline void pop_timeout() {
CPPA_REQUIRE(m_active_timeout_id > 0);
--m_active_timeout_id;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) {
return m_has_pending_timeout_request
&& m_active_timeout_id == timeout_id;
}
bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id;
public:
static constexpr int ready = 0x00;
static constexpr int done = 0x01;
static constexpr int blocked = 0x02;
static constexpr int pending = 0x03;
static constexpr int about_to_block = 0x04;
abstract_scheduled_actor(int state = done);
bool chained_enqueue(actor* sender, any_tuple msg);
bool chained_sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
void quit(std::uint32_t reason = exit_reason::normal);
void enqueue(actor* sender, any_tuple msg);
void sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
int compare_exchange_state(int expected, int new_value);
private:
bool enqueue_node(recursive_queue_node* node,
int next_state = ready,
bool* failed = nullptr);
};
} } // namespace cppa::detail
#endif // CPPA_SCHEDULED_ACTOR_HPP
......@@ -44,18 +44,11 @@
namespace cppa { namespace detail {
enum tuple_impl_info {
statically_typed,
dynamically_typed
};
class abstract_tuple : public ref_counted {
tuple_impl_info m_impl_type;
public:
inline abstract_tuple(tuple_impl_info tii) : m_impl_type(tii) { }
abstract_tuple(bool dynamically_typed);
abstract_tuple(const abstract_tuple& other);
// mutators
......@@ -76,7 +69,7 @@ class abstract_tuple : public ref_counted {
// A statically typed tuple implementation can use some optimizations,
// e.g., "impl_type() == statically_typed" implies that type_token()
// identifies all possible instances of a given tuple implementation
inline tuple_impl_info impl_type() const { return m_impl_type; }
inline bool dynamically_typed() const { return m_is_dynamic; }
// uniquely identifies this category (element types) of messages
// override this member function only if impl_type() == statically_typed
......@@ -93,6 +86,10 @@ class abstract_tuple : public ref_counted {
inline const_iterator end() const { return {this, size()}; }
inline const_iterator cend() const { return {this, size()}; }
private:
bool m_is_dynamic;
};
struct full_eq_type {
......
......@@ -195,6 +195,8 @@ class continuation_decorator : public behavior_impl {
};
typedef intrusive_ptr<behavior_impl> behavior_impl_ptr;
} } // namespace cppa::detail
#endif // BEHAVIOR_IMPL_HPP
......@@ -122,7 +122,7 @@ class behavior_stack
template<class Policy, class Client>
void exec(Policy& policy, Client* client) {
while (!empty()) {
invoke(policy, client, client->receive_node());
invoke(policy, client, client->await_message());
cleanup();
}
}
......@@ -140,7 +140,7 @@ class behavior_stack
}
}
inline void erase_at(std::vector<element_type>::reverse_iterator 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);
}
......@@ -152,7 +152,7 @@ class behavior_stack
template<typename UnaryPredicate>
inline void rerase_if(UnaryPredicate p) {
erase_at(std::find_if(m_elements.rbegin(), m_elements.rend(), p));
rerase_at(std::find_if(m_elements.rbegin(), m_elements.rend(), p));
}
};
......
......@@ -50,7 +50,7 @@ class container_tuple_view : public abstract_tuple {
typedef typename Container::value_type value_type;
container_tuple_view(Container* c, bool take_ownership = false)
: super(tuple_impl_info::dynamically_typed), m_ptr(c) {
: super(true), m_ptr(c) {
CPPA_REQUIRE(c != nullptr);
if (!take_ownership) m_ptr.get_deleter().disable();
}
......
......@@ -105,7 +105,7 @@ class decorated_tuple : public abstract_tuple {
vector_type m_mapping;
decorated_tuple(cow_pointer_type d, const vector_type& v)
: super(tuple_impl_info::statically_typed)
: super(false)
, m_decorated(std::move(d)), m_mapping(v) {
# ifdef CPPA_DEBUG
const cow_pointer_type& ptr = m_decorated; // prevent detaching
......@@ -116,7 +116,7 @@ class decorated_tuple : public abstract_tuple {
}
decorated_tuple(cow_pointer_type d, size_t offset)
: super(tuple_impl_info::statically_typed), m_decorated(std::move(d)) {
: super(false), m_decorated(std::move(d)) {
# ifdef CPPA_DEBUG
const cow_pointer_type& ptr = m_decorated; // prevent detaching
# endif
......
......@@ -46,7 +46,7 @@ struct matcher;
template<class Tuple, typename... T>
struct matcher<wildcard_position::nil, Tuple, T...> {
static inline bool tmatch(const Tuple& tup) {
if (tup.impl_type() == tuple_impl_info::statically_typed) {
if (not tup.dynamically_typed()) {
// statically typed tuples return &typeid(type_list<T...>)
// as type token
return typeid(util::type_list<T...>) == *(tup.type_token());
......
......@@ -108,7 +108,7 @@ class do_receive_helper {
template<typename... Args>
do_receive_helper(Args&&... args)
: m_bhvr(match_expr_concat(std::forward<Args>(args)...)) {
: m_bhvr(match_expr_convert(std::forward<Args>(args)...)) {
}
do_receive_helper(do_receive_helper&&) = default;
......@@ -122,21 +122,6 @@ class do_receive_helper {
sptr->dequeue(m_bhvr);
}
while (stmt() == false);
/*
if (m_bhvr.timeout().valid()) {
do {
sptr->dequeue(m_bhvr);
}
while (stmt() == false);
}
else {
partial_function& pfun = m_bhvr;
do {
sptr->dequeue(pfun);
}
while (stmt() == false);
}
*/
}
};
......
......@@ -36,6 +36,7 @@
#include <iostream>
#include <type_traits>
#include "cppa/logging.hpp"
#include "cppa/behavior.hpp"
#include "cppa/to_string.hpp"
#include "cppa/message_id.hpp"
......@@ -135,7 +136,7 @@ class receive_policy {
template<class Client, class FunOrBehavior>
inline void receive_wo_timeout(Client *client, FunOrBehavior& fun) {
if (!invoke_from_cache(client, fun)) {
while (!invoke(client, client->receive_node(), fun)) { }
while (!invoke(client, client->await_message(), fun)) { }
}
}
......@@ -152,7 +153,7 @@ class receive_policy {
else if (!invoke_from_cache(client, bhvr)) {
if (bhvr.timeout().is_zero()) {
pointer e = nullptr;
while ((e = client->try_receive_node()) != nullptr) {
while ((e = client->m_mailbox.try_pop()) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) {
return; // done
......@@ -163,7 +164,7 @@ class receive_policy {
else {
auto timeout = client->init_timeout(bhvr.timeout());
pointer e = nullptr;
while ((e = client->try_receive_node(timeout)) != nullptr) {
while ((e = client->await_message(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) {
return; // done
......@@ -182,7 +183,7 @@ class receive_policy {
CPPA_REQUIRE(bhvr.timeout().is_zero() == false);
auto timeout = client->init_timeout(bhvr.timeout());
pointer e = nullptr;
while ((e = client->try_receive_node(timeout)) != nullptr) {
while ((e = client->await_message(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr, mid)) {
return; // done
......@@ -190,9 +191,7 @@ class receive_policy {
}
handle_timeout(client, bhvr);
}
else {
while (!invoke(client, client->receive_node(), bhvr, mid)) { }
}
else while (!invoke(client, client->await_message(), bhvr, mid)) { }
}
}
......@@ -243,9 +242,8 @@ class receive_policy {
CPPA_REQUIRE(!message_id.valid());
if (client->m_trap_exit == false) {
if (v1 != exit_reason::normal) {
// TODO: check for possible memory leak here
// ('node' might not get destroyed)
client->quit(v1);
return non_normal_exit_signal;
}
return normal_exit_signal;
}
......@@ -327,7 +325,7 @@ class receive_policy {
if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout());
}
else client->m_has_pending_timeout_request = false;
else client->reset_timeout();
}
template<class Client>
......@@ -350,12 +348,25 @@ class receive_policy {
}
switch (this->filter_msg(client, node)) {
default: {
// one of:
// - normal_exit_signal
// - expired_sync_response
// - expired_timeout_message
CPPA_CRITICAL("illegal filter result");
}
case normal_exit_signal: {
CPPA_LOG_DEBUG("dropped message: normal exit signal");
return hm_drop_msg;
}
case expired_sync_response: {
CPPA_LOG_DEBUG("dropped message: expired sync response");
return hm_drop_msg;
}
case expired_timeout_message: {
CPPA_LOG_DEBUG("dropped message: expired timeout message");
return hm_drop_msg;
}
case non_normal_exit_signal: {
// this message was handled
// by calling client->quit(...)
return hm_msg_handled;
}
case timeout_message: {
handle_timeout(client, fun);
if (awaited_response.valid()) {
......@@ -372,6 +383,7 @@ class receive_policy {
if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = hm_begin(client, node, policy);
if (!fun(node->msg) && handle_sync_failure_on_mismatch) {
CPPA_LOG_WARNING("sync failure occured");
client->handle_sync_failure();
}
client->mark_arrived(awaited_response);
......@@ -390,6 +402,8 @@ class receive_policy {
auto id = node->mid;
auto sender = node->sender;
if (id.valid() && !id.is_answered() && sender) {
CPPA_LOG_WARNING("actor did not reply to a "
"synchronous request message");
sender->sync_enqueue(client,
id.response_id(),
make_any_tuple(atom("VOID")));
......
......@@ -37,14 +37,14 @@
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/memory_cached_mixin.hpp"
#include "cppa/memory_cached.hpp"
// needs access to constructor + destructor to initialize m_dummy_node
namespace cppa { class local_actor; }
namespace cppa { namespace detail {
class recursive_queue_node : public memory_cached_mixin<memory_managed> {
class recursive_queue_node : public memory_cached<memory_managed,recursive_queue_node> {
friend class memory;
friend class ::cppa::local_actor;
......@@ -64,11 +64,18 @@ class recursive_queue_node : public memory_cached_mixin<memory_managed> {
recursive_queue_node& operator=(recursive_queue_node&&) = delete;
recursive_queue_node& operator=(const recursive_queue_node&) = delete;
template<typename... Ts>
inline static recursive_queue_node* create(Ts&&... args) {
return memory::create<recursive_queue_node>(std::forward<Ts>(args)...);
}
private:
recursive_queue_node() = default;
recursive_queue_node(actor_ptr sptr, any_tuple data, message_id_t id = message_id_t());
recursive_queue_node(const actor_ptr& sptr,
any_tuple data,
message_id_t id = message_id_t{});
};
......
......@@ -31,22 +31,19 @@
#ifndef CPPA_SCHEDULED_ACTOR_DUMMY_HPP
#define CPPA_SCHEDULED_ACTOR_DUMMY_HPP
#include "cppa/detail/abstract_scheduled_actor.hpp"
#include "cppa/scheduled_actor.hpp"
namespace cppa { namespace detail {
struct scheduled_actor_dummy : abstract_scheduled_actor {
struct scheduled_actor_dummy : scheduled_actor {
scheduled_actor_dummy();
void enqueue(const actor_ptr&, any_tuple);
void sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
resume_result resume(util::fiber*, actor_ptr&);
void quit(std::uint32_t);
void dequeue(behavior&);
void dequeue(partial_function&);
void dequeue_response(behavior&, message_id_t);
void link_to(const intrusive_ptr<actor>&);
void unlink_from(const intrusive_ptr<actor>&);
bool establish_backlink(const intrusive_ptr<actor>&);
bool remove_backlink(const intrusive_ptr<actor>&);
void detach(const attachable::token&);
bool attach(attachable*);
void do_become(behavior&&, bool);
void become_waiting_for(behavior&&, message_id_t);
bool has_behavior();
......
......@@ -37,7 +37,6 @@ namespace cppa {
class logging;
class scheduler;
class msg_content;
} // namespace cppa
......@@ -82,7 +81,7 @@ class singleton_manager {
private:
/**
/*
* @brief Type @p T has to provide: <tt>static T* create_singleton()</tt>,
* <tt>void initialize()</tt>, <tt>void destroy()</tt>,
* and <tt>dispose()</tt>.
......
......@@ -28,22 +28,37 @@
\******************************************************************************/
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#ifndef CPPA_SYNC_REQUEST_BOUNCER_HPP
#define CPPA_SYNC_REQUEST_BOUNCER_HPP
namespace cppa { namespace detail {
#include <cstdint>
void inc_actor_count() {
singleton_manager::get_actor_registry()->inc_running();
}
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
void dec_actor_count() {
singleton_manager::get_actor_registry()->dec_running();
}
namespace cppa { namespace detail {
void actor_count_wait_until(size_t value) {
singleton_manager::get_actor_registry()->await_running_count_equal(value);
}
struct sync_request_bouncer {
std::uint32_t rsn;
constexpr sync_request_bouncer(std::uint32_t r)
: rsn(r == exit_reason::not_exited ? exit_reason::normal : r) { }
inline void operator()(const actor_ptr& sender, const message_id_t& mid) const {
CPPA_REQUIRE(rsn != exit_reason::not_exited);
if (mid.is_request() && sender != nullptr) {
actor_ptr nobody;
sender->sync_enqueue(nobody,
mid.response_id(),
make_any_tuple(atom("EXITED"), rsn));
}
}
inline void operator()(const recursive_queue_node& e) const {
(*this)(e.sender.get(), e.mid);
}
};
} } // namespace cppa::detail
#endif // CPPA_SYNC_REQUEST_BOUNCER_HPP
......@@ -176,8 +176,8 @@ struct tdata<> {
inline bool operator==(const tdata&) const { return true; }
inline tuple_impl_info impl_type() const {
return statically_typed;
inline bool dynamically_typed() const {
return false;
}
};
......
......@@ -37,7 +37,6 @@
#include "cppa/context_switching_actor.hpp"
#include "cppa/util/producer_consumer_list.hpp"
#include "cppa/detail/scheduled_actor_dummy.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace detail {
......
......@@ -59,9 +59,8 @@ class tuple_vals : public abstract_tuple {
tuple_vals(const tuple_vals&) = default;
template<typename... Args>
tuple_vals(Args&&... args)
: super(tuple_impl_info::statically_typed)
, m_data(std::forward<Args>(args)...) { }
tuple_vals(Args&&... args) : super(false)
, m_data(std::forward<Args>(args)...) { }
const void* native_data() const {
return &m_data;
......
......@@ -68,9 +68,7 @@ class tuple_view : public abstract_tuple {
/**
* @warning @p tuple_view does @b NOT takes ownership for given pointers
*/
tuple_view(Ts*... args)
: super(tuple_impl_info::statically_typed), m_data(args...) {
}
tuple_view(Ts*... args) : super(false), m_data(args...) { }
inline data_type& data() {
return m_data;
......
......@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef CPPA_ENABLE_WEAK_PTR_MIXIN_HPP
#define CPPA_ENABLE_WEAK_PTR_MIXIN_HPP
#ifndef CPPA_ENABLE_WEAK_PTR_HPP
#define CPPA_ENABLE_WEAK_PTR_HPP
#include <mutex>
#include <utility>
......@@ -47,10 +47,8 @@ namespace cppa {
/**
* @brief Enables derived classes to be used in {@link weak_intrusive_ptr}.
*/
template<class Base>
class enable_weak_ptr_mixin : public Base {
typedef Base super;
template<class Base, class Subtype>
class enable_weak_ptr : public Base {
template<typename T>
friend class weak_intrusive_ptr;
......@@ -60,9 +58,11 @@ class enable_weak_ptr_mixin : public Base {
protected:
typedef enable_weak_ptr combined_type;
template<typename... Args>
enable_weak_ptr_mixin(Args&&... args)
: super(std::forward<Args>(args)...)
enable_weak_ptr(Args&&... args)
: Base(std::forward<Args>(args)...)
, m_anchor(new weak_ptr_anchor(this)) { }
void request_deletion() {
......@@ -81,4 +81,4 @@ class enable_weak_ptr_mixin : public Base {
} // namespace cppa
#endif // CPPA_ENABLE_WEAK_PTR_MIXIN_HPP
#endif // CPPA_ENABLE_WEAK_PTR_HPP
......@@ -38,23 +38,21 @@
#include "cppa/config.hpp"
#include "cppa/either.hpp"
#include "cppa/extend.hpp"
#include "cppa/behavior.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @brief Base class for all event-based actor implementations.
*/
class event_based_actor : public scheduled_actor {
#else // CPPA_DOCUMENTATION
class event_based_actor : public detail::abstract_scheduled_actor {
#endif // CPPA_DOCUMENTATION
friend class detail::receive_policy;
typedef detail::abstract_scheduled_actor super;
typedef scheduled_actor super;
public:
......@@ -81,7 +79,7 @@ class event_based_actor : public detail::abstract_scheduled_actor {
*/
virtual void init() = 0;
void quit(std::uint32_t reason = exit_reason::normal);
virtual void quit(std::uint32_t reason = exit_reason::normal);
bool has_behavior();
......@@ -149,7 +147,7 @@ class event_based_actor : public detail::abstract_scheduled_actor {
static const detail::receive_policy_flag receive_flag = detail::rp_sequential;
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
m_has_pending_timeout_request = false;
reset_timeout();
bhvr.handle_timeout();
if (m_bhvr_stack.empty() == false) {
request_timeout(get_behavior().timeout());
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_MIXED_HPP
#define CPPA_MIXED_HPP
// saves some typing
#define CPPA_MIXIN template<class,class> class
namespace cppa {
namespace detail {
template<class B, class D, CPPA_MIXIN... Ms>
struct extend_helper;
template<class B, class D>
struct extend_helper<B,D> { typedef D type; };
template<class B, class D, CPPA_MIXIN M, CPPA_MIXIN... Ms>
struct extend_helper<B,D,M,Ms...> : extend_helper<B,M<D,B>,Ms...> { };
} // namespace detail
/**
* @brief Allows convenient definition of types using mixins.
* For example, @p extend<ar,T>::with<ob,fo> is an alias for
* @p fo<ob<ar,T>,T>.
*
* Mixins in libcppa always have two template parameters: base type and
* derived type. This allows mixins to make use of the curiously recurring
* template pattern (CRTP).
*/
template<class Base, class Derived>
struct extend {
/**
* @brief An alias for the result type.
*/
template<template<class,class> class... Mixins>
using with = typename detail::extend_helper<Derived,Base,Mixins...>::type;
};
} // namespace cppa
#endif // CPPA_MIXED_HPP
......@@ -35,7 +35,7 @@
namespace cppa { namespace factory {
#ifdef CPPA_DOCUMENTATION
void default_cleanup();
/**
* @brief Returns a factory for event-based actors using @p fun as
......@@ -57,7 +57,10 @@ namespace cppa { namespace factory {
* unless an initial value is passed to @p spawn.
*/
template<typename InitFun>
auto event_based(InitFun fun);
inline typename detail::ebaf_from_functor<InitFun, void (*)()>::type
event_based(InitFun init) {
return {std::move(init), default_cleanup};
}
/**
* @brief Returns a factory for event-based actors using @p fun0 as
......@@ -65,27 +68,12 @@ auto event_based(InitFun fun);
* and @p fun1 as implementation for
* {@link cppa::event_based_actor::on_exit() on_exit()}.
*/
template<typename InitFun, OnExitFun>
auto event_based(InitFun fun0, OnExitFun fun1);
#else // CPPA_DOCUMENTATION
void default_cleanup();
template<typename InitFun>
inline typename detail::ebaf_from_functor<InitFun, void (*)()>::type
event_based(InitFun init) {
return {std::move(init), default_cleanup};
}
template<typename InitFun, typename OnExitFun>
inline typename detail::ebaf_from_functor<InitFun, OnExitFun>::type
event_based(InitFun init, OnExitFun on_exit) {
return {std::move(init), on_exit};
}
#endif // CPPA_DOCUMENTATION
} } // namespace cppa::factory
#endif // CPPA_FACTORY_HPP
......@@ -31,9 +31,6 @@
#ifndef CPPA_GET_HPP
#define CPPA_GET_HPP
// functions are documented in the implementation headers
#ifndef CPPA_DOCUMENTATION
#include <tuple>
#include <cstddef>
......@@ -108,5 +105,4 @@ inline auto get_cv_aware(const Tuple& tup) -> decltype(get<Pos>(tup)) {
} // namespace cppa
#endif // CPPA_DOCUMENTATION
#endif // CPPA_GET_HPP
......@@ -728,11 +728,6 @@ struct mutable_gref_wrapped<T&> {
namespace placeholders {
// doxygen cannot handle anonymous namespaces
#ifndef CPPA_DOCUMENTATION
namespace {
#endif // CPPA_DOCUMENTATION
constexpr guard_placeholder<0> _x1;
constexpr guard_placeholder<1> _x2;
constexpr guard_placeholder<2> _x3;
......@@ -743,11 +738,6 @@ constexpr guard_placeholder<6> _x7;
constexpr guard_placeholder<7> _x8;
constexpr guard_placeholder<8> _x9;
// doxygen cannot handle anonymous namespaces
#ifndef CPPA_DOCUMENTATION
} // namespace <anonymous>
#endif // CPPA_DOCUMENTATION
} // namespace placeholders
} // namespace cppa
......
This diff is collapsed.
This diff is collapsed.
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_MAILBOX_BASED_HPP
#define CPPA_MAILBOX_BASED_HPP
#include <type_traits>
#include "cppa/detail/sync_request_bouncer.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/intrusive/blocking_single_reader_queue.hpp"
namespace cppa {
template<typename T>
struct has_blocking_receive;
template<class Base, class Subtype>
class mailbox_based : public Base {
typedef detail::disposer del;
public:
~mailbox_based() {
if (!m_mailbox.closed()) {
detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
}
}
protected:
typedef mailbox_based combined_type;
typedef detail::recursive_queue_node mailbox_element;
typedef typename std::conditional<
has_blocking_receive<Subtype>::value,
intrusive::blocking_single_reader_queue<mailbox_element,del>,
intrusive::single_reader_queue<mailbox_element,del>
>::type
mailbox_type;
template<typename... Ts>
mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
void cleanup(std::uint32_t reason) {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
}
template<typename... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
mailbox_type m_mailbox;
private:
inline Subtype* dthis() { return static_cast<Subtype*>(this); }
};
} // namespace cppa
#endif //CPPA_MAILBOX_BASED_HPP
This diff is collapsed.
......@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef MEMORY_CACHED_MIXIN_HPP
#define MEMORY_CACHED_MIXIN_HPP
#ifndef CPPA_MEMORY_CACHED_HPP
#define CPPA_MEMORY_CACHED_HPP
#include "cppa/detail/memory.hpp"
......@@ -39,19 +39,21 @@ namespace cppa {
* @brief This mixin adds all member functions and member variables needed
* by the memory management subsystem.
*/
template<typename Base>
class memory_cached_mixin : public Base {
template<class Base, class Subtype>
class memory_cached : public Base {
friend class detail::memory;
template<typename T>
template<typename>
friend class detail::basic_memory_cache;
protected:
template<typename... Args>
memory_cached_mixin(Args&&... args)
: Base(std::forward<Args>(args)...), outer_memory(nullptr) { }
typedef memory_cached combined_type;
template<typename... Ts>
memory_cached(Ts&&... args) : Base(std::forward<Ts>(args)...)
, outer_memory(nullptr) { }
virtual void request_deletion() {
auto mc = detail::memory::get_cache_map_entry(&typeid(*this));
......@@ -74,4 +76,4 @@ class memory_cached_mixin : public Base {
} // namespace cppa
#endif // MEMORY_CACHED_MIXIN_HPP
#endif // CPPA_MEMORY_CACHED_HPP
......@@ -31,9 +31,11 @@
#ifndef MESSAGE_FUTURE_HPP
#define MESSAGE_FUTURE_HPP
#include <cstdint>
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/atom.hpp"
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
......@@ -130,12 +132,8 @@ class message_future {
message_future(const message_future&) = default;
message_future& operator=(const message_future&) = default;
# ifndef CPPA_DOCUMENTATION
inline message_future(const message_id_t& from) : m_mid(from) { }
# endif
private:
message_id_t m_mid;
......@@ -147,7 +145,7 @@ class message_future {
return false;
};
return {
on<atom("EXITED"),std::uint32_t>() >> skip_message,
on(atom("EXITED"), val<std::uint32_t>) >> skip_message,
on(atom("VOID")) >> skip_message,
on(atom("TIMEOUT")) >> handle_sync_timeout,
(on(any_vals, arg_match) >> fs)...
......
......@@ -31,12 +31,11 @@
#ifndef DEFAULT_ACTOR_PROXY_HPP
#define DEFAULT_ACTOR_PROXY_HPP
#include "cppa/extend.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/memory_cached_mixin.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
......@@ -49,7 +48,8 @@ class basic_memory_cache;
namespace cppa { namespace network {
class sync_request_info : public memory_cached_mixin<memory_managed> {
class sync_request_info : public extend<memory_managed,sync_request_info>::
with<memory_cached> {
friend class detail::memory;
......@@ -67,9 +67,9 @@ class sync_request_info : public memory_cached_mixin<memory_managed> {
};
class default_actor_proxy : public detail::abstract_actor<actor_proxy> {
class default_actor_proxy : public actor_proxy {
typedef detail::abstract_actor<actor_proxy> super;
typedef actor_proxy super;
public:
......@@ -77,19 +77,19 @@ class default_actor_proxy : public detail::abstract_actor<actor_proxy> {
const process_information_ptr& pinfo,
const default_protocol_ptr& parent);
void enqueue(actor* sender, any_tuple msg);
void enqueue(const actor_ptr& sender, any_tuple msg);
void sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
void sync_enqueue(const actor_ptr& sender, message_id_t id, any_tuple msg);
void link_to(const intrusive_ptr<actor>& other);
void link_to(const actor_ptr& other);
void unlink_from(const intrusive_ptr<actor>& other);
void unlink_from(const actor_ptr& other);
bool remove_backlink(const intrusive_ptr<actor>& to);
bool remove_backlink(const actor_ptr& to);
bool establish_backlink(const intrusive_ptr<actor>& to);
bool establish_backlink(const actor_ptr& to);
void local_link_to(const intrusive_ptr<actor>& other);
void local_link_to(const actor_ptr& other);
void local_unlink_from(const actor_ptr& other);
......
......@@ -223,6 +223,11 @@ struct pattern_type : pattern_type_<util::is_callable<T>::value && !detail::is_b
namespace cppa {
/**
* @brief A wildcard that matches any number of any values.
*/
constexpr anything any_vals = anything{};
#ifdef CPPA_DOCUMENTATION
/**
......@@ -251,12 +256,6 @@ constexpr __unspecified__ skip_message;
template<typename T>
__unspecified__ val();
/**
* @brief A wildcard that matches any number of any values.
*/
constexpr anything any_vals;
/**
* @brief Left-hand side of a partial function expression that matches values.
*
......@@ -294,8 +293,6 @@ constexpr typename detail::boxed<T>::type val() {
return typename detail::boxed<T>::type();
}
constexpr anything any_vals = anything();
typedef typename detail::boxed<util::arg_match_t>::type boxed_arg_match_t;
constexpr boxed_arg_match_t arg_match = boxed_arg_match_t();
......@@ -414,7 +411,7 @@ class on_the_fly_rvalue_builder {
constexpr detail::on_the_fly_rvalue_builder on_arg_match;
#endif
#endif // CPPA_DOCUMENTATION
} // namespace cppa
......
......@@ -35,6 +35,7 @@
#include <vector>
#include <memory>
#include <utility>
#include <type_traits>
#include "cppa/behavior.hpp"
#include "cppa/any_tuple.hpp"
......@@ -62,6 +63,8 @@ class partial_function {
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
static constexpr bool may_have_timeout = false;
partial_function() = default;
partial_function(partial_function&&) = default;
partial_function(const partial_function&) = default;
......@@ -95,28 +98,23 @@ class partial_function {
return behavior{m_impl->or_else(other.m_impl)};
}
template<typename F>
inline behavior or_else(const timeout_definition<F>& tdef) const {
generic_timeout_definition gtd{tdef.timeout, tdef.handler};
return behavior{m_impl->copy(gtd)};
}
template<typename... Cases>
inline partial_function or_else(const match_expr<Cases...>& mexpr) const {
return m_impl->or_else(mexpr.as_behavior_impl());
}
template<typename Arg0, typename Arg1, typename... Args>
typename util::if_else<
util::disjunction<
is_timeout_definition<Arg0>,
is_timeout_definition<Arg1>,
is_timeout_definition<Args>...>,
behavior,
util::wrapped<partial_function> >::type
or_else(const Arg0& arg0, const Arg1& arg1, const Args&... args) const {
return m_impl->or_else(match_expr_concat(arg0, arg1, args...));
}
template<typename T>
typename std::conditional<T::may_have_timeout,behavior,partial_function>::type
or_else(const T& arg) const;
template<typename T0, typename T1, typename... Ts>
typename std::conditional<
util::disjunction<
T0::may_have_timeout,
T1::may_have_timeout,
Ts::may_have_timeout...
>::value,
behavior,
partial_function
>::type
or_else(const T0& arg0, const T1& arg1, const Ts&... args) const;
inline const impl_ptr& as_behavior_impl() const;
private:
......@@ -124,15 +122,45 @@ class partial_function {
};
template<typename Arg0, typename... Args>
typename util::if_else<
util::disjunction<
is_timeout_definition<Arg0>,
is_timeout_definition<Args>...>,
behavior,
util::wrapped<partial_function> >::type
match_expr_convert(const Arg0& arg0, const Args&... args) {
return {match_expr_concat(arg0, args...)};
template<typename T, typename... Ts>
typename std::conditional<
util::disjunction<
T::may_have_timeout,
Ts::may_have_timeout...
>::value,
behavior,
partial_function
>::type
match_expr_convert(const T& arg0, const Ts&... args) {
return detail::match_expr_concat(arg0, args...);
}
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline const partial_function::impl_ptr& partial_function::as_behavior_impl() const {
return m_impl;
}
template<typename T>
typename std::conditional<T::may_have_timeout,behavior,partial_function>::type
partial_function::or_else(const T& arg) const {
return m_impl->or_else(arg.as_behavior_impl());
}
template<typename T0, typename T1, typename... Ts>
typename std::conditional<
util::disjunction<
T0::may_have_timeout,
T1::may_have_timeout,
Ts::may_have_timeout...
>::value,
behavior,
partial_function
>::type
partial_function::or_else(const T0& arg0, const T1& arg1, const Ts&... args) const {
return m_impl->or_else(match_expr_convert(arg0, arg1, args...).as_behavior_impl());
}
} // namespace cppa
......
......@@ -293,8 +293,6 @@ T& get_ref(primitive_variant& pv) {
return pv.get_as<ptype>();
}
#ifdef CPPA_DOCUMENTATION
/**
* @ingroup TypeSystem
* @brief Casts a primitive variant to its C++ type.
......@@ -304,7 +302,11 @@ T& get_ref(primitive_variant& pv) {
* @returns A const reference to the value of @p pv of type @p T.
*/
template<primitive_type PT>
const T& get_ref(const primitive_variant& pv);
inline const typename detail::ptype_to_type<PT>::type&
get(const primitive_variant& pv) {
static_assert(PT != pt_null, "PT == pt_null");
return get<typename detail::ptype_to_type<PT>::type>(pv);
}
/**
* @ingroup TypeSystem
......@@ -314,18 +316,6 @@ const T& get_ref(const primitive_variant& pv);
* @param pv A primitive variant of type @p T.
* @returns A reference to the value of @p pv of type @p T.
*/
template<primitive_type PT>
T& get_ref(primitive_variant& pv);
#else
template<primitive_type PT>
inline const typename detail::ptype_to_type<PT>::type&
get(const primitive_variant& pv) {
static_assert(PT != pt_null, "PT == pt_null");
return get<typename detail::ptype_to_type<PT>::type>(pv);
}
template<primitive_type PT>
inline typename detail::ptype_to_type<PT>::type&
get_ref(primitive_variant& pv) {
......@@ -333,8 +323,6 @@ get_ref(primitive_variant& pv) {
return get_ref<typename detail::ptype_to_type<PT>::type>(pv);
}
#endif
/**
* @relates primitive_variant
*/
......
......@@ -50,6 +50,10 @@ class sb_actor : public Base {
static_assert(std::is_base_of<event_based_actor,Base>::value,
"Base must be either event_based_actor or a derived type");
protected:
typedef sb_actor combined_type;
public:
/**
......
......@@ -34,7 +34,14 @@
#include <functional>
#include "cppa/config.hpp"
#include "cppa/extend.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa {
......@@ -47,48 +54,167 @@ enum class resume_result {
};
enum scheduled_actor_type {
context_switching_impl,
event_based_impl,
context_switching_impl, // enqueued to the job queue on startup
event_based_impl, // not enqueued to the job queue on startup
default_event_based_impl // scheduler enqueues a 'RUN' message on startup
};
class scheduled_actor;
template<>
struct has_blocking_receive<scheduled_actor> : std::true_type { };
/**
* @brief A base class for cooperatively scheduled actors.
* @extends local_actor
*/
class scheduled_actor : public local_actor {
class scheduled_actor : public extend<local_actor,scheduled_actor>::with<mailbox_based>{
typedef combined_type super;
public:
scheduled_actor(bool enable_chained_send = false);
static constexpr bool has_blocking_receive = false;
~scheduled_actor();
/**
* @brief Intrusive next pointer needed by the scheduler's job queue.
*/
scheduled_actor* next;
// called from worker thread,
// actors sets next_job to its chained actor
/**
* @brief Continues execution of this actor.
* @param next_job Denotes the next actor to be resumed, should be
* set in case of chaining.
* @note This member function is called from the scheduler's worker threads.
*/
virtual resume_result resume(util::fiber* from, actor_ptr& next_job) = 0;
/**
* @brief Called once by the scheduler after actor is initialized,
* but before @p resume is called.
*/
void attach_to_scheduler(scheduler* sched, bool hidden);
/**
* @brief Returns @p false if this actor can be disposed immediately
* after construction, @p true otherwise.
*/
virtual bool has_behavior() = 0;
/**
* @brief Denotes wheter this actor is enqueued to the scheduler's
* job queue on startup.
*/
virtual scheduled_actor_type impl_type() = 0;
inline bool is_hidden() const { return m_hidden; }
/**
* @brief Returns @p true if this actor is ignored by
* {@link await_all_others_done()}, false otherwise.
*/
inline bool is_hidden() const;
void enqueue(const actor_ptr&, any_tuple);
bool chained_enqueue(const actor_ptr&, any_tuple);
void sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
bool chained_sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
void request_timeout(const util::duration& d);
inline bool has_pending_timeout() const {
return m_has_pending_tout;
}
inline void reset_timeout() {
if (m_has_pending_tout) {
++m_pending_tout;
m_has_pending_tout = false;
}
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void push_timeout() {
++m_pending_tout;
}
inline void pop_timeout() {
CPPA_REQUIRE(m_pending_tout > 0);
--m_pending_tout;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) {
return m_has_pending_tout && m_pending_tout == timeout_id;
}
protected:
scheduled_actor(actor_state init_state, bool enable_chained_send);
void cleanup(std::uint32_t reason);
typedef detail::recursive_queue_node mailbox_element;
typedef intrusive::single_reader_queue<mailbox_element,detail::disposer>
mailbox_type;
actor_state compare_exchange_state(actor_state expected, actor_state desired);
template<typename... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
inline void set_state(actor_state new_value) {
m_state.store(new_value);
}
inline actor_state state() const {
return m_state;
}
bool initialized() const;
private:
bool enqueue(actor_state next_state, bool* failed, mailbox_element* e);
template<typename... Ts>
inline bool enqueue_impl(actor_state next, Ts&&... args) {
auto ptr = this->new_mailbox_element(std::forward<Ts>(args)...);
return enqueue(next, nullptr, ptr);
}
bool sync_enqueue_impl(actor_state next,
const actor_ptr& sender,
any_tuple& msg,
message_id_t id);
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
std::atomic<actor_state> m_state;
protected:
scheduler* m_scheduler;
bool m_hidden;
bool initialized();
mailbox_type m_mailbox;
};
typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr;
inline bool scheduled_actor::is_hidden() const {
return m_hidden;
}
} // namespace cppa
#endif // CPPA_ACTOR_BEHAVIOR_HPP
......@@ -79,7 +79,7 @@ class scheduler {
scheduler_helper* m_helper;
channel* delayed_send_helper();
const actor_ptr& delayed_send_helper();
friend class detail::singleton_manager;
......@@ -173,18 +173,13 @@ class scheduler {
init_callback init_cb,
void_function actor_behavior) = 0;
// hide implementation details for documentation
# ifndef CPPA_DOCUMENTATION
template<typename F, typename T0, typename... Ts>
template<typename F, typename T, typename... Ts>
actor_ptr exec(spawn_options opts, init_callback cb,
F f, T0&& a0, Ts&&... as) {
return this->exec(opts, cb, std::bind(f, detail::fwd<T0>(a0),
F f, T&& a0, Ts&&... as) {
return this->exec(opts, cb, std::bind(f, detail::fwd<T>(a0),
detail::fwd<Ts>(as)...));
}
# endif // CPPA_DOCUMENTATION
};
/**
......@@ -201,11 +196,6 @@ void set_scheduler(scheduler* sched);
*/
void set_default_scheduler(size_t num_threads);
/**
* @brief Returns the currently running scheduler.
*/
scheduler* get_scheduler();
} // namespace cppa
#endif // CPPA_SCHEDULER_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_SINGLETONS_HPP
#define CPPA_SINGLETONS_HPP
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
inline logging* get_logger() {
return detail::singleton_manager::get_logger();
}
inline scheduler* get_scheduler() {
return detail::singleton_manager::get_scheduler();
}
inline detail::group_manager* get_group_manager() {
return detail::singleton_manager::get_group_manager();
}
inline detail::actor_registry* get_actor_registry() {
return detail::singleton_manager::get_actor_registry();
}
inline network::middleman* get_middleman() {
return detail::singleton_manager::get_middleman();
}
inline detail::uniform_type_info_map* get_uniform_type_info_map() {
return detail::singleton_manager::get_uniform_type_info_map();
}
inline detail::abstract_tuple* get_tuple_dummy() {
return detail::singleton_manager::get_tuple_dummy();
}
inline detail::empty_tuple* get_empty_tuple() {
return detail::singleton_manager::get_empty_tuple();
}
inline detail::decorated_names_map* get_decorated_names_map() {
return detail::singleton_manager::get_decorated_names_map();
}
} // namespace cppa
#endif // CPPA_SINGLETONS_HPP
......@@ -41,6 +41,9 @@ namespace cppa {
/**
* @brief Stores options passed to the @p spawn function family.
*/
#ifdef CPPA_DOCUMENTATION
class spawn_options { };
#else
enum class spawn_options : int {
no_flags = 0x00,
link_flag = 0x01,
......@@ -49,6 +52,7 @@ enum class spawn_options : int {
hide_flag = 0x08,
blocking_api_flag = 0x10
};
#endif
#ifndef CPPA_DOCUMENTATION
namespace {
......
......@@ -31,24 +31,36 @@
#ifndef CPPA_STACKED_ACTOR_MIXIN_HPP
#define CPPA_STACKED_ACTOR_MIXIN_HPP
#include <chrono>
#include <memory>
#include <functional>
#include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail {
namespace cppa {
template<class Derived, class Base>
class stacked_actor_mixin : public Base {
/**
* @tparam Base Either @p scheduled or @p threaded.
*/
template<class Base, class Subtype>
class stacked : public Base {
friend class receive_policy;
friend class detail::receive_policy;
friend class detail::behavior_stack;
protected:
typedef stacked combined_type;
public:
static constexpr auto receive_flag = detail::rp_nestable;
virtual void dequeue(partial_function& fun) {
m_recv_policy.receive(dthis(), fun);
}
......@@ -62,19 +74,28 @@ class stacked_actor_mixin : public Base {
}
virtual void run() {
if (!this->m_bhvr_stack.empty()) {
this->m_bhvr_stack.exec(m_recv_policy, dthis());
if (!dthis()->m_bhvr_stack.empty()) {
dthis()->exec_behavior_stack();
}
if (m_behavior) {
m_behavior();
}
}
protected:
inline void set_behavior(std::function<void()> fun) {
m_behavior = std::move(fun);
}
virtual void quit(std::uint32_t reason) {
this->cleanup(reason);
throw actor_exited(reason);
}
stacked_actor_mixin() = default;
protected:
stacked_actor_mixin(std::function<void()> f) : m_behavior(std::move(f)) { }
template<typename... Args>
stacked(std::function<void()> fun, Args&&... args)
: Base(std::forward<Args>(args)...), m_behavior(std::move(fun)) { }
virtual void do_become(behavior&& bhvr, bool discard_old) {
become_impl(std::move(bhvr), discard_old, message_id_t());
......@@ -85,36 +106,42 @@ class stacked_actor_mixin : public Base {
}
virtual bool has_behavior() {
return static_cast<bool>(m_behavior)
|| !this->m_bhvr_stack.empty();
return static_cast<bool>(m_behavior) || !dthis()->m_bhvr_stack.empty();
}
typedef std::chrono::time_point<std::chrono::high_resolution_clock>
timeout_type;
virtual timeout_type init_timeout(const util::duration& rel_time) = 0;
virtual detail::recursive_queue_node* await_message() = 0;
virtual detail::recursive_queue_node* await_message(const timeout_type& abs_time) = 0;
private:
std::function<void()> m_behavior;
receive_policy m_recv_policy;
inline Derived* dthis() {
return static_cast<Derived*>(this);
}
detail::receive_policy m_recv_policy;
void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) {
if (bhvr.timeout().valid()) {
dthis()->reset_timeout();
dthis()->request_timeout(bhvr.timeout());
}
if (!this->m_bhvr_stack.empty() && discard_old) {
this->m_bhvr_stack.pop_async_back();
if (!dthis()->m_bhvr_stack.empty() && discard_old) {
dthis()->m_bhvr_stack.pop_async_back();
}
this->m_bhvr_stack.push_back(std::move(bhvr), mid);
dthis()->m_bhvr_stack.push_back(std::move(bhvr), mid);
}
virtual void exec_behavior_stack() {
this->m_bhvr_stack.exec(m_recv_policy, dthis());
}
inline Subtype* dthis() { return static_cast<Subtype*>(this); }
};
} } // namespace cppa::detail
} // namespace cppa
#endif // CPPA_STACKED_ACTOR_MIXIN_HPP
......@@ -44,63 +44,39 @@
#include "cppa/atom.hpp"
#include "cppa/either.hpp"
#include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/stacked_actor_mixin.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/blocking_single_reader_queue.hpp"
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @brief An actor running in its own thread.
*/
class thread_mapped_actor : public local_actor {
protected:
/**
* @brief Implements the actor's behavior.
* Reimplemented this function for a class-based actor.
* Returning from this member function will end the
* execution of the actor.
*/
virtual void run();
};
#else // CPPA_DOCUMENTATION
class self_type;
class scheduler_helper;
class thread_mapped_actor;
namespace detail {
typedef intrusive::blocking_single_reader_queue<recursive_queue_node,disposer>
blocking_mailbox;
typedef stacked_actor_mixin<thread_mapped_actor,
abstract_actor<local_actor,blocking_mailbox> >
thread_mapped_actor_base;
template<>
struct has_blocking_receive<thread_mapped_actor> : std::true_type { };
} // namespace detail
class thread_mapped_actor : public detail::thread_mapped_actor_base {
friend class self_type; // needs access to cleanup()
/**
* @brief An actor running in its own thread.
* @extends local_actor
*/
class thread_mapped_actor : public extend<local_actor,thread_mapped_actor>::with<stacked,mailbox_based> {
friend class detail::behavior_stack; // needs access to receive_node()
friend class detail::receive_policy;
friend class self_type; // needs access to cleanup()
friend class scheduler_helper; // needs access to mailbox
friend class detail::receive_policy; // needs access to await_message(), etc.
friend class detail::behavior_stack; // needs same access as receive_policy
typedef detail::thread_mapped_actor_base super;
typedef combined_type super;
public:
......@@ -108,59 +84,48 @@ class thread_mapped_actor : public detail::thread_mapped_actor_base {
thread_mapped_actor(std::function<void()> fun);
void quit(std::uint32_t reason = exit_reason::normal); //override
inline void initialized(bool value) { m_initialized = value; }
void enqueue(actor* sender, any_tuple msg); //override
bool initialized() const;
void sync_enqueue(actor* sender, message_id_t id, any_tuple msg);
void enqueue(const actor_ptr& sender, any_tuple msg);
inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
void sync_enqueue(const actor_ptr& sender, message_id_t id, any_tuple msg);
inline void initialized(bool value) { m_initialized = value; }
// required by stacked_actor_mixin
inline void reset_timeout() { }
inline void request_timeout(const util::duration&) { }
inline void handle_timeout(behavior& bhvr) { bhvr.handle_timeout(); }
inline void pop_timeout() { }
inline void push_timeout() { }
inline bool waits_for_timeout(std::uint32_t) { return false; }
protected:
bool initialized();
typedef detail::recursive_queue_node mailbox_element;
typedef intrusive::blocking_single_reader_queue<mailbox_element,detail::disposer>
mailbox_type;
void cleanup(std::uint32_t reason);
private:
timeout_type init_timeout(const util::duration& rel_time);
detail::recursive_queue_node* await_message();
detail::recursive_queue_node* await_message(const timeout_type& abs_time);
bool m_initialized;
// required by nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_nestable;
inline void push_timeout() { }
inline void pop_timeout() { }
inline bool waits_for_timeout(std::uint32_t) { return false; }
inline detail::recursive_queue_node* receive_node() {
return m_mailbox.pop();
}
inline auto init_timeout(const util::duration& tout)
-> decltype(std::chrono::high_resolution_clock::now()) {
auto result = std::chrono::high_resolution_clock::now();
result += tout;
return result;
}
inline detail::recursive_queue_node* try_receive_node() {
return m_mailbox.try_pop();
}
template<typename Timeout>
inline detail::recursive_queue_node* try_receive_node(const Timeout& tout) {
return m_mailbox.try_pop(tout);
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
}
protected:
mailbox_type m_mailbox;
};
typedef intrusive_ptr<thread_mapped_actor> thread_mapped_actor_ptr;
#endif // CPPA_DOCUMENTATION
} // namespace cppa
#endif // CPPA_THREAD_BASED_ACTOR_HPP
......@@ -37,18 +37,16 @@
namespace cppa {
namespace detail { class behavior_impl; }
template<typename F>
struct timeout_definition {
static constexpr bool may_have_timeout = true;
util::duration timeout;
F handler;
detail::behavior_impl* as_behavior_impl() const;
};
template<typename T>
struct is_timeout_definition : std::false_type { };
template<typename F>
struct is_timeout_definition<timeout_definition<F> > : std::true_type { };
typedef timeout_definition<std::function<void()> > generic_timeout_definition;
} // namespace cppa
......
......@@ -35,12 +35,12 @@
namespace cppa { namespace util {
template<typename... BooleanConstants>
template<bool... Values>
struct disjunction;
template<typename Head, typename... Tail>
template<bool Head, bool... Tail>
struct disjunction<Head, Tail...>
: std::integral_constant<bool, Head::value || disjunction<Tail...>::value> {
: std::integral_constant<bool, Head || disjunction<Tail...>::value> {
};
template<>
......
......@@ -53,7 +53,7 @@ namespace cppa { namespace util {
template<typename What, typename With, typename... IfStmt>
struct replace_type {
static constexpr bool do_replace = disjunction<IfStmt...>::value;
static constexpr bool do_replace = disjunction<IfStmt::value...>::value;
typedef typename detail::replace_type<do_replace, What, With>::type
type;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace detail {
void abstract_scheduled_actor::request_timeout(const util::duration& d) {
if (d.valid()) {
if (d.is_zero()) {
// immediately enqueue timeout
auto node = super::fetch_node(this,
make_any_tuple(atom("SYNC_TOUT"),
++m_active_timeout_id));
this->m_mailbox.enqueue(node);
}
else {
get_scheduler()->delayed_send(
this, d,
make_any_tuple(
atom("SYNC_TOUT"), ++m_active_timeout_id));
}
m_has_pending_timeout_request = true;
}
else m_has_pending_timeout_request = false;
}
abstract_scheduled_actor::abstract_scheduled_actor(int state)
: super(true), m_state(state), m_has_pending_timeout_request(false)
, m_active_timeout_id(0) {
}
bool abstract_scheduled_actor::chained_enqueue(actor* sender, any_tuple msg) {
return enqueue_node(super::fetch_node(sender, std::move(msg)), pending);
}
bool abstract_scheduled_actor::chained_sync_enqueue(actor* sender, message_id_t id, any_tuple msg) {
bool failed = false;
bool result = enqueue_node(super::fetch_node(sender, std::move(msg), id), pending, &failed);
if (failed) {
sync_request_bouncer f{this, exit_reason()};
f(sender, id);
}
return result;
}
void abstract_scheduled_actor::quit(std::uint32_t reason) {
this->cleanup(reason);
throw actor_exited(reason);
}
void abstract_scheduled_actor::enqueue(actor* sender, any_tuple msg) {
enqueue_node(super::fetch_node(sender, std::move(msg)));
}
void abstract_scheduled_actor::sync_enqueue(actor* sender, message_id_t id, any_tuple msg) {
bool failed = false;
enqueue_node(super::fetch_node(sender, std::move(msg), id), ready, &failed);
if (failed) {
sync_request_bouncer f{this, exit_reason()};
f(sender, id);
}
}
int abstract_scheduled_actor::compare_exchange_state(int expected, int new_value) {
int e = expected;
do {
if (m_state.compare_exchange_weak(e, new_value)) {
return new_value;
}
}
while (e == expected);
return e;
}
bool abstract_scheduled_actor::enqueue_node(recursive_queue_node* node,
int next_state,
bool* failed) {
CPPA_REQUIRE(next_state == ready || next_state == pending);
CPPA_REQUIRE(node->marked == false);
switch (this->m_mailbox.enqueue(node)) {
case intrusive::first_enqueued: {
int state = m_state.load();
for (;;) {
switch (state) {
case blocked: {
if (m_state.compare_exchange_weak(state, next_state)) {
CPPA_REQUIRE(this->m_scheduler != nullptr);
if (next_state == ready) {
this->m_scheduler->enqueue(this);
}
return true;
}
break;
}
case about_to_block: {
if (m_state.compare_exchange_weak(state, ready)) {
return false;
}
break;
}
default: return false;
}
}
break;
}
case intrusive::queue_closed: {
if (failed) *failed = true;
break;
}
default: break;
}
return false;
}
} } // namespace cppa::detail
......@@ -32,6 +32,8 @@
namespace cppa { namespace detail {
abstract_tuple::abstract_tuple(bool is_dynamic) : m_is_dynamic(is_dynamic) { }
bool abstract_tuple::equals(const abstract_tuple &other) const {
return this == &other
|| ( size() == other.size()
......@@ -39,9 +41,7 @@ bool abstract_tuple::equals(const abstract_tuple &other) const {
}
abstract_tuple::abstract_tuple(const abstract_tuple& other)
: ref_counted()
, m_impl_type(other.m_impl_type) {
}
: m_is_dynamic(other.m_is_dynamic) { }
const std::type_info* abstract_tuple::type_token() const {
return &typeid(void);
......
......@@ -36,7 +36,10 @@
#include <stdexcept>
#include "cppa/actor.hpp"
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/singletons.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......@@ -46,26 +49,161 @@
namespace cppa {
namespace {
inline detail::actor_registry& registry() {
return *(detail::singleton_manager::get_actor_registry());
}
namespace { typedef std::unique_lock<std::mutex> guard_type; }
} // namespace <anonymous>
// m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
actor::actor(actor_id aid) : m_id(aid), m_is_proxy(true) { }
actor::actor(actor_id aid)
: m_id(aid), m_is_proxy(true), m_exit_reason(exit_reason::not_exited) { }
actor::actor() : m_id(registry().next_id()), m_is_proxy(false) { }
actor::actor()
: m_id(get_actor_registry()->next_id()), m_is_proxy(false)
, m_exit_reason(exit_reason::not_exited) {
CPPA_LOG_INFO("spawned new actor with ID " << id()
<< ", class = " << CPPA_CLASS_NAME);
}
bool actor::chained_enqueue(actor* sender, any_tuple msg) {
bool actor::chained_enqueue(const actor_ptr& sender, any_tuple msg) {
enqueue(sender, std::move(msg));
return false;
}
bool actor::chained_sync_enqueue(actor* ptr, message_id_t id, any_tuple msg) {
bool actor::chained_sync_enqueue(const actor_ptr& ptr, message_id_t id, any_tuple msg) {
sync_enqueue(ptr, id, std::move(msg));
return false;
}
bool actor::link_to_impl(const actor_ptr& other) {
if (other && other != this) {
guard_type guard(m_mtx);
// send exit message if already exited
if (exited()) {
other->enqueue(this, make_any_tuple(atom("EXIT"), exit_reason()));
}
// add link if not already linked to other
// (checked by establish_backlink)
else if (other->establish_backlink(this)) {
m_links.push_back(other);
return true;
}
}
return false;
}
bool actor::attach(attachable_ptr ptr) {
if (ptr == nullptr) {
guard_type guard(m_mtx);
return m_exit_reason == exit_reason::not_exited;
}
std::uint32_t reason;
{ // lifetime scope of guard
guard_type guard(m_mtx);
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
m_attachables.push_back(std::move(ptr));
return true;
}
}
ptr->actor_exited(reason);
return false;
}
void actor::detach(const attachable::token& what) {
attachable_ptr ptr;
{ // lifetime scope of guard
guard_type guard(m_mtx);
auto end = m_attachables.end();
auto i = std::find_if(
m_attachables.begin(), end,
[&](attachable_ptr& p) { return p->matches(what); });
if (i != end) {
ptr = std::move(*i);
m_attachables.erase(i);
}
}
// ptr will be destroyed here, without locked mutex
}
void actor::link_to(const actor_ptr& other) {
static_cast<void>(link_to_impl(other));
}
void actor::unlink_from(const actor_ptr& other) {
static_cast<void>(unlink_from_impl(other));
}
bool actor::remove_backlink(const actor_ptr& other) {
if (other && other != this) {
guard_type guard(m_mtx);
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i != m_links.end()) {
m_links.erase(i);
return true;
}
}
return false;
}
bool actor::establish_backlink(const actor_ptr& other) {
std::uint32_t reason = exit_reason::not_exited;
if (other && other != this) {
guard_type guard(m_mtx);
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i == m_links.end()) {
m_links.push_back(other);
return true;
}
}
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
other->enqueue(this, make_any_tuple(atom("EXIT"), reason));
}
return false;
}
bool actor::unlink_from_impl(const actor_ptr& other) {
guard_type guard(m_mtx);
// remove_backlink returns true if this actor is linked to other
if (other && !exited() && other->remove_backlink(this)) {
auto i = std::find(m_links.begin(), m_links.end(), other);
CPPA_REQUIRE(i != m_links.end());
m_links.erase(i);
return true;
}
return false;
}
void actor::cleanup(std::uint32_t reason) {
CPPA_REQUIRE(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
decltype(m_links) mlinks;
decltype(m_attachables) mattachables;
{ // lifetime scope of guard
guard_type guard(m_mtx);
if (m_exit_reason != exit_reason::not_exited) {
// already exited
return;
}
m_exit_reason = reason;
mlinks = std::move(m_links);
mattachables = std::move(m_attachables);
// make sure lists are empty
m_links.clear();
m_attachables.clear();
}
// send exit messages
for (actor_ptr& aptr : mlinks) {
aptr->enqueue(this, make_any_tuple(atom("EXIT"), reason));
}
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
}
}
} // namespace cppa
......@@ -76,7 +76,7 @@ void actor_registry::put(actor_id key, const actor_ptr& value) {
}
}
if (add_attachable) {
CPPA_LOG_INFO("added " << key);
CPPA_LOG_INFO("added actor with ID " << key);
struct eraser : attachable {
actor_id m_id;
actor_registry* m_registry;
......@@ -88,7 +88,7 @@ void actor_registry::put(actor_id key, const actor_ptr& value) {
return false;
}
};
value->attach(new eraser(key, this));
value->attach(attachable_ptr{new eraser(key, this)});
}
}
......@@ -97,7 +97,8 @@ void actor_registry::erase(actor_id key, std::uint32_t reason) {
auto i = m_entries.find(key);
if (i != m_entries.end()) {
auto& entry = i->second;
CPPA_LOG_INFO("erased " << key << ", reason = " << std::hex << reason);
CPPA_LOG_INFO("erased actor with ID " << key
<< ", reason " << reason);
entry.first = nullptr;
entry.second = reason;
}
......
......@@ -29,26 +29,20 @@
#include "cppa/any_tuple.hpp"
#include "cppa/singletons.hpp"
#include "cppa/detail/empty_tuple.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
namespace {
inline detail::empty_tuple* s_empty_tuple() {
return detail::singleton_manager::get_empty_tuple();
}
} // namespace <anonymous>
any_tuple::any_tuple() : m_vals(s_empty_tuple()) { }
any_tuple::any_tuple() : m_vals(get_empty_tuple()) { }
any_tuple::any_tuple(detail::abstract_tuple* ptr) : m_vals(ptr) { }
any_tuple::any_tuple(any_tuple&& other) : m_vals(s_empty_tuple()) {
any_tuple::any_tuple(any_tuple&& other) : m_vals(get_empty_tuple()) {
m_vals.swap(other.m_vals);
}
any_tuple::any_tuple(const value_ptr& vals) : m_vals(vals) { }
any_tuple::any_tuple(const data_ptr& vals) : m_vals(vals) { }
any_tuple& any_tuple::operator=(any_tuple&& other) {
m_vals.swap(other.m_vals);
......@@ -56,7 +50,7 @@ any_tuple& any_tuple::operator=(any_tuple&& other) {
}
void any_tuple::reset() {
m_vals.reset(s_empty_tuple());
m_vals.reset(get_empty_tuple());
}
size_t any_tuple::size() const {
......
......@@ -32,7 +32,9 @@
namespace cppa {
attachable::~attachable() {
}
attachable::~attachable() { }
attachable::token::token(const std::type_info& tinfo, const void* vptr)
: subtype(tinfo), ptr(vptr) { }
} // namespace cppa::detail
......@@ -38,12 +38,37 @@
namespace cppa {
context_switching_actor::context_switching_actor()
: m_fiber(&context_switching_actor::trampoline, this) {
context_switching_actor::context_switching_actor(std::function<void()> fun)
: super(std::move(fun), actor_state::ready, true)
, m_fiber(&context_switching_actor::trampoline, this) { }
auto context_switching_actor::init_timeout(const util::duration& tout) -> timeout_type {
// request explicit timeout message
request_timeout(tout);
return {};
}
context_switching_actor::context_switching_actor(std::function<void()> fun)
: super(std::move(fun)), m_fiber(&context_switching_actor::trampoline, this) {
detail::recursive_queue_node* context_switching_actor::await_message(const timeout_type&) {
// receives requested timeout message if timeout occured
return await_message();
}
detail::recursive_queue_node* context_switching_actor::await_message() {
auto e = m_mailbox.try_pop();
while (e == nullptr) {
if (m_mailbox.can_fetch_more() == false) {
set_state(actor_state::about_to_block);
// make sure mailbox is empty
if (m_mailbox.can_fetch_more()) {
// someone preempt us => continue
set_state(actor_state::ready);
}
// wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked);
}
e = m_mailbox.try_pop();
}
return e;
}
void context_switching_actor::trampoline(void* this_ptr) {
......@@ -64,31 +89,12 @@ void context_switching_actor::trampoline(void* this_ptr) {
detail::yield(detail::yield_state::done);
}
detail::recursive_queue_node* context_switching_actor::receive_node() {
detail::recursive_queue_node* e = m_mailbox.try_pop();
while (e == nullptr) {
if (m_mailbox.can_fetch_more() == false) {
m_state.store(abstract_scheduled_actor::about_to_block);
// make sure mailbox is empty
if (m_mailbox.can_fetch_more()) {
// someone preempt us => continue
m_state.store(abstract_scheduled_actor::ready);
}
else {
// wait until actor becomes rescheduled
detail::yield(detail::yield_state::blocked);
}
}
e = m_mailbox.try_pop();
}
return e;
}
scheduled_actor_type context_switching_actor::impl_type() {
return context_switching_impl;
}
resume_result context_switching_actor::resume(util::fiber* from, actor_ptr& next_job) {
CPPA_LOG_TRACE("id = " << id() << ", state = " << static_cast<int>(state()));
CPPA_REQUIRE(from != nullptr);
CPPA_REQUIRE(next_job == nullptr);
using namespace detail;
......@@ -107,15 +113,15 @@ resume_result context_switching_actor::resume(util::fiber* from, actor_ptr& next
CPPA_REQUIRE(next_job == nullptr);
m_chained_actor.swap(next_job);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (compare_exchange_state(abstract_scheduled_actor::about_to_block,
abstract_scheduled_actor::blocked)) {
case abstract_scheduled_actor::ready: {
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
m_chained_actor.swap(next_job);
CPPA_REQUIRE(next_job == nullptr);
break;
}
case abstract_scheduled_actor::blocked: {
case actor_state::blocked: {
// wait until someone re-schedules that actor
return resume_result::actor_blocked;
}
......
......@@ -63,7 +63,7 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
else {
// local actor?
if (!ptr->is_proxy()) {
detail::singleton_manager::get_actor_registry()->put(ptr->id(), ptr);
get_actor_registry()->put(ptr->id(), ptr);
}
auto pinf = m_pinf;
if (ptr->is_proxy()) {
......@@ -101,7 +101,7 @@ actor_ptr default_actor_addressing::read(deserializer* source) {
// local actor?
auto pinf = process_information::get();
if (pid == pinf->process_id() && nid == pinf->node_id()) {
return detail::singleton_manager::get_actor_registry()->get(aid);
return get_actor_registry()->get(aid);
}
else {
process_information tmp(pid, nid);
......
......@@ -36,6 +36,7 @@
#include "cppa/detail/memory.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
using namespace std;
......@@ -54,13 +55,13 @@ default_actor_proxy::default_actor_proxy(actor_id mid,
: super(mid), m_proto(parent), m_pinf(pinfo) { }
default_actor_proxy::~default_actor_proxy() {
CPPA_LOG_TRACE("node = " << to_string(*m_pinf) << ", aid = " << id());
auto aid = id();
auto node = m_pinf;
auto proto = m_proto;
proto->run_later([aid, node, proto] {
CPPA_LOGF_TRACE("lambda from ~default_actor_proxy"
<< "; node = " << to_string(*node) << ", aid " << aid
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"~default_actor_proxy$run_later",
"node = " << to_string(*node) << ", aid " << aid
<< ", proto = " << to_string(proto->identifier()));
proto->addressing()->erase(*node, aid);
auto p = proto->get_peer(*node);
......@@ -94,10 +95,12 @@ void default_actor_proxy::forward_msg(const actor_ptr& sender,
switch (m_pending_requests.enqueue(new_req_info(sender, mid))) {
case intrusive::queue_closed: {
if (sender) {
intrusive_ptr<default_actor_proxy> _this{this};
auto rsn = exit_reason();
m_proto->run_later([_this,rsn,sender,mid] {
detail::sync_request_bouncer f{_this.get(), rsn};
m_proto->run_later([rsn,sender,mid] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"forward_msg$bouncer",
"bounce message for reason " << rsn);
detail::sync_request_bouncer f{rsn};
f(sender.get(), mid);
});
}
......@@ -110,13 +113,16 @@ void default_actor_proxy::forward_msg(const actor_ptr& sender,
auto node = m_pinf;
auto proto = m_proto;
m_proto->run_later([hdr, msg, node, proto] {
CPPA_LOGF_TRACE("lambda from default_actor_proxy::forward_msg");
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"forward_msg$forwarder",
"");
proto->enqueue(*node, hdr, msg);
});
}
void default_actor_proxy::enqueue(actor* sender, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_ARG(sender) << ", " << CPPA_TARG(msg, to_string));
void default_actor_proxy::enqueue(const actor_ptr& sender, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< CPPA_TARG(msg, to_string));
auto& arr = detail::static_types_array<atom_value, uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
......@@ -126,10 +132,11 @@ void default_actor_proxy::enqueue(actor* sender, any_tuple msg) {
intrusive_ptr<default_actor_proxy> _this{this};
auto reason = msg.get_as<uint32_t>(1);
m_proto->run_later([_this, reason] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"enqueue$kill_proxy_helper",
"KILL_PROXY with exit reason " << reason);
_this->cleanup(reason);
// make sure cleanup is done before closing requests queue
std::atomic_thread_fence(std::memory_order_seq_cst);
detail::sync_request_bouncer f{_this.get(), reason};
detail::sync_request_bouncer f{reason};
_this->m_pending_requests.close([&](const sync_request_info& e) {
f(e.sender.get(), e.mid);
});
......@@ -139,9 +146,12 @@ void default_actor_proxy::enqueue(actor* sender, any_tuple msg) {
forward_msg(sender, move(msg));
}
void default_actor_proxy::sync_enqueue(actor* sender, message_id_t mid, any_tuple msg) {
CPPA_LOG_TRACE(CPPA_ARG(sender) << ", " << CPPA_MARG(mid, integer_value)
<< ", " << CPPA_TARG(msg, to_string));
void default_actor_proxy::sync_enqueue(const actor_ptr& sender,
message_id_t mid,
any_tuple msg) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< CPPA_MARG(mid, integer_value) << ", "
<< CPPA_TARG(msg, to_string));
forward_msg(sender, move(msg), mid);
}
......
......@@ -36,6 +36,7 @@
#include "cppa/match.hpp"
#include "cppa/logging.hpp"
#include "cppa/to_string.hpp"
#include "cppa/singletons.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/binary_serializer.hpp"
......@@ -195,7 +196,7 @@ void default_peer::monitor(const actor_ptr&,
CPPA_LOG_ERROR("received MONITOR from invalid peer");
return;
}
auto entry = detail::singleton_manager::get_actor_registry()->get_entry(aid);
auto entry = get_actor_registry()->get_entry(aid);
auto pself = process_information::get();
if (*node == *pself) {
......@@ -221,7 +222,9 @@ void default_peer::monitor(const actor_ptr&,
default_protocol_ptr proto = m_parent;
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
CPPA_LOGF_TRACE("lambda from default_peer::monitor");
CPPA_LOGC_TRACE("cppa::network::default_peer",
"monitor$kill_proxy_helper",
"reason = " << reason);
auto p = proto->get_peer(*node);
if (p) p->enqueue(make_any_tuple(atom("KILL_PROXY"), pself, aid, reason));
});
......
......@@ -34,6 +34,7 @@
#include "cppa/logging.hpp"
#include "cppa/to_string.hpp"
#include "cppa/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
......@@ -88,11 +89,12 @@ void default_protocol::publish(const actor_ptr& whom,
if (!whom) return;
CPPA_REQUIRE(args.size() == 0);
static_cast<void>(args); // keep compiler happy
singleton_manager::get_actor_registry()->put(whom->id(), whom);
get_actor_registry()->put(whom->id(), whom);
default_protocol_ptr proto = this;
auto impl = make_counted<default_peer_acceptor>(this, move(ptr), whom);
run_later([=] {
CPPA_LOGF_TRACE("lambda from default_protocol::publish");
CPPA_LOGC_TRACE("cppa::network::default_protocol",
"publish$add_acceptor", "");
proto->m_acceptors[whom].push_back(impl);
proto->continue_reader(impl.get());
});
......@@ -102,11 +104,10 @@ void default_protocol::unpublish(const actor_ptr& whom) {
CPPA_LOG_TRACE("whom = " << to_string(whom));
default_protocol_ptr proto = this;
run_later([=] {
CPPA_LOGF_TRACE("lambda from default_protocol::unpublish");
CPPA_LOGC_TRACE("cppa::network::default_protocol",
"unpublish$remove_acceptors", "");
auto& acceptors = m_acceptors[whom];
for (auto& ptr : acceptors) {
proto->stop_reader(ptr.get());
}
for (auto& ptr : acceptors) proto->stop_reader(ptr.get());
m_acceptors.erase(whom);
});
}
......@@ -123,8 +124,12 @@ void default_protocol::register_peer(const process_information& node,
auto tmp = entry.queue->pop();
ptr->enqueue(tmp.first, tmp.second);
}
CPPA_LOG_INFO("peer " << to_string(node) << " added");
}
else {
CPPA_LOG_WARNING("peer " << to_string(node) << " already defined, "
"multiple calls to remote_actor()?");
}
else { CPPA_LOG_ERROR("peer " << to_string(node) << " already defined"); }
}
default_peer_ptr default_protocol::get_peer(const process_information& n) {
......@@ -192,12 +197,13 @@ actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
std::cerr << "*** warning: remote_actor() called to access a local actor\n"
<< std::flush;
# endif
return singleton_manager::get_actor_registry()->get(remote_aid);
return get_actor_registry()->get(remote_aid);
}
default_protocol_ptr proto = this;
intrusive::blocking_single_reader_queue<remote_actor_result> q;
run_later([proto, io, pinfptr, remote_aid, &q] {
CPPA_LOGF_TRACE("lambda from default_protocol::remote_actor");
CPPA_LOGC_TRACE("cppa::network::default_protocol",
"remote_actor$create_connection", "");
auto pp = proto->get_peer(*pinfptr);
CPPA_LOGF_INFO_IF(pp, "connection already exists (re-use old one)");
if (!pp) proto->new_peer(io.first, io.second, pinfptr);
......
......@@ -33,7 +33,7 @@
namespace cppa { namespace detail {
empty_tuple::empty_tuple() : super(tuple_impl_info::statically_typed) { }
empty_tuple::empty_tuple() : super(false) { }
size_t empty_tuple::size() const {
return 0;
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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