source: XIOS/dev/dev_trunk_omp/extern/src_ep_dev2/ep_probe.cpp @ 1838

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

namespace ep_lib
{
  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;
  }


  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)
  {
    *flag = false;
    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_Iprobe with src\n");
      
      ::MPI_Iprobe(mpi_src, mpi_tag, to_mpi_comm(comm->mpi_comm), flag, &mpi_status);
      if(*flag)
      {
        //printf("proc %d probe from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, src);
        status->mpi_status = new ::MPI_Status(mpi_status);
      }
      return MPI_SUCCESS;
    }
    else
    {
      //printf("MPI_Iprobe with ANY_SOURCE\n");
      //for(int i=0; i<comm->ep_comm_ptr->size_rank_info[1].second; i++)
      for(int i=0; i<4; i++)
      {
        int 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), flag, &mpi_status);
        if(*flag) 
        {
          //printf("proc %d probe from proc %d\n", comm->ep_comm_ptr->size_rank_info[0].first, get_ep_rank(comm, i, mpi_status.MPI_SOURCE));
          status->mpi_status = new ::MPI_Status(mpi_status);
          //printf("get_ep_rank in MPI_Iprobe\n");
          status->ep_src = get_ep_rank(comm, i, mpi_status.MPI_SOURCE);
          return MPI_SUCCESS;
        }
    
      }
      
    }

  }



  int MPI_Iprobe2(int src, int tag, MPI_Comm comm, int *flag, MPI_Status *status)
  {
    if(!comm->is_ep)
    {
      Debug("MPI_Iprobe with MPI\n");
      return MPI_Iprobe_mpi(src, tag, comm, flag, status);
    }

    if(comm->is_intercomm)
    {
      if(src>=0) src = comm->inter_rank_map->at(src);
    } 
    
    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;
    
    #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;
        
        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;
          }
        }

        *flag = true;
        break;
      }
    }
    if(*flag) return 0;
    
    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)
    {
      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;
        
        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;
          }
        }

        *flag = true;
        break;
      }
    }
    if(*flag) return 0;
  }

  

  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;
    
    #pragma omp critical (_query)
    if(! comm->ep_comm_ptr->message_queue->empty())
    {
      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)
        {
          *flag = true;

          status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
          memcheck("new "<< status->mpi_status << " : in ep_lib::MPI_Improbe, status->mpi_status = new ::MPI_Status");
          status->ep_src = (*it)->ep_src;
          status->ep_tag = (*it)->ep_tag;

          (*message)->mpi_message = new ::MPI_Message(*static_cast< ::MPI_Message*>((*it)->mpi_message));
          memcheck("new "<< (*message)->mpi_message <<" : in ep_lib::MPI_Improbe, (*message)->mpi_message = new ::MPI_Message");
          (*message)->ep_src = (*it)->ep_src;
          (*message)->ep_tag = (*it)->ep_tag;
                                      

          #pragma omp critical (_query2)
          {              
            memcheck("delete "<< (*it)->mpi_message <<" : in ep_lib::Message_Check, delete (*it)->mpi_message");
            memcheck("delete "<< (*it)->mpi_status <<" : in ep_lib::Message_Check, delete (*it)->mpi_status");
            memcheck("delete "<< (*it) <<" : in ep_lib::Message_Check, delete (*it)");
            
            
            delete (*it)->mpi_message;     
            delete (*it)->mpi_status; 
            delete *it;
            
                       
            comm->ep_comm_ptr->message_queue->erase(it);
            memcheck("message_queue["<<mpi_rank<<","<<ep_rank_loc<<"]->size = "<<comm->ep_comm_ptr->message_queue->size());
            #pragma omp flush
          }
          
          break;
        }

      }
    }

    if(*flag) return 0;
   
    Message_Check(comm);
    
    #pragma omp flush

    #pragma omp critical (_query)
    if(! comm->ep_comm_ptr->message_queue->empty())
    {
      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)
        {
          *flag = true;

          status->mpi_status = new ::MPI_Status(*static_cast< ::MPI_Status*>((*it)->mpi_status));
          memcheck("new "<< status->mpi_status << " : in ep_lib::MPI_Improbe, status->mpi_status = new ::MPI_Status");
          status->ep_src = (*it)->ep_src;
          status->ep_tag = (*it)->ep_tag;

          (*message)->mpi_message = new ::MPI_Message(*static_cast< ::MPI_Message*>((*it)->mpi_message));
          memcheck("new "<< (*message)->mpi_message <<" : in ep_lib::MPI_Improbe, (*message)->mpi_message = new ::MPI_Message");
          (*message)->ep_src = (*it)->ep_src;
          (*message)->ep_tag = (*it)->ep_tag;
                                      

          #pragma omp critical (_query2)
          {              
            memcheck("delete "<< (*it)->mpi_message <<" : in ep_lib::Message_Check, delete (*it)->mpi_message");
            memcheck("delete "<< (*it)->mpi_status <<" : in ep_lib::Message_Check, delete (*it)->mpi_status");
            memcheck("delete "<< (*it) <<" : in ep_lib::Message_Check, delete (*it)");
            
            
            delete (*it)->mpi_message;     
            delete (*it)->mpi_status; 
            delete *it;
            
                       
            comm->ep_comm_ptr->message_queue->erase(it);
            memcheck("message_queue["<<mpi_rank<<","<<ep_rank_loc<<"]->size = "<<comm->ep_comm_ptr->message_queue->size());
            #pragma omp flush
          }
          
          break;
        }

      }
    }

    if(*flag) return 0;

  }

}


#endif