Commit c3beb4d6 authored by Dominik Charousset's avatar Dominik Charousset

revamped spawn to use event-based impl by default

this patch turns libcppa's spawn function family upside down;
actors are now event-based by default, but users can opt out
by using `spawn<blocking_api>(...)`;
this patch also removes the old `scheduling_hint` interface and
replaces it with the more versatile `spawn_options`, e.g.,
`spawn_monitor<detached>(...)` becomes `spawn<monitored + detached>(...)`
parent 55bc8869
......@@ -125,7 +125,6 @@ cppa/response_handle.hpp
cppa/sb_actor.hpp
cppa/scheduled_actor.hpp
cppa/scheduler.hpp
cppa/scheduling_hint.hpp
cppa/self.hpp
cppa/serializer.hpp
cppa/thread_mapped_actor.hpp
......@@ -293,3 +292,4 @@ cppa/memory_cached_mixin.hpp
unit_testing/test.cpp
src/exit_reason.cpp
src/on.cpp
cppa/spawn_options.hpp
......@@ -37,6 +37,7 @@
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/attachable.hpp"
#include "cppa/message_id.hpp"
......
......@@ -55,10 +55,10 @@
#include "cppa/tuple_cast.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/message_future.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/util/rm_ref.hpp"
......@@ -417,11 +417,11 @@ inline void send_impl(T* whom, any_tuple&& what) {
if (whom) self->send_message(whom, std::move(what));
}
template<typename T, typename... Args>
inline void send_tpl_impl(T* whom, Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send");
template<typename T, typename... Ts>
inline void send_tpl_impl(T* whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
if (whom) self->send_message(whom,
make_any_tuple(std::forward<Args>(what)...));
make_any_tuple(std::forward<Ts>(what)...));
}
} // namespace detail
......@@ -436,7 +436,7 @@ inline void send_tpl_impl(T* whom, Args&&... what) {
* @param whom Receiver of the message.
* @param what Message content as tuple.
*/
template<class C, typename... Args>
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send_tuple(const intrusive_ptr<C>& whom, any_tuple what) {
detail::send_impl(whom.get(), std::move(what));
......@@ -446,12 +446,12 @@ send_tuple(const intrusive_ptr<C>& whom, any_tuple what) {
* @brief Sends <tt>{what...}</tt> as a message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Args) > 0</tt>
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Args>
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send(const intrusive_ptr<C>& whom, Args&&... what) {
detail::send_tpl_impl(whom.get(), std::forward<Args>(what)...);
send(const intrusive_ptr<C>& whom, Ts&&... what) {
detail::send_tpl_impl(whom.get(), std::forward<Ts>(what)...);
}
/**
......@@ -460,9 +460,9 @@ send(const intrusive_ptr<C>& whom, Args&&... what) {
* @param from Sender as seen by @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Args) > 0</tt>
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Args>
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send_tuple_as(const actor_ptr& from, const intrusive_ptr<C>& whom, any_tuple what) {
if (whom) whom->enqueue(from.get(), std::move(what));
......@@ -474,12 +474,12 @@ send_tuple_as(const actor_ptr& from, const intrusive_ptr<C>& whom, any_tuple wha
* @param from Sender as seen by @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Args) > 0</tt>
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Args>
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send_as(const actor_ptr& from, const intrusive_ptr<C>& whom, Args&&... what) {
send_tuple_as(from, whom, make_any_tuple(std::forward<Args>(what)...));
send_as(const actor_ptr& from, const intrusive_ptr<C>& whom, Ts&&... what) {
send_tuple_as(from, whom, make_any_tuple(std::forward<Ts>(what)...));
}
/**
......@@ -522,13 +522,13 @@ inline message_future sync_send_tuple(const actor_ptr& whom, any_tuple what) {
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @pre <tt>sizeof...(Args) > 0</tt>
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Args>
inline message_future sync_send(const actor_ptr& whom, Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send");
return sync_send_tuple(whom, make_any_tuple(std::forward<Args>(what)...));
template<typename... Ts>
inline message_future sync_send(const actor_ptr& whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(whom, make_any_tuple(std::forward<Ts>(what)...));
}
/**
......@@ -543,7 +543,7 @@ inline message_future sync_send(const actor_ptr& whom, Args&&... what) {
* message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
message_future timed_sync_send_tuple(const actor_ptr& whom,
const std::chrono::duration<Rep,Period>& rel_time,
any_tuple what) {
......@@ -564,17 +564,17 @@ message_future timed_sync_send_tuple(const actor_ptr& whom,
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the sent
* message cannot be received by another actor.
* @pre <tt>sizeof...(Args) > 0</tt>
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
message_future timed_sync_send(const actor_ptr& whom,
const std::chrono::duration<Rep,Period>& rel_time,
Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send");
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(whom,
rel_time,
make_any_tuple(std::forward<Args>(what)...));
make_any_tuple(std::forward<Ts>(what)...));
}
/**
......@@ -589,18 +589,18 @@ inline void reply_tuple(any_tuple what) {
* @brief Sends a message to the sender of the last received message.
* @param what Message elements.
*/
template<typename... Args>
inline void reply(Args&&... what) {
self->reply_message(make_any_tuple(std::forward<Args>(what)...));
template<typename... Ts>
inline void reply(Ts&&... what) {
self->reply_message(make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends a message as reply to @p handle.
*/
template<typename... Args>
inline void reply_to(const response_handle& handle, Args&&... what) {
template<typename... Ts>
inline void reply_to(const response_handle& handle, Ts&&... what) {
if (handle.valid()) {
handle.apply(make_any_tuple(std::forward<Args>(what)...));
handle.apply(make_any_tuple(std::forward<Ts>(what)...));
}
}
......@@ -626,7 +626,7 @@ inline void forward_to(const actor_ptr& whom) {
* microseconds, milliseconds, seconds or minutes.
* @param what Message content as a tuple.
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
inline void delayed_send_tuple(const channel_ptr& whom,
const std::chrono::duration<Rep,Period>& rtime,
any_tuple what) {
......@@ -640,15 +640,15 @@ inline void delayed_send_tuple(const channel_ptr& whom,
* microseconds, milliseconds, seconds or minutes.
* @param what Message elements.
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
inline void delayed_send(const channel_ptr& whom,
const std::chrono::duration<Rep,Period>& rtime,
Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send");
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
if (whom) {
delayed_send_tuple(whom,
rtime,
make_any_tuple(std::forward<Args>(what)...));
make_any_tuple(std::forward<Ts>(what)...));
}
}
......@@ -659,7 +659,7 @@ inline void delayed_send(const channel_ptr& whom,
* @param what Message content as a tuple.
* @see delayed_send()
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
inline void delayed_reply_tuple(const std::chrono::duration<Rep, Period>& rtime,
any_tuple what) {
get_scheduler()->delayed_reply(self->last_sender(),
......@@ -675,10 +675,10 @@ inline void delayed_reply_tuple(const std::chrono::duration<Rep, Period>& rtime,
* @param what Message elements.
* @see delayed_send()
*/
template<class Rep, class Period, typename... Args>
template<class Rep, class Period, typename... Ts>
inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
Args&&... what) {
delayed_reply_tuple(rtime, make_any_tuple(std::forward<Args>(what)...));
Ts&&... what) {
delayed_reply_tuple(rtime, make_any_tuple(std::forward<Ts>(what)...));
}
/** @} */
......@@ -688,14 +688,19 @@ inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
inline void send_tuple(local_actor* whom, any_tuple what) {
detail::send_impl(whom, std::move(what));
}
template<typename... Args>
inline void send(local_actor* whom, Args&&... args) {
detail::send_tpl_impl(whom, std::forward<Args>(args)...);
template<typename... Ts>
inline void send(local_actor* whom, Ts&&... args) {
detail::send_tpl_impl(whom, std::forward<Ts>(args)...);
}
inline const self_type& operator<<(const self_type& s, any_tuple what) {
detail::send_impl(s.get(), std::move(what));
return s;
}
inline actor_ptr eval_sopts(spawn_options opts, actor_ptr ptr) {
if (has_monitor_flag(opts)) self->monitor(ptr);
if (has_link_flag(opts)) self->link_to(ptr);
return std::move(ptr);
}
#endif // CPPA_DOCUMENTATION
/**
......@@ -704,213 +709,66 @@ inline const self_type& operator<<(const self_type& s, any_tuple what) {
*/
/**
* @brief Spawns a new context-switching or thread-mapped {@link actor}
* that executes <tt>fun(args...)</tt>.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @tparam Hint A hint to the scheduler for the best scheduling strategy.
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<scheduling_hint Hint, typename Fun, typename... Args>
actor_ptr spawn(Fun&& fun, Args&&... args) {
return get_scheduler()->spawn_impl(Hint,
std::forward<Fun>(fun),
std::forward<Args>(args)...);
template<spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn(Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
return eval_sopts(Options,
get_scheduler()->exec(Options,
scheduler::init_callback{},
std::forward<Ts>(args)...));
}
/**
* @brief Spawns a new context-switching {@link actor}
* that executes <tt>fun(args...)</tt>.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note This function is equal to <tt>spawn<scheduled>(fun, args...)</tt>.
*/
template<typename Fun, typename... Args>
actor_ptr spawn(Fun&& fun, Args&&... args) {
return spawn<scheduled>(std::forward<Fun>(fun),
std::forward<Args>(args)...);
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn(Ts&&... args) {
auto rawptr = detail::memory::create<Impl>(std::forward<Ts>(args)...);
return eval_sopts(Options, get_scheduler()->exec(Options, rawptr));
}
/**
* @brief Spawns a new context-switching or thread-mapped {@link actor}
* that executes <tt>fun(args...)</tt> and
* joins @p grp immediately.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @tparam Hint A hint to the scheduler for the best scheduling strategy.
* @brief Spawns a new actor that evaluates given arguments and
* immediately joins @p grp.
* @param args A functor followed by its arguments.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note The spawned actor joins @p grp after its
* {@link local_actor::init() init} member function is called but
* before it is executed. Hence, the spawned actor already joined
* the group before this function returns.
* @note The spawned has joined the group before this function returns.
*/
template<scheduling_hint Hint, typename Fun, typename... Args>
actor_ptr spawn_in_group(const group_ptr& grp, Fun&& fun, Args&&... args) {
return get_scheduler()->spawn_cb_impl(Hint,
[grp](local_actor* ptr) {
template<spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
auto init_cb = [=](local_actor* ptr) {
ptr->join(grp);
},
std::forward<Fun>(fun),
std::forward<Args>(args)...);
}
/**
* @brief Spawns a new context-switching {@link actor}
* that executes <tt>fun(args...)</tt> and
* joins @p grp immediately.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note The spawned actor joins @p grp after its
* {@link local_actor::init() init} member function is called but
* before it is executed. Hence, the spawned actor already joined
* the group before this function returns.
* @note This function is equal to
* <tt>spawn_in_group<scheduled>(fun, args...)</tt>.
*/
template<typename Fun, typename... Args>
actor_ptr spawn_in_group(Fun&& fun, Args&&... args) {
return spawn_in_group<scheduled>(std::forward<Fun>(fun),
std::forward<Args>(args)...);
}
/**
* @brief Spawns an actor of type @p ActorImpl.
* @param args Optional constructor arguments.
* @tparam ActorImpl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn(Args&&... args) {
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr);
}
/**
* @brief Spawns an actor of type @p ActorImpl that joins @p grp immediately.
* @param grp The group that the newly created actor shall join.
* @param args Optional constructor arguments.
* @tparam ActorImpl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note The spawned actor joins @p grp after its
* {@link local_actor::init() init} member function is called but
* before it is executed. Hence, the spawned actor already joined
* the group before this function returns.
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn_in_group(const group_ptr& grp, Args&&... args) {
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, [&](local_actor* p) { p->join(grp); });
}
#ifndef CPPA_DOCUMENTATION
template<class ActorImpl, typename... Args>
actor_ptr spawn_hidden_in_group(const group_ptr& grp, Args&&... args) {
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, [&](local_actor* p) { p->join(grp); },
scheduled_and_hidden);
};
return eval_sopts(Options,
get_scheduler()->exec(Options,
init_cb,
std::forward<Ts>(args)...));
}
template<class ActorImpl, typename... Args>
actor_ptr spawn_hidden(Args&&... args) {
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, scheduled_and_hidden);
}
#endif
/**
* @brief Spawns a new context-switching or thread-mapped {@link actor}
* that executes <tt>fun(args...)</tt> and creates a link between the
* calling actor and the new actor.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @tparam Hint A hint to the scheduler for the best scheduling strategy.
* @brief Spawns an actor of type @p Impl that immediately joins @p grp.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns.
*/
template<scheduling_hint Hint, typename Fun, typename... Args>
actor_ptr spawn_link(Fun&& fun, Args&&... args) {
auto res = spawn<Hint>(std::forward<Fun>(fun), std::forward<Args>(args)...);
self->link_to(res);
return res;
}
/**
* @brief Spawns a new context-switching {@link actor}
* that executes <tt>fun(args...)</tt> and creates a link between the
* calling actor and the new actor.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note This function is equal to <tt>spawn<scheduled>(fun, args...)</tt>.
*/
template<typename Fun, typename... Args>
actor_ptr spawn_link(Fun&& fun, Args&&... args) {
auto res = spawn(std::forward<Fun>(fun), std::forward<Args>(args)...);
self->link_to(res);
return res;
}
/**
* @brief Spawns an actor of type @p ActorImpl and creates a link between the
* calling actor and the new actor.
* @param args Optional constructor arguments.
* @tparam ActorImpl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn_link(Args&&... args) {
auto res = spawn<ActorImpl>(std::forward<Args>(args)...);
self->link_to(res);
return res;
}
/**
* @brief Spawns a new context-switching or thread-mapped {@link actor}
* that executes <tt>fun(args...)</tt> and adds a monitor to the
* new actor.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @tparam Hint A hint to the scheduler for the best scheduling strategy.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<scheduling_hint Hint, typename Fun, typename... Args>
actor_ptr spawn_monitor(Fun&& fun, Args&&... args) {
auto res = spawn<Hint>(std::forward<Fun>(fun), std::forward<Args>(args)...);
self->monitor(res);
return res;
}
/**
* @brief Spawns a new context-switching {@link actor}
* that executes <tt>fun(args...)</tt> and adds a monitor to the
* new actor.
* @param fun A function implementing the actor's behavior.
* @param args Optional function parameters for @p fun.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @note This function is equal to <tt>spawn<scheduled>(fun, args...)</tt>.
*/
template<typename Fun, typename... Args>
actor_ptr spawn_monitor(Fun&& fun, Args&&... args) {
auto res = spawn(std::forward<Fun>(fun), std::forward<Args>(args)...);
self->monitor(res);
return res;
}
/**
* @brief Spawns an actor of type @p ActorImpl and adds a monitor to the
* new actor.
* @param args Optional constructor arguments.
* @tparam ActorImpl Subtype of {@link event_based_actor} or {@link sb_actor}.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn_monitor(Args&&... args) {
auto res = spawn<ActorImpl>(std::forward<Args>(args)...);
self->monitor(res);
return res;
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_in_group(const group_ptr& grp, Ts&&... args) {
auto rawptr = detail::memory::create<Impl>(std::forward<Ts>(args)...);
rawptr->join(grp);
return eval_sopts(Options, get_scheduler()->exec(Options, rawptr));
}
/** @} */
......@@ -988,6 +846,13 @@ inline void quit_actor(const actor_ptr& whom, std::uint32_t reason) {
send(whom, atom("EXIT"), reason);
}
template<typename... Ts>
inline void become(Ts&&... args) {
self->become(std::forward<Ts>(args)...);
}
inline void unbecome() { self->unbecome(); }
struct actor_ostream {
typedef const actor_ostream& (*fun_type)(const actor_ostream&);
......
......@@ -104,8 +104,10 @@ class event_based_actor_factory {
template<typename... Args>
actor_ptr spawn(Args&&... args) {
return get_scheduler()->spawn(memory::create<impl>(m_init, m_on_exit,
std::forward<Args>(args)...));
auto ptr = memory::create<impl>(m_init,
m_on_exit,
std::forward<Args>(args)...);
return get_scheduler()->exec(no_spawn_options, ptr);
}
private:
......
......@@ -47,31 +47,24 @@ class thread_pool_scheduler : public scheduler {
public:
using scheduler::init_callback;
using scheduler::void_function;
struct worker;
thread_pool_scheduler();
thread_pool_scheduler(size_t num_worker_threads);
void initialize() /*override*/;
void destroy() /*override*/;
void enqueue(scheduled_actor* what) /*override*/;
void initialize();
actor_ptr spawn(scheduled_actor* what,
scheduling_hint hint);
void destroy();
actor_ptr spawn(scheduled_actor* what,
init_callback init_cb,
scheduling_hint hint);
void enqueue(scheduled_actor* what);
actor_ptr spawn(void_function fun,
scheduling_hint hint);
actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr);
actor_ptr spawn(void_function what,
init_callback init_cb,
scheduling_hint hint);
actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f);
private:
......@@ -86,10 +79,6 @@ class thread_pool_scheduler : public scheduler {
static void worker_loop(worker*);
static void supervisor_loop(job_queue*, scheduled_actor*, size_t);
actor_ptr spawn_impl(scheduled_actor_ptr what);
actor_ptr spawn_as_thread(void_function fun, init_callback cb, bool hidden);
};
} } // namespace cppa::detail
......
......@@ -87,6 +87,8 @@ class event_based_actor : public detail::abstract_scheduled_actor {
scheduled_actor_type impl_type();
static intrusive_ptr<event_based_actor> from(std::function<void()> fun);
protected:
event_based_actor();
......
......@@ -31,6 +31,8 @@
#ifndef CPPA_ACTOR_BEHAVIOR_HPP
#define CPPA_ACTOR_BEHAVIOR_HPP
#include <functional>
#include "cppa/config.hpp"
#include "cppa/local_actor.hpp"
......@@ -46,7 +48,8 @@ enum class resume_result {
enum scheduled_actor_type {
context_switching_impl,
event_based_impl
event_based_impl,
default_event_based_impl // scheduler enqueues a 'RUN' message on startup
};
/**
......
......@@ -35,6 +35,7 @@
#include <memory>
#include <cstdint>
#include <functional>
#include <type_traits>
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
......@@ -42,32 +43,32 @@
#include "cppa/cow_tuple.hpp"
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
class self_type;
class scheduled_actor;
class scheduler_helper;
typedef std::function<void()> void_function;
typedef std::function<void(local_actor*)> init_callback;
namespace detail { class singleton_manager; } // namespace detail
namespace detail {
// forwards self_type as actor_ptr, otherwise equal to std::forward
template<typename T>
struct spawn_fwd_ {
static inline T&& _(T&& arg) { return std::move(arg); }
static inline T& _(T& arg) { return arg; }
static inline const T& _(const T& arg) { return arg; }
};
template<>
struct spawn_fwd_<self_type> {
static inline actor_ptr _(const self_type& s) { return s.get(); }
struct is_self {
typedef typename util::rm_ref<T>::type plain_type;
static constexpr bool value = std::is_same<plain_type,self_type>::value;
};
class singleton_manager;
template<typename T, typename U>
auto fwd(U& arg, typename std::enable_if<!is_self<T>::value>::type* = 0)
-> decltype(std::forward<T>(arg)) {
return std::forward<T>(arg);
}
template<typename T, typename U>
local_actor* fwd(U& arg, typename std::enable_if<is_self<T>::value>::type* = 0){
return arg;
}
} // namespace detail
/**
......@@ -107,6 +108,9 @@ class scheduler {
public:
typedef std::function<void(local_actor*)> init_callback;
typedef std::function<void()> void_function;
const actor_ptr& printer() const;
virtual void enqueue(scheduled_actor*) = 0;
......@@ -157,72 +161,26 @@ class scheduler {
}
/**
* @brief Spawns a new actor that executes <code>fun()</code>
* with the scheduling policy @p hint if possible.
*/
virtual actor_ptr spawn(void_function fun, scheduling_hint hint) = 0;
/**
* @brief Spawns a new actor that executes <code>behavior()</code>
* with the scheduling policy @p hint if possible and calls
* <code>init_cb</code> after the actor is initialized but before
* it starts execution.
*/
virtual actor_ptr spawn(void_function fun,
init_callback init_cb,
scheduling_hint hint) = 0;
/**
* @brief Spawns a new event-based actor.
* @brief Executes @p ptr in this scheduler.
*/
virtual actor_ptr spawn(scheduled_actor* what,
scheduling_hint hint = scheduled) = 0;
virtual actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) = 0;
/**
* @brief Spawns a new event-based actor and calls
* <code>init_cb</code> after the actor is initialized but before
* it starts execution.
* @brief Creates a new actor from @p actor_behavior and executes it
* in this scheduler.
*/
virtual actor_ptr spawn(scheduled_actor* what,
virtual actor_ptr exec(spawn_options opts,
init_callback init_cb,
scheduling_hint hint = scheduled) = 0;
void_function actor_behavior) = 0;
// hide implementation details for documentation
# ifndef CPPA_DOCUMENTATION
template<typename Fun, typename Arg0, typename... Args>
actor_ptr spawn_impl(scheduling_hint hint, Fun&& fun, Arg0&& arg0, Args&&... args) {
return this->spawn(
std::bind(
std::forward<Fun>(fun),
detail::spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0),
detail::spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...),
hint);
}
template<typename Fun>
actor_ptr spawn_impl(scheduling_hint hint, Fun&& fun) {
return this->spawn(std::forward<Fun>(fun), hint);
}
template<typename InitCallback, typename Fun, typename Arg0, typename... Args>
actor_ptr spawn_cb_impl(scheduling_hint hint,
InitCallback&& init_cb,
Fun&& fun, Arg0&& arg0, Args&&... args) {
return this->spawn(
std::bind(
std::forward<Fun>(fun),
detail::spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0),
detail::spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...),
std::forward<InitCallback>(init_cb),
hint);
}
template<typename InitCallback, typename Fun>
actor_ptr spawn_cb_impl(scheduling_hint hint, InitCallback&& init_cb, Fun&& fun) {
return this->spawn(std::forward<Fun>(fun),
std::forward<InitCallback>(init_cb),
hint);
template<typename F, typename T0, 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),
detail::fwd<Ts>(as)...));
}
# endif // CPPA_DOCUMENTATION
......
......@@ -28,41 +28,130 @@
\******************************************************************************/
#ifndef CPPA_SCHEDULING_HINT_HPP
#define CPPA_SCHEDULING_HINT_HPP
#ifndef CPPA_SPAWN_OPTIONS_HPP
#define CPPA_SPAWN_OPTIONS_HPP
namespace cppa {
/**
* @brief Denotes whether a user wants an actor to take part in
* cooperative scheduling or not.
* @ingroup ActorCreation
* @{
*/
enum scheduling_hint {
/**
* @brief Indicates that an actor takes part in cooperative scheduling.
/**
* @brief Stores options passed to the @p spawn function family.
*/
enum class spawn_options : int {
no_flags = 0x00,
link_flag = 0x01,
monitor_flag = 0x02,
detach_flag = 0x04,
hide_flag = 0x08,
blocking_api_flag = 0x10
};
#ifndef CPPA_DOCUMENTATION
namespace {
#endif
/**
* @brief Denotes default settings.
*/
scheduled,
constexpr spawn_options no_spawn_options = spawn_options::no_flags;
/**
* @brief Indicates that an actor should run in its own thread.
/**
* @brief Causes @p spawn to call <tt>self->monitor(...)</tt> immediately
* after the new actor was spawned.
*/
detached,
constexpr spawn_options monitored = spawn_options::monitor_flag;
/**
* @brief Indicates that an actor should run in its own thread,
* but it is ignored by {@link await_others_done()}.
/**
* @brief Causes @p spawn to call <tt>self->link_to(...)</tt> immediately
* after the new actor was spawned.
*/
detached_and_hidden,
constexpr spawn_options linked = spawn_options::link_flag;
/**
* @brief Indicates that an actor takes part in cooperative scheduling,
* but it is ignored by {@link await_others_done()}.
/**
* @brief Causes the new actor to opt out of the cooperative scheduling.
*/
scheduled_and_hidden
constexpr spawn_options detached = spawn_options::detach_flag;
};
/**
* @brief Causes the runtime to ignore the new actor in
* {@link await_all_others_done()}.
*/
constexpr spawn_options hidden = spawn_options::hide_flag;
/**
* @brief Causes the new actor to opt in to the blocking API of libcppa,
* i.e., the actor uses a context-switching or thread-based backend
* instead of the default event-based implementation.
*/
constexpr spawn_options blocking_api = spawn_options::blocking_api_flag;
#ifndef CPPA_DOCUMENTATION
} // namespace <anonymous>
#endif
/**
* @brief Concatenates two {@link spawn_options}.
* @relates spawn_options
*/
constexpr spawn_options operator+(const spawn_options& lhs,
const spawn_options& rhs) {
return static_cast<spawn_options>(static_cast<int>(lhs) | static_cast<int>(rhs));
}
/**
* @brief Checks wheter @p haystack contains @p needle.
* @relates spawn_options
*/
constexpr bool has_spawn_option(spawn_options haystack, spawn_options needle) {
return (static_cast<int>(haystack) & static_cast<int>(needle)) != 0;
}
/**
* @brief Checks wheter the {@link detached} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_detach_flag(spawn_options opts) {
return has_spawn_option(opts, detached);
}
/**
* @brief Checks wheter the {@link hidden} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_hide_flag(spawn_options opts) {
return has_spawn_option(opts, hidden);
}
/**
* @brief Checks wheter the {@link linked} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_link_flag(spawn_options opts) {
return has_spawn_option(opts, linked);
}
/**
* @brief Checks wheter the {@link monitored} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_monitor_flag(spawn_options opts) {
return has_spawn_option(opts, monitored);
}
/**
* @brief Checks wheter the {@link blocking_api} flag is set in @p opts.
* @relates spawn_options
*/
constexpr bool has_blocking_api_flag(spawn_options opts) {
return has_spawn_option(opts, blocking_api);
}
/** @} */
} // namespace cppa
#endif // CPPA_SCHEDULING_HINT_HPP
#endif // CPPA_SPAWN_OPTIONS_HPP
......@@ -17,33 +17,25 @@ using namespace std;
using namespace cppa;
// either taken by a philosopher or available
struct chopstick : sb_actor<chopstick> {
behavior& init_state; // a reference to available
behavior available;
behavior taken_by(const actor_ptr& philos) {
// create a behavior new on-the-fly
return (
on<atom("take"), actor_ptr>() >> [=](actor_ptr other) {
send(other, atom("busy"), this);
void chopstick() {
become(
on(atom("take"), arg_match) >> [=](const actor_ptr& philos) {
// tell philosopher it took this chopstick
send(philos, atom("taken"), self);
// await 'put' message and reject other 'take' messages
become(
keep_behavior, // "enables" unbecome()
on(atom("take"), arg_match) >> [=](const actor_ptr& other) {
send(other, atom("busy"), self);
},
on(atom("put"), philos) >> [=]() {
become(available);
on(atom("put"), philos) >> [=] {
// return to previous behaivor, i.e., await next 'take'
unbecome();
}
);
}
chopstick() : init_state(available) {
available = (
on<atom("take"), actor_ptr>() >> [=](actor_ptr philos) {
send(philos, atom("taken"), this);
become(taken_by(philos));
}
);
}
};
}
/* See: http://www.dalnefre.com/wp/2010/08/dining-philosophers-in-humus/
*
......@@ -95,10 +87,7 @@ struct philosopher : sb_actor<philosopher> {
// wait for second chopstick
behavior waiting_for(const actor_ptr& what) {
return (
on(atom("taken"), what) >> [=]() {
// create message in memory to avoid interleaved
// messages on the terminal
std::ostringstream oss;
on(atom("taken"), what) >> [=] {
aout << name
<< " has picked up chopsticks with IDs "
<< left->id()
......@@ -109,7 +98,7 @@ struct philosopher : sb_actor<philosopher> {
delayed_send(this, seconds(5), atom("think"));
become(eating);
},
on(atom("busy"), what) >> [=]() {
on(atom("busy"), what) >> [=] {
send((what == left) ? right : left, atom("put"), this);
send(this, atom("eat"));
become(thinking);
......@@ -121,7 +110,7 @@ struct philosopher : sb_actor<philosopher> {
: name(n), left(l), right(r) {
// a philosopher that receives {eat} stops thinking and becomes hungry
thinking = (
on(atom("eat")) >> [=]() {
on(atom("eat")) >> [=] {
become(hungry);
send(left, atom("take"), this);
send(right, atom("take"), this);
......@@ -129,43 +118,43 @@ struct philosopher : sb_actor<philosopher> {
);
// wait for the first answer of a chopstick
hungry = (
on(atom("taken"), left) >> [=]() {
on(atom("taken"), left) >> [=] {
become(waiting_for(right));
},
on(atom("taken"), right) >> [=]() {
on(atom("taken"), right) >> [=] {
become(waiting_for(left));
},
on<atom("busy"), actor_ptr>() >> [=]() {
on<atom("busy"), actor_ptr>() >> [=] {
become(denied);
}
);
// philosopher was not able to obtain the first chopstick
denied = (
on<atom("taken"), actor_ptr>() >> [=](actor_ptr& ptr) {
on(atom("taken"), arg_match) >> [=](const actor_ptr& ptr) {
send(ptr, atom("put"), this);
send(this, atom("eat"));
become(thinking);
},
on<atom("busy"), actor_ptr>() >> [=]() {
on<atom("busy"), actor_ptr>() >> [=] {
send(this, atom("eat"));
become(thinking);
}
);
// philosopher obtained both chopstick and eats (for five seconds)
eating = (
on(atom("think")) >> [=]() {
on(atom("think")) >> [=] {
send(left, atom("put"), this);
send(right, atom("put"), this);
delayed_send(this, seconds(5), atom("eat"));
aout << ( name
+ " puts down his chopsticks and starts to think\n");
aout << name
<< " puts down his chopsticks and starts to think\n";
become(thinking);
}
);
// philosophers start to think after receiving {think}
init_state = (
on(atom("think")) >> [=]() {
aout << (name + " starts to think\n");
on(atom("think")) >> [=] {
aout << name << " starts to think\n";
delayed_send(this, seconds(5), atom("eat"));
become(thinking);
}
......@@ -176,10 +165,10 @@ struct philosopher : sb_actor<philosopher> {
int main(int, char**) {
// create five chopsticks
aout << "chopstick ids:";
aout << "chopstick ids are:";
std::vector<actor_ptr> chopsticks;
for (size_t i = 0; i < 5; ++i) {
chopsticks.push_back(spawn<chopstick>());
chopsticks.push_back(spawn(chopstick));
aout << " " << chopsticks.back()->id();
}
aout << endl;
......@@ -194,7 +183,7 @@ int main(int, char**) {
chopsticks[i],
chopsticks[(i+1) % chopsticks.size()]);
}
// tell philosophers to start thinking
// tell all philosophers to start thinking
send(dinner_club, atom("think"));
// real philosophers are never done
await_all_others_done();
......
......@@ -4,35 +4,43 @@
using namespace cppa;
void echo_actor() {
// wait for a message
receive (
void mirror() {
// wait for messages
become (
// invoke this lambda expression if we receive a string
on<std::string>() >> [](const std::string& what) {
on_arg_match >> [](const std::string& what) {
// prints "Hello World!"
std::cout << what << std::endl;
// replies "!dlroW olleH"
reply(std::string(what.rbegin(), what.rend()));
// terminates this actor
self->quit();
}
);
}
int main() {
// create a new actor that invokes the function echo_actor
auto hello_actor = spawn(echo_actor);
// send "Hello World!" to our new actor
// note: libcppa converts string literals to std::string
send(hello_actor, "Hello World!");
// wait for a response and print it
receive (
on<std::string>() >> [](const std::string& what) {
void hello_world(const actor_ptr& buddy) {
// send "Hello World!" to the mirror
send(buddy, "Hello World!");
// wait for messages
become (
on_arg_match >> [](const std::string& what) {
// prints "!dlroW olleH"
std::cout << what << std::endl;
// terminate this actor
self->quit();
}
);
// wait until all other actors we've spawned are done
}
int main() {
// create a new actor that calls 'mirror()'
auto mirror_actor = spawn(mirror);
// create another actor that calls 'hello_world(mirror_actor)'
spawn(hello_world, mirror_actor);
// wait until all other actors we have spawned are done
await_all_others_done();
// done
// run cleanup code before exiting main
shutdown();
return 0;
}
......@@ -13,7 +13,7 @@ using std::endl;
using namespace cppa;
// implementation using the blocking API
void math_fun() {
void blocking_math_fun() {
bool done = false;
do_receive (
// "arg_match" matches the parameter types of given lambda expression
......@@ -23,22 +23,19 @@ void math_fun() {
on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b);
},
on<atom("minus"), int, int>() >> [](int a, int b) {
on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b);
},
on(atom("quit")) >> [&]() {
// note: quit(exit_reason::normal) would terminate the actor
// but is best avoided since it forces stack unwinding
// by throwing an exception
// note: this actor uses the blocking API, hence self->quit()
// would force stack unwinding by throwing an exception
done = true;
}
)
.until(gref(done));
).until(gref(done));
}
// implementation using the event-based API
struct math_actor : event_based_actor {
void init() {
void math_fun() {
// execute this behavior until actor terminates
become (
on(atom("plus"), arg_match) >> [](int a, int b) {
......@@ -49,14 +46,12 @@ struct math_actor : event_based_actor {
},
// the [=] capture copies the 'this' pointer into the lambda
// thus, it has access to all members and member functions
on(atom("quit")) >> [=]() {
// set an empty behavior
// (terminates actor with normal exit reason)
quit();
on(atom("quit")) >> [=] {
// terminate actor with normal exit reason
self->quit();
}
);
}
};
}
// utility function
int fetch_result(actor_ptr& calculator, atom_value operation, int a, int b) {
......@@ -72,9 +67,9 @@ int fetch_result(actor_ptr& calculator, atom_value operation, int a, int b) {
int main() {
// spawn a context-switching actor that invokes math_fun
auto a1 = spawn(math_fun);
auto a1 = spawn<blocking_api>(blocking_math_fun);
// spawn an event-based math actor
auto a2 = spawn<math_actor>();
auto a2 = spawn(math_fun);
// do some testing on both implementations
assert((fetch_result(a1, atom("plus"), 1, 2) == 3));
assert((fetch_result(a2, atom("plus"), 1, 2) == 3));
......
......@@ -31,11 +31,43 @@
#include <iostream>
#include "cppa/to_string.hpp"
#include "cppa/on.hpp"
#include "cppa/self.hpp"
#include "cppa/event_based_actor.hpp"
namespace cppa {
class default_scheduled_actor : public event_based_actor {
public:
typedef std::function<void()> fun_type;
default_scheduled_actor(fun_type&& fun) : m_fun(std::move(fun)) { }
void init() {
become (
on(atom("RUN")) >> [=] {
unbecome();
m_fun();
}
);
}
scheduled_actor_type impl_type() {
return default_event_based_impl;
}
private:
fun_type m_fun;
};
intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> fun) {
return detail::memory::create<default_scheduled_actor>(std::move(fun));
}
event_based_actor::event_based_actor() : super(super::blocked) { }
void event_based_actor::dequeue(behavior&) {
......
......@@ -108,7 +108,7 @@ struct group_nameserver : event_based_actor {
};
void publish_local_groups_at(std::uint16_t port, const char* addr) {
auto gn = spawn_hidden<group_nameserver>();
auto gn = spawn<group_nameserver, hidden>();
try {
publish(gn, port, addr);
}
......
......@@ -189,7 +189,7 @@ class local_group_proxy : public local_group {
CPPA_REQUIRE(remote_broker != nullptr);
CPPA_REQUIRE(remote_broker->is_proxy());
m_broker = move(remote_broker);
m_proxy_broker = spawn_hidden<proxy_broker>(this);
m_proxy_broker = spawn<proxy_broker, hidden>(this);
}
group::subscription subscribe(const channel_ptr& who) {
......@@ -426,7 +426,7 @@ class remote_group_module : public group::module {
auto sm = make_counted<shared_map>();
group::module_ptr _this = this;
m_map = sm;
auto worker = spawn<detached_and_hidden>([_this, sm] {
auto worker = spawn<hidden>([_this, sm] {
typedef map<string, pair<actor_ptr, vector<pair<string, remote_group_ptr>>>>
peer_map;
peer_map peers;
......@@ -529,7 +529,7 @@ local_group::local_group(bool spawn_local_broker,
local_group_module* mod,
string id)
: group(mod, move(id)) {
if (spawn_local_broker) m_broker = spawn_hidden<local_broker>(this);
if (spawn_local_broker) m_broker = spawn<local_broker, hidden>(this);
}
void local_group::serialize(serializer* sink) {
......
......@@ -28,8 +28,12 @@
\******************************************************************************/
#include "cppa/on.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/detail/memory.hpp"
namespace cppa {
......
......@@ -41,6 +41,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/context_switching_actor.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -108,7 +109,7 @@ class delayed_msg {
private:
either<async_send_fun, sync_reply_fun> fun;
either<async_send_fun,sync_reply_fun> fun;
channel_ptr ptr_a;
actor_ptr ptr_b;
......@@ -347,5 +348,4 @@ const actor_ptr& scheduler::printer() const {
return m_helper->m_printer;
}
} // namespace cppa
......@@ -34,6 +34,7 @@
#include <cstddef>
#include <iostream>
#include "cppa/on.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
......@@ -46,13 +47,6 @@ using std::endl;
namespace cppa { namespace detail {
namespace {
typedef std::unique_lock<std::mutex> guard_type;
typedef intrusive::single_reader_queue<thread_pool_scheduler::worker> worker_queue;
} // namespace <anonmyous>
struct thread_pool_scheduler::worker {
typedef scheduled_actor* job_ptr;
......@@ -71,38 +65,28 @@ struct thread_pool_scheduler::worker {
worker& operator=(const worker&) = delete;
job_ptr aggressive_polling() {
job_ptr result = nullptr;
bool aggressive(job_ptr& result) {
for (int i = 0; i < 100; ++i) {
result = m_job_queue->try_pop();
if (result) {
return result;
}
if (result) return true;
std::this_thread::yield();
}
return result;
return false;
}
job_ptr less_aggressive_polling() {
job_ptr result = nullptr;
bool moderate(job_ptr& result) {
for (int i = 0; i < 550; ++i) {
result = m_job_queue->try_pop();
if (result) {
return result;
}
//std::this_thread::sleep_for(std::chrono::milliseconds(1));
if (result) return true;
std::this_thread::sleep_for(std::chrono::microseconds(50));
}
return result;
return false;
}
job_ptr relaxed_polling() {
job_ptr result = nullptr;
bool relaxed(job_ptr& result) {
for (;;) {
result = m_job_queue->try_pop();
if (result) {
return result;
}
if (result) return true;
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
......@@ -110,37 +94,25 @@ struct thread_pool_scheduler::worker {
void operator()() {
util::fiber fself;
job_ptr job = nullptr;
actor_ptr next_job;
actor_ptr next;
for (;;) {
job = aggressive_polling();
if (job == nullptr) {
job = less_aggressive_polling();
if (job == nullptr) {
job = relaxed_polling();
}
}
aggressive(job) || moderate(job) || relaxed(job);
if (job == m_dummy) {
// dummy of doom received ...
m_job_queue->push_back(job); // kill the next guy
return; // and say goodbye
}
else {
do {
next_job.reset();
if (job->resume(&fself, next_job) == resume_result::actor_done) {
next.reset();
if (job->resume(&fself, next) == resume_result::actor_done) {
bool hidden = job->is_hidden();
job->deref();
//std::atomic_thread_fence(std::memory_order_seq_cst);
if (!hidden) dec_actor_count();
}
if (next_job) {
job = static_cast<job_ptr>(next_job.get());
//get_scheduler()->printer()->enqueue(job, make_any_tuple("fast-forwarded execution (chained actor)\n"));
}
if (next) job = static_cast<job_ptr>(next.get());
else job = nullptr;
}
while (job); // loops until next_job was nullptr
}
while (job); // loops until next == nullptr
}
}
......@@ -151,14 +123,13 @@ void thread_pool_scheduler::worker_loop(thread_pool_scheduler::worker* w) {
}
thread_pool_scheduler::thread_pool_scheduler() {
m_num_threads = std::max<size_t>(std::thread::hardware_concurrency() * 2, 4);
m_num_threads = std::max<size_t>(std::thread::hardware_concurrency()*2, 4);
}
thread_pool_scheduler::thread_pool_scheduler(size_t num_worker_threads) {
m_num_threads = num_worker_threads;
}
void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
scheduled_actor* dummy,
size_t num_threads) {
......@@ -201,104 +172,71 @@ void thread_pool_scheduler::enqueue(scheduled_actor* what) {
m_queue.push_back(what);
}
actor_ptr thread_pool_scheduler::spawn_as_thread(void_function fun,
init_callback cb,
bool hidden) {
if (!hidden) inc_actor_count();
thread_mapped_actor_ptr ptr{detail::memory::create<thread_mapped_actor>(std::move(fun))};
ptr->init();
ptr->initialized(true);
cb(ptr.get());
std::thread([hidden, ptr]() {
scoped_self_setter sss{ptr.get()};
try {
ptr->run();
ptr->on_exit();
}
template<typename F>
void exec_as_thread(bool is_hidden, local_actor_ptr p, F f) {
if (!is_hidden) { inc_actor_count(); }
std::thread([=] {
scoped_self_setter sss(p.get());
try { f(); }
catch (...) { }
if (!is_hidden) {
std::atomic_thread_fence(std::memory_order_seq_cst);
if (!hidden) dec_actor_count();
}).detach();
return ptr;
}
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) {
if (what->has_behavior()) {
if (!what->is_hidden()) { inc_actor_count(); }
what->ref();
// event-based actors are not pushed to the job queue on startup
if (what->impl_type() == context_switching_impl) {
m_queue.push_back(what.get());
dec_actor_count();
}
}
else {
what->on_exit();
}
return std::move(what);
}).detach();
}
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw,
scheduling_hint hint) {
scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
return spawn_impl(std::move(ptr));
actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
CPPA_REQUIRE(p != nullptr);
bool is_hidden = has_hide_flag(os);
if (has_detach_flag(os)) {
exec_as_thread(is_hidden, p, [p] {
p->exec_behavior_stack();
p->on_exit();
});
return std::move(p);
}
p->attach_to_scheduler(this, is_hidden);
if (p->has_behavior()) {
if (!is_hidden) { inc_actor_count(); }
p->ref(); // implicit reference that's released if actor dies
switch (p->impl_type()) {
case default_event_based_impl: {
p->enqueue(nullptr, make_any_tuple(atom("RUN")));
break;
}
case context_switching_impl: {
m_queue.push_back(p.get());
break;
}
default: break; // nothing to do
}
}
else p->on_exit();
return std::move(p);
}
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw,
actor_ptr thread_pool_scheduler::exec(spawn_options os,
init_callback cb,
scheduling_hint hint) {
scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
cb(ptr.get());
return spawn_impl(std::move(ptr));
}
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr thread_pool_scheduler::spawn(void_function fun, scheduling_hint hint) {
if (hint == scheduled || hint == scheduled_and_hidden) {
scheduled_actor_ptr ptr{detail::memory::create<context_switching_actor>(std::move(fun))};
ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
return spawn_impl(std::move(ptr));
}
else {
return spawn_as_thread(std::move(fun),
[](local_actor*) { },
hint == detached_and_hidden);
}
}
actor_ptr thread_pool_scheduler::spawn(void_function fun,
init_callback init_cb,
scheduling_hint hint) {
if (hint == scheduled || hint == scheduled_and_hidden) {
scheduled_actor_ptr ptr{detail::memory::create<context_switching_actor>(std::move(fun))};
ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
init_cb(ptr.get());
return spawn_impl(std::move(ptr));
}
else {
return spawn_as_thread(std::move(fun),
std::move(init_cb),
hint == detached_and_hidden);
}
}
#else // CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr thread_pool_scheduler::spawn(void_function what, scheduling_hint sh) {
return spawn_as_thread(std::move(what),
[](local_actor*) { },
sh == detached_and_hidden);
}
actor_ptr thread_pool_scheduler::spawn(void_function what,
init_callback init_cb,
scheduling_hint sh) {
return spawn_as_thread(std::move(what),
std::move(init_cb),
sh == detached_and_hidden);
void_function f) {
if (has_blocking_api_flag(os)) {
# ifndef CPPA_DISABLE_CONTEXT_SWITCHING
if (!has_detach_flag(os)) {
return exec(os,
memory::create<context_switching_actor>(std::move(f)));
}
# endif
thread_mapped_actor_ptr p;
p.reset(memory::create<thread_mapped_actor>(std::move(f)));
exec_as_thread(has_hide_flag(os), p, [p] {
p->run();
p->on_exit();
});
return std::move(p);
}
auto p = event_based_actor::from(std::move(f));
if (cb) cb(p.get());
return exec(os, p);
}
#endif
} } // namespace cppa::detail
......@@ -157,7 +157,7 @@ int client_part(const vector<string_pair>& args) {
send(server, atom("SpawnPing"));
receive (
on(atom("PingPtr"), arg_match) >> [](actor_ptr ping_actor) {
spawn<detached>(pong, ping_actor);
spawn<detached + blocking_api>(pong, ping_actor);
}
);
await_all_others_done();
......
......@@ -121,10 +121,10 @@ class testee_actor {
void wait4string() {
bool string_received = false;
do_receive (
on<string>() >> [&]() {
on<string>() >> [&] {
string_received = true;
},
on<atom("get_state")>() >> [&]() {
on<atom("get_state")>() >> [&] {
reply("wait4string");
}
)
......@@ -134,10 +134,10 @@ class testee_actor {
void wait4float() {
bool float_received = false;
do_receive (
on<float>() >> [&]() {
on<float>() >> [&] {
float_received = true;
},
on<atom("get_state")>() >> [&]() {
on<atom("get_state")>() >> [&] {
reply("wait4float");
}
)
......@@ -149,10 +149,10 @@ class testee_actor {
void operator()() {
receive_loop (
on<int>() >> [&]() {
on<int>() >> [&] {
wait4float();
},
on<atom("get_state")>() >> [&]() {
on<atom("get_state")>() >> [&] {
reply("wait4int");
}
);
......@@ -162,33 +162,19 @@ class testee_actor {
// receives one timeout and quits
void testee1() {
receive (
after(chrono::milliseconds(10)) >> []() { }
);
become(after(chrono::milliseconds(10)) >> [] { unbecome(); });
}
void testee2(actor_ptr other) {
self->link_to(other);
send(other, uint32_t(1));
receive_loop (
become (
on<uint32_t>() >> [](uint32_t sleep_time) {
// "sleep" for sleep_time milliseconds
receive(after(chrono::milliseconds(sleep_time)) >> []() {});
//reply(sleep_time * 2);
}
become (
keep_behavior,
after(chrono::milliseconds(sleep_time)) >> [] { unbecome(); }
);
}
void testee3(actor_ptr parent) {
// test a delayed_send / delayed_reply based loop
delayed_send(self, chrono::milliseconds(50), atom("Poll"));
int polls = 0;
receive_for(polls, 5) (
on(atom("Poll")) >> [&]() {
if (polls < 4) {
delayed_reply(chrono::milliseconds(50), atom("Poll"));
}
send(parent, atom("Push"), polls);
}
);
}
......@@ -277,9 +263,10 @@ class fixed_stack : public sb_actor<fixed_stack> {
};
void echo_actor() {
receive (
others() >> []() {
become (
others() >> [] {
reply_tuple(self->last_dequeued());
self->quit(exit_reason::normal);
}
);
}
......@@ -423,7 +410,7 @@ int main() {
await_all_others_done();
CPPA_CHECKPOINT();
auto sync_testee1 = spawn([]() {
auto sync_testee1 = spawn<blocking_api>([] {
receive (
on(atom("get")) >> []() {
reply(42, 2);
......@@ -629,27 +616,23 @@ int main() {
}).spawn();
await_all_others_done();
auto res1 = behavior_test<testee_actor>(spawn(testee_actor{}));
auto res1 = behavior_test<testee_actor>(spawn<blocking_api>(testee_actor{}));
CPPA_CHECK_EQUAL("wait4int", res1);
CPPA_CHECK_EQUAL(behavior_test<event_testee>(spawn<event_testee>()), "wait4int");
// create 20,000 actors linked to one single actor
// and kill them all through killing the link
auto twenty_thousand = spawn([]() {
auto twenty_thousand = spawn([] {
for (int i = 0; i < 20000; ++i) {
spawn_link<event_testee>();
}
receive_loop (
others() >> []() {
cout << "wtf? => " << to_string(self->last_dequeued()) << endl;
spawn<event_testee, linked>();
}
);
become(others() >> CPPA_UNEXPECTED_MSG_CB());
});
quit_actor(twenty_thousand, exit_reason::user_defined);
await_all_others_done();
self->trap_exit(true);
auto ping_actor = spawn_monitor(ping, 10);
auto pong_actor = spawn_monitor(pong, ping_actor);
auto ping_actor = spawn<monitored + blocking_api>(ping, 10);
auto pong_actor = spawn<monitored + blocking_api>(pong, ping_actor);
self->link_to(pong_actor);
int i = 0;
int flags = 0;
......
......@@ -171,9 +171,9 @@ int main() {
self->on_sync_failure([] {
CPPA_ERROR("received: " << to_string(self->last_dequeued()));
});
spawn_monitor([&] {
spawn<monitored + blocking_api>([&] {
int invocations = 0;
auto foi = spawn_link<float_or_int>();
auto foi = spawn<float_or_int, linked>();
send(foi, atom("i"));
receive(on_arg_match >> [](int i) { CPPA_CHECK_EQUAL(i, 0); });
self->on_sync_failure([] {
......@@ -228,11 +228,11 @@ int main() {
}
);
};
send(spawn_monitor<A>(self), atom("go"), spawn<B>(spawn<C>()));
send(spawn<A, monitored>(self), atom("go"), spawn<B>(spawn<C>()));
await_success_message();
CPPA_CHECKPOINT();
await_all_others_done();
send(spawn_monitor<A>(self), atom("go"), spawn<D>(spawn<C>()));
send(spawn<A, monitored>(self), atom("go"), spawn<D>(spawn<C>()));
await_success_message();
CPPA_CHECKPOINT();
await_all_others_done();
......@@ -276,10 +276,10 @@ int main() {
CPPA_CHECKPOINT();
// test use case 3
spawn_monitor([] { // client
auto s = spawn_link<server>(); // server
auto w = spawn_link([] { // worker
receive_loop(on(atom("request")) >> []{ reply(atom("response")); });
spawn<monitored>([] { // client
auto s = spawn<server, linked>(); // server
auto w = spawn<linked>([] { // worker
become(on(atom("request")) >> []{ reply(atom("response")); });
});
// first 'idle', then 'request'
send_as(w, s, atom("idle"));
......
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