Commit b88aaefe authored by neverlord's avatar neverlord

:Attach for proxies

parent 5fce51f5
......@@ -135,7 +135,7 @@
</data>
<data>
<variable>ProjectExplorer.Project.Updater.EnvironmentId</variable>
<value type="QString">{07fcd197-092d-45a0-8500-3be614e6ae31}</value>
<value type="QString">{23902c37-f07e-47cd-bb19-c366b9f708db}</value>
</data>
<data>
<variable>ProjectExplorer.Project.Updater.FileVersion</variable>
......
......@@ -17,12 +17,6 @@ namespace cppa {
class actor_proxy : public actor
{
process_information_ptr m_parent;
std::atomic<std::uint32_t> m_exit_reason;
std::mutex m_mtx;
std::vector<unique_attachable_ptr> m_attachables;
public:
actor_proxy(std::uint32_t mid, process_information_ptr&& parent);
......@@ -48,6 +42,17 @@ class actor_proxy : public actor
process_information_ptr parent_process_ptr() const;
private:
// implemented in unicast_network.cpp
static void forward_message(const process_information_ptr&, const message&);
process_information_ptr m_parent;
std::atomic<std::uint32_t> m_exit_reason;
std::mutex m_mtx;
std::vector<unique_attachable_ptr> m_attachables;
};
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
......
......@@ -71,6 +71,14 @@ inline void link(actor_ptr&& other)
self()->link_to(other);
}
void monitor(actor_ptr& whom);
void monitor(actor_ptr&& whom);
void demonitor(actor_ptr& whom);
void demonitor(actor_ptr&& whom);
inline void trap_exit(bool new_value)
{
self()->trap_exit(new_value);
......
......@@ -22,6 +22,7 @@ class actor_proxy_cache
std::map<key_tuple, actor_proxy_ptr> m_proxies;
process_information_ptr get_pinfo(const key_tuple& key);
void add(const actor_proxy_ptr& pptr, const key_tuple& key);
public:
......
......@@ -29,8 +29,8 @@ inline constexpr std::uint64_t next_interim(std::uint64_t tmp, size_t char_code)
constexpr std::uint64_t atom_val(const char* str, std::uint64_t interim = 0)
{
return (*str == '\0') ? interim
: atom_val(str + 1, next_interim(interim, *str));
return (*str <= 0) ? interim
: atom_val(str + 1, next_interim(interim, *str));
}
} } } // namespace cppa::detail::<anonymous>
......
......@@ -70,11 +70,25 @@ bool match(const any_tuple& what, const ValuesTuple& vals)
}
template<atom_value A0, typename... MatchRules>
bool match(const any_tuple& what)
inline bool match(const any_tuple& what)
{
return match<atom_value, MatchRules...>(what, make_tdata(A0));
}
template<atom_value A0, atom_value A1, typename... MatchRules>
inline bool match(const any_tuple& what)
{
return match<atom_value, atom_value, MatchRules...>(what,
make_tdata(A0, A1));
}
template<atom_value A0, atom_value A1, atom_value A2, typename... MatchRules>
bool match(const any_tuple& what)
{
auto vals = make_tdata(A0, A1, A2);
return match<atom_value, atom_value, atom_value, MatchRules...>(what, vals);
}
} // namespace cppa
#endif // MATCH_HPP
#include "cppa/atom.hpp"
#include "cppa/message.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/detail/actor_impl_util.hpp"
using cppa::exit_reason::not_exited;
namespace { typedef std::lock_guard<std::mutex> guard_type; }
namespace {
constexpr auto s_kp = cppa::atom(":KillProxy");
typedef std::lock_guard<std::mutex> guard_type;
} // namespace <anonymous>
namespace cppa {
......@@ -20,8 +27,33 @@ actor_proxy::actor_proxy(std::uint32_t mid, process_information_ptr&& pptr)
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
// implemented in unicast_network.cpp
// void actor_proxy::enqueue(const message& msg);
void actor_proxy::enqueue(const message& msg)
{
const any_tuple& content = msg.content();
if ( content.size() == 2
&& content.utype_info_at(0) == typeid(atom_value)
&& *reinterpret_cast<const atom_value*>(content.at(0)) == s_kp
&& content.utype_info_at(1) == typeid(std::uint32_t))
{
decltype(m_attachables) mattachables;
auto r = *reinterpret_cast<const std::uint32_t*>(content.at(1));
// lifetime scope of guard
{
guard_type guard(m_mtx);
m_exit_reason = r;
mattachables = std::move(m_attachables);
m_attachables.clear();
}
for (auto i = mattachables.begin(); i != mattachables.end(); ++i)
{
(*i)->detach(r);
}
}
else
{
forward_message(m_parent, msg);
}
}
bool actor_proxy::attach(attachable* ptr)
{
......
#include <boost/thread.hpp>
#include "cppa/atom.hpp"
#include "cppa/message.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
namespace {
......@@ -31,15 +33,24 @@ actor_proxy_ptr actor_proxy_cache::get(const key_tuple& key)
{
return i->second;
}
return new actor_proxy(std::get<0>(key), get_pinfo(key));
actor_proxy_ptr result(new actor_proxy(std::get<0>(key), get_pinfo(key)));
result->enqueue(message(result, nullptr, make_tuple(atom(":Monitor"),
result)));
add(result);
return result;
}
void actor_proxy_cache::add(const actor_proxy_ptr& pptr, const key_tuple& key)
{
m_pinfos.insert(std::make_pair(key, pptr->parent_process_ptr()));
m_proxies.insert(std::make_pair(key, pptr));
}
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));
add(pptr, key);
}
actor_proxy_cache& get_actor_proxy_cache()
......
#include "cppa/cppa.hpp"
namespace {
class observer : public cppa::attachable
{
cppa::actor_ptr m_client;
public:
observer(cppa::actor_ptr&& client) : m_client(std::move(client)) { }
void detach(std::uint32_t reason)
{
cppa::send(m_client, cppa::atom(":Down"), reason);
}
bool matches(const cppa::attachable::token& match_token)
{
if (match_token.subtype == typeid(observer))
{
auto ptr = reinterpret_cast<const cppa::context*>(match_token.ptr);
return m_client == ptr;
}
return false;
}
};
} // namespace <anonymous>
namespace cppa {
context* operator<<(context* whom, const any_tuple& what)
......@@ -15,4 +45,25 @@ context* operator<<(context* whom, any_tuple&& what)
return whom;
}
void monitor(actor_ptr& whom)
{
if (whom) whom->attach(new observer(self()));
}
void monitor(actor_ptr&& whom)
{
monitor(static_cast<actor_ptr&>(whom));
}
void demonitor(actor_ptr& whom)
{
attachable::token mtoken(typeid(observer), self());
if (whom) whom->detach(mtoken);
}
void demonitor(actor_ptr&& whom)
{
demonitor(static_cast<actor_ptr&>(whom));
}
} // namespace cppa
......@@ -47,10 +47,14 @@ namespace {
struct mailman_send_job
{
actor_proxy_ptr client;
process_information_ptr target_peer;
message original_message;
mailman_send_job(actor_proxy_ptr apptr, message omsg)
: client(apptr), original_message(omsg)
mailman_send_job(actor_proxy_ptr apptr, message msg)
: target_peer(apptr->parent_process_ptr()), original_message(msg)
{
}
mailman_send_job(process_information_ptr peer, message msg)
: target_peer(peer), original_message(msg)
{
}
};
......@@ -96,7 +100,13 @@ class mailman_job
public:
mailman_job(actor_proxy_ptr apptr, message omsg)
mailman_job(process_information_ptr piptr, const message& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(piptr, omsg);
}
mailman_job(actor_proxy_ptr apptr, const message& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(apptr, omsg);
......@@ -189,7 +199,7 @@ s_mailman_manager;
util::single_reader_queue<mailman_job>& s_mailman_queue()
{
return *s_mailman_manager.m_queue;
return *(s_mailman_manager.m_queue);
}
// a map that manages links between local actors and remote actors (proxies)
......@@ -204,7 +214,8 @@ void fake_exits_from_disconnected_links(link_map& links)
for (auto& rem_actor : remote_actors)
{
message msg(rem_actor, local_actor,
atom(":Exit"), exit_reason::remote_link_unreachable);
atom(":KillProxy"),
exit_reason::remote_link_unreachable);
local_actor->enqueue(msg);
}
}
......@@ -232,6 +243,26 @@ void remove_link(link_map& links,
}
}
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, make_tuple(atom(":KillProxy"), reason));
s_mailman_queue().push_back(new mailman_job(peer, msg));
}
};
// handles *all* outgoing messages
void mailman_loop()
{
......@@ -267,7 +298,7 @@ void mailman_loop()
}
*/
// forward message to receiver peer
auto peer_element = peers.find(sjob.client->parent_process());
auto peer_element = peers.find(*(sjob.target_peer));
if (peer_element != peers.end())
{
auto peer = peer_element->second;
......@@ -275,7 +306,7 @@ void mailman_loop()
{
bs << out_msg;
auto size32 = static_cast<std::uint32_t>(bs.size());
//cout << "--> " << to_string(out_msg) << endl;
cout << "--> " << to_string(out_msg) << endl;
auto sent = ::send(peer, &size32, sizeof(size32), flags);
if (sent != -1)
{
......@@ -284,7 +315,7 @@ void mailman_loop()
if (sent == -1)
{
// peer unreachable
peers.erase(sjob.client->parent_process());
peers.erase(*(sjob.target_peer));
}
}
catch (...)
......@@ -343,21 +374,31 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size)
while (urres < left);
}
// handles *one* socket
void post_office_loop(native_socket_t socket_fd, const actor_proxy_ptr& aptr)
template<typename T>
T& operator<<(T& o, const process_information& pinfo)
{
//cout << "--> post_office_loop" << endl;
return (o << pinfo.process_id << "@" << pinfo.node_id_as_string());
}
// handles *one* socket / peer
void post_office_loop(native_socket_t socket_fd,
process_information_ptr peer,
actor_proxy_ptr aptr)
{
cout << "--> post_office_loop; self() = "
<< process_information::get()
<< ", peer = "
<< *peer
<< 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;
std::uint32_t rsize;
char* buf = nullptr;
size_t buf_size = 0;
size_t buf_allocated = 0;
auto meta_msg = uniform_typeid<message>();
const std::type_info& atom_tinfo = typeid(atom_value);
auto& pinfo = process_information::get();
try
{
for (;;)
......@@ -376,24 +417,43 @@ void post_office_loop(native_socket_t socket_fd, const actor_proxy_ptr& aptr)
buf = new char[buf_allocated];
}
buf_size = rsize;
cout << "[" << pinfo << "] " << "received " << rsize << " bytes" << endl;
read_from_socket(socket_fd, buf, buf_size);
binary_deserializer bd(buf, buf_size);
meta_msg->deserialize(&msg, &bd);
//cout << "<-- " << to_string(msg) << endl;
cout << "<-- " << to_string(msg) << endl;
if ( msg.content().size() == 1
&& msg.content().utype_info_at(0) == atom_tinfo
&& *reinterpret_cast<const atom_value*>(msg.content().at(0))
== atom(":Monitor"))
{
actor_ptr sender = msg.sender();
if (sender->parent_process() == pinfo)
{
cout << pinfo.process_id << "@" << pinfo.node_id_as_string()
<< " :Monitor; actor id = " << sender->id() << endl;
// local actor?
// this message was send from a proxy
sender->attach(new remote_observer(peer));
}
// don't "deliver" message
continue;
}
auto r = msg.receiver();
if (r) r->enqueue(msg);
}
}
catch (std::ios_base::failure& e)
{
//cout << "std::ios_base::failure: " << e.what() << endl;
cout << "std::ios_base::failure: " << e.what() << endl;
}
catch (std::exception& e)
{
//cout << detail::to_uniform_name(typeid(e)) << ": "
// << e.what() << endl;
cout << "[" << process_information::get() << "] "
<< detail::to_uniform_name(typeid(e)) << ": "
<< e.what() << endl;
}
//cout << "<-- post_office_loop" << endl;
cout << "<-- post_office_loop" << endl;
}
struct mm_worker
......@@ -402,10 +462,11 @@ 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(native_socket_t sockfd, process_information_ptr peer)
: m_sockfd(sockfd), m_thread(post_office_loop,
sockfd,
peer,
actor_proxy_ptr())
{
}
......@@ -469,16 +530,16 @@ void middle_man_loop(native_socket_t server_socket_fd,
::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;
process_information_ptr peer(new process_information);
read_from_socket(sockfd,
&(peer_pinf->process_id),
&(peer->process_id),
sizeof(std::uint32_t));
read_from_socket(sockfd,
peer_pinf->node_id.data(),
peer->node_id.data(),
process_information::node_id_size);
s_mailman_queue().push_back(new mailman_job(sockfd, peer_pinf));
s_mailman_queue().push_back(new mailman_job(sockfd, peer));
// todo: check if connected peer is compatible
children.push_back(child_ptr(new mm_worker(sockfd)));
children.push_back(child_ptr(new mm_worker(sockfd, peer)));
//cout << "client connection done" << endl;
}
}
......@@ -493,9 +554,10 @@ void middle_man_loop(native_socket_t server_socket_fd,
} // namespace <anonmyous>
void actor_proxy::enqueue(const message& msg)
void actor_proxy::forward_message(const process_information_ptr& piptr,
const message& msg)
{
s_mailman_queue().push_back(new mailman_job(this, msg));
s_mailman_queue().push_back(new mailman_job(piptr, msg));
}
void publish(actor_ptr& whom, std::uint16_t port)
......@@ -570,7 +632,7 @@ actor_ptr remote_actor(const char* host, std::uint16_t port)
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();
boost::thread(post_office_loop, sockfd, peer_pinf, result).detach();
return result;
}
......
......@@ -708,8 +708,10 @@ uniform_type_info::by_type_info(const std::type_info& tinf)
auto result = detail::s_uniform_type_info_map().by_raw_name(raw_name(tinf));
if (!result)
{
throw std::runtime_error(std::string(raw_name(tinf))
+ " is an unknown typeid name");
std::string error = "uniform_type_info::by_type_info(): ";
error += detail::to_uniform_name(tinf);
error += " is an unknown typeid name";
throw std::runtime_error(error);
}
return result;
}
......
......@@ -11,12 +11,21 @@ void pong(actor_ptr ping_actor)
link(ping_actor);
bool done = false;
// kickoff
ping_actor << make_tuple(0); // or: send(ping_actor, 0);
ping_actor << make_tuple(atom("Pong"), 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); });
auto pattern =
(
on<atom("Ping"), std::int32_t>(9) >> [&]()
{
done = true;
},
on<atom("Ping"), std::int32_t>() >> [](int v)
{
reply(atom("Pong"), v+1);
}
);
// loop
while (!done) receive(rules);
while (!done) receive(pattern);
// terminate with non-normal exit reason
// to force ping actor to quit
quit(exit_reason::user_defined);
......@@ -26,13 +35,16 @@ void ping()
{
s_pongs = 0;
// invoke rule
auto rule = (on<std::int32_t>() >> [](std::int32_t v)
{
++s_pongs;
reply(v+1);
});
auto pattern =
(
on<atom("Pong"), std::int32_t>() >> [](std::int32_t v)
{
++s_pongs;
reply(atom("Ping"), v+1);
}
);
// loop
for (;;) receive(rule);
for (;;) receive(pattern);
}
int pongs()
......
......@@ -10,6 +10,7 @@
using std::cout;
using std::endl;
using std::string;
using namespace cppa;
using namespace cppa::util;
......@@ -28,17 +29,35 @@ void foo()
size_t test__atom()
{
bool matched_pattern[3] = { false, false, false };
CPPA_TEST(test__atom);
CPPA_CHECK_EQUAL(to_string(s_foo), "FooBar");
self() << make_tuple(atom("foo"), static_cast<std::uint32_t>(42));
receive(on<atom("foo"), std::uint32_t>() >> [&](std::uint32_t value) {
CPPA_CHECK_EQUAL(value, 42);
});
self() << make_tuple(atom("foo"), static_cast<std::uint32_t>(42))
<< make_tuple(atom(":Attach"), atom(":Baz"), "cstring")
<< make_tuple(atom("b"), atom("a"), atom("c"), 23.f)
<< make_tuple(atom("a"), atom("b"), atom("c"), 23.f);
auto pattern =
(
on<atom("foo"), std::uint32_t>() >> [&](std::uint32_t value)
{
matched_pattern[0] = true;
CPPA_CHECK_EQUAL(value, 42);
},
on<atom(":Attach"), atom(":Baz"), string>() >> [&](const string& str)
{
matched_pattern[1] = true;
CPPA_CHECK_EQUAL(str, "cstring");
},
on<atom("a"), atom("b"), atom("c"), float>() >> [&](float value)
{
matched_pattern[2] = true;
CPPA_CHECK_EQUAL(value, 23.f);
}
);
for (auto i = 0; i < 3; ++i) receive(pattern);
CPPA_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]);
message msg = receive();
CPPA_CHECK((match<atom("b"), atom("a"), atom("c"), float>(msg.content())));
CPPA_CHECK(try_receive(msg) == false);
return CPPA_TEST_RESULT;
}
......@@ -56,7 +56,7 @@ size_t test__remote_actor(const char* app_path, bool is_client,
std::string cmd;
{
std::ostringstream oss;
oss << app_path << " test__remote_actor " << port << " &>/dev/null";
oss << app_path << " test__remote_actor " << port;// << " &>/dev/null";
cmd = oss.str();
}
// execute client_part() in a separate process,
......
......@@ -8,7 +8,9 @@
#include "cppa/cppa.hpp"
#include "cppa/actor.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
using std::cerr;
using std::cout;
using std::endl;
......@@ -17,46 +19,36 @@ using namespace cppa;
size_t test__spawn()
{
CPPA_TEST(test__spawn);
spawn(pong, spawn(ping));
await_all_others_done();
CPPA_CHECK_EQUAL(pongs(), 5);
/*
actor_ptr self_ptr = self();
auto report_unexpected = [&] ()
{
spawn(echo, self_ptr, 1);
spawn(pong) << make_tuple(23.f) << make_tuple(2);
//send(sl, 23.f);
//send(sl, 2);
bool received_pong = false;
bool received_echo = false;
auto rules = (on<int>(42) >> [&]() { received_pong = true; },
on<int>(1) >> [&]() { received_echo = true; });
receive(rules);
receive(rules);
CPPA_CHECK(received_pong);
CPPA_CHECK(received_echo);
}
cerr << "unexpected message: " << to_string(last_received()) << endl;
CPPA_CHECK(false);
};
auto pong_actor = spawn(pong, spawn(ping));
monitor(pong_actor);
auto pattern =
(
on<atom(":Down"), std::uint32_t>() >> [&] (std::uint32_t reason)
{
CPPA_CHECK_EQUAL(reason, exit_reason::user_defined);
CPPA_CHECK_EQUAL(last_received().sender(), pong_actor);
},
others() >> [&] ()
{
report_unexpected();
}
);
// wait for :Down message of pong
receive(pattern);
// wait for termination of all spawned actors
await_all_others_done();
// mailbox has to be empty
message msg;
while (try_receive(msg))
{
auto sl = spawn([]() {
receive(on<int>() >> [](int value) {
reply((value * 20) + 2);
});
});
spawn([](actor_ptr whom, int what) { send(whom, what); },
self_ptr,
1);
send(sl, 23.f);
send(sl, 2);
bool received_pong = false;
bool received_echo = false;
auto rules = (on<int>(42) >> [&]() { received_pong = true; },
on<int>(1) >> [&]() { received_echo = true; });
receive(rules);
receive(rules);
CPPA_CHECK(received_pong);
CPPA_CHECK(received_echo);
report_unexpected();
}
await_all_others_done();
*/
// verify pong messages
CPPA_CHECK_EQUAL(pongs(), 5);
return CPPA_TEST_RESULT;
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment