X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/1d2a442708064da9020cf0574bc35ba0b6f712e6..5c471ea7cf2c1ea761f5896ecf97c72e8134ee30:/src/smpi/smpi_base.c

diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c
index 380f9dbe2e..7296b10763 100644
--- a/src/smpi/smpi_base.c
+++ b/src/smpi/smpi_base.c
@@ -20,10 +20,10 @@ void smpi_process_init(int* argc, char*** argv) {
 
   proc = SIMIX_process_self();
   index = atoi((*argv)[1]);
-  free((*argv)[1]);
   data = smpi_process_remote_data(index);
   SIMIX_process_set_data(proc, data);
   if (*argc > 2) {
+    free((*argv)[1]);
     memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
     (*argv)[(*argc) - 1] = NULL;
   }
@@ -48,8 +48,9 @@ MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst,
   request->tag = tag;
   request->size = smpi_datatype_size(datatype) * count;
   request->complete = 0;
+  request->data = request;
   smpi_process_post_send(comm, request);
-  request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, buf, request->size, request);
+  request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, buf, request->size, NULL);
   return request;
 }
 
@@ -63,6 +64,7 @@ MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src,
   request->tag = tag;
   request->size = smpi_datatype_size(datatype) * count;
   request->complete = 0;
+  request->data = MPI_REQUEST_NULL;
   smpi_process_post_recv(request);
   request->pair = SIMIX_network_irecv(request->rdv, buf, &request->size);
   return request;
@@ -96,13 +98,15 @@ void smpi_mpi_sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype, int
 }
 
 static void finish_wait(MPI_Request* request, MPI_Status* status) {
-  MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
+  MPI_Request data = (*request)->data;
 
   xbt_assert0(data != MPI_REQUEST_NULL, "Erroneous situation");
   if(status != MPI_STATUS_IGNORE) {
     status->MPI_SOURCE = (*request)->src;
     status->MPI_TAG = (*request)->tag;
     status->MPI_ERROR = MPI_SUCCESS;
+    status->_count = (*request)->size; // size in bytes
+    status->_cancelled = 0;            // FIXME: cancellation of requests not handled yet
   }
   DEBUG3("finishing wait for %p [data = %p, complete = %d]", *request, data, data->complete);
   // data == *request if sender is first to finish its wait
@@ -120,16 +124,18 @@ static void finish_wait(MPI_Request* request, MPI_Status* status) {
       // receiver cleans everything
       xbt_free(data);
     }
+    SIMIX_rdv_destroy((*request)->rdv);
     xbt_free(*request);
   }
   *request = MPI_REQUEST_NULL;
 }
 
 int smpi_mpi_test(MPI_Request* request, MPI_Status* status) {
-  MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
+  MPI_Request data = (*request)->data;
   int flag = data && data->complete == 1;
 
   if(flag) {
+    SIMIX_communication_destroy((*request)->pair);
     finish_wait(request, status);
   }
   return flag;
@@ -143,8 +149,9 @@ int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status*
   flag = 0;
   for(i = 0; i < count; i++) {
     if(requests[i] != MPI_REQUEST_NULL) {
-      data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
+      data = requests[i]->data;
       if(data != MPI_REQUEST_NULL && data->complete == 1) {
+        SIMIX_communication_destroy(requests[i]->pair);
         finish_wait(&requests[i], status);
         *index = i;
         flag = 1;
@@ -155,14 +162,25 @@ int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status*
   return flag;
 }
 
+
+void smpi_mpi_get_count(MPI_Status *status, MPI_Datatype datatype, int *count) {
+ int size = smpi_datatype_size(datatype);
+	  *count = (int)(status->_count / size);
+	  if ( (int)((*count) * size) != status->_count )
+		    *count = MPI_UNDEFINED;
+}
+
+
 void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) {
-  MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
+  MPI_Request data = (*request)->data;
 
   DEBUG6("wait for request %p (%p: %p) [src = %d, dst = %d, tag = %d]",
          *request, (*request)->pair, data, (*request)->src, (*request)->dst, (*request)->tag);
   // data is null if receiver waits before sender enters the rdv
   if(data == MPI_REQUEST_NULL || data->complete == 0) {
     SIMIX_network_wait((*request)->pair, -1.0);
+  } else {
+    SIMIX_communication_destroy((*request)->pair);
   }
   finish_wait(request, status);
 }
@@ -178,9 +196,10 @@ int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) {
     // First check for already completed requests
     for(i = 0; i < count; i++) {
       if(requests[i] != MPI_REQUEST_NULL) {
-        data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
+        data = requests[i]->data;
         if(data != MPI_REQUEST_NULL && data->complete == 1) {
           index = i;
+          SIMIX_communication_destroy(requests[index]->pair); // always succeeds (but cleans the simix layer)
           break;
         }
       }
@@ -238,10 +257,11 @@ int smpi_mpi_waitsome(int incount, MPI_Request requests[], int* indices, MPI_Sta
   int i, count;
 
   count = 0;
-  for(i = 0; i < count; i++) {
+  for(i = 0; i < incount; i++) {
     if(requests[i] != MPI_REQUEST_NULL) {
-      data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
+      data = requests[i]->data;
       if(data != MPI_REQUEST_NULL && data->complete == 1) {
+        SIMIX_communication_destroy(requests[i]->pair);
         finish_wait(&requests[i], status != MPI_STATUS_IGNORE ? &status[i] : MPI_STATUS_IGNORE);
         indices[count] = i;
         count++;
@@ -439,7 +459,7 @@ void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datat
   int rank, size, src, index, datasize;
   MPI_Request* requests;
   void** tmpbufs;
- 
+
   rank = smpi_comm_rank(comm);
   size = smpi_comm_size(comm);
   if(rank != root) {
@@ -448,7 +468,7 @@ void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datat
   } else {
     datasize = smpi_datatype_size(datatype);
     // Local copy from root
-    memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); 
+    memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
     // Receive buffers from senders
     //TODO: make a MPI_barrier here ?
     requests = xbt_new(MPI_Request, size - 1);
@@ -488,12 +508,12 @@ FIXME: buggy implementation
   int rank, size, other, index, datasize;
   MPI_Request* requests;
   void** tmpbufs;
- 
+
   rank = smpi_comm_rank(comm);
   size = smpi_comm_size(comm);
   datasize = smpi_datatype_size(datatype);
   // Local copy from self
-  memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); 
+  memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
   // Send/Recv buffers to/from others;
   //TODO: make a MPI_barrier here ?
   requests = xbt_new(MPI_Request, 2 * (size - 1));
@@ -525,3 +545,47 @@ FIXME: buggy implementation
   xbt_free(requests);
 */
 }
+
+void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
+  int system_tag = 666;
+  int rank, size, other, index, datasize;
+  int total;
+  MPI_Request* requests;
+  void** tmpbufs;
+
+  rank = smpi_comm_rank(comm);
+  size = smpi_comm_size(comm);
+  datasize = smpi_datatype_size(datatype);
+  // Local copy from self
+  memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
+  // Send/Recv buffers to/from others;
+  total = rank + (size - (rank + 1));
+  requests = xbt_new(MPI_Request, total);
+  tmpbufs = xbt_new(void*, rank);
+  index = 0;
+  for(other = 0; other < rank; other++) {
+    tmpbufs[index] = xbt_malloc(count * datasize);
+    requests[index] = smpi_mpi_irecv(tmpbufs[index], count, datatype, other, system_tag, comm);
+    index++;
+  }
+  for(other = rank + 1; other < size; other++) {
+    requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm);
+    index++;
+  }
+  // Wait for completion of all comms.
+  for(other = 0; other < total; other++) {
+    index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
+    if(index == MPI_UNDEFINED) {
+      break;
+    }
+    if(index < rank) {
+      // #Request is below rank: it's a irecv
+      smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
+    }
+  }
+  for(index = 0; index < size - 1; index++) {
+    xbt_free(tmpbufs[index]);
+  }
+  xbt_free(tmpbufs);
+  xbt_free(requests);
+}