src/msg/msg_comm.cpp

   1 /* Copyright (c) 2004-2019. 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 <cmath>
   7
   8 #include "simgrid/Exception.hpp"
   9 #include "simgrid/s4u/Actor.hpp"
  10 #include "simgrid/s4u/Comm.hpp"
  11 #include "simgrid/s4u/Exec.hpp"
  12 #include "simgrid/s4u/Mailbox.hpp"
  13 #include "src/instr/instr_private.hpp"
  14 #include "src/msg/msg_private.hpp"
  15
  16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_comm, msg, "Logging specific to MSG (comm)");
  17
  18 namespace simgrid {
  19 namespace msg {
  20
  21 bool Comm::test()
  22 {
  23   bool finished = false;
  24
  25   try {
  26     finished = s_comm->test();
  27     if (finished && task_received != nullptr) {
  28       /* I am the receiver */
  29       (*task_received)->set_not_used();
  30     }
  31   } catch (simgrid::TimeoutError& e) {
  32     status_  = MSG_TIMEOUT;
  33     finished = true;
  34   } catch (simgrid::CancelException& e) {
  35     status_  = MSG_TASK_CANCELED;
  36     finished = true;
  37   } catch (xbt_ex& e) {
  38     if (e.category == network_error) {
  39       status_  = MSG_TRANSFER_FAILURE;
  40       finished = true;
  41     } else {
  42       throw;
  43     }
  44   }
  45
  46   return finished;
  47 }
  48 msg_error_t Comm::wait_for(double timeout)
  49 {
  50   try {
  51     s_comm->wait_for(timeout);
  52
  53     if (task_received != nullptr) {
  54       /* I am the receiver */
  55       (*task_received)->set_not_used();
  56     }
  57
  58     /* FIXME: these functions are not traceable */
  59   } catch (simgrid::TimeoutError& e) {
  60     status_ = MSG_TIMEOUT;
  61   } catch (simgrid::CancelException& e) {
  62     status_ = MSG_TASK_CANCELED;
  63   } catch (xbt_ex& e) {
  64     if (e.category == network_error)
  65       status_ = MSG_TRANSFER_FAILURE;
  66     else
  67       throw;
  68   }
  69
  70   return status_;
  71 }
  72 } // namespace msg
  73 } // namespace simgrid
  74
  75 /**
  76  * @brief Checks whether a communication is done, and if yes, finalizes it.
  77  * @param comm the communication to test
  78  * @return 'true' if the communication is finished
  79  * (but it may have failed, use MSG_comm_get_status() to know its status)
  80  * or 'false' if the communication is not finished yet
  81  * If the status is 'false', don't forget to use MSG_process_sleep() after the test.
  82  */
  83 int MSG_comm_test(msg_comm_t comm)
  84 {
  85   return comm->test();
  86 }
  87
  88 /**
  89  * @brief This function checks if a communication is finished.
  90  * @param comms a vector of communications
  91  * @return the position of the finished communication if any
  92  * (but it may have failed, use MSG_comm_get_status() to know its status), or -1 if none is finished
  93  */
  94 int MSG_comm_testany(xbt_dynar_t comms)
  95 {
  96   int finished_index = -1;
  97
  98   /* Create the equivalent array with SIMIX objects: */
  99   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
 100   s_comms.reserve(xbt_dynar_length(comms));
 101   msg_comm_t comm;
 102   unsigned int cursor;
 103   xbt_dynar_foreach (comms, cursor, comm) {
 104     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
 105   }
 106
 107   msg_error_t status = MSG_OK;
 108   try {
 109     finished_index = simcall_comm_testany(s_comms.data(), s_comms.size());
 110   } catch (simgrid::TimeoutError& e) {
 111     finished_index = e.value;
 112     status         = MSG_TIMEOUT;
 113   } catch (simgrid::CancelException& e) {
 114     finished_index = e.value;
 115     status         = MSG_TASK_CANCELED;
 116   } catch (xbt_ex& e) {
 117     if (e.category != network_error)
 118       throw;
 119     finished_index = e.value;
 120     status         = MSG_TRANSFER_FAILURE;
 121   }
 122
 123   if (finished_index != -1) {
 124     comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 125     /* the communication is finished */
 126     comm->set_status(status);
 127
 128     if (status == MSG_OK && comm->task_received != nullptr) {
 129       /* I am the receiver */
 130       (*comm->task_received)->set_not_used();
 131     }
 132   }
 133
 134   return finished_index;
 135 }
 136
 137 /** @brief Destroys the provided communication. */
 138 void MSG_comm_destroy(msg_comm_t comm)
 139 {
 140   delete comm;
 141 }
 142
 143 /** @brief Wait for the completion of a communication.
 144  *
 145  * It takes two parameters.
 146  * @param comm the communication to wait.
 147  * @param timeout Wait until the communication terminates or the timeout occurs.
 148  *                You can provide a -1 timeout to obtain an infinite timeout.
 149  * @return msg_error_t
 150  */
 151 msg_error_t MSG_comm_wait(msg_comm_t comm, double timeout)
 152 {
 153   return comm->wait_for(timeout);
 154 }
 155
 156 /** @brief This function is called by a sender and permit to wait for each communication
 157  *
 158  * @param comm a vector of communication
 159  * @param nb_elem is the size of the comm vector
 160  * @param timeout for each call of MSG_comm_wait
 161  */
 162 void MSG_comm_waitall(msg_comm_t* comm, int nb_elem, double timeout)
 163 {
 164   for (int i = 0; i < nb_elem; i++)
 165     comm[i]->wait_for(timeout);
 166 }
 167
 168 /** @brief This function waits for the first communication finished in a list.
 169  * @param comms a vector of communications
 170  * @return the position of the first finished communication
 171  * (but it may have failed, use MSG_comm_get_status() to know its status)
 172  */
 173 int MSG_comm_waitany(xbt_dynar_t comms)
 174 {
 175   int finished_index = -1;
 176
 177   /* Create the equivalent array with SIMIX objects: */
 178   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
 179   s_comms.reserve(xbt_dynar_length(comms));
 180   msg_comm_t comm;
 181   unsigned int cursor;
 182   xbt_dynar_foreach (comms, cursor, comm) {
 183     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
 184   }
 185
 186   msg_error_t status = MSG_OK;
 187   try {
 188     finished_index = simcall_comm_waitany(s_comms.data(), s_comms.size(), -1);
 189   } catch (simgrid::TimeoutError& e) {
 190     finished_index = e.value;
 191     status         = MSG_TIMEOUT;
 192   } catch (simgrid::CancelException& e) {
 193     finished_index = e.value;
 194     status         = MSG_TASK_CANCELED;
 195   } catch (xbt_ex& e) {
 196     if (e.category == network_error) {
 197       finished_index = e.value;
 198       status         = MSG_TRANSFER_FAILURE;
 199     } else {
 200       throw;
 201     }
 202   }
 203
 204   xbt_assert(finished_index != -1, "WaitAny returned -1");
 205
 206   comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 207   /* the communication is finished */
 208   comm->set_status(status);
 209
 210   if (comm->task_received != nullptr) {
 211     /* I am the receiver */
 212     (*comm->task_received)->set_not_used();
 213   }
 214
 215   return finished_index;
 216 }
 217
 218 /**
 219  * @brief Returns the error (if any) that occurred during a finished communication.
 220  * @param comm a finished communication
 221  * @return the status of the communication, or #MSG_OK if no error occurred during the communication
 222  */
 223 msg_error_t MSG_comm_get_status(msg_comm_t comm)
 224 {
 225
 226   return comm->get_status();
 227 }
 228
 229 /** @brief Get a task (#msg_task_t) from a communication
 230  *
 231  * @param comm the communication where to get the task
 232  * @return the task from the communication
 233  */
 234 msg_task_t MSG_comm_get_task(msg_comm_t comm)
 235 {
 236   xbt_assert(comm, "Invalid parameter");
 237
 238   return comm->task_received ? *comm->task_received : comm->task_sent;
 239 }
 240
 241 /**
 242  * @brief This function is called by SIMIX in kernel mode to copy the data of a comm.
 243  * @param comm the comm
 244  * @param buff the data copied
 245  * @param buff_size size of the buffer
 246  */
 247 // deprecated but used by MSG_set_copy_callback. Should be removed in v325
 248 void MSG_comm_copy_data_from_SIMIX(simgrid::kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
 249 {
 250   SIMIX_comm_copy_pointer_callback(comm, buff, buff_size);
 251
 252   // notify the user callback if any
 253   if (msg_global->task_copy_callback) {
 254     msg_task_t task = static_cast<msg_task_t>(buff);
 255     msg_global->task_copy_callback(task, comm->src_actor_->ciface(), comm->dst_actor_->ciface());
 256   }
 257 }