Commit 181c56ec authored by Dominik Charousset's avatar Dominik Charousset

Add try-accept-connection feature to test mpx

parent eb2a9810
...@@ -132,7 +132,10 @@ public: ...@@ -132,7 +132,10 @@ public:
bool has_pending_scribe(std::string x, uint16_t y); bool has_pending_scribe(std::string x, uint16_t y);
/// Accepts a pending connect on `hdl`. /// Accepts a pending connect on `hdl`.
bool accept_connection(accept_handle hdl); void accept_connection(accept_handle hdl);
/// Tries to accept a pending connection.
bool try_accept_connection();
/// Poll data on all scribes. /// Poll data on all scribes.
bool read_data(); bool read_data();
......
...@@ -114,6 +114,7 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) { ...@@ -114,6 +114,7 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) {
guard_type guard{mx_}; guard_type guard{mx_};
impl_ptr(hdl) = sptr; impl_ptr(hdl) = sptr;
} }
CAF_LOG_INFO("opened connection" << sptr->hdl());
return sptr; return sptr;
} }
...@@ -159,6 +160,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) { ...@@ -159,6 +160,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) {
ch = i->second; ch = i->second;
pc.erase(i); pc.erase(i);
} }
CAF_LOG_INFO("accepted connection" << ch << "on acceptor" << hdl());
parent()->add_scribe(mpx_->new_scribe(ch)); parent()->add_scribe(mpx_->new_scribe(ch));
return doorman::new_connection(mpx_, ch); return doorman::new_connection(mpx_, ch);
} }
...@@ -192,6 +194,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) { ...@@ -192,6 +194,7 @@ doorman_ptr test_multiplexer::new_doorman(accept_handle hdl, uint16_t port) {
ref.ptr = dptr; ref.ptr = dptr;
ref.port = port; ref.port = port;
} }
CAF_LOG_INFO("opened port" << port << "on acceptor" << hdl);
return dptr; return dptr;
} }
...@@ -354,7 +357,11 @@ void test_multiplexer::prepare_connection(accept_handle src, ...@@ -354,7 +357,11 @@ void test_multiplexer::prepare_connection(accept_handle src,
auto res2 = peer.scribe_data_.emplace(peer_hdl, scribe_data{output, input}); auto res2 = peer.scribe_data_.emplace(peer_hdl, scribe_data{output, input});
if (!res2.second) if (!res2.second)
CAF_RAISE_ERROR("prepare_connection: peer handle already in use"); CAF_RAISE_ERROR("prepare_connection: peer handle already in use");
provide_acceptor(port, src); CAF_LOG_INFO("acceptor" << src << "has connection" << hdl
<< "ready for incoming connect from" << host << ":"
<< port << "from peer with connection handle" << peer_hdl);
if (doormen_.count(port) == 0)
provide_acceptor(port, src);
add_pending_connect(src, hdl); add_pending_connect(src, hdl);
peer.provide_scribe(std::move(host), port, peer_hdl); peer.provide_scribe(std::move(host), port, peer_hdl);
} }
...@@ -370,7 +377,7 @@ bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) { ...@@ -370,7 +377,7 @@ bool test_multiplexer::has_pending_scribe(std::string x, uint16_t y) {
return scribes_.count(std::make_pair(std::move(x), y)) > 0; return scribes_.count(std::make_pair(std::move(x), y)) > 0;
} }
bool test_multiplexer::accept_connection(accept_handle hdl) { void test_multiplexer::accept_connection(accept_handle hdl) {
CAF_ASSERT(std::this_thread::get_id() == tid_); CAF_ASSERT(std::this_thread::get_id() == tid_);
CAF_LOG_TRACE(CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(hdl));
// Filled / initialized in the critical section. // Filled / initialized in the critical section.
...@@ -382,7 +389,21 @@ bool test_multiplexer::accept_connection(accept_handle hdl) { ...@@ -382,7 +389,21 @@ bool test_multiplexer::accept_connection(accept_handle hdl) {
CAF_ASSERT(dd->ptr != nullptr); CAF_ASSERT(dd->ptr != nullptr);
if (!dd->ptr->new_connection()) if (!dd->ptr->new_connection())
dd->passive_mode = true; dd->passive_mode = true;
return true; }
bool test_multiplexer::try_accept_connection() {
CAF_ASSERT(std::this_thread::get_id() == tid_);
// Filled / initialized in the critical section.
std::vector<doorman_data*> doormen;
{ // Access `doorman_data_` and `pending_connects_` while holding `mx_`.
guard_type guard{mx_};
doormen.reserve(doorman_data_.size());
for (auto& kvp : doorman_data_)
doormen.emplace_back(&kvp.second);
}
// Try accepting a new connection on all existing doorman.
return std::any_of(doormen.begin(), doormen.end(),
[](doorman_data* x) { return x->ptr->new_connection(); });
} }
bool test_multiplexer::read_data() { bool test_multiplexer::read_data() {
......
...@@ -285,7 +285,7 @@ public: ...@@ -285,7 +285,7 @@ public:
<< ", acceptor ID = " << src.id()); << ", acceptor ID = " << src.id());
auto hdl = n.connection; auto hdl = n.connection;
mpx_->add_pending_connect(src, hdl); mpx_->add_pending_connect(src, hdl);
CAF_REQUIRE(mpx_->accept_connection(src)); mpx_->accept_connection(src);
// technically, the server handshake arrives // technically, the server handshake arrives
// before we send the client handshake // before we send the client handshake
mock(hdl, mock(hdl,
......
...@@ -181,7 +181,7 @@ public: ...@@ -181,7 +181,7 @@ public:
aut_ptr_->add_doorman(mpx_->new_doorman(acceptor_, 1u)); aut_ptr_->add_doorman(mpx_->new_doorman(acceptor_, 1u));
// "open" a new connection to our server // "open" a new connection to our server
mpx_->add_pending_connect(acceptor_, connection_); mpx_->add_pending_connect(acceptor_, connection_);
CAF_REQUIRE(mpx_->accept_connection(acceptor_)); mpx_->accept_connection(acceptor_);
} }
~fixture() { ~fixture() {
......
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