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 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_rdv_get_comm(smx_rdv_t rdv, 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 void SIMIX_rdv_free(void *data);
  26
  27 void SIMIX_network_init(void)
  28 {
  29   rdv_points = xbt_dict_new_homogeneous(SIMIX_rdv_free);
  30 }
  31
  32 void SIMIX_network_exit(void)
  33 {
  34   xbt_dict_free(&rdv_points);
  35 }
  36
  37 /******************************************************************************/
  38 /*                           Rendez-Vous Points                               */
  39 /******************************************************************************/
  40
  41 smx_rdv_t SIMIX_rdv_create(const char *name)
  42 {
  43   /* two processes may have pushed the same rdv_create simcall at the same time */
  44   smx_rdv_t rdv = name ? xbt_dict_get_or_null(rdv_points, name) : NULL;
  45
  46   if (!rdv) {
  47     rdv = xbt_new0(s_smx_rvpoint_t, 1);
  48     rdv->name = name ? xbt_strdup(name) : NULL;
  49     rdv->comm_fifo = xbt_fifo_new();
  50
  51     if (rdv->name)
  52       xbt_dict_set(rdv_points, rdv->name, rdv, NULL);
  53   }
  54   return rdv;
  55 }
  56
  57 void SIMIX_rdv_destroy(smx_rdv_t rdv)
  58 {
  59   if (rdv->name)
  60     xbt_dict_remove(rdv_points, rdv->name);
  61 }
  62
  63 void SIMIX_rdv_free(void *data)
  64 {
  65   smx_rdv_t rdv = (smx_rdv_t) data;
  66   xbt_free(rdv->name);
  67   xbt_fifo_free(rdv->comm_fifo);
  68   xbt_free(rdv);
  69 }
  70
  71 xbt_dict_t SIMIX_get_rdv_points()
  72 {
  73   return rdv_points;
  74 }
  75
  76 smx_rdv_t SIMIX_rdv_get_by_name(const char *name)
  77 {
  78   return xbt_dict_get_or_null(rdv_points, name);
  79 }
  80
  81 int SIMIX_rdv_comm_count_by_host(smx_rdv_t rdv, smx_host_t host)
  82 {
  83   smx_action_t comm = NULL;
  84   xbt_fifo_item_t item = NULL;
  85   int count = 0;
  86
  87   xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_action_t) {
  88     if (comm->comm.src_proc->smx_host == host)
  89       count++;
  90   }
  91
  92   return count;
  93 }
  94
  95 smx_action_t SIMIX_rdv_get_head(smx_rdv_t rdv)
  96 {
  97   return xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo));
  98 }
  99
 100 /**
 101  *  \brief Pushes a communication action into a rendez-vous point
 102  *  \param rdv The rendez-vous point
 103  *  \param comm The communication action
 104  */
 105 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm)
 106 {
 107   xbt_fifo_push(rdv->comm_fifo, comm);
 108   comm->comm.rdv = rdv;
 109 }
 110
 111 /**
 112  *  \brief Removes a communication action from a rendez-vous point
 113  *  \param rdv The rendez-vous point
 114  *  \param comm The communication action
 115  */
 116 XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm)
 117 {
 118   xbt_fifo_remove(rdv->comm_fifo, comm);
 119   comm->comm.rdv = NULL;
 120 }
 121
 122 /**
 123  *  \brief Checks if there is a communication action queued in a rendez-vous matching our needs
 124  *  \param type The type of communication we are looking for (comm_send, comm_recv)
 125  *  \return The communication action if found, NULL otherwise
 126  */
 127 smx_action_t SIMIX_rdv_get_comm(smx_rdv_t rdv, e_smx_comm_type_t type,
 128                                    int (*match_fun)(void *, void *,smx_action_t),
 129                                    void *this_user_data, smx_action_t my_action)
 130 {
 131   smx_action_t action;
 132   xbt_fifo_item_t item;
 133   void* other_user_data = NULL;
 134
 135   xbt_fifo_foreach(rdv->comm_fifo, item, action, smx_action_t) {
 136     if (action->comm.type == SIMIX_COMM_SEND) {
 137       other_user_data = action->comm.src_data;
 138     } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
 139       other_user_data = action->comm.dst_data;
 140     }
 141     if (action->comm.type == type &&
 142         (!match_fun              ||              match_fun(this_user_data,  other_user_data, action)) &&
 143         (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data,  my_action))) {
 144       XBT_DEBUG("Found a matching communication action %p", action);
 145       xbt_fifo_remove_item(rdv->comm_fifo, item);
 146       xbt_fifo_free_item(item);
 147       action->comm.refcount++;
 148       action->comm.rdv = NULL;
 149       return action;
 150     }
 151     XBT_DEBUG("Sorry, communication action %p does not match our needs:"
 152            " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
 153            action, (int)action->comm.type, (int)type);
 154   }
 155   XBT_DEBUG("No matching communication action found");
 156   return NULL;
 157 }
 158
 159
 160 /******************************************************************************/
 161 /*                            Communication Actions                            */
 162 /******************************************************************************/
 163
 164 /**
 165  *  \brief Creates a new communicate action
 166  *  \param type The direction of communication (comm_send, comm_recv)
 167  *  \return The new communicate action
 168  */
 169 smx_action_t SIMIX_comm_new(e_smx_comm_type_t type)
 170 {
 171   smx_action_t act;
 172
 173   /* alloc structures */
 174   act = xbt_mallocator_get(simix_global->action_mallocator);
 175
 176   act->type = SIMIX_ACTION_COMMUNICATE;
 177   act->state = SIMIX_WAITING;
 178
 179   /* set communication */
 180   act->comm.type = type;
 181   act->comm.refcount = 1;
 182
 183 #ifdef HAVE_LATENCY_BOUND_TRACKING
 184   //initialize with unknown value
 185   act->latency_limited = -1;
 186 #endif
 187
 188 #ifdef HAVE_TRACING
 189   act->category = NULL;
 190 #endif
 191
 192   XBT_DEBUG("Create communicate action %p", act);
 193   ++smx_total_comms;
 194
 195   return act;
 196 }
 197
 198 /**
 199  *  \brief Destroy a communicate action
 200  *  \param action The communicate action to be destroyed
 201  */
 202 void SIMIX_comm_destroy(smx_action_t action)
 203 {
 204   XBT_DEBUG("Destroy action %p (refcount: %d), state: %d",
 205             action, action->comm.refcount, (int)action->state);
 206
 207   if (action->comm.refcount <= 0) {
 208     xbt_backtrace_display_current();
 209     xbt_die("The refcount of comm %p is already 0 before decreasing it. "
 210             "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);
 211   }
 212   action->comm.refcount--;
 213   if (action->comm.refcount > 0)
 214     return;
 215   XBT_DEBUG("Really free communication %p; refcount is now %d", action,
 216         action->comm.refcount);
 217
 218 #ifdef HAVE_LATENCY_BOUND_TRACKING
 219     action->latency_limited = SIMIX_comm_is_latency_bounded( action ) ;
 220 #endif
 221
 222   xbt_free(action->name);
 223   SIMIX_comm_destroy_internal_actions(action);
 224
 225   if (action->comm.detached && action->state != SIMIX_DONE) {
 226     /* the communication has failed and was detached:
 227      * we have to free the buffer */
 228     if (action->comm.clean_fun) {
 229       action->comm.clean_fun(action->comm.src_buff);
 230     }
 231     action->comm.src_buff = NULL;
 232   }
 233
 234   xbt_mallocator_release(simix_global->action_mallocator, action);
 235 }
 236
 237 void SIMIX_comm_destroy_internal_actions(smx_action_t action)
 238 {
 239   if (action->comm.surf_comm){
 240 #ifdef HAVE_LATENCY_BOUND_TRACKING
 241     action->latency_limited = SIMIX_comm_is_latency_bounded(action);
 242 #endif
 243     action->comm.surf_comm->model_type->action_unref(action->comm.surf_comm);
 244     action->comm.surf_comm = NULL;
 245   }
 246
 247   if (action->comm.src_timeout){
 248     action->comm.src_timeout->model_type->action_unref(action->comm.src_timeout);
 249     action->comm.src_timeout = NULL;
 250   }
 251
 252   if (action->comm.dst_timeout){
 253     action->comm.dst_timeout->model_type->action_unref(action->comm.dst_timeout);
 254     action->comm.dst_timeout = NULL;
 255   }
 256 }
 257
 258 smx_action_t SIMIX_comm_isend(smx_process_t src_proc, smx_rdv_t rdv,
 259                               double task_size, double rate,
 260                               void *src_buff, size_t src_buff_size,
 261                               int (*match_fun)(void *, void *,smx_action_t),
 262                               void (*clean_fun)(void *), // used to free the action in case of problem after a detached send
 263                               void *data,
 264                               int detached)
 265 {
 266   /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
 267   smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_SEND);
 268
 269   /* Look for communication action matching our needs. We also provide a description of
 270    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 271    *
 272    * If it is not found then push our communication into the rendez-vous point */
 273   smx_action_t other_action = SIMIX_rdv_get_comm(rdv, SIMIX_COMM_RECEIVE, match_fun, data, this_action);
 274
 275   if (!other_action) {
 276     other_action = this_action;
 277     SIMIX_rdv_push(rdv, this_action);
 278   } else {
 279     SIMIX_comm_destroy(this_action);
 280     --smx_total_comms; // this creation was a pure waste
 281
 282     other_action->state = SIMIX_READY;
 283     other_action->comm.type = SIMIX_COMM_READY;
 284   }
 285   xbt_fifo_push(src_proc->comms, other_action);
 286
 287   /* if the communication action is detached then decrease the refcount
 288    * by one, so it will be eliminated by the receiver's destroy call */
 289   if (detached) {
 290     other_action->comm.detached = 1;
 291     other_action->comm.refcount--;
 292     other_action->comm.clean_fun = clean_fun;
 293   } else {
 294     other_action->comm.clean_fun = NULL;
 295   }
 296
 297   /* Setup the communication action */
 298   other_action->comm.src_proc = src_proc;
 299   other_action->comm.task_size = task_size;
 300   other_action->comm.rate = rate;
 301   other_action->comm.src_buff = src_buff;
 302   other_action->comm.src_buff_size = src_buff_size;
 303   other_action->comm.src_data = data;
 304
 305   other_action->comm.match_fun = match_fun;
 306
 307   if (MC_IS_ENABLED) {
 308     other_action->state = SIMIX_RUNNING;
 309     return other_action;
 310   }
 311
 312   SIMIX_comm_start(other_action);
 313   return (detached ? NULL : other_action);
 314 }
 315
 316 smx_action_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_rdv_t rdv,
 317                       void *dst_buff, size_t *dst_buff_size,
 318                       int (*match_fun)(void *, void *, smx_action_t), void *data)
 319 {
 320   /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
 321   smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
 322
 323   /* Look for communication action matching our needs. We also provide a description of
 324    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
 325    *
 326    * If it is not found then push our communication into the rendez-vous point */
 327   smx_action_t other_action = SIMIX_rdv_get_comm(rdv, SIMIX_COMM_SEND, match_fun, data, this_action);
 328
 329   if (!other_action) {
 330     other_action = this_action;
 331     SIMIX_rdv_push(rdv, this_action);
 332   } else {
 333     SIMIX_comm_destroy(this_action);
 334     --smx_total_comms; // this creation was a pure waste
 335
 336     other_action->state = SIMIX_READY;
 337     other_action->comm.type = SIMIX_COMM_READY;
 338   }
 339   xbt_fifo_push(dst_proc->comms, other_action);
 340
 341   /* Setup communication action */
 342   other_action->comm.dst_proc = dst_proc;
 343   other_action->comm.dst_buff = dst_buff;
 344   other_action->comm.dst_buff_size = dst_buff_size;
 345   other_action->comm.dst_data = data;
 346
 347   other_action->comm.match_fun = match_fun;
 348
 349   if (MC_IS_ENABLED) {
 350     other_action->state = SIMIX_RUNNING;
 351     return other_action;
 352   }
 353
 354   SIMIX_comm_start(other_action);
 355   return other_action;
 356 }
 357
 358 void SIMIX_pre_comm_wait(smx_simcall_t simcall, smx_action_t action, double timeout, int idx)
 359 {
 360
 361   /* the simcall may be a wait, a send or a recv */
 362   surf_action_t sleep;
 363
 364   /* Associate this simcall to the wait action */
 365   xbt_fifo_push(action->simcalls, simcall);
 366   simcall->issuer->waiting_action = action;
 367
 368   if (MC_IS_ENABLED) {
 369     if (idx == 0) {
 370       action->state = SIMIX_DONE;
 371     } else {
 372       /* If we reached this point, the wait simcall must have a timeout */
 373       /* Otherwise it shouldn't be enabled and executed by the MC */
 374       if (timeout == -1)
 375         THROW_IMPOSSIBLE;
 376
 377       if (action->comm.src_proc == simcall->issuer)
 378         action->state = SIMIX_SRC_TIMEOUT;
 379       else
 380         action->state = SIMIX_DST_TIMEOUT;
 381     }
 382
 383     SIMIX_comm_finish(action);
 384     return;
 385   }
 386
 387   /* If the action has already finish perform the error handling, */
 388   /* otherwise set up a waiting timeout on the right side         */
 389   if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
 390     SIMIX_comm_finish(action);
 391   } 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 */
 392     sleep = surf_workstation_model->extension.workstation.sleep(simcall->issuer->smx_host->host, timeout);
 393     surf_workstation_model->action_data_set(sleep, action);
 394
 395     if (simcall->issuer == action->comm.src_proc)
 396       action->comm.src_timeout = sleep;
 397     else
 398       action->comm.dst_timeout = sleep;
 399   }
 400 }
 401
 402 void SIMIX_pre_comm_test(smx_simcall_t simcall)
 403 {
 404   smx_action_t action = simcall->comm_test.comm;
 405
 406   if(MC_IS_ENABLED){
 407     simcall->comm_test.result = action->comm.src_proc && action->comm.dst_proc;
 408     if(simcall->comm_test.result){
 409       action->state = SIMIX_DONE;
 410       xbt_fifo_push(action->simcalls, simcall);
 411       SIMIX_comm_finish(action);
 412     }else{
 413       SIMIX_simcall_answer(simcall);
 414     }
 415     return;
 416   }
 417
 418   simcall->comm_test.result = (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING);
 419   if (simcall->comm_test.result) {
 420     xbt_fifo_push(action->simcalls, simcall);
 421     SIMIX_comm_finish(action);
 422   } else {
 423     SIMIX_simcall_answer(simcall);
 424   }
 425 }
 426
 427 void SIMIX_pre_comm_testany(smx_simcall_t simcall, int idx)
 428 {
 429   unsigned int cursor;
 430   smx_action_t action;
 431   xbt_dynar_t actions = simcall->comm_testany.comms;
 432   simcall->comm_testany.result = -1;
 433
 434   if (MC_IS_ENABLED){
 435     if(idx == -1){
 436       SIMIX_simcall_answer(simcall);
 437     }else{
 438       action = xbt_dynar_get_as(actions, idx, smx_action_t);
 439       simcall->comm_testany.result = idx;
 440       xbt_fifo_push(action->simcalls, simcall);
 441       action->state = SIMIX_DONE;
 442       SIMIX_comm_finish(action);
 443     }
 444     return;
 445   }
 446
 447   xbt_dynar_foreach(simcall->comm_testany.comms,cursor,action) {
 448     if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
 449       simcall->comm_testany.result = cursor;
 450       xbt_fifo_push(action->simcalls, simcall);
 451       SIMIX_comm_finish(action);
 452       return;
 453     }
 454   }
 455   SIMIX_simcall_answer(simcall);
 456 }
 457
 458 void SIMIX_pre_comm_waitany(smx_simcall_t simcall, int idx)
 459 {
 460   smx_action_t action;
 461   unsigned int cursor = 0;
 462   xbt_dynar_t actions = simcall->comm_waitany.comms;
 463
 464   if (MC_IS_ENABLED){
 465     action = xbt_dynar_get_as(actions, idx, smx_action_t);
 466     xbt_fifo_push(action->simcalls, simcall);
 467     simcall->comm_waitany.result = idx;
 468     action->state = SIMIX_DONE;
 469     SIMIX_comm_finish(action);
 470     return;
 471   }
 472
 473   xbt_dynar_foreach(actions, cursor, action){
 474     /* associate this simcall to the the action */
 475     xbt_fifo_push(action->simcalls, simcall);
 476
 477     /* see if the action is already finished */
 478     if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING){
 479       SIMIX_comm_finish(action);
 480       break;
 481     }
 482   }
 483 }
 484
 485 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
 486 {
 487   smx_action_t action;
 488   unsigned int cursor = 0;
 489   xbt_dynar_t actions = simcall->comm_waitany.comms;
 490
 491   xbt_dynar_foreach(actions, cursor, action) {
 492     xbt_fifo_remove(action->simcalls, simcall);
 493   }
 494 }
 495
 496 /**
 497  *  \brief Starts the simulation of a communication action.
 498  *  \param action the communication action
 499  */
 500 XBT_INLINE void SIMIX_comm_start(smx_action_t action)
 501 {
 502   /* If both the sender and the receiver are already there, start the communication */
 503   if (action->state == SIMIX_READY) {
 504
 505     smx_host_t sender = action->comm.src_proc->smx_host;
 506     smx_host_t receiver = action->comm.dst_proc->smx_host;
 507
 508     XBT_DEBUG("Starting communication %p from '%s' to '%s'", action,
 509            SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
 510
 511     action->comm.surf_comm = surf_workstation_model->extension.workstation.
 512         communicate(sender->host, receiver->host, action->comm.task_size, action->comm.rate);
 513
 514     surf_workstation_model->action_data_set(action->comm.surf_comm, action);
 515
 516     action->state = SIMIX_RUNNING;
 517
 518     /* If a link is failed, detect it immediately */
 519     if (surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED) {
 520       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
 521     SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
 522       action->state = SIMIX_LINK_FAILURE;
 523       SIMIX_comm_destroy_internal_actions(action);
 524     }
 525
 526     /* If any of the process is suspend, create the action but stop its execution,
 527        it will be restarted when the sender process resume */
 528     if (SIMIX_process_is_suspended(action->comm.src_proc) ||
 529         SIMIX_process_is_suspended(action->comm.dst_proc)) {
 530       /* FIXME: check what should happen with the action state */
 531
 532       if (SIMIX_process_is_suspended(action->comm.src_proc))
 533         XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
 534             SIMIX_host_get_name(action->comm.src_proc->smx_host), action->comm.src_proc->name);
 535       else
 536         XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
 537             SIMIX_host_get_name(action->comm.dst_proc->smx_host), action->comm.dst_proc->name);
 538
 539       surf_workstation_model->suspend(action->comm.surf_comm);
 540
 541     }
 542   }
 543 }
 544
 545 /**
 546  * \brief Answers the SIMIX simcalls associated to a communication action.
 547  * \param action a finished communication action
 548  */
 549 void SIMIX_comm_finish(smx_action_t action)
 550 {
 551   unsigned int destroy_count = 0;
 552   smx_simcall_t simcall;
 553
 554   while ((simcall = xbt_fifo_shift(action->simcalls))) {
 555
 556     /* If a waitany simcall is waiting for this action to finish, then remove
 557        it from the other actions in the waitany list. Afterwards, get the
 558        position of the actual action in the waitany dynar and
 559        return it as the result of the simcall */
 560     if (simcall->call == SIMCALL_COMM_WAITANY) {
 561       SIMIX_waitany_remove_simcall_from_actions(simcall);
 562       if (!MC_IS_ENABLED)
 563         simcall->comm_waitany.result = xbt_dynar_search(simcall->comm_waitany.comms, &action);
 564     }
 565
 566     /* If the action is still in a rendez-vous point then remove from it */
 567     if (action->comm.rdv)
 568       SIMIX_rdv_remove(action->comm.rdv, action);
 569
 570     XBT_DEBUG("SIMIX_comm_finish: action state = %d", (int)action->state);
 571
 572     /* Check out for errors */
 573     switch (action->state) {
 574
 575       case SIMIX_DONE:
 576         XBT_DEBUG("Communication %p complete!", action);
 577         SIMIX_comm_copy_data(action);
 578         break;
 579
 580       case SIMIX_SRC_TIMEOUT:
 581         SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
 582                   "Communication timeouted because of sender");
 583         break;
 584
 585       case SIMIX_DST_TIMEOUT:
 586         SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
 587                   "Communication timeouted because of receiver");
 588         break;
 589
 590       case SIMIX_SRC_HOST_FAILURE:
 591         if (simcall->issuer == action->comm.src_proc)
 592           simcall->issuer->context->iwannadie = 1;
 593 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 594         else
 595           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 596         break;
 597
 598       case SIMIX_DST_HOST_FAILURE:
 599         if (simcall->issuer == action->comm.dst_proc)
 600           simcall->issuer->context->iwannadie = 1;
 601 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
 602         else
 603           SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
 604         break;
 605
 606       case SIMIX_LINK_FAILURE:
 607         XBT_DEBUG("Link failure in action %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
 608             action,
 609             action->comm.src_proc ? action->comm.src_proc->smx_host->name : NULL,
 610             action->comm.dst_proc ? action->comm.dst_proc->smx_host->name : NULL,
 611             simcall->issuer->name, simcall->issuer, action->comm.detached);
 612         if (action->comm.src_proc == simcall->issuer) {
 613           XBT_DEBUG("I'm source");
 614         } else if (action->comm.dst_proc == simcall->issuer) {
 615           XBT_DEBUG("I'm dest");
 616         } else {
 617           XBT_DEBUG("I'm neither source nor dest");
 618         }
 619         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
 620         break;
 621
 622       case SIMIX_CANCELED:
 623         if (simcall->issuer == action->comm.dst_proc)
 624           SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
 625                     "Communication canceled by the sender");
 626         else
 627           SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
 628                     "Communication canceled by the receiver");
 629         break;
 630
 631       default:
 632         xbt_die("Unexpected action state in SIMIX_comm_finish: %d", (int)action->state);
 633     }
 634
 635     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
 636     if (simcall->issuer->doexception) {
 637       if (simcall->call == SIMCALL_COMM_WAITANY) {
 638         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall->comm_waitany.comms, &action);
 639       }
 640       else if (simcall->call == SIMCALL_COMM_TESTANY) {
 641         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall->comm_testany.comms, &action);
 642       }
 643     }
 644
 645     if (surf_workstation_model->extension.
 646         workstation.get_state(simcall->issuer->smx_host->host) != SURF_RESOURCE_ON) {
 647       simcall->issuer->context->iwannadie = 1;
 648     }
 649
 650     simcall->issuer->waiting_action = NULL;
 651     xbt_fifo_remove(simcall->issuer->comms, action);
 652     SIMIX_simcall_answer(simcall);
 653     destroy_count++;
 654   }
 655
 656   while (destroy_count-- > 0)
 657     SIMIX_comm_destroy(action);
 658 }
 659
 660 /**
 661  * \brief This function is called when a Surf communication action is finished.
 662  * \param action the corresponding Simix communication
 663  */
 664 void SIMIX_post_comm(smx_action_t action)
 665 {
 666   /* Update action state */
 667   if (action->comm.src_timeout &&
 668      surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_DONE)
 669      action->state = SIMIX_SRC_TIMEOUT;
 670   else if (action->comm.dst_timeout &&
 671           surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_DONE)
 672      action->state = SIMIX_DST_TIMEOUT;
 673   else if (action->comm.src_timeout &&
 674           surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_FAILED)
 675      action->state = SIMIX_SRC_HOST_FAILURE;
 676   else if (action->comm.dst_timeout &&
 677           surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_FAILED)
 678      action->state = SIMIX_DST_HOST_FAILURE;
 679   else if (action->comm.surf_comm &&
 680           surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED) {
 681     XBT_DEBUG("Puta madre. Surf says that the link broke");
 682      action->state = SIMIX_LINK_FAILURE;
 683   } else
 684     action->state = SIMIX_DONE;
 685
 686   XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
 687       action, (int)action->state, action->comm.src_proc, action->comm.dst_proc, action->comm.detached);
 688
 689   /* destroy the surf actions associated with the Simix communication */
 690   SIMIX_comm_destroy_internal_actions(action);
 691
 692   /* remove the communication action from the list of pending communications
 693    * of both processes (if they still exist) */
 694   if (action->comm.src_proc) {
 695     xbt_fifo_remove(action->comm.src_proc->comms, action);
 696   }
 697   if (action->comm.dst_proc) {
 698     xbt_fifo_remove(action->comm.dst_proc->comms, action);
 699   }
 700
 701   /* if there are simcalls associated with the action, then answer them */
 702   if (xbt_fifo_size(action->simcalls)) {
 703     SIMIX_comm_finish(action);
 704   }
 705 }
 706
 707 void SIMIX_comm_cancel(smx_action_t action)
 708 {
 709   /* if the action is a waiting state means that it is still in a rdv */
 710   /* so remove from it and delete it */
 711   if (action->state == SIMIX_WAITING) {
 712     SIMIX_rdv_remove(action->comm.rdv, action);
 713     action->state = SIMIX_CANCELED;
 714   }
 715   else if (!MC_IS_ENABLED /* when running the MC there are no surf actions */
 716       && (action->state == SIMIX_READY || action->state == SIMIX_RUNNING)) {
 717
 718     surf_workstation_model->action_cancel(action->comm.surf_comm);
 719   }
 720 }
 721
 722 void SIMIX_comm_suspend(smx_action_t action)
 723 {
 724   /*FIXME: shall we suspend also the timeout actions? */
 725   if (action->comm.surf_comm)
 726     surf_workstation_model->suspend(action->comm.surf_comm);
 727   /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
 728 }
 729
 730 void SIMIX_comm_resume(smx_action_t action)
 731 {
 732   /*FIXME: check what happen with the timeouts */
 733   if (action->comm.surf_comm)
 734     surf_workstation_model->resume(action->comm.surf_comm);
 735   /* in the other case, the action were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
 736 }
 737
 738
 739 /************* Action Getters **************/
 740
 741 /**
 742  *  \brief get the amount remaining from the communication
 743  *  \param action The communication
 744  */
 745 double SIMIX_comm_get_remains(smx_action_t action)
 746 {
 747   double remains;
 748
 749   if(!action){
 750       return 0;
 751   }
 752
 753   switch (action->state) {
 754
 755     case SIMIX_RUNNING:
 756       remains = surf_workstation_model->get_remains(action->comm.surf_comm);
 757       break;
 758
 759     case SIMIX_WAITING:
 760     case SIMIX_READY:
 761       remains = 0; /*FIXME: check what should be returned */
 762       break;
 763
 764     default:
 765       remains = 0; /*FIXME: is this correct? */
 766       break;
 767   }
 768   return remains;
 769 }
 770
 771 e_smx_state_t SIMIX_comm_get_state(smx_action_t action)
 772 {
 773   return action->state;
 774 }
 775
 776 /**
 777  *  \brief Return the user data associated to the sender of the communication
 778  *  \param action The communication
 779  *  \return the user data
 780  */
 781 void* SIMIX_comm_get_src_data(smx_action_t action)
 782 {
 783   return action->comm.src_data;
 784 }
 785
 786 /**
 787  *  \brief Return the user data associated to the receiver of the communication
 788  *  \param action The communication
 789  *  \return the user data
 790  */
 791 void* SIMIX_comm_get_dst_data(smx_action_t action)
 792 {
 793   return action->comm.dst_data;
 794 }
 795
 796 smx_process_t SIMIX_comm_get_src_proc(smx_action_t action)
 797 {
 798   return action->comm.src_proc;
 799 }
 800
 801 smx_process_t SIMIX_comm_get_dst_proc(smx_action_t action)
 802 {
 803   return action->comm.dst_proc;
 804 }
 805
 806 #ifdef HAVE_LATENCY_BOUND_TRACKING
 807 /**
 808  *  \brief verify if communication is latency bounded
 809  *  \param comm The communication
 810  */
 811 XBT_INLINE int SIMIX_comm_is_latency_bounded(smx_action_t action)
 812 {
 813   if(!action){
 814       return 0;
 815   }
 816   if (action->comm.surf_comm){
 817       XBT_DEBUG("Getting latency limited for surf_action (%p)", action->comm.surf_comm);
 818       action->latency_limited = surf_workstation_model->get_latency_limited(action->comm.surf_comm);
 819       XBT_DEBUG("Action limited is %d", action->latency_limited);
 820   }
 821   return action->latency_limited;
 822 }
 823 #endif
 824
 825 /******************************************************************************/
 826 /*                    SIMIX_comm_copy_data callbacks                       */
 827 /******************************************************************************/
 828 static void (*SIMIX_comm_copy_data_callback) (smx_action_t, void*, size_t) =
 829     &SIMIX_comm_copy_pointer_callback;
 830
 831 void
 832 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_action_t, void*, size_t))
 833 {
 834   SIMIX_comm_copy_data_callback = callback;
 835 }
 836
 837 void SIMIX_comm_copy_pointer_callback(smx_action_t comm, void* buff, size_t buff_size)
 838 {
 839   xbt_assert((buff_size == sizeof(void *)),
 840               "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
 841   *(void **) (comm->comm.dst_buff) = buff;
 842 }
 843
 844 void SIMIX_comm_copy_buffer_callback(smx_action_t comm, void* buff, size_t buff_size)
 845 {
 846   XBT_DEBUG("Copy the data over");
 847   memcpy(comm->comm.dst_buff, buff, buff_size);
 848 }
 849
 850 void smpi_comm_copy_data_callback(smx_action_t comm, void* buff, size_t buff_size)
 851 {
 852   XBT_DEBUG("Copy the data over");
 853   memcpy(comm->comm.dst_buff, buff, buff_size);
 854   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
 855     xbt_free(buff);
 856     comm->comm.src_buff = NULL;
 857   }
 858 }
 859
 860 /**
 861  *  \brief Copy the communication data from the sender's buffer to the receiver's one
 862  *  \param comm The communication
 863  */
 864 void SIMIX_comm_copy_data(smx_action_t comm)
 865 {
 866   size_t buff_size = comm->comm.src_buff_size;
 867   /* If there is no data to be copy then return */
 868   if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
 869     return;
 870
 871   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
 872          comm,
 873          comm->comm.src_proc ? comm->comm.src_proc->smx_host->name : "a finished process",
 874          comm->comm.src_buff,
 875          comm->comm.dst_proc ? comm->comm.dst_proc->smx_host->name : "a finished process",
 876          comm->comm.dst_buff, buff_size);
 877
 878   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
 879   if (comm->comm.dst_buff_size)
 880     buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
 881
 882   /* Update the receiver's buffer size to the copied amount */
 883   if (comm->comm.dst_buff_size)
 884     *comm->comm.dst_buff_size = buff_size;
 885
 886   if (buff_size > 0)
 887     SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
 888
 889   /* Set the copied flag so we copy data only once */
 890   /* (this function might be called from both communication ends) */
 891   comm->comm.copied = 1;
 892 }