src/simix/smx_network.c

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