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