src/simix/smx_network.c

   1 /*      $Id$     */
   2
   3 /* Copyright (c) 2009 Cristian Rosa.
   4    All rights reserved.                                          */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include "private.h"
  10 #include "xbt/log.h"
  11
  12 /******************************************************************************/
  13 /*                           Rendez-Vous Points                               */
  14 /******************************************************************************/
  15
  16 /**
  17  *  \brief Creates a new rendez-vous point
  18  *  \param name The name of the rendez-vous point
  19  *  \return The created rendez-vous point
  20  */
  21 smx_rdv_t SIMIX_rdv_create(const char *name)
  22 {
  23   smx_rdv_t rvp = xbt_new0(s_smx_rvpoint_t, 1);
  24   rvp->name = name ? xbt_strdup(name) : NULL;
  25   rvp->read = SIMIX_mutex_init();
  26   rvp->write = SIMIX_mutex_init();
  27   rvp->comm_fifo = xbt_fifo_new();
  28
  29   return rvp;
  30 }
  31
  32 /**
  33  *  \brief Destroy a rendez-vous point
  34  *  \param name The rendez-vous point to destroy
  35  */
  36 void SIMIX_rdv_destroy(smx_rdv_t rvp)
  37 {
  38   if(rvp->name)
  39     xbt_free(rvp->name);
  40   SIMIX_mutex_destroy(rvp->read);
  41   SIMIX_mutex_destroy(rvp->write);
  42   xbt_fifo_free(rvp->comm_fifo);
  43   xbt_free(rvp);
  44 }
  45
  46 /**
  47  *  \brief Push a communication request into a rendez-vous point
  48  *  The communications request are dequeued by the two functions below
  49  *  \param rvp The rendez-vous point
  50  *  \param comm The communication request
  51  */
  52 static inline void SIMIX_rdv_push(smx_rdv_t rvp, smx_comm_t comm)
  53 {
  54   xbt_fifo_push(rvp->comm_fifo, comm);
  55 }
  56
  57 /**
  58  *  \brief Checks if there is a communication request queued in a rendez-vous matching our needs
  59  *  \param rvp The rendez-vous with the queue
  60  *  \param look_for_src boolean. True: we are receiver looking for sender; False: other way round
  61  *  \return The communication request if found, or a newly created one otherwise.
  62  */
  63 smx_comm_t SIMIX_rdv_get_request(smx_rdv_t rvp, int (filter)(smx_comm_t, void*), void *arg) {
  64   smx_comm_t comm;
  65   xbt_fifo_item_t item;
  66
  67   /* Traverse the rendez-vous queue looking for a comm request matching the
  68      filter conditions. If found return it and remove it from the list. */
  69   xbt_fifo_foreach(rvp->comm_fifo, item, comm, smx_comm_t) {
  70     if(filter(comm, arg)){
  71       SIMIX_communication_use(comm);
  72       xbt_fifo_remove_item(rvp->comm_fifo, item);
  73       return comm;
  74     }
  75   }
  76
  77   /* no relevant request found. Return NULL */
  78   return NULL;
  79 }
  80
  81 /******************************************************************************/
  82 /*                           Communication Requests                           */
  83 /******************************************************************************/
  84
  85 /**
  86  *  \brief Creates a new communication request
  87  *  \param sender The process starting the communication (by send)
  88  *  \param receiver The process receiving the communication (by recv)
  89  *  \return the communication request
  90  */
  91 smx_comm_t SIMIX_communication_new(smx_comm_type_t type, smx_rdv_t rdv)
  92 {
  93   /* alloc structures */
  94   smx_comm_t comm = xbt_new0(s_smx_comm_t, 1);
  95   comm->type = type;
  96   comm->cond = SIMIX_cond_init();
  97   comm->rdv = rdv;
  98   comm->refcount = 1;
  99
 100   return comm;
 101 }
 102
 103 /**
 104  *  \brief Destroy a communication request
 105  *  \param comm The request to be destroyed
 106  */
 107 void SIMIX_communication_destroy(smx_comm_t comm)
 108 {
 109   comm->refcount--;
 110   if(comm->refcount == 0){
 111     if(comm->act != NULL)
 112       SIMIX_action_destroy(comm->act);
 113
 114     xbt_free(comm->cond);
 115     xbt_free(comm);
 116   }
 117 }
 118
 119 /**
 120  *  \brief Increase the number of users of the communication.
 121  *  \param comm The communication request
 122  *  Each communication request can be used by more than one process, so it is
 123  *  necessary to know number of them at destroy time, to avoid freeing stuff that
 124  *  maybe is in use by others.
 125  *  \
 126  */
 127 static inline void SIMIX_communication_use(smx_comm_t comm)
 128 {
 129   comm->refcount++;
 130 }
 131
 132 /**
 133  *  \brief Start the simulation of a communication request
 134  *  \param comm The communication request
 135  */
 136 static inline void SIMIX_communication_start(smx_comm_t comm)
 137 {
 138   /* If both the sender and the receiver are already there, start the communication */
 139   if(comm->src_host != NULL && comm->dst_host != NULL){
 140     comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL,
 141                                          comm->task_size, comm->rate);
 142     /* Add the communication as user data into the action, so it can be reached from it later */
 143     comm->act->data = comm;
 144
 145     SIMIX_register_action_to_condition(comm->act, comm->cond);
 146   }
 147 }
 148
 149 /**
 150  *  \brief Waits for communication completion and performs error checking
 151  *  \param comm The communication
 152  *  \param timeout The max amount of time to wait for the communication to finish
 153  *
 154  *  Throws:
 155  *   - host_error if peer failed
 156  *   - timeout_error if communication reached the timeout specified
 157  *   - network_error if network failed or peer issued a timeout
 158  */
 159 static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, double timeout)
 160 {
 161   xbt_ex_t e;
 162
 163   if(timeout > 0){
 164     TRY{
 165       SIMIX_cond_wait_timeout(comm->cond, NULL, timeout);
 166     }
 167     CATCH(e){
 168       /* If it's a timeout then cancel the communication and signal the other peer */
 169       if(e.category == timeout_error)
 170         SIMIX_action_cancel(comm->act);
 171         SIMIX_cond_signal(comm->cond);
 172       RETHROW;
 173     }
 174   }else{
 175     SIMIX_cond_wait(comm->cond, NULL);
 176   }
 177
 178   SIMIX_unregister_action_to_condition(comm->act, comm->cond);
 179
 180   /* Check for errors */
 181   if (SIMIX_host_get_state(comm->dst_host) == 0){
 182     THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name);
 183   } else if (SIMIX_host_get_state(comm->src_host) == 0){
 184     THROW1(host_error, 0, "Source host %s failed", comm->src_host->name);
 185   } else if (SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){
 186     THROW0(network_error, 0, "Link failure");
 187   }
 188 }
 189
 190 /**
 191  *  \brief Copy the communication data from the sender's buffer to the receiver's one
 192  *  \param comm The communication
 193  */
 194 void SIMIX_network_copy_data(smx_comm_t comm)
 195 {
 196   /* Copy the minimum between the size of the sender's message and the size of the
 197      receiver's buffer */
 198   *comm->dest_buff_size = *comm->dest_buff_size < comm->data_size ?
 199                             *comm->dest_buff_size : comm->data_size;
 200
 201   memcpy(comm->dest_buff, comm->data, *comm->dest_buff_size);
 202 }
 203
 204 /**
 205  *  \brief Checks if a communication is a send request
 206  *  \param comm The communication
 207  *  \return Boolean value
 208  */
 209 int SIMIX_communication_isSend(smx_comm_t comm)
 210 {
 211   return comm->type == comm_send ? TRUE : FALSE;
 212 }
 213
 214 /**
 215  *  \brief Checks if a communication is a recv request
 216  *  \param comm The communication
 217  *  \return Boolean value
 218  */
 219 int SIMIX_communication_isRecv(smx_comm_t comm)
 220 {
 221   return comm->type == comm_recv ? TRUE : FALSE;
 222 }
 223
 224 /* FIXME: move to some other place */
 225 int comm_filter_get(smx_comm_t comm, void *arg)
 226 {
 227   if(comm->type == comm_send){
 228     if(arg && comm->src_host != (smx_host_t)arg)
 229      return FALSE;
 230     else
 231      return TRUE;
 232   }else{
 233     return FALSE;
 234   }
 235 }
 236
 237 int comm_filter_put(smx_comm_t comm, void *arg)
 238 {
 239   return comm->type == comm_recv ? TRUE : FALSE;
 240 }
 241 /******************************************************************************/
 242 /*                        Synchronous Communication                           */
 243 /******************************************************************************/
 244 /*  Throws:
 245  *   - host_error if peer failed
 246  *   - timeout_error if communication reached the timeout specified
 247  *   - network_error if network failed or peer issued a timeout
 248  */
 249 void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate,
 250                         double timeout, void *data, size_t data_size,
 251                         int (filter)(smx_comm_t, void *), void *arg)
 252 {
 253   smx_comm_t comm;
 254
 255   /* Look for communication request matching our needs.
 256      If it is not found then create it and push it into the rendez-vous point */
 257   comm = SIMIX_rdv_get_request(rdv, filter, arg);
 258
 259   if(comm == NULL){
 260     comm = SIMIX_communication_new(comm_send, rdv);
 261     SIMIX_rdv_push(rdv, comm);
 262   }
 263
 264   /* Setup the communication request */
 265   comm->src_host = SIMIX_host_self();
 266   comm->task_size = task_size;
 267   comm->rate = rate;
 268   comm->data = data;
 269   comm->data_size = data_size;
 270
 271   SIMIX_communication_start(comm);
 272
 273   /* Wait for communication completion */
 274   /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 275   /* FIXME: add timeout checking stuff */
 276   SIMIX_communication_wait_for_completion(comm, timeout);
 277
 278   SIMIX_communication_destroy(comm);
 279 }
 280
 281 /*  Throws:
 282  *   - host_error if peer failed
 283  *   - timeout_error if communication reached the timeout specified
 284  *   - network_error if network failed or peer issued a timeout
 285  */
 286 void SIMIX_network_recv(smx_rdv_t rdv, double timeout, void *data,
 287                         size_t *data_size, int (filter)(smx_comm_t, void *), void *arg)
 288 {
 289   smx_comm_t comm;
 290
 291   /* Look for communication request matching our needs.
 292      If it is not found then create it and push it into the rendez-vous point */
 293   comm = SIMIX_rdv_get_request(rdv, filter, arg);
 294
 295   if(comm == NULL){
 296     comm = SIMIX_communication_new(comm_send, rdv);
 297     SIMIX_rdv_push(rdv, comm);
 298   }
 299
 300   /* Setup communication request */
 301   comm->dst_host = SIMIX_host_self();
 302   comm->dest_buff = data;
 303   comm->dest_buff_size = data_size;
 304
 305   SIMIX_communication_start(comm);
 306
 307   /* Wait for communication completion */
 308   /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 309   /* FIXME: add timeout checking stuff */
 310   SIMIX_communication_wait_for_completion(comm, timeout);
 311
 312   SIMIX_communication_destroy(comm);
 313 }
 314
 315 /******************************************************************************/
 316 /*                        Asynchronous Communication                          */
 317 /******************************************************************************/
 318
 319 /*
 320 void SIMIX_network_wait(smx_action_t comm, double timeout)
 321 {
 322     if (timeout > 0)
 323       SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time);
 324     else
 325       SIMIX_cond_wait(rvp_cond, rvp_comm_mutex);
 326
 327 }
 328
 329
 330 XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm)
 331 {
 332   if(SIMIX_action_get_state (comm) == SURF_ACTION_DONE){
 333     memcpy(comm->data
 334
 335   return SIMIX_action_get_state (comm) == SURF_ACTION_DONE ? TRUE : FALSE;
 336 }*/
 337
 338
 339
 340
 341
 342
 343