src/simix/smx_network.cpp

   1 /* Copyright (c) 2009-2016. 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 <boost/range/algorithm.hpp>
   7
   8 #include "src/surf/surf_interface.hpp"
   9 #include "src/simix/smx_private.h"
  10 #include "xbt/log.h"
  11 #include "mc/mc.h"
  12 #include "src/mc/mc_replay.h"
  13 #include "xbt/dict.h"
  14 #include "simgrid/s4u/mailbox.hpp"
  15
  16 #include "src/simix/SynchroComm.hpp"
  17
  18 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization");
  19
  20 static void SIMIX_mbox_free(void *data);
  21 static xbt_dict_t mailboxes = xbt_dict_new_homogeneous(SIMIX_mbox_free);
  22
  23 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
  24 static void SIMIX_comm_copy_data(smx_synchro_t comm);
  25 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t comm);
  26 static smx_synchro_t _find_matching_comm(std::deque<smx_synchro_t> *deque, e_smx_comm_type_t type,
  27     int (*match_fun)(void *, void *,smx_synchro_t), void *user_data, smx_synchro_t my_synchro, bool remove_matching);
  28 static void SIMIX_comm_start(smx_synchro_t synchro);
  29
  30 void SIMIX_mailbox_exit(void)
  31 {
  32   xbt_dict_free(&mailboxes);
  33 }
  34
  35 /******************************************************************************/
  36 /*                           Rendez-Vous Points                               */
  37 /******************************************************************************/
  38
  39 smx_mailbox_t SIMIX_mbox_create(const char *name)
  40 {
  41   xbt_assert(name, "Mailboxes must have a name");
  42   /* two processes may have pushed the same mbox_create simcall at the same time */
  43   smx_mailbox_t mbox = (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
  44
  45   if (!mbox) {
  46     mbox = xbt_new0(s_smx_mailbox_t, 1);
  47     mbox->name = xbt_strdup(name);
  48     mbox->comm_queue = new std::deque<smx_synchro_t>();
  49     mbox->done_comm_queue = nullptr; // Allocated on need only
  50     mbox->permanent_receiver=nullptr;
  51
  52     XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name);
  53     xbt_dict_set(mailboxes, mbox->name, mbox, nullptr);
  54   }
  55   return mbox;
  56 }
  57
  58 void SIMIX_mbox_free(void *data)
  59 {
  60   XBT_DEBUG("mbox free %p", data);
  61   smx_mailbox_t mbox = (smx_mailbox_t) data;
  62   xbt_free(mbox->name);
  63   delete mbox->comm_queue;
  64   delete mbox->done_comm_queue;
  65
  66   xbt_free(mbox);
  67 }
  68
  69 smx_mailbox_t SIMIX_mbox_get_by_name(const char *name)
  70 {
  71   return (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
  72 }
  73
  74 /**
  75  *  \brief set the receiver of the rendez vous point to allow eager sends
  76  *  \param mbox The rendez-vous point
  77  *  \param process The receiving process
  78  */
  79 void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_process_t process)
  80 {
  81   mbox->permanent_receiver=process;
  82   if (mbox->done_comm_queue == nullptr)
  83     mbox->done_comm_queue = new std::deque<smx_synchro_t>();
  84 }
  85
  86 /**
  87  *  \brief Pushes a communication synchro into a rendez-vous point
  88  *  \param mbox The mailbox
  89  *  \param synchro The communication synchro
  90  */
  91 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t synchro)
  92 {
  93   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
  94
  95   mbox->comm_queue->push_back(comm);
  96   comm->mbox = mbox;
  97 }
  98
  99 /**
 100  *  \brief Removes a communication synchro from a rendez-vous point
 101  *  \param mbox The rendez-vous point
 102  *  \param synchro The communication synchro
 103  */
 104 void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_synchro_t synchro)
 105 {
 106   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 107
 108   comm->mbox = nullptr;
 109   for (auto it = mbox->comm_queue->begin(); it != mbox->comm_queue->end(); it++)
 110     if (*it == comm) {
 111       mbox->comm_queue->erase(it);
 112       return;
 113     }
 114   xbt_die("Cannot remove this comm that is not part of the mailbox");
 115 }
 116
 117 /**
 118  *  \brief Checks if there is a communication synchro queued in a deque matching our needs
 119  *  \param type The type of communication we are looking for (comm_send, comm_recv)
 120  *  \return The communication synchro if found, nullptr otherwise
 121  */
 122 static smx_synchro_t _find_matching_comm(std::deque<smx_synchro_t> *deque, e_smx_comm_type_t type,
 123     int (*match_fun)(void *, void *,smx_synchro_t), void *this_user_data, smx_synchro_t my_synchro, bool remove_matching)
 124 {
 125   void* other_user_data = nullptr;
 126
 127   for(auto it = deque->begin(); it != deque->end(); it++){
 128     smx_synchro_t synchro = *it;
 129     simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 130
 131     if (comm->type == SIMIX_COMM_SEND) {
 132       other_user_data = comm->src_data;
 133     } else if (comm->type == SIMIX_COMM_RECEIVE) {
 134       other_user_data = comm->dst_data;
 135     }
 136     if (comm->type == type &&
 137         (!      match_fun ||       match_fun(this_user_data,  other_user_data, synchro)) &&
 138         (!comm->match_fun || comm->match_fun(other_user_data, this_user_data,  my_synchro))) {
 139       XBT_DEBUG("Found a matching communication synchro %p", comm);
 140       if (remove_matching)
 141         deque->erase(it);
 142       comm->ref();
 143 #if HAVE_MC
 144       comm->mbox_cpy = comm->mbox;
 145 #endif
 146       comm->mbox = nullptr;
 147       return comm;
 148     }
 149     XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
 150               " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
 151               comm, (int)comm->type, (int)type);
 152   }
 153   XBT_DEBUG("No matching communication synchro found");
 154   return nullptr;
 155 }
 156
 157 /******************************************************************************/
 158 /*                          Communication synchros                            */
 159 /******************************************************************************/
 160 XBT_PRIVATE void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_process_t src, smx_mailbox_t mbox,
 161                                   double task_size, double rate,
 162                                   void *src_buff, size_t src_buff_size,
 163                                   int (*match_fun)(void *, void *,smx_synchro_t),
 164                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 165           void *data, double timeout){
 166   smx_synchro_t comm = simcall_HANDLER_comm_isend(simcall, src, mbox, task_size, rate,
 167                            src_buff, src_buff_size, match_fun, nullptr, copy_data_fun,
 168                data, 0);
 169   SIMCALL_SET_MC_VALUE(simcall, 0);
 170   simcall_HANDLER_comm_wait(simcall, comm, timeout);
 171 }
 172 XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_process_t src_proc, smx_mailbox_t mbox,
 173                                   double task_size, double rate,
 174                                   void *src_buff, size_t src_buff_size,
 175                                   int (*match_fun)(void *, void *,smx_synchro_t),
 176                                   void (*clean_fun)(void *), // used to free the synchro in case of problem after a detached send
 177                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),// used to copy data if not default one
 178                           void *data, int detached)
 179 {
 180   XBT_DEBUG("send from %p", mbox);
 181
 182   /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
 183   simgrid::simix::Comm* this_synchro = new simgrid::simix::Comm(SIMIX_COMM_SEND);
 184
 185   /* Look for communication synchro matching our needs. We also provide a description of
 186    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 187    *
 188    * If it is not found then push our communication into the rendez-vous point */
 189   smx_synchro_t other_synchro =
 190       _find_matching_comm(mbox->comm_queue, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro, /*remove_matching*/true);
 191   simgrid::simix::Comm *other_comm = static_cast<simgrid::simix::Comm*>(other_synchro);
 192
 193
 194   if (!other_synchro) {
 195     other_synchro = this_synchro;
 196     other_comm = static_cast<simgrid::simix::Comm*>(other_synchro);
 197
 198     if (mbox->permanent_receiver!=nullptr){
 199       //this mailbox is for small messages, which have to be sent right now
 200       other_synchro->state = SIMIX_READY;
 201       other_comm->dst_proc=mbox->permanent_receiver;
 202       other_comm->ref();
 203       mbox->done_comm_queue->push_back(other_synchro);
 204       other_comm->mbox=mbox;
 205       XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", mbox, &(other_comm));
 206
 207     }else{
 208       SIMIX_mbox_push(mbox, this_synchro);
 209     }
 210   } else {
 211     XBT_DEBUG("Receive already pushed");
 212     this_synchro->unref();
 213
 214     other_comm->state = SIMIX_READY;
 215     other_comm->type = SIMIX_COMM_READY;
 216
 217   }
 218   xbt_fifo_push(src_proc->comms, other_synchro);
 219
 220
 221   if (detached) {
 222     other_comm->detached = true;
 223     other_comm->clean_fun = clean_fun;
 224   } else {
 225     other_comm->clean_fun = nullptr;
 226   }
 227
 228   /* Setup the communication synchro */
 229   other_comm->src_proc = src_proc;
 230   other_comm->task_size = task_size;
 231   other_comm->rate = rate;
 232   other_comm->src_buff = src_buff;
 233   other_comm->src_buff_size = src_buff_size;
 234   other_comm->src_data = data;
 235
 236   other_comm->match_fun = match_fun;
 237   other_comm->copy_data_fun = copy_data_fun;
 238
 239
 240   if (MC_is_active() || MC_record_replay_is_active()) {
 241     other_comm->state = SIMIX_RUNNING;
 242     return (detached ? nullptr : other_comm);
 243   }
 244
 245   SIMIX_comm_start(other_comm);
 246   return (detached ? nullptr : other_comm);
 247 }
 248
 249 XBT_PRIVATE void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
 250                          void *dst_buff, size_t *dst_buff_size,
 251                          int (*match_fun)(void *, void *, smx_synchro_t),
 252                          void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 253                          void *data, double timeout, double rate)
 254 {
 255   smx_synchro_t comm = SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
 256   SIMCALL_SET_MC_VALUE(simcall, 0);
 257   simcall_HANDLER_comm_wait(simcall, comm, timeout);
 258 }
 259
 260 XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_irecv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
 261     void *dst_buff, size_t *dst_buff_size,
 262     int (*match_fun)(void *, void *, smx_synchro_t),
 263     void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 264     void *data, double rate)
 265 {
 266   return SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
 267 }
 268
 269 smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void *dst_buff, size_t *dst_buff_size,
 270     int (*match_fun)(void *, void *, smx_synchro_t),
 271     void (*copy_data_fun)(smx_synchro_t, void*, size_t), // used to copy data if not default one
 272     void *data, double rate)
 273 {
 274   XBT_DEBUG("recv from %p %p", mbox, mbox->comm_queue);
 275   simgrid::simix::Comm* this_synchro = new simgrid::simix::Comm(SIMIX_COMM_RECEIVE);
 276
 277   smx_synchro_t other_synchro;
 278   //communication already done, get it inside the fifo of completed comms
 279   if (mbox->permanent_receiver && ! mbox->done_comm_queue->empty()) {
 280
 281     XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
 282     //find a match in the already received fifo
 283     other_synchro = _find_matching_comm(mbox->done_comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
 284     //if not found, assume the receiver came first, register it to the mailbox in the classical way
 285     if (!other_synchro)  {
 286       XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo");
 287       other_synchro = this_synchro;
 288       SIMIX_mbox_push(mbox, this_synchro);
 289     } else {
 290       simgrid::simix::Comm *other_comm = static_cast<simgrid::simix::Comm*>(other_synchro);
 291
 292       if(other_comm->surf_comm && other_comm->remains()==0.0) {
 293         XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",other_comm);
 294         other_comm->state = SIMIX_DONE;
 295         other_comm->type = SIMIX_COMM_DONE;
 296         other_comm->mbox = nullptr;
 297       }
 298       other_comm->unref();
 299       static_cast<simgrid::simix::Comm*>(this_synchro)->unref();
 300     }
 301   } else {
 302     /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
 303
 304     /* Look for communication synchro matching our needs. We also provide a description of
 305      * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 306      *
 307      * If it is not found then push our communication into the rendez-vous point */
 308     other_synchro = _find_matching_comm(mbox->comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
 309
 310     if (!other_synchro) {
 311       XBT_DEBUG("Receive pushed first %zu", mbox->comm_queue->size());
 312       other_synchro = this_synchro;
 313       SIMIX_mbox_push(mbox, this_synchro);
 314     } else {
 315       this_synchro->unref();
 316       simgrid::simix::Comm *other_comm = static_cast<simgrid::simix::Comm*>(other_synchro);
 317
 318       other_comm->state = SIMIX_READY;
 319       other_comm->type = SIMIX_COMM_READY;
 320     }
 321     xbt_fifo_push(dst_proc->comms, other_synchro);
 322   }
 323
 324   /* Setup communication synchro */
 325   simgrid::simix::Comm *other_comm = static_cast<simgrid::simix::Comm*>(other_synchro);
 326   other_comm->dst_proc = dst_proc;
 327   other_comm->dst_buff = dst_buff;
 328   other_comm->dst_buff_size = dst_buff_size;
 329   other_comm->dst_data = data;
 330
 331   if (rate != -1.0 && (other_comm->rate == -1.0 || rate < other_comm->rate))
 332     other_comm->rate = rate;
 333
 334   other_comm->match_fun = match_fun;
 335   other_comm->copy_data_fun = copy_data_fun;
 336
 337   if (MC_is_active() || MC_record_replay_is_active()) {
 338     other_synchro->state = SIMIX_RUNNING;
 339     return other_synchro;
 340   }
 341
 342   SIMIX_comm_start(other_synchro);
 343   return other_synchro;
 344 }
 345
 346 smx_synchro_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t mbox,
 347                                    int type, int src, int tag,
 348                                    int (*match_fun)(void *, void *, smx_synchro_t),
 349                                    void *data){
 350   return SIMIX_comm_iprobe(simcall->issuer, mbox, type, src, tag, match_fun, data);
 351 }
 352
 353 smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t mbox, int type, int src,
 354                               int tag, int (*match_fun)(void *, void *, smx_synchro_t), void *data)
 355 {
 356   XBT_DEBUG("iprobe from %p %p", mbox, mbox->comm_queue);
 357   simgrid::simix::Comm* this_comm;
 358   int smx_type;
 359   if(type == 1){
 360     this_comm = new simgrid::simix::Comm(SIMIX_COMM_SEND);
 361     smx_type = SIMIX_COMM_RECEIVE;
 362   } else{
 363     this_comm = new simgrid::simix::Comm(SIMIX_COMM_RECEIVE);
 364     smx_type = SIMIX_COMM_SEND;
 365   }
 366   smx_synchro_t other_synchro=nullptr;
 367   if(mbox->permanent_receiver && ! mbox->done_comm_queue->empty()){
 368     XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something");
 369     other_synchro =
 370         _find_matching_comm(mbox->done_comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false);
 371   }
 372   if (!other_synchro){
 373     XBT_DEBUG("check if we have more luck in the normal mailbox");
 374     other_synchro = _find_matching_comm(mbox->comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false);
 375   }
 376
 377   if(other_synchro)
 378     other_synchro->unref();
 379
 380   this_comm->unref();
 381   return other_synchro;
 382 }
 383
 384 void simcall_HANDLER_comm_wait(smx_simcall_t simcall, smx_synchro_t synchro, double timeout)
 385 {
 386   /* Associate this simcall to the wait synchro */
 387   XBT_DEBUG("simcall_HANDLER_comm_wait, %p", synchro);
 388
 389   synchro->simcalls.push_back(simcall);
 390   simcall->issuer->waiting_synchro = synchro;
 391
 392   if (MC_is_active() || MC_record_replay_is_active()) {
 393     int idx = SIMCALL_GET_MC_VALUE(simcall);
 394     if (idx == 0) {
 395       synchro->state = SIMIX_DONE;
 396     } else {
 397       /* If we reached this point, the wait simcall must have a timeout */
 398       /* Otherwise it shouldn't be enabled and executed by the MC */
 399       if (timeout == -1)
 400         THROW_IMPOSSIBLE;
 401
 402       simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 403       if (comm->src_proc == simcall->issuer)
 404         comm->state = SIMIX_SRC_TIMEOUT;
 405       else
 406         comm->state = SIMIX_DST_TIMEOUT;
 407     }
 408
 409     SIMIX_comm_finish(synchro);
 410     return;
 411   }
 412
 413   /* If the synchro has already finish perform the error handling, */
 414   /* otherwise set up a waiting timeout on the right side          */
 415   if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
 416     SIMIX_comm_finish(synchro);
 417   } else { /* if (timeout >= 0) { we need a surf sleep action even when there is no timeout, otherwise surf won't tell us when the host fails */
 418     surf_action_t sleep = surf_host_sleep(simcall->issuer->host, timeout);
 419     sleep->setData(synchro);
 420
 421     simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 422     if (simcall->issuer == comm->src_proc)
 423       comm->src_timeout = sleep;
 424     else
 425       comm->dst_timeout = sleep;
 426   }
 427 }
 428
 429 void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_synchro_t synchro)
 430 {
 431   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 432
 433   if (MC_is_active() || MC_record_replay_is_active()){
 434     simcall_comm_test__set__result(simcall, comm->src_proc && comm->dst_proc);
 435     if (simcall_comm_test__get__result(simcall)){
 436       synchro->state = SIMIX_DONE;
 437       synchro->simcalls.push_back(simcall);
 438       SIMIX_comm_finish(synchro);
 439     } else {
 440       SIMIX_simcall_answer(simcall);
 441     }
 442     return;
 443   }
 444
 445   simcall_comm_test__set__result(simcall, (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING));
 446   if (simcall_comm_test__get__result(simcall)) {
 447     synchro->simcalls.push_back(simcall);
 448     SIMIX_comm_finish(synchro);
 449   } else {
 450     SIMIX_simcall_answer(simcall);
 451   }
 452 }
 453
 454 void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros)
 455 {
 456   unsigned int cursor;
 457   smx_synchro_t synchro;
 458   // The default result is -1 -- this means, "nothing is ready".
 459   // It can be changed below, but only if something matches.
 460   simcall_comm_testany__set__result(simcall, -1);
 461
 462   if (MC_is_active() || MC_record_replay_is_active()){
 463     int idx = SIMCALL_GET_MC_VALUE(simcall);
 464     if(idx == -1){
 465       SIMIX_simcall_answer(simcall);
 466     }else{
 467       synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
 468       simcall_comm_testany__set__result(simcall, idx);
 469       synchro->simcalls.push_back(simcall);
 470       synchro->state = SIMIX_DONE;
 471       SIMIX_comm_finish(synchro);
 472     }
 473     return;
 474   }
 475
 476   xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,synchro) {
 477     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
 478       simcall_comm_testany__set__result(simcall, cursor);
 479       synchro->simcalls.push_back(simcall);
 480       SIMIX_comm_finish(synchro);
 481       return;
 482     }
 483   }
 484   SIMIX_simcall_answer(simcall);
 485 }
 486
 487 void simcall_HANDLER_comm_waitany(smx_simcall_t simcall, xbt_dynar_t synchros)
 488 {
 489   smx_synchro_t synchro;
 490   unsigned int cursor = 0;
 491
 492   if (MC_is_active() || MC_record_replay_is_active()){
 493     int idx = SIMCALL_GET_MC_VALUE(simcall);
 494     synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
 495     synchro->simcalls.push_back(simcall);
 496     simcall_comm_waitany__set__result(simcall, idx);
 497     synchro->state = SIMIX_DONE;
 498     SIMIX_comm_finish(synchro);
 499     return;
 500   }
 501
 502   xbt_dynar_foreach(synchros, cursor, synchro){
 503     /* associate this simcall to the the synchro */
 504     synchro->simcalls.push_back(simcall);
 505
 506     /* see if the synchro is already finished */
 507     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING){
 508       SIMIX_comm_finish(synchro);
 509       break;
 510     }
 511   }
 512 }
 513
 514 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
 515 {
 516   smx_synchro_t synchro;
 517   unsigned int cursor = 0;
 518   xbt_dynar_t synchros = simcall_comm_waitany__get__comms(simcall);
 519
 520   xbt_dynar_foreach(synchros, cursor, synchro) {
 521     // Remove the first occurence of simcall:
 522     auto i = boost::range::find(synchro->simcalls, simcall);
 523     if (i !=  synchro->simcalls.end())
 524       synchro->simcalls.erase(i);
 525   }
 526 }
 527
 528 /**
 529  *  \brief Starts the simulation of a communication synchro.
 530  *  \param synchro the communication synchro
 531  */
 532 static inline void SIMIX_comm_start(smx_synchro_t synchro)
 533 {
 534   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 535
 536   /* If both the sender and the receiver are already there, start the communication */
 537   if (synchro->state == SIMIX_READY) {
 538
 539     sg_host_t sender   = comm->src_proc->host;
 540     sg_host_t receiver = comm->dst_proc->host;
 541
 542     XBT_DEBUG("Starting communication %p from '%s' to '%s'", synchro, sg_host_get_name(sender), sg_host_get_name(receiver));
 543
 544     comm->surf_comm = surf_network_model_communicate(surf_network_model, sender, receiver, comm->task_size, comm->rate);
 545     comm->surf_comm->setData(synchro);
 546     comm->state = SIMIX_RUNNING;
 547
 548     /* If a link is failed, detect it immediately */
 549     if (comm->surf_comm->getState() == simgrid::surf::Action::State::failed) {
 550       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
 551                 sg_host_get_name(sender), sg_host_get_name(receiver));
 552       comm->state = SIMIX_LINK_FAILURE;
 553       comm->cleanupSurf();
 554     }
 555
 556     /* If any of the process is suspend, create the synchro but stop its execution,
 557        it will be restarted when the sender process resume */
 558     if (SIMIX_process_is_suspended(comm->src_proc) || SIMIX_process_is_suspended(comm->dst_proc)) {
 559       if (SIMIX_process_is_suspended(comm->src_proc))
 560         XBT_DEBUG("The communication is suspended on startup because src (%s@%s) was suspended since it initiated the communication",
 561             comm->src_proc->name.c_str(), sg_host_get_name(comm->src_proc->host));
 562       else
 563         XBT_DEBUG("The communication is suspended on startup because dst (%s@%s) was suspended since it initiated the communication",
 564             comm->dst_proc->name.c_str(), sg_host_get_name(comm->dst_proc->host));
 565
 566       comm->surf_comm->suspend();
 567     }
 568   }
 569 }
 570
 571 /**
 572  * \brief Answers the SIMIX simcalls associated to a communication synchro.
 573  * \param synchro a finished communication synchro
 574  */
 575 void SIMIX_comm_finish(smx_synchro_t synchro)
 576 {
 577   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 578   unsigned int destroy_count = 0;
 579
 580   while (!synchro->simcalls.empty()) {
 581     smx_simcall_t simcall = synchro->simcalls.front();
 582     synchro->simcalls.pop_front();
 583
 584     /* If a waitany simcall is waiting for this synchro to finish, then remove
 585        it from the other synchros in the waitany list. Afterwards, get the
 586        position of the actual synchro in the waitany dynar and
 587        return it as the result of the simcall */
 588
 589     if (simcall->call == SIMCALL_NONE) //FIXME: maybe a better way to handle this case
 590       continue; // if process handling comm is killed
 591     if (simcall->call == SIMCALL_COMM_WAITANY) {
 592       SIMIX_waitany_remove_simcall_from_actions(simcall);
 593       if (!MC_is_active() && !MC_record_replay_is_active())
 594         simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro));
 595     }
 596
 597     /* If the synchro is still in a rendez-vous point then remove from it */
 598     if (comm->mbox)
 599       SIMIX_mbox_remove(comm->mbox, synchro);
 600
 601     XBT_DEBUG("SIMIX_comm_finish: synchro state = %d", (int)synchro->state);
 602
 603     /* Check out for errors */
 604
 605     if (simcall->issuer->host->isOff()) {
 606       simcall->issuer->context->iwannadie = 1;
 607       SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 608     } else {
 609       switch (synchro->state) {
 610
 611       case SIMIX_DONE:
 612         XBT_DEBUG("Communication %p complete!", synchro);
 613         SIMIX_comm_copy_data(synchro);
 614         break;
 615
 616       case SIMIX_SRC_TIMEOUT:
 617         SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of sender");
 618         break;
 619
 620       case SIMIX_DST_TIMEOUT:
 621         SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of receiver");
 622         break;
 623
 624       case SIMIX_SRC_HOST_FAILURE:
 625         if (simcall->issuer == comm->src_proc)
 626           simcall->issuer->context->iwannadie = 1;
 627   //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 628         else
 629           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 630         break;
 631
 632       case SIMIX_DST_HOST_FAILURE:
 633         if (simcall->issuer == comm->dst_proc)
 634           simcall->issuer->context->iwannadie = 1;
 635   //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 636         else
 637           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 638         break;
 639
 640       case SIMIX_LINK_FAILURE:
 641
 642         XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
 643                   synchro,
 644                   comm->src_proc ? sg_host_get_name(comm->src_proc->host) : nullptr,
 645                   comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : nullptr,
 646                   simcall->issuer->name.c_str(), simcall->issuer, comm->detached);
 647         if (comm->src_proc == simcall->issuer) {
 648           XBT_DEBUG("I'm source");
 649         } else if (comm->dst_proc == simcall->issuer) {
 650           XBT_DEBUG("I'm dest");
 651         } else {
 652           XBT_DEBUG("I'm neither source nor dest");
 653         }
 654         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
 655         break;
 656
 657       case SIMIX_CANCELED:
 658         if (simcall->issuer == comm->dst_proc)
 659           SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the sender");
 660         else
 661           SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the receiver");
 662         break;
 663
 664       default:
 665         xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state);
 666       }
 667     }
 668
 669     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
 670     if (simcall->issuer->exception) {
 671       // In order to modify the exception we have to rethrow it:
 672       try {
 673         std::rethrow_exception(simcall->issuer->exception);
 674       }
 675       catch(xbt_ex& e) {
 676         if (simcall->call == SIMCALL_COMM_WAITANY) {
 677           e.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro);
 678         }
 679         else if (simcall->call == SIMCALL_COMM_TESTANY) {
 680           e.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &synchro);
 681         }
 682         simcall->issuer->exception = std::make_exception_ptr(e);
 683       }
 684       catch(...) {
 685         // Nothing to do
 686       }
 687     }
 688
 689     if (simcall->issuer->host->isOff()) {
 690       simcall->issuer->context->iwannadie = 1;
 691     }
 692
 693     simcall->issuer->waiting_synchro = nullptr;
 694     xbt_fifo_remove(simcall->issuer->comms, synchro);
 695     if(comm->detached){
 696       if(simcall->issuer == comm->src_proc){
 697         if(comm->dst_proc)
 698           xbt_fifo_remove(comm->dst_proc->comms, synchro);
 699       }
 700       if(simcall->issuer == comm->dst_proc){
 701         if(comm->src_proc)
 702           xbt_fifo_remove(comm->src_proc->comms, synchro);
 703         //in case of a detached comm we have an extra ref to remove, as the sender won't do it
 704         destroy_count++;
 705       }
 706     }
 707     SIMIX_simcall_answer(simcall);
 708     destroy_count++;
 709   }
 710
 711   while (destroy_count-- > 0)
 712     static_cast<simgrid::simix::Comm*>(synchro)->unref();
 713 }
 714
 715 /******************************************************************************/
 716 /*                    SIMIX_comm_copy_data callbacks                       */
 717 /******************************************************************************/
 718 static void (*SIMIX_comm_copy_data_callback) (smx_synchro_t, void*, size_t) = &SIMIX_comm_copy_pointer_callback;
 719
 720 void SIMIX_comm_set_copy_data_callback(void (*callback) (smx_synchro_t, void*, size_t))
 721 {
 722   SIMIX_comm_copy_data_callback = callback;
 723 }
 724
 725 void SIMIX_comm_copy_pointer_callback(smx_synchro_t synchro, void* buff, size_t buff_size)
 726 {
 727   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 728
 729   xbt_assert((buff_size == sizeof(void *)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
 730   *(void **) (comm->dst_buff) = buff;
 731 }
 732
 733 void SIMIX_comm_copy_buffer_callback(smx_synchro_t synchro, void* buff, size_t buff_size)
 734 {
 735   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 736
 737   XBT_DEBUG("Copy the data over");
 738   memcpy(comm->dst_buff, buff, buff_size);
 739   if (comm->detached) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the original buffer available to the application ASAP
 740     xbt_free(buff);
 741     comm->src_buff = nullptr;
 742   }
 743 }
 744
 745
 746 /**
 747  *  \brief Copy the communication data from the sender's buffer to the receiver's one
 748  *  \param comm The communication
 749  */
 750 void SIMIX_comm_copy_data(smx_synchro_t synchro)
 751 {
 752   simgrid::simix::Comm *comm = static_cast<simgrid::simix::Comm*>(synchro);
 753
 754   size_t buff_size = comm->src_buff_size;
 755   /* If there is no data to copy then return */
 756   if (!comm->src_buff || !comm->dst_buff || comm->copied)
 757     return;
 758
 759   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
 760             comm,
 761             comm->src_proc ? sg_host_get_name(comm->src_proc->host) : "a finished process",
 762             comm->src_buff,
 763             comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : "a finished process",
 764             comm->dst_buff, buff_size);
 765
 766   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
 767   if (comm->dst_buff_size)
 768     buff_size = MIN(buff_size, *(comm->dst_buff_size));
 769
 770   /* Update the receiver's buffer size to the copied amount */
 771   if (comm->dst_buff_size)
 772     *comm->dst_buff_size = buff_size;
 773
 774   if (buff_size > 0){
 775       if(comm->copy_data_fun)
 776         comm->copy_data_fun (comm, comm->src_buff, buff_size);
 777       else
 778         SIMIX_comm_copy_data_callback (comm, comm->src_buff, buff_size);
 779   }
 780
 781
 782   /* Set the copied flag so we copy data only once */
 783   /* (this function might be called from both communication ends) */
 784   comm->copied = 1;
 785 }