source: XIOS/dev/branch_yushan/extern/src_ep_dev/ep_intercomm_kernel.cpp @ 1067

Last change on this file since 1067 was 1067, checked in by yushan, 7 years ago

server mode OK tested with test_complete
File size: 25.2 KB
Line 
1#include "ep_lib.hpp"
2#include <mpi.h>
3#include "ep_declaration.hpp"
4
5using namespace std;
6
7
8namespace ep_lib
9{
10  int MPI_Intercomm_create_kernel(MPI_Comm local_comm, int local_leader, MPI_Comm peer_comm, int remote_leader, int tag, MPI_Comm *newintercomm)
11  {
12
13    int ep_rank, ep_rank_loc, mpi_rank;
14    int ep_size, num_ep, mpi_size;
15
16    ep_rank = local_comm.ep_comm_ptr->size_rank_info[0].first;
17    ep_rank_loc = local_comm.ep_comm_ptr->size_rank_info[1].first;
18    mpi_rank = local_comm.ep_comm_ptr->size_rank_info[2].first;
19    ep_size = local_comm.ep_comm_ptr->size_rank_info[0].second;
20    num_ep = local_comm.ep_comm_ptr->size_rank_info[1].second;
21    mpi_size = local_comm.ep_comm_ptr->size_rank_info[2].second;
22
23    std::vector<int> rank_info[4];  //! 0->rank_in_world of local_comm,  1->rank_in_local_parent of local_comm
24                                    //! 2->rank_in_world of remote_comm, 3->rank_in_local_parent of remote_comm
25
26    int rank_in_world;
27    int rank_in_local_parent;
28
29    int rank_in_peer_mpi[2];
30
31    int local_ep_size = ep_size;
32    int remote_ep_size;
33
34
35    ::MPI_Comm local_mpi_comm = static_cast< ::MPI_Comm>(local_comm.mpi_comm);
36
37   
38    ::MPI_Comm_rank(MPI_COMM_WORLD_STD, &rank_in_world);
39    ::MPI_Comm_rank(static_cast< ::MPI_Comm>(local_comm.mpi_comm), &rank_in_local_parent);
40   
41
42    bool is_proc_master = false;
43    bool is_local_leader = false;
44    bool is_final_master = false;
45
46
47    if(ep_rank == local_leader) { is_proc_master = true; is_local_leader = true; is_final_master = true;}
48    if(ep_rank_loc == 0 && mpi_rank != local_comm.rank_map->at(local_leader).second) is_proc_master = true;
49
50
51    int size_info[4]; //! used for choose size of rank_info 0-> mpi_size of local_comm, 1-> mpi_size of remote_comm
52
53    int leader_info[4]; //! 0->world rank of local_leader, 1->world rank of remote leader
54
55
56    std::vector<int> ep_info[2]; //! 0-> num_ep in local_comm, 1->num_ep in remote_comm
57
58    std::vector<int> new_rank_info[4];
59    std::vector<int> new_ep_info[2];
60
61    std::vector<int> offset;
62
63    if(is_proc_master)
64    {
65
66      size_info[0] = mpi_size;
67
68      rank_info[0].resize(size_info[0]);
69      rank_info[1].resize(size_info[0]);
70
71
72
73      ep_info[0].resize(size_info[0]);
74
75      vector<int> send_buf(6);
76      vector<int> recv_buf(3*size_info[0]);
77
78      send_buf[0] = rank_in_world;
79      send_buf[1] = rank_in_local_parent;
80      send_buf[2] = num_ep;
81
82      ::MPI_Allgather(send_buf.data(), 3, MPI_INT_STD, recv_buf.data(), 3, MPI_INT_STD, local_mpi_comm);
83
84      for(int i=0; i<size_info[0]; i++)
85      {
86        rank_info[0][i] = recv_buf[3*i];
87        rank_info[1][i] = recv_buf[3*i+1];
88        ep_info[0][i]   = recv_buf[3*i+2];
89      }
90
91      if(is_local_leader)
92      {
93        leader_info[0] = rank_in_world;
94        leader_info[1] = remote_leader;
95
96        ::MPI_Comm_rank(static_cast< ::MPI_Comm>(peer_comm.mpi_comm), &rank_in_peer_mpi[0]);
97
98        MPI_Status status;
99
100        send_buf[0] = size_info[0];
101        send_buf[1] = local_ep_size;
102        send_buf[2] = rank_in_peer_mpi[0];
103
104        MPI_Send(send_buf.data(), 3, MPI_INT_STD, remote_leader, tag, peer_comm);
105        MPI_Recv(recv_buf.data(), 3, MPI_INT_STD, remote_leader, tag, peer_comm, &status);
106
107        size_info[1] = recv_buf[0];
108        remote_ep_size = recv_buf[1];
109        rank_in_peer_mpi[1] = recv_buf[2];
110
111      }
112
113      send_buf[0] = size_info[1];
114      send_buf[1] = leader_info[0];
115      send_buf[2] = leader_info[1];
116      send_buf[3] = rank_in_peer_mpi[0];
117      send_buf[4] = rank_in_peer_mpi[1];
118
119      ::MPI_Bcast(send_buf.data(), 5, MPI_INT_STD, local_comm.rank_map->at(local_leader).second, local_mpi_comm);
120
121      size_info[1] = send_buf[0];
122      leader_info[0] = send_buf[1];
123      leader_info[1] = send_buf[2];
124      rank_in_peer_mpi[0] = send_buf[3];
125      rank_in_peer_mpi[1] = send_buf[4];
126
127
128      rank_info[2].resize(size_info[1]);
129      rank_info[3].resize(size_info[1]);
130
131      ep_info[1].resize(size_info[1]);
132
133      send_buf.resize(3*size_info[0]);
134      recv_buf.resize(3*size_info[1]);
135
136      if(is_local_leader)
137      {
138        MPI_Status status;
139
140        std::copy ( rank_info[0].data(), rank_info[0].data() + size_info[0], send_buf.begin() );
141        std::copy ( rank_info[1].data(), rank_info[1].data() + size_info[0], send_buf.begin() + size_info[0] );
142        std::copy ( ep_info[0].data(),   ep_info[0].data()   + size_info[0], send_buf.begin() + 2*size_info[0] );
143
144        MPI_Send(send_buf.data(), 3*size_info[0], MPI_INT_STD, remote_leader, tag, peer_comm);
145        MPI_Recv(recv_buf.data(), 3*size_info[1], MPI_INT_STD, remote_leader, tag, peer_comm, &status);
146      }
147
148      ::MPI_Bcast(recv_buf.data(), 3*size_info[1], MPI_INT_STD, local_comm.rank_map->at(local_leader).second, local_mpi_comm);
149
150      std::copy ( recv_buf.data(), recv_buf.data() + size_info[1], rank_info[2].begin() );
151      std::copy ( recv_buf.data() + size_info[1], recv_buf.data() + 2*size_info[1], rank_info[3].begin()  );
152      std::copy ( recv_buf.data() + 2*size_info[1], recv_buf.data() + 3*size_info[1], ep_info[1].begin() );
153
154
155      offset.resize(size_info[0]);
156
157      if(leader_info[0]<leader_info[1]) // erase all ranks doubled with remote_comm, except the local leader
158      {
159
160        bool found = false;
161        int ep_local;
162        int ep_remote;
163        for(int i=0; i<size_info[0]; i++)
164        {
165          int target = rank_info[0][i];
166          found = false;
167          for(int j=0; j<size_info[1]; j++)
168          {
169            if(target == rank_info[2][j])
170            {
171              found = true;
172              ep_local = ep_info[0][j];
173              ep_remote = ep_info[1][j];
174              break;
175            }
176          }
177          if(found)
178          {
179
180            if(target == leader_info[0]) // the leader is doubled in remote
181            {
182              new_rank_info[0].push_back(target);
183              new_rank_info[1].push_back(rank_info[1][i]);
184
185              new_ep_info[0].push_back(ep_local + ep_remote);
186              offset[i] = 0;
187            }
188            else
189            {
190              offset[i] = ep_local;
191            }
192          }
193          else
194          {
195            new_rank_info[0].push_back(target);
196            new_rank_info[1].push_back(rank_info[1][i]);
197
198            new_ep_info[0].push_back(ep_info[0][i]);
199
200            offset[i] = 0;
201          }
202
203        }
204      }
205
206      else // erase rank doubled with remote leader
207      {
208
209        bool found = false;
210        int ep_local;
211        int ep_remote;
212        for(int i=0; i<size_info[0]; i++)
213        {
214          int target = rank_info[0][i];
215          found = false;
216          for(int j=0; j<size_info[1]; j++)
217          {
218
219            if(target == rank_info[2][j])
220            {
221              found = true;
222              ep_local = ep_info[0][j];
223              ep_remote = ep_info[1][j];
224              break;
225            }
226          }
227          if(found)
228          {
229            if(target != leader_info[1])
230            {
231              new_rank_info[0].push_back(target);
232              new_rank_info[1].push_back(rank_info[1][i]);
233
234              new_ep_info[0].push_back(ep_local + ep_remote);
235              offset[i] = 0;
236            }
237            else // found remote leader
238            {
239              offset[i] = ep_remote;
240            }
241          }
242          else
243          {
244            new_rank_info[0].push_back(target);
245            new_rank_info[1].push_back(rank_info[1][i]);
246
247            new_ep_info[0].push_back(ep_info[0][i]);
248            offset[i] = 0;
249          }
250        }
251      }
252
253      if(offset[mpi_rank] == 0)
254      {
255        is_final_master = true;
256      }
257
258
259      //! size_info[4]: 2->size of new_ep_info for local, 3->size of new_ep_info for remote
260
261      if(is_local_leader)
262      {
263        size_info[2] = new_ep_info[0].size();
264        MPI_Status status;
265        MPI_Send(&size_info[2], 1, MPI_INT_STD, remote_leader, tag, peer_comm);
266        MPI_Recv(&size_info[3], 1, MPI_INT_STD, remote_leader, tag, peer_comm, &status);
267      }
268
269      ::MPI_Bcast(&size_info[2], 2, MPI_INT_STD, local_comm.rank_map->at(local_leader).second, local_mpi_comm);
270
271      new_rank_info[2].resize(size_info[3]);
272      new_rank_info[3].resize(size_info[3]);
273      new_ep_info[1].resize(size_info[3]);
274
275      send_buf.resize(size_info[2]);
276      recv_buf.resize(size_info[3]);
277
278      if(is_local_leader)
279      {
280        MPI_Status status;
281
282        std::copy ( new_rank_info[0].data(), new_rank_info[0].data() + size_info[2], send_buf.begin() );
283        std::copy ( new_rank_info[1].data(), new_rank_info[1].data() + size_info[2], send_buf.begin() + size_info[2] );
284        std::copy ( new_ep_info[0].data(),   new_ep_info[0].data()   + size_info[0], send_buf.begin() + 2*size_info[2] );
285
286        MPI_Send(send_buf.data(), 3*size_info[2], MPI_INT_STD, remote_leader, tag, peer_comm);
287        MPI_Recv(recv_buf.data(), 3*size_info[3], MPI_INT_STD, remote_leader, tag, peer_comm, &status);
288      }
289
290      ::MPI_Bcast(recv_buf.data(),   3*size_info[3], MPI_INT_STD, local_comm.rank_map->at(local_leader).second, local_mpi_comm);
291
292      std::copy ( recv_buf.data(), recv_buf.data() + size_info[3], new_rank_info[2].begin() );
293      std::copy ( recv_buf.data() + size_info[3], recv_buf.data() + 2*size_info[3], new_rank_info[3].begin()  );
294      std::copy ( recv_buf.data() + 2*size_info[3], recv_buf.data() + 3*size_info[3], new_ep_info[1].begin() );
295
296    }
297
298
299
300    if(is_proc_master)
301    {
302      //! leader_info[4]: 2-> rank of local leader in new_group generated comm;
303                      // 3-> rank of remote leader in new_group generated comm;
304      ::MPI_Group local_group;
305      ::MPI_Group new_group;
306      ::MPI_Comm new_comm;
307      ::MPI_Comm intercomm;
308
309      ::MPI_Comm_group(local_mpi_comm, &local_group);
310
311      ::MPI_Group_incl(local_group, size_info[2], new_rank_info[1].data(), &new_group);
312
313      ::MPI_Comm_create(local_mpi_comm, new_group, &new_comm);
314
315
316
317      if(is_local_leader)
318      {
319        ::MPI_Comm_rank(new_comm, &leader_info[2]);
320      }
321
322      ::MPI_Bcast(&leader_info[2], 1, MPI_INT_STD, local_comm.rank_map->at(local_leader).second, local_mpi_comm);
323
324      if(new_comm != MPI_COMM_NULL_STD)
325      {
326
327        ::MPI_Barrier(new_comm);
328
329        ::MPI_Intercomm_create(new_comm, leader_info[2], static_cast< ::MPI_Comm>(peer_comm.mpi_comm), rank_in_peer_mpi[1], tag, &intercomm);
330
331        int id;
332
333        ::MPI_Comm_rank(new_comm, &id);
334        int my_num_ep = new_ep_info[0][id];
335
336        MPI_Comm *ep_intercomm;
337        MPI_Info info;
338        MPI_Comm_create_endpoints(new_comm, my_num_ep, info, ep_intercomm);
339
340
341        for(int i= 0; i<my_num_ep; i++)
342        {
343          ep_intercomm[i].is_intercomm = true;
344
345          ep_intercomm[i].ep_comm_ptr->intercomm = new ep_lib::ep_intercomm;
346          ep_intercomm[i].ep_comm_ptr->intercomm->mpi_inter_comm = intercomm;
347          ep_intercomm[i].ep_comm_ptr->comm_label = leader_info[0];
348        }
349
350
351        #pragma omp critical (write_to_tag_list)
352        tag_list.push_back(make_pair( make_pair(tag, min(leader_info[0], leader_info[1])) , ep_intercomm));
353        printf("tag_list size = %lu\n", tag_list.size());
354      }
355
356
357    }
358
359
360    MPI_Barrier_local(local_comm);
361
362    vector<int> bcast_buf(8);
363    if(is_local_leader)
364    {
365      std::copy(size_info, size_info+4, bcast_buf.begin());
366      std::copy(leader_info, leader_info+4, bcast_buf.begin()+4);
367    }
368
369    MPI_Bcast(bcast_buf.data(), 8, MPI_INT_STD, local_leader, local_comm);
370
371    if(!is_local_leader)
372    {
373      std::copy(bcast_buf.begin(), bcast_buf.begin()+4, size_info);
374      std::copy(bcast_buf.begin()+4, bcast_buf.begin()+8, leader_info);
375    }
376
377    if(!is_local_leader)
378    {
379      new_rank_info[1].resize(size_info[2]);
380      ep_info[1].resize(size_info[1]);
381      offset.resize(size_info[0]);
382    }
383
384    bcast_buf.resize(size_info[2]+size_info[1]+size_info[0]+1);
385
386    if(is_local_leader)
387    {
388      bcast_buf[0] = remote_ep_size;
389      std::copy(new_rank_info[1].data(), new_rank_info[1].data()+size_info[2], bcast_buf.begin()+1);
390      std::copy(ep_info[1].data(), ep_info[1].data()+size_info[1], bcast_buf.begin()+size_info[2]+1);
391      std::copy(offset.data(), offset.data()+size_info[0], bcast_buf.begin()+size_info[2]+size_info[1]+1);
392    }
393
394    MPI_Bcast(bcast_buf.data(), size_info[2]+size_info[1]+size_info[0]+1, MPI_INT_STD, local_leader, local_comm);
395
396    if(!is_local_leader)
397    {
398      remote_ep_size = bcast_buf[0];
399      std::copy(bcast_buf.data()+1, bcast_buf.data()+1+size_info[2], new_rank_info[1].begin());
400      std::copy(bcast_buf.data()+1+size_info[2], bcast_buf.data()+1+size_info[2]+size_info[1], ep_info[1].begin());
401      std::copy(bcast_buf.data()+1+size_info[2]+size_info[1], bcast_buf.data()+1+size_info[2]+size_info[1]+size_info[0], offset.begin());
402    }
403
404    int my_position = offset[rank_in_local_parent]+ep_rank_loc;
405
406
407    MPI_Barrier_local(local_comm);
408    #pragma omp flush
409
410
411    #pragma omp critical (read_from_tag_list)
412    {
413      bool found = false;
414      while(!found)
415      {
416        for(std::list<std::pair < std::pair<int,int>, MPI_Comm* > >::iterator iter = tag_list.begin(); iter!=tag_list.end(); iter++)
417        {
418          if((*iter).first == make_pair(tag, min(leader_info[0], leader_info[1])))
419          {
420            *newintercomm =  iter->second[my_position];
421            found = true;
422            tag_list.erase(iter);
423            break;
424          }
425        }
426      }
427    }
428
429    MPI_Barrier_local(local_comm);
430    // if(is_proc_master)
431    // {
432    //   for(std::list<std::pair < std::pair<int,int>, MPI_Comm* > >::iterator iter = tag_list.begin(); iter!=tag_list.end(); iter++)
433    //   {
434    //     if((*iter).first == make_pair(tag, min(leader_info[0], leader_info[1])))
435    //     {
436    //       tag_list.erase(iter);
437    //       break;
438    //     }
439    //   }
440    // }
441
442    int intercomm_ep_rank, intercomm_ep_rank_loc, intercomm_mpi_rank;
443    int intercomm_ep_size, intercomm_num_ep, intercomm_mpi_size;
444
445    intercomm_ep_rank = newintercomm->ep_comm_ptr->size_rank_info[0].first;
446    intercomm_ep_rank_loc = newintercomm->ep_comm_ptr->size_rank_info[1].first;
447    intercomm_mpi_rank = newintercomm->ep_comm_ptr->size_rank_info[2].first;
448    intercomm_ep_size = newintercomm->ep_comm_ptr->size_rank_info[0].second;
449    intercomm_num_ep = newintercomm->ep_comm_ptr->size_rank_info[1].second;
450    intercomm_mpi_size = newintercomm->ep_comm_ptr->size_rank_info[2].second;
451
452    MPI_Bcast(&remote_ep_size, 1, MPI_INT_STD, local_leader, local_comm);
453
454    int my_rank_map_elem[2];
455
456    my_rank_map_elem[0] = intercomm_ep_rank;
457    my_rank_map_elem[1] = (*newintercomm).ep_comm_ptr->comm_label;
458
459    vector<pair<int, int> > local_rank_map_array;
460    vector<pair<int, int> > remote_rank_map_array;
461
462
463    (*newintercomm).ep_comm_ptr->intercomm->local_rank_map = new RANK_MAP;
464    (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->resize(local_ep_size);
465
466    MPI_Allgather(my_rank_map_elem, 2, MPI_INT_STD, (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->data(), 2, MPI_INT_STD, local_comm);
467
468    (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map = new RANK_MAP;
469    (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->resize(remote_ep_size);
470
471    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[0] = local_comm.ep_comm_ptr->size_rank_info[0];
472    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[1] = local_comm.ep_comm_ptr->size_rank_info[1];
473    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[2] = local_comm.ep_comm_ptr->size_rank_info[2];
474
475    int local_intercomm_size = intercomm_ep_size;
476    int remote_intercomm_size;
477
478    int new_bcast_root_0 = 0;
479    int new_bcast_root = 0;
480
481
482    if(is_local_leader)
483    {
484      MPI_Status status;
485      MPI_Send((*newintercomm).ep_comm_ptr->intercomm->local_rank_map->data(), 2*local_ep_size, MPI_INT_STD, remote_leader, tag, peer_comm);
486      MPI_Recv((*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->data(), 2*remote_ep_size, MPI_INT_STD, remote_leader, tag, peer_comm, &status);
487
488      MPI_Send(&local_intercomm_size, 1, MPI_INT_STD, remote_leader, tag, peer_comm);
489      MPI_Recv(&remote_intercomm_size, 1, MPI_INT_STD, remote_leader, tag, peer_comm, &status);
490
491      new_bcast_root_0 = intercomm_ep_rank;
492    }
493
494    MPI_Allreduce(&new_bcast_root_0, &new_bcast_root, 1, MPI_INT_STD, MPI_SUM_STD, *newintercomm);
495
496
497    MPI_Bcast((*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->data(), 2*remote_ep_size, MPI_INT_STD, local_leader, local_comm);
498    MPI_Bcast(&remote_intercomm_size, 1, MPI_INT_STD, new_bcast_root, *newintercomm);
499
500
501    (*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map = new RANK_MAP;
502    (*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map->resize(remote_intercomm_size);
503
504
505
506
507    if(is_local_leader)
508    {
509      MPI_Status status;
510      MPI_Send((*newintercomm).rank_map->data(), 2*local_intercomm_size, MPI_INT_STD, remote_leader, tag, peer_comm);
511      MPI_Recv((*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map->data(), 2*remote_intercomm_size, MPI_INT_STD, remote_leader, tag, peer_comm, &status);
512    }
513
514    MPI_Bcast((*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map->data(), 2*remote_intercomm_size, MPI_INT_STD, new_bcast_root, *newintercomm);
515
516    (*newintercomm).ep_comm_ptr->intercomm->local_comm = &(local_comm.ep_comm_ptr->comm_list[ep_rank_loc]);
517    (*newintercomm).ep_comm_ptr->intercomm->intercomm_tag = tag;
518
519/*
520    for(int i=0; i<local_ep_size; i++)
521    if(local_comm.ep_comm_ptr->comm_label == 0) printf("ep_rank (from EP) = %d, local_rank_map[%d] = (%d,%d)\n", intercomm_ep_rank, i,
522          (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->at(i).first, (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->at(i).second);
523
524    for(int i=0; i<remote_ep_size; i++)
525    if(local_comm.ep_comm_ptr->comm_label == 0) printf("ep_rank (from EP) = %d, remote_rank_map[%d] = (%d,%d)\n", intercomm_ep_rank, i,
526          (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->at(i).first, (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->at(i).second);
527
528    for(int i=0; i<remote_intercomm_size; i++)
529    if(local_comm.ep_comm_ptr->comm_label == 0) printf("ep_rank (from EP) = %d, intercomm_rank_map[%d] = (%d,%d)\n", intercomm_ep_rank, i,
530          (*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map->at(i).first, (*newintercomm).ep_comm_ptr->intercomm->intercomm_rank_map->at(i).second);
531*/
532
533//    for(int i=0; i<(*newintercomm).rank_map->size(); i++)
534//    if(local_comm.ep_comm_ptr->comm_label != 99) printf("ep_rank = %d, rank_map[%d] = (%d,%d)\n", intercomm_ep_rank, i,
535//          (*newintercomm).rank_map->at(i).first, (*newintercomm).rank_map->at(i).second);
536
537//    MPI_Comm *test_comm = newintercomm->ep_comm_ptr->intercomm->local_comm;
538//    int test_rank;
539//    MPI_Comm_rank(*test_comm, &test_rank);
540//    printf("=================test_rank = %d\n", test_rank);
541
542    return MPI_SUCCESS;
543
544  }
545
546
547
548
549  int MPI_Intercomm_create_unique_leader(MPI_Comm local_comm, int local_leader, MPI_Comm peer_comm, int remote_leader, int tag, MPI_Comm *newintercomm)
550  {
551    //! mpi_size of local comm = 1
552    //! same world rank of leaders
553
554    int ep_rank, ep_rank_loc, mpi_rank;
555    int ep_size, num_ep, mpi_size;
556
557    ep_rank = local_comm.ep_comm_ptr->size_rank_info[0].first;
558    ep_rank_loc = local_comm.ep_comm_ptr->size_rank_info[1].first;
559    mpi_rank = local_comm.ep_comm_ptr->size_rank_info[2].first;
560    ep_size = local_comm.ep_comm_ptr->size_rank_info[0].second;
561    num_ep = local_comm.ep_comm_ptr->size_rank_info[1].second;
562    mpi_size = local_comm.ep_comm_ptr->size_rank_info[2].second;
563
564
565
566    std::vector<int> rank_info[4];  //! 0->rank_in_world of local_comm,  1->rank_in_local_parent of local_comm
567                                    //! 2->rank_in_world of remote_comm, 3->rank_in_local_parent of remote_comm
568
569    int rank_in_world;
570
571    int rank_in_peer_mpi[2];
572
573    ::MPI_Comm_rank(MPI_COMM_WORLD_STD, &rank_in_world);
574
575
576    int local_num_ep = num_ep;
577    int remote_num_ep;
578    int total_num_ep;
579
580    int leader_rank_in_peer[2];
581
582    int my_position;
583    int tag_label[2];
584
585    vector<int> send_buf(4);
586    vector<int> recv_buf(4);
587
588
589    if(ep_rank == local_leader)
590    {
591      MPI_Status status;
592
593
594
595      MPI_Comm_rank(peer_comm, &leader_rank_in_peer[0]);
596
597      send_buf[0] = local_num_ep;
598      send_buf[1] = leader_rank_in_peer[0];
599
600      MPI_Request req_s, req_r;
601
602      MPI_Isend(send_buf.data(), 2, MPI_INT_STD, remote_leader, tag, peer_comm, &req_s);
603      MPI_Irecv(recv_buf.data(), 2, MPI_INT_STD, remote_leader, tag, peer_comm, &req_r);
604
605
606      MPI_Wait(&req_s, &status);
607      MPI_Wait(&req_r, &status);
608
609      recv_buf[2] = leader_rank_in_peer[0];
610
611    }
612
613    MPI_Bcast(recv_buf.data(), 3, MPI_INT_STD, local_leader, local_comm);
614
615    remote_num_ep = recv_buf[0];
616    leader_rank_in_peer[1] = recv_buf[1];
617    leader_rank_in_peer[0] = recv_buf[2];
618
619    total_num_ep = local_num_ep + remote_num_ep;
620
621
622    if(leader_rank_in_peer[0] < leader_rank_in_peer[1])
623    {
624      my_position = ep_rank_loc;
625      //! LEADER create EP
626      if(ep_rank == local_leader)
627      {
628        ::MPI_Comm mpi_dup;
629       
630        ::MPI_Comm_dup(static_cast< ::MPI_Comm>(local_comm.mpi_comm), &mpi_dup);
631
632        MPI_Comm *ep_intercomm;
633        MPI_Info info;
634        MPI_Comm_create_endpoints(mpi_dup, total_num_ep, info, ep_intercomm);
635
636
637        for(int i=0; i<total_num_ep; i++)
638        {
639          ep_intercomm[i].is_intercomm = true;
640          ep_intercomm[i].ep_comm_ptr->intercomm = new ep_lib::ep_intercomm;
641          ep_intercomm[i].ep_comm_ptr->intercomm->mpi_inter_comm = 0;
642
643          ep_intercomm[i].ep_comm_ptr->comm_label = leader_rank_in_peer[0];
644        }
645
646        tag_label[0] = TAG++;
647        tag_label[1] = rank_in_world;
648
649        #pragma omp critical (write_to_tag_list)
650        tag_list.push_back(make_pair( make_pair(tag_label[0], tag_label[1]) , ep_intercomm));
651
652        MPI_Request req_s;
653        MPI_Status sta_s;
654        MPI_Isend(tag_label, 2, MPI_INT_STD, remote_leader, tag, peer_comm, &req_s);
655
656        MPI_Wait(&req_s, &sta_s);
657
658      }
659    }
660    else
661    {
662      //! Wait for EP creation
663      my_position = remote_num_ep + ep_rank_loc;
664      if(ep_rank == local_leader)
665      {
666        MPI_Status status;
667        MPI_Request req_r;
668        MPI_Irecv(tag_label, 2, MPI_INT_STD, remote_leader, tag, peer_comm, &req_r);
669        MPI_Wait(&req_r, &status);
670      }
671    }
672
673    MPI_Bcast(tag_label, 2, MPI_INT_STD, local_leader, local_comm);
674
675
676
677
678    #pragma omp critical (read_from_tag_list)
679    {
680      bool found = false;
681      while(!found)
682      {
683        for(std::list<std::pair < std::pair<int,int>, MPI_Comm* > >::iterator iter = tag_list.begin(); iter!=tag_list.end(); iter++)
684        {
685          if((*iter).first == make_pair(tag_label[0], tag_label[1]))
686          {
687            *newintercomm =  iter->second[my_position];
688            found = true;
689            tag_list.erase(iter);
690            break;
691          }
692        }
693      }
694    }
695
696    // if(leader_rank_in_peer[0] < leader_rank_in_peer[1])
697    // {
698    //   for(std::list<std::pair < std::pair<int,int>, MPI_Comm* > >::iterator iter = tag_list.begin(); iter!=tag_list.end(); iter++)
699    //     {
700    //       if((*iter).first == make_pair(tag_label[0], tag_label[1]))
701    //       {
702    //         tag_list.erase(iter);
703    //       }
704    //     }
705    // }
706
707
708
709    int intercomm_ep_rank, intercomm_ep_rank_loc, intercomm_mpi_rank;
710    int intercomm_ep_size, intercomm_num_ep, intercomm_mpi_size;
711
712    intercomm_ep_rank = newintercomm->ep_comm_ptr->size_rank_info[0].first;
713    intercomm_ep_rank_loc = newintercomm->ep_comm_ptr->size_rank_info[1].first;
714    intercomm_mpi_rank = newintercomm->ep_comm_ptr->size_rank_info[2].first;
715    intercomm_ep_size = newintercomm->ep_comm_ptr->size_rank_info[0].second;
716    intercomm_num_ep = newintercomm->ep_comm_ptr->size_rank_info[1].second;
717    intercomm_mpi_size = newintercomm->ep_comm_ptr->size_rank_info[2].second;
718
719
720
721    (*newintercomm).ep_comm_ptr->intercomm->local_rank_map  = new RANK_MAP;
722    (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map = new RANK_MAP;
723    (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->resize(local_num_ep);
724    (*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->resize(remote_num_ep);
725
726    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[0] = local_comm.ep_comm_ptr->size_rank_info[0];
727    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[1] = local_comm.ep_comm_ptr->size_rank_info[1];
728    (*newintercomm).ep_comm_ptr->intercomm->size_rank_info[2] = local_comm.ep_comm_ptr->size_rank_info[2];
729
730
731
732    int local_rank_map_ele[2];
733    local_rank_map_ele[0] = intercomm_ep_rank;
734    local_rank_map_ele[1] = (*newintercomm).ep_comm_ptr->comm_label;
735
736    MPI_Allgather(local_rank_map_ele, 2, MPI_INT_STD, (*newintercomm).ep_comm_ptr->intercomm->local_rank_map->data(), 2, MPI_INT_STD, local_comm);
737
738    if(ep_rank == local_leader)
739    {
740      MPI_Status status;
741      MPI_Request req_s, req_r;
742
743      MPI_Isend((*newintercomm).ep_comm_ptr->intercomm->local_rank_map->data(), 2*local_num_ep, MPI_INT_STD, remote_leader, tag, peer_comm, &req_s);
744      MPI_Irecv((*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->data(), 2*remote_num_ep, MPI_INT_STD, remote_leader, tag, peer_comm, &req_r);
745
746
747      MPI_Wait(&req_s, &status);
748      MPI_Wait(&req_r, &status);
749
750    }
751
752    MPI_Bcast((*newintercomm).ep_comm_ptr->intercomm->remote_rank_map->data(), 2*remote_num_ep, MPI_INT_STD, local_leader, local_comm);
753    (*newintercomm).ep_comm_ptr->intercomm->local_comm = &(local_comm.ep_comm_ptr->comm_list[ep_rank_loc]);
754    (*newintercomm).ep_comm_ptr->intercomm->intercomm_tag = tag;
755
756
757    return MPI_SUCCESS;
758  }
759
760
761}
Note: See TracBrowser for help on using the repository browser.