src/amok/Bandwidth/saturate.c

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