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