Commit 6f113f7c authored by neverlord's avatar neverlord

refactoring

parent dfea897e
...@@ -60,15 +60,15 @@ ...@@ -60,15 +60,15 @@
<valuemap type="QVariantMap"> <valuemap type="QVariantMap">
<value key="ProjectExplorer.BuildConfiguration.DisplayName" type="QString">all</value> <value key="ProjectExplorer.BuildConfiguration.DisplayName" type="QString">all</value>
<valuelist key="abstractProcess.Environment" type="QVariantList"> <valuelist key="abstractProcess.Environment" type="QVariantList">
<value type="QString">Apple_PubSub_Socket_Render=/tmp/launch-Ii2KqP/Render</value> <value type="QString">Apple_PubSub_Socket_Render=/tmp/launch-qKmNvj/Render</value>
<value type="QString">COMMAND_MODE=unix2003</value> <value type="QString">COMMAND_MODE=unix2003</value>
<value type="QString">DISPLAY=/tmp/launch-6FAyZP/org.x:0</value> <value type="QString">DISPLAY=/tmp/launch-bKBD8m/org.x:0</value>
<value type="QString">HOME=/Users/neverlord</value> <value type="QString">HOME=/Users/neverlord</value>
<value type="QString">LOGNAME=neverlord</value> <value type="QString">LOGNAME=neverlord</value>
<value type="QString">PATH=/usr/bin:/bin:/usr/sbin:/sbin</value> <value type="QString">PATH=/usr/bin:/bin:/usr/sbin:/sbin</value>
<value type="QString">SHELL=/bin/bash</value> <value type="QString">SHELL=/bin/bash</value>
<value type="QString">SSH_AUTH_SOCK=/tmp/launch-vifjfO/Listeners</value> <value type="QString">SSH_AUTH_SOCK=/tmp/launch-voyypI/Listeners</value>
<value type="QString">TMPDIR=/var/folders/1p/1p6iuPgbH7GoDkrjrT20tU+++TI/-Tmp-/</value> <value type="QString">TMPDIR=/var/folders/SE/SEReyCW0H-yDPCnNCe8mN++++TI/-Tmp-/</value>
<value type="QString">USER=neverlord</value> <value type="QString">USER=neverlord</value>
<value type="QString">__CF_USER_TEXT_ENCODING=0x1F5:0:3</value> <value type="QString">__CF_USER_TEXT_ENCODING=0x1F5:0:3</value>
</valuelist> </valuelist>
......
cppa/ref_counted.hpp cppa/ref_counted.hpp
cppa/tuple.hpp cppa/tuple.hpp
unit_testing/main.cpp unit_testing/main.cpp
src/ref_counted.cpp
cppa/util/void_type.hpp cppa/util/void_type.hpp
cppa/util/type_list.hpp cppa/util/type_list.hpp
cppa/util/type_at.hpp cppa/util/type_at.hpp
...@@ -82,17 +81,18 @@ cppa/detail/actor_public.hpp ...@@ -82,17 +81,18 @@ cppa/detail/actor_public.hpp
cppa/detail/comparable.hpp cppa/detail/comparable.hpp
cppa/util/single_reader_queue.hpp cppa/util/single_reader_queue.hpp
cppa/util/singly_linked_list.hpp cppa/util/singly_linked_list.hpp
cppa/detail/spawn_impl.hpp
cppa/detail/actor_private.hpp cppa/detail/actor_private.hpp
cppa/reply.hpp
unit_testing/test__local_group.cpp unit_testing/test__local_group.cpp
unit_testing/hash_of.cpp unit_testing/hash_of.cpp
cppa/spawn.hpp
unit_testing/hash_of.hpp unit_testing/hash_of.hpp
cppa/detail/channel.hpp cppa/detail/channel.hpp
cppa/message_receiver.hpp
src/message_receiver.cpp
src/actor.cpp
src/actor_private.cpp
cppa/util/eval_first_n.hpp
cppa/util/first_n.hpp cppa/util/first_n.hpp
cppa/util/eval_first_n.hpp
cppa/context.hpp
cppa/channel.hpp
src/channel.cpp
cppa/actor_behavior.hpp
cppa/message_queue.hpp
cppa/cppa.hpp
cppa/scheduling_hint.hpp
src/actor_behavior.cpp
#ifndef ACTOR_HPP #ifndef ACTOR_HPP
#define ACTOR_HPP #define ACTOR_HPP
#include "cppa/message_receiver.hpp" #include "cppa/channel.hpp"
#include "cppa/detail/actor_public.hpp"
#include "cppa/detail/actor_private.hpp"
namespace cppa { namespace cppa {
class actor : public message_receiver class actor : public channel
{ {
typedef message_receiver super;
public: public:
typedef cppa::intrusive_ptr<detail::actor_private> ptr_type; virtual void link(const intrusive_ptr<actor>& other) = 0;
actor() = default;
~actor();
inline actor(detail::actor_private* ptr) : super(ptr) { }
inline actor(const ptr_type& ptr) : super(ptr) { }
inline actor(ptr_type&& ptr) : super(ptr) { }
inline actor(const actor& other) : super(other) { }
inline actor(actor&& other) : super(other) { }
actor& operator=(const actor&);
actor& operator=(actor&& other);
}; };
typedef intrusive_ptr<actor> actor_ptr;
} // namespace cppa } // namespace cppa
#endif // ACTOR_HPP #endif // ACTOR_HPP
#ifndef ACTOR_BEHAVIOR_HPP
#define ACTOR_BEHAVIOR_HPP
namespace cppa {
class actor_behavior
{
public:
virtual void act() = 0;
virtual void on_exit();
};
} // namespace cppa
#endif // ACTOR_BEHAVIOR_HPP
#ifndef CHANNEL_HPP
#define CHANNEL_HPP
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa {
// forward declaration
class message;
class channel : public ref_counted
{
public:
virtual ~channel();
virtual void enqueue(const message&) = 0;
};
typedef intrusive_ptr<channel> channel_ptr;
} // namespace cppa
#endif // CHANNEL_HPP
#ifndef CONTEXT_HPP
#define CONTEXT_HPP
#include "cppa/actor.hpp"
#include "cppa/message_queue.hpp"
namespace cppa {
class context : public actor
{
public:
virtual message_queue& mailbox() = 0;
virtual void unlink(const actor_ptr& other) = 0;
};
} // namespace cppa
#endif // CONTEXT_HPP
#ifndef CPPA_HPP
#define CPPA_HPP
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/context.hpp"
#include "cppa/message.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/detail/scheduler.hpp"
namespace cppa {
template<typename F>
actor_ptr spawn(F fun, scheduling_hint hint = scheduled)
{
struct fun_behavior : actor_behavior
{
F m_fun;
fun_behavior(const F& fun_arg) : m_fun(fun_arg) { }
virtual void act()
{
m_fun();
}
};
return detail::scheduler::spawn(new fun_behavior(fun), hint);
}
inline actor_ptr spawn(actor_behavior* ab, scheduling_hint hint = scheduled)
{
return detail::scheduler::spawn(ab, hint);
}
inline context* self()
{
return detail::scheduler::get_context();
}
inline const message& receive()
{
return self()->mailbox().dequeue();
}
inline void receive(invoke_rules& rules)
{
self()->mailbox().dequeue(rules);
}
inline void receive(invoke_rules&& rules)
{
self()->mailbox().dequeue(rules);
}
inline bool try_receive(message& msg)
{
return self()->mailbox().try_dequeue(msg);
}
inline bool try_receive(invoke_rules& rules)
{
return self()->mailbox().try_dequeue(rules);
}
inline const message& last_received()
{
return self()->mailbox().last_dequeued();
}
template<typename Arg0, typename... Args>
void send(channel_ptr whom, const Arg0& arg0, const Args&... args)
{
if (whom) whom->enqueue(message(self(), whom, arg0, args...));
}
template<typename Arg0, typename... Args>
void reply(const Arg0& arg0, const Args&... args)
{
context* sptr = self();
actor_ptr whom = sptr->mailbox().last_dequeued().sender();
if (whom) whom->enqueue(message(sptr, whom, arg0, args...));
}
inline void await_all_actors_done()
{
detail::scheduler::await_all_done();
}
} // namespace cppa
#endif // CPPA_HPP
#ifndef SCHEDULER_HPP #ifndef SCHEDULER_HPP
#define SCHEDULER_HPP #define SCHEDULER_HPP
#include "cppa/message.hpp" #include "cppa/actor.hpp"
#include "cppa/detail/spawn_impl.hpp" #include "cppa/context.hpp"
#include "cppa/detail/actor_public.hpp" #include "cppa/actor_behavior.hpp"
#include "cppa/detail/actor_private.hpp" #include "cppa/scheduling_hint.hpp"
namespace cppa { namespace detail {
struct scheduler
{
static actor_ptr spawn(actor_behavior*, scheduling_hint);
static context* get_context();
static void await_all_done();
};
} } // namespace cppa::detail
#endif // SCHEDULER_HPP #endif // SCHEDULER_HPP
#ifndef SPAWN_IMPL_HPP
#define SPAWN_IMPL_HPP
#include "cppa/actor.hpp"
#include "cppa/detail/scheduler.hpp"
namespace cppa { namespace detail {
struct behavior
{
virtual void act() = 0;
virtual void on_exit() = 0;
};
actor spawn_impl(behavior*);
} } // namespace cppa::detail
#endif // SPAWN_IMPL_HPP
...@@ -25,7 +25,7 @@ template<std::size_t N, typename... Types> ...@@ -25,7 +25,7 @@ template<std::size_t N, typename... Types>
const typename util::type_at<N, util::type_list<Types...>>::type& const typename util::type_at<N, util::type_list<Types...>>::type&
get(const tuple_view<Types...>& t); get(const tuple_view<Types...>& t);
// forward declarations of get(...) // forward declarations of get_ref(...)
template<std::size_t N, typename... Types> template<std::size_t N, typename... Types>
typename util::type_at<N, util::type_list<Types...>>::type& typename util::type_at<N, util::type_list<Types...>>::type&
get_ref(tuple<Types...>& t); get_ref(tuple<Types...>& t);
......
...@@ -3,9 +3,9 @@ ...@@ -3,9 +3,9 @@
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/tuple.hpp" #include "cppa/tuple.hpp"
#include "cppa/channel.hpp"
#include "cppa/untyped_tuple.hpp" #include "cppa/untyped_tuple.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
#include "cppa/message_receiver.hpp"
#include "cppa/detail/channel.hpp" #include "cppa/detail/channel.hpp"
...@@ -18,17 +18,14 @@ class message ...@@ -18,17 +18,14 @@ class message
struct content : ref_counted struct content : ref_counted
{ {
const actor sender; const actor_ptr sender;
const message_receiver receiver; const channel_ptr receiver;
const untyped_tuple data; const untyped_tuple data;
content(const actor& s, const message_receiver& r, content(const actor_ptr& s, const channel_ptr& r,
const untyped_tuple& ut) const untyped_tuple& ut)
: ref_counted(), sender(s), receiver(r), data(ut) : ref_counted(), sender(s), receiver(r), data(ut)
{ {
} }
~content()
{
}
}; };
private: private:
...@@ -38,22 +35,22 @@ class message ...@@ -38,22 +35,22 @@ class message
public: public:
template<typename... Args> template<typename... Args>
message(const actor& from, const message_receiver& to, const Args&... args) message(const actor_ptr& from, const channel_ptr& to, const Args&... args)
: m_content(new content(from, to, tuple<Args...>(args...))) { } : m_content(new content(from, to, tuple<Args...>(args...))) { }
message(const actor& from, const message_receiver& to, const untyped_tuple& ut) message(const actor_ptr& from, const channel_ptr& to, const untyped_tuple& ut)
: m_content(new content(from, to, ut)) { } : m_content(new content(from, to, ut)) { }
// message(const actor& from, const message_receiver& to, untyped_tuple&& ut) // message(const actor_ptr& from, const channel_ptr& to, untyped_tuple&& ut)
// : m_content(new content(from, to, std::move(ut))) { } // : m_content(new content(from, to, std::move(ut))) { }
message() : m_content(new content(0, 0, tuple<int>(0))) { } message() : m_content(new content(0, 0, tuple<int>(0))) { }
const actor& sender() const const actor_ptr& sender() const
{ {
return m_content->sender; return m_content->sender;
} }
const message_receiver& receiver() const const channel_ptr& receiver() const
{ {
return m_content->receiver; return m_content->receiver;
} }
...@@ -67,6 +64,4 @@ class message ...@@ -67,6 +64,4 @@ class message
} // namespace cppa } // namespace cppa
#include "cppa/reply.hpp"
#endif // MESSAGE_HPP #endif // MESSAGE_HPP
#ifndef MESSAGE_QUEUE_HPP
#define MESSAGE_QUEUE_HPP
#include "cppa/channel.hpp"
// forward declaration
namespace cppa { class invoke_rules; }
namespace cppa {
class message_queue : public channel
{
public:
virtual const message& dequeue() = 0;
virtual void dequeue(invoke_rules&) = 0;
virtual bool try_dequeue(message&) = 0;
virtual bool try_dequeue(invoke_rules&) = 0;
virtual const message& last_dequeued() = 0;
};
} // namespace cppa
#endif // MESSAGE_QUEUE_HPP
#ifndef MESSAGE_RECEIVER_HPP
#define MESSAGE_RECEIVER_HPP
#include "cppa/tuple.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/channel.hpp"
#include "cppa/detail/comparable.hpp"
#include "cppa/detail/actor_private.hpp"
namespace cppa {
class message_receiver :
detail::comparable<message_receiver, intrusive_ptr<detail::channel>>,
detail::comparable<message_receiver, message_receiver>
{
protected:
intrusive_ptr<detail::channel> m_channel;
public:
inline message_receiver(detail::channel* ch_ptr) : m_channel(ch_ptr) { }
inline message_receiver(const intrusive_ptr<detail::channel>& ch_ptr)
: m_channel(ch_ptr)
{
}
inline message_receiver(intrusive_ptr<detail::channel>&& ch_ptr)
: m_channel(std::move(ch_ptr))
{
}
virtual ~message_receiver();
message_receiver() = default;
message_receiver(message_receiver&&) = default;
message_receiver(const message_receiver&) = default;
message_receiver& operator=(const message_receiver&) = default;
message_receiver& operator=(detail::channel*);
message_receiver& operator=(const intrusive_ptr<detail::channel>&);
message_receiver& operator=(intrusive_ptr<detail::channel>&&);
void enqueue_msg(const message& msg)
{
m_channel->enqueue_msg(msg);
}
template<typename... Args>
void send(const Args&... args)
{
this_actor()->send(m_channel.get(), tuple<Args...>(args...));
}
inline bool equal_to(const intrusive_ptr<detail::channel>& ch_ptr) const
{
return m_channel == ch_ptr;
}
inline bool equal_to(const message_receiver& other) const
{
return m_channel == other.m_channel;
}
};
} // namespace cppa
#endif // MESSAGE_RECEIVER_HPP
#ifndef REPLY_HPP
#define REPLY_HPP
#include "cppa/message.hpp"
namespace cppa {
template<typename... Args>
void reply(const Args&... args)
{
auto whom = this_actor()->last_dequeued().sender();
whom.send(args...);
}
} // namespace cppa
#endif // REPLY_HPP
#ifndef SCHEDULING_HINT_HPP
#define SCHEDULING_HINT_HPP
namespace cppa {
enum scheduling_hint
{
scheduled,
detached,
as_task
};
} // namespace cppa
#endif // SCHEDULING_HINT_HPP
#ifndef SPAWN_HPP
#define SPAWN_HPP
#include <functional>
#include "cppa/actor.hpp"
#include "cppa/detail/scheduler.hpp"
using std::cout;
using std::endl;
using namespace cppa;
template<typename F>
actor spawn(F act_fun)
{
struct bhv : cppa::detail::behavior
{
std::function<void ()> m_act;
bhv(const F& invokable) : m_act(invokable) { }
virtual void act()
{
m_act();
}
virtual void on_exit()
{
}
};
return cppa::detail::spawn_impl(new bhv(act_fun));
}
#endif // SPAWN_HPP
...@@ -2,6 +2,7 @@ ...@@ -2,6 +2,7 @@
#define UTIL_HPP #define UTIL_HPP
#include "cppa/util/a_matches_b.hpp" #include "cppa/util/a_matches_b.hpp"
#include "cppa/util/abstract_type_list.hpp"
#include "cppa/util/callable_trait.hpp" #include "cppa/util/callable_trait.hpp"
#include "cppa/util/compare_tuples.hpp" #include "cppa/util/compare_tuples.hpp"
#include "cppa/util/concat_type_lists.hpp" #include "cppa/util/concat_type_lists.hpp"
...@@ -23,6 +24,8 @@ ...@@ -23,6 +24,8 @@
#include "cppa/util/remove_const_reference.hpp" #include "cppa/util/remove_const_reference.hpp"
#include "cppa/util/replace_type.hpp" #include "cppa/util/replace_type.hpp"
#include "cppa/util/reverse_type_list.hpp" #include "cppa/util/reverse_type_list.hpp"
#include "cppa/util/single_reader_queue.hpp"
#include "cppa/util/singly_linked_list.hpp"
#include "cppa/util/sink.hpp" #include "cppa/util/sink.hpp"
#include "cppa/util/source.hpp" #include "cppa/util/source.hpp"
#include "cppa/util/type_at.hpp" #include "cppa/util/type_at.hpp"
......
#ifndef EVAL_FIRST_N_HPP
#define EVAL_FIRST_N_HPP
#include <type_traits>
#include "cppa/util/first_n.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/util/eval_type_lists.hpp"
namespace cppa { namespace util {
template <std::size_t N,
class ListA, class ListB,
template <typename, typename> class What>
struct eval_first_n;
template <std::size_t N, typename... TypesA, typename... TypesB,
template <typename, typename> class What>
struct eval_first_n<N, type_list<TypesA...>, type_list<TypesB...>, What>
{
typedef type_list<TypesA...> first_list;
typedef type_list<TypesB...> second_list;
typedef typename first_n<N, first_list>::type slist_a;
typedef typename first_n<N, second_list>::type slist_b;
static const bool value = eval_type_lists<slist_a, slist_b, What>::value;
};
} } // namespace cppa::util
#endif // EVAL_FIRST_N_HPP
#ifndef FIRST_N_HPP
#define FIRST_N_HPP
#include <cstddef>
#include <type_traits>
#include "cppa/util/type_at.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/util/concat_type_lists.hpp"
namespace cppa { namespace detail {
template<bool Stmt, typename IfType, typename ElseType>
struct if_t
{
typedef ElseType type;
};
template<typename IfType, typename ElseType>
struct if_t<true, IfType, ElseType>
{
typedef IfType type;
};
template<std::size_t N, typename List>
struct n_
{
typedef typename util::type_at<N - 1, List>::type head_type;
typedef typename if_t<std::is_same<util::void_type, head_type>::value,
util::type_list<>,
util::type_list<head_type>>::type
head_list;
typedef typename n_<N - 1, List>::type tail_list;
typedef typename util::concat_type_lists<tail_list, head_list>::type type;
};
template<typename List>
struct n_<0, List>
{
typedef util::type_list<> type;
};
} } // namespace cppa::detail
namespace cppa { namespace util {
template<std::size_t N, typename List>
struct first_n;
template<std::size_t N, typename... ListTypes>
struct first_n<N, util::type_list<ListTypes...>>
{
typedef typename detail::n_<N, util::type_list<ListTypes...>>::type type;
};
} } // namespace cppa::util
#endif // FIRST_N_HPP
...@@ -59,7 +59,7 @@ class single_reader_queue ...@@ -59,7 +59,7 @@ class single_reader_queue
* @warning call only from the reader (owner) * @warning call only from the reader (owner)
*/ */
template<typename List> template<typename List>
void prepend(List&& list) void push_front(List&& list)
{ {
if (!list.empty()) if (!list.empty())
{ {
...@@ -103,6 +103,11 @@ class single_reader_queue ...@@ -103,6 +103,11 @@ class single_reader_queue
} }
} }
bool empty()
{
return !m_head && !(m_tail.load());
}
single_reader_queue() : m_tail(0), m_head(0) { } single_reader_queue() : m_tail(0), m_head(0) { }
private: private:
......
...@@ -54,13 +54,11 @@ HEADERS = cppa/actor.hpp \ ...@@ -54,13 +54,11 @@ HEADERS = cppa/actor.hpp \
cppa/util/utype_iterator.hpp \ cppa/util/utype_iterator.hpp \
cppa/util/void_type.hpp cppa/util/void_type.hpp
SOURCES = src/actor.cpp \ SOURCES = src/actor_behavior.cpp \
src/actor_private.cpp \ src/channel.cpp \
src/decorated_tuple.cpp \ src/decorated_tuple.cpp \
src/deserializer.cpp \ src/deserializer.cpp \
src/message_receiver.cpp \
src/mock_scheduler.cpp \ src/mock_scheduler.cpp \
src/ref_counted.cpp \
src/serializer.cpp \ src/serializer.cpp \
src/uniform_type_info.cpp \ src/uniform_type_info.cpp \
src/untyped_tuple.cpp src/untyped_tuple.cpp
......
#include "cppa/actor.hpp"
#include <boost/thread.hpp>
namespace cppa {
actor& actor::operator=(const actor& other)
{
static_cast<super&>(*this) = other;
return *this;
}
actor& actor::operator=(actor&& other)
{
static_cast<super&>(*this) = other;
return *this;
}
actor::~actor()
{
/*
if (m_channel && m_channel->unique())
{
detail::actor_private* ap = dynamic_cast<detail::actor_private*>(m_channel.get());
if (ap && ap->m_thread)
{
if (boost::this_thread::get_id() != ap->m_thread->get_id())
{
ap->m_thread->join();
delete ap->m_thread;
ap->m_thread = 0;
}
}
}
*/
}
} // namespace cppa
#include "cppa/actor_behavior.hpp"
namespace cppa {
void actor_behavior::on_exit() { }
} // namespace cppa
#include "cppa/ref_counted.hpp" #include "cppa/channel.hpp"
namespace cppa { namespace cppa {
//ref_counted::~ref_counted() { } channel::~channel() { }
} // namespace cppa } // namespace cppa
#include "cppa/message_receiver.hpp"
namespace cppa {
message_receiver& message_receiver::operator=(detail::channel* ch_ptr)
{
m_channel.reset(ch_ptr);
return *this;
}
message_receiver&
message_receiver::operator=(const intrusive_ptr<detail::channel>& ch_ptr)
{
m_channel = ch_ptr;
return *this;
}
message_receiver&
message_receiver::operator=(intrusive_ptr<detail::channel>&& ch_ptr)
{
m_channel.swap(ch_ptr);
return *this;
}
message_receiver::~message_receiver()
{
}
} // namespace cppa
#include <boost/thread.hpp> #include <boost/thread.hpp>
#include "cppa/message.hpp"
#include "cppa/context.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/detail/scheduler.hpp" #include "cppa/detail/scheduler.hpp"
#include "cppa/util/singly_linked_list.hpp" #include "cppa/util/singly_linked_list.hpp"
#include "cppa/util/single_reader_queue.hpp" #include "cppa/util/single_reader_queue.hpp"
using namespace cppa;
namespace { namespace {
struct actor_message struct actor_message
{ {
actor_message* next; actor_message* next;
cppa::message msg; message msg;
actor_message(const cppa::message& from) : next(0), msg(from) { } actor_message(const message& from) : next(0), msg(from) { }
}; };
struct actor_impl; struct actor_impl;
void cleanup_fun(actor_impl* what); void cleanup_fun(context* what);
boost::thread_specific_ptr<actor_impl> m_this_actor(cleanup_fun); boost::thread_specific_ptr<context> m_this_context(cleanup_fun);
struct actor_impl : cppa::detail::actor_private struct mbox : message_queue
{ {
cppa::message m_last_dequeued; message m_last_dequeued;
util::single_reader_queue<actor_message> m_impl;
cppa::util::single_reader_queue<actor_message> mailbox;
cppa::detail::behavior* m_behavior;
actor_impl(cppa::detail::behavior* b = 0) : m_behavior(b) { }
~actor_impl() void enqueue(const message& msg)
{ {
if (m_behavior) delete m_behavior; m_impl.push_back(new actor_message(msg));
} }
virtual void enqueue_msg(const cppa::message& msg) virtual const message& dequeue()
{ {
mailbox.push_back(new actor_message(msg)); actor_message* msg = m_impl.pop();
}
virtual const cppa::message& receive()
{
actor_message* msg = mailbox.pop();
m_last_dequeued = std::move(msg->msg); m_last_dequeued = std::move(msg->msg);
delete msg; delete msg;
return m_last_dequeued; return m_last_dequeued;
} }
virtual const cppa::message& last_dequeued() const virtual void dequeue(invoke_rules& rules)
{
return m_last_dequeued;
}
virtual void receive(cppa::invoke_rules& rules)
{ {
actor_message* amsg = mailbox.pop(); actor_message* amsg = m_impl.pop();
cppa::util::singly_linked_list<actor_message> buffer; util::singly_linked_list<actor_message> buffer;
cppa::intrusive_ptr<cppa::detail::intermediate> imd; intrusive_ptr<detail::intermediate> imd;
while (!(imd = rules.get_intermediate(amsg->msg.data()))) while (!(imd = rules.get_intermediate(amsg->msg.data())))
{ {
buffer.push_back(amsg); buffer.push_back(amsg);
amsg = mailbox.pop(); amsg = m_impl.pop();
} }
m_last_dequeued = amsg->msg; m_last_dequeued = amsg->msg;
if (!buffer.empty()) mailbox.prepend(std::move(buffer)); if (!buffer.empty()) m_impl.push_front(std::move(buffer));
imd->invoke(); imd->invoke();
delete amsg; delete amsg;
} }
virtual void send(cppa::detail::channel* whom, virtual bool try_dequeue(message& msg)
cppa::untyped_tuple&& what)
{ {
if (whom) whom->enqueue_msg(cppa::message(this, whom, std::move(what))); if (!m_impl.empty())
{
msg = dequeue();
return true;
}
return false;
} }
void operator()() virtual bool try_dequeue(invoke_rules& rules)
{ {
if (m_behavior) if (!m_impl.empty())
{ {
try dequeue(rules);
{ return true;
m_behavior->act();
}
catch (...) { }
m_behavior->on_exit();
} }
return false;
}
virtual const message& last_dequeued()
{
return m_last_dequeued;
} }
}; };
void cleanup_fun(actor_impl* what) struct actor_impl : context
{ {
if (what)
mbox m_mbox;
actor_behavior* m_behavior;
actor_impl(actor_behavior* b = 0) : m_behavior(b) { }
~actor_impl()
{ {
if (!what->deref()) delete what; if (m_behavior) delete m_behavior;
} }
}
struct actor_ptr virtual void enqueue(const message& msg)
{
cppa::intrusive_ptr<actor_impl> m_impl;
actor_ptr(actor_impl* ai) : m_impl(ai) { }
actor_ptr(const actor_ptr&) = default;
void operator()()
{ {
m_this_actor.reset(m_impl.get()); m_mbox.enqueue(msg);
(*m_impl)();
} }
virtual void link(const intrusive_ptr<actor>&) { }
virtual void unlink(const intrusive_ptr<actor>&) { }
virtual message_queue& mailbox()
{
return m_mbox;
}
}; };
} // namespace <anonymous> void cleanup_fun(context* what)
{
if (what && !what->deref()) delete what;
}
namespace cppa { std::atomic<int> m_running_actors(0);
boost::mutex m_ra_mtx;
boost::condition_variable m_ra_cv;
detail::actor_private* this_actor() void run_actor_impl(intrusive_ptr<actor_impl> m_impl)
{ {
actor_impl* res = m_this_actor.get(); actor_behavior* ab = m_impl->m_behavior;
if (!res) if (ab)
{ {
res = new actor_impl; try
res->ref(); {
m_this_actor.reset(res); ab->act();
}
catch(...) { }
ab->on_exit();
}
if (--m_running_actors == 0)
{
boost::mutex::scoped_lock lock(m_ra_mtx);
m_ra_cv.notify_all();
} }
return res;
} }
} // namespace cppa } // namespace <anonymous>
namespace cppa { namespace detail { namespace cppa { namespace detail {
actor spawn_impl(behavior* actor_behavior) actor_ptr scheduler::spawn(actor_behavior* ab, scheduling_hint)
{
++m_running_actors;
intrusive_ptr<actor_impl> result(new actor_impl(ab));
boost::thread(run_actor_impl, result).detach();
return result;
}
context* scheduler::get_context()
{
context* result = m_this_context.get();
if (!result)
{
result = new actor_impl;
result->ref();
m_this_context.reset(result);
}
return result;
}
void scheduler::await_all_done()
{ {
actor_ptr aptr(new actor_impl(actor_behavior)); boost::mutex::scoped_lock lock(m_ra_mtx);
aptr.m_impl->m_thread = new boost::thread(aptr); while (m_running_actors.load() > 0)
// boost::thread(aptr).detach(); {
return actor(aptr.m_impl); m_ra_cv.wait(lock);
// return actor(std::move(aptr.m_impl)); }
} }
} } // namespace cppa::detail } } // namespace detail
...@@ -2,8 +2,7 @@ ...@@ -2,8 +2,7 @@
#include "hash_of.hpp" #include "hash_of.hpp"
#include "cppa/on.hpp" #include "cppa/on.hpp"
#include "cppa/spawn.hpp" #include "cppa/cppa.hpp"
#include "cppa/reply.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
...@@ -22,14 +21,14 @@ using namespace cppa; ...@@ -22,14 +21,14 @@ using namespace cppa;
namespace { namespace {
struct group : detail::channel struct group : channel
{ {
// NOT thread safe // NOT thread safe
class subscription : public detail::ref_counted_impl<std::size_t> class subscription : public detail::ref_counted_impl<std::size_t>
{ {
actor m_self; actor_ptr m_self;
intrusive_ptr<group> m_group; intrusive_ptr<group> m_group;
public: public:
...@@ -38,7 +37,7 @@ struct group : detail::channel ...@@ -38,7 +37,7 @@ struct group : detail::channel
subscription(const subscription&) = delete; subscription(const subscription&) = delete;
subscription& operator=(const subscription&) = delete; subscription& operator=(const subscription&) = delete;
subscription(const actor& s, const intrusive_ptr<group>& g) subscription(const actor_ptr& s, const intrusive_ptr<group>& g)
: m_self(s), m_group(g) : m_self(s), m_group(g)
{ {
} }
...@@ -50,15 +49,9 @@ struct group : detail::channel ...@@ -50,15 +49,9 @@ struct group : detail::channel
}; };
virtual intrusive_ptr<subscription> subscribe(const actor& who) = 0; virtual intrusive_ptr<subscription> subscribe(const actor_ptr& who) = 0;
virtual void unsubscribe(const actor& who) = 0; virtual void unsubscribe(const actor_ptr& who) = 0;
template<typename... Args>
void send(const Args&... args)
{
enqueue_msg(message(this_actor(), this, args...));
}
}; };
...@@ -66,25 +59,25 @@ class local_group : public group ...@@ -66,25 +59,25 @@ class local_group : public group
{ {
boost::mutex m_mtx; boost::mutex m_mtx;
std::list<actor> m_subscribers; std::list<actor_ptr> m_subscribers;
inline std::list<actor>::iterator find(const actor& what) inline std::list<actor_ptr>::iterator find(const actor_ptr& what)
{ {
return std::find(m_subscribers.begin(), m_subscribers.end(), what); return std::find(m_subscribers.begin(), m_subscribers.end(), what);
} }
public: public:
virtual void enqueue_msg(const message& msg) virtual void enqueue(const message& msg)
{ {
boost::mutex::scoped_lock guard(m_mtx); boost::mutex::scoped_lock guard(m_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i) for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
{ {
i->enqueue_msg(msg); (*i)->enqueue(msg);
} }
} }
virtual intrusive_ptr<group::subscription> subscribe(const actor& who) virtual intrusive_ptr<group::subscription> subscribe(const actor_ptr& who)
{ {
boost::mutex::scoped_lock guard(m_mtx); boost::mutex::scoped_lock guard(m_mtx);
auto i = find(who); auto i = find(who);
...@@ -96,7 +89,7 @@ class local_group : public group ...@@ -96,7 +89,7 @@ class local_group : public group
return new group::subscription(0, 0); return new group::subscription(0, 0);
} }
virtual void unsubscribe(const actor& who) virtual void unsubscribe(const actor_ptr& who)
{ {
boost::mutex::scoped_lock guard(m_mtx); boost::mutex::scoped_lock guard(m_mtx);
auto i = find(who); auto i = find(who);
...@@ -110,9 +103,9 @@ class local_group : public group ...@@ -110,9 +103,9 @@ class local_group : public group
//local_group* m_local_group = new local_group; //local_group* m_local_group = new local_group;
local_group m_local_group; intrusive_ptr<local_group> m_local_group;
group& local(const char*) intrusive_ptr<local_group> local(const char*)
{ {
return m_local_group; return m_local_group;
} }
...@@ -145,18 +138,7 @@ struct storage ...@@ -145,18 +138,7 @@ struct storage
}; };
/* void foo_actor_ptr()
template<typename... Args>
void send(std::list<actor>& actors, const Args&... args)
{
for (auto i = actors.begin(); i != actors.end(); ++i)
{
i->send(args...);
}
}
*/
void foo_actor()
{ {
receive(on<int>() >> [](int i) { receive(on<int>() >> [](int i) {
reply(i); reply(i);
...@@ -170,13 +152,13 @@ std::size_t test__local_group() ...@@ -170,13 +152,13 @@ std::size_t test__local_group()
std::list<intrusive_ptr<group::subscription>> m_subscriptions; std::list<intrusive_ptr<group::subscription>> m_subscriptions;
group& lg = local("foobar"); auto lg = local("foobar");
for (int i = 0; i < 5; ++i) for (int i = 0; i < 5; ++i)
{ {
m_subscriptions.push_back(lg.subscribe(spawn(foo_actor))); m_subscriptions.push_back(lg->subscribe(spawn(foo_actor_ptr)));
} }
lg.send(1); send(lg, 1);
int result = 0; int result = 0;
......
...@@ -403,11 +403,11 @@ std::size_t test__serialization() ...@@ -403,11 +403,11 @@ std::size_t test__serialization()
std::string str = "Hello World"; std::string str = "Hello World";
CPPA_CHECK_EQUAL(to_string(make_tuple(str)), // CPPA_CHECK_EQUAL(to_string(make_tuple(str)),
"{ \"Hello World\" }"); // "{ \"Hello World\" }");
CPPA_CHECK_EQUAL(to_string(make_tuple(str, 42)), // CPPA_CHECK_EQUAL(to_string(make_tuple(str, 42)),
"{ \"Hello World\", 42 }"); // "{ \"Hello World\", 42 }");
char v0 = 0x11; char v0 = 0x11;
short v1 = 0x1122; short v1 = 0x1122;
......
...@@ -4,8 +4,8 @@ ...@@ -4,8 +4,8 @@
#include "test.hpp" #include "test.hpp"
#include "cppa/on.hpp" #include "cppa/on.hpp"
#include "cppa/cppa.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/spawn.hpp"
#include "cppa/detail/scheduler.hpp" #include "cppa/detail/scheduler.hpp"
using std::cout; using std::cout;
...@@ -26,14 +26,16 @@ std::size_t test__spawn() ...@@ -26,14 +26,16 @@ std::size_t test__spawn()
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
{ {
actor sl = spawn(pong); auto sl = spawn(pong);
sl.send(23.f); send(sl, 23.f);
sl.send(2); send(sl, 2);
receive(on<int>() >> [&](int value) { receive(on<int>() >> [&](int value) {
CPPA_CHECK_EQUAL(value, 42); CPPA_CHECK_EQUAL(value, 42);
}); });
} }
await_all_actors_done();
return CPPA_TEST_RESULT; return CPPA_TEST_RESULT;
} }
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