src/ParallelContainer/Parallel_Salome_file_i.cxx

   1 // Copyright (C) 2007  OPEN CASCADE, CEA/DEN, EDF R&D, PRINCIPIA R&D
   2 //
   3 //  This library is free software; you can redistribute it and/or
   4 //  modify it under the terms of the GNU Lesser General Public
   5 //  License as published by the Free Software Foundation; either
   6 //  version 2.1 of the License.
   7 //
   8 //  This library is distributed in the hope that it will be useful,
   9 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 //  Lesser General Public License for more details.
  12 //
  13 //  You should have received a copy of the GNU Lesser General Public
  14 //  License along with this library; if not, write to the Free Software
  15 //  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 //
  17 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 //
  19 //
  20 //
  21 //  File   : Parallel_Salome_file_i.cxx
  22 //  Author : André RIBES, EDF
  23 //  Module : SALOME
  24 //  $Header:
  25
  26 #include "Parallel_Salome_file_i.hxx"
  27 #include "utilities.h"
  28
  29 Parallel_Salome_file_i::Parallel_Salome_file_i(CORBA::ORB_ptr orb, const char * ior) :
  30   InterfaceParallel_impl(orb,ior),
  31   Engines::Salome_file_serv(orb,ior),
  32   Engines::fileTransfer_serv(orb,ior),
  33   Engines::Parallel_Salome_file_serv(orb,ior)
  34 {
  35   CORBA::Object_ptr obj = _orb->string_to_object(ior);
  36   proxy = Engines::Parallel_Salome_file::_narrow(obj);
  37   parallel_file = NULL;
  38 }
  39
  40 Parallel_Salome_file_i::~Parallel_Salome_file_i() {}
  41
  42 void
  43 Parallel_Salome_file_i::load(const char* hdf5_file) {
  44   MESSAGE("Parallel_Salome_file_i::load : NOT YET IMPLEMENTED");
  45   SALOME::ExceptionStruct es;
  46   es.type = SALOME::INTERNAL_ERROR;
  47   es.text = "Parallel_Salome_file_i::load : NOT YET IMPLEMENTED";
  48   throw SALOME::SALOME_Exception(es);
  49 }
  50
  51 void
  52 Parallel_Salome_file_i::save(const char* hdf5_file) {
  53   MESSAGE("Parallel_Salome_file_i::save : NOT YET IMPLEMENTED");
  54   SALOME::ExceptionStruct es;
  55   es.type = SALOME::INTERNAL_ERROR;
  56   es.text = "Parallel_Salome_file_i::save : NOT YET IMPLEMENTED";
  57   throw SALOME::SALOME_Exception(es);
  58 }
  59
  60 void
  61 Parallel_Salome_file_i::save_all(const char* hdf5_file) {
  62   MESSAGE("Parallel_Salome_file_i::save_all : NOT YET IMPLEMENTED");
  63   SALOME::ExceptionStruct es;
  64   es.type = SALOME::INTERNAL_ERROR;
  65   es.text = "Parallel_Salome_file_i::save_all : NOT YET IMPLEMENTED";
  66   throw SALOME::SALOME_Exception(es);
  67 }
  68
  69 void
  70 Parallel_Salome_file_i::connect(Engines::Salome_file_ptr source_Salome_file) {
  71   // only one file managed case
  72   Salome_file_i::connect(source_Salome_file);
  73
  74   // Test if the file is managed in an another node
  75   // If yes, node is updated
  76   _t_fileManaged::iterator begin = _fileManaged.begin();
  77   _t_fileManaged::iterator end = _fileManaged.end();
  78   for(;begin!=end;begin++) {
  79     std::string file_name = begin->first;
  80     if (_fileManaged[file_name].node > 0 && getMyRank() == 0) {
  81       if (parallel_file == NULL)
  82         parallel_file = Engines::PaCO_Parallel_Salome_file::PaCO_narrow(proxy, _orb);
  83       parallel_file->connect(source_Salome_file, _fileManaged[file_name].node);
  84     }
  85   }
  86 }
  87
  88 void
  89 Parallel_Salome_file_i::connectDistributedFile(const char * file_name,
  90                                                Engines::Salome_file_ptr source_Salome_file) {
  91   Salome_file_i::connectDistributedFile(file_name, source_Salome_file);
  92
  93   // Test if the file is managed in an another node
  94   // If yes, node is updated
  95   std::string fname(file_name);
  96   if (_fileManaged[fname].node > 0 && getMyRank() == 0) {
  97     if (parallel_file == NULL)
  98       parallel_file = Engines::PaCO_Parallel_Salome_file::PaCO_narrow(proxy, _orb);
  99     parallel_file->connectDistributedFile(file_name, source_Salome_file, _fileManaged[fname].node);
 100   }
 101 }
 102
 103 void
 104 Parallel_Salome_file_i::setDistributedSourceFile(const char* file_name,
 105                                                  const char * source_file_name) {
 106   Salome_file_i::setDistributedSourceFile(file_name, source_file_name);
 107   // Test if the file is managed in an another node
 108   // If yes, node is updated
 109   std::string fname(file_name);
 110   if (_fileManaged[fname].node > 0 && getMyRank() == 0) {
 111     if (parallel_file == NULL)
 112       parallel_file = Engines::PaCO_Parallel_Salome_file::PaCO_narrow(proxy, _orb);
 113     parallel_file->setDistributedSourceFile(file_name, source_file_name, _fileManaged[fname].node);
 114   }
 115 }
 116
 117 void
 118 Parallel_Salome_file_i::recvFiles() {
 119   if (parallel_file == NULL)
 120     parallel_file = Engines::PaCO_Parallel_Salome_file::PaCO_narrow(proxy, _orb);
 121
 122   std::string files_not_ok("");
 123   int total = getTotalNode();
 124   for (int i =0; i<total; i++) {
 125     try {
 126      parallel_file->recvFiles_node(i);
 127     }
 128     catch (SALOME::SALOME_Exception & ex) {
 129       files_not_ok = files_not_ok + std::string(ex.details.text.in());
 130     }
 131   }
 132
 133   if (files_not_ok != "")
 134   {
 135     SALOME::ExceptionStruct es;
 136     es.type = SALOME::INTERNAL_ERROR;
 137     std::string text = "files not ready : " + files_not_ok;
 138     es.text = CORBA::string_dup(text.c_str());
 139     throw SALOME::SALOME_Exception(es);
 140   }
 141   else
 142   {
 143     // We change the state of the Salome_file
 144     _state.files_ok = true;
 145   }
 146 }
 147
 148 void
 149 Parallel_Salome_file_i::recvFiles_node() {
 150
 151   std::string files_not_ok("");
 152   _t_fileManaged::iterator begin = _fileManaged.begin();
 153   _t_fileManaged::iterator end = _fileManaged.end();
 154   for(;begin!=end;begin++)
 155   {
 156     bool result = true;
 157     Engines::file file_infos = begin->second;
 158     if (file_infos.node == getMyRank()) {
 159       // Test if the file is local or distributed
 160       if (std::string(file_infos.type.in()) == "local")
 161       {
 162         if (std::string(file_infos.status.in()) == "not_ok")
 163           result = checkLocalFile(file_infos.file_name.in());
 164       }
 165       else
 166       {
 167         if (std::string(file_infos.status.in()) == "not_ok") {
 168           // 2 cases :
 169           // Source file is a Salome_file
 170           // Source file is a Parallel_Salome_file
 171           PaCO::ParallelKernel_var interface_manager =
 172             PaCO::ParallelKernel::_narrow(_fileDistributedSource[file_infos.file_name.in()]);
 173           if (CORBA::is_nil(interface_manager))
 174             result = getDistributedFile(file_infos.file_name.in());
 175           else
 176             result = getParallelDistributedFile(file_infos.file_name.in());
 177         }
 178       }
 179       // if the result is false
 180       // we add this file to files_not_ok
 181       if (!result)
 182       {
 183         files_not_ok.append(" ");
 184         files_not_ok.append(file_infos.file_name.in());
 185       }
 186     }
 187   }
 188   if (files_not_ok != "")
 189   {
 190     SALOME::ExceptionStruct es;
 191     es.type = SALOME::INTERNAL_ERROR;
 192     std::string text = files_not_ok;
 193     es.text = CORBA::string_dup(text.c_str());
 194     throw SALOME::SALOME_Exception(es);
 195   }
 196 }
 197
 198 bool
 199 Parallel_Salome_file_i::getParallelDistributedFile(std::string file_name) {
 200
 201   bool result = true;
 202   const char * source_file_name = _fileManaged[file_name].source_file_name.in();
 203   int fileId;
 204   FILE* fp;
 205   std::string comp_file_name(_fileManaged[file_name].path.in());
 206   comp_file_name.append("/");
 207   comp_file_name.append(_fileManaged[file_name].file_name.in());
 208
 209   // Test if the process can write on disk
 210   if ((fp = fopen(comp_file_name.c_str(),"wb")) == NULL)
 211   {
 212     INFOS("file " << comp_file_name << " cannot be open for writing");
 213     _fileManaged[file_name].status = CORBA::string_dup("not_ok");
 214     result = false;
 215     return result;
 216   }
 217
 218   Engines::PaCO_Parallel_Salome_file * parallel_source_file =
 219     Engines::PaCO_Parallel_Salome_file::PaCO_narrow(_fileDistributedSource[file_name], _orb);
 220
 221   int node = parallel_source_file->getFileNode(source_file_name);
 222
 223   try
 224   {
 225     fileId = parallel_source_file->open(source_file_name, node);
 226   }
 227   catch (...)
 228   {
 229     _fileManaged[file_name].status = CORBA::string_dup("not_ok");
 230     fclose(fp);
 231     result = false;
 232     return result;
 233   }
 234
 235   if (fileId > 0)
 236   {
 237     Engines::fileBlock* aBlock;
 238     int toFollow = 1;
 239     int ctr=0;
 240     MESSAGE("begin of transfer of " << comp_file_name);
 241     while (toFollow)
 242     {
 243       ctr++;
 244       aBlock = parallel_source_file->getBlock(fileId, node);
 245       toFollow = aBlock->length();
 246       CORBA::Octet *buf = aBlock->get_buffer();
 247       int nbWri = fwrite(buf, sizeof(CORBA::Octet), toFollow, fp);
 248       ASSERT(nbWri == toFollow);
 249     }
 250     fclose(fp);
 251     MESSAGE("end of transfer of " << comp_file_name);
 252     parallel_source_file->close(fileId, node);
 253   }
 254   else
 255   {
 256     INFOS("open reference file for copy impossible");
 257     result = false;
 258     fclose(fp);
 259     _fileManaged[file_name].status = CORBA::string_dup("not_ok");
 260     return result;
 261   }
 262
 263   _fileManaged[file_name].status = CORBA::string_dup("ok");
 264   return result;
 265 }
 266
 267 void
 268 Parallel_Salome_file_i::setContainer(Engines::Container_ptr container) {
 269   _container = Engines::Container::_duplicate(container);
 270
 271   // Update All the files managed by the node
 272   _t_fileManaged::iterator begin = _fileManaged.begin();
 273   _t_fileManaged::iterator end = _fileManaged.end();
 274   for(;begin!=end;begin++) {
 275     begin->second.container = Engines::Container::_duplicate(container);
 276   }
 277 }
 278
 279 void
 280 Parallel_Salome_file_i::setFileNode(const char* file_name, CORBA::Long node) {
 281
 282   // Test if this file is managed
 283   std::string fname(file_name);
 284   _t_fileManaged::iterator it = _fileManaged.find(fname);
 285   if (it == _fileManaged.end())
 286   {
 287     SALOME::ExceptionStruct es;
 288     es.type = SALOME::INTERNAL_ERROR;
 289     es.text = "file is not managed";
 290     throw SALOME::SALOME_Exception(es);
 291   }
 292
 293   // Update file infos into this node (node 0)
 294   // and into the node that actually managed it
 295   _fileManaged[fname].node = node;
 296
 297   if (node > 0) {
 298     if (parallel_file == NULL)
 299       parallel_file = Engines::PaCO_Parallel_Salome_file::PaCO_narrow(proxy, _orb);
 300
 301     Engines::Container_ptr cont = parallel_file->updateFile(_fileManaged[fname], node);
 302     parallel_file->connectDistributedFile(fname.c_str(),
 303                                           _fileDistributedSource[fname],
 304                                           node);
 305
 306     // Update file infos with the new reference of the container
 307     _fileManaged[fname].container = Engines::Container::_duplicate(cont);
 308   }
 309 }
 310
 311 Engines::Container_ptr
 312 Parallel_Salome_file_i::updateFile(const Engines::file& file) {
 313   // Copy file
 314   Engines::file new_file_infos(file);
 315
 316   // Adding it to node list
 317   new_file_infos.container = Engines::Container::_duplicate(_container);
 318   std::string fname(new_file_infos.file_name.in());
 319   _fileManaged[fname] = new_file_infos;
 320
 321   // Return the new reference of the container associated to the file
 322   return Engines::Container::_duplicate(_container);
 323 }
 324
 325 CORBA::Long
 326 Parallel_Salome_file_i::getFileNode(const char* file_name) {
 327
 328   // Test if this file is managed
 329   std::string fname(file_name);
 330   if (fname == "") {
 331     // We enter in the simple case where the user
 332     // has not used setDistributedSourceFile.
 333     // In this case we try to see if the Salome_file
 334     if (_fileManaged.size() == 1)
 335     {
 336       // only one file managed
 337       _t_fileManaged::iterator it = _fileManaged.begin();
 338       fname = it->first;
 339     }
 340     else
 341     {
 342       SALOME::ExceptionStruct es;
 343       es.type = SALOME::INTERNAL_ERROR;
 344       es.text = "Error : there is more than one file that is managed";
 345       throw SALOME::SALOME_Exception(es);
 346     }
 347   }
 348   _t_fileManaged::iterator it = _fileManaged.find(fname);
 349   if (it == _fileManaged.end())
 350   {
 351     SALOME::ExceptionStruct es;
 352     es.type = SALOME::INTERNAL_ERROR;
 353     es.text = "file is not managed";
 354     throw SALOME::SALOME_Exception(es);
 355   }
 356
 357   return _fileManaged[fname].node;
 358 }