Commit 6221cc05 authored by Dominik Charousset's avatar Dominik Charousset

Work around broken `std::promise` implementation

parent e1da00b4
...@@ -66,9 +66,8 @@ class basp_broker : public broker, public actor_namespace::backend { ...@@ -66,9 +66,8 @@ class basp_broker : public broker, public actor_namespace::backend {
struct client_handshake_data { struct client_handshake_data {
id_type remote_id; id_type remote_id;
std::promise<abstract_actor_ptr>* result; actor client;
const std::set<std::string>* expected_ifs; std::set<std::string> expected_ifs;
}; };
void init_client(connection_handle hdl, client_handshake_data* data); void init_client(connection_handle hdl, client_handshake_data* data);
...@@ -117,9 +116,6 @@ class basp_broker : public broker, public actor_namespace::backend { ...@@ -117,9 +116,6 @@ class basp_broker : public broker, public actor_namespace::backend {
void write(binary_serializer& bs, const basp::header& msg); void write(binary_serializer& bs, const basp::header& msg);
void send(const connection_context& ctx, const basp::header& msg,
message payload);
void send_kill_proxy_instance(const id_type& nid, actor_id aid, void send_kill_proxy_instance(const id_type& nid, actor_id aid,
uint32_t reason); uint32_t reason);
......
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs, abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port); const std::string& host, uint16_t port);
template <class List> template <class List>
......
...@@ -74,8 +74,7 @@ behavior basp_broker::make_behavior() { ...@@ -74,8 +74,7 @@ behavior basp_broker::make_behavior() {
if (j != m_ctx.end()) { if (j != m_ctx.end()) {
auto hd = j->second.handshake_data; auto hd = j->second.handshake_data;
if (hd) { if (hd) {
network_error err{"disconnect during handshake"}; send(hd->client, atom("ERROR"), "disconnect during handshake");
hd->result->set_exception(std::make_exception_ptr(err));
} }
m_ctx.erase(j); m_ctx.erase(j);
} }
...@@ -435,7 +434,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -435,7 +434,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
auto str = bd.read<string>(); auto str = bd.read<string>();
remote_ifs.insert(std::move(str)); remote_ifs.insert(std::move(str));
} }
auto& ifs = *(ctx.handshake_data->expected_ifs); auto& ifs = ctx.handshake_data->expected_ifs;
if (!std::includes(ifs.begin(), ifs.end(), if (!std::includes(ifs.begin(), ifs.end(),
remote_ifs.begin(), remote_ifs.end())) { remote_ifs.begin(), remote_ifs.end())) {
auto tostr = [](const std::set<string>& what) -> string { auto tostr = [](const std::set<string>& what) -> string {
...@@ -476,15 +475,14 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -476,15 +475,14 @@ basp_broker::handle_basp_header(connection_context& ctx,
+ iface_str; + iface_str;
} }
// abort with error // abort with error
std::runtime_error err{error_msg}; send(ctx.handshake_data->client, atom("ERROR"), std::move(error_msg));
ctx.handshake_data->result->set_exception(std::make_exception_ptr(err));
return close_connection; return close_connection;
} }
auto nid = ctx.handshake_data->remote_id; auto nid = ctx.handshake_data->remote_id;
if (nid == node()) { if (nid == node()) {
CAF_LOG_INFO("incoming connection from self: drop connection"); CAF_LOG_INFO("incoming connection from self: drop connection");
auto res = detail::singletons::get_actor_registry()->get(remote_aid); auto res = detail::singletons::get_actor_registry()->get(remote_aid);
ctx.handshake_data->result->set_value(std::move(res)); send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(res));
ctx.handshake_data = nullptr; ctx.handshake_data = nullptr;
return close_connection; return close_connection;
} }
...@@ -493,7 +491,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -493,7 +491,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
<< " (re-use old one)"); << " (re-use old one)");
auto proxy = m_namespace.get_or_put(nid, remote_aid); auto proxy = m_namespace.get_or_put(nid, remote_aid);
// discard this peer; there's already an open connection // discard this peer; there's already an open connection
ctx.handshake_data->result->set_value(std::move(proxy)); send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(proxy));
ctx.handshake_data = nullptr; ctx.handshake_data = nullptr;
return close_connection; return close_connection;
} }
...@@ -506,7 +504,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -506,7 +504,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
// prepare to receive messages // prepare to receive messages
auto proxy = m_namespace.get_or_put(nid, remote_aid); auto proxy = m_namespace.get_or_put(nid, remote_aid);
ctx.published_actor = proxy; ctx.published_actor = proxy;
ctx.handshake_data->result->set_value(std::move(proxy)); send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(proxy));
ctx.handshake_data = nullptr; ctx.handshake_data = nullptr;
parent().notify<hook::new_connection_established>(nid); parent().notify<hook::new_connection_established>(nid);
break; break;
......
...@@ -17,15 +17,16 @@ ...@@ -17,15 +17,16 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <future> #include "caf/io/publish.hpp"
#include "caf/send.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
#include "caf/io/publish.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
...@@ -36,21 +37,35 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -36,21 +37,35 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) { const char* in, bool reuse_addr) {
using namespace detail; using namespace detail;
auto mm = middleman::instance(); auto mm = middleman::instance();
std::promise<bool> res; scoped_actor self;
actor selfhdl = self;
mm->run_later([&] { mm->run_later([&] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
try { try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr); auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr);
bro->add_published_actor(hdl, whom, port); bro->add_published_actor(hdl, whom, port);
mm->notify<hook::actor_published>(whom->address(), port); mm->notify<hook::actor_published>(whom->address(), port);
res.set_value(true); anon_send(selfhdl, atom("OK"));
}
catch (bind_failure& e) {
anon_send(selfhdl, atom("BIND_FAIL"), e.what());
} }
catch (...) { catch (network_error& e) {
res.set_exception(std::current_exception()); anon_send(selfhdl, atom("ERROR"), e.what());
} }
}); });
// block caller and re-throw exception here in case of an error // block caller and re-throw exception here in case of an error
res.get_future().get(); self->receive(
on(atom("OK")) >> [] {
// success
},
on(atom("BIND_FAIL"), arg_match) >> [](std::string& str) {
throw bind_failure(std::move(str));
},
on(atom("ERROR"), arg_match) >> [](std::string& str) {
throw network_error(std::move(str));
}
);
} }
} // namespace io } // namespace io
......
...@@ -27,6 +27,8 @@ ...@@ -27,6 +27,8 @@
#include <cstdint> #include <cstdint>
#include <algorithm> #include <algorithm>
#include "caf/send.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
...@@ -39,28 +41,39 @@ ...@@ -39,28 +41,39 @@
namespace caf { namespace caf {
namespace io { namespace io {
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs, abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
auto mm = middleman::instance(); auto mm = middleman::instance();
std::promise<abstract_actor_ptr> res; scoped_actor self;
basp_broker::client_handshake_data hdata{invalid_node_id, &res, &ifs}; actor selfhdl = self;
basp_broker::client_handshake_data hdata{invalid_node_id,
selfhdl, std::move(ifs)};
mm->run_later([&] { mm->run_later([&] {
std::exception_ptr eptr; std::string err;
try { try {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
auto hdl = mm->backend().add_tcp_scribe(bro.get(), host, port); auto hdl = mm->backend().add_tcp_scribe(bro.get(), host, port);
bro->init_client(hdl, &hdata); bro->init_client(hdl, &hdata);
} }
catch (...) { catch (std::exception& e) {
eptr = std::current_exception(); err = e.what();
} }
// accessing `res` inside the catch block triggers // accessing variables from the outer scope inside the
// a silly compiler error on GCC 4.7 // catch block triggers a silly compiler error on GCC 4.7
if (eptr) { if (!err.empty()) {
res.set_exception(std::move(eptr)); anon_send(selfhdl, atom("ERROR"), std::move(err));
} }
}); });
return res.get_future().get(); abstract_actor_ptr result;
self->receive(
on(atom("OK"), arg_match) >> [&](const actor& res) {
result = actor_cast<abstract_actor_ptr>(res);
},
on(atom("ERROR"), arg_match) >> [](std::string& str) {
throw network_error(std::move(str));
}
);
return result;
} }
} // namespace io } // namespace io
......
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