src/amok/Bandwidth/saturate.c

   1 /* $Id$ */
   2
   3 /* amok_saturate - Link saturating facilities (for ALNeM's BW testing)      */
   4
   5 /* Copyright (c) 2003, 2004 Martin Quinson. All rights reserved.            */
   6
   7 /* This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the license (GNU LGPL) which comes with this package. */
   9
  10 #include "amok/Bandwidth/bandwidth_private.h"
  11 #include "gras/Msg/msg_private.h"       /* FIXME: This mucks with contextes to answer RPC directly */
  12 #include "xbt/ex.h"
  13
  14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(amok_bw_sat, amok_bw,
  15                                 "Everything concerning the SATuration part of the amok_bw module");
  16
  17 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload);
  18 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload);
  19
  20
  21 void amok_bw_sat_init(void)
  22 {
  23   gras_datadesc_type_t bw_res_desc = gras_datadesc_by_name("bw_res_t");
  24   gras_datadesc_type_t sat_request_desc;
  25   /* Build the saturation datatype descriptions */
  26
  27   sat_request_desc = gras_datadesc_struct("s_sat_request_desc_t");
  28   gras_datadesc_struct_append(sat_request_desc, "peer",
  29                               gras_datadesc_by_name("s_xbt_peer_t"));
  30   gras_datadesc_struct_append(sat_request_desc, "msg_size",
  31                               gras_datadesc_by_name("unsigned int"));
  32   gras_datadesc_struct_append(sat_request_desc, "duration",
  33                               gras_datadesc_by_name("unsigned int"));
  34   gras_datadesc_struct_close(sat_request_desc);
  35   sat_request_desc = gras_datadesc_ref("sat_request_t", sat_request_desc);
  36
  37   /* Register the saturation messages */
  38   gras_msgtype_declare_rpc("amok_bw_sat start", sat_request_desc, NULL);
  39   gras_msgtype_declare_rpc("amok_bw_sat begin", sat_request_desc,
  40                            sat_request_desc);
  41   gras_msgtype_declare_rpc("amok_bw_sat stop", NULL, bw_res_desc);
  42
  43 }
  44
  45 void amok_bw_sat_join(void)
  46 {
  47   gras_cb_register("amok_bw_sat start", &amok_bw_cb_sat_start);
  48   gras_cb_register("amok_bw_sat begin", &amok_bw_cb_sat_begin);
  49 }
  50
  51 void amok_bw_sat_leave(void)
  52 {
  53   gras_cb_unregister("amok_bw_sat start", &amok_bw_cb_sat_start);
  54   gras_cb_unregister("amok_bw_sat begin", &amok_bw_cb_sat_begin);
  55 }
  56
  57 /* ***************************************************************************
  58  * Link saturation
  59  * ***************************************************************************/
  60
  61 /**
  62  * @brief Ask 'from_name:from_port' to stop saturating going to to_name:to_name.
  63  *
  64  * @param from_name: Name of the peer we are asking to do a experiment with (to_name:to_port)
  65  * @param from_port: port on which the process we are asking for an experiment is listening
  66  * (for message, do not give a raw socket here. The needed raw socket will be negociated
  67  * between the peers)
  68  * @param to_name: Name of the peer with which we should conduct the experiment
  69  * @param to_port: port on which the peer process is listening for message
  70  * @param msg_size: Size of each message sent.
  71  * @param duration: How long in maximum should be the saturation.
  72  *
  73  * Ask the process 'from_name:from_port' to start to saturate the link between itself
  74  * and to_name:to_name.
  75  */
  76 void amok_bw_saturate_start(const char *from_name, unsigned int from_port,
  77                             const char *to_name, unsigned int to_port,
  78                             unsigned int msg_size, double duration)
  79 {
  80   gras_socket_t sock;
  81   sat_request_t request = xbt_new(s_sat_request_t, 1);
  82
  83   VERB4("Start from_name %s:%d -> to_name %s:%d",
  84         from_name, from_port, to_name, to_port);
  85   sock = gras_socket_client(from_name, from_port);
  86
  87   request->peer.name = (char *) to_name;
  88   request->peer.port = to_port;
  89
  90   request->duration = duration;
  91   request->msg_size = msg_size;
  92
  93   gras_msg_rpccall(sock, 60, "amok_bw_sat start", &request, NULL);
  94
  95   free(request);
  96   gras_socket_close(sock);
  97 }
  98
  99 /* Asked to begin a saturation */
 100 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload)
 101 {
 102   sat_request_t request = *(sat_request_t *) payload;
 103   gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
 104
 105   VERB4("Asked by %s:%d to start a saturation to %s:%d",
 106         gras_socket_peer_name(expeditor), gras_socket_peer_port(expeditor),
 107         request->peer.name, request->peer.port);
 108
 109   gras_msg_rpcreturn(60, ctx, NULL);
 110
 111   amok_bw_saturate_begin(request->peer.name, request->peer.port,
 112                          request->msg_size, request->duration, NULL, NULL);
 113
 114   free(request->peer.name);
 115
 116   free(request);
 117   return 0;
 118 }
 119
 120 /**
 121  * @brief Start saturating between the current process and the designated peer
 122  *
 123  * Note that the only way to break this function before the end of the timeout
 124  * is to have a remote peer calling amok_bw_saturate_stop to this process.
 125  *
 126  * If duration=0, the experiment will never timeout (you then have to manually
 127  * stop it).
 128  *
 129  * If msg_size=0, the size will be automatically computed to make sure that
 130  * each of the messages occupy the connexion one second
 131  */
 132 void amok_bw_saturate_begin(const char *to_name, unsigned int to_port,
 133                             unsigned int msg_size, double duration,
 134                             /*out */ double *elapsed_res, double *bw_res)
 135 {
 136
 137   xbt_ex_t e;
 138
 139   gras_socket_t peer_cmd = gras_socket_client(to_name, to_port);
 140   gras_msg_cb_ctx_t ctx;
 141
 142   gras_socket_t meas;
 143
 144   s_gras_msg_t msg_got;
 145
 146   unsigned int packet_sent = 0;
 147   double start, elapsed = -1;   /* timer */
 148   double bw;
 149
 150   volatile int saturate_further;        /* boolean in the main loop */
 151
 152   /* Negociate the saturation with the peer */
 153   sat_request_t request = xbt_new(s_sat_request_t, 1);
 154
 155   DEBUG2("Begin to saturate to %s:%d", to_name, to_port);
 156   memset(&msg_got, 0, sizeof(msg_got));
 157
 158   request->msg_size = msg_size;
 159   request->duration = duration;
 160   request->peer.name = NULL;
 161   request->peer.port = 0;
 162
 163
 164   /* Size autodetection on need */
 165   if (!msg_size) {
 166     double bw;
 167     double sec;
 168     amok_bw_test(peer_cmd, 0,   /* check buffsize yourself */
 169                  512 * 1024,    /* 512k as first guess */
 170                  1,             /* One packet only */
 171                  1,             /* at least one sec */
 172                  &sec, &bw);
 173     msg_size = request->msg_size = (int) bw;
 174     DEBUG1("Saturate with packets of %d bytes", request->msg_size);
 175   }
 176
 177   /* Launch the saturation */
 178
 179   ctx = 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   INFO4("Saturation(%s:%d->%s:%d) 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     gras_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     DEBUG3("elapsed %f duration %f (msg_size=%d)", 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     INFO6("Saturation from %s:%d to %s:%d stopped by %s:%d",
 236           gras_os_myname(), gras_os_myport(), to_name, to_port,
 237           gras_socket_peer_name(msg_got.expe),
 238           gras_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     INFO6
 251       ("Saturation from %s:%d to %s:%d 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   gras_socket_t measMaster = NULL, meas = NULL;
 267   gras_socket_t from = gras_msg_cb_ctx_from(ctx);
 268
 269   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         RETHROW0
 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 = gras_socket_meas_accept(measMaster);
 295     DEBUG0("saturation handshake answered");
 296   }
 297   CATCH(e) {
 298     gras_socket_close(measMaster);
 299     RETHROW0("Error during saturation handshake: %s");
 300   }
 301
 302   while (saturate_further) {
 303     TRY {
 304       gras_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   INFO4("Saturation comming from %s:%d stopped on %s:%d",
 312         gras_socket_peer_name(from), gras_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_ex_t e;
 334
 335   gras_socket_t sock = gras_socket_client(from_name, from_port);
 336   bw_res_t answer;
 337   VERB2("Ask %s:%d to stop the saturation", from_name, from_port);
 338   TRY {
 339     gras_msg_rpccall(sock, 60, "amok_bw_sat stop", NULL, &answer);
 340   } CATCH(e) {
 341     RETHROW2("Cannot ask %s:%d to stop saturation: %s", from_name, from_port);
 342   }
 343   gras_socket_close(sock);
 344   if (time)
 345     *time = answer->sec;
 346   if (bw)
 347     *bw = answer->bw;
 348   free(answer);
 349 }