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