source: XIOS/dev/dev_trunk_omp/extern/src_ep_dev2/ep_recv.cpp @ 1710

#ifdef _usingEP
/*!
   \file ep_recv.cpp
   \since 2 may 2016

   \brief Definitions of MPI receive functions: MPI_Recv, MPI_Mrecv, MPI_Irecv, MPI_Imrecv
 */


#include "ep_lib.hpp"
#include <mpi.h>
#include "ep_declaration.hpp"
#include "ep_mpi.hpp"

using namespace std;

extern std::list< ep_lib::MPI_Request* > * EP_PendingRequests;
#pragma omp threadprivate(EP_PendingRequests)

namespace ep_lib 
{

  int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status *status)
  {

    if(src>=0) status->ep_src = src;
    
    if(comm->is_intercomm)
    {
      if(src>=0) src = comm->inter_rank_map->at(src);
    }

    ::MPI_Status mpi_status;
    if(src>=0)
    {

      int mpi_src = comm->ep_rank_map->at(src).second;
      int mpi_tag = tag_combine(tag, comm->ep_rank_map->at(src).first, comm->ep_comm_ptr->size_rank_info[1].first);

      //printf("MPI_Recv with src\n");
      //printf("proc %d recv from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, src);
      ::MPI_Recv(buf, count, to_mpi_type(datatype), mpi_src, mpi_tag, to_mpi_comm(comm->mpi_comm), &mpi_status);
      status->mpi_status = new ::MPI_Status(mpi_status);
      return MPI_SUCCESS;
    }
    else
    {
      //printf("MPI_Recv with ANY_SOURCE\n");
      int probed=false;
      int mpi_tag;
      while(!probed)
      {
        //for(int i=0; i<comm->ep_comm_ptr->size_rank_info[1].second; i++)
        for(int i=0; i<4; i++)
        {
          mpi_tag = tag_combine(tag, i, comm->ep_comm_ptr->size_rank_info[1].first);
          ::MPI_Iprobe(MPI_ANY_SOURCE, mpi_tag, to_mpi_comm(comm->mpi_comm), &probed, &mpi_status);
          if(probed) 
          {
            //printf("proc %d recv from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, get_ep_rank(comm, i, mpi_status.MPI_SOURCE));
            ::MPI_Recv(buf, count, to_mpi_type(datatype), mpi_status.MPI_SOURCE, mpi_status.MPI_TAG, to_mpi_comm(comm->mpi_comm), &mpi_status);
            //printf("get_ep_rank in MPI_Recv\n");
            status->ep_src = get_ep_rank(comm, i, mpi_status.MPI_SOURCE);
            status->mpi_status = new ::MPI_Status(mpi_status);
            return MPI_SUCCESS;
          }
        }
      }
    }
    
  }

  int MPI_Irecv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request *request)
  {
    if(comm->is_intercomm)
    {
      if(src>=0) src = comm->inter_rank_map->at(src);
    }

    *request = new ep_request;
    //printf("MPI_Irecv : *request = new ep_request => *request = %p\n", *request);

    (*request)->mpi_request = new ::MPI_Request;
    //printf("MPI_Irecv : (*request)->mpi_request = new ::MPI_Request => (*request)->mpi_request = %p\n", (*request)->mpi_request);


    if(src>=0)
    {
      int mpi_src = comm->ep_rank_map->at(src).second;
      int mpi_tag = tag_combine(tag, comm->ep_rank_map->at(src).first, comm->ep_comm_ptr->size_rank_info[1].first);
      //printf("MPI_Irecv with src\n");
      //printf("proc %d recv from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, src);
      return ::MPI_Irecv(buf, count, to_mpi_type(datatype), mpi_src, mpi_tag, to_mpi_comm(comm->mpi_comm), to_mpi_request_ptr(*request));
    }
    else
    {
      //printf("MPI_Irecv with ANY_SOURCE\n");
      int probed=false;
      int mpi_tag;
      ::MPI_Status mpi_status;
      while(!probed)
      {
        //for(int i=0; i<comm->ep_comm_ptr->size_rank_info[1].second; i++)
        for(int i=0; i<4; i++)
        {
          mpi_tag = tag_combine(tag, i, comm->ep_comm_ptr->size_rank_info[1].first);
          ::MPI_Iprobe(MPI_ANY_SOURCE, mpi_tag, to_mpi_comm(comm->mpi_comm), &probed, &mpi_status);
          if(probed)
          {
            //printf("proc %d recv from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, get_ep_rank(comm, i, mpi_status.MPI_SOURCE));
            return ::MPI_Irecv(buf, count, to_mpi_type(datatype), mpi_status.MPI_SOURCE, mpi_status.MPI_TAG, to_mpi_comm(comm->mpi_comm), to_mpi_request_ptr(*request));
          }
        }
      }
    }
    

  }

  


  int MPI_Irecv2(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request *request)
  {
    if(!comm->is_ep) return MPI_Irecv_mpi(buf, count, datatype, src, tag, comm, request);

    if(comm->is_intercomm)
    {
      if(src>=0) src = comm->inter_rank_map->at(src);
    }
    
    Debug("MPI_Irecv with EP");
    int dest_rank;
    MPI_Comm_rank(comm, &dest_rank);
    
    *request = new ep_request;   
    memcheck("new " << *request <<" : in ep_lib::MPI_Irecv, *request = new ep_request");

    (*request)->mpi_request = new ::MPI_Request;
    memcheck("new " << (*request)->mpi_request << " : in ep_lib::MPI_Irecv, (*request)->mpi_request = new ::MPI_Request");
    
    (*request)->buf = buf;
    (*request)->comm = comm;
    (*request)->type = 2;
    (*request)->probed = false;
    (*request)->state = 0;

    (*request)->ep_src = src;
    (*request)->ep_tag = tag;
    (*request)->ep_datatype = datatype;
  
    if(EP_PendingRequests == 0 ) EP_PendingRequests = new std::list< MPI_Request* >;

    EP_PendingRequests->push_back(request); 
    
    //memcheck("EP_PendingRequests["<<ep_rank<<"]->size() = " << EP_PendingRequests->size());    
    
    show_EP_PendingRequests(EP_PendingRequests);

#ifdef _showinfo
    if(comm->is_intercomm)
    {
      int ep_dest_loc  = comm->ep_rank_map->at(dest_rank).first;
      int ep_src_loc = comm->ep_comm_ptr->intercomm->intercomm_rank_map->at(src).first;
      int mpi_tag     = tag_combine(tag, ep_src_loc, ep_dest_loc);
      int mpi_dest    = comm->ep_comm_ptr->intercomm->intercomm_rank_map->at(src).second.first;

      printf("Irecv : ep_src_loc = %d, ep_dest_loc = %d, mpi_src = %d, mpi_dest = %d, mpi_tag = %d\n", ep_src_loc, ep_dest_loc, comm->ep_comm_ptr->size_rank_info[2].first, mpi_dest, mpi_tag);
    }                                                          
#endif                                      

    return Request_Check();
  }
  
  int MPI_Mrecv(void *buf, int count, MPI_Datatype datatype, MPI_Message *message, MPI_Status *status)
  {
    Debug("MPI_Mrecv with MPI/EP");

    status->mpi_status = new ::MPI_Status;
    memcheck("new " << status->mpi_status << " : in ep_lib::MPI_Mrecv, status->mpi_status = new ::MPI_Status");
    
    ::MPI_Mrecv(buf, count, to_mpi_type(datatype), static_cast< ::MPI_Message* >((*message)->mpi_message), to_mpi_status_ptr(*status));

    
    status->ep_src = (*message)->ep_src;
    status->ep_datatype = datatype;
    status->ep_tag = (*message)->ep_tag;

    memcheck("delete " << (*message)->mpi_message << " : in ep_lib::MPI_Mrecv, delete (*message)->mpi_message");
    delete (*message)->mpi_message;
    memcheck("delete " << *message << " : in ep_lib::MPI_Imrecv, delete *message");
    delete *message;

#ifdef _check_sum
    check_sum_recv(buf, count, datatype, message->ep_src, message->ep_tag);
#endif

    return Request_Check();
  }


  int MPI_Imrecv(void *buf, int count, MPI_Datatype datatype, MPI_Message *message, MPI_Request *request)
  {
    Debug("MPI_Imrecv with MPI/EP");

    (*request)->type = 3;
    (*request)->ep_datatype = datatype;
    (*request)->ep_tag = (*message)->ep_tag;
    (*request)->ep_src = (*message)->ep_src;
    
    (*request)->probed = true;
    (*request)->state = 1;
        
    ::MPI_Imrecv(buf, count, to_mpi_type(datatype), to_mpi_message_ptr(*message), to_mpi_request_ptr(*request));                
    
    memcheck("delete " << (*message)->mpi_message << " : in ep_lib::MPI_Imrecv, delete (*message)->mpi_message");
    delete (*message)->mpi_message;


#ifdef _check_sum
    check_sum_recv(buf, count, datatype, message->ep_src, message->ep_tag);
#endif

    
    //return Request_Check();
  }


   int MPI_Recv_mpi(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status *status)
  {
    Debug("MPI_Recv with MPI");
    status->ep_src = src;
    status->ep_tag = tag;
    status->ep_datatype = datatype;
    
    return ::MPI_Recv(buf, count, to_mpi_type(datatype), src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), to_mpi_status_ptr(*status)); 
  }
  
  int MPI_Irecv_mpi(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request *request)
  {
    Debug("MPI_Irecv with MPI");
    int dest_rank;
    MPI_Comm_rank(comm, &dest_rank);
    
    *request = new ep_request;
    memcheck("new "<< *request <<" : in ep_lib::MPI_Irecv, *request = new ep_request");

    (*request)->mpi_request = new ::MPI_Request;
    memcheck("new "<< (*request)->mpi_request <<" : in ep_lib::MPI_Irecv, (*request)->mpi_request = new ::MPI_Request");
      
    (*request)->ep_src = src;
    (*request)->ep_datatype = datatype;
    (*request)->type = 2;
    (*request)->ep_tag = tag;
    
    return ::MPI_Irecv(buf, count, to_mpi_type(datatype), src<0? MPI_ANY_SOURCE : src, tag<0? MPI_ANY_TAG: tag, to_mpi_comm(comm->mpi_comm), to_mpi_request_ptr(*request));
  }
}



#endif