Commit 904b5b3c authored by Dominik Charousset's avatar Dominik Charousset

improved broker impl and example

parent 076d830d
...@@ -565,17 +565,19 @@ actor_ptr spawn_io(F fun, ...@@ -565,17 +565,19 @@ actor_ptr spawn_io(F fun,
return eval_sopts(Options, io::init_and_launch(move(ptr))); return eval_sopts(Options, io::init_and_launch(move(ptr)));
} }
/*
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts> template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) { actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host, port); auto ptr = io::ipv4_io_stream::connect_to(host, port);
return spawn_io<Impl>(ptr, ptr, std::forward<Ts>(args)...); return spawn_io<Impl>(ptr, ptr, std::forward<Ts>(args)...);
} }
*/
template<spawn_options Options = no_spawn_options, template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>, typename F = std::function<void (io::broker*)>,
typename... Ts> typename... Ts>
actor_ptr spawn_io(F fun, const char* host, uint16_t port, Ts&&... args) { actor_ptr spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host, port); auto ptr = io::ipv4_io_stream::connect_to(host.c_str(), port);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...); return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
} }
......
...@@ -87,7 +87,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -87,7 +87,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
bool initialized() const; bool initialized() const;
void quit(std::uint32_t reason); void quit(std::uint32_t reason = exit_reason::normal) override;
void receive_policy(const connection_handle& hdl, void receive_policy(const connection_handle& hdl,
broker::policy_flag policy, broker::policy_flag policy,
...@@ -99,7 +99,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -99,7 +99,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
void write(const connection_handle& hdl, util::buffer&& buf); void write(const connection_handle& hdl, util::buffer&& buf);
static broker_ptr from(std::function<void (broker*)> fun, static broker_ptr from(std::function<void (broker*, connection_handle)> fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out); output_stream_ptr out);
...@@ -108,6 +108,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -108,6 +108,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
T0&& arg0, Ts&&... args) { T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun), return from(std::bind(std::move(fun),
std::placeholders::_1, std::placeholders::_1,
std::placeholders::_2,
detail::fwd<T0>(arg0), detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...), detail::fwd<Ts>(args)...),
std::move(in), std::move(in),
...@@ -125,16 +126,17 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -125,16 +126,17 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
std::move(in)); std::move(in));
} }
actor_ptr fork(std::function<void (broker*)> fun, actor_ptr fork(std::function<void (broker*, connection_handle)> fun,
const connection_handle& hdl); connection_handle hdl);
template<typename F, typename T0, typename... Ts> template<typename F, typename T0, typename... Ts>
actor_ptr fork(F fun, actor_ptr fork(F fun,
const connection_handle& hdl, connection_handle hdl,
T0&& arg0, T0&& arg0,
Ts&&... args) { Ts&&... args) {
return this->fork(std::bind(std::move(fun), return this->fork(std::bind(std::move(fun),
std::placeholders::_1, std::placeholders::_1,
std::placeholders::_2,
detail::fwd<T0>(arg0), detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...), detail::fwd<Ts>(args)...),
hdl); hdl);
......
...@@ -49,6 +49,12 @@ using namespace std; ...@@ -49,6 +49,12 @@ using namespace std;
using namespace cppa; using namespace cppa;
using namespace cppa::io; using namespace cppa::io;
void print_on_exit(const actor_ptr& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) {
aout << name << " exited with reason " << reason << endl;
});
}
void ping(size_t num_pings) { void ping(size_t num_pings) {
auto count = make_shared<size_t>(0); auto count = make_shared<size_t>(0);
become ( become (
...@@ -56,7 +62,7 @@ void ping(size_t num_pings) { ...@@ -56,7 +62,7 @@ void ping(size_t num_pings) {
send(pong, atom("ping"), 1); send(pong, atom("ping"), 1);
become ( become (
on(atom("pong"), arg_match) >> [=](int value) { on(atom("pong"), arg_match) >> [=](int value) {
cout << "<- pong " << value << endl; aout << "pong: " << value << endl;
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1); else reply(atom("ping"), value + 1);
} }
...@@ -68,23 +74,24 @@ void ping(size_t num_pings) { ...@@ -68,23 +74,24 @@ void ping(size_t num_pings) {
void pong() { void pong() {
become ( become (
on(atom("ping"), arg_match) >> [](int value) { on(atom("ping"), arg_match) >> [](int value) {
cout << "<- ping " << value << endl; aout << "ping: " << value << endl;
reply(atom("pong"), value); reply(atom("pong"), value);
} }
); );
} }
void protobuf_io(broker* ios, const actor_ptr& buddy) { void protobuf_io(broker* thisptr, connection_handle hdl, const actor_ptr& buddy) {
self->monitor(buddy); self->monitor(buddy);
auto write = [=](const org::libcppa::PingOrPong& p) { auto write = [=](const org::libcppa::PingOrPong& p) {
string buf = p.SerializeAsString(); string buf = p.SerializeAsString();
int32_t s = htonl(static_cast<int32_t>(buf.size())); int32_t s = htonl(static_cast<int32_t>(buf.size()));
ios->write(sizeof(int32_t), &s); thisptr->write(hdl, sizeof(int32_t), &s);
ios->write(buf.size(), buf.data()); thisptr->write(hdl, buf.size(), buf.data());
}; };
auto default_bhvr = ( auto default_bhvr = (
on(atom("IO_closed"), arg_match) >> [=](uint32_t) { on(atom("IO_closed"), hdl) >> [=] {
cout << "IO_closed" << endl; cout << "IO_closed" << endl;
send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable); self->quit(exit_reason::remote_link_unreachable);
}, },
on(atom("ping"), arg_match) >> [=](int i) { on(atom("ping"), arg_match) >> [=](int i) {
...@@ -105,12 +112,9 @@ void protobuf_io(broker* ios, const actor_ptr& buddy) { ...@@ -105,12 +112,9 @@ void protobuf_io(broker* ios, const actor_ptr& buddy) {
} }
); );
partial_function await_protobuf_data { partial_function await_protobuf_data {
on(atom("IO_read"), arg_match) >> [=](uint32_t, const util::buffer& buf) { on(atom("IO_read"), hdl, arg_match) >> [=](const util::buffer& buf) {
org::libcppa::PingOrPong p; org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), static_cast<int>(buf.size())); p.ParseFromArray(buf.data(), static_cast<int>(buf.size()));
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) { if (p.has_ping()) {
send(buddy, atom("ping"), p.ping().id()); send(buddy, atom("ping"), p.ping().id());
} }
...@@ -119,62 +123,74 @@ void protobuf_io(broker* ios, const actor_ptr& buddy) { ...@@ -119,62 +123,74 @@ void protobuf_io(broker* ios, const actor_ptr& buddy) {
} }
else { else {
self->quit(exit_reason::user_defined); self->quit(exit_reason::user_defined);
cerr << "neither Pong nor Ping!" << endl; cerr << "neither Ping nor Pong!" << endl;
} }
// receive next length prefix // receive next length prefix
ios->receive_policy(broker::exactly, 4); thisptr->receive_policy(hdl, broker::exactly, 4);
unbecome(); unbecome();
}, },
default_bhvr default_bhvr
}; };
partial_function await_length_prefix { partial_function await_length_prefix {
on(atom("IO_read"), arg_match) >> [=](uint32_t, const util::buffer& buf) { on(atom("IO_read"), hdl, arg_match) >> [=](const util::buffer& buf) {
int num_bytes; int32_t num_bytes;
memcpy(&num_bytes, buf.data(), 4); memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes); num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) { if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
aout << "someone is trying something nasty" << endl;
self->quit(exit_reason::user_defined); self->quit(exit_reason::user_defined);
return; return;
} }
// receive protobuf data // receive protobuf data
ios->receive_policy(broker::exactly, (size_t) num_bytes); thisptr->receive_policy(hdl, broker::exactly, static_cast<size_t>(num_bytes));
become(keep_behavior, await_protobuf_data); become(keep_behavior, await_protobuf_data);
}, },
default_bhvr default_bhvr
}; };
// initial setup // initial setup
ios->receive_policy(broker::exactly, 4); thisptr->receive_policy(hdl, broker::exactly, 4);
become(await_length_prefix); become(await_length_prefix);
} }
void server(broker* thisptr, actor_ptr buddy) {
aout << "server is running" << endl;
become (
on(atom("IO_accept"), arg_match) >> [=](accept_handle, connection_handle hdl) {
aout << "server: IO_accept" << endl;
auto io_actor = thisptr->fork(protobuf_io, hdl, buddy);
print_on_exit(io_actor, "protobuf_io");
// only accept 1 connection
thisptr->quit();
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
);
}
optional<uint16_t> as_u16(const std::string& str) {
return static_cast<uint16_t>(stoul(str));
}
int main(int argc, char** argv) { int main(int argc, char** argv) {
auto print_exit = [](const actor_ptr& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) {
cout << name << " exited with reason " << reason << endl;
});
};
match(std::vector<string>{argv + 1, argv + argc}) ( match(std::vector<string>{argv + 1, argv + argc}) (
on("-s") >> [&] { on("-s", as_u16) >> [&](uint16_t port) {
cout << "run in server mode" << endl; cout << "run in server mode" << endl;
std::string hi = "hello"; auto pong_actor = spawn(pong);
auto po = spawn(pong); auto sever_actor = spawn_io_server(server, port, pong_actor);
auto ack = io::ipv4_acceptor::create(4242); print_on_exit(sever_actor, "server");
for (;;) { print_on_exit(pong_actor, "pong");
auto p = ack->accept_connection(); },
//spawn_io<protobuf_io>(p.first, p.second); on("-c", val<string>, as_u16) >> [&](const string& host, uint16_t port) {
auto s = spawn_io(protobuf_io, std::move(p.first), std::move(p.second), po); auto ping_actor = spawn(ping, 20);
print_exit(s, "io actor"); auto io_actor = spawn_io(protobuf_io, host, port, ping_actor);
print_exit(po, "pong"); print_on_exit(io_actor, "protobuf_io");
} print_on_exit(ping_actor, "ping");
send_as(io_actor, ping_actor, atom("kickoff"), io_actor);
}, },
on_arg_match >> [&](const string& host, const string& port_str) { others() >> [] {
auto port = static_cast<uint16_t>(stoul(port_str)); cerr << "use with eihter '-s PORT' as server or '-c HOST PORT' as client"
auto io = io::ipv4_io_stream::connect_to(host.c_str(), port); << endl;
auto pi = spawn(ping, 20);
auto pr = spawn_io(protobuf_io, io, io, pi);
send_as(pr, pi, atom("kickoff"), pr);
print_exit(pr, "io actor");
print_exit(pi, "ping");
} }
); );
await_all_others_done(); await_all_others_done();
......
...@@ -63,11 +63,21 @@ class default_broker : public broker { ...@@ -63,11 +63,21 @@ class default_broker : public broker {
typedef std::function<void (broker*)> function_type; typedef std::function<void (broker*)> function_type;
struct fork_flag { };
template<typename... Ts> template<typename... Ts>
default_broker(function_type&& fun, default_broker(function_type&& fun, Ts&&... args)
Ts&&... args)
: broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { } : broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { }
template<typename... Ts>
default_broker(fork_flag,
std::function<void (broker*, connection_handle)>&& fun,
connection_handle hdl,
Ts&&... args)
: broker{std::forward<Ts>(args)...}
, m_fun{std::bind(move(fun), std::placeholders::_1, hdl)} { }
void init() override { void init() override {
enqueue(nullptr, make_any_tuple(atom("INITMSG"))); enqueue(nullptr, make_any_tuple(atom("INITMSG")));
become( become(
...@@ -249,13 +259,13 @@ class broker::doorman : public broker::servant { ...@@ -249,13 +259,13 @@ class broker::doorman : public broker::servant {
doorman(broker_ptr parent, acceptor_uptr ptr) doorman(broker_ptr parent, acceptor_uptr ptr)
: super{move(parent), ptr->file_handle()} : super{move(parent), ptr->file_handle()}
//, m_ptr{move(ptr)}
, m_accept_msg{atom("IO_accept"), , m_accept_msg{atom("IO_accept"),
accept_handle::from_int(ptr->file_handle())} { accept_handle::from_int(ptr->file_handle())} {
m_ptr.reset(ptr.release()); m_ptr.reset(ptr.release());
} }
continue_reading_result continue_reading() override { continue_reading_result continue_reading() override {
CPPA_REQUIRE(parent() != nullptr);
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
for (;;) { for (;;) {
optional<stream_ptr_pair> opt{none}; optional<stream_ptr_pair> opt{none};
...@@ -427,10 +437,15 @@ local_actor_ptr init_and_launch(broker_ptr ptr) { ...@@ -427,10 +437,15 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
return move(ptr); return move(ptr);
} }
broker_ptr broker::from(std::function<void (broker*)> fun, broker_ptr broker::from(std::function<void (broker*, connection_handle)> fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out) { output_stream_ptr out) {
return make_counted<default_broker>(move(fun), move(in), move(out)); auto hdl = connection_handle::from_int(in->read_handle());
return make_counted<default_broker>(std::bind(move(fun),
std::placeholders::_1,
hdl),
move(in),
move(out));
} }
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) { broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
...@@ -460,12 +475,15 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) { ...@@ -460,12 +475,15 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) {
return id; return id;
} }
actor_ptr broker::fork(std::function<void (broker*)> fun, actor_ptr broker::fork(std::function<void (broker*, connection_handle)> fun,
const connection_handle& hdl) { connection_handle hdl) {
auto i = m_io.find(hdl); auto i = m_io.find(hdl);
if (i == m_io.end()) throw std::invalid_argument("invalid handle"); if (i == m_io.end()) throw std::invalid_argument("invalid handle");
scribe* sptr = i->second.get(); // non-owning pointer scribe* sptr = i->second.get(); // non-owning pointer
auto result = make_counted<default_broker>(move(fun), move(i->second)); auto result = make_counted<default_broker>(default_broker::fork_flag{},
move(fun),
hdl,
move(i->second));
init_and_launch(result); init_and_launch(result);
sptr->set_parent(result); // set new parent sptr->set_parent(result); // set new parent
m_io.erase(i); m_io.erase(i);
......
...@@ -80,20 +80,20 @@ void pong() { ...@@ -80,20 +80,20 @@ void pong() {
); );
} }
void peer(io::broker* thisptr, const actor_ptr& buddy) { void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy) {
self->monitor(buddy); self->monitor(buddy);
if (thisptr->num_connections() == 0) { if (thisptr->num_connections() == 0) {
cout << "num_connections() != 1" << endl; cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1"); throw std::logic_error("num_connections() != 1");
} }
thisptr->for_each_connection([=](io::connection_handle hdl) { //thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->receive_policy(hdl, io::broker::exactly, message_size); // thisptr->receive_policy(hdl, io::broker::exactly, message_size);
}); //});
auto write = [=](atom_value type, int value) { auto write = [=](atom_value type, int value) {
thisptr->for_each_connection([=](io::connection_handle hdl) { //thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->write(hdl, sizeof(type), &type); thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value); thisptr->write(hdl, sizeof(value), &value);
}); //});
}; };
become ( become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) { on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
...@@ -122,6 +122,7 @@ void peer(io::broker* thisptr, const actor_ptr& buddy) { ...@@ -122,6 +122,7 @@ void peer(io::broker* thisptr, const actor_ptr& buddy) {
void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) { void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
become ( become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) { on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) {
CPPA_CHECKPOINT();
CPPA_LOGF_INFO("received IO_accept"); CPPA_LOGF_INFO("received IO_accept");
thisptr->fork(peer, hdl, buddy); thisptr->fork(peer, hdl, buddy);
self->quit(); self->quit();
......
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