src/msg/msg_comm.cpp

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