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_rvpoint_t SIMIX_rvpoint_create(char *name)
  22 {
  23   smx_rvpoint_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_mutex = SIMIX_mutex_init();
  28   rvp->comm_fifo = xbt_fifo_new();
  29
  30   return rvp;
  31 }
  32
  33 /**
  34  *  \brief Destroy a rendez-vous point
  35  *  \param name The rendez-vous point to destroy
  36  */
  37 void SIMIX_rvpoint_destroy(smx_rvpoint_t rvp)
  38 {
  39   xbt_free(rvp->name);
  40   SIMIX_mutex_destroy(rvp->read);
  41   SIMIX_mutex_destroy(rvp->write);
  42   SIMIX_mutex_destroy(rvp->comm_mutex);
  43   xbt_fifo_free(rvp->comm_fifo);
  44   xbt_free(rvp);
  45 }
  46
  47 /**
  48  *  \brief Push a communication request into a rendez-vous point
  49  *  The communications request are dequeued by the two functions below
  50  *  \param rvp The rendez-vous point
  51  *  \param comm The communication request
  52  */
  53 static inline void SIMIX_rvpoint_push(smx_rvpoint_t rvp, smx_comm_t comm)
  54 {
  55   xbt_fifo_push(rvp->comm_fifo, comm);
  56 }
  57
  58 /**
  59  *  \brief Checks if there is a receive communication request queued in a rendez-vous
  60  *  \param rvp The rendez-vous with the queue
  61  *  \return The communication request if found, NULL otherwise.
  62  */
  63 smx_comm_t SIMIX_rvpoint_get_receiver(smx_rvpoint_t rvp)
  64 {
  65   /* Get a communication request from the rendez-vous queue. If it is a receive
  66      request then return it, otherwise put it again in the queue and return NULL
  67    */
  68   smx_comm_t comm_head = xbt_fifo_shift(rvp->comm_fifo);
  69
  70   if(comm_head != NULL && comm_head->dst_host != NULL)
  71     return comm_head;
  72
  73   xbt_fifo_unshift(rvp->comm_fifo, comm_head);
  74   return NULL;
  75 }
  76
  77 /**
  78  *  \brief Checks if there is a send communication request queued in a rendez-vous
  79  *  \param rvp The rendez-vous with the queue
  80  *  \return The communication request if found, NULL otherwise.
  81  */
  82 smx_comm_t SIMIX_rvpoint_get_sender(smx_rvpoint_t rvp)
  83 {
  84   /* Get a communication request from the rendez-vous queue. If it is a send
  85      request then return it, otherwise put it again in the queue and return NULL
  86    */
  87   smx_comm_t comm_head = xbt_fifo_shift(rvp->comm_fifo);
  88
  89   if(comm_head != NULL && comm_head->src_host != NULL)
  90     return comm_head;
  91
  92   xbt_fifo_unshift(rvp->comm_fifo, comm_head);
  93   return NULL;
  94 }
  95
  96 /**
  97  *  \brief Get the communication mutex of the rendez-vous point
  98  *  \param rvp The rendez-vous point
  99  */
 100 static inline smx_mutex_t SIMIX_rvpoint_get_comm_mutex(smx_rvpoint_t rvp)
 101 {
 102   return rvp->comm_mutex;
 103 }
 104
 105 /******************************************************************************/
 106 /*                           Communication Requests                           */
 107 /******************************************************************************/
 108
 109 /**
 110  *  \brief Creates a new communication request
 111  *  \param sender The process starting the communication (by send)
 112  *  \param receiver The process receiving the communication (by recv)
 113  *  \return the communication request
 114  */
 115 smx_comm_t SIMIX_communication_new(smx_host_t src_host, smx_host_t dst_host,
 116                                    smx_rvpoint_t rdv)
 117 {
 118   /* alloc structures */
 119   smx_comm_t comm = xbt_new0(s_smx_comm_t, 1);
 120   comm->cond = SIMIX_cond_init();
 121
 122   /* initialize them */
 123   comm->src_host = src_host;
 124   comm->dst_host = dst_host;
 125   comm->rdv = rdv;
 126
 127   return comm;
 128 }
 129
 130 /**
 131  *  \brief Destroy a communication request
 132  *  \param comm The request to be destroyed
 133  */
 134 void SIMIX_communication_destroy(smx_comm_t comm)
 135 {
 136   comm->refcount--;
 137   if(comm->refcount == 0){
 138     if(comm->act != NULL)
 139       SIMIX_action_destroy(comm->act);
 140
 141     if(comm->data != NULL)
 142       xbt_free(comm->data);
 143
 144     xbt_free(comm->cond);
 145     xbt_free(comm);
 146   }
 147 }
 148
 149 /**
 150  *  \brief Increase the number of users of the communication.
 151  *  \param comm The communication request
 152  *  Each communication request can be used by more than one process, so it is
 153  *  necessary to know number of them at destroy time, to avoid freeing stuff that
 154  *  maybe is in use by others.
 155  *  \
 156  */
 157 static inline void SIMIX_communication_use(smx_comm_t comm)
 158 {
 159   comm->refcount++;
 160 }
 161
 162 /**
 163  *  \brief Start the simulation of a communication request
 164  *  \param comm The communicatino request
 165  */
 166 static inline void SIMIX_communication_start(smx_comm_t comm)
 167 {
 168   comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL,
 169                                        comm->data_size, comm->rate);
 170
 171   SIMIX_register_action_to_condition(comm->act, comm->cond);
 172   __SIMIX_cond_wait(comm->cond);
 173   SIMIX_unregister_action_to_condition(comm->act, comm->cond);
 174 }
 175
 176 /******************************************************************************/
 177 /*                        Synchronous Communication                           */
 178 /******************************************************************************/
 179
 180 void SIMIX_network_send(smx_rvpoint_t rdv, void *data, size_t size, double timeout, double rate)
 181 {
 182   /*double start_time = SIMIX_get_clock();*/
 183   void *smx_net_data;
 184   smx_host_t my_host = SIMIX_host_self();
 185   smx_comm_t comm;
 186   smx_mutex_t rvp_comm_mutex = SIMIX_rvpoint_get_comm_mutex(rdv);
 187
 188   /* Lock the rendez-vous point */
 189   SIMIX_mutex_lock(rvp_comm_mutex);
 190   /* Copy the message to the network */
 191   /*FIXME here the MC should allocate the space from the network storage area */
 192   smx_net_data = xbt_malloc(size);
 193   memcpy(smx_net_data, data, size);
 194
 195   /* Check if there is already a receive waiting in the rendez-vous point */
 196   if((comm = SIMIX_rvpoint_get_receiver(rdv)) != NULL){
 197     comm->src_host = my_host;
 198     comm->data = smx_net_data;
 199     comm->data_size = size;
 200     comm->rate = rate;
 201     SIMIX_communication_use(comm);
 202
 203     /* Unlock the rendez-vous point and start the communication action.*/
 204     /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 205     SIMIX_mutex_unlock(rvp_comm_mutex);
 206     SIMIX_communication_start(comm);
 207
 208   /* Nobody is at the rendez-vous point, so push the comm action into it */
 209   }else{
 210     comm = SIMIX_communication_new(my_host, NULL, rdv);
 211     comm->data = smx_net_data;
 212     comm->data_size = size;
 213     comm->rate = rate;
 214     SIMIX_communication_use(comm);
 215     SIMIX_rvpoint_push(rdv, comm);
 216
 217     /* Wait for communication completion */
 218     /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 219     /* FIXME: add timeout checking stuff */
 220     SIMIX_mutex_unlock (rvp_comm_mutex);
 221     __SIMIX_cond_wait(comm->cond);
 222   }
 223
 224   /* Check for errors */
 225   if (SIMIX_host_get_state(comm->dst_host) == 0){
 226     THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name);
 227   }else if(SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){
 228     THROW0(network_error, 0, "Link failure");
 229   }
 230
 231   SIMIX_communication_destroy(comm);
 232   return;
 233 }
 234
 235 void SIMIX_network_recv(smx_rvpoint_t rdv, void **data, size_t *size, double timeout)
 236 {
 237   /*double start_time = SIMIX_get_clock();*/
 238   smx_comm_t comm;
 239   smx_host_t my_host = SIMIX_host_self();
 240   smx_mutex_t rvp_comm_mutex = SIMIX_rvpoint_get_comm_mutex(rdv);
 241
 242   /* Lock the rendez-vous point */
 243   SIMIX_mutex_lock(rvp_comm_mutex);
 244
 245   /* Check if there is already a send waiting in the rendez-vous point */
 246   if((comm = SIMIX_rvpoint_get_sender(rdv)) != NULL){
 247     comm->dst_host = my_host;
 248     comm->dest_buff = data;
 249     SIMIX_communication_use(comm);
 250
 251     /* Unlock the rendez-vous point and start the communication action.*/
 252     /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 253     SIMIX_mutex_unlock(rvp_comm_mutex);
 254     SIMIX_communication_start(comm);
 255
 256   /* Nobody is at the rendez-vous point, so push the comm action into it */
 257   }else{
 258     comm = SIMIX_communication_new(NULL, my_host, rdv);
 259     comm->dest_buff = data;
 260     SIMIX_communication_use(comm);
 261     SIMIX_rvpoint_push(rdv, comm);
 262
 263     /* Wait for communication completion */
 264     /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */
 265     /* FIXME: add timeout checking stuff*/
 266     SIMIX_mutex_unlock (rvp_comm_mutex);
 267     __SIMIX_cond_wait(comm->cond);
 268   }
 269
 270   /* Check for errors */
 271   if (SIMIX_host_get_state(comm->src_host) == 0){
 272     THROW1(host_error, 0, "Source host %s failed", comm->src_host->name);
 273   }else if(SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){
 274     THROW0(network_error, 0, "Link failure");
 275   }
 276
 277   /* We are OK, let's copy the message to receiver's buffer */
 278   *size = *size < comm->data_size ? *size : comm->data_size;
 279   memcpy(*data, comm->data, *size);
 280
 281   SIMIX_communication_destroy(comm);
 282   return;
 283 }
 284
 285 /******************************************************************************/
 286 /*                        Asynchronous Communication                          */
 287 /******************************************************************************/
 288
 289 /*
 290 void SIMIX_network_wait(smx_action_t comm, double timeout)
 291 {
 292     if (timeout > 0)
 293       SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time);
 294     else
 295       SIMIX_cond_wait(rvp_cond, rvp_comm_mutex);
 296
 297 }
 298
 299
 300 XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm)
 301 {
 302   if(SIMIX_action_get_state (comm) == SURF_ACTION_DONE){
 303     memcpy(comm->data
 304
 305   return SIMIX_action_get_state (comm) == SURF_ACTION_DONE ? TRUE : FALSE;
 306 }*/
 307
 308
 309
 310
 311
 312
 313