Commit 880c838d authored by neverlord's avatar neverlord

maintenance

parent c8cb7827
......@@ -22,7 +22,6 @@ cppa/detail/tuple_vals.hpp
cppa/match.hpp
cppa/detail/decorated_tuple.hpp
cppa/cow_ptr.hpp
cppa/util/detach.hpp
cppa/detail/ref_counted_impl.hpp
cppa/intrusive_ptr.hpp
unit_testing/test__spawn.cpp
......@@ -184,3 +183,7 @@ src/native_socket.cpp
cppa/detail/post_office.hpp
src/post_office.cpp
cppa/detail/buffer.hpp
cppa/detail/task_scheduler.hpp
src/task_scheduler.cpp
cppa/detail/actor_count.hpp
src/actor_count.cpp
......@@ -15,6 +15,9 @@ namespace cppa {
class attachable
{
attachable(const attachable&) = delete;
attachable& operator=(const attachable&) = delete;
protected:
attachable() = default;
......
......@@ -35,7 +35,7 @@ class context : public actor
*
* Calls <code>mailbox().enqueue(msg)</code>.
*/
virtual void enqueue /*[[override]]*/ (const message& msg);
virtual void enqueue(const message& msg) /*override*/;
void trap_exit(bool new_value);
......
......@@ -2,73 +2,109 @@
#define COW_PTR_HPP
#include <stdexcept>
#include "cppa/util/detach.hpp"
#include <type_traits>
#include "cppa/intrusive_ptr.hpp"
#include "cppa/util/is_copyable.hpp"
#include "cppa/util/has_copy_member_fun.hpp"
namespace cppa { namespace detail {
template<typename T>
constexpr int copy_method()
{
return util::has_copy_member_fun<T>::value
? 2
: (util::is_copyable<T>::value ? 1 : 0);
}
// is_copyable
template<typename T>
T* copy_of(const T* what, std::integral_constant<int, 1>)
{
return new T(*what);
}
// has_copy_member_fun
template<typename T>
T* copy_of(const T* what, std::integral_constant<int, 2>)
{
return what->copy();
}
} } // namespace cppa::detail
namespace cppa {
template<typename T>
class cow_ptr
{
intrusive_ptr<T> m_ptr;
T* do_detach()
{
T* ptr = m_ptr.get();
if (!ptr->unique())
{
T* new_ptr = util::detach(ptr);
cow_ptr tmp(new_ptr);
swap(tmp);
return new_ptr;
}
return ptr;
}
static_assert(detail::copy_method<T>() != 0, "T is not copyable");
typedef std::integral_constant<int, detail::copy_method<T>()> copy_of_token;
intrusive_ptr<T> m_ptr;
T* detach_ptr()
{
T* ptr = m_ptr.get();
if (!ptr->unique())
{
T* new_ptr = detail::copy_of(ptr, copy_of_token());
cow_ptr tmp(new_ptr);
swap(tmp);
return new_ptr;
}
return ptr;
}
public:
template<typename Y>
cow_ptr(Y* raw_ptr) : m_ptr(raw_ptr) { }
template<typename Y>
cow_ptr(Y* raw_ptr) : m_ptr(raw_ptr) { }
cow_ptr(T* raw_ptr) : m_ptr(raw_ptr) { }
cow_ptr(T* raw_ptr) : m_ptr(raw_ptr) { }
cow_ptr(const cow_ptr& other) : m_ptr(other.m_ptr) { }
cow_ptr(const cow_ptr& other) : m_ptr(other.m_ptr) { }
template<typename Y>
cow_ptr(const cow_ptr<Y>& other) : m_ptr(const_cast<Y*>(other.get())) { }
template<typename Y>
cow_ptr(const cow_ptr<Y>& other) : m_ptr(const_cast<Y*>(other.get())) { }
inline void swap(cow_ptr& other)
{
m_ptr.swap(other.m_ptr);
}
const T* get() const { return m_ptr.get(); }
cow_ptr& operator=(const cow_ptr& other)
{
cow_ptr tmp(other);
swap(tmp);
return *this;
}
inline void swap(cow_ptr& other)
{
m_ptr.swap(other.m_ptr);
}
template<typename Y>
cow_ptr& operator=(const cow_ptr<Y>& other)
{
cow_ptr tmp(other);
swap(tmp);
return *this;
}
cow_ptr& operator=(const cow_ptr& other)
{
cow_ptr tmp(other);
swap(tmp);
return *this;
}
inline T* get() { return detach_ptr(); }
template<typename Y>
cow_ptr& operator=(const cow_ptr<Y>& other)
{
cow_ptr tmp(other);
swap(tmp);
return *this;
}
inline const T* get() const { return m_ptr.get(); }
T* operator->() { return do_detach(); }
inline T* operator->() { return detach_ptr(); }
T& operator*() { return do_detach(); }
inline T& operator*() { return detach_ptr(); }
const T* operator->() const { return m_ptr.get(); }
inline const T* operator->() const { return m_ptr.get(); }
const T& operator*() const { return *m_ptr.get(); }
inline const T& operator*() const { return *m_ptr.get(); }
explicit operator bool() const { return m_ptr.get() != nullptr; }
inline explicit operator bool() const { return static_cast<bool>(m_ptr); }
};
......
......@@ -140,7 +140,7 @@ class abstract_actor /*[[base_check]]*/ : public Base
public:
bool attach /*[[override]]*/ (attachable* ptr)
bool attach(attachable* ptr) /*override*/
{
if (ptr == nullptr)
{
......@@ -166,7 +166,7 @@ class abstract_actor /*[[base_check]]*/ : public Base
}
}
void detach /*[[override]]*/ (const attachable::token& what)
void detach(const attachable::token& what) /*override*/
{
attachable_ptr uptr;
// lifetime scope of guard
......@@ -186,17 +186,17 @@ class abstract_actor /*[[base_check]]*/ : public Base
// uptr will be destroyed here, without locked mutex
}
void link_to /*[[override]]*/ (intrusive_ptr<actor>& other)
void link_to(intrusive_ptr<actor>& other) /*override*/
{
(void) link_to_impl(other);
}
void unlink_from /*[[override]]*/ (intrusive_ptr<actor>& other)
void unlink_from(intrusive_ptr<actor>& other) /*override*/
{
(void) unlink_from_impl(other);
}
bool remove_backlink /*[[override]]*/ (intrusive_ptr<actor>& other)
bool remove_backlink(intrusive_ptr<actor>& other) /*override*/
{
if (other && other != this)
{
......@@ -206,7 +206,7 @@ class abstract_actor /*[[base_check]]*/ : public Base
return false;
}
bool establish_backlink /*[[override]]*/ (intrusive_ptr<actor>& other)
bool establish_backlink(intrusive_ptr<actor>& other) /*override*/
{
bool result = false;
std::uint32_t reason = exit_reason::not_exited;
......
#ifndef ACTOR_COUNT_HPP
#define ACTOR_COUNT_HPP
namespace cppa { namespace detail {
void inc_actor_count();
void dec_actor_count();
// @pre expected <= 1
void actor_count_wait_until(size_t expected);
} } // namespace cppa::detail
#endif // ACTOR_COUNT_HPP
......@@ -28,17 +28,17 @@ class blocking_message_queue : public message_queue
virtual void trap_exit(bool new_value);
virtual void enqueue /*[[override]]*/ (const message& msg);
virtual void enqueue(const message& msg) /*override*/;
virtual const message& dequeue /*[[override]]*/ ();
virtual const message& dequeue() /*override*/;
virtual void dequeue /*[[override]]*/ (invoke_rules& rules);
virtual void dequeue(invoke_rules& rules) /*override*/;
virtual bool try_dequeue /*[[override]]*/ (message& msg);
virtual bool try_dequeue(message& msg) /*override*/;
virtual bool try_dequeue /*[[override]]*/ (invoke_rules& rules);
virtual bool try_dequeue(invoke_rules& rules) /*override*/;
virtual const message& last_dequeued /*[[override]]*/ ();
virtual const message& last_dequeued() /*override*/;
};
......
......@@ -28,12 +28,12 @@ class converted_thread_context : public abstract_actor<context>
public:
message_queue& mailbox /*[[override]]*/ ();
message_queue& mailbox() /*override*/;
// called if the converted thread finished execution
void cleanup(std::uint32_t reason = exit_reason::normal);
void quit /*[[override]]*/ (std::uint32_t reason);
void quit(std::uint32_t reason) /*override*/;
};
......
......@@ -35,7 +35,7 @@ class list_member : public util::abstract_uniform_type_info<List>
for (size_t i = 0; i < ls_size; ++i)
{
primitive_variant val = d->read_value(vptype);
ls.push_back(std::move(get<value_type>(val)));
ls.push_back(std::move(get_ref<value_type>(val)));
}
d->end_sequence();
}
......
#ifndef TASK_SCHEDULER_HPP
#define TASK_SCHEDULER_HPP
#include "cppa/scheduler.hpp"
namespace cppa { namespace detail {
class task_scheduler : public scheduler
{
public:
void await_others_done();
void register_converted_context(context*);
//void unregister_converted_context(context*);
actor_ptr spawn(actor_behavior*, scheduling_hint);
attachable* register_hidden_context();
};
} } // namespace cppa::detail
#endif // TASK_SCHEDULER_HPP
......@@ -10,6 +10,9 @@
namespace cppa {
/**
* @brief A multicast group.
*/
class group : public channel
{
......@@ -28,6 +31,7 @@ class group : public channel
class unsubscriber;
// needs access to unsubscribe()
friend class unsubscriber;
// unsubscribes its channel from the group on destruction
......@@ -39,15 +43,11 @@ class group : public channel
channel_ptr m_self;
intrusive_ptr<group> m_group;
unsubscriber() = delete;
unsubscriber(const unsubscriber&) = delete;
unsubscriber& operator=(const unsubscriber&) = delete;
public:
unsubscriber(const channel_ptr& s, const intrusive_ptr<group>& g);
// matches on group
// matches on m_group
bool matches(const attachable::token& what);
virtual ~unsubscriber();
......@@ -56,25 +56,69 @@ class group : public channel
typedef std::unique_ptr<unsubscriber> subscription;
/**
* @brief Module interface.
*/
class module
{
std::string m_name;
protected:
module(std::string&& module_name);
module(const std::string& module_name);
public:
virtual const std::string& name() = 0;
/**
* @brief Get the name of this module implementation.
* @return The name of this module implementation.
* @threadsafe
*/
const std::string& name();
/**
* @brief Get a pointer to the group associated with
* the name @p group_name.
* @threadsafe
*/
virtual intrusive_ptr<group> get(const std::string& group_name) = 0;
};
/**
* @brief A string representation of the group identifier.
* @return The group identifier as string (e.g. "224.0.0.1" for IPv4
* multicast or a user-defined string for local groups).
*/
const std::string& identifier() const;
/**
* @brief The name of the module.
* @return The module name of this group (e.g. "local").
*/
const std::string& module_name() const;
/**
* @brief Subscribe @p who to this group.
* @return A {@link subscription} object that unsubscribes @p who
* if the lifetime of @p who ends.
*/
virtual subscription subscribe(const channel_ptr& who) = 0;
/**
* @brief Get a pointer to the group associated with
* @p group_identifier from the module @p module_name.
* @threadsafe
*/
static intrusive_ptr<group> get(const std::string& module_name,
const std::string& group_name);
const std::string& group_identifier);
/**
* @brief Add a new group module to the libcppa group management.
* @threadsafe
*/
static void add_module(module*);
};
......
......@@ -61,9 +61,9 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, const T*>,
}
}
T* get() { return m_ptr; }
inline T* get() { return m_ptr; }
const T* get() const { return m_ptr; }
inline const T* get() const { return m_ptr; }
T* take()
{
......@@ -123,13 +123,13 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, const T*>,
return *this;
}
T* operator->() { return m_ptr; }
inline T* operator->() { return m_ptr; }
T& operator*() { return *m_ptr; }
inline T& operator*() { return *m_ptr; }
const T* operator->() const { return m_ptr; }
inline const T* operator->() const { return m_ptr; }
const T& operator*() const { return *m_ptr; }
inline const T& operator*() const { return *m_ptr; }
inline explicit operator bool() const { return m_ptr != nullptr; }
......@@ -166,9 +166,9 @@ bool operator==(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
}
template<typename X, typename Y>
bool operator!=(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
inline bool operator!=(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
{
return lhs.get() != rhs.get();
return !(lhs == rhs);
}
} // namespace cppa
......
......@@ -244,6 +244,20 @@ const typename detail::ptype_to_type<PT>::type& get(const primitive_variant& pv)
return pv.get_as<PT>();
}
template<typename T>
T& get_ref(primitive_variant& pv)
{
static const primitive_type ptype = detail::type_to_ptype<T>::ptype;
return pv.get_as<ptype>();
}
template<primitive_type PT>
typename detail::ptype_to_type<PT>::type& get_ref(primitive_variant& pv)
{
static_assert(PT != pt_null, "PT == pt_null");
return pv.get_as<PT>();
}
template<typename T>
typename util::enable_if<util::is_primitive<T>, bool>::type
operator==(const T& lhs, const primitive_variant& rhs)
......
......@@ -39,6 +39,7 @@ class scheduler
/**
* @brief Informs the scheduler about a converted context
* (a thread that acts as actor).
* @note Calls <tt>what->attach(...)</tt>.
*/
virtual void register_converted_context(context* what) = 0;
......@@ -61,7 +62,7 @@ class scheduler
/**
* @brief Gets the actual used scheduler implementation.
* @return The active scheduler or @c nullptr.
* @return The active scheduler (default constructed).
*/
scheduler* get_scheduler();
......
......@@ -8,7 +8,6 @@
#include "cppa/util/compare_tuples.hpp"
#include "cppa/util/concat_type_lists.hpp"
#include "cppa/util/conjunction.hpp"
#include "cppa/util/detach.hpp"
#include "cppa/util/disjunction.hpp"
#include "cppa/util/enable_if.hpp"
#include "cppa/util/eval_first_n.hpp"
......
#ifndef DETACH_HPP
#define DETACH_HPP
#include "cppa/util/is_copyable.hpp"
#include "cppa/util/has_copy_member_fun.hpp"
namespace cppa { namespace detail {
template<bool IsCopyable, bool HasCpyMemFun>
struct detach_;
template<bool HasCpyMemFun>
struct detach_<true, HasCpyMemFun>
{
template<typename T>
inline static T* cpy(const T* what)
{
return new T(*what);
}
};
template<>
struct detach_<false, true>
{
template<typename T>
inline static T* cpy(const T* what)
{
return what->copy();
}
};
} } // namespace cppa::detail
namespace cppa { namespace util {
template<typename T>
inline T* detach(const T* what)
{
return detail::detach_<is_copyable<T>::value,
has_copy_member_fun<T>::value>::cpy(what);
}
} } // namespace cppa::util
#endif // DETACH_HPP
......@@ -43,13 +43,8 @@ class is_copyable_<true, T>
namespace cppa { namespace util {
template<typename T>
struct is_copyable
struct is_copyable : std::integral_constant<bool, detail::is_copyable_<std::is_abstract<T>::value, T>::value>
{
public:
static const bool value = detail::is_copyable_<std::is_abstract<T>::value, T>::value;
};
} } // namespace cppa::util
......
#include <limits>
#include <atomic>
#include <stdexcept>
#include <boost/thread.hpp>
#include "cppa/detail/actor_count.hpp"
namespace {
boost::mutex s_ra_mtx;
boost::condition_variable s_ra_cv;
std::atomic<size_t> s_running_actors(0);
typedef boost::unique_lock<boost::mutex> guard_type;
} // namespace <anonymous>
namespace cppa { namespace detail {
void inc_actor_count()
{
++s_running_actors;
}
void dec_actor_count()
{
size_t new_val = --s_running_actors;
if (new_val == std::numeric_limits<size_t>::max())
{
throw std::underflow_error("dec_actor_count()");
}
else if (new_val <= 1)
{
guard_type guard(s_ra_mtx);
s_ra_cv.notify_all();
}
}
void actor_count_wait_until(size_t expected)
{
guard_type lock(s_ra_mtx);
while (s_running_actors.load() != expected)
{
s_ra_cv.wait(lock);
}
}
} } // namespace cppa::detail
......@@ -18,19 +18,20 @@ 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;
typedef util::shared_spinlock shared_mutex_type;
typedef std::lock_guard<shared_mutex_type> exclusive_guard;
typedef util::shared_lock_guard<shared_mutex_type> shared_guard;
typedef util::upgrade_lock_guard<shared_mutex_type> upgrade_guard;
std::mutex s_mtx;
modules_map s_mmap;
std::mutex s_mmap_mtx;
class local_group : public group
{
friend class local_group_module;
util::shared_spinlock m_mtx;
shared_mutex_type m_shared_mtx;
std::set<channel_ptr> m_subscribers;
// allow access to local_group_module only
......@@ -42,7 +43,7 @@ class local_group : public group
virtual void enqueue(const message& msg)
{
shared_guard guard(m_mtx);
shared_guard guard(m_shared_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
{
const_cast<channel_ptr&>(*i)->enqueue(msg);
......@@ -52,7 +53,7 @@ class local_group : public group
virtual group::subscription subscribe(const channel_ptr& who)
{
group::subscription result;
exclusive_guard guard(m_mtx);
exclusive_guard guard(m_shared_mtx);
if (m_subscribers.insert(who).second)
{
result.reset(new group::unsubscriber(who, this));
......@@ -62,7 +63,7 @@ class local_group : public group
virtual void unsubscribe(const channel_ptr& who)
{
exclusive_guard guard(m_mtx);
exclusive_guard guard(m_shared_mtx);
m_subscribers.erase(who);
}
......@@ -71,20 +72,16 @@ class local_group : public group
class local_group_module : public group::module
{
std::string m_name;
typedef group::module super;
util::shared_spinlock m_mtx;
std::map<std::string, group_ptr> m_instances;
public:
local_group_module() : m_name("local")
{
}
const std::string& name()
local_group_module() : super("local")
{
return m_name;
}
group_ptr get(const std::string& group_name)
......@@ -110,6 +107,7 @@ class local_group_module : public group::module
};
// @pre: s_mmap_mtx is locked
modules_map& mmap()
{
if (s_mmap.empty())
......@@ -128,7 +126,7 @@ intrusive_ptr<group> group::get(const std::string& module_name,
{
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(s_mtx);
std::lock_guard<std::mutex> guard(s_mmap_mtx);
auto& mmref = mmap();
auto i = mmref.find(module_name);
if (i != mmref.end())
......@@ -148,7 +146,7 @@ void group::add_module(group::module* ptr)
std::unique_ptr<group::module> mptr(ptr);
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(s_mtx);
std::lock_guard<std::mutex> guard(s_mmap_mtx);
if (mmap().insert(std::make_pair(mname, std::move(mptr))).second)
{
return; // success; don't throw exception
......@@ -172,6 +170,19 @@ group::unsubscriber::~unsubscriber()
if (m_group) m_group->unsubscribe(m_self);
}
group::module::module(const std::string& name) : m_name(name)
{
}
group::module::module(std::string&& name) : m_name(std::move(name))
{
}
const std::string& group::module::name()
{
return m_name;
}
bool group::unsubscriber::matches(const attachable::token& what)
{
if (what.subtype == typeid(group::unsubscriber))
......
......@@ -12,6 +12,7 @@
#include "cppa/attachable.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/converted_thread_context.hpp"
......@@ -21,26 +22,6 @@ using std::endl;
namespace {
boost::mutex s_ra_mtx;
boost::condition_variable s_ra_cv;
std::atomic<int> s_running_actors(0);
typedef boost::unique_lock<boost::mutex> guard_type;
void inc_actor_count()
{
++s_running_actors;
}
void dec_actor_count()
{
if (--s_running_actors <= 1)
{
guard_type guard(s_ra_mtx);
s_ra_cv.notify_all();
}
}
void run_actor(cppa::intrusive_ptr<cppa::context> m_self,
cppa::actor_behavior* behavior)
{
......@@ -53,14 +34,14 @@ void run_actor(cppa::intrusive_ptr<cppa::context> m_self,
catch (...) { }
delete behavior;
}
dec_actor_count();
cppa::detail::dec_actor_count();
}
struct exit_observer : cppa::attachable
{
~exit_observer()
{
dec_actor_count();
cppa::detail::dec_actor_count();
}
};
......@@ -93,12 +74,7 @@ attachable* mock_scheduler::register_hidden_context()
void mock_scheduler::await_others_done()
{
auto expected = (unchecked_self() == nullptr) ? 0 : 1;
guard_type lock(s_ra_mtx);
while (s_running_actors.load() != expected)
{
s_ra_cv.wait(lock);
}
actor_count_wait_until((unchecked_self() == nullptr) ? 0 : 1);
}
} } // namespace detail
#include <new> // placement new
#include <ios> // ios_base::failure
#include <list> // std::list
#include <cstdint> // std::uint32_t
#include <vector> // std::vector
#include <cstring> // strerror
#include <cstdint> // std::uint32_t, std::uint64_t
#include <iostream> // std::cout, std::endl
#include <exception> // std::logic_error
#include <algorithm> // std::find_if
#include <stdexcept> // std::underflow_error
#include <cstdio>
#include <fcntl.h>
......@@ -166,54 +169,21 @@ struct post_office_manager
typedef util::single_reader_queue<post_office_msg> queue_t;
typedef util::single_reader_queue<mailman_job> mailman_queue_t;
// m_pipe[0] is for reading, m_pipe[1] is for writing
int m_pipe[2];
mailman_queue_t m_mailman_queue;
queue_t m_queue;
boost::thread* m_loop;
int m_pipe[2]; // m_pipe[0]: read; m_pipe[1]: write
queue_t m_queue; // post office queue
boost::thread* m_loop; // post office thread
mailman_queue_t m_mailman_queue; // mailman queue
post_office_manager()
{
//if (socketpair(AF_UNIX, SOCK_STREAM, 0, m_pipe) != 0)
if (pipe(m_pipe) != 0)
{
switch (errno)
{
case EFAULT:
{
throw std::logic_error("EFAULT: invalid pipe() argument");
}
case EMFILE:
{
throw std::logic_error("EMFILE: Too many file "
"descriptors in use");
}
case ENFILE:
{
throw std::logic_error("The system limit on the total "
"number of open files "
"has been reached");
}
default:
{
throw std::logic_error("unknown error");
}
}
char* error_cstr = strerror(errno);
std::string error_str = "pipe(): ";
error_str += error_cstr;
free(error_cstr);
throw std::logic_error(error_str);
}
/*
int flags = fcntl(m_pipe[0], F_GETFL, 0);
if (flags == -1 || fcntl(m_pipe[0], F_SETFL, flags | O_NONBLOCK) == -1)
{
throw std::logic_error("unable to set pipe to nonblocking");
}
*/
/*
int flags = fcntl(m_pipe[0], F_GETFL, 0);
if (flags == -1 || fcntl(m_pipe[0], F_SETFL, flags | O_ASYNC) == -1)
{
throw std::logic_error("unable to set pipe to O_ASYNC");
}
*/
m_loop = new boost::thread(post_office_loop, m_pipe[0]);
}
......@@ -251,59 +221,6 @@ util::single_reader_queue<mailman_job>& mailman_queue()
namespace cppa { namespace detail { namespace {
class remote_observer : public attachable
{
process_information_ptr peer;
public:
remote_observer(const process_information_ptr& piptr) : peer(piptr)
{
}
void detach(std::uint32_t reason)
{
actor_ptr self_ptr = self();
message msg(self_ptr, self_ptr, atom(":KillProxy"), reason);
detail::mailman_queue().push_back(new detail::mailman_job(peer, msg));
}
};
void handle_message(const message& msg,
const std::type_info& atom_tinfo,
const process_information& pself,
const process_information_ptr& peer)
{
if ( msg.content().size() == 1
&& msg.content().utype_info_at(0) == atom_tinfo
&& *reinterpret_cast<const atom_value*>(msg.content().at(0))
== atom(":Monitor"))
{
DEBUG("<-- :Monitor");
actor_ptr sender = msg.sender();
if (sender->parent_process() == pself)
{
//cout << pinfo << " ':Monitor'; actor id = "
// << sender->id() << endl;
// local actor?
// this message was send from a proxy
sender->attach(new remote_observer(peer));
}
else
{
DEBUG(":Monitor received for an remote actor");
}
}
else
{
DEBUG("<-- " << to_string(msg));
auto r = msg.receiver();
if (r) r->enqueue(msg);
}
}
class post_office_worker
{
......@@ -314,10 +231,14 @@ class post_office_worker
native_socket_t m_socket;
// caches process_information::get()
const process_information& m_pself;
post_office_worker(native_socket_t fd, native_socket_t parent_fd = -1)
: m_rc((parent_fd != -1) ? 1 : 0)
, m_parent(parent_fd)
, m_socket(fd)
, m_pself(process_information::get())
{
}
......@@ -325,6 +246,7 @@ class post_office_worker
: m_rc(other.m_rc)
, m_parent(other.m_parent)
, m_socket(other.m_socket)
, m_pself(process_information::get())
{
other.m_rc = 0;
other.m_socket = -1;
......@@ -345,7 +267,10 @@ class post_office_worker
inline size_t dec_ref_count()
{
if (m_rc == 0) throw std::runtime_error("dec_ref_count(): underflow");
if (m_rc == 0)
{
throw std::underflow_error("post_office_worker::dec_ref_count()");
}
return --m_rc;
}
......@@ -406,6 +331,8 @@ class po_peer : public post_office_worker
buffer<s_chunk_size, s_max_buffer_size> m_rdbuf;
std::list<actor_proxy_ptr> m_children;
const uniform_type_info* m_meta_msg;
public:
explicit po_peer(add_peer_msg& from)
......@@ -413,12 +340,14 @@ class po_peer : public post_office_worker
, m_state(wait_for_msg_size)
, m_peer(std::move(from.peer))
, m_observer(std::move(from.attachable_ptr))
, m_meta_msg(uniform_typeid<message>())
{
}
explicit po_peer(native_socket_t sockfd, native_socket_t parent_socket)
: super(sockfd, parent_socket)
, m_state(wait_for_process_info)
, m_meta_msg(uniform_typeid<message>())
{
m_rdbuf.reset( sizeof(std::uint32_t)
+ process_information::node_id_size);
......@@ -431,6 +360,7 @@ class po_peer : public post_office_worker
, m_observer(std::move(other.m_observer))
, m_rdbuf(std::move(other.m_rdbuf))
, m_children(std::move(other.m_children))
, m_meta_msg(uniform_typeid<message>())
{
}
......@@ -452,8 +382,7 @@ class po_peer : public post_office_worker
}
// @return false if an error occured; otherwise true
bool read_and_continue(const uniform_type_info* meta_msg,
const process_information& pself)
bool read_and_continue()
{
switch (m_state)
{
......@@ -508,24 +437,62 @@ class po_peer : public post_office_worker
}
case read_message:
{
if (!m_rdbuf.append_from(m_socket, s_rdflag)) return false;
if (m_rdbuf.ready())
if (!m_rdbuf.append_from(m_socket, s_rdflag))
{
// could not read from socket
return false;
}
if (m_rdbuf.ready() == false)
{
// wait for new data
break;
}
message msg;
binary_deserializer bd(m_rdbuf.data(), m_rdbuf.size());
try
{
m_meta_msg->deserialize(&msg, &bd);
}
catch (std::exception& e)
{
message msg;
binary_deserializer bd(m_rdbuf.data(), m_rdbuf.size());
try
// unable to deserialize message (format error)
DEBUG(to_uniform_name(typeid(e)) << ": " << e.what());
return false;
}
auto& content = msg.content();
if ( content.size() == 1
&& content.utype_info_at(0) == typeid(atom_value)
&& *reinterpret_cast<const atom_value*>(content.at(0))
== atom(":Monitor"))
{
actor_ptr sender = msg.sender();
if (sender->parent_process() == process_information::get())
{
meta_msg->deserialize(&msg, &bd);
//cout << pinfo << " ':Monitor'; actor id = "
// << sender->id() << endl;
// local actor?
// this message was send from a proxy
sender->attach_functor([=](std::uint32_t reason)
{
message msg(sender, sender,
atom(":KillProxy"), reason);
auto mjob = new detail::mailman_job(m_peer, msg);
detail::mailman_queue().push_back(mjob);
});
}
catch (std::exception& e)
else
{
DEBUG(to_uniform_name(typeid(e)) << ": " << e.what());
return false;
DEBUG(":Monitor received for an remote actor");
}
handle_message(msg, typeid(atom_value), pself, m_peer);
m_rdbuf.reset();
m_state = wait_for_msg_size;
}
else
{
DEBUG("<-- " << to_string(msg));
auto r = msg.receiver();
if (r) r->enqueue(msg);
}
m_rdbuf.reset();
m_state = wait_for_msg_size;
break;
}
default:
......@@ -546,23 +513,26 @@ class po_doorman : public post_office_worker
// server socket
actor_ptr published_actor;
std::list<po_peer>* m_peers;
public:
explicit po_doorman(add_server_socket_msg& assm)
explicit po_doorman(add_server_socket_msg& assm, std::list<po_peer>* peers)
: super(assm.server_sockfd)
, published_actor(assm.published_actor)
, m_peers(peers)
{
}
po_doorman(po_doorman&& other)
: super(std::move(other))
, published_actor(std::move(other.published_actor))
, m_peers(other.m_peers)
{
}
// @return false if an error occured; otherwise true
bool read_and_continue(const process_information& pself,
std::list<po_peer>& peers)
bool read_and_continue()
{
sockaddr addr;
socklen_t addrlen;
......@@ -586,37 +556,33 @@ class po_doorman : public post_office_worker
}
auto id = published_actor->id();
::send(sfd, &id, sizeof(std::uint32_t), 0);
::send(sfd, &(pself.process_id), sizeof(std::uint32_t), 0);
::send(sfd, pself.node_id.data(), pself.node_id.size(), 0);
peers.push_back(po_peer(sfd, m_socket));
::send(sfd, &(m_pself.process_id), sizeof(std::uint32_t), 0);
::send(sfd, m_pself.node_id.data(), m_pself.node_id.size(), 0);
m_peers->push_back(po_peer(sfd, m_socket));
DEBUG("socket accepted; published actor: " << id);
return true;
}
};
// starts and stops mailman_loop
struct mailman_worker
{
boost::thread m_thread;
mailman_worker() : m_thread(mailman_loop)
{
}
~mailman_worker()
{
mailman_queue().push_back(mailman_job::kill_job());
m_thread.join();
}
};
void post_office_loop(int pipe_read_handle)
{
// starts and stops mailman
mailman_worker mworker;
// map of all published actors
// map of all published actors (actor_id => list<doorman>)
std::map<std::uint32_t, std::list<po_doorman> > doormen;
// list of all connected peers
std::list<po_peer> peers;
......@@ -624,9 +590,6 @@ void post_office_loop(int pipe_read_handle)
fd_set readset;
// maximum number of all socket descriptors
int maxfd = 0;
// cache some used global data
auto meta_msg = uniform_typeid<message>();
auto& pself = process_information::get();
// initialize variables
FD_ZERO(&readset);
maxfd = pipe_read_handle;
......@@ -676,71 +639,32 @@ void post_office_loop(int pipe_read_handle)
});
for (;;)
{
//std::cout << __LINE__ << std::endl;
if (select(maxfd + 1, &readset, nullptr, nullptr, nullptr) < 0)
{
// must not happen
perror("select()");
exit(3);
}
//std::cout << __LINE__ << std::endl;
// iterate over all peers; lifetime scope of i, end
// iterate over all peers and remove peers on error
peers.remove_if([&](po_peer& peer) -> bool
{
auto i = peers.begin();
auto end = peers.end();
while (i != end)
if (FD_ISSET(peer.get_socket(), &readset))
{
if (FD_ISSET(i->get_socket(), &readset))
{
selected_peer = &(*i);
//DEBUG("read message from peer");
if (i->read_and_continue(meta_msg, pself))
{
// no errors detected; next iteration
++i;
}
else
{
// peer detected an error; erase from list
DEBUG("connection to peer lost");
i = peers.erase(i);
}
}
else
{
// next iteration
++i;
}
selected_peer = &peer;
return peer.read_and_continue() == false;
}
}
return false;
});
selected_peer = nullptr;
// new connections to accept?
// iterate over all doormen (accept new connections)
for (auto& kvp : doormen)
{
auto& list = kvp.second;
auto i = list.begin();
auto end = list.end();
while (i != end)
// iterate over all doormen and remove doormen on error
kvp.second.remove_if([&](po_doorman& doorman) -> bool
{
if (FD_ISSET(i->get_socket(), &readset))
{
DEBUG("accept new socket...");
if (i->read_and_continue(pself, peers))
{
DEBUG("ok");
++i;
}
else
{
DEBUG("failed; erased doorman");
i = list.erase(i);
}
}
else
{
++i;
}
}
return FD_ISSET(doorman.get_socket(), &readset)
&& doorman.read_and_continue() == false;
});
}
// read events from pipe
if (FD_ISSET(pipe_read_handle, &readset))
......@@ -774,23 +698,26 @@ void post_office_loop(int pipe_read_handle)
{
auto& assm = pom->as_add_server_socket_msg();
auto& pactor = assm.published_actor;
if (!pactor)
if (pactor)
{
throw std::logic_error("nullptr published");
auto actor_id = pactor->id();
auto callback = [actor_id](std::uint32_t)
{
DEBUG("call post_office_unpublish() ...");
post_office_unpublish(actor_id);
};
if (pactor->attach_functor(std::move(callback)))
{
auto& dm = doormen[actor_id];
dm.push_back(po_doorman(assm, &peers));
DEBUG("new doorman");
}
// else: actor already exited!
}
auto actor_id = pactor->id();
auto callback = [actor_id](std::uint32_t)
else
{
DEBUG("call post_office_unpublish() ...");
post_office_unpublish(actor_id);
};
if (pactor->attach_functor(std::move(callback)))
{
auto& dm = doormen[actor_id];
dm.push_back(po_doorman(assm));
DEBUG("new doorman");
DEBUG("nullptr published");
}
// else: actor already exited!
}
delete pom;
break;
......@@ -801,22 +728,14 @@ void post_office_loop(int pipe_read_handle)
auto kvp = doormen.find(pmsg[1]);
if (kvp != doormen.end())
{
// decrease ref count of all children of this doorman
for (po_doorman& dm : kvp->second)
{
auto i = peers.begin();
auto end = peers.end();
while (i != end)
auto parent_fd = dm.get_socket();
peers.remove_if([parent_fd](po_peer& ppeer) -> bool
{
if (i->parent_exited(dm.get_socket()) == 0)
{
DEBUG("socket closed; parent exited");
i = peers.erase(i);
}
else
{
++i;
}
}
return ppeer.parent_exited(parent_fd) == 0;
});
}
doormen.erase(kvp);
}
......@@ -824,25 +743,19 @@ void post_office_loop(int pipe_read_handle)
}
case dec_socket_ref_event:
{
auto sockfd = static_cast<native_socket_t>(pmsg[1]);
release_socket(sockfd);
release_socket(static_cast<native_socket_t>(pmsg[1]));
break;
}
case close_socket_event:
{
auto sockfd = static_cast<native_socket_t>(pmsg[1]);
auto i = peers.begin();
auto end = peers.end();
while (i != end)
auto i = std::find_if(peers.begin(), end,
[sockfd](po_peer& peer) -> bool
{
if (i->get_socket() == sockfd)
{
// exit loop
peers.erase(i);
i = end;
}
else ++i;
}
return peer.get_socket() == sockfd;
});
if (i != end) peers.erase(i);
break;
}
case shutdown_event:
......@@ -864,9 +777,9 @@ void post_office_loop(int pipe_read_handle)
{
release_socket(sockfd);
}
released_socks.clear();
}
// recalculate readset if needed
//DEBUG("recalculate readset");
// recalculate readset
FD_ZERO(&readset);
FD_SET(pipe_read_handle, &readset);
maxfd = pipe_read_handle;
......@@ -876,10 +789,10 @@ void post_office_loop(int pipe_read_handle)
if (fd > maxfd) maxfd = fd;
FD_SET(fd, &readset);
}
// iterate over key value pairs
// iterate over key (actor id) value (doormen) pairs
for (auto& kvp : doormen)
{
// iterate over value (= list of doormen)
// iterate over values (doormen)
for (auto& dm : kvp.second)
{
auto fd = dm.get_socket();
......@@ -900,7 +813,7 @@ void post_office_add_peer(native_socket_t a0,
std::unique_ptr<attachable>&& a3)
{
s_po_manager.m_queue.push_back(new post_office_msg(a0, a1, a2,
std::move(a3)));
std::move(a3)));
pipe_msg msg = { rd_queue_event, 0 };
write(s_po_manager.write_handle(), msg, pipe_msg_size);
}
......
#include "cppa/detail/task_scheduler.hpp"
namespace cppa { namespace detail {
void task_scheduler::await_others_done()
{
}
void task_scheduler::register_converted_context(context*)
{
}
actor_ptr task_scheduler::spawn(actor_behavior*, scheduling_hint)
{
return nullptr;
}
attachable* task_scheduler::register_hidden_context()
{
return nullptr;
}
} } // namespace cppa::detail
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