source: XIOS/dev/branch_openmp/extern/ep_dev/ep_probe.cpp @ 1532

Last change on this file since 1532 was 1532, checked in by yushan, 3 years ago

save dev.

File size: 5.3 KB
RevLine 
[1381]1#include "ep_lib.hpp"
2#include <mpi.h>
3#include "ep_declaration.hpp"
4#include "ep_mpi.hpp"
5
6namespace ep_lib
7{
[1522]8  int MPI_Iprobe_mpi(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
9  {
[1527]10    ::MPI_Status mpi_status;
11
12    ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
13
14    status->mpi_status = new ::MPI_Status(mpi_status);
[1522]15    status->ep_src = src;
16    status->ep_tag = tag;
17  }
[1381]18
[1527]19
20  int MPI_Improbe_mpi(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
21  {
22    ::MPI_Status mpi_status;
23    ::MPI_Message mpi_message;
24
25    #ifdef _openmpi
26    #pragma omp critical (_mpi_call)
27    {
28      ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
29      if(*flag)
30      {
31        ::MPI_Mprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), &mpi_message, &mpi_status);
32      }
33    }
34    #elif _intelmpi
35    ::MPI_Improbe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_message, &mpi_status);
36    #endif
37     
38    status->mpi_status = new ::MPI_Status(mpi_status);
39    status->ep_src = src;
40    status->ep_tag = tag;
41
42    (*message)->mpi_message = &message;
43    (*message)->ep_src = src;
44    (*message)->ep_tag = tag;
45  }
46
47
[1381]48  int MPI_Iprobe(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
49  {
[1500]50    if(!comm->is_ep)
[1381]51    {
[1522]52      Debug("MPI_Iprobe with MPI\n");
53      return MPI_Iprobe_mpi(src, tag, comm, flag, status);
[1381]54    }
[1532]55
56    if(comm->is_intercomm)
57    {
58      src = comm->inter_rank_map->at(src);
59    } 
[1522]60   
[1532]61    return MPI_Iprobe_endpoint(src, tag, comm, flag, status);
62  }
[1527]63
[1532]64  int MPI_Iprobe_endpoint(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
65  {
66    Debug("MPI_Iprobe with EP\n");
67
68    *flag = false;
[1522]69   
[1532]70    Message_Check(comm);
[1381]71
[1532]72    #pragma omp flush
[1381]73
[1532]74    #pragma omp critical (_query)
75    for(Message_list::iterator it = comm->ep_comm_ptr->message_queue->begin(); it!= comm->ep_comm_ptr->message_queue->end(); ++it)
76    {
77      bool src_matched = src<0? true: (*it)->ep_src == src;
78      bool tag_matched = tag<0? true: (*it)->ep_tag == tag;
79     
80      if(src_matched && tag_matched)       
[1381]81      {
[1532]82        Debug("find message\n");
83         
84        status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
85        status->ep_src = (*it)->ep_src;
86        status->ep_tag = (*it)->ep_tag;
[1381]87
[1532]88        *flag = true;
89        break;
[1381]90      }
91    }
92  }
93
[1527]94 
[1381]95
96  int MPI_Improbe(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
97  {
[1500]98    if(!comm->is_ep)
[1381]99    {
[1532]100      Debug("MPI_Iprobe with MPI\n");
[1527]101      return MPI_Improbe_mpi(src, tag, comm, flag, message, status);
[1532]102    }
103
104    if(comm->is_intercomm)
105    {
106      src = comm->inter_rank_map->at(src);
107      *message = new ep_message;
108      printf("============= new *message = %p\n", *message);
[1527]109    } 
[1532]110   
111    return MPI_Improbe_endpoint(src, tag, comm, flag, message, status);
112  }
[1381]113
[1532]114
115
116  int MPI_Improbe_endpoint(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
117  {
118    int ep_rank_loc = comm->ep_comm_ptr->size_rank_info[1].first;
119    int mpi_rank    = comm->ep_comm_ptr->size_rank_info[2].first;
120
121    *flag = false;
122   
123    Message_Check(comm);
124   
[1381]125    #pragma omp flush
126
127    #pragma omp critical (_query)
[1500]128    if(! comm->ep_comm_ptr->message_queue->empty())
[1381]129    {
[1500]130      for(Message_list::iterator it = comm->ep_comm_ptr->message_queue->begin(); it!= comm->ep_comm_ptr->message_queue->end(); ++it)
[1381]131      {
[1500]132                                         
133        bool src_matched = src<0? true: (*it)->ep_src == src;
134        bool tag_matched = tag<0? true: (*it)->ep_tag == tag;
[1381]135       
136        if(src_matched && tag_matched)
137        {
138          *flag = true;
139
[1503]140          status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
141          memcheck("new "<< status->mpi_status << " : in ep_lib::MPI_Improbe, status->mpi_status = new ::MPI_Status");
[1500]142          status->ep_src = (*it)->ep_src;
143          status->ep_tag = (*it)->ep_tag;
[1381]144
[1503]145          (*message)->mpi_message = new ::MPI_Message(*static_cast< ::MPI_Message*>((*it)->mpi_message));
146          memcheck("new "<< (*message)->mpi_message <<" : in ep_lib::MPI_Improbe, (*message)->mpi_message = new ::MPI_Message");
[1500]147          (*message)->ep_src = (*it)->ep_src;
148          (*message)->ep_tag = (*it)->ep_tag;
149                                     
[1381]150
151          #pragma omp critical (_query2)
152          {             
[1503]153            memcheck("delete "<< (*it)->mpi_message <<" : in ep_lib::Message_Check, delete (*it)->mpi_message");
154            memcheck("delete "<< (*it)->mpi_status <<" : in ep_lib::Message_Check, delete (*it)->mpi_status");
155            memcheck("delete "<< (*it) <<" : in ep_lib::Message_Check, delete (*it)");
[1500]156           
[1503]157           
[1500]158            delete (*it)->mpi_message;     
159            delete (*it)->mpi_status; 
160            delete *it;
161           
[1503]162                       
[1500]163            comm->ep_comm_ptr->message_queue->erase(it);
[1503]164            memcheck("message_queue["<<mpi_rank<<","<<ep_rank_loc<<"]->size = "<<comm->ep_comm_ptr->message_queue->size());
[1381]165            #pragma omp flush
166          }
[1500]167         
[1381]168          break;
169        }
170
171      }
172    }
173  }
174
175}
176
[1532]177
Note: See TracBrowser for help on using the repository browser.