src/simix/smx_network.cpp

   1 /* Copyright (c) 2009-2015. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "src/surf/surf_interface.hpp"
   8 #include "src/simix/smx_private.h"
   9 #include "xbt/log.h"
  10 #include "mc/mc.h"
  11 #include "src/mc/mc_replay.h"
  12 #include "xbt/dict.h"
  13 #include "simgrid/s4u/mailbox.hpp"
  14
  15 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization");
  16
  17 static void SIMIX_mbox_free(void *data);
  18 static xbt_dict_t mailboxes = xbt_dict_new_homogeneous(SIMIX_mbox_free);
  19
  20 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
  21 static void SIMIX_comm_copy_data(smx_synchro_t comm);
  22 static smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type);
  23 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t comm);
  24 static smx_synchro_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
  25                                         int (*match_fun)(void *, void *,smx_synchro_t),
  26                                         void *user_data, smx_synchro_t my_synchro);
  27 static smx_synchro_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
  28                                         int (*match_fun)(void *, void *,smx_synchro_t),
  29                                         void *user_data, smx_synchro_t my_synchro);
  30 static void SIMIX_comm_start(smx_synchro_t synchro);
  31
  32 void SIMIX_mailbox_exit(void)
  33 {
  34   xbt_dict_free(&mailboxes);
  35 }
  36
  37 /******************************************************************************/
  38 /*                           Rendez-Vous Points                               */
  39 /******************************************************************************/
  40
  41 smx_mailbox_t SIMIX_mbox_create(const char *name)
  42 {
  43   xbt_assert(name, "Mailboxes must have a name");
  44   /* two processes may have pushed the same mbox_create simcall at the same time */
  45   smx_mailbox_t mbox = (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
  46
  47   if (!mbox) {
  48     mbox = xbt_new0(s_smx_mailbox_t, 1);
  49     mbox->name = xbt_strdup(name);
  50     mbox->comm_fifo = xbt_fifo_new();
  51     mbox->done_comm_fifo = xbt_fifo_new();
  52     mbox->permanent_receiver=NULL;
  53
  54     XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name);
  55     xbt_dict_set(mailboxes, mbox->name, mbox, NULL);
  56   }
  57   return mbox;
  58 }
  59
  60 void SIMIX_mbox_free(void *data)
  61 {
  62   XBT_DEBUG("mbox free %p", data);
  63   smx_mailbox_t mbox = (smx_mailbox_t) data;
  64   xbt_free(mbox->name);
  65   xbt_fifo_free(mbox->comm_fifo);
  66   xbt_fifo_free(mbox->done_comm_fifo);
  67
  68   xbt_free(mbox);
  69 }
  70
  71 smx_mailbox_t SIMIX_mbox_get_by_name(const char *name)
  72 {
  73   return (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
  74 }
  75
  76 smx_synchro_t SIMIX_mbox_get_head(smx_mailbox_t mbox)
  77 {
  78   return (smx_synchro_t) xbt_fifo_get_item_content(
  79     xbt_fifo_get_first_item(mbox->comm_fifo));
  80 }
  81
  82 /**
  83  *  \brief get the receiver (process associated to the mailbox)
  84  *  \param mbox The rendez-vous point
  85  *  \return process The receiving process (NULL if not set)
  86  */
  87 smx_process_t SIMIX_mbox_get_receiver(smx_mailbox_t mbox)
  88 {
  89   return mbox->permanent_receiver;
  90 }
  91
  92 /**
  93  *  \brief set the receiver of the rendez vous point to allow eager sends
  94  *  \param mbox The rendez-vous point
  95  *  \param process The receiving process
  96  */
  97 void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_process_t process)
  98 {
  99   mbox->permanent_receiver=process;
 100 }
 101
 102 /**
 103  *  \brief Pushes a communication synchro into a rendez-vous point
 104  *  \param mbox The mailbox
 105  *  \param comm The communication synchro
 106  */
 107 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t comm)
 108 {
 109   xbt_fifo_push(mbox->comm_fifo, comm);
 110   comm->comm.mbox = mbox;
 111 }
 112
 113 /**
 114  *  \brief Removes a communication synchro from a rendez-vous point
 115  *  \param mbox The rendez-vous point
 116  *  \param comm The communication synchro
 117  */
 118 void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_synchro_t comm)
 119 {
 120   xbt_fifo_remove(mbox->comm_fifo, comm);
 121   comm->comm.mbox = NULL;
 122 }
 123
 124 /**
 125  *  \brief Checks if there is a communication synchro queued in a fifo matching our needs
 126  *  \param type The type of communication we are looking for (comm_send, comm_recv)
 127  *  \return The communication synchro if found, NULL otherwise
 128  */
 129 smx_synchro_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
 130                                  int (*match_fun)(void *, void *,smx_synchro_t),
 131                                  void *this_user_data, smx_synchro_t my_synchro)
 132 {
 133   smx_synchro_t synchro;
 134   xbt_fifo_item_t item;
 135   void* other_user_data = NULL;
 136
 137   xbt_fifo_foreach(fifo, item, synchro, smx_synchro_t) {
 138     if (synchro->comm.type == SIMIX_COMM_SEND) {
 139       other_user_data = synchro->comm.src_data;
 140     } else if (synchro->comm.type == SIMIX_COMM_RECEIVE) {
 141       other_user_data = synchro->comm.dst_data;
 142     }
 143     if (synchro->comm.type == type &&
 144         (!match_fun              ||              match_fun(this_user_data,  other_user_data, synchro)) &&
 145         (!synchro->comm.match_fun || synchro->comm.match_fun(other_user_data, this_user_data,  my_synchro))) {
 146       XBT_DEBUG("Found a matching communication synchro %p", synchro);
 147       xbt_fifo_remove_item(fifo, item);
 148       xbt_fifo_free_item(item);
 149       synchro->comm.refcount++;
 150 #if HAVE_MC
 151       synchro->comm.mbox_cpy = synchro->comm.mbox;
 152 #endif
 153       synchro->comm.mbox = NULL;
 154       return synchro;
 155     }
 156     XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
 157               " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
 158               synchro, (int)synchro->comm.type, (int)type);
 159   }
 160   XBT_DEBUG("No matching communication synchro found");
 161   return NULL;
 162 }
 163
 164
 165 /**
 166  *  \brief Checks if there is a communication synchro queued in a fifo matching our needs, but leave it there
 167  *  \param type The type of communication we are looking for (comm_send, comm_recv)
 168  *  \return The communication synchro if found, NULL otherwise
 169  */
 170 smx_synchro_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
 171                                  int (*match_fun)(void *, void *,smx_synchro_t),
 172                                  void *this_user_data, smx_synchro_t my_synchro)
 173 {
 174   smx_synchro_t synchro;
 175   xbt_fifo_item_t item;
 176   void* other_user_data = NULL;
 177
 178   xbt_fifo_foreach(fifo, item, synchro, smx_synchro_t) {
 179     if (synchro->comm.type == SIMIX_COMM_SEND) {
 180       other_user_data = synchro->comm.src_data;
 181     } else if (synchro->comm.type == SIMIX_COMM_RECEIVE) {
 182       other_user_data = synchro->comm.dst_data;
 183     }
 184     if (synchro->comm.type == type &&
 185         (!match_fun              ||              match_fun(this_user_data,  other_user_data, synchro)) &&
 186         (!synchro->comm.match_fun || synchro->comm.match_fun(other_user_data, this_user_data,  my_synchro))) {
 187       XBT_DEBUG("Found a matching communication synchro %p", synchro);
 188       synchro->comm.refcount++;
 189
 190       return synchro;
 191     }
 192     XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
 193               " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
 194               synchro, (int)synchro->comm.type, (int)type);
 195   }
 196   XBT_DEBUG("No matching communication synchro found");
 197   return NULL;
 198 }
 199 /******************************************************************************/
 200 /*                          Communication synchros                            */
 201 /******************************************************************************/
 202
 203 /**
 204  *  \brief Creates a new communicate synchro
 205  *  \param type The direction of communication (comm_send, comm_recv)
 206  *  \return The new communicate synchro
 207  */
 208 smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type)
 209 {
 210   smx_synchro_t synchro;
 211
 212   /* alloc structures */
 213   synchro = (smx_synchro_t) xbt_mallocator_get(simix_global->synchro_mallocator);
 214
 215   synchro->type = SIMIX_SYNC_COMMUNICATE;
 216   synchro->state = SIMIX_WAITING;
 217
 218   /* set communication */
 219   synchro->comm.type = type;
 220   synchro->comm.refcount = 1;
 221   synchro->comm.src_data=NULL;
 222   synchro->comm.dst_data=NULL;
 223
 224   synchro->category = NULL;
 225
 226   XBT_DEBUG("Create communicate synchro %p", synchro);
 227
 228   return synchro;
 229 }
 230
 231 /**
 232  *  \brief Destroy a communicate synchro
 233  *  \param synchro The communicate synchro to be destroyed
 234  */
 235 void SIMIX_comm_destroy(smx_synchro_t synchro)
 236 {
 237   XBT_DEBUG("Destroy synchro %p (refcount: %d), state: %d",
 238             synchro, synchro->comm.refcount, (int)synchro->state);
 239
 240   if (synchro->comm.refcount <= 0) {
 241     xbt_backtrace_display_current();
 242     xbt_die("The refcount of comm %p is already 0 before decreasing it. "
 243             "That's a bug! If you didn't test and/or wait the same communication twice in your code, then the bug is SimGrid's...", synchro);
 244   }
 245   synchro->comm.refcount--;
 246   if (synchro->comm.refcount > 0)
 247       return;
 248   XBT_DEBUG("Really free communication %p; refcount is now %d", synchro,
 249       synchro->comm.refcount);
 250
 251   xbt_free(synchro->name);
 252   SIMIX_comm_destroy_internal_actions(synchro);
 253
 254   if (synchro->comm.detached && synchro->state != SIMIX_DONE) {
 255     /* the communication has failed and was detached:
 256      * we have to free the buffer */
 257     if (synchro->comm.clean_fun) {
 258       synchro->comm.clean_fun(synchro->comm.src_buff);
 259     }
 260     synchro->comm.src_buff = NULL;
 261   }
 262
 263   if(synchro->comm.mbox)
 264     SIMIX_mbox_remove(synchro->comm.mbox, synchro);
 265
 266   xbt_mallocator_release(simix_global->synchro_mallocator, synchro);
 267 }
 268
 269 void SIMIX_comm_destroy_internal_actions(smx_synchro_t synchro)
 270 {
 271   if (synchro->comm.surf_comm){
 272     synchro->comm.surf_comm->unref();
 273     synchro->comm.surf_comm = NULL;
 274   }
 275
 276   if (synchro->comm.src_timeout){
 277     synchro->comm.src_timeout->unref();
 278     synchro->comm.src_timeout = NULL;
 279   }
 280
 281   if (synchro->comm.dst_timeout){
 282     synchro->comm.dst_timeout->unref();
 283     synchro->comm.dst_timeout = NULL;
 284   }
 285 }
 286
 287 void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_process_t src, smx_mailbox_t mbox,
 288                                   double task_size, double rate,
 289                                   void *src_buff, size_t src_buff_size,
 290                                   int (*match_fun)(void *, void *,smx_synchro_t),
 291                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 292           void *data, double timeout){
 293   smx_synchro_t comm = simcall_HANDLER_comm_isend(simcall, src, mbox, task_size, rate,
 294                            src_buff, src_buff_size, match_fun, NULL, copy_data_fun,
 295                data, 0);
 296   SIMCALL_SET_MC_VALUE(simcall, 0);
 297   simcall_HANDLER_comm_wait(simcall, comm, timeout);
 298 }
 299 smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_process_t src_proc, smx_mailbox_t mbox,
 300                                   double task_size, double rate,
 301                                   void *src_buff, size_t src_buff_size,
 302                                   int (*match_fun)(void *, void *,smx_synchro_t),
 303                                   void (*clean_fun)(void *), // used to free the synchro in case of problem after a detached send
 304                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),// used to copy data if not default one
 305                           void *data, int detached)
 306 {
 307   XBT_DEBUG("send from %p", mbox);
 308
 309   /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
 310   smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_SEND);
 311
 312   /* Look for communication synchro matching our needs. We also provide a description of
 313    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 314    *
 315    * If it is not found then push our communication into the rendez-vous point */
 316   smx_synchro_t other_synchro = SIMIX_fifo_get_comm(mbox->comm_fifo, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro);
 317
 318   if (!other_synchro) {
 319     other_synchro = this_synchro;
 320
 321     if (mbox->permanent_receiver!=NULL){
 322       //this mailbox is for small messages, which have to be sent right now
 323       other_synchro->state = SIMIX_READY;
 324       other_synchro->comm.dst_proc=mbox->permanent_receiver;
 325       other_synchro->comm.refcount++;
 326       xbt_fifo_push(mbox->done_comm_fifo,other_synchro);
 327       other_synchro->comm.mbox=mbox;
 328       XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", mbox, &(other_synchro->comm));
 329
 330     }else{
 331       SIMIX_mbox_push(mbox, this_synchro);
 332     }
 333   } else {
 334     XBT_DEBUG("Receive already pushed");
 335
 336     SIMIX_comm_destroy(this_synchro);
 337
 338     other_synchro->state = SIMIX_READY;
 339     other_synchro->comm.type = SIMIX_COMM_READY;
 340
 341   }
 342   xbt_fifo_push(src_proc->comms, other_synchro);
 343
 344   /* if the communication synchro is detached then decrease the refcount
 345    * by one, so it will be eliminated by the receiver's destroy call */
 346   if (detached) {
 347     other_synchro->comm.detached = 1;
 348     other_synchro->comm.refcount--;
 349     other_synchro->comm.clean_fun = clean_fun;
 350   } else {
 351     other_synchro->comm.clean_fun = NULL;
 352   }
 353
 354   /* Setup the communication synchro */
 355   other_synchro->comm.src_proc = src_proc;
 356   other_synchro->comm.task_size = task_size;
 357   other_synchro->comm.rate = rate;
 358   other_synchro->comm.src_buff = src_buff;
 359   other_synchro->comm.src_buff_size = src_buff_size;
 360   other_synchro->comm.src_data = data;
 361
 362   other_synchro->comm.match_fun = match_fun;
 363   other_synchro->comm.copy_data_fun = copy_data_fun;
 364
 365
 366   if (MC_is_active() || MC_record_replay_is_active()) {
 367     other_synchro->state = SIMIX_RUNNING;
 368     return (detached ? NULL : other_synchro);
 369   }
 370
 371   SIMIX_comm_start(other_synchro);
 372   return (detached ? NULL : other_synchro);
 373 }
 374
 375 void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
 376                          void *dst_buff, size_t *dst_buff_size,
 377                          int (*match_fun)(void *, void *, smx_synchro_t),
 378                          void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 379                          void *data, double timeout, double rate)
 380 {
 381   smx_synchro_t comm = SIMIX_comm_irecv(receiver, mbox, dst_buff,
 382                            dst_buff_size, match_fun, copy_data_fun, data, rate);
 383   SIMCALL_SET_MC_VALUE(simcall, 0);
 384   simcall_HANDLER_comm_wait(simcall, comm, timeout);
 385 }
 386
 387 smx_synchro_t simcall_HANDLER_comm_irecv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
 388     void *dst_buff, size_t *dst_buff_size,
 389     int (*match_fun)(void *, void *, smx_synchro_t),
 390     void (*copy_data_fun)(smx_synchro_t, void*, size_t),
 391     void *data, double rate)
 392 {
 393   return SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
 394 }
 395
 396 smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void *dst_buff, size_t *dst_buff_size,
 397     int (*match_fun)(void *, void *, smx_synchro_t),
 398     void (*copy_data_fun)(smx_synchro_t, void*, size_t), // used to copy data if not default one
 399     void *data, double rate)
 400 {
 401   XBT_DEBUG("recv from %p %p", mbox, mbox->comm_fifo);
 402   smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
 403
 404   smx_synchro_t other_synchro;
 405   //communication already done, get it inside the fifo of completed comms
 406   if (mbox->permanent_receiver && xbt_fifo_size(mbox->done_comm_fifo)!=0) {
 407
 408     XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
 409     //find a match in the already received fifo
 410     other_synchro = SIMIX_fifo_get_comm(mbox->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_synchro);
 411     //if not found, assume the receiver came first, register it to the mailbox in the classical way
 412     if (!other_synchro)  {
 413       XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo");
 414       other_synchro = this_synchro;
 415       SIMIX_mbox_push(mbox, this_synchro);
 416     } else {
 417       if(other_synchro->comm.surf_comm && SIMIX_comm_get_remains(other_synchro)==0.0) {
 418         XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",&(other_synchro->comm));
 419         other_synchro->state = SIMIX_DONE;
 420         other_synchro->comm.type = SIMIX_COMM_DONE;
 421         other_synchro->comm.mbox = NULL;
 422       }
 423       other_synchro->comm.refcount--;
 424       SIMIX_comm_destroy(this_synchro);
 425     }
 426   } else {
 427     /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
 428
 429     /* Look for communication synchro matching our needs. We also provide a description of
 430      * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 431      *
 432      * If it is not found then push our communication into the rendez-vous point */
 433     other_synchro = SIMIX_fifo_get_comm(mbox->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_synchro);
 434
 435     if (!other_synchro) {
 436       XBT_DEBUG("Receive pushed first %d", xbt_fifo_size(mbox->comm_fifo));
 437       other_synchro = this_synchro;
 438       SIMIX_mbox_push(mbox, this_synchro);
 439     } else {
 440       SIMIX_comm_destroy(this_synchro);
 441       other_synchro->state = SIMIX_READY;
 442       other_synchro->comm.type = SIMIX_COMM_READY;
 443       //other_synchro->comm.refcount--;
 444     }
 445     xbt_fifo_push(dst_proc->comms, other_synchro);
 446   }
 447
 448   /* Setup communication synchro */
 449   other_synchro->comm.dst_proc = dst_proc;
 450   other_synchro->comm.dst_buff = dst_buff;
 451   other_synchro->comm.dst_buff_size = dst_buff_size;
 452   other_synchro->comm.dst_data = data;
 453
 454   if (rate != -1.0 && (other_synchro->comm.rate == -1.0 || rate < other_synchro->comm.rate))
 455     other_synchro->comm.rate = rate;
 456
 457   other_synchro->comm.match_fun = match_fun;
 458   other_synchro->comm.copy_data_fun = copy_data_fun;
 459
 460   if (MC_is_active() || MC_record_replay_is_active()) {
 461     other_synchro->state = SIMIX_RUNNING;
 462     return other_synchro;
 463   }
 464
 465   SIMIX_comm_start(other_synchro);
 466   return other_synchro;
 467 }
 468
 469 smx_synchro_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t mbox,
 470                                    int type, int src, int tag,
 471                                    int (*match_fun)(void *, void *, smx_synchro_t),
 472                                    void *data){
 473   return SIMIX_comm_iprobe(simcall->issuer, mbox, type, src, tag, match_fun, data);
 474 }
 475
 476 smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t mbox, int type, int src,
 477                               int tag, int (*match_fun)(void *, void *, smx_synchro_t), void *data)
 478 {
 479   XBT_DEBUG("iprobe from %p %p", mbox, mbox->comm_fifo);
 480   smx_synchro_t this_synchro;
 481   int smx_type;
 482   if(type == 1){
 483     this_synchro=SIMIX_comm_new(SIMIX_COMM_SEND);
 484     smx_type = SIMIX_COMM_RECEIVE;
 485   } else{
 486     this_synchro=SIMIX_comm_new(SIMIX_COMM_RECEIVE);
 487     smx_type = SIMIX_COMM_SEND;
 488   }
 489   smx_synchro_t other_synchro=NULL;
 490   if(mbox->permanent_receiver && xbt_fifo_size(mbox->done_comm_fifo)!=0){
 491     //find a match in the already received fifo
 492       XBT_DEBUG("first try in the perm recv mailbox");
 493
 494     other_synchro = SIMIX_fifo_probe_comm(
 495       mbox->done_comm_fifo, (e_smx_comm_type_t) smx_type,
 496       match_fun, data, this_synchro);
 497   }
 498  // }else{
 499     if(!other_synchro){
 500         XBT_DEBUG("try in the normal mailbox");
 501         other_synchro = SIMIX_fifo_probe_comm(
 502           mbox->comm_fifo, (e_smx_comm_type_t) smx_type,
 503           match_fun, data, this_synchro);
 504     }
 505 //  }
 506   if(other_synchro)other_synchro->comm.refcount--;
 507
 508   SIMIX_comm_destroy(this_synchro);
 509   return other_synchro;
 510 }
 511
 512 void simcall_HANDLER_comm_wait(smx_simcall_t simcall, smx_synchro_t synchro, double timeout)
 513 {
 514   /* the simcall may be a wait, a send or a recv */
 515   surf_action_t sleep;
 516
 517   /* Associate this simcall to the wait synchro */
 518   XBT_DEBUG("simcall_HANDLER_comm_wait, %p", synchro);
 519
 520   xbt_fifo_push(synchro->simcalls, simcall);
 521   simcall->issuer->waiting_synchro = synchro;
 522
 523   if (MC_is_active() || MC_record_replay_is_active()) {
 524     int idx = SIMCALL_GET_MC_VALUE(simcall);
 525     if (idx == 0) {
 526       synchro->state = SIMIX_DONE;
 527     } else {
 528       /* If we reached this point, the wait simcall must have a timeout */
 529       /* Otherwise it shouldn't be enabled and executed by the MC */
 530       if (timeout == -1)
 531         THROW_IMPOSSIBLE;
 532
 533       if (synchro->comm.src_proc == simcall->issuer)
 534         synchro->state = SIMIX_SRC_TIMEOUT;
 535       else
 536         synchro->state = SIMIX_DST_TIMEOUT;
 537     }
 538
 539     SIMIX_comm_finish(synchro);
 540     return;
 541   }
 542
 543   /* If the synchro has already finish perform the error handling, */
 544   /* otherwise set up a waiting timeout on the right side          */
 545   if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
 546     SIMIX_comm_finish(synchro);
 547   } 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 */
 548     sleep = surf_host_sleep(simcall->issuer->host, timeout);
 549     sleep->setData(synchro);
 550
 551     if (simcall->issuer == synchro->comm.src_proc)
 552       synchro->comm.src_timeout = sleep;
 553     else
 554       synchro->comm.dst_timeout = sleep;
 555   }
 556 }
 557
 558 void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_synchro_t synchro)
 559 {
 560   if(MC_is_active() || MC_record_replay_is_active()){
 561     simcall_comm_test__set__result(simcall, synchro->comm.src_proc && synchro->comm.dst_proc);
 562     if(simcall_comm_test__get__result(simcall)){
 563       synchro->state = SIMIX_DONE;
 564       xbt_fifo_push(synchro->simcalls, simcall);
 565       SIMIX_comm_finish(synchro);
 566     }else{
 567       SIMIX_simcall_answer(simcall);
 568     }
 569     return;
 570   }
 571
 572   simcall_comm_test__set__result(simcall, (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING));
 573   if (simcall_comm_test__get__result(simcall)) {
 574     xbt_fifo_push(synchro->simcalls, simcall);
 575     SIMIX_comm_finish(synchro);
 576   } else {
 577     SIMIX_simcall_answer(simcall);
 578   }
 579 }
 580
 581 void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros)
 582 {
 583   unsigned int cursor;
 584   smx_synchro_t synchro;
 585   simcall_comm_testany__set__result(simcall, -1);
 586
 587   if (MC_is_active() || MC_record_replay_is_active()){
 588     int idx = SIMCALL_GET_MC_VALUE(simcall);
 589     if(idx == -1){
 590       SIMIX_simcall_answer(simcall);
 591     }else{
 592       synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
 593       simcall_comm_testany__set__result(simcall, idx);
 594       xbt_fifo_push(synchro->simcalls, simcall);
 595       synchro->state = SIMIX_DONE;
 596       SIMIX_comm_finish(synchro);
 597     }
 598     return;
 599   }
 600
 601   xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,synchro) {
 602     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
 603       simcall_comm_testany__set__result(simcall, cursor);
 604       xbt_fifo_push(synchro->simcalls, simcall);
 605       SIMIX_comm_finish(synchro);
 606       return;
 607     }
 608   }
 609   SIMIX_simcall_answer(simcall);
 610 }
 611
 612 void simcall_HANDLER_comm_waitany(smx_simcall_t simcall, xbt_dynar_t synchros)
 613 {
 614   smx_synchro_t synchro;
 615   unsigned int cursor = 0;
 616
 617   if (MC_is_active() || MC_record_replay_is_active()){
 618     int idx = SIMCALL_GET_MC_VALUE(simcall);
 619     synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
 620     xbt_fifo_push(synchro->simcalls, simcall);
 621     simcall_comm_waitany__set__result(simcall, idx);
 622     synchro->state = SIMIX_DONE;
 623     SIMIX_comm_finish(synchro);
 624     return;
 625   }
 626
 627   xbt_dynar_foreach(synchros, cursor, synchro){
 628     /* associate this simcall to the the synchro */
 629     xbt_fifo_push(synchro->simcalls, simcall);
 630
 631     /* see if the synchro is already finished */
 632     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING){
 633       SIMIX_comm_finish(synchro);
 634       break;
 635     }
 636   }
 637 }
 638
 639 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
 640 {
 641   smx_synchro_t synchro;
 642   unsigned int cursor = 0;
 643   xbt_dynar_t synchros = simcall_comm_waitany__get__comms(simcall);
 644
 645   xbt_dynar_foreach(synchros, cursor, synchro) {
 646     xbt_fifo_remove(synchro->simcalls, simcall);
 647   }
 648 }
 649
 650 /**
 651  *  \brief Starts the simulation of a communication synchro.
 652  *  \param synchro the communication synchro
 653  */
 654 static inline void SIMIX_comm_start(smx_synchro_t synchro)
 655 {
 656   /* If both the sender and the receiver are already there, start the communication */
 657   if (synchro->state == SIMIX_READY) {
 658
 659     sg_host_t sender = synchro->comm.src_proc->host;
 660     sg_host_t receiver = synchro->comm.dst_proc->host;
 661
 662     XBT_DEBUG("Starting communication %p from '%s' to '%s'", synchro,
 663               sg_host_get_name(sender), sg_host_get_name(receiver));
 664
 665     synchro->comm.surf_comm = surf_network_model_communicate(surf_network_model,
 666                                                             sender, receiver,
 667                                                             synchro->comm.task_size, synchro->comm.rate);
 668
 669     synchro->comm.surf_comm->setData(synchro);
 670
 671     synchro->state = SIMIX_RUNNING;
 672
 673     /* If a link is failed, detect it immediately */
 674     if (synchro->comm.surf_comm->getState() == simgrid::surf::Action::State::failed) {
 675       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
 676                 sg_host_get_name(sender), sg_host_get_name(receiver));
 677       synchro->state = SIMIX_LINK_FAILURE;
 678       SIMIX_comm_destroy_internal_actions(synchro);
 679     }
 680
 681     /* If any of the process is suspend, create the synchro but stop its execution,
 682        it will be restarted when the sender process resume */
 683     if (SIMIX_process_is_suspended(synchro->comm.src_proc) ||
 684         SIMIX_process_is_suspended(synchro->comm.dst_proc)) {
 685       /* FIXME: check what should happen with the synchro state */
 686
 687       if (SIMIX_process_is_suspended(synchro->comm.src_proc))
 688         XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
 689                   sg_host_get_name(synchro->comm.src_proc->host), synchro->comm.src_proc->name);
 690       else
 691         XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
 692                   sg_host_get_name(synchro->comm.dst_proc->host), synchro->comm.dst_proc->name);
 693
 694       synchro->comm.surf_comm->suspend();
 695
 696     }
 697   }
 698 }
 699
 700 /**
 701  * \brief Answers the SIMIX simcalls associated to a communication synchro.
 702  * \param synchro a finished communication synchro
 703  */
 704 void SIMIX_comm_finish(smx_synchro_t synchro)
 705 {
 706   unsigned int destroy_count = 0;
 707   smx_simcall_t simcall;
 708
 709   while ((simcall = (smx_simcall_t) xbt_fifo_shift(synchro->simcalls))) {
 710
 711     /* If a waitany simcall is waiting for this synchro to finish, then remove
 712        it from the other synchros in the waitany list. Afterwards, get the
 713        position of the actual synchro in the waitany dynar and
 714        return it as the result of the simcall */
 715
 716     if (simcall->call == SIMCALL_NONE) //FIXME: maybe a better way to handle this case
 717       continue; // if process handling comm is killed
 718     if (simcall->call == SIMCALL_COMM_WAITANY) {
 719       SIMIX_waitany_remove_simcall_from_actions(simcall);
 720       if (!MC_is_active() && !MC_record_replay_is_active())
 721         simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro));
 722     }
 723
 724     /* If the synchro is still in a rendez-vous point then remove from it */
 725     if (synchro->comm.mbox)
 726       SIMIX_mbox_remove(synchro->comm.mbox, synchro);
 727
 728     XBT_DEBUG("SIMIX_comm_finish: synchro state = %d", (int)synchro->state);
 729
 730     /* Check out for errors */
 731
 732     if (simcall->issuer->host->isOff()) {
 733       simcall->issuer->context->iwannadie = 1;
 734       SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 735     } else
 736
 737     switch (synchro->state) {
 738
 739     case SIMIX_DONE:
 740       XBT_DEBUG("Communication %p complete!", synchro);
 741       SIMIX_comm_copy_data(synchro);
 742       break;
 743
 744     case SIMIX_SRC_TIMEOUT:
 745       SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
 746                     "Communication timeouted because of sender");
 747       break;
 748
 749     case SIMIX_DST_TIMEOUT:
 750       SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
 751                     "Communication timeouted because of receiver");
 752       break;
 753
 754     case SIMIX_SRC_HOST_FAILURE:
 755       if (simcall->issuer == synchro->comm.src_proc)
 756         simcall->issuer->context->iwannadie = 1;
 757 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 758       else
 759         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 760       break;
 761
 762     case SIMIX_DST_HOST_FAILURE:
 763       if (simcall->issuer == synchro->comm.dst_proc)
 764         simcall->issuer->context->iwannadie = 1;
 765 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 766       else
 767         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 768       break;
 769
 770     case SIMIX_LINK_FAILURE:
 771
 772       XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
 773                 synchro,
 774                 synchro->comm.src_proc ? sg_host_get_name(synchro->comm.src_proc->host) : NULL,
 775                 synchro->comm.dst_proc ? sg_host_get_name(synchro->comm.dst_proc->host) : NULL,
 776                 simcall->issuer->name, simcall->issuer, synchro->comm.detached);
 777       if (synchro->comm.src_proc == simcall->issuer) {
 778         XBT_DEBUG("I'm source");
 779       } else if (synchro->comm.dst_proc == simcall->issuer) {
 780         XBT_DEBUG("I'm dest");
 781       } else {
 782         XBT_DEBUG("I'm neither source nor dest");
 783       }
 784       SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
 785       break;
 786
 787     case SIMIX_CANCELED:
 788       if (simcall->issuer == synchro->comm.dst_proc)
 789         SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
 790                       "Communication canceled by the sender");
 791       else
 792         SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
 793                       "Communication canceled by the receiver");
 794       break;
 795
 796     default:
 797       xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state);
 798     }
 799
 800     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
 801     if (simcall->issuer->doexception) {
 802       if (simcall->call == SIMCALL_COMM_WAITANY) {
 803         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro);
 804       }
 805       else if (simcall->call == SIMCALL_COMM_TESTANY) {
 806         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &synchro);
 807       }
 808     }
 809
 810     if (simcall->issuer->host->isOff()) {
 811       simcall->issuer->context->iwannadie = 1;
 812     }
 813
 814     simcall->issuer->waiting_synchro = NULL;
 815     xbt_fifo_remove(simcall->issuer->comms, synchro);
 816     if(synchro->comm.detached){
 817       if(simcall->issuer == synchro->comm.src_proc){
 818         if(synchro->comm.dst_proc)
 819           xbt_fifo_remove(synchro->comm.dst_proc->comms, synchro);
 820       }
 821       if(simcall->issuer == synchro->comm.dst_proc){
 822         if(synchro->comm.src_proc)
 823           xbt_fifo_remove(synchro->comm.src_proc->comms, synchro);
 824       }
 825     }
 826     SIMIX_simcall_answer(simcall);
 827     destroy_count++;
 828   }
 829
 830   while (destroy_count-- > 0)
 831     SIMIX_comm_destroy(synchro);
 832 }
 833
 834 /**
 835  * \brief This function is called when a Surf communication synchro is finished.
 836  * \param synchro the corresponding Simix communication
 837  */
 838 void SIMIX_post_comm(smx_synchro_t synchro)
 839 {
 840   /* Update synchro state */
 841   if (synchro->comm.src_timeout &&
 842       synchro->comm.src_timeout->getState() == simgrid::surf::Action::State::done)
 843     synchro->state = SIMIX_SRC_TIMEOUT;
 844   else if (synchro->comm.dst_timeout &&
 845     synchro->comm.dst_timeout->getState() == simgrid::surf::Action::State::done)
 846     synchro->state = SIMIX_DST_TIMEOUT;
 847   else if (synchro->comm.src_timeout &&
 848     synchro->comm.src_timeout->getState() == simgrid::surf::Action::State::failed)
 849     synchro->state = SIMIX_SRC_HOST_FAILURE;
 850   else if (synchro->comm.dst_timeout &&
 851       synchro->comm.dst_timeout->getState() == simgrid::surf::Action::State::failed)
 852     synchro->state = SIMIX_DST_HOST_FAILURE;
 853   else if (synchro->comm.surf_comm &&
 854     synchro->comm.surf_comm->getState() == simgrid::surf::Action::State::failed) {
 855     XBT_DEBUG("Puta madre. Surf says that the link broke");
 856     synchro->state = SIMIX_LINK_FAILURE;
 857   } else
 858     synchro->state = SIMIX_DONE;
 859
 860   XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
 861             synchro, (int)synchro->state, synchro->comm.src_proc, synchro->comm.dst_proc, synchro->comm.detached);
 862
 863   /* destroy the surf actions associated with the Simix communication */
 864   SIMIX_comm_destroy_internal_actions(synchro);
 865
 866   /* if there are simcalls associated with the synchro, then answer them */
 867   if (xbt_fifo_size(synchro->simcalls)) {
 868     SIMIX_comm_finish(synchro);
 869   }
 870 }
 871
 872 void SIMIX_comm_cancel(smx_synchro_t synchro)
 873 {
 874   /* if the synchro is a waiting state means that it is still in a mbox */
 875   /* so remove from it and delete it */
 876   if (synchro->state == SIMIX_WAITING) {
 877     SIMIX_mbox_remove(synchro->comm.mbox, synchro);
 878     synchro->state = SIMIX_CANCELED;
 879   }
 880   else if (!MC_is_active() /* when running the MC there are no surf actions */
 881            && !MC_record_replay_is_active()
 882            && (synchro->state == SIMIX_READY || synchro->state == SIMIX_RUNNING)) {
 883
 884     synchro->comm.surf_comm->cancel();
 885   }
 886 }
 887
 888 void SIMIX_comm_suspend(smx_synchro_t synchro)
 889 {
 890   /*FIXME: shall we suspend also the timeout synchro? */
 891   if (synchro->comm.surf_comm)
 892     synchro->comm.surf_comm->suspend();
 893   /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
 894 }
 895
 896 void SIMIX_comm_resume(smx_synchro_t synchro)
 897 {
 898   /*FIXME: check what happen with the timeouts */
 899   if (synchro->comm.surf_comm)
 900     synchro->comm.surf_comm->resume();
 901   /* in the other case, the synchro were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
 902 }
 903
 904
 905 /************* synchro Getters **************/
 906
 907 /**
 908  *  \brief get the amount remaining from the communication
 909  *  \param synchro The communication
 910  */
 911 double SIMIX_comm_get_remains(smx_synchro_t synchro)
 912 {
 913   double remains;
 914
 915   if(!synchro){
 916     return 0;
 917   }
 918
 919   switch (synchro->state) {
 920
 921   case SIMIX_RUNNING:
 922     remains = synchro->comm.surf_comm->getRemains();
 923     break;
 924
 925   case SIMIX_WAITING:
 926   case SIMIX_READY:
 927     remains = 0; /*FIXME: check what should be returned */
 928     break;
 929
 930   default:
 931     remains = 0; /*FIXME: is this correct? */
 932     break;
 933   }
 934   return remains;
 935 }
 936
 937 e_smx_state_t SIMIX_comm_get_state(smx_synchro_t synchro)
 938 {
 939   return synchro->state;
 940 }
 941
 942 /**
 943  *  \brief Return the user data associated to the sender of the communication
 944  *  \param synchro The communication
 945  *  \return the user data
 946  */
 947 void* SIMIX_comm_get_src_data(smx_synchro_t synchro)
 948 {
 949   return synchro->comm.src_data;
 950 }
 951
 952 /**
 953  *  \brief Return the user data associated to the receiver of the communication
 954  *  \param synchro The communication
 955  *  \return the user data
 956  */
 957 void* SIMIX_comm_get_dst_data(smx_synchro_t synchro)
 958 {
 959   return synchro->comm.dst_data;
 960 }
 961
 962 smx_process_t SIMIX_comm_get_src_proc(smx_synchro_t synchro)
 963 {
 964   return synchro->comm.src_proc;
 965 }
 966
 967 smx_process_t SIMIX_comm_get_dst_proc(smx_synchro_t synchro)
 968 {
 969   return synchro->comm.dst_proc;
 970 }
 971
 972 /******************************************************************************/
 973 /*                    SIMIX_comm_copy_data callbacks                       */
 974 /******************************************************************************/
 975 static void (*SIMIX_comm_copy_data_callback) (smx_synchro_t, void*, size_t) =
 976   &SIMIX_comm_copy_pointer_callback;
 977
 978 void
 979 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_synchro_t, void*, size_t))
 980 {
 981   SIMIX_comm_copy_data_callback = callback;
 982 }
 983
 984 void SIMIX_comm_copy_pointer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
 985 {
 986   xbt_assert((buff_size == sizeof(void *)),
 987              "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
 988   *(void **) (comm->comm.dst_buff) = buff;
 989 }
 990
 991 void SIMIX_comm_copy_buffer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
 992 {
 993   XBT_DEBUG("Copy the data over");
 994   memcpy(comm->comm.dst_buff, buff, buff_size);
 995   if (comm->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
 996     xbt_free(buff);
 997     comm->comm.src_buff = NULL;
 998   }
 999 }
1000
1001
1002 /**
1003  *  \brief Copy the communication data from the sender's buffer to the receiver's one
1004  *  \param comm The communication
1005  */
1006 void SIMIX_comm_copy_data(smx_synchro_t comm)
1007 {
1008   size_t buff_size = comm->comm.src_buff_size;
1009   /* If there is no data to be copy then return */
1010   if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
1011     return;
1012
1013   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
1014             comm,
1015             comm->comm.src_proc ? sg_host_get_name(comm->comm.src_proc->host) : "a finished process",
1016             comm->comm.src_buff,
1017             comm->comm.dst_proc ? sg_host_get_name(comm->comm.dst_proc->host) : "a finished process",
1018             comm->comm.dst_buff, buff_size);
1019
1020   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
1021   if (comm->comm.dst_buff_size)
1022     buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
1023
1024   /* Update the receiver's buffer size to the copied amount */
1025   if (comm->comm.dst_buff_size)
1026     *comm->comm.dst_buff_size = buff_size;
1027
1028   if (buff_size > 0){
1029       if(comm->comm.copy_data_fun)
1030         comm->comm.copy_data_fun (comm, comm->comm.src_buff, buff_size);
1031       else
1032         SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
1033   }
1034
1035
1036   /* Set the copied flag so we copy data only once */
1037   /* (this function might be called from both communication ends) */
1038   comm->comm.copied = 1;
1039 }