source: XIOS/dev/branch_openmp/extern/ep_dev/ep_message.cpp @ 1500

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

save dev

File size: 6.9 KB
Line 
1/*!
2   \file ep_message.cpp
3   \since 2 may 2016
4
5   \brief Definitions of MPI endpoint function: Message_Check
6 */
7
8#include "ep_lib.hpp"
9#include <mpi.h>
10#include "ep_declaration.hpp"
11#include "ep_mpi.hpp"
12
13using namespace std;
14
15extern std::list< ep_lib::MPI_Request* > * EP_PendingRequests;
16#pragma omp threadprivate(EP_PendingRequests)
17
18namespace ep_lib
19{
20  int Request_Check()
21  {
22    MPI_Status status;
23    MPI_Message *message;
24    int probed = false;
25    int recv_count = 0;
26    std::list<MPI_Request* >::iterator it;
27   
28    if(EP_PendingRequests == 0 ) return 0;
29   
30    for(it = EP_PendingRequests->begin(); it!=EP_PendingRequests->end(); it++)
31    { 
32      Message_Check(((*(*it))->comm));
33    }
34
35
36    for(it = EP_PendingRequests->begin(); it!=EP_PendingRequests->end(); )
37    {
38      if((*(*it))->state == 0)
39      {
40         
41        message = new MPI_Message;
42        *message = new ep_message;
43       
44        printf("new %p : in ep_lib::Request_Check, message = new MPI_Message\n", message);
45        printf("new %p : in ep_lib::Request_Check, *message = new ep_message\n", *message);
46         
47         
48        MPI_Improbe((*(*it))->ep_src, (*(*it))->ep_tag, ((*(*it))->comm), &probed, message, &status);
49        printf("in Request_Check, after improbe, mpi_status = %p\n", to_mpi_status_ptr(status));
50     
51       
52        if(probed)
53        {
54          ::MPI_Get_count(to_mpi_status_ptr(status), to_mpi_type((*(*it))->ep_datatype), &recv_count);
55         
56          printf("in Request_Check, imrecv, buf = %p, recv_count = %d, status = %p\n", (*(*it))->buf, recv_count, to_mpi_status_ptr(status));
57          MPI_Imrecv((*(*it))->buf, recv_count, (*(*it))->ep_datatype, message, *it);
58          (*(*it))->type = 3;
59          (*(*it))->state = 1;
60         
61
62          printf("delete %p : in ep_lib::Request_Check, delete *message\n", *message);
63          printf("delete %p : in ep_lib::Request_Check, delete message\n", message);
64
65          delete *message;
66          delete message;
67       
68          it++;
69          continue;     
70        }     
71      }
72     
73      if((*(*it))->state == 2)
74      {
75        printf("delete %p : in ep_lib::Request_Check, delete (*(*it))\n", (*(*it)));
76        delete (*(*it));
77       
78        EP_PendingRequests->erase(it);
79        printf("EP_PendingRequests->size() = %lu\n", EP_PendingRequests->size());
80        it = EP_PendingRequests->begin();
81        continue;
82      }
83      else it++;
84    }
85  }
86 
87 
88 
89  int Message_Check(MPI_Comm comm)
90  {
91    if(!comm->is_ep) return MPI_SUCCESS;
92
93    if(comm->is_intercomm)
94    {
95      return  Message_Check_intercomm(comm);
96    }
97   
98    return Message_Check_intracomm(comm);
99
100  }
101 
102 
103  int Message_Check_intracomm(MPI_Comm comm)
104  {
105   
106    int flag = true;
107    ::MPI_Status status;
108    ::MPI_Message message;
109
110    while(flag) // loop until the end of global queue
111    {
112      Debug("Message probing for intracomm\n");
113     
114      #pragma omp critical (_mpi_call)
115      {
116        ::MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, to_mpi_comm(comm->mpi_comm), &flag, &status);
117        if(flag)
118        {
119          Debug("find message in mpi comm \n");
120          ::MPI_Mprobe(status.MPI_SOURCE, status.MPI_TAG, to_mpi_comm(comm->mpi_comm), &message, &status);
121        }
122      }
123
124     
125      if(flag)
126      {
127        MPI_Message msg = new ep_message; 
128        msg->mpi_message = new ::MPI_Message(message);
129
130        printf("new %p : in ep_lib::Message_Check, msg = new ep_message\n", msg);
131        printf("new %p : in ep_lib::Message_Check, msg->mpi_message = new ::MPI_Message\n", msg->mpi_message);
132             
133
134        msg->ep_tag  = bitset<15>(status.MPI_TAG >> 16).to_ulong(); 
135        int src_loc  = bitset<8> (status.MPI_TAG >> 8) .to_ulong(); 
136        int dest_loc = bitset<8> (status.MPI_TAG)           .to_ulong();
137        int src_mpi  = status.MPI_SOURCE;
138             
139        msg->ep_src  = get_ep_rank(comm, src_loc,  src_mpi);       
140        msg->mpi_status = new ::MPI_Status(status); 
141        printf("new %p : in ep_lib::Message_Check, msg->mpi_status = new ::MPI_Status\n", msg->mpi_status);
142
143        #pragma omp critical (_query)
144        {
145          #pragma omp flush
146          comm->ep_comm_ptr->comm_list[dest_loc]->ep_comm_ptr->message_queue->push_back(msg); 
147          int test_count;
148          ::MPI_Get_count(static_cast< ::MPI_Status* >(msg->mpi_status), 1275070475, &test_count);
149          printf("status1 = %p, test_count2 = %d\n", static_cast< ::MPI_Status* >(msg->mpi_status), test_count);
150         
151          ::MPI_Get_count(static_cast< ::MPI_Status* >(comm->ep_comm_ptr->comm_list[dest_loc]->ep_comm_ptr->message_queue->back()->mpi_status), 1275070475, &test_count);
152          printf("status2 = %p, test_count2 = %d\n", static_cast< ::MPI_Status* >(comm->ep_comm_ptr->comm_list[dest_loc]->ep_comm_ptr->message_queue->back()->mpi_status), test_count);
153                   
154          #pragma omp flush
155        }
156      }
157    }
158
159    return MPI_SUCCESS;
160  }
161
162
163 
164 
165 
166
167
168  int Message_Check_intercomm(MPI_Comm comm)
169  {
170    if(!comm->ep_comm_ptr->intercomm->mpi_inter_comm) return 0;
171
172    Debug("Message probing for intercomm\n");
173
174    int flag = true;
175    ::MPI_Message message;
176    ::MPI_Status status;
177    int current_ep_rank;
178    MPI_Comm_rank(comm, &current_ep_rank);
179
180    while(flag) // loop until the end of global queue "comm->ep_comm_ptr->intercomm->mpi_inter_comm"
181    {
182      Debug("Message probing for intracomm\n");
183
184      #pragma omp critical (_mpi_call)
185      {
186        ::MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, to_mpi_comm(comm->ep_comm_ptr->intercomm->mpi_inter_comm), &flag, &status);
187        if(flag)
188        {
189          Debug("find message in mpi comm \n");
190          ::MPI_Mprobe(status.MPI_SOURCE, status.MPI_TAG, to_mpi_comm(comm->ep_comm_ptr->intercomm->mpi_inter_comm), &message, &status);
191        }
192      }
193     
194
195      if(flag)
196      {
197
198        MPI_Message msg = new ep_message; 
199        msg->mpi_message = new ::MPI_Message(message);
200
201        printf("new %p : in ep_lib::Message_Check, msg = new ep_message\n", msg);
202        printf("new %p : in ep_lib::Message_Check, msg->mpi_message = new ::MPI_Message\n", msg->mpi_message);
203             
204
205        msg->ep_tag  = bitset<15>(status.MPI_TAG >> 16).to_ulong(); 
206        int src_loc  = bitset<8> (status.MPI_TAG >> 8) .to_ulong(); 
207        int dest_loc = bitset<8> (status.MPI_TAG)           .to_ulong();
208        int src_mpi  = status.MPI_SOURCE;
209        int current_inter = comm->ep_comm_ptr->intercomm->local_rank_map->at(current_ep_rank).first;
210             
211        msg->ep_src  = get_ep_rank(comm, src_loc,  src_mpi);       
212        msg->mpi_status = new ::MPI_Status(status); 
213        printf("new %p : in ep_lib::Message_Check, msg->mpi_status = new ::MPI_Status\n", msg->mpi_status);
214
215        #pragma omp critical (_query)
216        {
217          #pragma omp flush
218          comm->ep_comm_ptr->comm_list[dest_loc]->ep_comm_ptr->message_queue->push_back(msg);
219          #pragma omp flush
220        }
221      }
222    }
223
224    Message_Check_intracomm(comm);
225
226    return MPI_SUCCESS;
227  }
228
229 
230
231}
Note: See TracBrowser for help on using the repository browser.