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   xbt_datadesc_type_t bw_res_desc = xbt_datadesc_by_name("bw_res_t");
  23   xbt_datadesc_type_t sat_request_desc;
  24   /* Build the saturation datatype descriptions */
  25
  26   sat_request_desc = xbt_datadesc_struct("s_sat_request_desc_t");
  27   xbt_datadesc_struct_append(sat_request_desc, "peer",
  28                               xbt_datadesc_by_name("s_xbt_peer_t"));
  29   xbt_datadesc_struct_append(sat_request_desc, "msg_size",
  30                               xbt_datadesc_by_name("unsigned int"));
  31   xbt_datadesc_struct_append(sat_request_desc, "duration",
  32                               xbt_datadesc_by_name("unsigned int"));
  33   xbt_datadesc_struct_close(sat_request_desc);
  34   sat_request_desc = xbt_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   xbt_socket_t sock;
  80   sat_request_t request = xbt_new(s_sat_request_t, 1);
  81
  82   XBT_VERB("Start from_name %s:%u -> to_name %s:%u",
  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   xbt_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 103
 104   XBT_VERB("Asked by %s:%d to start a saturation to %s:%d",
 105         xbt_socket_peer_name(expeditor), xbt_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                             volatile unsigned int msg_size, double duration,
 133                             /*out */ double *elapsed_res, double *bw_res)
 134 {
 135
 136   xbt_ex_t e;
 137
 138   xbt_socket_t peer_cmd = gras_socket_client(to_name, to_port);
 139   gras_msg_cb_ctx_t ctx;
 140
 141   xbt_socket_t meas;
 142
 143   s_gras_msg_t msg_got;
 144
 145   volatile 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   XBT_DEBUG("Begin to saturate to %s:%u", 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     XBT_DEBUG("Saturate with packets of %u bytes", request->msg_size);
 174   }
 175
 176   /* Launch the saturation */
 177
 178   ctx =
 179       gras_msg_rpc_async_call(peer_cmd, 60, "amok_bw_sat begin", &request);
 180   free(request);
 181   gras_msg_rpc_async_wait(ctx, &request);
 182   meas = gras_socket_client_ext(to_name, request->peer.port,
 183                                 0 /*bufsize: auto */ ,
 184                                 1 /*meas: true */ );
 185
 186   free(request);
 187
 188   gras_socket_close(peer_cmd);
 189   XBT_INFO("Saturation(%s:%d->%s:%u) started", gras_os_myname(),
 190         gras_os_myport(), to_name, to_port);
 191
 192   /* Start experiment */
 193   start = gras_os_time();
 194
 195   do {
 196     /* do send it */
 197     xbt_socket_meas_send(meas, 120, msg_size, 1);
 198     packet_sent++;
 199
 200     /* Check whether someone asked us to stop saturation */
 201     saturate_further = 0;
 202     TRY {
 203       gras_msg_wait_ext(0 /*no wait */ , "amok_bw_sat stop",
 204                         NULL /* accept any sender */ ,
 205                         NULL, NULL,     /* No specific filter */
 206                         &msg_got);
 207     }
 208     CATCH(e) {
 209       if (e.category == timeout_error) {
 210         saturate_further = 1;
 211         memset(&msg_got, 0, sizeof(msg_got));   /* may be overprotectiv here */
 212       }
 213       xbt_ex_free(e);
 214     }
 215
 216     /* Check whether the experiment has to be terminated by now */
 217     elapsed = gras_os_time() - start;
 218     XBT_DEBUG("elapsed %f duration %f (msg_size=%u)", elapsed, duration,
 219            msg_size);
 220
 221   } while (saturate_further && (duration == 0 || elapsed < duration));
 222
 223   bw = ((double) (packet_sent * msg_size)) / elapsed;
 224
 225   if (elapsed_res)
 226     *elapsed_res = elapsed;
 227   if (bw_res)
 228     *bw_res = bw;
 229
 230   /* If someone stopped us, inform him about the achieved bandwidth */
 231   if (msg_got.expe) {
 232     bw_res_t answer = xbt_new(s_bw_res_t, 1);
 233     s_gras_msg_cb_ctx_t ctx;
 234
 235     XBT_INFO("Saturation from %s:%d to %s:%u stopped by %s:%d",
 236           gras_os_myname(), gras_os_myport(), to_name, to_port,
 237           xbt_socket_peer_name(msg_got.expe),
 238           xbt_socket_peer_port(msg_got.expe));
 239     answer->timestamp = gras_os_time();
 240     answer->sec = elapsed;
 241     answer->bw = bw;
 242
 243     ctx.expeditor = msg_got.expe;
 244     ctx.ID = msg_got.ID;
 245     ctx.msgtype = msg_got.type;
 246
 247     gras_msg_rpcreturn(60, &ctx, &answer);
 248     free(answer);
 249   } else {
 250     XBT_INFO
 251         ("Saturation from %s:%d to %s:%u elapsed after %f sec (achieving %f kb/s)",
 252          gras_os_myname(), gras_os_myport(), to_name, to_port, elapsed,
 253          bw / 1024.0);
 254   }
 255
 256   gras_socket_close(meas);
 257 }
 258
 259 /* Sender will saturate link to us */
 260 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload)
 261 {
 262   sat_request_t request = *(sat_request_t *) payload;
 263   sat_request_t answer = xbt_new0(s_sat_request_t, 1);
 264   volatile int saturate_further = 1;
 265   xbt_ex_t e;
 266   volatile xbt_socket_t measMaster = NULL, meas = NULL;
 267   volatile xbt_socket_t from = gras_msg_cb_ctx_from(ctx);
 268
 269   volatile int port = 6000;
 270   while (port <= 10000 && measMaster == NULL) {
 271     TRY {
 272       measMaster = gras_socket_server_ext(port, 0 /*bufsize: auto */ ,
 273                                           1 /*meas: true */ );
 274     }
 275     CATCH(e) {
 276       measMaster = NULL;
 277       if (port < 10000)
 278         xbt_ex_free(e);
 279       else
 280         RETHROWF
 281             ("Error encountered while opening a measurement server socket: %s");
 282     }
 283     if (measMaster == NULL)
 284       port++;                   /* prepare for a new loop */
 285   }
 286   answer->peer.port = port;
 287
 288   gras_msg_rpcreturn(60, ctx, &answer);
 289   free(answer);
 290
 291   gras_os_sleep(5);             /* Wait for the accept */
 292
 293   TRY {
 294     meas = xbt_socket_meas_accept(measMaster);
 295     XBT_DEBUG("saturation handshake answered");
 296   }
 297   CATCH_ANONYMOUS {
 298     gras_socket_close(measMaster);
 299     RETHROWF("Error during saturation handshake: %s");
 300   }
 301
 302   while (saturate_further) {
 303     TRY {
 304       xbt_socket_meas_recv(meas, 5, request->msg_size, 1);
 305     }
 306     CATCH(e) {
 307       saturate_further = 0;
 308       xbt_ex_free(e);
 309     }
 310   }
 311   XBT_INFO("Saturation comming from %s:%d stopped on %s:%d",
 312         xbt_socket_peer_name(from), xbt_socket_peer_port(from),
 313         gras_os_myname(), gras_os_myport());
 314
 315   gras_socket_close(meas);
 316   if (gras_if_RL())             /* On SG, accepted=master */
 317     gras_socket_close(measMaster);
 318   free(request);
 319   return 0;
 320 }
 321
 322 /**
 323  * @brief Ask 'from_name:from_port' to stop any saturation experiments
 324  * @param from_name: Name of the peer we are asking to do a experiment with (to_name:to_port)
 325  * @param from_port: port on which the process we are asking for an experiment is listening
 326  * @param time: the duration of the experiment
 327  * @param bw: the achieved bandwidth
 328  *
 329  */
 330 void amok_bw_saturate_stop(const char *from_name, unsigned int from_port,
 331                            /*out */ double *time, double *bw)
 332 {
 333   xbt_socket_t sock = gras_socket_client(from_name, from_port);
 334   bw_res_t answer;
 335   XBT_VERB("Ask %s:%u to stop the saturation", from_name, from_port);
 336   TRY {
 337     gras_msg_rpccall(sock, 60, "amok_bw_sat stop", NULL, &answer);
 338   }
 339   CATCH_ANONYMOUS {
 340     RETHROWF("Cannot ask %s:%u 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 }