Changeset 1539 for XIOS/dev/branch_openmp/extern/src_ep_dev/ep_probe.cpp

Timestamp:

06/12/18 11:54:13 (6 years ago)

Author:

yushan

Message:

update intercomm_merge and remove redundant files

File:

• XIOS/dev/branch_openmp/extern/src_ep_dev/ep_probe.cpp (modified) (3 diffs)

XIOS/dev/branch_openmp/extern/src_ep_dev/ep_probe.cpp

@@ -8 +8 @@
   int MPI_Iprobe_mpi(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
   {
+    ::MPI_Status mpi_status;
+
+    ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
+
+    status->mpi_status = new ::MPI_Status(mpi_status);
     status->ep_src = src;
     status->ep_tag = tag;
-    return ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, to_mpi_status_ptr(*status));
   }
+
+
+  int MPI_Improbe_mpi(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
+  {
+    ::MPI_Status mpi_status;
+    ::MPI_Message mpi_message;
+
+    #ifdef _openmpi
+    #pragma omp critical (_mpi_call)
+    {
+      ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
+      if(*flag)
+      {
+        ::MPI_Mprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), &mpi_message, &mpi_status);
+      }
+    }
+    #elif _intelmpi
+    ::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);
+    #endif
+      
+    status->mpi_status = new ::MPI_Status(mpi_status);
+    status->ep_src = src;
+    status->ep_tag = tag;
+
+    (*message)->mpi_message = &message;
+    (*message)->ep_src = src;
+    (*message)->ep_tag = tag;
+  }
+
 
   int MPI_Iprobe(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
@@ -20 +53 @@
       return MPI_Iprobe_mpi(src, tag, comm, flag, status);
     }
+
+    if(comm->is_intercomm)
+    {
+      if(src>=0) src = comm->inter_rank_map->at(src);
+    } 
     
-    else
+    return MPI_Iprobe_endpoint(src, tag, comm, flag, status);
+  }
+
+  int MPI_Iprobe_endpoint(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
+  {
+    Debug("MPI_Iprobe with EP\n");
+
+    *flag = false;
+    
+    Message_Check(comm);
+
+    #pragma omp flush
+
+    #pragma omp critical (_query)
+    for(Message_list::iterator it = comm->ep_comm_ptr->message_queue->begin(); it!= comm->ep_comm_ptr->message_queue->end(); ++it)
     {
-      Debug("MPI_Iprobe with EP\n");
+      bool src_matched = src<0? true: (*it)->ep_src == src;
+      bool tag_matched = tag<0? true: (*it)->ep_tag == tag;
       
-      *flag = false;
-    
-      Message_Check(comm);
+      if(src_matched && tag_matched)        
+      {
+        Debug("find message\n");
+         
+        status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
+        status->ep_src = (*it)->ep_src;
+        status->ep_tag = (*it)->ep_tag;
+        
+        if(comm->is_intercomm)
+        {
+          for(INTER_RANK_MAP::iterator iter = comm->inter_rank_map->begin(); iter != comm->inter_rank_map->end(); iter++)
+          {
+            if(iter->second == (*it)->ep_src) status->ep_src=iter->first;
+          }
+        }
 
-      #pragma omp flush
-
-      #pragma omp critical (_query)
-      for(Message_list::iterator it = comm->ep_comm_ptr->message_queue->begin(); it!= comm->ep_comm_ptr->message_queue->end(); ++it)
-      {
-        bool src_matched = src<0? true: (*it)->ep_src == src;
-        bool tag_matched = tag<0? true: (*it)->ep_tag == tag;
-        
-        if(src_matched && tag_matched)        
-        {
-          Debug("find message\n");
-          
-
-          status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
-          status->ep_src = (*it)->ep_src;
-          status->ep_tag = (*it)->ep_tag;
-
-          *flag = true;
-          break;
-        }
+        *flag = true;
+        break;
       }
     }
   }
 
-
+  
 
   int MPI_Improbe(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
   {
+    if(!comm->is_ep)
+    {
+      Debug("MPI_Iprobe with MPI\n");
+      return MPI_Improbe_mpi(src, tag, comm, flag, message, status);
+    }
+
+    if(comm->is_intercomm)
+    {
+      src = comm->inter_rank_map->at(src);
+      *message = new ep_message;
+      printf("============= new *message = %p\n", *message);
+    } 
+    
+    return MPI_Improbe_endpoint(src, tag, comm, flag, message, status);
+  }
+
+
+
+  int MPI_Improbe_endpoint(int src, int tag, MPI_Comm comm, int *flag, MPI_Message *message, MPI_Status *status)
+  {
     int ep_rank_loc = comm->ep_comm_ptr->size_rank_info[1].first;
     int mpi_rank    = comm->ep_comm_ptr->size_rank_info[2].first;
+
     *flag = false;
-    if(!comm->is_ep)
-    {
-      Debug("calling MPI_Improbe MPI\n");
-
-      ::MPI_Status mpi_status;
-      ::MPI_Message mpi_message;
-
-      #ifdef _openmpi
-      #pragma omp critical (_mpi_call)
-      {
-        ::MPI_Iprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
-        if(*flag)
-        {
-          ::MPI_Mprobe(src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), &mpi_message, &mpi_status);
-        }
-      }
-      #elif _intelmpi
-        ::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);
-      #endif
-        
-      status->mpi_status = &mpi_status;
-      status->ep_src = src;
-      status->ep_tag = tag;
-
-      (*message)->mpi_message = &message;
-      (*message)->ep_src = src;
-      (*message)->ep_tag = tag;
-      
-      
-      return 0;
-    }
-
-   
-
+    
+    Message_Check(comm);
+    
     #pragma omp flush
 
@@ -146 +183 @@
 }
 
+