X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/99bdf0f53ee37b983b0e1e0c8e060ccb9ad5198d..dc1369e2055686dde3d178efb2001dbad8972d54:/src/smpi/mpi/smpi_file.cpp?ds=sidebyside

diff --git a/src/smpi/mpi/smpi_file.cpp b/src/smpi/mpi/smpi_file.cpp
index 6006b4d848..46937ceed6 100644
--- a/src/smpi/mpi/smpi_file.cpp
+++ b/src/smpi/mpi/smpi_file.cpp
@@ -8,18 +8,42 @@
 #include "smpi_coll.hpp"
 #include "smpi_datatype.hpp"
 #include "smpi_info.hpp"
+#include "smpi_win.hpp"
+#include "smpi_request.hpp"
+
+//setup here, because we have templates in smpi_file we want to log
+XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_io, smpi, "Logging specific to SMPI (RMA operations)");
+
 #include "smpi_file.hpp"
+#include "smpi_status.hpp"
 #include "simgrid/plugins/file_system.h"
 
-XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_io, smpi, "Logging specific to SMPI (RMA operations)");
+#define FP_SIZE sizeof(MPI_Offset)
 
 
 namespace simgrid{
 namespace smpi{
-  File::File(MPI_Comm comm, char *filename, int amode, MPI_Info info): comm_(comm), flags_(amode), info_(info){
+
+  File::File(MPI_Comm comm, char *filename, int amode, MPI_Info info): comm_(comm), flags_(amode), info_(info) {
     file_= new simgrid::s4u::File(filename, nullptr);
+    list_=nullptr;
+    if (comm_->rank() == 0) {
+      int size= comm_->size() + FP_SIZE;
+      list_ = new char[size];
+      memset(list_, 0, size);
+      shared_file_pointer_ = new MPI_Offset[1];
+      shared_mutex_ = s4u::Mutex::create();
+      *shared_file_pointer_ = 0;
+      win_=new Win(list_, size, 1, MPI_INFO_NULL, comm_);
+    }else{
+      win_=new Win(list_, 0, 1, MPI_INFO_NULL, comm_);
+    }
+    simgrid::smpi::Colls::bcast(&shared_file_pointer_, 1, MPI_AINT, 0, comm);
+    simgrid::smpi::Colls::bcast(&shared_mutex_, 1, MPI_AINT, 0, comm);
+    if(comm_->rank() != 0)
+      intrusive_ptr_add_ref(&*shared_mutex_);
   }
-  
+
   File::~File(){
     delete file_;
   }
@@ -45,18 +69,25 @@ namespace smpi{
     return MPI_SUCCESS;
   }
 
+  int File::get_position_shared(MPI_Offset* offset){
+    shared_mutex_->lock();
+    *offset=*shared_file_pointer_;
+    shared_mutex_->unlock();
+    return MPI_SUCCESS;
+  }
+
   int File::seek(MPI_Offset offset, int whence){
     switch(whence){
       case(MPI_SEEK_SET):
-        XBT_DEBUG("Seeking in MPI_File %s, setting offset %lld", file_->get_path(), offset);
+        XBT_VERB("Seeking in MPI_File %s, setting offset %lld", file_->get_path(), offset);
         file_->seek(offset,SEEK_SET);
         break;
       case(MPI_SEEK_CUR):
-        XBT_DEBUG("Seeking in MPI_File %s, current offset + %lld", file_->get_path(), offset);
+        XBT_VERB("Seeking in MPI_File %s, current offset + %lld", file_->get_path(), offset);
         file_->seek(offset,SEEK_CUR);
         break;
       case(MPI_SEEK_END):
-        XBT_DEBUG("Seeking in MPI_File %s, end offset + %lld", file_->get_path(), offset);
+        XBT_VERB("Seeking in MPI_File %s, end offset + %lld", file_->get_path(), offset);
         file_->seek(offset,SEEK_END);
         break;
       default:
@@ -64,7 +95,15 @@ namespace smpi{
     }
     return MPI_SUCCESS;
   }
-  
+
+  int File::seek_shared(MPI_Offset offset, int whence){
+    shared_mutex_->lock();
+    seek(offset,whence);
+    *shared_file_pointer_=offset;
+    shared_mutex_->unlock();
+    return MPI_SUCCESS;
+  }
+
   int File::read(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
     //get position first as we may be doing non contiguous reads and it will probably be updated badly
     MPI_Offset position = fh->file_->tell();
@@ -72,14 +111,56 @@ namespace smpi{
     MPI_Offset readsize = datatype->size()*count;
     XBT_DEBUG("Position before read in MPI_File %s : %llu",fh->file_->get_path(),fh->file_->tell());
     MPI_Offset read = fh->file_->read(readsize);
-    XBT_DEBUG("Read in MPI_File %s, %lld bytes read, readsize %lld bytes, movesize %lld", fh->file_->get_path(), read, readsize, movesize);
+    XBT_VERB("Read in MPI_File %s, %lld bytes read, readsize %lld bytes, movesize %lld", fh->file_->get_path(), read, readsize, movesize);
     if(readsize!=movesize){
       fh->file_->seek(position+movesize, SEEK_SET);
     }
-    XBT_DEBUG("Position after read in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());
+    XBT_VERB("Position after read in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());
+    status->count=count*datatype->size();
     return MPI_SUCCESS;
   }
 
+  /*Ordered and Shared Versions, with RMA-based locks : Based on the model described in :*/
+  /* @InProceedings{10.1007/11557265_15,*/
+  /* author="Latham, Robert and Ross, Robert and Thakur, Rajeev and Toonen, Brian",*/ 
+  /* title="Implementing MPI-IO Shared File Pointers Without File System Support",*/
+  /* booktitle="Recent Advances in Parallel Virtual Machine and Message Passing Interface",*/
+  /* year="2005",*/
+  /* publisher="Springer Berlin Heidelberg",*/
+  /* address="Berlin, Heidelberg",*/
+  /* pages="84--93"*/
+  /* }*/
+  int File::read_shared(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
+    fh->shared_mutex_->lock();
+    fh->seek(*(fh->shared_file_pointer_),MPI_SEEK_SET);
+    read(fh, buf, count, datatype, status);
+    *(fh->shared_file_pointer_)=fh->file_->tell();
+    fh->shared_mutex_->unlock();
+    return MPI_SUCCESS;
+  }
+
+  int File::read_ordered(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
+    //0 needs to get the shared pointer value
+    MPI_Offset val;
+    if(fh->comm_->rank()==0){
+      val=*(fh->shared_file_pointer_);
+    }else{
+      val=count*datatype->size();
+    }
+
+    MPI_Offset result;
+    simgrid::smpi::Colls::scan(&val, &result, 1, MPI_OFFSET, MPI_SUM, fh->comm_);
+    fh->seek(result, MPI_SEEK_SET);
+    int ret = fh->op_all<simgrid::smpi::File::read>(buf, count, datatype, status);
+    if(fh->comm_->rank()==fh->comm_->size()-1){
+      fh->shared_mutex_->lock();
+      *(fh->shared_file_pointer_)=fh->file_->tell();
+      fh->shared_mutex_->unlock();
+    }
+    char c;
+    simgrid::smpi::Colls::bcast(&c, 1, MPI_BYTE, fh->comm_->size()-1, fh->comm_);
+    return ret;
+  }
 
   int File::write(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
     //get position first as we may be doing non contiguous reads and it will probably be updated badly
@@ -88,14 +169,46 @@ namespace smpi{
     MPI_Offset writesize = datatype->size()*count;
     XBT_DEBUG("Position before write in MPI_File %s : %llu",fh->file_->get_path(),fh->file_->tell());
     MPI_Offset write = fh->file_->write(writesize);
-    XBT_DEBUG("Write in MPI_File %s, %lld bytes read, readsize %lld bytes, movesize %lld", fh->file_->get_path(), write, writesize, movesize);
+    XBT_VERB("Write in MPI_File %s, %lld bytes written, readsize %lld bytes, movesize %lld", fh->file_->get_path(), write, writesize, movesize);
     if(writesize!=movesize){
       fh->file_->seek(position+movesize, SEEK_SET);
     }
-    XBT_DEBUG("Position after write in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());
+    XBT_VERB("Position after write in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());
+    status->count=count*datatype->size();
+    return MPI_SUCCESS;
+  }
+
+  int File::write_shared(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
+    fh->shared_mutex_->lock();
+    fh->seek(*(fh->shared_file_pointer_),MPI_SEEK_SET);
+    write(fh, buf, count, datatype, status);
+    *(fh->shared_file_pointer_)=fh->file_->tell();
+    fh->shared_mutex_->unlock();
     return MPI_SUCCESS;
   }
 
+  int File::write_ordered(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){
+    //0 needs to get the shared pointer value
+    MPI_Offset val;
+    if(fh->comm_->rank()==0){
+      val=*(fh->shared_file_pointer_);
+    }else{
+      val=count*datatype->size();
+    }
+    MPI_Offset result;
+    simgrid::smpi::Colls::scan(&val, &result, 1, MPI_OFFSET, MPI_SUM, fh->comm_);
+    fh->seek(result, MPI_SEEK_SET);
+    int ret = fh->op_all<simgrid::smpi::File::write>(buf, count, datatype, status);
+    if(fh->comm_->rank()==fh->comm_->size()-1){
+      fh->shared_mutex_->lock();
+      *(fh->shared_file_pointer_)=fh->file_->tell();
+      fh->shared_mutex_->unlock();
+    }
+    char c;
+    simgrid::smpi::Colls::bcast(&c, 1, MPI_BYTE, fh->comm_->size()-1, fh->comm_);
+    return ret;
+  }
+
   int File::size(){
     return file_->size();
   }
@@ -108,6 +221,7 @@ namespace smpi{
     //no idea
     return simgrid::smpi::Colls::barrier(comm_);
   }
+
 MPI_Info File::info(){
   if(info_== MPI_INFO_NULL)
     info_ = new Info();