Commit 50543c98 authored by neverlord's avatar neverlord

exit_signal and serialization of group_ptr + channel_ptr

parent 904adf60
......@@ -148,3 +148,6 @@ src/object_array.cpp
cppa/any_tuple.hpp
src/any_tuple.cpp
cppa/util/abstract_uniform_type_info.hpp
cppa/util/upgrade_lock_guard.hpp
cppa/exit_signal.hpp
src/exit_signal.cpp
......@@ -6,12 +6,21 @@
namespace cppa {
// forward declaration
// forward declarations
class actor;
class group;
class message;
class channel : public ref_counted
{
friend class actor;
friend class group;
// channel is a sealed class and the only
// allowed subclasses are actor and group.
channel() = default;
public:
virtual ~channel();
......
......@@ -15,15 +15,15 @@ namespace cppa { namespace detail {
class converted_thread_context : public context
{
// true if the associated thread has finished execution
bool m_exited;
// mailbox implementation
detail::blocking_message_queue m_mailbox;
// guards access to m_exited, m_subscriptions and m_links
std::mutex m_mtx;
// true if the associated thread has finished execution
bool m_exited;
// manages group subscriptions
std::map<group_ptr, group::subscription> m_subscriptions;
......@@ -32,6 +32,8 @@ class converted_thread_context : public context
public:
converted_thread_context();
message_queue& mailbox /*[[override]]*/ ();
void join /*[[override]]*/ (group_ptr& what);
......
......@@ -2,8 +2,10 @@
#define DEFAULT_UNIFORM_TYPE_INFO_IMPL_HPP
#include "cppa/any_type.hpp"
#include "cppa/util/void_type.hpp"
#include "cppa/util/rm_ref.hpp"
#include "cppa/util/void_type.hpp"
#include "cppa/util/conjunction.hpp"
#include "cppa/util/disjunction.hpp"
#include "cppa/util/is_iterable.hpp"
#include "cppa/util/is_primitive.hpp"
......@@ -98,11 +100,11 @@ class default_uniform_type_info_impl : public util::abstract_uniform_type_info<T
std::function<void (const uniform_type_info*,
const void*,
serializer* )> m_serialize;
serializer* )> m_serialize;
std::function<void (const uniform_type_info*,
void*,
deserializer* )> m_deserialize;
deserializer* )> m_deserialize;
member(const member&) = delete;
member& operator=(const member&) = delete;
......@@ -141,6 +143,32 @@ class default_uniform_type_info_impl : public util::abstract_uniform_type_info<T
};
}
template<typename GRes, typename SRes, typename SArg, class C>
member(uniform_type_info* mtptr,
GRes (C::*getter)() const,
SRes (C::*setter)(SArg)) : m_meta(mtptr)
{
typedef typename util::rm_ref<GRes>::type getter_result;
typedef typename util::rm_ref<SArg>::type setter_arg;
static_assert(std::is_same<getter_result, setter_arg>::value,
"getter result doesn't match setter argument");
m_serialize = [getter] (const uniform_type_info* mt,
const void* obj,
serializer* s)
{
GRes v = (*reinterpret_cast<const C*>(obj).*getter)();
mt->serialize(&v, s);
};
m_deserialize = [setter] (const uniform_type_info* mt,
void* obj,
deserializer* d)
{
setter_arg value;
mt->deserialize(&value, d);
(*reinterpret_cast<C*>(obj).*setter)(value);
};
}
member(member&& other) : m_meta(nullptr)
{
swap(other);
......@@ -201,6 +229,16 @@ class default_uniform_type_info_impl : public util::abstract_uniform_type_info<T
push_back(args...);
}
// pr.first = getter member function pointer
// pr.second = setter member function pointer
template<typename GRes, typename SRes, typename SArg, class C>
typename util::enable_if<util::is_primitive<typename util::rm_ref<GRes>::type> >::type
push_back(std::pair<GRes (C::*)() const, SRes (C::*)(SArg)> pr)
{
typedef typename util::rm_ref<GRes>::type memtype;
m_members.push_back({ new primitive_member<memtype>(), pr.first, pr.second });
}
template<typename R, class C, typename... Args>
typename util::enable_if<util::is_primitive<R> >::type
push_back(R C::*mem_ptr, const Args&... args)
......
#ifndef EXIT_SIGNAL_HPP
#define EXIT_SIGNAL_HPP
#include <cstdint>
namespace cppa {
enum class exit_reason : std::uint32_t
{
/**
* @brief Indicates that an actor finished execution.
*/
normal = 0x00,
/**
* @brief Indicates that an actor finished execution
* because of an unhandled exception.
*/
unhandled_exception = 0x01,
/**
* @brief Any user defined exit reason should have a
* value greater or equal to prevent collisions
* with default defined exit reasons.
*/
user_defined = 0x10000
};
/**
* @brief Converts {@link exit_reason} to @c std::uint32_t.
*/
constexpr std::uint32_t to_uint(exit_reason r)
{
return static_cast<std::uint32_t>(r);
}
class exit_signal
{
std::uint32_t m_reason;
public:
/**
* @brief Creates an exit signal with
* <tt>reason() == exit_reason::normal</tt>.
*/
exit_signal();
/**
* @brief Creates an exit signal with
* <tt>reason() == @p r</tt>.
* @pre {@code r != exit_reason::user_defined}.
*/
exit_signal(exit_reason r);
/**
* @brief Creates an exit signal with
* <tt>reason() == @p r</tt>.
* @pre {@code r >= exit_reason::user_defined}.
*/
exit_signal(std::uint32_t r);
/**
* @brief Reads the exit reason.
*/
inline std::uint32_t reason() const
{
return m_reason;
}
// an unambiguous member function
void set_uint_reason(std::uint32_t value);
/**
* @brief Sets the exit reason to @p value.
*/
void set_reason(std::uint32_t value);
/**
* @brief Sets the exit reason to @p value.
*/
void set_reason(exit_reason value);
};
} // namespace cppa
#endif // EXIT_SIGNAL_HPP
......@@ -11,8 +11,15 @@ namespace cppa {
class group : public channel
{
std::string m_identifier;
std::string m_module_name;
protected:
group(std::string&& id, std::string&& mod_name);
group(const std::string& id, const std::string& mod_name);
virtual void unsubscribe(const channel_ptr& who) = 0;
public:
......@@ -33,6 +40,11 @@ class group : public channel
public:
inline explicit operator bool()
{
return m_self != nullptr && m_group != nullptr;
}
subscription(const channel_ptr& s, const intrusive_ptr<group>& g);
subscription(subscription&& other);
~subscription();
......@@ -49,6 +61,10 @@ class group : public channel
};
const std::string& identifier() const;
const std::string& module_name() const;
virtual subscription subscribe(const channel_ptr& who) = 0;
static intrusive_ptr<group> get(const std::string& module_name,
......
......@@ -65,6 +65,13 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, T*>,
const T* get() const { return m_ptr; }
T* take()
{
auto result = m_ptr;
m_ptr = nullptr;
return result;
}
void swap(intrusive_ptr& other)
{
std::swap(m_ptr, other.m_ptr);
......@@ -111,8 +118,8 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, T*>,
{
static_assert(std::is_convertible<Y*, T*>::value,
"Y* is not assignable to T*");
m_ptr = other.m_ptr;
other.m_ptr = nullptr;
reset();
m_ptr = other.take();
return *this;
}
......
#ifndef MESSAGE_QUEUE_HPP
#define MESSAGE_QUEUE_HPP
#include "cppa/channel.hpp"
#include "cppa/ref_counted.hpp"
namespace cppa {
// forward declaration
class invoke_rules;
class message_queue : public channel
class message_queue : public ref_counted
{
public:
virtual void enqueue(const message&) = 0;
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;
virtual void enqueue(const message&) = 0;
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;
};
......
......@@ -29,18 +29,16 @@ const uniform_type_info* uniform_typeid(const std::type_info&);
* @brief Provides a platform independent type name and a (very primitive)
* kind of reflection in combination with {@link cppa::object object}.
*
* The platform dependent type name (from GCC or Microsofts VC++ Compiler) is
* translated to a (usually) shorter and platform independent name.
*
* This name is equal to the "in-sourcecode-name" but with a few exceptions:
* The platform independent name is equal to the "in-sourcecode-name"
* with a few exceptions:
* - @c std::string is named @c \@str
* - @c std::u16string is named @c \@u16str
* - @c std::u32string is named @c \@u32str
* - @c integers are named <tt>\@(i|u)$size</tt>\n
* e.g.: @c \@i32 is a 32 bit signed integer; @c \@u16
* is a 16 bit unsigned integer
* - the <em>anonymous namespace</em> is named @c \@_\n
* e.g.: @code namespace { class foo { }; } @endcode is mapped to
* - the <em>anonymous namespace</em> is named @c \@_ \n
* e.g.: <tt>namespace { class foo { }; }</tt> is mapped to
* @c \@_::foo
* - {@link cppa::util::void_type} is named @c \@0
*/
......
......@@ -7,20 +7,31 @@ template<typename SharedLockable>
class shared_lock_guard
{
SharedLockable& m_lockable;
SharedLockable* m_lockable;
public:
shared_lock_guard(SharedLockable& lockable) : m_lockable(lockable)
explicit shared_lock_guard(SharedLockable& lockable) : m_lockable(&lockable)
{
m_lockable.lock_shared();
m_lockable->lock_shared();
}
~shared_lock_guard()
{
m_lockable.unlock_shared();
if (m_lockable) m_lockable->unlock_shared();
}
bool owns_lock() const
{
return m_lockable != nullptr;
}
SharedLockable* release()
{
auto result = m_lockable;
m_lockable = nullptr;
return result;
}
};
} } // namespace cppa::util
......
......@@ -23,6 +23,12 @@ class shared_spinlock
void unlock_shared();
bool try_lock_shared();
/**
* @brief Upgrades a shared lock to an exclusive lock (must be released
* through a call to {@link lock()} afterwards).
*/
void lock_upgrade();
};
} } // namespace cppa::util
......
#ifndef UPGRADE_LOCK_GUARD_HPP
#define UPGRADE_LOCK_GUARD_HPP
namespace cppa { namespace util {
/**
* @brief Upgrades shared ownership to exclusive ownership.
*/
template<typename UpgradeLockable>
class upgrade_lock_guard
{
UpgradeLockable* m_lockable;
public:
template<template <typename> class LockType>
upgrade_lock_guard(LockType<UpgradeLockable>& other)
{
m_lockable = other.release();
if (m_lockable) m_lockable->lock_upgrade();
}
~upgrade_lock_guard()
{
if (m_lockable) m_lockable->unlock();
}
bool owns_lock() const
{
return m_lockable != nullptr;
}
};
} } // namespace cppa::util
#endif // UPGRADE_LOCK_GUARD_HPP
......@@ -15,6 +15,7 @@ HEADERS = \
cppa/cow_ptr.hpp \
cppa/cppa.hpp \
cppa/deserializer.hpp \
cppa/exit_signal.hpp \
cppa/get.hpp \
cppa/get_view.hpp \
cppa/group.hpp \
......@@ -102,6 +103,7 @@ HEADERS = \
cppa/util/type_list.hpp \
cppa/util/type_list_apply.hpp \
cppa/util/type_list_pop_back.hpp \
cppa/util/upgrade_lock_guard.hpp \
cppa/util/void_type.hpp \
cppa/util/wrapped_type.hpp
......@@ -118,6 +120,7 @@ SOURCES = \
src/converted_thread_context.cpp \
src/demangle.cpp \
src/deserializer.cpp \
src/exit_signal.cpp \
src/group.cpp \
src/message.cpp \
src/mock_scheduler.cpp \
......
......@@ -2,6 +2,10 @@
namespace cppa { namespace detail {
converted_thread_context::converted_thread_context() : m_exited(false)
{
}
void converted_thread_context::link(intrusive_ptr<actor>& other)
{
std::lock_guard<std::mutex> guard(m_mtx);
......@@ -21,15 +25,30 @@ bool converted_thread_context::remove_backlink(const intrusive_ptr<actor>& other
return false;
}
bool converted_thread_context::establish_backlink(const intrusive_ptr<actor>& other)
{
bool send_exit_message = false;
bool result = false;
if (other && other != this)
{
std::lock_guard<std::mutex> guard(m_mtx);
return m_links.insert(other).second;
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(m_mtx);
if (!m_exited)
{
result = m_links.insert(other).second;
}
else
{
send_exit_message = true;
}
}
}
return false;
if (send_exit_message)
{
}
return result;
}
void converted_thread_context::unlink(intrusive_ptr<actor>& other)
......@@ -46,8 +65,10 @@ void converted_thread_context::join(group_ptr& what)
std::lock_guard<std::mutex> guard(m_mtx);
if (!m_exited && m_subscriptions.count(what) == 0)
{
m_subscriptions.insert(std::make_pair(what, what->subscribe(this)));
//m_subscriptions[what] = what->subscribe(this);
auto s = what->subscribe(this);
// insert only valid subscriptions
// (a subscription is invalid if this actor already joined the group)
if (s) m_subscriptions.insert(std::make_pair(what, std::move(s)));
}
}
......
#include "cppa/exit_signal.hpp"
namespace cppa {
exit_signal::exit_signal() : m_reason(to_uint(exit_reason::normal))
{
}
exit_signal::exit_signal(exit_reason r) : m_reason(to_uint(r))
{
}
exit_signal::exit_signal(std::uint32_t r) : m_reason(r)
{
}
void exit_signal::set_reason(std::uint32_t value)
{
m_reason = value;
}
void exit_signal::set_reason(exit_reason value)
{
m_reason = to_uint(value);
}
void exit_signal::set_uint_reason(std::uint32_t value)
{
set_reason(value);
}
} // namespace cppa
#include "cppa/config.hpp"
#include <map>
#include <set>
#include <list>
#include <mutex>
#include <memory>
#include <stdexcept>
#include "cppa/group.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"
namespace cppa {
......@@ -13,19 +18,133 @@ namespace {
typedef std::map< std::string, std::unique_ptr<group::module> > modules_map;
typedef std::lock_guard<util::shared_spinlock> exclusive_guard;
typedef util::shared_lock_guard<util::shared_spinlock> shared_guard;
typedef util::upgrade_lock_guard<util::shared_spinlock> upgrade_guard;
std::mutex s_mtx;
modules_map s_mmap;
class local_group : public group
{
friend class local_group_module;
util::shared_spinlock m_mtx;
std::set<channel_ptr> m_subscribers;
// allow access to local_group_module only
local_group(std::string&& gname) : group(std::move(gname), "local")
{
}
public:
virtual void enqueue(const message& msg)
{
shared_guard guard(m_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
{
const_cast<channel_ptr&>(*i)->enqueue(msg);
}
}
virtual group::subscription subscribe(const channel_ptr& who)
{
exclusive_guard guard(m_mtx);
if (m_subscribers.insert(who).second)
{
return { who, this };
}
else
{
return { nullptr, nullptr };
}
}
virtual void unsubscribe(const channel_ptr& who)
{
exclusive_guard guard(m_mtx);
m_subscribers.erase(who);
}
};
class local_group_module : public group::module
{
std::string m_name;
util::shared_spinlock m_mtx;
std::map<std::string, group_ptr> m_instances;
public:
local_group_module() : m_name("local")
{
}
const std::string& name()
{
return m_name;
}
group_ptr get(const std::string& group_name)
{
shared_guard guard(m_mtx);
auto i = m_instances.find(group_name);
if (i != m_instances.end())
{
return i->second;
}
else
{
group_ptr tmp(new local_group(std::string(group_name)));
// lifetime scope of uguard
{
upgrade_guard uguard(guard);
auto p = m_instances.insert(std::make_pair(group_name, tmp));
if (p.second == false)
{
// someone preempt us
return p.first->second;
}
else
{
// ok, inserted tmp
return tmp;
}
}
}
}
};
modules_map& mmap()
{
if (s_mmap.empty())
{
// lazy initialization
s_mmap.insert(std::make_pair(std::string("local"),
new local_group_module));
}
return s_mmap;
}
} // namespace <anonymous>
intrusive_ptr<group> group::get(const std::string& module_name,
const std::string& group_name)
{
std::lock_guard<std::mutex> guard(s_mtx);
auto i = s_mmap.find(module_name);
if (i != s_mmap.end())
// lifetime scope of guard
{
return (i->second)->get(group_name);
std::lock_guard<std::mutex> guard(s_mtx);
auto& mmref = mmap();
auto i = mmref.find(module_name);
if (i != mmref.end())
{
return (i->second)->get(group_name);
}
}
std::string error_msg = "no module named \"";
error_msg += module_name;
......@@ -40,14 +159,16 @@ void group::add_module(group::module* ptr)
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(s_mtx);
if (!s_mmap.insert(std::make_pair(mname, std::move(mptr))).second)
if (mmap().insert(std::make_pair(mname, std::move(mptr))).second)
{
std::string error_msg = "module name \"";
error_msg += mname;
error_msg += "\" already defined";
throw std::logic_error(error_msg);
return; // success; don't throw exception
}
}
std::string error_msg = "module name \"";
error_msg += mname;
error_msg += "\" already defined";
throw std::logic_error(error_msg);
}
group::subscription::subscription(const channel_ptr& s,
......@@ -66,4 +187,24 @@ group::subscription::~subscription()
if (m_group) m_group->unsubscribe(m_self);
}
group::group(std::string&& id, std::string&& mod_name)
: m_identifier(std::move(id)), m_module_name(std::move(mod_name))
{
}
group::group(const std::string& id, const std::string& mod_name)
: m_identifier(id), m_module_name(mod_name)
{
}
const std::string& group::identifier() const
{
return m_identifier;
}
const std::string& group::module_name() const
{
return m_module_name;
}
} // namespace cppa
......@@ -35,6 +35,28 @@ void shared_spinlock::lock()
}
}
void shared_spinlock::lock_upgrade()
{
unlock_shared();
lock();
/*
long v = m_flag.load();
for (;;)
{
if (v != 1)
{
//std::this_thread::yield();
v = m_flag.load();
}
else if (m_flag.compare_exchange_weak(v, min_long))
{
return;
}
// else: next iteration
}
*/
}
void shared_spinlock::unlock()
{
m_flag.store(0);
......
......@@ -15,6 +15,7 @@
#include "cppa/announce.hpp"
#include "cppa/any_type.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/exit_signal.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/uniform_type_info.hpp"
......@@ -96,20 +97,184 @@ void push(std::map<int, std::pair<string_set, string_set>>& ints)
namespace cppa { namespace detail { namespace {
class actor_ptr_type_info : public util::abstract_uniform_type_info<actor_ptr>
const std::string nullptr_type_name = "@0";
void serialize_nullptr(serializer* sink)
{
sink->begin_object(nullptr_type_name);
sink->end_object();
}
void deserialize_nullptr(deserializer* source)
{
source->begin_object(nullptr_type_name);
source->end_object();
}
class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
{
public:
static void s_serialize(const actor* ptr, serializer* sink,
const std::string name)
{
if (!ptr)
{
serialize_nullptr(sink);
}
else
{
sink->begin_object(name);
sink->write_value(ptr->id());
sink->end_object();
}
}
static void s_deserialize(actor_ptr& ptrref, deserializer* source,
const std::string name)
{
std::string cname = source->seek_object();
if (cname != name)
{
if (cname == nullptr_type_name)
{
deserialize_nullptr(source);
ptrref.reset();
}
else
{
throw std::logic_error("wrong type name found");
}
}
else
{
source->begin_object(cname);
auto id = get<std::uint32_t>(source->read_value(pt_uint32));
ptrref = actor::by_id(id);
source->end_object();
}
}
protected:
void serialize(const void* ptr, serializer* sink) const
{
s_serialize(reinterpret_cast<const actor_ptr*>(ptr)->get(),
sink,
name());
}
void deserialize(void* ptr, deserializer* source) const
{
s_deserialize(*reinterpret_cast<actor_ptr*>(ptr), source, name());
}
};
class group_ptr_tinfo : public util::abstract_uniform_type_info<group_ptr>
{
public:
static void s_serialize(const group* ptr, serializer* sink,
const std::string name)
{
if (!ptr)
{
serialize_nullptr(sink);
}
else
{
sink->begin_object(name);
sink->write_value(ptr->module_name());
sink->write_value(ptr->identifier());
sink->end_object();
}
}
static void s_deserialize(group_ptr& ptrref, deserializer* source,
const std::string name)
{
std::string cname = source->seek_object();
if (cname != name)
{
if (cname == nullptr_type_name)
{
deserialize_nullptr(source);
ptrref.reset();
}
else
{
throw std::logic_error("wrong type name found");
}
}
else
{
source->begin_object(name);
auto modname = source->read_value(pt_u8string);
auto groupid = source->read_value(pt_u8string);
ptrref = group::get(get<std::string>(modname),
get<std::string>(groupid));
source->end_object();
}
}
protected:
void serialize(const void* ptr, serializer* sink) const
{
s_serialize(reinterpret_cast<const group_ptr*>(ptr)->get(),
sink,
name());
}
void deserialize(void* ptr, deserializer* source) const
{
s_deserialize(*reinterpret_cast<group_ptr*>(ptr), source, name());
}
};
class channel_ptr_tinfo : public util::abstract_uniform_type_info<channel_ptr>
{
std::string group_ptr_name;
std::string actor_ptr_name;
protected:
void serialize(const void* instance, serializer* sink) const
{
sink->begin_object(name());
auto id = (*reinterpret_cast<const actor_ptr*>(ptr))->id();
sink->write_value(id);
auto ptr = reinterpret_cast<const channel_ptr*>(instance)->get();
if (!ptr)
{
serialize_nullptr(sink);
}
else
{
const group* gptr;
auto aptr = dynamic_cast<const actor*>(ptr);
if (aptr)
{
actor_ptr_tinfo::s_serialize(aptr, sink, actor_ptr_name);
}
else if ((gptr = dynamic_cast<const group*>(ptr)) != nullptr)
{
group_ptr_tinfo::s_serialize(gptr, sink, group_ptr_name);
}
else
{
throw std::logic_error("channel is neither "
"an actor nor a group");
}
}
sink->end_object();
}
void deserialize(void* ptr, deserializer* source) const
void deserialize(void* instance, deserializer* source) const
{
std::string cname = source->seek_object();
if (cname != name())
......@@ -117,14 +282,43 @@ class actor_ptr_type_info : public util::abstract_uniform_type_info<actor_ptr>
throw std::logic_error("wrong type name found");
}
source->begin_object(cname);
auto id = get<std::uint32_t>(source->read_value(pt_uint32));
*reinterpret_cast<actor_ptr*>(ptr) = actor::by_id(id);
std::string subobj = source->peek_object();
if (subobj == actor_ptr_name)
{
actor_ptr tmp;
actor_ptr_tinfo::s_deserialize(tmp, source, actor_ptr_name);
*reinterpret_cast<channel_ptr*>(instance) = tmp;
}
else if (subobj == group_ptr_name)
{
group_ptr tmp;
group_ptr_tinfo::s_deserialize(tmp, source, group_ptr_name);
*reinterpret_cast<channel_ptr*>(instance) = tmp;
}
else if (subobj == nullptr_type_name)
{
(void) source->seek_object();
deserialize_nullptr(source);
reinterpret_cast<channel_ptr*>(instance)->reset();
}
else
{
throw std::logic_error("unexpected type name: " + subobj);
}
source->end_object();
}
public:
channel_ptr_tinfo() : group_ptr_name(to_uniform_name(typeid(group_ptr)))
, actor_ptr_name(to_uniform_name(typeid(actor_ptr)))
{
}
};
class any_tuple_type_info : public util::abstract_uniform_type_info<any_tuple>
class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple>
{
protected:
......@@ -205,8 +399,14 @@ class uniform_type_info_map
insert<std::string>();
insert<std::u16string>();
insert<std::u32string>();
insert(new actor_ptr_type_info, { raw_name<actor_ptr>() });
insert(new any_tuple_type_info, { raw_name<any_tuple>() });
insert(new default_uniform_type_info_impl<exit_reason>(
std::make_pair(&exit_signal::reason,
&exit_signal::set_uint_reason)),
{ raw_name<exit_signal>() });
insert(new any_tuple_tinfo, { raw_name<any_tuple>() });
insert(new actor_ptr_tinfo, { raw_name<actor_ptr>() });
insert(new group_ptr_tinfo, { raw_name<actor_ptr>() });
insert(new channel_ptr_tinfo, { raw_name<channel_ptr>() });
insert<float>();
insert<cppa::util::void_type>();
if (sizeof(double) == sizeof(long double))
......
......@@ -21,90 +21,6 @@ using std::endl;
using namespace cppa;
class local_group : public group
{
friend class local_group_module;
std::mutex m_mtx;
std::list<channel_ptr> m_subscribers;
inline std::list<channel_ptr>::iterator find(const channel_ptr& what)
{
return std::find(m_subscribers.begin(), m_subscribers.end(), what);
}
local_group() = default;
public:
virtual void enqueue(const message& msg)
{
std::lock_guard<std::mutex> guard(m_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
{
(*i)->enqueue(msg);
}
}
virtual group::subscription subscribe(const channel_ptr& who)
{
std::lock_guard<std::mutex> guard(m_mtx);
auto i = find(who);
if (i == m_subscribers.end())
{
m_subscribers.push_back(who);
return { who, this };
}
return { nullptr, nullptr };
}
virtual void unsubscribe(const channel_ptr& who)
{
std::lock_guard<std::mutex> guard(m_mtx);
auto i = find(who);
if (i != m_subscribers.end())
{
m_subscribers.erase(i);
}
}
};
class local_group_module : public group::module
{
std::string m_name;
std::mutex m_mtx;
std::map<std::string, group_ptr> m_instances;
public:
local_group_module() : m_name("local")
{
}
const std::string& name()
{
return m_name;
}
group_ptr get(const std::string& group_name)
{
std::lock_guard<std::mutex> guard(m_mtx);
auto i = m_instances.find(group_name);
if (i == m_instances.end())
{
group_ptr result(new local_group);
m_instances.insert(std::make_pair(group_name, result));
return result;
}
else return i->second;
}
};
void worker()
{
receive(on<int>() >> [](int value) {
......@@ -115,11 +31,10 @@ void worker()
size_t test__local_group()
{
CPPA_TEST(test__local_group);
group::add_module(new local_group_module);
auto foo_group = group::get("local", "foo");
for (int i = 0; i < 5; ++i)
{
// spawn workers in group local:foo
// spawn workers and let them join local/foo
spawn(worker)->join(foo_group);
}
send(foo_group, 2);
......
This diff is collapsed.
......@@ -89,7 +89,10 @@ size_t test__uniform_type()
// default announced cppa types
"cppa::any_type",
"cppa::any_tuple",
"cppa::intrusive_ptr<cppa::actor>"
"cppa::exit_reason",
"cppa::intrusive_ptr<cppa::actor>",
"cppa::intrusive_ptr<cppa::group>",
"cppa::intrusive_ptr<cppa::channel>"
};
if (sizeof(double) != sizeof(long double))
{
......
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