Commit ad937fb8 authored by neverlord's avatar neverlord

event based ping pong

parent 6c3a8c43
...@@ -36,15 +36,18 @@ ...@@ -36,15 +36,18 @@
#include "cppa/binary_serializer.hpp" #include "cppa/binary_serializer.hpp"
#include "cppa/detail/post_office.hpp" #include "cppa/detail/post_office.hpp"
/*
#define DEBUG(arg) \ #define DEBUG(arg) \
std::cout << "[process id: " \ std::cout << "[process id: " \
<< cppa::process_information::get()->process_id() \ << cppa::process_information::get()->process_id() \
<< "] " << arg << std::endl << "] " << arg << std::endl
*/
#undef DEBUG
#define DEBUG(unused) ((void) 0) #define DEBUG(unused) ((void) 0)
using std::cout;
using std::cerr;
using std::endl;
// implementation of mailman.hpp // implementation of mailman.hpp
namespace cppa { namespace detail { namespace cppa { namespace detail {
......
...@@ -48,6 +48,7 @@ ...@@ -48,6 +48,7 @@
// used cppa classes // used cppa classes
#include "cppa/atom.hpp" #include "cppa/atom.hpp"
#include "cppa/config.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
#include "cppa/deserializer.hpp" #include "cppa/deserializer.hpp"
#include "cppa/binary_deserializer.hpp" #include "cppa/binary_deserializer.hpp"
...@@ -69,15 +70,15 @@ ...@@ -69,15 +70,15 @@
#include "cppa/detail/actor_proxy_cache.hpp" #include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/addressed_message.hpp" #include "cppa/detail/addressed_message.hpp"
/*
#define DEBUG(arg) \ #define DEBUG(arg) \
std::cout << "[process id: " \ std::cout << "[process id: " \
<< cppa::process_information::get()->process_id() \ << cppa::process_information::get()->process_id() \
<< "] " << arg << std::endl << "] " << arg << std::endl
*/
#undef DEBUG
#define DEBUG(unused) ((void) 0) #define DEBUG(unused) ((void) 0)
using std::cout;
using std::cerr; using std::cerr;
using std::endl; using std::endl;
...@@ -206,16 +207,16 @@ class po_peer ...@@ -206,16 +207,16 @@ class po_peer
// removes pptr from the list of children and returns // removes pptr from the list of children and returns
// a <bool, size_t> pair, whereas: first = true if pptr is a child of this // a <bool, size_t> pair, whereas: first = true if pptr is a child of this
// second = number of remaining children // second = number of remaining children
std::pair<bool, size_t> remove_child(const actor_proxy_ptr& pptr) bool remove_child(const actor_proxy_ptr& pptr)
{ {
auto end = m_children.end(); auto end = m_children.end();
auto i = std::find(m_children.begin(), end, pptr); auto i = std::find(m_children.begin(), end, pptr);
if (i != end) if (i != end)
{ {
m_children.erase(i); m_children.erase(i);
return { true, m_children.size() }; return true;
} }
return { false, m_children.size() }; return false;
} }
...@@ -375,7 +376,7 @@ class po_peer ...@@ -375,7 +376,7 @@ class po_peer
else if ( content.size() == 2 else if ( content.size() == 2
&& t_atom_actor_ptr_types[0] == content.type_at(0) && t_atom_actor_ptr_types[0] == content.type_at(0)
&& t_atom_actor_ptr_types[1] == content.type_at(1) && t_atom_actor_ptr_types[1] == content.type_at(1)
&& content.get_as<atom_value>(0) == atom("LINK")) && content.get_as<atom_value>(0) == atom("UNLINK"))
{ {
CPPA_REQUIRE(msg.sender()->is_proxy()); CPPA_REQUIRE(msg.sender()->is_proxy());
auto whom = msg.sender().downcast<actor_proxy>(); auto whom = msg.sender().downcast<actor_proxy>();
...@@ -425,7 +426,7 @@ class po_doorman ...@@ -425,7 +426,7 @@ class po_doorman
public: public:
explicit po_doorman(post_office_msg::add_server_socket& assm, po_doorman(post_office_msg::add_server_socket& assm,
std::list<po_peer>* peers) std::list<po_peer>* peers)
: m_socket(assm.server_sockfd) : m_socket(assm.server_sockfd)
, published_actor(assm.published_actor) , published_actor(assm.published_actor)
...@@ -544,7 +545,6 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle) ...@@ -544,7 +545,6 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
// is guaranteed to be executed in the same thread // is guaranteed to be executed in the same thread
if (selected_peer == nullptr) if (selected_peer == nullptr)
{ {
if (!pptr) DEBUG("pptr == nullptr");
throw std::logic_error("selected_peer == nullptr"); throw std::logic_error("selected_peer == nullptr");
} }
pptr->enqueue(nullptr, make_cow_tuple(atom("MONITOR"))); pptr->enqueue(nullptr, make_cow_tuple(atom("MONITOR")));
...@@ -565,7 +565,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle) ...@@ -565,7 +565,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
}); });
for (;;) for (;;)
{ {
if (select(maxfd + 1, &readset, nullptr, nullptr, nullptr) < 0) if (select(maxfd + 1, &readset, nullptr, nullptr, nullptr) <= 0)
{ {
// must not happen // must not happen
perror("select()"); perror("select()");
...@@ -609,7 +609,8 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle) ...@@ -609,7 +609,8 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
case rd_queue_event: case rd_queue_event:
{ {
DEBUG("rd_queue_event"); DEBUG("rd_queue_event");
std::unique_ptr<post_office_msg> pom{msg_queue.pop()}; std::unique_ptr<post_office_msg> pom{msg_queue.try_pop()};
CPPA_REQUIRE(pom.get() != nullptr);
if (pom->is_add_peer_msg()) if (pom->is_add_peer_msg())
{ {
DEBUG("add_peer_msg"); DEBUG("add_peer_msg");
...@@ -661,11 +662,10 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle) ...@@ -661,11 +662,10 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
DEBUG("search parent of exited proxy"); DEBUG("search parent of exited proxy");
while (i != end) while (i != end)
{ {
auto result = i->remove_child(pptr); if (i->remove_child(pptr))
if (result.first) // true if pptr is a child
{ {
DEBUG("found parent of proxy"); DEBUG("found parent of proxy");
if (result.second == 0) // no more children? if (i->children_count() == 0)
{ {
// disconnect peer if we don't know any // disconnect peer if we don't know any
// actor of it and if the published // actor of it and if the published
...@@ -757,7 +757,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle) ...@@ -757,7 +757,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
for (auto& dm : kvp.second) for (auto& dm : kvp.second)
{ {
auto fd = dm.get_socket(); auto fd = dm.get_socket();
if (fd > maxfd) maxfd = fd; maxfd = std::max(maxfd, fd);
FD_SET(fd, &readset); FD_SET(fd, &readset);
} }
} }
...@@ -775,8 +775,9 @@ void post_office_add_peer(native_socket_type a0, ...@@ -775,8 +775,9 @@ void post_office_add_peer(native_socket_type a0,
std::unique_ptr<attachable>&& a3) std::unique_ptr<attachable>&& a3)
{ {
auto nm = singleton_manager::get_network_manager(); auto nm = singleton_manager::get_network_manager();
nm->post_office_queue().push_back(new post_office_msg(a0, a1, a2, nm->post_office_queue()._push_back(new post_office_msg(a0, a1, a2,
std::move(a3))); std::move(a3)));
CPPA_MEMORY_BARRIER();
pipe_msg msg = { rd_queue_event, 0 }; pipe_msg msg = { rd_queue_event, 0 };
nm->write_to_pipe(msg); nm->write_to_pipe(msg);
} }
...@@ -786,8 +787,9 @@ void post_office_publish(native_socket_type server_socket, ...@@ -786,8 +787,9 @@ void post_office_publish(native_socket_type server_socket,
{ {
DEBUG("post_office_publish(" << published_actor->id() << ")"); DEBUG("post_office_publish(" << published_actor->id() << ")");
auto nm = singleton_manager::get_network_manager(); auto nm = singleton_manager::get_network_manager();
nm->post_office_queue().push_back(new post_office_msg(server_socket, nm->post_office_queue()._push_back(new post_office_msg(server_socket,
published_actor)); published_actor));
CPPA_MEMORY_BARRIER();
pipe_msg msg = { rd_queue_event, 0 }; pipe_msg msg = { rd_queue_event, 0 };
nm->write_to_pipe(msg); nm->write_to_pipe(msg);
} }
...@@ -796,6 +798,7 @@ void post_office_unpublish(actor_id whom) ...@@ -796,6 +798,7 @@ void post_office_unpublish(actor_id whom)
{ {
DEBUG("post_office_unpublish(" << whom << ")"); DEBUG("post_office_unpublish(" << whom << ")");
auto nm = singleton_manager::get_network_manager(); auto nm = singleton_manager::get_network_manager();
CPPA_MEMORY_BARRIER();
pipe_msg msg = { unpublish_actor_event, whom }; pipe_msg msg = { unpublish_actor_event, whom };
nm->write_to_pipe(msg); nm->write_to_pipe(msg);
} }
...@@ -803,6 +806,7 @@ void post_office_unpublish(actor_id whom) ...@@ -803,6 +806,7 @@ void post_office_unpublish(actor_id whom)
void post_office_close_socket(native_socket_type sfd) void post_office_close_socket(native_socket_type sfd)
{ {
auto nm = singleton_manager::get_network_manager(); auto nm = singleton_manager::get_network_manager();
CPPA_MEMORY_BARRIER();
pipe_msg msg = { close_socket_event, static_cast<std::uint32_t>(sfd) }; pipe_msg msg = { close_socket_event, static_cast<std::uint32_t>(sfd) };
nm->write_to_pipe(msg); nm->write_to_pipe(msg);
} }
......
...@@ -161,9 +161,22 @@ int main(int argc, char** argv) ...@@ -161,9 +161,22 @@ int main(int argc, char** argv)
test__remote_actor(argv[0], true, args); test__remote_actor(argv[0], true, args);
exit(0); exit(0);
}, },
on("run", "threaded_ping_pong") >> []()
{
spawn<detached>(pong, spawn<detached>(ping, 1000));
await_all_others_done();
exit(0);
},
on("run", "ping_pong") >> []()
{
spawn_event_based_pong(spawn_event_based_ping(1000000));
await_all_others_done();
exit(0);
},
on("run_ping", arg_match) >> [&](std::string const& num_pings) on("run_ping", arg_match) >> [&](std::string const& num_pings)
{ {
auto ping_actor = spawn(ping, std::stoi(num_pings)); //auto ping_actor = spawn<detached>(ping, std::stoi(num_pings));
auto ping_actor = spawn_event_based_ping(std::stoi(num_pings));
std::uint16_t port = 4242; std::uint16_t port = 4242;
bool success = false; bool success = false;
do do
......
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
#include "ping_pong.hpp" #include "ping_pong.hpp"
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
#include "cppa/fsm_actor.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
namespace { size_t s_pongs = 0; } namespace { size_t s_pongs = 0; }
...@@ -23,7 +24,7 @@ void ping(size_t num_pings) ...@@ -23,7 +24,7 @@ void ping(size_t num_pings)
( (
on<atom("pong"), int>() >> [&](int value) on<atom("pong"), int>() >> [&](int value)
{ {
//cout << to_string(self->last_dequeued()) << endl; //cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs == num_pings) if (++s_pongs == num_pings)
{ {
reply(atom("EXIT"), exit_reason::user_defined); reply(atom("EXIT"), exit_reason::user_defined);
...@@ -40,7 +41,40 @@ void ping(size_t num_pings) ...@@ -40,7 +41,40 @@ void ping(size_t num_pings)
} }
) )
.until(gref(s_pongs) == num_pings); .until(gref(s_pongs) == num_pings);
cout << "ping is done" << endl; }
actor_ptr spawn_event_based_ping(size_t num_pings)
{
s_pongs = 0;
struct impl : public fsm_actor<impl>
{
behavior init_state;
impl(size_t num_pings)
{
init_state =
(
on<atom("pong"), int>() >> [num_pings, this](int value)
{
//cout << to_string(self->last_dequeued()) << endl;
if (++s_pongs >= num_pings)
{
reply(atom("EXIT"), exit_reason::user_defined);
become_void();
}
else
{
reply(atom("ping"), value);
}
},
others() >> []()
{
cout << __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
}
);
}
};
return spawn(new impl{num_pings});
} }
void pong(actor_ptr ping_actor) void pong(actor_ptr ping_actor)
...@@ -51,7 +85,30 @@ void pong(actor_ptr ping_actor) ...@@ -51,7 +85,30 @@ void pong(actor_ptr ping_actor)
( (
on<atom("ping"), int>() >> [](int value) on<atom("ping"), int>() >> [](int value)
{ {
//cout << to_string(self->last_dequeued()) << endl; //cout << to_string(self->last_dequeued()) << endl;
reply(atom("pong"), value + 1);
},
others() >> []()
{
cout << __FILE__ << " line " << __LINE__ << ": "
<< to_string(self->last_dequeued()) << endl;
}
);
}
actor_ptr spawn_event_based_pong(actor_ptr ping_actor)
{
CPPA_REQUIRE(ping_actor.get() != nullptr);
struct impl : public fsm_actor<impl>
{
behavior init_state;
impl()
{
init_state =
(
on<atom("ping"), int>() >> [](int value)
{
//cout << to_string(self->last_dequeued()) << endl;
reply(atom("pong"), value + 1); reply(atom("pong"), value + 1);
}, },
others() >> []() others() >> []()
...@@ -60,4 +117,10 @@ void pong(actor_ptr ping_actor) ...@@ -60,4 +117,10 @@ void pong(actor_ptr ping_actor)
<< to_string(self->last_dequeued()) << endl; << to_string(self->last_dequeued()) << endl;
} }
); );
}
};
auto pptr = spawn(new impl);
// kickoff
ping_actor->enqueue(pptr.get(), make_any_tuple(atom("pong"), 0));
return pptr;
} }
...@@ -5,8 +5,12 @@ ...@@ -5,8 +5,12 @@
void ping(size_t num_pings); void ping(size_t num_pings);
cppa::actor_ptr spawn_event_based_ping(size_t num_pings);
void pong(cppa::actor_ptr ping_actor); void pong(cppa::actor_ptr ping_actor);
cppa::actor_ptr spawn_event_based_pong(cppa::actor_ptr ping_actor);
// returns the number of messages ping received // returns the number of messages ping received
size_t pongs(); size_t pongs();
......
...@@ -24,11 +24,9 @@ void client_part(std::vector<string_pair> const& args) ...@@ -24,11 +24,9 @@ void client_part(std::vector<string_pair> const& args)
{ {
throw std::runtime_error("no port specified"); throw std::runtime_error("no port specified");
} }
std::istringstream iss(i->second); auto port = static_cast<std::uint16_t>(std::stoi(i->second));
std::uint16_t port;
iss >> port;
auto ping_actor = remote_actor("localhost", port); auto ping_actor = remote_actor("localhost", port);
spawn(pong, ping_actor); spawn_event_based_pong(ping_actor);
await_all_others_done(); await_all_others_done();
} }
...@@ -43,7 +41,8 @@ size_t test__remote_actor(char const* app_path, bool is_client, ...@@ -43,7 +41,8 @@ size_t test__remote_actor(char const* app_path, bool is_client,
return 0; return 0;
} }
CPPA_TEST(test__remote_actor); CPPA_TEST(test__remote_actor);
auto ping_actor = spawn(ping, 10); auto ping_actor = spawn_event_based_ping(10);
//auto ping_actor = spawn(ping, 10);
std::uint16_t port = 4242; std::uint16_t port = 4242;
bool success = false; bool success = false;
do do
......
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