Commit a78b7351 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #428 from ufownl/topic/actor-system

Fix thread sync issue in test `io_basp[_asio]`
parents 779d9171 1dec42fa
...@@ -327,7 +327,6 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -327,7 +327,6 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
learned_new_node(nid); learned_new_node(nid);
if (! enable_automatic_connections) if (! enable_automatic_connections)
return; return;
actor bb = self; // a handle for our helper back to this BASP broker
// this member function gets only called once, after adding a new // this member function gets only called once, after adding a new
// indirect connection to the routing table; hence, spawning // indirect connection to the routing table; hence, spawning
// our helper here exactly once and there is no need to track // our helper here exactly once and there is no need to track
...@@ -353,7 +352,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -353,7 +352,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
// gotcha! send scribe to our BASP broker // gotcha! send scribe to our BASP broker
// to initiate handshake etc. // to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr)); CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
helper->send(bb, connect_atom::value, hdl, port); helper->send(s, connect_atom::value, hdl, port);
return; return;
} }
catch (...) { catch (...) {
......
...@@ -89,6 +89,7 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr, ...@@ -89,6 +89,7 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr,
private: private:
test_multiplexer* mpx_; test_multiplexer* mpx_;
}; };
CAF_LOG_TRACE(CAF_ARG(hdl));
auto sptr = make_counted<impl>(ptr, hdl, this); auto sptr = make_counted<impl>(ptr, hdl, this);
impl_ptr(hdl) = sptr; impl_ptr(hdl) = sptr;
ptr->add_scribe(sptr); ptr->add_scribe(sptr);
...@@ -266,8 +267,8 @@ void test_multiplexer::accept_connection(accept_handle hdl) { ...@@ -266,8 +267,8 @@ void test_multiplexer::accept_connection(accept_handle hdl) {
void test_multiplexer::read_data(connection_handle hdl) { void test_multiplexer::read_data(connection_handle hdl) {
flush_runnables(); flush_runnables();
scribe_data& sd = scribe_data_[hdl]; scribe_data& sd = scribe_data_[hdl];
if (sd.ptr == nullptr) while (sd.ptr == nullptr)
throw std::logic_error("scribe data contains a nullptr"); exec_runnable();
switch (sd.recv_conf.first) { switch (sd.recv_conf.first) {
case receive_policy_flag::exactly: case receive_policy_flag::exactly:
while (sd.xbuf.size() >= sd.recv_conf.second) { while (sd.xbuf.size() >= sd.recv_conf.second) {
...@@ -302,12 +303,14 @@ void test_multiplexer::read_data(connection_handle hdl) { ...@@ -302,12 +303,14 @@ void test_multiplexer::read_data(connection_handle hdl) {
void test_multiplexer::virtual_send(connection_handle hdl, void test_multiplexer::virtual_send(connection_handle hdl,
const buffer_type& buf) { const buffer_type& buf) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto& vb = virtual_network_buffer(hdl); auto& vb = virtual_network_buffer(hdl);
vb.insert(vb.end(), buf.begin(), buf.end()); vb.insert(vb.end(), buf.begin(), buf.end());
read_data(hdl); read_data(hdl);
} }
void test_multiplexer::exec_runnable() { void test_multiplexer::exec_runnable() {
CAF_LOG_TRACE("");
resumable_ptr ptr; resumable_ptr ptr;
{ // critical section { // critical section
guard_type guard{mx_}; guard_type guard{mx_};
...@@ -320,6 +323,7 @@ void test_multiplexer::exec_runnable() { ...@@ -320,6 +323,7 @@ void test_multiplexer::exec_runnable() {
} }
bool test_multiplexer::try_exec_runnable() { bool test_multiplexer::try_exec_runnable() {
CAF_LOG_TRACE("");
resumable_ptr ptr; resumable_ptr ptr;
{ // critical section { // critical section
guard_type guard{mx_}; guard_type guard{mx_};
...@@ -333,6 +337,7 @@ bool test_multiplexer::try_exec_runnable() { ...@@ -333,6 +337,7 @@ bool test_multiplexer::try_exec_runnable() {
} }
void test_multiplexer::flush_runnables() { void test_multiplexer::flush_runnables() {
CAF_LOG_TRACE("");
// execute runnables in bursts, pick a small size to // execute runnables in bursts, pick a small size to
// minimize time in the critical section // minimize time in the critical section
constexpr size_t max_runnable_count = 8; constexpr size_t max_runnable_count = 8;
...@@ -357,6 +362,7 @@ void test_multiplexer::flush_runnables() { ...@@ -357,6 +362,7 @@ void test_multiplexer::flush_runnables() {
void test_multiplexer::exec_later(resumable* ptr) { void test_multiplexer::exec_later(resumable* ptr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0); CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0);
CAF_LOG_TRACE("");
switch (ptr->subtype()) { switch (ptr->subtype()) {
case resumable::io_actor: case resumable::io_actor:
case resumable::function_object: { case resumable::function_object: {
...@@ -375,6 +381,7 @@ void test_multiplexer::exec_later(resumable* ptr) { ...@@ -375,6 +381,7 @@ void test_multiplexer::exec_later(resumable* ptr) {
void test_multiplexer::exec(resumable_ptr& ptr) { void test_multiplexer::exec(resumable_ptr& ptr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0); CAF_ASSERT(ptr->as_ref_counted_ptr()->get_reference_count() > 0);
CAF_LOG_TRACE("");
switch (ptr->resume(this, 1)) { switch (ptr->resume(this, 1)) {
case resumable::resume_later: case resumable::resume_later:
exec_later(ptr.get()); exec_later(ptr.get());
......
...@@ -476,7 +476,9 @@ CAF_TEST(remote_address_and_port) { ...@@ -476,7 +476,9 @@ CAF_TEST(remote_address_and_port) {
connect_node(1); connect_node(1);
auto mm = system.middleman().actor_handle(); auto mm = system.middleman().actor_handle();
self()->send(mm, get_atom::value, remote_node(1)); self()->send(mm, get_atom::value, remote_node(1));
mpx()->exec_runnable(); do {
mpx()->exec_runnable();
} while (! self()->has_next_message());
self()->receive( self()->receive(
[&](const node_id& nid, const std::string& addr, uint16_t port) { [&](const node_id& nid, const std::string& addr, uint16_t port) {
CAF_CHECK(nid == remote_node(1)); CAF_CHECK(nid == remote_node(1));
......
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