src/amok/Bandwidth/saturate.c

   1 /* amok_saturate - Link saturating facilities (for ALNeM's BW testing)      */
   2
   3 /* Copyright (c) 2004, 2005, 2006, 2007, 2009, 2010. The SimGrid Team.
   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 "amok/Bandwidth/bandwidth_private.h"
  10 #include "gras/Msg/msg_private.h"       /* FIXME: This mucks with contextes to answer RPC directly */
  11 #include "xbt/ex.h"
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(amok_bw_sat, amok_bw,
  14                                 "Everything concerning the SATuration part of the amok_bw module");
  15
  16 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload);
  17 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload);
  18
  19
  20 void amok_bw_sat_init(void)
  21 {
  22   gras_datadesc_type_t bw_res_desc = gras_datadesc_by_name("bw_res_t");
  23   gras_datadesc_type_t sat_request_desc;
  24   /* Build the saturation datatype descriptions */
  25
  26   sat_request_desc = gras_datadesc_struct("s_sat_request_desc_t");
  27   gras_datadesc_struct_append(sat_request_desc, "peer",
  28                               gras_datadesc_by_name("s_xbt_peer_t"));
  29   gras_datadesc_struct_append(sat_request_desc, "msg_size",
  30                               gras_datadesc_by_name("unsigned int"));
  31   gras_datadesc_struct_append(sat_request_desc, "duration",
  32                               gras_datadesc_by_name("unsigned int"));
  33   gras_datadesc_struct_close(sat_request_desc);
  34   sat_request_desc = gras_datadesc_ref("sat_request_t", sat_request_desc);
  35
  36   /* Register the saturation messages */
  37   gras_msgtype_declare_rpc("amok_bw_sat start", sat_request_desc, NULL);
  38   gras_msgtype_declare_rpc("amok_bw_sat begin", sat_request_desc,
  39                            sat_request_desc);
  40   gras_msgtype_declare_rpc("amok_bw_sat stop", NULL, bw_res_desc);
  41
  42 }
  43
  44 void amok_bw_sat_join(void)
  45 {
  46   gras_cb_register("amok_bw_sat start", &amok_bw_cb_sat_start);
  47   gras_cb_register("amok_bw_sat begin", &amok_bw_cb_sat_begin);
  48 }
  49
  50 void amok_bw_sat_leave(void)
  51 {
  52   gras_cb_unregister("amok_bw_sat start", &amok_bw_cb_sat_start);
  53   gras_cb_unregister("amok_bw_sat begin", &amok_bw_cb_sat_begin);
  54 }
  55
  56 /* ***************************************************************************
  57  * Link saturation
  58  * ***************************************************************************/
  59
  60 /**
  61  * @brief Ask 'from_name:from_port' to stop saturating going to to_name:to_name.
  62  *
  63  * @param from_name: Name of the peer we are asking to do a experiment with (to_name:to_port)
  64  * @param from_port: port on which the process we are asking for an experiment is listening
  65  * (for message, do not give a raw socket here. The needed raw socket will be negociated
  66  * between the peers)
  67  * @param to_name: Name of the peer with which we should conduct the experiment
  68  * @param to_port: port on which the peer process is listening for message
  69  * @param msg_size: Size of each message sent.
  70  * @param duration: How long in maximum should be the saturation.
  71  *
  72  * Ask the process 'from_name:from_port' to start to saturate the link between itself
  73  * and to_name:to_name.
  74  */
  75 void amok_bw_saturate_start(const char *from_name, unsigned int from_port,
  76                             const char *to_name, unsigned int to_port,
  77                             unsigned int msg_size, double duration)
  78 {
  79   gras_socket_t sock;
  80   sat_request_t request = xbt_new(s_sat_request_t, 1);
  81
  82   VERB4("Start from_name %s:%d -> to_name %s:%d",
  83         from_name, from_port, to_name, to_port);
  84   sock = gras_socket_client(from_name, from_port);
  85
  86   request->peer.name = (char *) to_name;
  87   request->peer.port = to_port;
  88
  89   request->duration = duration;
  90   request->msg_size = msg_size;
  91
  92   gras_msg_rpccall(sock, 60, "amok_bw_sat start", &request, NULL);
  93
  94   free(request);
  95   gras_socket_close(sock);
  96 }
  97
  98 /* Asked to begin a saturation */
  99 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload)
 100 {
 101   sat_request_t request = *(sat_request_t *) payload;
 102   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 103
 104   VERB4("Asked by %s:%d to start a saturation to %s:%d",
 105         gras_socket_peer_name(expeditor), gras_socket_peer_port(expeditor),
 106         request->peer.name, request->peer.port);
 107
 108   gras_msg_rpcreturn(60, ctx, NULL);
 109
 110   amok_bw_saturate_begin(request->peer.name, request->peer.port,
 111                          request->msg_size, request->duration, NULL, NULL);
 112
 113   free(request->peer.name);
 114
 115   free(request);
 116   return 0;
 117 }
 118
 119 /**
 120  * @brief Start saturating between the current process and the designated peer
 121  *
 122  * Note that the only way to break this function before the end of the timeout
 123  * is to have a remote peer calling amok_bw_saturate_stop to this process.
 124  *
 125  * If duration=0, the experiment will never timeout (you then have to manually
 126  * stop it).
 127  *
 128  * If msg_size=0, the size will be automatically computed to make sure that
 129  * each of the messages occupy the connexion one second
 130  */
 131 void amok_bw_saturate_begin(const char *to_name, unsigned int to_port,
 132                             unsigned int msg_size, double duration,
 133                             /*out */ double *elapsed_res, double *bw_res)
 134 {
 135
 136   xbt_ex_t e;
 137
 138   gras_socket_t peer_cmd = gras_socket_client(to_name, to_port);
 139   gras_msg_cb_ctx_t ctx;
 140
 141   gras_socket_t meas;
 142
 143   s_gras_msg_t msg_got;
 144
 145   unsigned int packet_sent = 0;
 146   double start, elapsed = -1;   /* timer */
 147   double bw;
 148
 149   volatile int saturate_further;        /* boolean in the main loop */
 150
 151   /* Negociate the saturation with the peer */
 152   sat_request_t request = xbt_new(s_sat_request_t, 1);
 153
 154   DEBUG2("Begin to saturate to %s:%d", to_name, to_port);
 155   memset(&msg_got, 0, sizeof(msg_got));
 156
 157   request->msg_size = msg_size;
 158   request->duration = duration;
 159   request->peer.name = NULL;
 160   request->peer.port = 0;
 161
 162
 163   /* Size autodetection on need */
 164   if (!msg_size) {
 165     double bw;
 166     double sec;
 167     amok_bw_test(peer_cmd, 0,   /* check buffsize yourself */
 168                  512 * 1024,    /* 512k as first guess */
 169                  1,             /* One packet only */
 170                  1,             /* at least one sec */
 171                  &sec, &bw);
 172     msg_size = request->msg_size = (int) bw;
 173     DEBUG1("Saturate with packets of %d bytes", request->msg_size);
 174   }
 175
 176   /* Launch the saturation */
 177
 178   ctx = gras_msg_rpc_async_call(peer_cmd, 60, "amok_bw_sat begin", &request);
 179   free(request);
 180   gras_msg_rpc_async_wait(ctx, &request);
 181   meas = gras_socket_client_ext(to_name, request->peer.port,
 182                                 0 /*bufsize: auto */ ,
 183                                 1 /*meas: true */ );
 184
 185   free(request);
 186
 187   gras_socket_close(peer_cmd);
 188   INFO4("Saturation(%s:%d->%s:%d) started", gras_os_myname(),
 189         gras_os_myport(), to_name, to_port);
 190
 191   /* Start experiment */
 192   start = gras_os_time();
 193
 194   do {
 195     /* do send it */
 196     gras_socket_meas_send(meas, 120, msg_size, 1);
 197     packet_sent++;
 198
 199     /* Check whether someone asked us to stop saturation */
 200     saturate_further = 0;
 201     TRY {
 202       gras_msg_wait_ext(0 /*no wait */ , "amok_bw_sat stop",
 203                         NULL /* accept any sender */ ,
 204                         NULL, NULL,     /* No specific filter */
 205                         &msg_got);
 206     }
 207     CATCH(e) {
 208       if (e.category == timeout_error) {
 209         saturate_further = 1;
 210         memset(&msg_got, 0, sizeof(msg_got));   /* may be overprotectiv here */
 211       }
 212       xbt_ex_free(e);
 213     }
 214
 215     /* Check whether the experiment has to be terminated by now */
 216     elapsed = gras_os_time() - start;
 217     DEBUG3("elapsed %f duration %f (msg_size=%d)", elapsed, duration,
 218            msg_size);
 219
 220   } while (saturate_further && (duration == 0 || elapsed < duration));
 221
 222   bw = ((double) (packet_sent * msg_size)) / elapsed;
 223
 224   if (elapsed_res)
 225     *elapsed_res = elapsed;
 226   if (bw_res)
 227     *bw_res = bw;
 228
 229   /* If someone stopped us, inform him about the achieved bandwidth */
 230   if (msg_got.expe) {
 231     bw_res_t answer = xbt_new(s_bw_res_t, 1);
 232     s_gras_msg_cb_ctx_t ctx;
 233
 234     INFO6("Saturation from %s:%d to %s:%d stopped by %s:%d",
 235           gras_os_myname(), gras_os_myport(), to_name, to_port,
 236           gras_socket_peer_name(msg_got.expe),
 237           gras_socket_peer_port(msg_got.expe));
 238     answer->timestamp = gras_os_time();
 239     answer->sec = elapsed;
 240     answer->bw = bw;
 241
 242     ctx.expeditor = msg_got.expe;
 243     ctx.ID = msg_got.ID;
 244     ctx.msgtype = msg_got.type;
 245
 246     gras_msg_rpcreturn(60, &ctx, &answer);
 247     free(answer);
 248   } else {
 249     INFO6
 250       ("Saturation from %s:%d to %s:%d elapsed after %f sec (achieving %f kb/s)",
 251        gras_os_myname(), gras_os_myport(), to_name, to_port, elapsed,
 252        bw / 1024.0);
 253   }
 254
 255   gras_socket_close(meas);
 256 }
 257
 258 /* Sender will saturate link to us */
 259 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload)
 260 {
 261   sat_request_t request = *(sat_request_t *) payload;
 262   sat_request_t answer = xbt_new0(s_sat_request_t, 1);
 263   volatile int saturate_further = 1;
 264   xbt_ex_t e;
 265   gras_socket_t measMaster = NULL, meas = NULL;
 266   gras_socket_t from = gras_msg_cb_ctx_from(ctx);
 267
 268   int port = 6000;
 269   while (port <= 10000 && measMaster == NULL) {
 270     TRY {
 271       measMaster = gras_socket_server_ext(port, 0 /*bufsize: auto */ ,
 272                                           1 /*meas: true */ );
 273     }
 274     CATCH(e) {
 275       measMaster = NULL;
 276       if (port < 10000)
 277         xbt_ex_free(e);
 278       else
 279         RETHROW0
 280           ("Error encountered while opening a measurement server socket: %s");
 281     }
 282     if (measMaster == NULL)
 283       port++;                   /* prepare for a new loop */
 284   }
 285   answer->peer.port = port;
 286
 287   gras_msg_rpcreturn(60, ctx, &answer);
 288   free(answer);
 289
 290   gras_os_sleep(5);             /* Wait for the accept */
 291
 292   TRY {
 293     meas = gras_socket_meas_accept(measMaster);
 294     DEBUG0("saturation handshake answered");
 295   }
 296   CATCH(e) {
 297     gras_socket_close(measMaster);
 298     RETHROW0("Error during saturation handshake: %s");
 299   }
 300
 301   while (saturate_further) {
 302     TRY {
 303       gras_socket_meas_recv(meas, 5, request->msg_size, 1);
 304     }
 305     CATCH(e) {
 306       saturate_further = 0;
 307       xbt_ex_free(e);
 308     }
 309   }
 310   INFO4("Saturation comming from %s:%d stopped on %s:%d",
 311         gras_socket_peer_name(from), gras_socket_peer_port(from),
 312         gras_os_myname(), gras_os_myport());
 313
 314   gras_socket_close(meas);
 315   if (gras_if_RL())             /* On SG, accepted=master */
 316     gras_socket_close(measMaster);
 317   free(request);
 318   return 0;
 319 }
 320
 321 /**
 322  * @brief Ask 'from_name:from_port' to stop any saturation experiments
 323  * @param from_name: Name of the peer we are asking to do a experiment with (to_name:to_port)
 324  * @param from_port: port on which the process we are asking for an experiment is listening
 325  * @param time: the duration of the experiment
 326  * @param bw: the achieved bandwidth
 327  *
 328  */
 329 void amok_bw_saturate_stop(const char *from_name, unsigned int from_port,
 330                            /*out */ double *time, double *bw)
 331 {
 332   xbt_ex_t e;
 333
 334   gras_socket_t sock = gras_socket_client(from_name, from_port);
 335   bw_res_t answer;
 336   VERB2("Ask %s:%d to stop the saturation", from_name, from_port);
 337   TRY {
 338     gras_msg_rpccall(sock, 60, "amok_bw_sat stop", NULL, &answer);
 339   } CATCH(e) {
 340     RETHROW2("Cannot ask %s:%d to stop saturation: %s", from_name, from_port);
 341   }
 342   gras_socket_close(sock);
 343   if (time)
 344     *time = answer->sec;
 345   if (bw)
 346     *bw = answer->bw;
 347   free(answer);
 348 }