| 223 | } |
| 224 | |
| 225 | void fields::step_boundaries(field_type ft) { |
| 226 | connect_chunks(); // re-connect if !chunk_connections_valid |
| 227 | |
| 228 | { |
| 229 | // Initiate receive operations as early as possible. |
| 230 | std::unique_ptr<comms_manager> manager = create_comms_manager(); |
| 231 | |
| 232 | const auto &sequence = comms_sequence_for_field[ft]; |
| 233 | for (const comms_operation &op : sequence.receive_ops) { |
| 234 | if (chunks[op.other_chunk_idx]->is_mine()) { continue; } |
| 235 | chunk_pair comm_pair{op.other_chunk_idx, op.my_chunk_idx}; |
| 236 | comms_manager::receive_callback cb = [this, ft, comm_pair]() { |
| 237 | process_incoming_chunk_data(ft, comm_pair); |
| 238 | }; |
| 239 | manager->receive_real_async(comm_blocks[ft][op.pair_idx], static_cast<int>(op.transfer_size), |
| 240 | op.other_proc_id, op.tag, cb); |
| 241 | } |
| 242 | |
| 243 | // Do the metals first! |
| 244 | for (int i = 0; i < num_chunks; i++) |
| 245 | if (chunks[i]->is_mine()) chunks[i]->zero_metal(ft); |
| 246 | |
| 247 | // Copy outgoing data into buffers while following the predefined sequence of comms operations. |
| 248 | // Trigger the asynchronous send immediately once the outgoing comms buffer has been filled. |
| 249 | am_now_working_on(Boundaries); |
| 250 | |
| 251 | for (const comms_operation &op : sequence.send_ops) { |
| 252 | const std::pair<int, int> comm_pair{op.my_chunk_idx, op.other_chunk_idx}; |
| 253 | const int pair_idx = op.pair_idx; |
| 254 | |
| 255 | realnum *outgoing_comm_block = comm_blocks[ft][pair_idx]; |
| 256 | for (connect_phase ip : all_connect_phases) { |
| 257 | const comms_key key = {ft, ip, comm_pair}; |
| 258 | const size_t pair_comm_size = get_comm_size(key); |
| 259 | if (pair_comm_size) { |
| 260 | const std::vector<realnum *> &outgoing_connection = |
| 261 | chunks[op.my_chunk_idx]->connections_out.at(key); |
| 262 | for (size_t n = 0; n < pair_comm_size; ++n) { |
| 263 | outgoing_comm_block[n] = *(outgoing_connection[n]); |
| 264 | } |
| 265 | outgoing_comm_block += pair_comm_size; |
| 266 | } |
| 267 | } |
| 268 | if (chunks[op.other_chunk_idx]->is_mine()) { continue; } |
| 269 | manager->send_real_async(comm_blocks[ft][pair_idx], static_cast<int>(op.transfer_size), |
| 270 | op.other_proc_id, op.tag); |
| 271 | } |
| 272 | |
| 273 | // Process local transfers, which do not depend on a communication mechanism across nodes. |
| 274 | for (const comms_operation &op : sequence.receive_ops) { |
| 275 | if (chunks[op.other_chunk_idx]->is_mine()) { |
| 276 | process_incoming_chunk_data(ft, {op.other_chunk_idx, op.my_chunk_idx}); |
| 277 | } |
| 278 | } |
| 279 | finished_working(); |
| 280 | |
| 281 | am_now_working_on(MpiOneTime); |
| 282 | // Let the communication manager drop out of scope to complete all outstanding requests. |
no test coverage detected