Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add MPI_Bsend, MPI_Ibsend, MPI_Bsend_init, MPI_Buffer_attach, MPI_Buffer_detach.
[simgrid.git] / src / smpi / mpi / smpi_request.cpp
index 05cdc18..c5e7c52 100644 (file)
@@ -2,8 +2,7 @@
 
 /* This program is free software; you can redistribute it and/or modify it
  * under the terms of the license (GNU LGPL) which comes with this package. */
-#include "simgrid/s4u/Mutex.hpp"
-#include "simgrid/s4u/ConditionVariable.hpp"
+
 #include "smpi_request.hpp"
 
 #include "mc/mc.h"
@@ -17,8 +16,6 @@
 #include "src/kernel/activity/CommImpl.hpp"
 #include "src/mc/mc_replay.hpp"
 #include "src/smpi/include/smpi_actor.hpp"
-#include "xbt/config.hpp"
-
 
 #include <algorithm>
 
@@ -173,6 +170,14 @@ void Request::print_request(const char *message)
 
 
 /* factories, to hide the internal flags from the caller */
+MPI_Request Request::bsend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+{
+
+  return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
+                     comm->group()->actor(dst)->get_pid(), tag, comm,
+                     MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED | MPI_REQ_BSEND);
+}
+
 MPI_Request Request::send_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
 
@@ -245,6 +250,16 @@ MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int
                      MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED);
 }
 
+MPI_Request Request::ibsend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+{
+  MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
+  request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
+                        comm->group()->actor(dst)->get_pid(), tag, comm,
+                        MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_BSEND);
+  request->start();
+  return request;
+}
+
 MPI_Request Request::isend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
@@ -284,6 +299,17 @@ void Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag
   request = nullptr;
 }
 
+void Request::bsend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
+{
+  MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
+  request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(),
+                        comm->group()->actor(dst)->get_pid(), tag, comm, MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND | MPI_REQ_BSEND);
+
+  request->start();
+  wait(&request, MPI_STATUS_IGNORE);
+  request = nullptr;
+}
+
 void Request::send(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm)
 {
   MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */
@@ -401,6 +427,7 @@ void Request::start()
       mut->unlock();
   } else { /* the RECV flag was not set, so this is a send */
     simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
+    xbt_assert(process, "Actor pid=%d is gone??", dst_);
     int rank = src_;
     if (TRACE_smpi_view_internals()) {
       TRACE_smpi_send(rank, rank, dst_, tag_, size_);
@@ -409,7 +436,7 @@ void Request::start()
 
     void* buf = buf_;
     if ((flags_ & MPI_REQ_SSEND) == 0 &&
-        ((flags_ & MPI_REQ_RMA) != 0 ||
+        ((flags_ & MPI_REQ_RMA) != 0 || (flags_ & MPI_REQ_BSEND) != 0 ||
          static_cast<int>(size_) < simgrid::config::get_value<int>("smpi/send-is-detached-thresh"))) {
       void *oldbuf = nullptr;
       detached_    = true;
@@ -424,6 +451,8 @@ void Request::start()
             XBT_DEBUG("Privatization : We are sending from a zone inside global memory. Switch data segment ");
             smpi_switch_data_segment(simgrid::s4u::Actor::by_pid(src_));
           }
+          //we need this temporary buffer even for bsend, as it will be released in the copy callback and we don't have a way to differentiate it
+          //so actually ... don't use manually attached buffer space.
           buf = xbt_malloc(size_);
           memcpy(buf,oldbuf,size_);
           XBT_DEBUG("buf %p copied into %p",oldbuf,buf);
@@ -441,7 +470,7 @@ void Request::start()
     }
 
     if(sleeptime > 0.0){
-      simcall_process_sleep(sleeptime);
+      simgrid::s4u::this_actor::sleep_for(sleeptime);
       XBT_DEBUG("sending size of %zu : sleep %f ", size_, sleeptime);
     }
 
@@ -542,12 +571,12 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) {
   }
   
   if(smpi_test_sleep > 0)
-    simcall_process_sleep(nsleeps*smpi_test_sleep);
+    simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep);
 
   Status::empty(status);
   *flag = 1;
   if (((*request)->flags_ & MPI_REQ_PREPARED) == 0) {
-    if ((*request)->action_ != nullptr){
+    if ((*request)->action_ != nullptr && (*request)->cancelled_ != 1){
       try{
         *flag = simcall_comm_test((*request)->action_);
       } catch (const Exception&) {
@@ -594,15 +623,15 @@ int Request::testsome(int incount, MPI_Request requests[], int *count, int *indi
 
   *count = 0;
   for (int i = 0; i < incount; i++) {
-    if (requests[i] != MPI_REQUEST_NULL) {
+    if (requests[i] != MPI_REQUEST_NULL && not (requests[i]->flags_ & MPI_REQ_FINISHED)) {
       ret = test(&requests[i], pstat, &flag);
       if(ret!=MPI_SUCCESS)
         error = 1;
       if(flag) {
-        indices[i] = 1;
-        (*count)++;
+        indices[*count] = i;
         if (status != MPI_STATUSES_IGNORE)
-          status[i] = *pstat;
+          status[*count] = *pstat;
+        (*count)++;
         if ((requests[i] != MPI_REQUEST_NULL) && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT))
           requests[i] = MPI_REQUEST_NULL;
       }
@@ -638,7 +667,7 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M
     //multiplier to the sleeptime, to increase speed of execution, each failed testany will increase it
     static int nsleeps = 1;
     if(smpi_test_sleep > 0)
-      simcall_process_sleep(nsleeps*smpi_test_sleep);
+      simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep);
     try{
       i = simcall_comm_testany(comms.data(), comms.size()); // The i-th element in comms matches!
     } catch (const Exception&) {
@@ -758,7 +787,7 @@ void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status*
   s4u::Mailbox* mailbox;
 
   request->print_request("New iprobe");
-  // We have to test both mailboxes as we don't know if we will receive one one or another
+  // We have to test both mailboxes as we don't know if we will receive one or another
   if (simgrid::config::get_value<int>("smpi/async-small-thresh") > 0) {
     mailbox = smpi_process()->mailbox_small();
     XBT_DEBUG("Trying to probe the perm recv mailbox");
@@ -856,8 +885,8 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status)
     //integrate pseudo-timing for buffering of small messages, do not bother to execute the simcall if 0
     double sleeptime =
         simgrid::s4u::Actor::self()->get_host()->extension<simgrid::smpi::Host>()->orecv(req->real_size());
-    if(sleeptime > 0.0){
-      simcall_process_sleep(sleeptime);
+    if (sleeptime > 0.0) {
+      simgrid::s4u::this_actor::sleep_for(sleeptime);
       XBT_DEBUG("receiving size of %zu : sleep %f ", req->real_size_, sleeptime);
     }
     unref(&(req->detached_sender_));
@@ -1060,7 +1089,6 @@ int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
   int index = 0;
   MPI_Status stat;
   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
-
   index = waitany(incount, (MPI_Request*)requests, pstat);
   if(index==MPI_UNDEFINED) return MPI_UNDEFINED;
   if(status != MPI_STATUSES_IGNORE) {
@@ -1069,9 +1097,8 @@ int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
   indices[count] = index;
   count++;
   for (int i = 0; i < incount; i++) {
-    if (i==index)
-      continue;
-    if (requests[i] != MPI_REQUEST_NULL) {
+    if (i!=index && requests[i] != MPI_REQUEST_NULL 
+        && not(requests[i]->flags_ & MPI_REQ_FINISHED)) {
       test(&requests[i], pstat,&flag);
       if (flag==1){
         indices[count] = i;
@@ -1091,29 +1118,17 @@ MPI_Request Request::f2c(int id) {
   char key[KEY_SIZE];
   if(id==MPI_FORTRAN_REQUEST_NULL)
     return static_cast<MPI_Request>(MPI_REQUEST_NULL);
-  return static_cast<MPI_Request>(F2C::f2c_lookup()->at(get_key_id(key, id)));
-}
-
-int Request::add_f()
-{
-  if (F2C::f2c_lookup() == nullptr) {
-    F2C::set_f2c_lookup(new std::unordered_map<std::string, F2C*>);
-  }
-  char key[KEY_SIZE];
-  (*(F2C::f2c_lookup()))[get_key_id(key, F2C::f2c_id())] = this;
-  F2C::f2c_id_increment();
-  return F2C::f2c_id()-1;
+  return static_cast<MPI_Request>(F2C::f2c_lookup()->at(get_key(key,id)));
 }
 
 void Request::free_f(int id)
 {
   if (id != MPI_FORTRAN_REQUEST_NULL) {
     char key[KEY_SIZE];
-    F2C::f2c_lookup()->erase(get_key_id(key, id));
+    F2C::f2c_lookup()->erase(get_key(key, id));
   }
 }
 
-
 int Request::get_status(MPI_Request req, int* flag, MPI_Status * status){
   *flag=0;