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