src/msg/msg_gos.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_gos, msg, "Logging specific to MSG (gos)");
  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 Receives a task from a mailbox.
  77  *
  78  * This is a blocking function, the execution flow will be blocked until the task is received. See #MSG_task_irecv
  79  * for receiving tasks asynchronously.
  80  *
  81  * @param task a memory location for storing a #msg_task_t.
  82  * @param alias name of the mailbox to receive the task from
  83  *
  84  * @return Returns
  85  * #MSG_OK if the task was successfully received,
  86  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
  87  */
  88 msg_error_t MSG_task_receive(msg_task_t * task, const char *alias)
  89 {
  90   return MSG_task_receive_with_timeout(task, alias, -1);
  91 }
  92
  93 /**
  94  * @brief Receives a task from a mailbox at a given rate.
  95  *
  96  * @param task a memory location for storing a #msg_task_t.
  97  * @param alias name of the mailbox to receive the task from
  98  * @param rate limit the reception to rate bandwidth (byte/sec)
  99  *
 100  * The rate parameter can be used to receive a task with a limited bandwidth (smaller than the physical available
 101  * value). Use MSG_task_receive() if you don't limit the rate (or pass -1 as a rate value do disable this feature).
 102  *
 103  * @return Returns
 104  * #MSG_OK if the task was successfully received,
 105  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
 106  */
 107 msg_error_t MSG_task_receive_bounded(msg_task_t * task, const char *alias, double rate)
 108 {
 109   return MSG_task_receive_with_timeout_bounded(task, alias, -1, rate);
 110 }
 111
 112 /**
 113  * @brief Receives a task from a mailbox with a given timeout.
 114  *
 115  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
 116  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously.  You can provide a -1 timeout
 117  * to obtain an infinite timeout.
 118  *
 119  * @param task a memory location for storing a #msg_task_t.
 120  * @param alias name of the mailbox to receive the task from
 121  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
 122  *
 123  * @return Returns
 124  * #MSG_OK if the task was successfully received,
 125  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 126  */
 127 msg_error_t MSG_task_receive_with_timeout(msg_task_t * task, const char *alias, double timeout)
 128 {
 129   return MSG_task_receive_ext_bounded(task, alias, timeout, nullptr, -1);
 130 }
 131
 132 /**
 133  * @brief Receives a task from a mailbox with a given timeout and at a given rate.
 134  *
 135  * @param task a memory location for storing a #msg_task_t.
 136  * @param alias name of the mailbox to receive the task from
 137  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
 138  * @param rate limit the reception to rate bandwidth (byte/sec)
 139  *
 140  * The rate parameter can be used to send a task with a limited
 141  * bandwidth (smaller than the physical available value). Use
 142  * MSG_task_receive() if you don't limit the rate (or pass -1 as a
 143  * rate value do disable this feature).
 144  *
 145  * @return Returns
 146  * #MSG_OK if the task was successfully received,
 147  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 148  */
 149 msg_error_t MSG_task_receive_with_timeout_bounded(msg_task_t* task, const char* alias, double timeout, double rate)
 150 {
 151   return MSG_task_receive_ext_bounded(task, alias, timeout, nullptr, rate);
 152 }
 153
 154 /**
 155  * @brief Receives a task from a mailbox from a specific host with a given timeout.
 156  *
 157  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
 158  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously. You can provide a -1 timeout
 159  * to obtain an infinite timeout.
 160  *
 161  * @param task a memory location for storing a #msg_task_t.
 162  * @param alias name of the mailbox to receive the task from
 163  * @param timeout is the maximum wait time for completion (provide -1 for no timeout)
 164  * @param host a #msg_host_t host from where the task was sent
 165  *
 166  * @return Returns
 167  * #MSG_OK if the task was successfully received,
 168  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 169  */
 170 msg_error_t MSG_task_receive_ext(msg_task_t * task, const char *alias, double timeout, msg_host_t host)
 171 {
 172   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
 173   return MSG_task_receive_ext_bounded(task, alias, timeout, host, -1.0);
 174 }
 175
 176 /**
 177  * @brief Receives a task from a mailbox from a specific host with a given timeout  and at a given rate.
 178  *
 179  * @param task a memory location for storing a #msg_task_t.
 180  * @param alias name of the mailbox to receive the task from
 181  * @param timeout is the maximum wait time for completion (provide -1 for no timeout)
 182  * @param host a #msg_host_t host from where the task was sent
 183  * @param rate limit the reception to rate bandwidth (byte/sec)
 184  *
 185  * The rate parameter can be used to receive a task with a limited bandwidth (smaller than the physical available
 186  * value). Use MSG_task_receive_ext() if you don't limit the rate (or pass -1 as a rate value do disable this feature).
 187  *
 188  * @return Returns
 189  * #MSG_OK if the task was successfully received,
 190  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 191  */
 192 msg_error_t MSG_task_receive_ext_bounded(msg_task_t * task, const char *alias, double timeout, msg_host_t host,
 193                                          double rate)
 194 {
 195   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
 196   msg_error_t ret = MSG_OK;
 197   /* We no longer support getting a task from a specific host */
 198   if (host)
 199     THROW_UNIMPLEMENTED;
 200
 201   /* Sanity check */
 202   xbt_assert(task, "Null pointer for the task storage");
 203
 204   if (*task)
 205     XBT_WARN("Asked to write the received task in a non empty struct -- proceeding.");
 206
 207   /* Try to receive it by calling SIMIX network layer */
 208   try {
 209     void* payload;
 210     simgrid::s4u::Mailbox::by_name(alias)
 211         ->get_init()
 212         ->set_dst_data(&payload, sizeof(msg_task_t*))
 213         ->set_rate(rate)
 214         ->wait_for(timeout);
 215     *task = static_cast<msg_task_t>(payload);
 216     XBT_DEBUG("Got task %s from %s", (*task)->get_cname(), alias);
 217     (*task)->set_not_used();
 218   } catch (simgrid::HostFailureException& e) {
 219     ret = MSG_HOST_FAILURE;
 220   } catch (simgrid::TimeoutError& e) {
 221     ret = MSG_TIMEOUT;
 222   } catch (simgrid::CancelException& e) {
 223     ret = MSG_TASK_CANCELED;
 224   } catch (xbt_ex& e) {
 225     if (e.category == network_error)
 226       ret = MSG_TRANSFER_FAILURE;
 227     else
 228       throw;
 229   }
 230
 231   if (TRACE_actor_is_enabled() && ret != MSG_HOST_FAILURE && ret != MSG_TRANSFER_FAILURE && ret != MSG_TIMEOUT) {
 232     container_t process_container = simgrid::instr::Container::by_name(instr_pid(MSG_process_self()));
 233
 234     std::string key = std::string("p") + std::to_string((*task)->get_id());
 235     simgrid::instr::Container::get_root()->get_link("ACTOR_TASK_LINK")->end_event(process_container, "SR", key);
 236   }
 237   return ret;
 238 }
 239
 240
 241 /**
 242  * @brief Starts listening for receiving a task from an asynchronous communication.
 243  *
 244  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
 245  *
 246  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
 247  * @param name of the mailbox to receive the task on
 248  * @return the msg_comm_t communication created
 249  */
 250 msg_comm_t MSG_task_irecv(msg_task_t *task, const char *name)
 251 {
 252   return MSG_task_irecv_bounded(task, name, -1.0);
 253 }
 254
 255 /**
 256  * @brief Starts listening for receiving a task from an asynchronous communication at a given rate.
 257  *
 258  * The rate parameter can be used to receive a task with a limited
 259  * bandwidth (smaller than the physical available value). Use
 260  * MSG_task_irecv() if you don't limit the rate (or pass -1 as a rate
 261  * value do disable this feature).
 262  *
 263  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
 264  * @param name of the mailbox to receive the task on
 265  * @param rate limit the bandwidth to the given rate (byte/sec)
 266  * @return the msg_comm_t communication created
 267  */
 268 msg_comm_t MSG_task_irecv_bounded(msg_task_t *task, const char *name, double rate)
 269 {
 270   simgrid::s4u::MailboxPtr mbox = simgrid::s4u::Mailbox::by_name(name);
 271
 272   /* FIXME: these functions are not traceable */
 273   /* Sanity check */
 274   xbt_assert(task, "Null pointer for the task storage");
 275
 276   if (*task)
 277     XBT_CRITICAL("MSG_task_irecv() was asked to write in a non empty task struct.");
 278
 279   /* Try to receive it by calling SIMIX network layer */
 280   simgrid::s4u::CommPtr comm = simgrid::s4u::Mailbox::by_name(name)
 281                                    ->get_init()
 282                                    ->set_dst_data((void**)task, sizeof(msg_task_t*))
 283                                    ->set_rate(rate)
 284                                    ->start();
 285
 286   return new simgrid::msg::Comm(nullptr, task, comm);
 287 }
 288
 289 /**
 290  * @brief Checks whether a communication is done, and if yes, finalizes it.
 291  * @param comm the communication to test
 292  * @return 'true' if the communication is finished
 293  * (but it may have failed, use MSG_comm_get_status() to know its status)
 294  * or 'false' if the communication is not finished yet
 295  * If the status is 'false', don't forget to use MSG_process_sleep() after the test.
 296  */
 297 int MSG_comm_test(msg_comm_t comm)
 298 {
 299   return comm->test();
 300 }
 301
 302 /**
 303  * @brief This function checks if a communication is finished.
 304  * @param comms a vector of communications
 305  * @return the position of the finished communication if any
 306  * (but it may have failed, use MSG_comm_get_status() to know its status), or -1 if none is finished
 307  */
 308 int MSG_comm_testany(xbt_dynar_t comms)
 309 {
 310   int finished_index = -1;
 311
 312   /* Create the equivalent array with SIMIX objects: */
 313   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
 314   s_comms.reserve(xbt_dynar_length(comms));
 315   msg_comm_t comm;
 316   unsigned int cursor;
 317   xbt_dynar_foreach(comms, cursor, comm) {
 318     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
 319   }
 320
 321   msg_error_t status = MSG_OK;
 322   try {
 323     finished_index = simcall_comm_testany(s_comms.data(), s_comms.size());
 324   } catch (simgrid::TimeoutError& e) {
 325     finished_index = e.value;
 326     status         = MSG_TIMEOUT;
 327   } catch (simgrid::CancelException& e) {
 328     finished_index = e.value;
 329     status         = MSG_TASK_CANCELED;
 330   }
 331   catch (xbt_ex& e) {
 332     if (e.category != network_error)
 333       throw;
 334     finished_index = e.value;
 335     status         = MSG_TRANSFER_FAILURE;
 336   }
 337
 338   if (finished_index != -1) {
 339     comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 340     /* the communication is finished */
 341     comm->set_status(status);
 342
 343     if (status == MSG_OK && comm->task_received != nullptr) {
 344       /* I am the receiver */
 345       (*comm->task_received)->set_not_used();
 346     }
 347   }
 348
 349   return finished_index;
 350 }
 351
 352 /** @brief Destroys the provided communication. */
 353 void MSG_comm_destroy(msg_comm_t comm)
 354 {
 355   delete comm;
 356 }
 357
 358 /** @brief Wait for the completion of a communication.
 359  *
 360  * It takes two parameters.
 361  * @param comm the communication to wait.
 362  * @param timeout Wait until the communication terminates or the timeout occurs.
 363  *                You can provide a -1 timeout to obtain an infinite timeout.
 364  * @return msg_error_t
 365  */
 366 msg_error_t MSG_comm_wait(msg_comm_t comm, double timeout)
 367 {
 368   return comm->wait_for(timeout);
 369 }
 370
 371 /** @brief This function is called by a sender and permit to wait for each communication
 372  *
 373  * @param comm a vector of communication
 374  * @param nb_elem is the size of the comm vector
 375  * @param timeout for each call of MSG_comm_wait
 376  */
 377 void MSG_comm_waitall(msg_comm_t * comm, int nb_elem, double timeout)
 378 {
 379   for (int i = 0; i < nb_elem; i++)
 380     comm[i]->wait_for(timeout);
 381 }
 382
 383 /** @brief This function waits for the first communication finished in a list.
 384  * @param comms a vector of communications
 385  * @return the position of the first finished communication
 386  * (but it may have failed, use MSG_comm_get_status() to know its status)
 387  */
 388 int MSG_comm_waitany(xbt_dynar_t comms)
 389 {
 390   int finished_index = -1;
 391
 392   /* Create the equivalent array with SIMIX objects: */
 393   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
 394   s_comms.reserve(xbt_dynar_length(comms));
 395   msg_comm_t comm;
 396   unsigned int cursor;
 397   xbt_dynar_foreach(comms, cursor, comm) {
 398     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
 399   }
 400
 401   msg_error_t status = MSG_OK;
 402   try {
 403     finished_index = simcall_comm_waitany(s_comms.data(), s_comms.size(), -1);
 404   } catch (simgrid::TimeoutError& e) {
 405     finished_index = e.value;
 406     status         = MSG_TIMEOUT;
 407   } catch (simgrid::CancelException& e) {
 408     finished_index = e.value;
 409     status         = MSG_TASK_CANCELED;
 410   }
 411   catch(xbt_ex& e) {
 412     if (e.category == network_error) {
 413       finished_index = e.value;
 414       status         = MSG_TRANSFER_FAILURE;
 415     } else {
 416       throw;
 417     }
 418   }
 419
 420   xbt_assert(finished_index != -1, "WaitAny returned -1");
 421
 422   comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 423   /* the communication is finished */
 424   comm->set_status(status);
 425
 426   if (comm->task_received != nullptr) {
 427     /* I am the receiver */
 428     (*comm->task_received)->set_not_used();
 429   }
 430
 431   return finished_index;
 432 }
 433
 434 /**
 435  * @brief Returns the error (if any) that occurred during a finished communication.
 436  * @param comm a finished communication
 437  * @return the status of the communication, or #MSG_OK if no error occurred during the communication
 438  */
 439 msg_error_t MSG_comm_get_status(msg_comm_t comm) {
 440
 441   return comm->get_status();
 442 }
 443
 444 /** @brief Get a task (#msg_task_t) from a communication
 445  *
 446  * @param comm the communication where to get the task
 447  * @return the task from the communication
 448  */
 449 msg_task_t MSG_comm_get_task(msg_comm_t comm)
 450 {
 451   xbt_assert(comm, "Invalid parameter");
 452
 453   return comm->task_received ? *comm->task_received : comm->task_sent;
 454 }
 455
 456 /**
 457  * @brief This function is called by SIMIX in kernel mode to copy the data of a comm.
 458  * @param comm the comm
 459  * @param buff the data copied
 460  * @param buff_size size of the buffer
 461  */
 462 // deprecated but used by MSG_set_copy_callback. Should be removed in v325
 463 void MSG_comm_copy_data_from_SIMIX(simgrid::kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
 464 {
 465   SIMIX_comm_copy_pointer_callback(comm, buff, buff_size);
 466
 467   // notify the user callback if any
 468   if (msg_global->task_copy_callback) {
 469     msg_task_t task = static_cast<msg_task_t>(buff);
 470     msg_global->task_copy_callback(task, comm->src_actor_->ciface(), comm->dst_actor_->ciface());
 471   }
 472 }
 473
 474 /**
 475  * @brief Look if there is a communication on a mailbox and return the PID of the sender process.
 476  *
 477  * @param alias the name of the mailbox to be considered
 478  *
 479  * @return Returns the PID of sender process,
 480  * -1 if there is no communication in the mailbox.#include <cmath>
 481  *
 482  */
 483 int MSG_task_listen_from(const char *alias)
 484 {
 485   /* looks inside the rdv directly. Not clean. */
 486   simgrid::kernel::activity::CommImplPtr comm = simgrid::s4u::Mailbox::by_name(alias)->front();
 487
 488   if (comm && comm->src_actor_)
 489     return comm->src_actor_->get_pid();
 490   else
 491     return -1;
 492 }