Commit 5fce51f5 authored by neverlord's avatar neverlord

unification of converted_thread_context and actor_proxy implementation and actor_proxy improvements

parent e0042862
...@@ -176,3 +176,4 @@ cppa/detail/atom_val.hpp ...@@ -176,3 +176,4 @@ cppa/detail/atom_val.hpp
src/atom.cpp src/atom.cpp
src/cppa.cpp src/cppa.cpp
cppa/exit_reason.hpp cppa/exit_reason.hpp
cppa/detail/actor_impl_util.hpp
#ifndef ACTOR_PROXY_HPP #ifndef ACTOR_PROXY_HPP
#define ACTOR_PROXY_HPP #define ACTOR_PROXY_HPP
#include "cppa/config.hpp"
#include <list>
#include <mutex>
#include <atomic>
#include <vector>
#include <memory>
#include <cstdint>
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
namespace cppa { namespace cppa {
...@@ -9,6 +18,10 @@ class actor_proxy : public actor ...@@ -9,6 +18,10 @@ class actor_proxy : public actor
{ {
process_information_ptr m_parent; 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: public:
...@@ -20,12 +33,9 @@ class actor_proxy : public actor ...@@ -20,12 +33,9 @@ class actor_proxy : public actor
void detach(const attachable::token&); void detach(const attachable::token&);
// implemented in unicast_network.cpp
void enqueue(const message& msg); void enqueue(const message& msg);
void join(group_ptr& what);
void leave(const group_ptr& what);
void link_to(intrusive_ptr<actor>& other); void link_to(intrusive_ptr<actor>& other);
void unlink_from(intrusive_ptr<actor>& other); void unlink_from(intrusive_ptr<actor>& other);
......
#ifndef ATTACHABLE_HPP #ifndef ATTACHABLE_HPP
#define ATTACHABLE_HPP #define ATTACHABLE_HPP
#include <memory>
#include <cstdint> #include <cstdint>
#include <typeinfo> #include <typeinfo>
...@@ -45,6 +46,8 @@ class attachable ...@@ -45,6 +46,8 @@ class attachable
}; };
typedef std::unique_ptr<attachable> unique_attachable_ptr;
} // namespace cppa } // namespace cppa
#endif // ATTACHABLE_HPP #endif // ATTACHABLE_HPP
#ifndef ACTOR_IMPL_UTIL_HPP
#define ACTOR_IMPL_UTIL_HPP
#include <atomic>
#include <memory>
#include "cppa/attachable.hpp"
#include "cppa/exit_reason.hpp"
namespace cppa { namespace detail {
template<class Guard, class List, class Mutex>
bool do_attach(std::atomic<std::uint32_t>& reason,
unique_attachable_ptr&& uptr,
List& ptr_list,
Mutex& mtx)
{
if (uptr == nullptr)
{
Guard guard(mtx);
return reason.load() == exit_reason::not_exited;
}
else
{
std::uint32_t reason_value;
// lifetime scope of guard
{
Guard guard(mtx);
reason_value = reason.load();
if (reason_value == exit_reason::not_exited)
{
ptr_list.push_back(std::move(uptr));
return true;
}
}
uptr->detach(reason_value);
return false;
}
}
template<class Guard, class List, class Mutex>
void do_detach(const attachable::token& what, List& ptr_list, Mutex& mtx)
{
Guard guard(mtx);
for (auto i = ptr_list.begin(); i != ptr_list.end(); ++i)
{
if ((*i)->matches(what))
{
ptr_list.erase(i);
return;
}
}
}
} } // namespace cppa::detail
#endif // ACTOR_IMPL_UTIL_HPP
...@@ -6,8 +6,10 @@ ...@@ -6,8 +6,10 @@
#include <map> #include <map>
#include <list> #include <list>
#include <mutex> #include <mutex>
#include <atomic>
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <cstdint>
#include "cppa/context.hpp" #include "cppa/context.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
...@@ -19,7 +21,7 @@ class converted_thread_context : public context ...@@ -19,7 +21,7 @@ class converted_thread_context : public context
{ {
// true if the associated thread has finished execution // true if the associated thread has finished execution
std::uint32_t m_exit_reason; std::atomic<std::uint32_t> m_exit_reason;
// mailbox implementation // mailbox implementation
detail::blocking_message_queue m_mailbox; detail::blocking_message_queue m_mailbox;
...@@ -30,7 +32,7 @@ class converted_thread_context : public context ...@@ -30,7 +32,7 @@ class converted_thread_context : public context
// manages actor links // manages actor links
std::list<actor_ptr> m_links; std::list<actor_ptr> m_links;
std::vector<std::unique_ptr<attachable>> m_attachables; std::vector<unique_attachable_ptr> m_attachables;
// @pre m_mtx is locked // @pre m_mtx is locked
inline bool exited() const inline bool exited() const
......
#include "cppa/actor_proxy.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 cppa { namespace cppa {
actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr) actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
: actor(mid), m_parent(pptr) : actor(mid), m_parent(pptr), m_exit_reason(not_exited)
{ {
if (!m_parent) throw std::runtime_error("parent == nullptr"); if (!m_parent) throw std::runtime_error("parent == nullptr");
} }
actor_proxy::actor_proxy(std::uint32_t mid, process_information_ptr&& pptr) actor_proxy::actor_proxy(std::uint32_t mid, process_information_ptr&& pptr)
: actor(mid), m_parent(std::move(pptr)) : actor(mid), m_parent(std::move(pptr)), m_exit_reason(not_exited)
{ {
if (!m_parent) throw std::runtime_error("parent == nullptr"); if (!m_parent) throw std::runtime_error("parent == nullptr");
} }
//implemented in unicast_network.cpp // implemented in unicast_network.cpp
//void actor_proxy::enqueue(const message& msg) // void actor_proxy::enqueue(const message& msg);
//{
//}
bool actor_proxy::attach(attachable* ptr) bool actor_proxy::attach(attachable* ptr)
{ {
delete ptr; return detail::do_attach<guard_type>(m_exit_reason,
return false; unique_attachable_ptr(ptr),
} m_attachables,
m_mtx);
void actor_proxy::detach(const attachable::token&)
{
}
void actor_proxy::join(group_ptr& what)
{
} }
void actor_proxy::leave(const group_ptr& what) void actor_proxy::detach(const attachable::token& what)
{ {
detail::do_detach<guard_type>(what, m_attachables, m_mtx);
} }
void actor_proxy::link_to(intrusive_ptr<actor>& other) void actor_proxy::link_to(intrusive_ptr<actor>& other)
......
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/detail/actor_impl_util.hpp"
#include "cppa/detail/converted_thread_context.hpp" #include "cppa/detail/converted_thread_context.hpp"
namespace { namespace {
...@@ -37,6 +38,8 @@ int erase_all(List& lst, const Element& e) ...@@ -37,6 +38,8 @@ int erase_all(List& lst, const Element& e)
return erase_all(lst, lst.begin(), lst.end(), e); return erase_all(lst, lst.begin(), lst.end(), e);
} }
typedef std::lock_guard<std::mutex> guard_type;
} // namespace <anonymous> } // namespace <anonymous>
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -48,41 +51,15 @@ converted_thread_context::converted_thread_context() ...@@ -48,41 +51,15 @@ converted_thread_context::converted_thread_context()
bool converted_thread_context::attach(attachable* ptr) bool converted_thread_context::attach(attachable* ptr)
{ {
std::unique_ptr<attachable> uptr(ptr); return detail::do_attach<guard_type>(m_exit_reason,
if (!ptr) unique_attachable_ptr(ptr),
{ m_attachables,
std::lock_guard<std::mutex> guard(m_mtx); 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::detach(const attachable::token& what) void converted_thread_context::detach(const attachable::token& what)
{ {
std::lock_guard<std::mutex> guard(m_mtx); detail::do_detach<guard_type>(what, m_attachables, m_mtx);
for (auto i = m_attachables.begin(); i != m_attachables.end(); ++i)
{
if ((*i)->matches(what))
{
m_attachables.erase(i);
return;
}
}
} }
void converted_thread_context::cleanup(std::uint32_t reason) void converted_thread_context::cleanup(std::uint32_t reason)
......
...@@ -235,14 +235,12 @@ void remove_link(link_map& links, ...@@ -235,14 +235,12 @@ void remove_link(link_map& links,
// handles *all* outgoing messages // handles *all* outgoing messages
void mailman_loop() void mailman_loop()
{ {
//cout << "mailman_loop()" << endl;
cout << "mailman_loop()" << endl; //link_map links;
link_map links;
int flags = 0; int flags = 0;
binary_serializer bs; binary_serializer bs;
mailman_job* job = nullptr; mailman_job* job = nullptr;
const auto& pself = process_information::get(); //const auto& pself = process_information::get();
std::map<process_information, native_socket_t> peers; std::map<process_information, native_socket_t> peers;
for (;;) for (;;)
{ {
...@@ -250,21 +248,24 @@ void mailman_loop() ...@@ -250,21 +248,24 @@ void mailman_loop()
if (job->is_send_job()) if (job->is_send_job())
{ {
mailman_send_job& sjob = job->send_job(); mailman_send_job& sjob = job->send_job();
const message& out_msg = sjob.original_message;
/*
// keep track about link states of local actors // keep track about link states of local actors
// (remove link states between local and remote actors if needed) // (remove link states between local and remote actors if needed)
if (match<atom(":Exit"), std::uint32_t>(sjob.original_message.content())) if (match<atom(":Exit"), std::uint32_t>(out_msg.content()))
{ {
auto sender = sjob.original_message.sender(); auto sender = out_msg.sender();
if (pself == sender->parent_process()) if (pself == sender->parent_process())
{ {
// local to remote // local to remote (local actor just died)
sjob.client->unlink_from(sender); //sjob.client->unlink_from(sender);
} }
else else
{ {
// remote to remote (ignored) // remote to remote (ignored)
} }
} }
*/
// forward message to receiver peer // forward message to receiver peer
auto peer_element = peers.find(sjob.client->parent_process()); auto peer_element = peers.find(sjob.client->parent_process());
if (peer_element != peers.end()) if (peer_element != peers.end())
...@@ -272,9 +273,9 @@ void mailman_loop() ...@@ -272,9 +273,9 @@ void mailman_loop()
auto peer = peer_element->second; auto peer = peer_element->second;
try try
{ {
bs << sjob.original_message; bs << out_msg;
auto size32 = static_cast<std::uint32_t>(bs.size()); auto size32 = static_cast<std::uint32_t>(bs.size());
cout << "--> " << to_string(sjob.original_message) << endl; //cout << "--> " << to_string(out_msg) << endl;
auto sent = ::send(peer, &size32, sizeof(size32), flags); auto sent = ::send(peer, &size32, sizeof(size32), flags);
if (sent != -1) if (sent != -1)
{ {
...@@ -303,8 +304,8 @@ cout << "--> " << to_string(sjob.original_message) << endl; ...@@ -303,8 +304,8 @@ cout << "--> " << to_string(sjob.original_message) << endl;
auto i = peers.find(*(pjob.pinfo)); auto i = peers.find(*(pjob.pinfo));
if (i == peers.end()) if (i == peers.end())
{ {
cout << "mailman added " << pjob.pinfo->process_id << "@" //cout << "mailman added " << pjob.pinfo->process_id << "@"
<< pjob.pinfo->node_id_as_string() << endl; // << pjob.pinfo->node_id_as_string() << endl;
peers.insert(std::make_pair(*(pjob.pinfo), pjob.sockfd)); peers.insert(std::make_pair(*(pjob.pinfo), pjob.sockfd));
} }
else else
...@@ -345,7 +346,7 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size) ...@@ -345,7 +346,7 @@ void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size)
// handles *one* socket // handles *one* socket
void post_office_loop(native_socket_t socket_fd, const actor_proxy_ptr& aptr) void post_office_loop(native_socket_t socket_fd, const actor_proxy_ptr& aptr)
{ {
cout << "--> post_office_loop" << endl; //cout << "--> post_office_loop" << endl;
if (aptr) detail::get_actor_proxy_cache().add(aptr); if (aptr) detail::get_actor_proxy_cache().add(aptr);
auto meta_msg = uniform_typeid<message>(); auto meta_msg = uniform_typeid<message>();
if (!meta_msg) if (!meta_msg)
...@@ -378,20 +379,21 @@ cout << "--> post_office_loop" << endl; ...@@ -378,20 +379,21 @@ cout << "--> post_office_loop" << endl;
read_from_socket(socket_fd, buf, buf_size); read_from_socket(socket_fd, buf, buf_size);
binary_deserializer bd(buf, buf_size); binary_deserializer bd(buf, buf_size);
meta_msg->deserialize(&msg, &bd); meta_msg->deserialize(&msg, &bd);
cout << "<-- " << to_string(msg) << endl; //cout << "<-- " << to_string(msg) << endl;
auto r = msg.receiver(); auto r = msg.receiver();
if (r) r->enqueue(msg); if (r) r->enqueue(msg);
} }
} }
catch (std::ios_base::failure& e) 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) catch (std::exception& e)
{ {
cout << detail::to_uniform_name(typeid(e)) << ": " << e.what() << endl; //cout << detail::to_uniform_name(typeid(e)) << ": "
// << e.what() << endl;
} }
cout << "<-- post_office_loop" << endl; //cout << "<-- post_office_loop" << endl;
} }
struct mm_worker struct mm_worker
...@@ -436,10 +438,10 @@ struct mm_handle : attachable ...@@ -436,10 +438,10 @@ struct mm_handle : attachable
virtual ~mm_handle() virtual ~mm_handle()
{ {
cout << "--> ~mm_handle()" << endl; //cout << "--> ~mm_handle()" << endl;
closesocket(m_sockfd); closesocket(m_sockfd);
if (m_barrier) m_barrier->wait(); if (m_barrier) m_barrier->wait();
cout << "<-- ~mm_handle()" << endl; //cout << "<-- ~mm_handle()" << endl;
} }
}; };
...@@ -463,7 +465,7 @@ void middle_man_loop(native_socket_t server_socket_fd, ...@@ -463,7 +465,7 @@ void middle_man_loop(native_socket_t server_socket_fd,
{ {
throw std::ios_base::failure("invalid socket accepted"); throw std::ios_base::failure("invalid socket accepted");
} }
cout << "socket accepted" << endl; //cout << "socket accepted" << endl;
::send(sockfd, &id, sizeof(id), 0); ::send(sockfd, &id, sizeof(id), 0);
::send(sockfd, &(pinf.process_id), sizeof(pinf.process_id), 0); ::send(sockfd, &(pinf.process_id), sizeof(pinf.process_id), 0);
::send(sockfd, pinf.node_id.data(), pinf.node_id.size(), 0); ::send(sockfd, pinf.node_id.data(), pinf.node_id.size(), 0);
...@@ -477,7 +479,7 @@ cout << "socket accepted" << endl; ...@@ -477,7 +479,7 @@ cout << "socket accepted" << endl;
s_mailman_queue().push_back(new mailman_job(sockfd, peer_pinf)); s_mailman_queue().push_back(new mailman_job(sockfd, peer_pinf));
// todo: check if connected peer is compatible // 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)));
cout << "client connection done" << endl; //cout << "client connection done" << endl;
} }
} }
catch (...) catch (...)
......
...@@ -14,7 +14,7 @@ using std::endl; ...@@ -14,7 +14,7 @@ using std::endl;
using namespace cppa; using namespace cppa;
using namespace cppa::util; using namespace cppa::util;
static constexpr auto s_foo = atom("abc"); static constexpr auto s_foo = atom("FooBar");
template<atom_value AtomValue, typename... Types> template<atom_value AtomValue, typename... Types>
void foo() void foo()
...@@ -31,35 +31,14 @@ size_t test__atom() ...@@ -31,35 +31,14 @@ size_t test__atom()
CPPA_TEST(test__atom); CPPA_TEST(test__atom);
CPPA_CHECK_EQUAL(to_string(s_foo), "FooBar");
self() << make_tuple(atom("foo"), static_cast<std::uint32_t>(42)); self() << make_tuple(atom("foo"), static_cast<std::uint32_t>(42));
receive(on<atom("foo"), std::uint32_t>() >> [&](std::uint32_t value) { receive(on<atom("foo"), std::uint32_t>() >> [&](std::uint32_t value) {
CPPA_CHECK_EQUAL(value, 42); CPPA_CHECK_EQUAL(value, 42);
}); });
/*
foo<static_cast<atom_value_type>(atom("abc").value()), int, float>();
cout << "a = " << get_atom_value<'a'>::_(0) << endl;
cout << "ab = " << get_atom_value<'a', 'b'>::_(0) << endl;
cout << "abc = " << get_atom_value<'a', 'b', 'c'>::_(0) << endl;
cout << "abcd = " << get_atom_value<'a', 'b', 'c', 'd'>::_(0) << endl;
cout << "__exit = " << get_atom_value<'_','_','e','x','i','t'>::_(0) << endl;
cout << "cppa:exit = " << get_atom_value<'c','p','p','a',':','e','x','i','t'>::_(0) << endl;
cout << "cppa::exit = " << get_atom_value<'c','p','p','a',':',':','e','x','i','t'>::_(0) << endl;
atom a1 = "cppa::exit";
cout << "a1 = " << to_string(a1) << endl;
atom a3 = "abc";
cout << "to_string(a3) = " << to_string(a3) << endl;
cout << "s_a3 = " << to_string(s_a3) << endl;
cout << "a3.value() = " << a3.value() << endl;
cout << "s_a1.value() = " << s_a1.value() << endl;
cout << "s_foo = " << s_foo << endl;
*/
return CPPA_TEST_RESULT; return CPPA_TEST_RESULT;
} }
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment