src/simix/smx_network.cpp

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