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