various cleanups in actors/processes

[simgrid.git] / src / smpi / smpi_request.cpp

diff --git a/src/smpi/smpi_request.cpp b/src/smpi/smpi_request.cpp
index 27d5142..0e8c01b 100644 (file)
--- a/src/smpi/smpi_request.cpp
+++ b/src/smpi/smpi_request.cpp
@@ -117,7 +117,8 @@ void Request::unref(MPI_Request* request)
   }
 }
 
-int Request::match_recv(void* a, void* b, smx_activity_t ignored) {
+int Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl* ignored)
+{
   MPI_Request ref = static_cast<MPI_Request>(a);
   MPI_Request req = static_cast<MPI_Request>(b);
   XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
@@ -140,7 +141,8 @@ int Request::match_recv(void* a, void* b, smx_activity_t ignored) {
   }else return 0;
 }
 
-int Request::match_send(void* a, void* b,smx_activity_t ignored) {
+int Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl* ignored)
+{
   MPI_Request ref = static_cast<MPI_Request>(a);
   MPI_Request req = static_cast<MPI_Request>(b);
   XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_);
@@ -530,7 +532,7 @@ int Request::testsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
         count++;
         if (status != MPI_STATUSES_IGNORE)
           status[i] = *pstat;
-        if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->flags_ & NON_PERSISTENT)
+        if ((requests[i] != MPI_REQUEST_NULL) && (requests[i]->flags_ & NON_PERSISTENT))
           requests[i] = MPI_REQUEST_NULL;
       }
     } else {
@@ -544,7 +546,7 @@ int Request::testsome(int incount, MPI_Request requests[], int *indices, MPI_Sta
 
 int Request::testany(int count, MPI_Request requests[], int *index, MPI_Status * status)
 {
-  std::vector<simgrid::kernel::activity::ActivityImpl*> comms;
+  std::vector<simgrid::kernel::activity::ActivityImplPtr> comms;
   comms.reserve(count);
 
   int i;
@@ -652,8 +654,8 @@ void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status*
   }
 
   if (request->action_ != nullptr){
-    simgrid::kernel::activity::CommImpl* sync_comm =
-        static_cast<simgrid::kernel::activity::CommImpl*>(request->action_);
+    simgrid::kernel::activity::CommImplPtr sync_comm =
+        boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(request->action_);
     MPI_Request req                            = static_cast<MPI_Request>(sync_comm->src_data);
     *flag = 1;
     if(status != MPI_STATUS_IGNORE && (req->flags_ & PREPARED) == 0) {
@@ -757,22 +759,23 @@ void Request::wait(MPI_Request * request, MPI_Status * status)
 int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
 {
   s_xbt_dynar_t comms; // Keep it on stack to save some extra mallocs
-  int i;
-  int size = 0;
   int index = MPI_UNDEFINED;
-  int *map;
 
   if(count > 0) {
+    int size = 0;
     // Wait for a request to complete
-    xbt_dynar_init(&comms, sizeof(smx_activity_t), nullptr);
-    map = xbt_new(int, count);
+    xbt_dynar_init(&comms, sizeof(smx_activity_t), [](void*ptr){
+      intrusive_ptr_release(*(simgrid::kernel::activity::ActivityImpl**)ptr);
+    });
+    int *map = xbt_new(int, count);
     XBT_DEBUG("Wait for one of %d", count);
-    for(i = 0; i < count; i++) {
+    for(int i = 0; i < count; i++) {
       if (requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & PREPARED) &&
           not(requests[i]->flags_ & FINISHED)) {
         if (requests[i]->action_ != nullptr) {
           XBT_DEBUG("Waiting any %p ", requests[i]);
-          xbt_dynar_push(&comms, &requests[i]->action_);
+          intrusive_ptr_add_ref(requests[i]->action_.get());
+          xbt_dynar_push_as(&comms, simgrid::kernel::activity::ActivityImpl*, requests[i]->action_.get());
           map[size] = i;
           size++;
         } else {
@@ -786,8 +789,9 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status)
         }
       }
     }
-    if(size > 0) {
-      i = simcall_comm_waitany(&comms, -1);
+    if (size > 0) {
+      XBT_DEBUG("Enter waitany for %lu comms", xbt_dynar_length(&comms));
+      int i = simcall_comm_waitany(&comms, -1);
 
       // not MPI_UNDEFINED, as this is a simix return code
       if (i != -1) {
@@ -822,7 +826,7 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
   std::vector<MPI_Request> accumulates;
   int index;
   MPI_Status stat;
-  MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
+  MPI_Status *pstat = (status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat);
   int retvalue = MPI_SUCCESS;
   //tag invalid requests in the set
   if (status != MPI_STATUSES_IGNORE) {
@@ -840,7 +844,7 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
       wait(&requests[c],pstat);
       index = c;
     } else {
-      index = waitany(count, requests, pstat);
+      index = waitany(count, (MPI_Request*)requests, pstat);
       if (index == MPI_UNDEFINED)
         break;
 
@@ -870,15 +874,12 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[])
 
 int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[])
 {
-  int i;
   int count = 0;
-  int index;
   MPI_Status stat;
   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 
-  for(i = 0; i < incount; i++)
-  {
-    index=waitany(incount, requests, pstat);
+  for (int i = 0; i < incount; i++) {
+    int index = waitany(incount, requests, pstat);
     if(index!=MPI_UNDEFINED){
       indices[count] = index;
       count++;
@@ -912,9 +913,10 @@ int Request::add_f() {
 }
 
 void Request::free_f(int id) {
-  char key[KEY_SIZE];
-  if(id!=MPI_FORTRAN_REQUEST_NULL)
+  if (id != MPI_FORTRAN_REQUEST_NULL) {
+    char key[KEY_SIZE];
     xbt_dict_remove(F2C::f2c_lookup(), get_key_id(key, id));
+  }
 }
 
 }