Commit 9ed4db36 authored by Dominik Charousset's avatar Dominik Charousset

Fix possible race in test multiplexer

parent ab865a9c
...@@ -109,7 +109,7 @@ public: ...@@ -109,7 +109,7 @@ public:
using pending_scribes_map = std::map<std::pair<std::string, uint16_t>, using pending_scribes_map = std::map<std::pair<std::string, uint16_t>,
connection_handle>; connection_handle>;
pending_scribes_map& pending_scribes(); bool has_pending_scribe(std::string host, uint16_t port);
/// Accepts a pending connect on `hdl`. /// Accepts a pending connect on `hdl`.
void accept_connection(accept_handle hdl); void accept_connection(accept_handle hdl);
...@@ -157,6 +157,7 @@ private: ...@@ -157,6 +157,7 @@ private:
intrusive_ptr<doorman> ptr; intrusive_ptr<doorman> ptr;
}; };
// guards resumables_ and scribes_
std::mutex mx_; std::mutex mx_;
std::condition_variable cv_; std::condition_variable cv_;
std::list<resumable_ptr> resumables_; std::list<resumable_ptr> resumables_;
......
...@@ -40,6 +40,7 @@ test_multiplexer::~test_multiplexer() { ...@@ -40,6 +40,7 @@ test_multiplexer::~test_multiplexer() {
connection_handle connection_handle
test_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port_hint) { test_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port_hint) {
guard_type guard{mx_};
connection_handle result; connection_handle result;
auto i = scribes_.find(std::make_pair(host, port_hint)); auto i = scribes_.find(std::make_pair(host, port_hint));
if (i != scribes_.end()) { if (i != scribes_.end()) {
...@@ -199,6 +200,7 @@ void test_multiplexer::run() { ...@@ -199,6 +200,7 @@ void test_multiplexer::run() {
void test_multiplexer::provide_scribe(std::string host, uint16_t desired_port, void test_multiplexer::provide_scribe(std::string host, uint16_t desired_port,
connection_handle hdl) { connection_handle hdl) {
guard_type guard{mx_};
scribes_.emplace(std::make_pair(std::move(host), desired_port), hdl); scribes_.emplace(std::make_pair(std::move(host), desired_port), hdl);
} }
...@@ -262,8 +264,9 @@ test_multiplexer::pending_connects_map& test_multiplexer::pending_connects() { ...@@ -262,8 +264,9 @@ test_multiplexer::pending_connects_map& test_multiplexer::pending_connects() {
return pending_connects_; return pending_connects_;
} }
test_multiplexer::pending_scribes_map& test_multiplexer::pending_scribes() { bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) {
return scribes_; guard_type guard{mx_};
return scribes_.count(std::make_pair(std::move(x), y)) > 0;
} }
void test_multiplexer::accept_connection(accept_handle hdl) { void test_multiplexer::accept_connection(accept_handle hdl) {
......
...@@ -579,7 +579,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -579,7 +579,7 @@ CAF_TEST(remote_actor_and_send) {
constexpr const char* lo = "localhost"; constexpr const char* lo = "localhost";
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
mpx()->provide_scribe(lo, 4242, remote_hdl(0)); mpx()->provide_scribe(lo, 4242, remote_hdl(0));
CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 1); CAF_REQUIRE(mpx()->has_pending_scribe(lo, 4242));
auto mm1 = system.middleman().actor_handle(); auto mm1 = system.middleman().actor_handle();
actor result{unsafe_actor_handle_init}; actor result{unsafe_actor_handle_init};
auto f = self()->request(mm1, infinite, auto f = self()->request(mm1, infinite,
...@@ -587,7 +587,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -587,7 +587,7 @@ CAF_TEST(remote_actor_and_send) {
// wait until BASP broker has received and processed the connect message // wait until BASP broker has received and processed the connect message
while (! aut()->valid(remote_hdl(0))) while (! aut()->valid(remote_hdl(0)))
mpx()->exec_runnable(); mpx()->exec_runnable();
CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 0); CAF_REQUIRE(! mpx()->has_pending_scribe(lo, 4242));
// build a fake server handshake containing the id of our first pseudo actor // build a fake server handshake containing the id of our first pseudo actor
CAF_MESSAGE("server handshake => client handshake + proxy announcement"); CAF_MESSAGE("server handshake => client handshake + proxy announcement");
auto na = registry()->named_actors(); auto na = registry()->named_actors();
...@@ -760,8 +760,7 @@ CAF_TEST(automatic_connection) { ...@@ -760,8 +760,7 @@ CAF_TEST(automatic_connection) {
// (this node receives a message from jupiter via mars and responds via mars, // (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
mpx()->provide_scribe("jupiter", 8080, remote_hdl(0)); mpx()->provide_scribe("jupiter", 8080, remote_hdl(0));
CAF_CHECK_EQUAL(mpx()->pending_scribes().count(make_pair("jupiter", 8080)), CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
1u);
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242); auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax); mpx()->provide_acceptor(4242, ax);
...@@ -816,7 +815,7 @@ CAF_TEST(automatic_connection) { ...@@ -816,7 +815,7 @@ CAF_TEST(automatic_connection) {
make_message(uint16_t{8080}, std::move(res)))); make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and // our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker // send the scribe handle over to the BASP broker
while (mpx()->pending_scribes().count(make_pair("jupiter", 8080)) != 0) while (mpx()->has_pending_scribe("jupiter", 8080))
mpx()->flush_runnables(); mpx()->flush_runnables();
CAF_REQUIRE(mpx()->output_buffer(remote_hdl(1)).size() == 0); CAF_REQUIRE(mpx()->output_buffer(remote_hdl(1)).size() == 0);
// send handshake from jupiter // send handshake from jupiter
......
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