src/smpi/mpi/smpi_file.cpp

   1 /* Copyright (c) 2007-2019. The SimGrid Team. All rights reserved.          */\r
   2 \r
   3 /* This program is free software; you can redistribute it and/or modify it\r
   4  * under the terms of the license (GNU LGPL) which comes with this package. */\r
   5 #include "private.hpp"\r
   6 \r
   7 #include "smpi_comm.hpp"\r
   8 #include "smpi_coll.hpp"\r
   9 #include "smpi_datatype.hpp"\r
  10 #include "smpi_info.hpp"\r
  11 #include "smpi_win.hpp"\r
  12 #include "smpi_request.hpp"\r
  13 \r
  14 //setup here, because we have templates in smpi_file we want to log\r
  15 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_io, smpi, "Logging specific to SMPI (RMA operations)");\r
  16 \r
  17 #include "smpi_file.hpp"\r
  18 #include "smpi_status.hpp"\r
  19 #include "simgrid/plugins/file_system.h"\r
  20 \r
  21 #define FP_SIZE sizeof(MPI_Offset)\r
  22 \r
  23 \r
  24 namespace simgrid{\r
  25 namespace smpi{\r
  26 \r
  27   File::File(MPI_Comm comm, char *filename, int amode, MPI_Info info): comm_(comm), flags_(amode), info_(info) {\r
  28     file_= new simgrid::s4u::File(filename, nullptr);\r
  29     list_=nullptr;\r
  30     if (comm_->rank() == 0) {\r
  31       int size= comm_->size() + FP_SIZE;\r
  32       list_ = new char[size];\r
  33       memset(list_, 0, size);\r
  34       shared_file_pointer_ = new MPI_Offset[1];\r
  35       shared_mutex_ = s4u::Mutex::create();\r
  36       *shared_file_pointer_ = 0;\r
  37       win_=new Win(list_, size, 1, MPI_INFO_NULL, comm_);\r
  38     }else{\r
  39       win_=new Win(list_, 0, 1, MPI_INFO_NULL, comm_);\r
  40     }\r
  41     simgrid::smpi::Colls::bcast(&shared_file_pointer_, 1, MPI_AINT, 0, comm);\r
  42     simgrid::smpi::Colls::bcast(&shared_mutex_, 1, MPI_AINT, 0, comm);\r
  43     if(comm_->rank() != 0)\r
  44       intrusive_ptr_add_ref(&*shared_mutex_);\r
  45   }\r
  46 \r
  47   File::~File(){\r
  48     delete file_;\r
  49   }\r
  50 \r
  51   int File::close(MPI_File *fh){\r
  52     XBT_DEBUG("Closing MPI_File %s", (*fh)->file_->get_path());\r
  53     (*fh)->sync();\r
  54     if((*fh)->flags() & MPI_MODE_DELETE_ON_CLOSE)\r
  55       (*fh)->file_->unlink();\r
  56     delete (*fh);\r
  57     return MPI_SUCCESS;\r
  58   }\r
  59 \r
  60   int File::del(char *filename, MPI_Info info){\r
  61     //get the file with MPI_MODE_DELETE_ON_CLOSE and then close it\r
  62     File* f = new File(MPI_COMM_SELF,filename,MPI_MODE_DELETE_ON_CLOSE|MPI_MODE_RDWR, nullptr);\r
  63     close(&f);\r
  64     return MPI_SUCCESS;\r
  65   }\r
  66 \r
  67   int File::get_position(MPI_Offset* offset){\r
  68     *offset=file_->tell();\r
  69     return MPI_SUCCESS;\r
  70   }\r
  71 \r
  72   int File::get_position_shared(MPI_Offset* offset){\r
  73     shared_mutex_->lock();\r
  74     *offset=*shared_file_pointer_;\r
  75     shared_mutex_->unlock();\r
  76     return MPI_SUCCESS;\r
  77   }\r
  78 \r
  79   int File::seek(MPI_Offset offset, int whence){\r
  80     switch(whence){\r
  81       case(MPI_SEEK_SET):\r
  82         XBT_VERB("Seeking in MPI_File %s, setting offset %lld", file_->get_path(), offset);\r
  83         file_->seek(offset,SEEK_SET);\r
  84         break;\r
  85       case(MPI_SEEK_CUR):\r
  86         XBT_VERB("Seeking in MPI_File %s, current offset + %lld", file_->get_path(), offset);\r
  87         file_->seek(offset,SEEK_CUR);\r
  88         break;\r
  89       case(MPI_SEEK_END):\r
  90         XBT_VERB("Seeking in MPI_File %s, end offset + %lld", file_->get_path(), offset);\r
  91         file_->seek(offset,SEEK_END);\r
  92         break;\r
  93       default:\r
  94         return MPI_ERR_FILE;\r
  95     }\r
  96     return MPI_SUCCESS;\r
  97   }\r
  98 \r
  99   int File::seek_shared(MPI_Offset offset, int whence){\r
 100     shared_mutex_->lock();\r
 101     seek(offset,whence);\r
 102     *shared_file_pointer_=offset;\r
 103     shared_mutex_->unlock();\r
 104     return MPI_SUCCESS;\r
 105   }\r
 106 \r
 107   int File::read(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 108     //get position first as we may be doing non contiguous reads and it will probably be updated badly\r
 109     MPI_Offset position = fh->file_->tell();\r
 110     MPI_Offset movesize = datatype->get_extent()*count;\r
 111     MPI_Offset readsize = datatype->size()*count;\r
 112     XBT_DEBUG("Position before read in MPI_File %s : %llu",fh->file_->get_path(),fh->file_->tell());\r
 113     MPI_Offset read = fh->file_->read(readsize);\r
 114     XBT_VERB("Read in MPI_File %s, %lld bytes read, readsize %lld bytes, movesize %lld", fh->file_->get_path(), read, readsize, movesize);\r
 115     if(readsize!=movesize){\r
 116       fh->file_->seek(position+movesize, SEEK_SET);\r
 117     }\r
 118     XBT_VERB("Position after read in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());\r
 119     status->count=count*datatype->size();\r
 120     return MPI_SUCCESS;\r
 121   }\r
 122 \r
 123   /*Ordered and Shared Versions, with RMA-based locks : Based on the model described in :*/\r
 124   /* @InProceedings{10.1007/11557265_15,*/\r
 125   /* author="Latham, Robert and Ross, Robert and Thakur, Rajeev and Toonen, Brian",*/ \r
 126   /* title="Implementing MPI-IO Shared File Pointers Without File System Support",*/\r
 127   /* booktitle="Recent Advances in Parallel Virtual Machine and Message Passing Interface",*/\r
 128   /* year="2005",*/\r
 129   /* publisher="Springer Berlin Heidelberg",*/\r
 130   /* address="Berlin, Heidelberg",*/\r
 131   /* pages="84--93"*/\r
 132   /* }*/\r
 133   int File::read_shared(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 134     fh->shared_mutex_->lock();\r
 135     fh->seek(*(fh->shared_file_pointer_),MPI_SEEK_SET);\r
 136     read(fh, buf, count, datatype, status);\r
 137     *(fh->shared_file_pointer_)=fh->file_->tell();\r
 138     fh->shared_mutex_->unlock();\r
 139     return MPI_SUCCESS;\r
 140   }\r
 141 \r
 142   int File::read_ordered(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 143     //0 needs to get the shared pointer value\r
 144     MPI_Offset val;\r
 145     if(fh->comm_->rank()==0){\r
 146       val=*(fh->shared_file_pointer_);\r
 147     }else{\r
 148       val=count*datatype->size();\r
 149     }\r
 150 \r
 151     MPI_Offset result;\r
 152     simgrid::smpi::Colls::scan(&val, &result, 1, MPI_OFFSET, MPI_SUM, fh->comm_);\r
 153     fh->seek(result, MPI_SEEK_SET);\r
 154     int ret = fh->op_all<simgrid::smpi::File::read>(buf, count, datatype, status);\r
 155     if(fh->comm_->rank()==fh->comm_->size()-1){\r
 156       fh->shared_mutex_->lock();\r
 157       *(fh->shared_file_pointer_)=fh->file_->tell();\r
 158       fh->shared_mutex_->unlock();\r
 159     }\r
 160     char c;\r
 161     simgrid::smpi::Colls::bcast(&c, 1, MPI_BYTE, fh->comm_->size()-1, fh->comm_);\r
 162     return ret;\r
 163   }\r
 164 \r
 165   int File::write(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 166     //get position first as we may be doing non contiguous reads and it will probably be updated badly\r
 167     MPI_Offset position = fh->file_->tell();\r
 168     MPI_Offset movesize = datatype->get_extent()*count;\r
 169     MPI_Offset writesize = datatype->size()*count;\r
 170     XBT_DEBUG("Position before write in MPI_File %s : %llu",fh->file_->get_path(),fh->file_->tell());\r
 171     MPI_Offset write = fh->file_->write(writesize);\r
 172     XBT_VERB("Write in MPI_File %s, %lld bytes written, readsize %lld bytes, movesize %lld", fh->file_->get_path(), write, writesize, movesize);\r
 173     if(writesize!=movesize){\r
 174       fh->file_->seek(position+movesize, SEEK_SET);\r
 175     }\r
 176     XBT_VERB("Position after write in MPI_File %s : %llu",fh->file_->get_path(), fh->file_->tell());\r
 177     status->count=count*datatype->size();\r
 178     return MPI_SUCCESS;\r
 179   }\r
 180 \r
 181   int File::write_shared(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 182     fh->shared_mutex_->lock();\r
 183     fh->seek(*(fh->shared_file_pointer_),MPI_SEEK_SET);\r
 184     write(fh, buf, count, datatype, status);\r
 185     *(fh->shared_file_pointer_)=fh->file_->tell();\r
 186     fh->shared_mutex_->unlock();\r
 187     return MPI_SUCCESS;\r
 188   }\r
 189 \r
 190   int File::write_ordered(MPI_File fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status){\r
 191     //0 needs to get the shared pointer value\r
 192     MPI_Offset val;\r
 193     if(fh->comm_->rank()==0){\r
 194       val=*(fh->shared_file_pointer_);\r
 195     }else{\r
 196       val=count*datatype->size();\r
 197     }\r
 198     MPI_Offset result;\r
 199     simgrid::smpi::Colls::scan(&val, &result, 1, MPI_OFFSET, MPI_SUM, fh->comm_);\r
 200     fh->seek(result, MPI_SEEK_SET);\r
 201     int ret = fh->op_all<simgrid::smpi::File::write>(buf, count, datatype, status);\r
 202     if(fh->comm_->rank()==fh->comm_->size()-1){\r
 203       fh->shared_mutex_->lock();\r
 204       *(fh->shared_file_pointer_)=fh->file_->tell();\r
 205       fh->shared_mutex_->unlock();\r
 206     }\r
 207     char c;\r
 208     simgrid::smpi::Colls::bcast(&c, 1, MPI_BYTE, fh->comm_->size()-1, fh->comm_);\r
 209     return ret;\r
 210   }\r
 211 \r
 212   int File::size(){\r
 213     return file_->size();\r
 214   }\r
 215 \r
 216   int File::flags(){\r
 217     return flags_;\r
 218   }\r
 219 \r
 220   int File::sync(){\r
 221     //no idea\r
 222     return simgrid::smpi::Colls::barrier(comm_);\r
 223   }\r
 224 \r
 225 MPI_Info File::info(){\r
 226   if(info_== MPI_INFO_NULL)\r
 227     info_ = new Info();\r
 228   info_->ref();\r
 229   return info_;\r
 230 }\r
 231 \r
 232 void File::set_info(MPI_Info info){\r
 233   if(info_!= MPI_INFO_NULL)\r
 234     info->ref();\r
 235   info_=info;\r
 236 }\r
 237 \r
 238 }\r
 239 }\r