src/msg/msg_io.cpp

   1 /* Copyright (c) 2004-2017. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #include "simgrid/s4u/Host.hpp"
   7 #include "simgrid/s4u/Storage.hpp"
   8 #include "src/msg/msg_private.hpp"
   9 #include "src/plugins/file_system/FileSystem.hpp"
  10 #include <numeric>
  11
  12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_io, msg, "Logging specific to MSG (io)");
  13
  14 extern "C" {
  15
  16 /** @addtogroup msg_file
  17  * (#msg_file_t) and the functions for managing it.
  18  *
  19  *  \see #msg_file_t
  20  */
  21
  22 /** \ingroup msg_file
  23  * \brief Read a file (local or remote)
  24  *
  25  * \param size of the file to read
  26  * \param fd is a the file descriptor
  27  * \return the number of bytes successfully read or -1 if an error occurred
  28  */
  29 sg_size_t MSG_file_read(msg_file_t fd, sg_size_t size)
  30 {
  31   sg_size_t read_size;
  32
  33   if (fd->size() == 0) /* Nothing to read, return */
  34     return 0;
  35
  36   /* Find the host where the file is physically located and read it */
  37   msg_storage_t storage_src           = fd->localStorage;
  38   msg_host_t attached_host            = storage_src->getHost();
  39   read_size                           = fd->read(size);
  40
  41   if (strcmp(attached_host->getCname(), MSG_host_self()->getCname())) {
  42     /* the file is hosted on a remote host, initiate a communication between src and dest hosts for data transfer */
  43     XBT_DEBUG("File is on %s remote host, initiate data transfer of %llu bytes.", attached_host->getCname(), read_size);
  44     msg_host_t m_host_list[] = {MSG_host_self(), attached_host};
  45     double flops_amount[]    = {0, 0};
  46     double bytes_amount[]    = {0, 0, static_cast<double>(read_size), 0};
  47
  48     msg_task_t task = MSG_parallel_task_create("file transfer for read", 2, m_host_list, flops_amount, bytes_amount,
  49                       nullptr);
  50     msg_error_t transfer = MSG_parallel_task_execute(task);
  51     MSG_task_destroy(task);
  52
  53     if(transfer != MSG_OK){
  54       if (transfer == MSG_HOST_FAILURE)
  55         XBT_WARN("Transfer error, %s remote host just turned off!", attached_host->getCname());
  56       if (transfer == MSG_TASK_CANCELED)
  57         XBT_WARN("Transfer error, task has been canceled!");
  58
  59       return -1;
  60     }
  61   }
  62   return read_size;
  63 }
  64
  65 /** \ingroup msg_file
  66  * \brief Write into a file (local or remote)
  67  *
  68  * \param size of the file to write
  69  * \param fd is a the file descriptor
  70  * \return the number of bytes successfully write or -1 if an error occurred
  71  */
  72 sg_size_t MSG_file_write(msg_file_t fd, sg_size_t size)
  73 {
  74   if (size == 0) /* Nothing to write, return */
  75     return 0;
  76
  77   /* Find the host where the file is physically located (remote or local)*/
  78   msg_storage_t storage_src = fd->localStorage;
  79   msg_host_t attached_host  = storage_src->getHost();
  80
  81   if (strcmp(attached_host->getCname(), MSG_host_self()->getCname())) {
  82     /* the file is hosted on a remote host, initiate a communication between src and dest hosts for data transfer */
  83     XBT_DEBUG("File is on %s remote host, initiate data transfer of %llu bytes.", attached_host->getCname(), size);
  84     msg_host_t m_host_list[] = {MSG_host_self(), attached_host};
  85     double flops_amount[]    = {0, 0};
  86     double bytes_amount[]    = {0, static_cast<double>(size), 0, 0};
  87
  88     msg_task_t task = MSG_parallel_task_create("file transfer for write", 2, m_host_list, flops_amount, bytes_amount,
  89                                                nullptr);
  90     msg_error_t transfer = MSG_parallel_task_execute(task);
  91     MSG_task_destroy(task);
  92
  93     if(transfer != MSG_OK){
  94       if (transfer == MSG_HOST_FAILURE)
  95         XBT_WARN("Transfer error, %s remote host just turned off!", attached_host->getCname());
  96       if (transfer == MSG_TASK_CANCELED)
  97         XBT_WARN("Transfer error, task has been canceled!");
  98
  99       return -1;
 100     }
 101   }
 102   /* Write file on local or remote host */
 103   sg_size_t write_size = fd->write(size);
 104
 105   return write_size;
 106 }
 107
 108 /**
 109  * \ingroup msg_file
 110  * \brief Copy a file to another location on a remote host.
 111  * \param file : the file to move
 112  * \param host : the remote host where the file has to be copied
 113  * \param fullpath : the complete path destination on the remote host
 114  * \return If successful, the function returns MSG_OK. Otherwise, it returns MSG_TASK_CANCELED.
 115  */
 116 msg_error_t MSG_file_rcopy (msg_file_t file, msg_host_t host, const char* fullpath)
 117 {
 118   /* Find the host where the file is physically located and read it */
 119   msg_storage_t storage_src = file->localStorage;
 120   msg_host_t src_host       = storage_src->getHost();
 121   file->seek(0, SEEK_SET);
 122   sg_size_t read_size = file->read(file->size());
 123
 124   /* Find the host that owns the storage where the file has to be copied */
 125   msg_storage_t storage_dest = nullptr;
 126   msg_host_t dst_host;
 127   size_t longest_prefix_length = 0;
 128
 129   for (auto const& elm : host->getMountedStorages()) {
 130     std::string mount_point = std::string(fullpath).substr(0, elm.first.size());
 131     if (mount_point == elm.first && elm.first.length() > longest_prefix_length) {
 132       /* The current mount name is found in the full path and is bigger than the previous*/
 133       longest_prefix_length = elm.first.length();
 134       storage_dest          = elm.second;
 135     }
 136   }
 137
 138   if (storage_dest != nullptr) {
 139     /* Mount point found, retrieve the host the storage is attached to */
 140     dst_host = storage_dest->getHost();
 141   }else{
 142     XBT_WARN("Can't find mount point for '%s' on destination host '%s'", fullpath, host->getCname());
 143     return MSG_TASK_CANCELED;
 144   }
 145
 146   XBT_DEBUG("Initiate data transfer of %llu bytes between %s and %s.", read_size, src_host->getCname(),
 147             storage_dest->getHost()->getCname());
 148   msg_host_t m_host_list[] = {src_host, dst_host};
 149   double flops_amount[]    = {0, 0};
 150   double bytes_amount[]    = {0, static_cast<double>(read_size), 0, 0};
 151
 152   msg_task_t task =
 153       MSG_parallel_task_create("file transfer for write", 2, m_host_list, flops_amount, bytes_amount, nullptr);
 154   msg_error_t err = MSG_parallel_task_execute(task);
 155   MSG_task_destroy(task);
 156
 157   if (err != MSG_OK) {
 158     if (err == MSG_HOST_FAILURE)
 159       XBT_WARN("Transfer error, %s remote host just turned off!", storage_dest->getHost()->getCname());
 160     if (err == MSG_TASK_CANCELED)
 161       XBT_WARN("Transfer error, task has been canceled!");
 162
 163     return err;
 164   }
 165
 166   /* Create file on remote host, write it and close it */
 167   msg_file_t fd = new simgrid::s4u::File(fullpath, dst_host, nullptr);
 168   fd->write(read_size);
 169   delete fd;
 170   return MSG_OK;
 171 }
 172
 173 /**
 174  * \ingroup msg_file
 175  * \brief Move a file to another location on a remote host.
 176  * \param file : the file to move
 177  * \param host : the remote host where the file has to be moved
 178  * \param fullpath : the complete path destination on the remote host
 179  * \return If successful, the function returns MSG_OK. Otherwise, it returns MSG_TASK_CANCELED.
 180  */
 181 msg_error_t MSG_file_rmove (msg_file_t file, msg_host_t host, const char* fullpath)
 182 {
 183   msg_error_t res = MSG_file_rcopy(file, host, fullpath);
 184   file->unlink();
 185   return res;
 186 }
 187
 188 /********************************* Storage **************************************/
 189 /** @addtogroup msg_storage_management
 190  * (#msg_storage_t) and the functions for managing it.
 191  */
 192
 193 /** \ingroup msg_storage_management
 194  *
 195  * \brief Returns the name of the #msg_storage_t.
 196  *
 197  * This functions checks whether a storage is a valid pointer or not and return its name.
 198  */
 199 const char* MSG_storage_get_name(msg_storage_t storage)
 200 {
 201   xbt_assert((storage != nullptr), "Invalid parameters");
 202   return storage->getCname();
 203 }
 204
 205 const char* MSG_storage_get_host(msg_storage_t storage)
 206 {
 207   xbt_assert((storage != nullptr), "Invalid parameters");
 208   return storage->getHost()->getCname();
 209 }
 210
 211 /** \ingroup msg_storage_management
 212  * \brief Returns a xbt_dict_t consisting of the list of properties assigned to this storage
 213  * \param storage a storage
 214  * \return a dict containing the properties
 215  */
 216 xbt_dict_t MSG_storage_get_properties(msg_storage_t storage)
 217 {
 218   xbt_assert((storage != nullptr), "Invalid parameters (storage is nullptr)");
 219   xbt_dict_t as_dict = xbt_dict_new_homogeneous(xbt_free_f);
 220   std::map<std::string, std::string>* props = storage->getProperties();
 221   if (props == nullptr)
 222     return nullptr;
 223   for (auto const& elm : *props) {
 224     xbt_dict_set(as_dict, elm.first.c_str(), xbt_strdup(elm.second.c_str()), nullptr);
 225   }
 226   return as_dict;
 227 }
 228
 229 /** \ingroup msg_storage_management
 230  * \brief Change the value of a given storage property
 231  *
 232  * \param storage a storage
 233  * \param name a property name
 234  * \param value what to change the property to
 235  */
 236 void MSG_storage_set_property_value(msg_storage_t storage, const char* name, char* value)
 237 {
 238   storage->setProperty(name, value);
 239 }
 240
 241 /** \ingroup m_storage_management
 242  * \brief Returns the value of a given storage property
 243  *
 244  * \param storage a storage
 245  * \param name a property name
 246  * \return value of a property (or nullptr if property not set)
 247  */
 248 const char *MSG_storage_get_property_value(msg_storage_t storage, const char *name)
 249 {
 250   return storage->getProperty(name);
 251 }
 252
 253 /** \ingroup msg_storage_management
 254  * \brief Finds a msg_storage_t using its name.
 255  * \param name the name of a storage
 256  * \return the corresponding storage
 257  */
 258 msg_storage_t MSG_storage_get_by_name(const char *name)
 259 {
 260   return simgrid::s4u::Storage::byName(name);
 261 }
 262
 263 /** \ingroup msg_storage_management
 264  * \brief Returns a dynar containing all the storage elements declared at a given point of time
 265  */
 266 xbt_dynar_t MSG_storages_as_dynar()
 267 {
 268   std::map<std::string, simgrid::s4u::Storage*>* storage_map = simgrid::s4u::allStorages();
 269   xbt_dynar_t res = xbt_dynar_new(sizeof(msg_storage_t),nullptr);
 270   for (auto const& s : *storage_map)
 271     xbt_dynar_push(res, &(s.second));
 272   delete storage_map;
 273   return res;
 274 }
 275
 276 void* MSG_storage_get_data(msg_storage_t storage)
 277 {
 278   xbt_assert((storage != nullptr), "Invalid parameters");
 279   return storage->getUserdata();
 280 }
 281
 282 msg_error_t MSG_storage_set_data(msg_storage_t storage, void *data)
 283 {
 284   storage->setUserdata(data);
 285   return MSG_OK;
 286 }
 287
 288 sg_size_t MSG_storage_read(msg_storage_t storage, sg_size_t size)
 289 {
 290   return storage->read(size);
 291 }
 292
 293 sg_size_t MSG_storage_write(msg_storage_t storage, sg_size_t size)
 294 {
 295   return storage->write(size);
 296 }
 297
 298 }