Changeset 1187

Timestamp:

06/28/17 17:46:00 (7 years ago)

Author:

yushan

Message:

add pending list for irecv

Location:

XIOS/dev/branch_yushan_merged

Files:

• bld.cfg (modified) (1 diff)
• extern/src_ep_dev/ep_lib.cpp (modified) (2 diffs)
• extern/src_ep_dev/ep_message.cpp (modified) (2 diffs)
• extern/src_ep_dev/ep_recv.cpp (modified) (1 diff)
• extern/src_ep_dev/ep_type.hpp (modified) (2 diffs)
• extern/src_ep_dev/ep_wait.cpp (modified) (6 diffs)
• src/client.cpp (modified) (1 diff)
• src/server.cpp (modified) (1 diff)

XIOS/dev/branch_yushan_merged/bld.cfg

@@ -34 +34 @@
 #bld::target generate_fortran_interface.exe 
 #bld::target xios_server.exe 
-bld::target test_remap.exe
+#bld::target test_remap.exe test_remap_omp.exe
 #bld::target test_regular.exe
 #bld::target test_expand_domain.exe
 #bld::target test_new_features.exe test_unstruct_complete.exe 
-bld::target test_omp.exe test_complete_omp.exe test_remap_omp.exe test_unstruct_omp.exe
+bld::target test_omp.exe #test_complete_omp.exe test_remap_omp.exe test_unstruct_omp.exe
 #bld::target test_client.exe test_complete.exe #test_xios2_cmip6.exe
 #bld::target test_connectivity_expand.exe

XIOS/dev/branch_yushan_merged/extern/src_ep_dev/ep_lib.cpp

@@ -9 +9 @@
 
 namespace ep_lib
-{
-
+{ 
 
   int tag_combine(int real_tag, int src, int dest)
@@ -368 +367 @@
 
   }
+
+  int test_sendrecv(MPI_Comm comm)
+  {
+    int myRank;
+    MPI_Comm_rank(comm, &myRank);
+    bool amClient = false;
+    bool amServer = false;
+    if(myRank<=3) amClient = true;
+    else amServer = true;
+
+    if(amServer)
+    {
+      int send_buf[4];
+      MPI_Request send_request[8];
+      MPI_Status send_status[8];
+
+      
+      
+      for(int j=0; j<4; j++)  // 4 buffers
+      {
+        for(int i=0; i<2; i++)
+        {
+          send_buf[j] = (myRank+1)*100 + j;
+          MPI_Isend(&send_buf[j], 1, MPI_INT, i*2, 9999, comm, &send_request[i*4+j]);
+        }
+      }
+      
+
+      MPI_Waitall(8, send_request, send_status);
+    }
+
+
+    if(amClient&&myRank%2==0) // Clients leaders
+    {
+      int recv_buf[8];
+      MPI_Request recv_request[8];
+      MPI_Status recv_status[8];
+
+      for(int i=0; i<2; i++)  // 2 servers
+      {
+        for(int j=0; j<4; j++)
+        {
+          MPI_Irecv(&recv_buf[i*4+j], 1, MPI_INT, i+4, 9999, comm, &recv_request[i*4+j]);
+        }
+      }
+
+      MPI_Waitall(8, recv_request, recv_status);
+      printf("============ client %d, recv_buf = %d, %d, %d, %d, %d, %d, %d, %d ================\n", 
+              myRank, recv_buf[0], recv_buf[1], recv_buf[2], recv_buf[3], recv_buf[4], recv_buf[5], recv_buf[6], recv_buf[7]);
+    }
+
+    MPI_Barrier(comm);
+
+  }
+
 }
 

XIOS/dev/branch_yushan_merged/extern/src_ep_dev/ep_message.cpp

@@ -79 +79 @@
         int dest_mpi = comm.ep_comm_ptr->size_rank_info[2].first;
         int ep_dest = get_ep_rank(comm, dest_loc, dest_mpi);
-        printf("myRank = %d, probed one message, ep_src = %d, ep_dest = %d, tag = %d, message = %d\n", myRank, msg_block->ep_src, ep_dest, msg_block->ep_tag, msg_block->mpi_message);
+        //printf("myRank = %d, probed one message, ep_src = %d, ep_dest = %d, tag = %d, message = %d\n", myRank, msg_block->ep_src, ep_dest, msg_block->ep_tag, msg_block->mpi_message);
         msg_block->mpi_status = new ::MPI_Status(status);
 
@@ -90 +90 @@
           #pragma omp flush
           ptr_comm_target->ep_comm_ptr->message_queue->push_back(*msg_block);  
-          printf("myRank = %d, push_back OK, ep_src = %d, ep_tag = %d, dest = %d(%d)\n", myRank, 
-                                                                                             ptr_comm_target->ep_comm_ptr->message_queue->back().ep_src,
-                                                                                             ptr_comm_target->ep_comm_ptr->message_queue->back().ep_tag,
-                                                                                             ep_dest, dest_loc);
+          //printf("myRank = %d, push_back OK, ep_src = %d, ep_tag = %d, dest = %d(%d)\n", myRank, 
+          //                                                                                   ptr_comm_target->ep_comm_ptr->message_queue->back().ep_src,
+          //                                                                                   ptr_comm_target->ep_comm_ptr->message_queue->back().ep_tag,
+          //                                                                                   ep_dest, dest_loc);
     
           #pragma omp flush

XIOS/dev/branch_yushan_merged/extern/src_ep_dev/ep_recv.cpp

@@ -76 +76 @@
     request->ep_datatype = datatype;
 
+
+
     /* With Improbe*/
     Message_Check(comm);
 
+    if(EP_PendingRequests == 0 ) 
+    {
+      EP_PendingRequests = new std::list< MPI_Request* >; 
+      printf("proc %d : EP_PendingRequests allocated, add = %p\n", dest_rank, EP_PendingRequests);  
+    }
+
+    request->pending_ptr = EP_PendingRequests;
+    printf("proc %d : &EP_PendingRequests add = %p, ptr = %p\n", dest_rank, EP_PendingRequests, request->pending_ptr);  
+    
     EP_PendingRequests->push_back(request);
-    printf("proc %d : EP_PendingRequests insert one request, add = %p(%p), buf = %p(%p)\n", dest_rank, EP_PendingRequests->back(), request, buf, request->buf);
+    //printf("proc %d : EP_PendingRequests insert one request, src = %d, tag = %d, size = %d\n", dest_rank, request->ep_src, request->ep_tag, EP_PendingRequests->size());
     
     // check all EP_PendingRequests
     
-    //if(EP_PendingRequests == 0 ) EP_PendingRequests = new std::list< MPI_Request* >; 
-    if(!EP_PendingRequests->empty())
+    //printf("proc %d have %d pending irecv request\n", dest_rank, EP_PendingRequests->size());
+      
+    for(std::list<MPI_Request* >::iterator it = EP_PendingRequests->begin(); it!=EP_PendingRequests->end(); )
     {
-      printf("proc %d have %d pending irecv request\n", dest_rank, EP_PendingRequests->size());
-      
-      for(std::list<MPI_Request* >::iterator it = EP_PendingRequests->begin(); it!=EP_PendingRequests->end(); )
+      if((*it)->type == 3) 
       {
-        if((*it)->type != 2) 
-        {
-          printf("proc %d : pending request type = %d, src= %d, tag = %d, add = %p skip\n", dest_rank, (*it)->type, (*it)->ep_src, (*it)->ep_tag, *it);
-          EP_PendingRequests->erase(it);
-          it = EP_PendingRequests->begin();
-          printf("proc %d : pending request processed, size = %d, it = %p\n", dest_rank, EP_PendingRequests->size(), *it);
-          continue;
-        }
+        //printf("proc %d : pending request type = %d, src= %d, tag = %d    skip\n", dest_rank, (*it)->type, (*it)->ep_src, (*it)->ep_tag);
+        EP_PendingRequests->erase(it);
+        it = EP_PendingRequests->begin();
+        //printf("proc %d : pending request processed, size = %d\n", dest_rank, EP_PendingRequests->size());
+        continue;
+      }
         
-        printf("proc %d : pending irecv request src = %d, tag = %d, type = %d, add = %p\n", dest_rank, (*it)->ep_src, (*it)->ep_tag, (*it)->type, *it);
-        int probed = false;
-        MPI_Message pending_message;
-        MPI_Status pending_status;
+      //printf("proc %d : pending irecv request src = %d, tag = %d, type = %d\n", dest_rank, (*it)->ep_src, (*it)->ep_tag, (*it)->type);
+      int probed = false;
+      MPI_Message pending_message;
+      MPI_Status pending_status;
     
-        MPI_Improbe((*it)->ep_src, (*it)->ep_tag, (*it)->comm, &probed, &pending_message, &pending_status);
-        printf("proc %d : pending irecv request probed to be %d, add = %p\n",dest_rank, probed, *it);
+      MPI_Improbe((*it)->ep_src, (*it)->ep_tag, (*it)->comm, &probed, &pending_message, &pending_status);
+      //printf("proc %d : pending irecv request probed to be %d, src = %d, tag = %d\n",dest_rank, probed, (*it)->ep_src, (*it)->ep_tag);
     
-        if(probed) 
-        { 
-          int count;
-          MPI_Get_count(&pending_status, (*it)->ep_datatype, &count);
-          MPI_Imrecv((*it)->buf, count, (*it)->ep_datatype, &pending_message, *it);
-          printf("proc %d : pending request is imrecving src = %d, tag = %d, add = %p, buf = %p, count = %d\n", dest_rank, (*it)->ep_src, (*it)->ep_tag, *it, (*it)->buf, count);
-          EP_PendingRequests->erase(it);
-          it = EP_PendingRequests->begin();
-          printf("proc %d : pending request processed, size = %d\n", dest_rank, EP_PendingRequests->size());
-          continue;
-        }
-        else it++;
+      if(probed) 
+      { 
+        int count;
+        MPI_Get_count(&pending_status, (*it)->ep_datatype, &count);
+        MPI_Imrecv((*it)->buf, count, (*it)->ep_datatype, &pending_message, *it);
+        //printf("proc %d : pending request is imrecving src = %d, tag = %d, count = %d\n", dest_rank, (*it)->ep_src, (*it)->ep_tag, count);
+        EP_PendingRequests->erase(it);
+        it = EP_PendingRequests->begin();
+        //printf("proc %d : pending request processed, size = %d\n", dest_rank, EP_PendingRequests->size());
+        continue;
       }
+
+      it++;
     }
-/*
-    int flag = false;
-    MPI_Message message;
-    MPI_Status status;
     
-    MPI_Improbe(src, tag, comm, &flag, &message, &status);
-            
-    if(flag)
-    {
-      MPI_Imrecv(buf, count, datatype, &message, request);
-      printf("proc %d : found message in local queue, src = %d, tag = %d\n", dest_rank, src, tag);
-    }
-  */  
-
     return 0;
   }

XIOS/dev/branch_yushan_merged/extern/src_ep_dev/ep_type.hpp

@@ -453 +453 @@
       MPI_Request(void* request): mpi_request(request) {}
       #endif
+
+      std::list< MPI_Request* > * pending_ptr;
   };
 
@@ -490 +492 @@
             //    <MPI_Fint,thread_num>   EP_Comm
 
-  static std::list< MPI_Request* > * EP_PendingRequests = new std::list< MPI_Request* >;
+  static std::list< MPI_Request* > * EP_PendingRequests = 0;
   #pragma omp threadprivate(EP_PendingRequests)
 }

XIOS/dev/branch_yushan_merged/extern/src_ep_dev/ep_wait.cpp

@@ -20 +20 @@
   {
 
-    if(request->type == 1)
+    if(request->type == 1)  //=>isend
     {
       ::MPI_Request mpi_request = static_cast< ::MPI_Request >(request->mpi_request);
@@ -44 +44 @@
     }
 
-    if(request->type == 2)
-    {
-      int flag = false;
-      MPI_Message message;
-
-      while(!flag)
-      {
-        Message_Check(request->comm);
-        #pragma omp flush
-        MPI_Improbe(request->ep_src, request->ep_tag, request->comm, &flag, &message, status);
-      }
-
-      int count;
-      request->type = 3;
-      MPI_Get_count(status, request->ep_datatype, &count);
-      MPI_Mrecv(request->buf, count, request->ep_datatype, &message, status);
-      status->ep_datatype = request->ep_datatype;
-
-      //check_sum_recv(request->buf, count, request->ep_datatype, request->ep_src, request->ep_tag, request->comm, 2);
-
-      return 0;
-    }
-
-    if(request->type == 3)
+    if(request->type == 3) //=>imrecv
     {
       ::MPI_Request mpi_request = static_cast< ::MPI_Request >(request->mpi_request);
@@ -90 +67 @@
 
       //check_sum_recv(request->buf, count, request->ep_datatype, request->ep_src, request->ep_tag, request->comm, 2);
+      return 0;
+
+    }
+
+    if(request->type == 2) //=>irecv not probed
+    {
+      
+      while(true)
+      {
+        Message_Check(request->comm);
+        // parcours pending list
+        for(std::list<MPI_Request* >::iterator it = (request->pending_ptr)->begin(); it!=(request->pending_ptr)->end(); )
+        {
+          if(*it == request)
+          { 
+            int probed = false;
+            MPI_Message message;
+
+            MPI_Improbe(request->ep_src, request->ep_tag, request->comm, &probed, &message, status);
+                  
+            if(probed)
+            {
+              int recv_count;
+              MPI_Get_count(status, request->ep_datatype, &recv_count);
+              MPI_Mrecv(request->buf, recv_count, request->ep_datatype, &message, status);
+              (request->pending_ptr)->erase(it);
+              //printf("wait  : pending request processed, size = %d\n", (request->pending_ptr)->size());
+              request->type = 3;
+            
+              return 0;
+            }
+
+            it++;
+          }
+          else 
+          {
+            int probed = false;
+            MPI_Message message;
+            MPI_Status status;
+
+            MPI_Improbe((*it)->ep_src, (*it)->ep_tag, (*it)->comm, &probed, &message, &status);
+                  
+            if(probed)
+            {
+              int recv_count;
+              MPI_Get_count(&status, (*it)->ep_datatype, &recv_count);
+              MPI_Imrecv((*it)->buf, recv_count, (*it)->ep_datatype, &message, *it);
+                  
+              (request->pending_ptr)->erase(it);
+                  
+              it = (request->pending_ptr)->begin();
+              //printf("wait  : pending request processed, size = %d\n", (request->pending_ptr)->size());
+            }
+            else it++;
+          }
+        }
+
+      }
+      
+
     }
 
@@ -111 +148 @@
 
     for(int i=0; i<count; i++)
+      printf("pending add = %p\n",  array_of_requests[i].pending_ptr);
+
+    //if(EP_PendingRequests == 0) EP_PendingRequests = new std::list< MPI_Request* >;  
+    //printf("pending size = %d, add = %p\n", EP_PendingRequests->size(), EP_PendingRequests);
+
+    for(int i=0; i<count; i++)
     {
       finished_index[i] = false;
@@ -117 +160 @@
     while(finished < count)
     {
-
+      
       for(int i=0; i<count; i++)
       {
         if(finished_index[i] == false) // this request has not been tested.
         {
-          if(array_of_requests[i].type != 2) // isend or imrecv
+          if(array_of_requests[i].type == 1 || array_of_requests[i].type == 3) // isend or imrecv
           {      
             //MPI_Wait(&array_of_requests[i], &array_of_statuses[i]);
@@ -133 +176 @@
           else // irecv
           {
+            
+            Message_Check(array_of_requests[i].comm);
             // parcours pending list
-            // find request in waitall
-                Message_Check(array_of_requests[i].comm);
-               // improbe + mrecv
-               // erase element in pending list
- //             finished++;
- //             finished_index[i] = true;
-            
-
-            int flag = false;
-            MPI_Message message;
-
-            MPI_Improbe(array_of_requests[i].ep_src, array_of_requests[i].ep_tag, array_of_requests[i].comm, &flag, &message, &array_of_statuses[i]);
-
-            if(flag)
+            for(std::list<MPI_Request* >::iterator it = (array_of_requests[i].pending_ptr)->begin(); it!=(array_of_requests[i].pending_ptr)->end(); )
             {
-              int recv_count;
-              MPI_Get_count(&array_of_statuses[i], array_of_requests[i].ep_datatype, &recv_count);
-              MPI_Mrecv(array_of_requests[i].buf, recv_count, array_of_requests[i].ep_datatype, &message, &array_of_statuses[i]);
-              //check_sum_recv(array_of_requests[i].buf, recv_count, array_of_requests[i].ep_datatype, array_of_requests[i].ep_src, array_of_requests[i].ep_tag, array_of_requests[i].comm, 2);
-
-              finished++;
-              finished_index[i] = true;
+              bool matched = false;
+              for(int j=0; j<count; j++)
+              {
+                if(*it == &array_of_requests[j])
+                { 
+                  int probed = false;
+                  MPI_Message message;
+
+                  MPI_Improbe((*it)->ep_src, (*it)->ep_tag, (*it)->comm, &probed, &message, &array_of_statuses[j]);
+                  
+                  if(probed)
+                  {
+                    int recv_count;
+                    MPI_Get_count(&array_of_statuses[j], array_of_requests[j].ep_datatype, &recv_count);
+                    MPI_Mrecv(array_of_requests[j].buf, recv_count, array_of_requests[j].ep_datatype, &message, &array_of_statuses[j]);
+                    //check_sum_recv(array_of_requests[i].buf, recv_count, array_of_requests[i].ep_datatype, array_of_requests[i].ep_src, array_of_requests[i].ep_tag, array_of_requests[i].comm, 2);
+                    (array_of_requests[i].pending_ptr)->erase(it);
+                    array_of_requests[j].type = 3;
+                    finished++;
+                    finished_index[j] = true;
+                    matched = true;
+                    it = (array_of_requests[i].pending_ptr)->begin();
+                    j=count;
+                    //printf("waitall  : pending request processed, size = %d\n", (*(array_of_requests[i].pending_ptr))->size());
+                    //printf("matched = %d, j=%d, src = %d, tag = %d, probed = %d\n", matched, j, (*it)->ep_src, (*it)->ep_tag, probed);
+                  }
+                }
+
+              }
+
+              if(!matched)
+              {
+                int probed = false;
+                MPI_Message message;
+                MPI_Status status;
+
+                MPI_Improbe((*it)->ep_src, (*it)->ep_tag, (*it)->comm, &probed, &message, &status);
+                  
+                if(probed)
+                {
+                  int recv_count;
+                  MPI_Get_count(&status, (*it)->ep_datatype, &recv_count);
+                  MPI_Imrecv((*it)->buf, recv_count, (*it)->ep_datatype, &message, *it);
+                  
+                  (array_of_requests[i].pending_ptr)->erase(it);
+                  
+                  it = (array_of_requests[i].pending_ptr)->begin();
+                  //printf("waitall  : pending request processed, size = %d\n", (*(array_of_requests[i].pending_ptr))->size());
+                }
+                else it++;
+              }
             }
+
           }
         }

XIOS/dev/branch_yushan_merged/src/client.cpp

@@ -107 +107 @@
             MPI_Comm_rank(intraComm,&intraCommRank) ;
             
-            /*#pragma omp critical(_output)
+            #pragma omp critical(_output)
             {
               info(10)<<"intercommCreate::client "<<test_omp_rank<< " "<< &test_omp_rank <<" intraCommSize : "<<intraCommSize
                  <<" intraCommRank :"<<intraCommRank<<"  serverLeader "<< serverLeader
                  <<" globalComm : "<< &(CXios::globalComm) << endl ;  
-            }*/
+            }
 
             

XIOS/dev/branch_yushan_merged/src/server.cpp

@@ -90 +90 @@
              MPI_Comm_size(intraComm,&intraCommSize) ;
              MPI_Comm_rank(intraComm,&intraCommRank) ;
-             /*info(50)<<"intercommCreate::server "<<rank<<" intraCommSize : "<<intraCommSize
-                     <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;*/
+             info(50)<<"intercommCreate::server "<<rank<<" intraCommSize : "<<intraCommSize
+                     <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
 
              test_sendrecv(CXios::globalComm);