Commit 98e2a20a authored by Dominik Charousset's avatar Dominik Charousset

parallelized connection establishments

parent ac81fea9
...@@ -296,52 +296,57 @@ void client_mode(Iterator first, Iterator last) { ...@@ -296,52 +296,57 @@ void client_mode(Iterator first, Iterator last) {
for (size_t j = 0; j < remotes.size(); ++j) { for (size_t j = 0; j < remotes.size(); ++j) {
if (i != j) { if (i != j) {
auto& r = remotes[j]; auto& r = remotes[j];
send(remote_actors[i], send(remote_actors[i], atom("add_pong"), r.first, r.second);
atom("add_pong"), r.first, r.second);
receive (
on(atom("ok")) >> []() {
},
on(atom("error"), arg_match) >> [&](string const& str) {
cout << "error on node " << i << ": " << str << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw std::logic_error("");
},
others() >> []() {
cout << "expected {ok|error}, received: "
<< to_string(self->last_dequeued())
<< endl;
throw std::logic_error("");
},
after(std::chrono::seconds(10)) >> [&]() {
cout << "remote didn't answer within 10sec." << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw std::logic_error("");
}
);
} }
} }
} }
{ // collect {ok} messages
size_t i = 0;
size_t end = remote_actors.size() * (remote_actors.size() - 1);
receive_for(i, end) (
on(atom("ok")) >> []() {
},
on(atom("error"), arg_match) >> [&](string const& str) {
cout << "error: " << str << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw std::logic_error("");
},
others() >> []() {
cout << "expected {ok|error}, received: "
<< to_string(self->last_dequeued())
<< endl;
throw std::logic_error("");
},
after(std::chrono::seconds(10)) >> [&]() {
cout << "remote didn't answer within 10sec." << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw std::logic_error("");
}
);
}
// kickoff // kickoff
//cout << "setup done" << endl; //cout << "setup done" << endl;
//cout << "kickoff, init value = " << init_value << endl; //cout << "kickoff, init value = " << init_value << endl;
for (auto& r : remote_actors) { for (auto& r : remote_actors) {
send(r, atom("kickoff"), init_value); send(r, atom("kickoff"), init_value);
} }
size_t i = 0; { // collect {done} messages
size_t num_pings = remote_actors.size() * (remote_actors.size() - 1); size_t i = 0;
receive_for(i, num_pings) ( size_t end = remote_actors.size() * (remote_actors.size() - 1);
on(atom("done")) >> []() { receive_for(i, end) (
//cout << "...done..." << endl; on(atom("done")) >> []() {
}, //cout << "...done..." << endl;
others() >> []() { },
cout << "unexpected: " << to_string(self->last_dequeued()) << endl; others() >> []() {
throw std::logic_error(""); cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
} throw std::logic_error("");
); }
);
}
await_all_others_done(); await_all_others_done();
} }
......
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