Commit 82939f63 authored by neverlord's avatar neverlord

unicast network

parent 178334ac
......@@ -162,5 +162,14 @@ src/process_information.cpp
cppa/util/ripemd_160.hpp
src/ripemd_160.cpp
src/unicast_network.cpp
cppa/actor_exited.hpp
src/actor_exited.cpp
unit_testing/test__remote_actor.cpp
unit_testing/ping_pong.hpp
unit_testing/ping_pong.cpp
cppa/exception.hpp
src/exception.cpp
cppa/actor_proxy.hpp
src/actor_proxy.cpp
cppa/detail/actor_proxy_cache.hpp
src/actor_proxy_cache.cpp
src/attachable.cpp
cppa/attachable.hpp
......@@ -5,11 +5,13 @@
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/attachable.hpp"
#include "cppa/process_information.hpp"
namespace cppa {
class serializer;
class actor_proxy;
class deserializer;
/**
......@@ -18,8 +20,13 @@ class deserializer;
class actor : public channel
{
friend class actor_proxy;
bool m_is_proxy;
std::uint32_t m_id;
actor(std::uint32_t aid);
protected:
actor();
......@@ -28,6 +35,18 @@ class actor : public channel
~actor();
/**
* @brief Attaches @p ptr to this actor.
*
* @p ptr will be deleted if actor finished execution of immediately
* if the actor already exited.
*
* @return @c true if @p ptr was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*
*/
virtual bool attach(attachable* ptr) = 0;
/**
* @brief
*/
......@@ -41,12 +60,12 @@ class actor : public channel
/**
* @brief
*/
virtual void link(intrusive_ptr<actor>& other) = 0;
virtual void link_to(intrusive_ptr<actor>& other) = 0;
/**
* @brief
*/
virtual void unlink(intrusive_ptr<actor>& other) = 0;
virtual void unlink_from(intrusive_ptr<actor>& other) = 0;
/**
* @brief
......
#ifndef ACTOR_EXITED_HPP
#define ACTOR_EXITED_HPP
#include <string>
#include <cstdint>
#include <exception>
namespace cppa {
class actor_exited : public std::exception
{
std::uint32_t m_reason;
std::string m_what;
public:
~actor_exited() throw();
actor_exited(std::uint32_t exit_reason);
std::uint32_t reason() const;
virtual const char* what() const throw();
};
} // namespace cppa
#endif // ACTOR_EXITED_HPP
#ifndef ACTOR_PROXY_HPP
#define ACTOR_PROXY_HPP
#include "cppa/actor.hpp"
namespace cppa {
class actor_proxy : public actor
{
process_information_ptr m_parent;
public:
actor_proxy(std::uint32_t mid, process_information_ptr&& parent);
actor_proxy(std::uint32_t mid, const process_information_ptr& parent);
bool attach(attachable* ptr);
void enqueue(const message& msg);
void join(group_ptr& what);
void leave(const group_ptr& what);
void link_to(intrusive_ptr<actor>& other);
void unlink_from(intrusive_ptr<actor>& other);
bool remove_backlink(const intrusive_ptr<actor>& to);
bool establish_backlink(const intrusive_ptr<actor>& to);
const process_information& parent_process() const;
process_information_ptr parent_process_ptr() const;
};
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
} // namespace cppa
#endif // ACTOR_PROXY_HPP
#ifndef ATTACHABLE_HPP
#define ATTACHABLE_HPP
#include <cstdint>
namespace cppa {
// utilty class;
// provides a virtual destructor that's executed after an actor
// finished execution
/**
* @brief Callback utility class.
*/
class attachable
{
protected:
attachable() = default;
public:
virtual ~attachable();
/**
* @brief Executed if the actor finished execution
* with given @p reason.
*
* The default implementation does nothing.
*/
virtual void detach(std::uint32_t reason);
};
} // namespace cppa
#endif // ATTACHABLE_HPP
......@@ -25,11 +25,6 @@ class context : public actor
*/
virtual void quit(std::uint32_t reason) = 0;
inline void quit(exit_reason reason)
{
quit(static_cast<std::uint32_t>(reason));
}
/**
* @brief
*/
......
......@@ -33,6 +33,7 @@
#include <cstdint>
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/tuple.hpp"
#include "cppa/actor.hpp"
#include "cppa/invoke.hpp"
......@@ -58,7 +59,7 @@ namespace cppa {
*/
inline void link(actor_ptr& other)
{
self()->link(other);
self()->link_to(other);
}
/**
......@@ -66,7 +67,7 @@ inline void link(actor_ptr& other)
*/
inline void link(actor_ptr&& other)
{
self()->link(other);
self()->link_to(other);
}
inline void trap_exit(bool new_value)
......@@ -103,14 +104,6 @@ inline void quit(std::uint32_t reason)
self()->quit(reason);
}
/**
* @brief Quits execution of the calling actor.
*/
inline void quit(exit_reason reason)
{
self()->quit(reason);
}
/**
* @brief Dequeues the next message from the mailbox.
*/
......@@ -220,7 +213,9 @@ inline void await_all_others_done()
/**
* @brief Publishes @p whom at given @p port.
*/
void publish(const actor_ptr& whom, std::uint16_t port);
void publish(actor_ptr& whom, std::uint16_t port);
void publish(actor_ptr&& whom, std::uint16_t port);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
......
#ifndef ACTOR_PROXY_CACHE_HPP
#define ACTOR_PROXY_CACHE_HPP
#include <string>
#include "cppa/actor_proxy.hpp"
#include "cppa/process_information.hpp"
namespace cppa { namespace detail {
class actor_proxy_cache
{
public:
typedef std::tuple<std::uint32_t, std::uint32_t,
process_information::node_id_type> key_tuple;
private:
std::map<key_tuple, process_information_ptr> m_pinfos;
std::map<key_tuple, actor_proxy_ptr> m_proxies;
process_information_ptr get_pinfo(const key_tuple& key);
public:
actor_proxy_ptr get(const key_tuple& key);
void add(const actor_proxy_ptr& pptr);
};
// get the thread-local cache object
actor_proxy_cache& get_actor_proxy_cache();
} } // namespace cppa::detail
#endif // ACTOR_PROXY_CACHE_HPP
......@@ -6,6 +6,8 @@
#include <map>
#include <list>
#include <mutex>
#include <vector>
#include <memory>
#include "cppa/context.hpp"
#include "cppa/detail/blocking_message_queue.hpp"
......@@ -16,7 +18,7 @@ class converted_thread_context : public context
{
// true if the associated thread has finished execution
bool m_exited;
std::uint32_t m_exit_reason;
// mailbox implementation
detail::blocking_message_queue m_mailbox;
......@@ -30,21 +32,34 @@ class converted_thread_context : public context
// manages actor links
std::list<actor_ptr> m_links;
std::vector<std::unique_ptr<attachable>> m_attachables;
// @pre m_mtx is locked
inline bool exited() const
{
return m_exit_reason != exit_reason::not_exited;
}
public:
converted_thread_context();
message_queue& mailbox /*[[override]]*/ ();
// called if the converted thread finished execution
void cleanup(std::uint32_t reason = exit_reason::normal);
bool attach /*[[override]]*/ (attachable* ptr);
void join /*[[override]]*/ (group_ptr& what);
void quit /*[[override]]*/ (std::uint32_t reason);
void leave /*[[override]]*/ (const group_ptr& what);
void link /*[[override]]*/ (intrusive_ptr<actor>& other);
void link_to /*[[override]]*/ (intrusive_ptr<actor>& other);
void unlink /*[[override]]*/ (intrusive_ptr<actor>& other);
void unlink_from /*[[override]]*/ (intrusive_ptr<actor>& other);
bool remove_backlink /*[[override]]*/ (const intrusive_ptr<actor>& to);
......
#ifndef EXCEPTION_HPP
#define EXCEPTION_HPP
#include <string>
#include <cstdint>
#include <exception>
namespace cppa {
class exception : public std::exception
{
std::string m_what;
protected:
exception(std::string&& what_str);
exception(const std::string& what_str);
public:
~exception() throw();
const char* what() const throw();
};
class actor_exited : public exception
{
std::uint32_t m_reason;
public:
actor_exited(std::uint32_t exit_reason);
std::uint32_t reason() const throw();
};
class network_exception : public exception
{
public:
network_exception(std::string&& what_str);
network_exception(const std::string& what_str);
};
class bind_failure : public network_exception
{
int m_errno;
public:
bind_failure(int bind_errno);
int error_code() const throw();
};
} // namespace cppa
#endif // EXCEPTION_HPP
......@@ -5,45 +5,39 @@
namespace cppa {
enum class exit_reason : std::uint32_t
namespace exit_reason
{
/**
* @brief Indicates that the actor is still alive.
*/
static constexpr std::uint32_t not_exited = 0x00000;
/**
* @brief Indicates that an actor finished execution.
*/
normal = 0x00000,
static constexpr std::uint32_t normal = 0x00001;
/**
* @brief Indicates that an actor finished execution
* because of an unhandled exception.
*/
unhandled_exception = 0x00001,
static constexpr std::uint32_t unhandled_exception = 0x00002;
/**
* @brief Indicates that an actor finishied execution
* because a connection to a remote link was
* closed unexpectedly.
*/
remote_link_unreachable = 0x00101,
static constexpr std::uint32_t remote_link_unreachable = 0x00101;
/**
* @brief Any user defined exit reason should have a
* value greater or equal to prevent collisions
* with default defined exit reasons.
*/
user_defined = 0x10000
};
static constexpr std::uint32_t user_defined = 0x10000;
static constexpr std::uint32_t user_defined_exit_reason
= static_cast<std::uint32_t>(exit_reason::user_defined);
/**
* @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
......@@ -59,13 +53,6 @@ class exit_signal
*/
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>.
......@@ -81,19 +68,11 @@ class exit_signal
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);
};
inline bool operator==(const exit_signal& lhs, const exit_signal& rhs)
......
#ifndef PROCESS_INFORMATION_HPP
#define PROCESS_INFORMATION_HPP
#include <array>
#include <string>
#include <cstdint>
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
......@@ -20,8 +22,12 @@ class process_information : public ref_counted,
static constexpr size_t node_id_size = 20;
typedef std::array<std::uint8_t, node_id_size> node_id_type;
process_information();
process_information(std::uint32_t process_id, const node_id_type& node_id);
process_information(const process_information& other);
process_information& operator=(const process_information& other);
......@@ -37,7 +43,8 @@ class process_information : public ref_counted,
* A hash build from the MAC address of the first network device
* and the serial number from the root HD (mounted in "/" or "C:").
*/
std::uint8_t node_id[node_id_size];
node_id_type node_id;
//std::uint8_t node_id[node_id_size];
/**
* @brief Converts {@link node_id} to an hexadecimal string.
......@@ -50,11 +57,15 @@ class process_information : public ref_counted,
*/
static const process_information& get();
static void node_id_from_string(const std::string& str, node_id_type& storage);
// "inherited" from comparable<process_information>
int compare(const process_information& other) const;
};
typedef intrusive_ptr<process_information> process_information_ptr;
} // namespace cppa
#endif // PROCESS_INFORMATION_HPP
......@@ -71,6 +71,9 @@ class single_reader_queue
}
}
/**
* @reentrant
*/
void push_back(element_type* new_element)
{
element_type* e = m_tail.load();
......
......@@ -18,7 +18,11 @@ cppa::util::shared_spinlock s_instances_mtx;
namespace cppa {
actor::actor() : m_id(s_ids.fetch_add(1))
actor::actor(std::uint32_t aid) : m_is_proxy(true), m_id(aid)
{
}
actor::actor() : m_is_proxy(false), m_id(s_ids.fetch_add(1))
{
std::lock_guard<util::shared_spinlock> guard(s_instances_mtx);
s_instances.insert(std::make_pair(m_id, this));
......@@ -26,8 +30,11 @@ actor::actor() : m_id(s_ids.fetch_add(1))
actor::~actor()
{
std::lock_guard<util::shared_spinlock> guard(s_instances_mtx);
s_instances.erase(m_id);
if (!m_is_proxy)
{
std::lock_guard<util::shared_spinlock> guard(s_instances_mtx);
s_instances.erase(m_id);
}
}
intrusive_ptr<actor> actor::by_id(std::uint32_t actor_id)
......
#include <sstream>
#include "cppa/actor_exited.hpp"
namespace cppa {
actor_exited::~actor_exited() throw()
{
}
actor_exited::actor_exited(std::uint32_t exit_reason) : m_reason(exit_reason)
{
std::ostringstream oss;
oss << "actor exited with reason " << exit_reason;
m_what = oss.str();
}
std::uint32_t actor_exited::reason() const
{
return m_reason;
}
const char* actor_exited::what() const throw()
{
return m_what.c_str();
}
} // namespace cppa
#include "cppa/actor_proxy.hpp"
namespace cppa {
actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
: actor(mid), m_parent(pptr)
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
actor_proxy::actor_proxy(std::uint32_t mid, process_information_ptr&& pptr)
: actor(mid), m_parent(std::move(pptr))
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
//implemented in unicast_network.cpp
//void actor_proxy::enqueue(const message& msg)
//{
//}
bool actor_proxy::attach(attachable* ptr)
{
delete ptr;
return false;
}
void actor_proxy::join(group_ptr& what)
{
}
void actor_proxy::leave(const group_ptr& what)
{
}
void actor_proxy::link_to(intrusive_ptr<actor>& other)
{
}
void actor_proxy::unlink_from(intrusive_ptr<actor>& other)
{
}
bool actor_proxy::remove_backlink(const intrusive_ptr<actor>& to)
{
return true;
}
bool actor_proxy::establish_backlink(const intrusive_ptr<actor>& to)
{
return true;
}
const process_information& actor_proxy::parent_process() const
{
return *m_parent;
}
process_information_ptr actor_proxy::parent_process_ptr() const
{
return m_parent;
}
} // namespace cppa
#include <boost/thread.hpp>
#include "cppa/detail/actor_proxy_cache.hpp"
namespace {
boost::thread_specific_ptr<cppa::detail::actor_proxy_cache> t_proxy_cache;
} // namespace <anonymous>
namespace cppa { namespace detail {
process_information_ptr
actor_proxy_cache::get_pinfo(const actor_proxy_cache::key_tuple& key)
{
auto i = m_pinfos.find(key);
if (i != m_pinfos.end())
{
return i->second;
}
process_information_ptr tmp(new process_information(std::get<1>(key),
std::get<2>(key)));
m_pinfos.insert(std::make_pair(key, tmp));
return tmp;
}
actor_proxy_ptr actor_proxy_cache::get(const key_tuple& key)
{
auto i = m_proxies.find(key);
if (i != m_proxies.end())
{
return i->second;
}
return new actor_proxy(std::get<0>(key), get_pinfo(key));
}
void actor_proxy_cache::add(const actor_proxy_ptr& pptr)
{
auto pinfo = pptr->parent_process_ptr();
key_tuple key(pptr->id(), pinfo->process_id, pinfo->node_id);
m_pinfos.insert(std::make_pair(key, pinfo));
m_proxies.insert(std::make_pair(key, pptr));
}
actor_proxy_cache& get_actor_proxy_cache()
{
if (t_proxy_cache.get() == nullptr)
{
t_proxy_cache.reset(new actor_proxy_cache);
}
return *t_proxy_cache.get();
}
} } // namespace cppa::detail
#include "cppa/attachable.hpp"
namespace cppa {
attachable::~attachable()
{
}
void attachable::detach(std::uint32_t)
{
}
} // namespace cppa::detail
......@@ -6,13 +6,20 @@
#include "cppa/detail/converted_thread_context.hpp"
using cppa::detail::converted_thread_context;
namespace {
void cleanup_fun(cppa::context* what)
{
if (what && !what->deref())
if (what)
{
delete what;
auto converted = dynamic_cast<converted_thread_context*>(what);
if (converted)
{
converted->cleanup();
}
if (!what->deref()) delete what;
}
}
......@@ -42,7 +49,7 @@ context* self()
context* result = m_this_context.get();
if (!result)
{
result = new detail::converted_thread_context;
result = new converted_thread_context;
result->ref();
get_scheduler()->register_converted_context(result);
m_this_context.reset(result);
......
#include <memory>
#include <algorithm>
#include "cppa/exception.hpp"
#include "cppa/exit_signal.hpp"
#include "cppa/actor_exited.hpp"
#include "cppa/detail/converted_thread_context.hpp"
namespace {
......@@ -40,38 +41,77 @@ int erase_all(List& lst, const Element& e)
namespace cppa { namespace detail {
converted_thread_context::converted_thread_context() : m_exited(false)
converted_thread_context::converted_thread_context()
: m_exit_reason(exit_reason::not_exited)
{
}
void converted_thread_context::quit(std::uint32_t reason)
bool converted_thread_context::attach(attachable* ptr)
{
std::unique_ptr<attachable> uptr(ptr);
if (!ptr)
{
std::lock_guard<std::mutex> guard(m_mtx);
return m_exit_reason == exit_reason::not_exited;
}
else
{
std::uint32_t reason;
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(m_mtx);
reason = m_exit_reason;
if (reason == exit_reason::not_exited)
{
m_attachables.push_back(std::move(uptr));
return true;
}
}
uptr->detach(reason);
return false;
}
}
void converted_thread_context::cleanup(std::uint32_t reason)
{
if (reason == exit_reason::not_exited) return;
decltype(m_links) mlinks;
decltype(m_subscriptions) msubscriptions;
decltype(m_attachables) mattachables;
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(m_mtx);
m_exited = true;
m_exit_reason = reason;
mlinks = std::move(m_links);
msubscriptions = std::move(m_subscriptions);
// make sure m_links and m_subscriptions definitely are empty
mattachables = std::move(m_attachables);
// make sure lists are definitely empty
m_links.clear();
m_subscriptions.clear();
m_attachables.clear();
}
actor_ptr mself = self();
// send exit messages
for (actor_ptr& aptr : mlinks)
{
aptr->enqueue(message(mself, aptr, exit_signal(reason)));
}
for (std::unique_ptr<attachable>& ptr : mattachables)
{
ptr->detach(reason);
}
}
void converted_thread_context::quit(std::uint32_t reason)
{
cleanup(reason);
throw actor_exited(reason);
}
void converted_thread_context::link(intrusive_ptr<actor>& other)
void converted_thread_context::link_to(intrusive_ptr<actor>& other)
{
std::lock_guard<std::mutex> guard(m_mtx);
if (other && !m_exited && other->establish_backlink(this))
if (other && !exited() && other->establish_backlink(this))
{
m_links.push_back(other);
//m_links.insert(other);
......@@ -97,7 +137,7 @@ bool converted_thread_context::establish_backlink(const intrusive_ptr<actor>& ot
// lifetime scope of guard
{
std::lock_guard<std::mutex> guard(m_mtx);
if (!m_exited)
if (!exited())
{
result = unique_insert(m_links, other);
//result = m_links.insert(other).second;
......@@ -115,10 +155,10 @@ bool converted_thread_context::establish_backlink(const intrusive_ptr<actor>& ot
return result;
}
void converted_thread_context::unlink(intrusive_ptr<actor>& other)
void converted_thread_context::unlink_from(intrusive_ptr<actor>& other)
{
std::lock_guard<std::mutex> guard(m_mtx);
if (other && !m_exited && other->remove_backlink(this))
if (other && !exited() && other->remove_backlink(this))
{
erase_all(m_links, other);
//m_links.erase(other);
......@@ -128,7 +168,7 @@ void converted_thread_context::unlink(intrusive_ptr<actor>& other)
void converted_thread_context::join(group_ptr& what)
{
std::lock_guard<std::mutex> guard(m_mtx);
if (!m_exited && m_subscriptions.count(what) == 0)
if (!exited() && m_subscriptions.count(what) == 0)
{
auto s = what->subscribe(this);
// insert only valid subscriptions
......
#include <sstream>
#include "cppa/exception.hpp"
#include <sys/socket.h>
#include <sys/un.h>
#include <stdlib.h>
namespace cppa {
exception::exception(const std::string &what_str) : m_what(what_str)
{
}
exception::exception(std::string&& what_str) : m_what(std::move(what_str))
{
}
exception::~exception() throw()
{
}
const char* exception::what() const throw()
{
return m_what.c_str();
}
namespace // <anonymous>
{
std::string ae_what(std::uint32_t reason)
{
std::ostringstream oss;
oss << "actor exited with reason " << reason;
return oss.str();
}
} // namespace <anonymous>
actor_exited::actor_exited(std::uint32_t reason) : exception(ae_what(reason))
{
m_reason = reason;
}
std::uint32_t actor_exited::reason() const throw()
{
return m_reason;
}
network_exception::network_exception(const std::string& str) : exception(str)
{
}
network_exception::network_exception(std::string&& str)
: exception(std::move(str))
{
}
namespace // <anonymous>
{
std::string be_what(int err_code)
{
switch (err_code)
{
case EACCES: return "EACCES: address is protected; root access needed";
case EADDRINUSE: return "EADDRINUSE: address is already in use";
case EBADF: return "EBADF: no valid socket descriptor";
case EINVAL: return "EINVAL: socket already bound to an address";
case ENOTSOCK: return "ENOTSOCK: file descriptor given";
default: break;
}
std::stringstream oss;
oss << "an unknown error occurred (code: " << err_code << ")";
return oss.str();
}
} // namespace <anonymous>
bind_failure::bind_failure(int err_code) : network_exception(be_what(err_code))
{
m_errno = err_code;
}
int bind_failure::error_code() const throw()
{
return m_errno;
}
} // namespace cppa
......@@ -2,11 +2,7 @@
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() : m_reason(exit_reason::normal)
{
}
......@@ -19,14 +15,4 @@ 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
......@@ -48,8 +48,8 @@ cppa::process_information compute_proc_info()
}
pclose(cmd);
erase_trailing_newline(first_mac_addr);
auto tmp = cppa::util::ripemd_160(first_mac_addr + hd_serial);
memcpy(result.node_id, tmp.data(), cppa::process_information::node_id_size);
result.node_id = cppa::util::ripemd_160(first_mac_addr + hd_serial);
//memcpy(result.node_id, tmp.data(), cppa::process_information::node_id_size);
return result;
}
......@@ -59,13 +59,18 @@ namespace cppa {
process_information::process_information() : process_id(0)
{
memset(node_id, 0, node_id_size);
memset(node_id.data(), 0, node_id_size);
}
process_information::process_information(const process_information& other)
: ref_counted(), process_id(other.process_id)
: ref_counted(), process_id(other.process_id), node_id(other.node_id)
{
//memcpy(node_id, other.node_id, node_id_size);
}
process_information::process_information(std::uint32_t a, const node_id_type& b)
: ref_counted(), process_id(a), node_id(b)
{
memcpy(node_id, other.node_id, node_id_size);
}
process_information&
......@@ -75,7 +80,8 @@ process_information::operator=(const process_information& other)
if (this != &other)
{
process_id = other.process_id;
memcpy(node_id, other.node_id, node_id_size);
node_id = other.node_id;
//memcpy(node_id, other.node_id, node_id_size);
}
return *this;
}
......@@ -102,8 +108,8 @@ const process_information& process_information::get()
int process_information::compare(const process_information& other) const
{
int tmp = strncmp(reinterpret_cast<const char*>(node_id),
reinterpret_cast<const char*>(other.node_id),
int tmp = strncmp(reinterpret_cast<const char*>(node_id.data()),
reinterpret_cast<const char*>(other.node_id.data()),
node_id_size);
if (tmp == 0)
{
......@@ -114,4 +120,51 @@ int process_information::compare(const process_information& other) const
return tmp;
}
void process_information::node_id_from_string(const std::string& str,
process_information::node_id_type& arr)
{
if (str.size() != (arr.size() * 2))
{
throw std::logic_error("str is not a node id hash");
}
auto j = str.begin();
for (size_t i = 0; i < arr.size(); ++i)
{
auto& val = arr[i];
val = 0;
for (int tmp = 0; tmp < 2; ++tmp)
{
char c = *j;
++j;
if (isdigit(c))
{
val |= static_cast<std::uint8_t>(c - '0');
}
else if (isalpha(c))
{
if (c >= 'a' && c <= 'f')
{
val |= static_cast<std::uint8_t>((c - 'a') + 10);
}
else if (c >= 'A' && c <= 'F')
{
val |= static_cast<std::uint8_t>((c - 'A') + 10);
}
else
{
throw std::logic_error(std::string("illegal character: ") + c);
}
}
else
{
throw std::logic_error(std::string("illegal character: ") + c);
}
if (tmp == 0)
{
val <<= 4;
}
}
}
}
} // namespace cppa
......@@ -3,6 +3,7 @@
#include <ios> // ios_base::failure
#include <list>
#include <memory>
#include <iostream>
#include <stdexcept>
#include <netdb.h>
......@@ -14,9 +15,18 @@
#include "cppa/cppa.hpp"
#include "cppa/match.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exception.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/util/single_reader_queue.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
using std::cout;
using std::cerr;
using std::endl;
using cppa::detail::get_actor_proxy_cache;
namespace cppa {
......@@ -33,88 +43,151 @@ namespace {
inline void closesocket(native_socket_t s) { close(s); }
#endif
typedef intrusive_ptr<process_information> pinfo_ptr;
struct mailman_send_job
{
actor_proxy_ptr client;
message original_message;
mailman_send_job(actor_proxy_ptr apptr, message omsg)
: client(apptr), original_message(omsg)
{
}
};
struct mailman_add_peer
{
native_socket_t sockfd;
process_information_ptr pinfo;
mailman_add_peer(native_socket_t sfd, const process_information_ptr& pinf)
: sockfd(sfd), pinfo(pinf)
{
}
};
class actor_proxy : public actor
class mailman_job
{
pinfo_ptr m_parent;
friend class
util::single_reader_queue<mailman_job>;
public:
actor_proxy(const pinfo_ptr& parent) : m_parent(parent)
enum job_type
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
send_job_type,
add_peer_type,
kill_type
};
private:
actor_proxy(pinfo_ptr&& parent) : m_parent(std::move(parent))
mailman_job* next;
job_type m_type;
union
{
mailman_send_job m_send_job;
mailman_add_peer m_add_socket;
};
mailman_job(job_type jt) : next(0), m_type(jt)
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
void enqueue(const message& msg);
public:
void join(group_ptr& what);
mailman_job(actor_proxy_ptr apptr, message omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(apptr, omsg);
}
void leave(const group_ptr& what);
mailman_job(native_socket_t sockfd, const process_information_ptr& pinfo)
: next(0), m_type(add_peer_type)
{
new (&m_add_socket) mailman_add_peer(sockfd, pinfo);
}
void link(intrusive_ptr<actor>& other);
static mailman_job* kill_job()
{
return new mailman_job(kill_type);
}
void unlink(intrusive_ptr<actor>& other);
~mailman_job()
{
switch (m_type)
{
case send_job_type:
m_send_job.~mailman_send_job();
break;
case add_peer_type:
m_add_socket.~mailman_add_peer();
break;
default: break;
}
}
bool remove_backlink(const intrusive_ptr<actor>& to);
mailman_send_job& send_job()
{
return m_send_job;
}
bool establish_backlink(const intrusive_ptr<actor>& to);
mailman_add_peer& add_peer_job()
{
return m_add_socket;
}
const process_information& parent_process() const
job_type type() const
{
return *m_parent;
return m_type;
}
};
bool is_send_job() const
{
return m_type == send_job_type;
}
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
bool is_add_peer_job() const
{
return m_type == add_peer_type;
}
struct mailman_job
{
mailman_job* next;
actor_proxy_ptr client;
message original_message;
mailman_job(actor_proxy_ptr apptr, message omsg)
: next(0), client(apptr), original_message(omsg)
bool is_kill_job() const
{
return m_type == kill_type;
}
};
util::single_reader_queue<mailman_job> s_mailman_queue;
void mailman_loop();
void actor_proxy::enqueue(const message& msg)
{
s_mailman_queue.push_back(new mailman_job(this, msg));
}
//util::single_reader_queue<mailman_job> s_mailman_queue;
//boost::thread s_mailman_thread(mailman_loop);
void actor_proxy::join(group_ptr& what)
struct mailman_manager
{
}
void actor_proxy::leave(const group_ptr& what)
{
}
boost::thread* m_loop;
util::single_reader_queue<mailman_job>* m_queue;
void actor_proxy::link(intrusive_ptr<actor>& other)
{
}
mailman_manager()
{
m_queue = new util::single_reader_queue<mailman_job>;
m_loop = new boost::thread(mailman_loop);
}
void actor_proxy::unlink(intrusive_ptr<actor>& other)
{
}
~mailman_manager()
{
m_queue->push_back(mailman_job::kill_job());
m_loop->join();
delete m_loop;
delete m_queue;
}
bool actor_proxy::remove_backlink(const intrusive_ptr<actor>& to)
{
}
s_mailman_manager;
bool actor_proxy::establish_backlink(const intrusive_ptr<actor>& to)
util::single_reader_queue<mailman_job>& s_mailman_queue()
{
return *s_mailman_manager.m_queue;
}
// a map that manages links between local actors and remote actors (proxies)
......@@ -160,56 +233,89 @@ void remove_link(link_map& links,
// handles *all* outgoing messages
void mailman_loop()
{
cout << "mailman_loop()" << endl;
link_map links;
int send_flags = 0;
int flags = 0;
binary_serializer bs;
mailman_job* job = nullptr;
const auto& pself = process_information::get();
std::map<process_information, native_socket_t> peers;
for (;;)
{
job = s_mailman_queue.pop();
// keep track about link states of local actors
// (remove link states between local and remote actors if needed)
if (match<exit_signal>(job->original_message.content()))
job = s_mailman_queue().pop();
if (job->is_send_job())
{
auto sender = job->original_message.sender();
if (pself == sender->parent_process())
mailman_send_job& sjob = job->send_job();
// keep track about link states of local actors
// (remove link states between local and remote actors if needed)
if (match<exit_signal>(sjob.original_message.content()))
{
auto sender = sjob.original_message.sender();
if (pself == sender->parent_process())
{
// local to remote
sjob.client->unlink_from(sender);
}
else
{
// remote to remote (ignored)
}
}
// forward message to receiver peer
auto peer_element = peers.find(sjob.client->parent_process());
if (peer_element != peers.end())
{
// an exit message from a local actor to a remote actor
job->client->unlink(sender);
auto peer = peer_element->second;
try
{
bs << sjob.original_message;
auto size32 = static_cast<std::uint32_t>(bs.size());
cout << "--> " << to_string(sjob.original_message) << endl;
auto sent = ::send(peer, &size32, sizeof(size32), flags);
if (sent != -1)
{
sent = ::send(peer, bs.data(), bs.size(), flags);
}
if (sent == -1)
{
// peer unreachable
peers.erase(sjob.client->parent_process());
}
}
catch (...)
{
// TODO: some kind of error handling?
}
bs.reset();
}
else
{
// an exit message from a remote actor to another remote actor
// TODO: some kind of error handling?
}
}
// forward message to receiver peer
auto peer_element = peers.find(job->client->parent_process());
if (peer_element != peers.end())
else if (job->is_add_peer_job())
{
auto peer = peer_element->second;
try
mailman_add_peer& pjob = job->add_peer_job();
auto i = peers.find(*(pjob.pinfo));
if (i == peers.end())
{
bs << job->original_message;
auto size32 = static_cast<std::uint32_t>(bs.size());
auto sent = ::send(peer, sizeof(size32), &size32, send_flags);
if (sent != -1)
{
sent = ::send(peer, bs.data(), bs.size(), send_flags);
}
if (sent == -1)
{
// peer unreachable
peers.erase(job->client->parent_process());
}
cout << "mailman added " << pjob.pinfo->process_id << "@"
<< pjob.pinfo->node_id_as_string() << endl;
peers.insert(std::make_pair(*(pjob.pinfo), pjob.sockfd));
}
catch (...)
else
{
// TODO: some kind of error handling?
}
bs.reset();
}
else if (job->is_kill_job())
{
delete job;
return;
}
delete job;
}
}
......@@ -235,11 +341,16 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size)
}
// handles *one* socket
void post_office_loop(native_socket_t socket_fd) // socket file descriptor
void post_office_loop(native_socket_t socket_fd, const actor_proxy_ptr& aptr)
{
cout << "--> post_office_loop" << endl;
if (aptr) detail::get_actor_proxy_cache().add(aptr);
auto meta_msg = uniform_typeid<message>();
if (!meta_msg)
{
throw std::logic_error("meta_msg == nullptr");
}
message msg;
int read_flags = 0;
std::uint32_t rsize;
char* buf = nullptr;
size_t buf_size = 0;
......@@ -264,25 +375,83 @@ void post_office_loop(native_socket_t socket_fd) // socket file descriptor
buf_size = rsize;
read_from_socket(socket_fd, buf, buf_size);
binary_deserializer bd(buf, buf_size);
uniform_type_info* meta_msg;
meta_msg->deserialize(&msg, &bd);
cout << "<-- " << to_string(msg) << endl;
auto r = msg.receiver();
if (r) r->enqueue(msg);
}
}
catch (std::ios_base::failure& e)
{
cout << "std::ios_base::failure: " << e.what() << endl;
}
catch (std::exception& e)
{
cout << detail::to_uniform_name(typeid(e)) << ": " << e.what() << endl;
}
cout << "<-- post_office_loop" << endl;
}
void middle_man_loop(native_socket_t server_socket_fd)
struct mm_worker
{
native_socket_t m_sockfd;
boost::thread m_thread;
mm_worker(native_socket_t sockfd) : m_sockfd(sockfd)
, m_thread(post_office_loop,
sockfd,
actor_proxy_ptr())
{
}
~mm_worker()
{
closesocket(m_sockfd);
m_thread.join();
}
};
struct shared_barrier : ref_counted
{
boost::barrier m_barrier;
shared_barrier() : m_barrier(2) { }
~shared_barrier() { }
void wait() { m_barrier.wait(); }
};
struct mm_handle : attachable
{
native_socket_t m_sockfd;
intrusive_ptr<shared_barrier> m_barrier;
mm_handle(native_socket_t sockfd,
const intrusive_ptr<shared_barrier>& mbarrier)
: m_sockfd(sockfd), m_barrier(mbarrier)
{
}
virtual ~mm_handle()
{
cout << "--> ~mm_handle()" << endl;
closesocket(m_sockfd);
if (m_barrier) m_barrier->wait();
cout << "<-- ~mm_handle()" << endl;
}
};
void middle_man_loop(native_socket_t server_socket_fd,
const actor_ptr& published_actor,
intrusive_ptr<shared_barrier> barrier)
{
if (!published_actor) return;
sockaddr addr;
socklen_t addrlen;
std::vector<boost::thread> children;
typedef std::unique_ptr<mm_worker> child_ptr;
std::vector<child_ptr> children;
auto id = published_actor->id();
const process_information& pinf = process_information::get();
try
{
for (;;)
......@@ -292,29 +461,48 @@ void middle_man_loop(native_socket_t server_socket_fd)
{
throw std::ios_base::failure("invalid socket accepted");
}
cout << "socket accepted" << endl;
::send(sockfd, &id, sizeof(id), 0);
::send(sockfd, &(pinf.process_id), sizeof(pinf.process_id), 0);
::send(sockfd, pinf.node_id.data(), pinf.node_id.size(), 0);
auto peer_pinf = new process_information;
read_from_socket(sockfd,
&(peer_pinf->process_id),
sizeof(std::uint32_t));
read_from_socket(sockfd,
peer_pinf->node_id.data(),
process_information::node_id_size);
s_mailman_queue().push_back(new mailman_job(sockfd, peer_pinf));
// todo: check if connected peer is compatible
children.push_back(boost::thread(post_office_loop, sockfd));
children.push_back(child_ptr(new mm_worker(sockfd)));
cout << "client connection done" << endl;
}
}
catch (...)
{
}
for (auto& child : children)
{
child.join();
}
// calls destructors of all children
children.clear();
// wait for handshake
barrier->wait();
}
} // namespace <anonmyous>
void publish(const actor_ptr& whom, std::uint16_t port)
void actor_proxy::enqueue(const message& msg)
{
s_mailman_queue().push_back(new mailman_job(this, msg));
}
void publish(actor_ptr& whom, std::uint16_t port)
{
if (!whom) return;
native_socket_t sockfd;
struct sockaddr_in serv_addr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
throw std::ios_base::failure("could not create server socket");
throw network_exception("could not create server socket");
}
memset((char*) &serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
......@@ -322,18 +510,64 @@ void publish(const actor_ptr& whom, std::uint16_t port)
serv_addr.sin_port = htons(port);
if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
{
throw std::ios_base::failure("could not bind socket to port");
throw bind_failure(errno);
}
if (listen(sockfd, 10) != 0)
{
throw std::ios_base::failure("listen() failed");
throw network_exception("listen() failed");
}
boost::thread(middle_man_loop, sockfd).detach();
intrusive_ptr<shared_barrier> barrier_ptr(new shared_barrier);
boost::thread(middle_man_loop, sockfd, whom, barrier_ptr).detach();
whom->attach(new mm_handle(sockfd, barrier_ptr));
}
actor_ptr remote_actor(const char* host, std::uint16_t port)
void publish(actor_ptr&& whom, std::uint16_t port)
{
publish(static_cast<actor_ptr&>(whom), port);
}
actor_ptr remote_actor(const char* host, std::uint16_t port)
{
native_socket_t sockfd;
struct sockaddr_in serv_addr;
struct hostent* server;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
throw network_exception("socket creation failed");
}
server = gethostbyname(host);
if (!server)
{
std::string errstr = "no such host: ";
errstr += host;
throw network_exception(std::move(errstr));
}
memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
serv_addr.sin_port = htons(port);
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
{
throw network_exception("could not connect to host");
}
auto& pinf = process_information::get();
::send(sockfd, &(pinf.process_id), sizeof(pinf.process_id), 0);
::send(sockfd, pinf.node_id.data(), pinf.node_id.size(), 0);
auto peer_pinf = new process_information;
std::uint32_t remote_actor_id;
read_from_socket(sockfd, &remote_actor_id, sizeof(remote_actor_id));
read_from_socket(sockfd,
&(peer_pinf->process_id),
sizeof(std::uint32_t));
read_from_socket(sockfd,
peer_pinf->node_id.data(),
peer_pinf->node_id.size());
process_information_ptr pinfptr(peer_pinf);
actor_proxy_ptr result(new actor_proxy(remote_actor_id, pinfptr));
s_mailman_queue().push_back(new mailman_job(sockfd, pinfptr));
boost::thread(post_office_loop, sockfd, result).detach();
return result;
}
} // namespace cppa
......@@ -6,6 +6,7 @@
#include <limits>
#include <cstdint>
#include <sstream>
#include <iostream>
#include <type_traits>
#include "cppa/config.hpp"
......@@ -25,8 +26,11 @@
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
using std::cout;
using std::endl;
using cppa::util::void_type;
namespace std {
......@@ -126,8 +130,12 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
}
else
{
primitive_variant ptup[3];
ptup[0] = ptr->id();
ptup[1] = ptr->parent_process().process_id;
ptup[2] = ptr->parent_process().node_id_as_string();
sink->begin_object(name);
sink->write_value(ptr->id());
sink->write_tuple(3, ptup);
sink->end_object();
}
}
......@@ -150,10 +158,27 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
}
else
{
primitive_variant ptup[3];
primitive_type ptypes[] = { pt_uint32, pt_uint32, pt_u8string };
source->begin_object(cname);
auto id = get<std::uint32_t>(source->read_value(pt_uint32));
ptrref = actor::by_id(id);
source->read_tuple(3, ptypes, ptup);
source->end_object();
const std::string& nstr = get<std::string>(ptup[2]);
// local actor?
auto& pinf = process_information::get();
if ( pinf.process_id == get<std::uint32_t>(ptup[1])
&& pinf.node_id_as_string() == nstr)
{
ptrref = actor::by_id(get<std::uint32_t>(ptup[0]));
}
else
{
actor_proxy_cache::key_tuple key;
std::get<0>(key) = get<std::uint32_t>(ptup[0]);
std::get<1>(key) = get<std::uint32_t>(ptup[1]);
process_information::node_id_from_string(nstr,std::get<2>(key));
ptrref = detail::get_actor_proxy_cache().get(key);
}
}
}
......@@ -505,7 +530,7 @@ class uniform_type_info_map
insert<std::u32string>();
insert(new default_uniform_type_info_impl<exit_signal>(
std::make_pair(&exit_signal::reason,
&exit_signal::set_uint_reason)),
&exit_signal::set_reason)),
{ raw_name<exit_signal>() });
insert(new any_tuple_tinfo, { raw_name<any_tuple>() });
insert(new actor_ptr_tinfo, { raw_name<actor_ptr>() });
......
......@@ -7,16 +7,19 @@ LIB_FLAGS = $(LIBS) -L../ -lcppa
EXECUTABLE = ../test
HEADERS = hash_of.hpp \
test.hpp
test.hpp \
ping_pong.hpp
SOURCES = hash_of.cpp \
main.cpp \
ping_pong.cpp \
test__a_matches_b.cpp \
test__atom.cpp \
test__intrusive_ptr.cpp \
test__local_group.cpp \
test__primitive_variant.cpp \
test__queue_performance.cpp \
test__remote_actor.cpp \
test__ripemd_160.cpp \
test__serialization.cpp \
test__spawn.cpp \
......
......@@ -25,6 +25,21 @@ std::cout << "run " << #fun_name << " ..." << std::endl; \
errors += fun_name (); \
std::cout << std::endl
#define RUN_TEST_A1(fun_name, arg1) \
std::cout << "run " << #fun_name << " ..." << std::endl; \
errors += fun_name ((arg1)); \
std::cout << std::endl
#define RUN_TEST_A2(fun_name, arg1, arg2) \
std::cout << "run " << #fun_name << " ..." << std::endl; \
errors += fun_name ((arg1), (arg2)); \
std::cout << std::endl
#define RUN_TEST_A3(fun_name, arg1, arg2, arg3) \
std::cout << "run " << #fun_name << " ..." << std::endl; \
errors += fun_name ((arg1), (arg2), (arg3)); \
std::cout << std::endl
using std::cout;
using std::cerr;
using std::endl;
......@@ -44,8 +59,6 @@ void print_node_id()
int main(int argc, char** c_argv)
{
print_node_id();
cout << endl << endl;
//return 0;
std::vector<std::string> argv;
......@@ -61,6 +74,12 @@ int main(int argc, char** c_argv)
test__queue_performance();
return 0;
}
else if (argv.size() == 2 && argv.front() == "test__remote_actor")
{
test__remote_actor(c_argv[0], true, argv);
cout << "BLABLUBB" << endl;
return 0;
}
else
{
cerr << "usage: test [performance_test]" << endl;
......@@ -68,6 +87,8 @@ int main(int argc, char** c_argv)
}
else
{
print_node_id();
cout << endl << endl;
std::cout << std::boolalpha;
size_t errors = 0;
RUN_TEST(test__ripemd_160);
......@@ -81,6 +102,7 @@ int main(int argc, char** c_argv)
RUN_TEST(test__spawn);
RUN_TEST(test__local_group);
RUN_TEST(test__atom);
RUN_TEST_A3(test__remote_actor, c_argv[0], false, argv);
cout << endl
<< "error(s) in all tests: " << errors
<< endl;
......
#include "ping_pong.hpp"
#include "cppa/cppa.hpp"
namespace { int s_pongs = 0; }
using namespace cppa;
void pong(actor_ptr ping_actor)
{
link(ping_actor);
bool done = false;
// kickoff
ping_actor << make_tuple(0); // or: send(ping_actor, 0);
// invoke rules
auto rules = (on<std::int32_t>(9) >> [&]() { done = true; },
on<std::int32_t>() >> [](int v) { reply(v+1); });
// loop
while (!done) receive(rules);
// terminate with non-normal exit reason
// to force ping actor to quit
quit(exit_reason::user_defined);
}
void ping()
{
s_pongs = 0;
// invoke rule
auto rule = (on<std::int32_t>() >> [](std::int32_t v)
{
++s_pongs;
reply(v+1);
});
// loop
for (;;) receive(rule);
}
int pongs()
{
return s_pongs;
}
#ifndef PING_PONG_HPP
#define PING_PONG_HPP
#include "cppa/actor.hpp"
void ping();
void pong(cppa::actor_ptr ping_actor);
// returns the number of messages ping received
int pongs();
#endif // PING_PONG_HPP
#ifndef TEST_HPP
#define TEST_HPP
#include <vector>
#include <string>
#include <cstddef>
#include <iostream>
......@@ -46,6 +48,8 @@ std::cerr << err_msg << std::endl; \
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) CPPA_CHECK(((lhs_loc) == (rhs_loc)))
size_t test__remote_actor(const char* app_path, bool is_client,
const std::vector<std::string>& argv);
size_t test__ripemd_160();
size_t test__uniform_type();
size_t test__type_list();
......
#include <string>
#include <boost/thread.hpp>
#include "test.hpp"
#include "ping_pong.hpp"
#include "cppa/cppa.hpp"
#include "cppa/exception.hpp"
using namespace cppa;
namespace {
void client_part(const std::vector<std::string>& argv)
{
if (argv.size() != 2) throw std::logic_error("argv.size() != 2");
std::istringstream iss(argv[1]);
std::uint16_t port;
iss >> port;
auto ping_actor = remote_actor("localhost", port);
try
{
pong(ping_actor);
}
catch (...)
{
}
}
} // namespace <anonymous>
size_t test__remote_actor(const char* app_path, bool is_client,
const std::vector<std::string>& argv)
{
CPPA_TEST(test__remote_actor);
if (is_client)
{
client_part(argv);
return 0;
}
auto ping_actor = spawn(ping);
std::uint16_t port = 4242;
bool success = false;
while (!success)
{
try
{
publish(ping_actor, port);
success = true;
}
catch (bind_failure&)
{
// try next port
++port;
}
}
std::string cmd;
{
std::ostringstream oss;
oss << app_path << " test__remote_actor " << port << " &>/dev/null";
cmd = oss.str();
}
// execute client_part() in a separate process,
// connected via localhost socket
boost::thread child([&cmd] () { system(cmd.c_str()); });
await_all_others_done();
CPPA_CHECK_EQUAL(pongs(), 5);
// wait until separate process (in sep. thread) finished execution
child.join();
return CPPA_TEST_RESULT;
}
......@@ -2,6 +2,7 @@
#include <functional>
#include "test.hpp"
#include "ping_pong.hpp"
#include "cppa/on.hpp"
#include "cppa/cppa.hpp"
......@@ -11,59 +12,14 @@
using std::cout;
using std::endl;
namespace { int pings = 0; }
using namespace cppa;
void pong(actor_ptr ping_actor)
{
link(ping_actor);
bool done = false;
// kickoff
ping_actor << make_tuple(0); // or: send(ping_actor, 0);
// invoke rules
auto rules = (on<std::int32_t>(9) >> [&]() { done = true; },
on<std::int32_t>() >> [](int v) { reply(v+1); });
// loop
while (!done) receive(rules);
// terminate with non-normal exit reason
// to force ping actor to quit
quit(user_defined_exit_reason);
}
void ping()
{
// invoke rule
auto rule = (on<std::int32_t>() >> [](std::int32_t v)
{
++pings;
reply(v+1);
},
others() >> []()
{
cout << to_string(last_received())
<< endl;
});
// loop
for (;;) receive(rule);
}
void pong_example()
{
spawn(pong, spawn(ping));
await_all_others_done();
}
void echo(actor_ptr whom, int what)
{
send(whom, what);
}
size_t test__spawn()
{
CPPA_TEST(test__spawn);
pong_example();
CPPA_CHECK_EQUAL(pings, 5);
spawn(pong, spawn(ping));
await_all_others_done();
CPPA_CHECK_EQUAL(pongs(), 5);
/*
actor_ptr self_ptr = self();
{
......
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