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(char *name)
  22 {
  23   smx_rdv_t rvp = xbt_new0(s_smx_rvpoint_t, 1);
  24   rvp->name = xbt_strdup(name);
  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   xbt_free(rvp->name);
  39   SIMIX_mutex_destroy(rvp->read);
  40   SIMIX_mutex_destroy(rvp->write);
  41   xbt_fifo_free(rvp->comm_fifo);
  42   xbt_free(rvp);
  43 }
  44
  45 /**
  46  *  \brief Push a communication request into a rendez-vous point
  47  *  The communications request are dequeued by the two functions below
  48  *  \param rvp The rendez-vous point
  49  *  \param comm The communication request
  50  */
  51 static inline void SIMIX_rdv_push(smx_rdv_t rvp, smx_comm_t comm)
  52 {
  53   xbt_fifo_push(rvp->comm_fifo, comm);
  54 }
  55
  56 /**
  57  *  \brief Checks if there is a communication request queued in a rendez-vous matching our needs
  58  *  \param rvp The rendez-vous with the queue
  59  *  \param look_for_src boolean. True: we are receiver looking for sender; False: other way round
  60  *  \return The communication request if found, or a newly created one otherwise.
  61  */
  62 smx_comm_t SIMIX_rdv_get_request_or_create(smx_rdv_t rvp, int look_for_src) {
  63   /* Get a communication request from the rendez-vous queue. If it is the kind
  64      of request we are looking for then return it, otherwise put it again in the
  65      queue.
  66    */
  67   smx_comm_t comm = xbt_fifo_shift(rvp->comm_fifo);
  68
  69   if(comm != NULL) {
  70     if (( look_for_src && comm->src_host != NULL) ||
  71         (!look_for_src && comm->dst_host != NULL)) {
  72
  73       SIMIX_communication_use(comm);
  74       return comm;
  75     }
  76   }
  77   xbt_fifo_unshift(rvp->comm_fifo, comm);
  78
  79   /* no relevant request found. Create a new comm action and set it up */
  80   comm = SIMIX_communication_new(rvp);
  81   SIMIX_rdv_push(rvp, comm);
  82   SIMIX_communication_use(comm);
  83
  84   return comm;
  85 }
  86
  87 /******************************************************************************/
  88 /*                           Communication Requests                           */
  89 /******************************************************************************/
  90
  91 /**
  92  *  \brief Creates a new communication request
  93  *  \param sender The process starting the communication (by send)
  94  *  \param receiver The process receiving the communication (by recv)
  95  *  \return the communication request
  96  */
  97 smx_comm_t SIMIX_communication_new(smx_rdv_t rdv)
  98 {
  99   /* alloc structures */
 100   smx_comm_t comm = xbt_new0(s_smx_comm_t, 1);
 101   comm->cond = SIMIX_cond_init();
 102   comm->rdv = rdv;
 103
 104   return comm;
 105 }
 106
 107 /**
 108  *  \brief Destroy a communication request
 109  *  \param comm The request to be destroyed
 110  */
 111 void SIMIX_communication_destroy(smx_comm_t comm)
 112 {
 113   comm->refcount--;
 114   if(comm->refcount == 0){
 115     if(comm->act != NULL)
 116       SIMIX_action_destroy(comm->act);
 117
 118     xbt_free(comm->cond);
 119     xbt_free(comm);
 120   }
 121 }
 122
 123 /**
 124  *  \brief Increase the number of users of the communication.
 125  *  \param comm The communication request
 126  *  Each communication request can be used by more than one process, so it is
 127  *  necessary to know number of them at destroy time, to avoid freeing stuff that
 128  *  maybe is in use by others.
 129  *  \
 130  */
 131 static inline void SIMIX_communication_use(smx_comm_t comm)
 132 {
 133   comm->refcount++;
 134 }
 135
 136 /**
 137  *  \brief Start the simulation of a communication request
 138  *  \param comm The communication request
 139  */
 140 static inline void SIMIX_communication_start(smx_comm_t comm)
 141 {
 142   comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL,
 143       comm->task_size, comm->rate);
 144
 145   SIMIX_register_action_to_condition(comm->act, comm->cond);
 146 }
 147
 148 /**
 149  *  \brief Waits for communication completion and performs error checking
 150  *  \param comm The communication
 151  *  \param timeout The max amount of time to wait for the communication to finish
 152  *
 153  *  Throws:
 154  *   - host_error if peer failed
 155  *   - timeout_error if communication reached the timeout specified
 156  *   - network_error if network failed or peer issued a timeout
 157  */
 158 static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, double timeout)
 159 {
 160   xbt_ex_t e;
 161
 162   if(timeout > 0){
 163     TRY{
 164       SIMIX_cond_wait_timeout(comm->cond, NULL, timeout);
 165     }
 166     CATCH(e){
 167       /* If it's a timeout then cancel the communication and signal the other peer */
 168       if(e.category == timeout_error)
 169         SIMIX_action_cancel(comm->act);
 170         SIMIX_cond_signal(comm->cond);
 171       RETHROW;
 172     }
 173   }else{
 174     SIMIX_cond_wait(comm->cond, NULL);
 175   }
 176
 177   SIMIX_unregister_action_to_condition(comm->act, comm->cond);
 178
 179   /* Check for errors */
 180   if (SIMIX_host_get_state(comm->dst_host) == 0){
 181     THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name);
 182   } else if (SIMIX_host_get_state(comm->src_host) == 0){
 183     THROW1(host_error, 0, "Source host %s failed", comm->src_host->name);
 184   } else if (SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){
 185     THROW0(network_error, 0, "Link failure");
 186   }
 187 }
 188
 189 /******************************************************************************/
 190 /*                        Synchronous Communication                           */
 191 /******************************************************************************/
 192 /*  Throws:
 193  *   - host_error if peer failed
 194  *   - timeout_error if communication reached the timeout specified
 195  *   - network_error if network failed or peer issued a timeout
 196  */
 197 void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, double timeout, void *data, size_t data_size)
 198 {
 199   smx_comm_t comm;
 200
 201   /* Setup communication request */
 202   comm = SIMIX_rdv_get_request_or_create(rdv, 0);
 203   comm->src_host = SIMIX_host_self();
 204   comm->task_size = task_size;
 205   comm->rate = rate;
 206   comm->data = data;
 207   comm->data_size = data_size;
 208
 209   /* If the receiver is already there, start the communication */
 210   /* If it's not here already, it will start that communication itself later on */
 211   if(comm->dst_host != NULL)
 212     SIMIX_communication_start(comm);
 213
 214   /* Wait for communication completion */
 215   /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 216   /* FIXME: add timeout checking stuff */
 217   SIMIX_communication_wait_for_completion(comm, timeout);
 218
 219   SIMIX_communication_destroy(comm);
 220 }
 221
 222 /*  Throws:
 223  *   - host_error if peer failed
 224  *   - timeout_error if communication reached the timeout specified
 225  *   - network_error if network failed or peer issued a timeout
 226  */
 227 void SIMIX_network_recv(smx_rdv_t rdv, double timeout, void **data, size_t *data_size)
 228 {
 229   smx_comm_t comm;
 230
 231   /* Setup communication request */
 232   comm = SIMIX_rdv_get_request_or_create(rdv, 1);
 233   comm->dst_host = SIMIX_host_self();
 234   comm->dest_buff = data;
 235   comm->dest_buff_size = data_size;
 236
 237   /* If the sender is already there, start the communication */
 238   /* If it's not here already, it will start that communication itself later on */
 239   if (comm->src_host != NULL)
 240     SIMIX_communication_start(comm);
 241
 242   /* Wait for communication completion */
 243   /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 244   /* FIXME: add timeout checking stuff */
 245   SIMIX_communication_wait_for_completion(comm, timeout);
 246
 247   SIMIX_communication_destroy(comm);
 248 }
 249
 250 /******************************************************************************/
 251 /*                        Asynchronous Communication                          */
 252 /******************************************************************************/
 253
 254 /*
 255 void SIMIX_network_wait(smx_action_t comm, double timeout)
 256 {
 257     if (timeout > 0)
 258       SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time);
 259     else
 260       SIMIX_cond_wait(rvp_cond, rvp_comm_mutex);
 261
 262 }
 263
 264
 265 XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm)
 266 {
 267   if(SIMIX_action_get_state (comm) == SURF_ACTION_DONE){
 268     memcpy(comm->data
 269
 270   return SIMIX_action_get_state (comm) == SURF_ACTION_DONE ? TRUE : FALSE;
 271 }*/
 272
 273
 274
 275   /* FIXME: MOVE DATA COPY TO MAESTRO AT ACTION SIGNAL TIME
 276    We are OK, let's copy the message to receiver's buffer
 277   *size = *size < comm->data_size ? *size : comm->data_size;
 278   memcpy(*data, comm->data, *size);*/
 279
 280
 281
 282