src/msg/msg_comm.cpp

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