src/simix/smx_network.c

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