| 163 | } |
| 164 | |
| 165 | void Linkers::Construct() { |
| 166 | // save ranks that need to connect with |
| 167 | std::unordered_map<int, int> need_connect; |
| 168 | for (int i = 0; i < num_machines_; ++i) { |
| 169 | if (i != rank_) { |
| 170 | need_connect[i] = 1; |
| 171 | } |
| 172 | } |
| 173 | int need_connect_cnt = 0; |
| 174 | int incoming_cnt = 0; |
| 175 | for (auto it = need_connect.begin(); it != need_connect.end(); ++it) { |
| 176 | int machine_rank = it->first; |
| 177 | if (machine_rank >= 0 && machine_rank != rank_) { |
| 178 | ++need_connect_cnt; |
| 179 | } |
| 180 | if (machine_rank < rank_) { |
| 181 | ++incoming_cnt; |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // start listener |
| 186 | listener_->SetTimeout(socket_timeout_); |
| 187 | listener_->Listen(incoming_cnt); |
| 188 | std::thread listen_thread(&Linkers::ListenThread, this, incoming_cnt); |
| 189 | const int connect_fail_retry_cnt = 20; |
| 190 | const int connect_fail_retry_first_delay_interval = 200; // 0.2 s |
| 191 | const float connect_fail_retry_delay_factor = 1.3f; |
| 192 | // start connect |
| 193 | for (auto it = need_connect.begin(); it != need_connect.end(); ++it) { |
| 194 | int out_rank = it->first; |
| 195 | // let smaller rank connect to larger rank |
| 196 | if (out_rank > rank_) { |
| 197 | TcpSocket cur_socket; |
| 198 | int connect_fail_delay_time = connect_fail_retry_first_delay_interval; |
| 199 | for (int i = 0; i < connect_fail_retry_cnt; ++i) { |
| 200 | if (cur_socket.Connect(client_ips_[out_rank].c_str(), client_ports_[out_rank])) { |
| 201 | break; |
| 202 | } else { |
| 203 | Log::Warning("Connecting to rank %d failed, waiting for %d milliseconds", out_rank, connect_fail_delay_time); |
| 204 | std::this_thread::sleep_for(std::chrono::milliseconds(connect_fail_delay_time)); |
| 205 | connect_fail_delay_time = static_cast<int>(connect_fail_delay_time * connect_fail_retry_delay_factor); |
| 206 | } |
| 207 | } |
| 208 | // send local rank |
| 209 | cur_socket.Send(reinterpret_cast<const char*>(&rank_), sizeof(rank_)); |
| 210 | SetLinker(out_rank, cur_socket); |
| 211 | } |
| 212 | } |
| 213 | // wait for listener |
| 214 | listen_thread.join(); |
| 215 | // print connected linkers |
| 216 | PrintLinkers(); |
| 217 | } |
| 218 | |
| 219 | bool Linkers::CheckLinker(int rank) { |
| 220 | if (linkers_[rank] == nullptr || linkers_[rank]->IsClosed()) { |