src/amok/bandwidth.c

   1 /* $Id$ */
   2
   3 /* amok_bandwidth - Bandwidth tests facilities    */
   4
   5 /* Copyright (c) 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 <stdio.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13
  14 #include "gras/messages.h"
  15 #include "amok/bandwidth.h"
  16
  17 /**
  18  * bw_request_t:
  19  *
  20  * Request for a BW experiment.
  21  * If host==NULL, it should be between the sender and the receiver.
  22  * If not, it should be between between the receiver and @host (3-tiers).
  23  */
  24 typedef struct {
  25   xbt_host_t host; /* host+raw socket to use */
  26   unsigned int buf_size;
  27   unsigned int exp_size;
  28   unsigned int msg_size;
  29 } bw_request_t;
  30
  31 /**
  32  * bw_res_t:
  33  *
  34  * Result of a BW experiment (payload when answering).
  35  * if err.msg != NULL, it wasn't sucessful. Check err.msg and err.code to see why.
  36  * else
  37  */
  38 typedef struct {
  39   amok_remoterr_t err;
  40   unsigned int timestamp;
  41   double seconds;
  42   double bw;
  43 } bw_res_t;
  44
  45
  46 /**
  47  * sat_request_t:
  48  *
  49  * Description of a saturation experiment (payload asking some host to collaborate for that)
  50  */
  51 typedef struct {
  52   xbt_host_t host; /* host+raw socket to use */
  53   unsigned int msg_size;
  54   unsigned int timeout;
  55 } sat_request_t;
  56
  57 /* Prototypes of local callbacks */
  58 int amok_bw_cb_bw_handshake(gras_socket_t *expeditor,
  59                             void          *payload);
  60 int amok_bw_cb_bw_request(gras_socket_t   *expeditor,
  61                           void            *payload);
  62
  63 int amok_bw_cb_sat_start(gras_socket_t    *expeditor,
  64                          void             *payload);
  65 int amok_bw_cb_sat_begin(gras_socket_t    *expeditor,
  66                          void             *payload);
  67
  68 /**** code ****/
  69 void amok_bw_init(void) {
  70   xbt_error_t errcode;
  71   gras_datadesc_type_t *bw_request_desc, *bw_res_desc, *sat_request_desc;
  72
  73   amok_base_init();
  74
  75   /* Build the datatype descriptions */
  76   bw_request_desc = gras_datadesc_struct("bw_request_t");
  77   gras_datadesc_struct_append(bw_request_desc,"host",gras_datadesc_by_name("xbt_host_t*"));
  78   gras_datadesc_struct_append(bw_request_desc,"buf_size",gras_datadesc_by_name("unsigned int"));
  79   gras_datadesc_struct_append(bw_request_desc,"exp_size",gras_datadesc_by_name("unsigned int"));
  80   gras_datadesc_struct_append(bw_request_desc,"msg_size",gras_datadesc_by_name("unsigned int"));
  81   gras_datadesc_struct_close(bw_request_desc);
  82   bw_request_desc = gras_datadesc_ref("bw_request_t*",bw_request_desc);
  83
  84   bw_res_desc = gras_datadesc_struct("bw_res_t");
  85   gras_datadesc_struct_append(bw_res_desc,"err",gras_datadesc_by_name("amok_remoterr_t"));
  86   gras_datadesc_struct_append(bw_res_desc,"timestamp",gras_datadesc_by_name("unsigned int"));
  87   gras_datadesc_struct_append(bw_res_desc,"seconds",gras_datadesc_by_name("double"));
  88   gras_datadesc_struct_append(bw_res_desc,"bw",gras_datadesc_by_name("double"));
  89   gras_datadesc_struct_close(bw_res_desc);
  90   bw_res_desc = gras_datadesc_ref("bw_res_t*",bw_res_desc);
  91
  92   sat_request_desc = gras_datadesc_struct("sat_request_desc");
  93   gras_datadesc_struct_append(sat_request_desc,"host",gras_datadesc_by_name("xbt_host_t"));
  94   gras_datadesc_struct_append(sat_request_desc,"msg_size",gras_datadesc_by_name("unsigned int"));
  95   gras_datadesc_struct_append(sat_request_desc,"timeout",gras_datadesc_by_name("unsigned int"));
  96   gras_datadesc_struct_close(sat_request_desc);
  97   sat_request_desc = gras_datadesc_ref("sat_request_t*",sat_request_desc);
  98
  99   /* Register the bandwidth messages */
 100   gras_msgtype_declare("BW request",       bw_request_desc);
 101   gras_msgtype_declare("BW result",        bw_res_desc);
 102   gras_msgtype_declare("BW handshake",     bw_request_desc);
 103   gras_msgtype_declare("BW handshake ACK", bw_request_desc);
 104
 105   /* Register the saturation messages */
 106   gras_msgtype_declare("SAT start",   sat_request_desc);
 107   gras_msgtype_declare("SAT started", gras_datadesc_by_name("amok_remoterr_t"));
 108   gras_msgtype_declare("SAT begin",   sat_request_desc);
 109   gras_msgtype_declare("SAT begun",   gras_datadesc_by_name("amok_remoterr_t"));
 110   gras_msgtype_declare("SAT end",     NULL);
 111   gras_msgtype_declare("SAT ended",   gras_datadesc_by_name("amok_remoterr_t"));
 112   gras_msgtype_declare("SAT stop",    NULL);
 113   gras_msgtype_declare("SAT stopped", gras_datadesc_by_name("amok_remoterr_t"));
 114
 115   /* Register the callbacks */
 116   gras_cb_register(gras_msgtype_by_name("BW request"),
 117                    &amok_bw_cb_bw_request);
 118   gras_cb_register(gras_msgtype_by_name("BW handshake"),
 119                    &amok_bw_cb_bw_handshake);
 120
 121   gras_cb_register(gras_msgtype_by_name("SAT start"),
 122                    &amok_bw_cb_sat_start);
 123   gras_cb_register(gras_msgtype_by_name("SAT begin"),
 124                    &amok_bw_cb_sat_begin);
 125 }
 126
 127 /* ***************************************************************************
 128  * Bandwidth tests
 129  * ***************************************************************************/
 130 /* Function to do a test from local to given host */
 131 /**
 132  * amok_bw_test:
 133  *
 134  * Conduct a test between the local host and @peer, and
 135  * report the result in last args
 136  */
 137 xbt_error_t amok_bw_test(gras_socket_t *peer,
 138                           unsigned int bufSize,unsigned int expSize,unsigned int msgSize,
 139                           /*OUT*/ double *sec, double *bw) {
 140   gras_socket_t *rawIn,*rawOut;
 141   int port;
 142   xbt_error_t errcode;
 143   bw_request_t *request;
 144   gras_msg_t *answer;
 145
 146   for (port = 5000, errcode = system_error;
 147        errcode == system_error;
 148        errcode = gras_socket_server_ext(++port,bufSize,1,&rawIn));
 149   if (errcode != no_error) {
 150     ERROR1("Error %s encountered while opening a raw socket\n",
 151            xbt_error_name(errcode));
 152     return errcode;
 153   }
 154
 155   request=xbt_new(bw_request_t);
 156   request->bufSize=bufSize;
 157   request->expSize=expSize;
 158   request->msgSize=msgSize;
 159   request->host.name = NULL;
 160   request->host.port = gras_socket_peer_port(rawIn);
 161
 162   if ((errcode=gras_send(peer,gras_msgtype_by_name("BW handshake"),request))) {
 163     ERROR1("Error %s encountered while sending the request.", xbt_error_name(errcode));
 164     gras_sock_close(sock);
 165     return errcode;
 166   }
 167   if ((errcode=gras_msg_wait(60,GRASMSG_BW_HANDSHAKED,&answer))) {
 168     fprintf(stderr,"grasbw_test(): Error %s encountered while waiting for the answer.\n",
 169             xbt_error_name(errcode));
 170     gras_sock_close(sock);
 171     return errcode;
 172   }
 173   if((errcode=gras_rawsock_client_open(to_name,gras_msg_ctn(answer,0,0,BwExp_t).port,
 174                                     bufSize,&rawOut))) {
 175     fprintf(stderr,"grasbw_test(): Error %s encountered while opening the raw socket to %s:%d\n",
 176             xbt_error_name(errcode),to_name,gras_msg_ctn(answer,0,0,BwExp_t).port);
 177     return errcode;
 178   }
 179
 180   *sec=gras_time();
 181   if ((errcode=gras_rawsock_send(rawOut,expSize,msgSize)) ||
 182       (errcode=gras_rawsock_recv(rawIn,1,1,120))) {
 183     fprintf(stderr,"grasbw_test(): Error %s encountered while sending the experiment.\n",
 184             xbt_error_name(errcode));
 185     gras_rawsock_close(rawOut);
 186     gras_rawsock_close(rawIn);
 187     return errcode;
 188   }
 189   *sec = gras_time() - *sec;
 190   *bw = ((double)expSize /* 8.0*/) / *sec / (1024.0 *1024.0);
 191
 192   gras_rawsock_close(rawIn);
 193   gras_rawsock_close(rawOut);
 194   gras_sock_close(sock);
 195   gras_msg_free(answer);
 196   return no_error;
 197 }
 198
 199 #if 0
 200
 201
 202 /* Callback to the GRASMSG_BW_HANDSHAKE message:
 203    opens a server raw socket,
 204    indicate its port in an answer GRASMSG_BW_HANDSHAKED message,
 205    receive the corresponding data on the raw socket,
 206    close the raw socket
 207 */
 208 int grasbw_cbBWHandshake(gras_msg_t *msg) {
 209   gras_rawsock_t *rawIn,*rawOut;
 210   BwExp_t *ans;
 211   xbt_error_t errcode;
 212
 213   if ((errcode=gras_rawsock_server_open(6666,8000,gras_msg_ctn(msg,0,0,BwExp_t).bufSize,&rawIn))) {
 214     fprintf(stderr,"grasbw_cbHandshake(): Error %s encountered while opening a raw socket\n",
 215             xbt_error_name(errcode));
 216     return 1;
 217   }
 218   if ((errcode=gras_rawsock_client_open(gras_sock_get_peer_name(msg->sock),gras_msg_ctn(msg,0,0,BwExp_t).port,
 219                                      gras_msg_ctn(msg,0,0,BwExp_t).bufSize,&rawOut))) {
 220     fprintf(stderr,"grasbw_cbHandshake(): Error %s encountered while opening a raw socket\n",
 221             xbt_error_name(errcode));
 222     return 1;
 223   }
 224   if (!(ans=(BwExp_t *)malloc(sizeof(BwExp_t)))) {
 225     fprintf(stderr,"grasbw_cbHandshake(): Malloc error.\n");
 226     gras_rawsock_close(rawIn);
 227     gras_rawsock_close(rawOut);
 228     return 1;
 229   }
 230   ans->bufSize=gras_msg_ctn(msg,0,0,BwExp_t).bufSize;
 231   ans->expSize=gras_msg_ctn(msg,0,0,BwExp_t).expSize;
 232   ans->msgSize=gras_msg_ctn(msg,0,0,BwExp_t).msgSize;
 233   ans->port=gras_rawsock_get_peer_port(rawIn);
 234   /*  fprintf(stderr,"grasbw_cbHandshake. bufSize=%d expSize=%d msgSize=%d port=%d\n",
 235           ans->bufSize,ans->expSize,ans->msgSize,ans->port);*/
 236
 237   if ((errcode=gras_msg_new_and_send(msg->sock,GRASMSG_BW_HANDSHAKED, 1,
 238                               ans, 1))) {
 239     fprintf(stderr,"grasbw_cbHandshake(): Error %s encountered while sending the answer.\n",
 240             xbt_error_name(errcode));
 241     gras_rawsock_close(rawIn);
 242     gras_rawsock_close(rawOut);
 243     return 1;
 244   }
 245
 246   if ((errcode=gras_rawsock_recv(rawIn,
 247                                gras_msg_ctn(msg,0,0,BwExp_t).expSize,
 248                                gras_msg_ctn(msg,0,0,BwExp_t).msgSize,
 249                                120)) ||
 250       (errcode=gras_rawsock_send(rawOut,1,1))) {
 251     fprintf(stderr,"grasbw_cbHandshake(): Error %s encountered while receiving the experiment.\n",
 252             xbt_error_name(errcode));
 253     gras_rawsock_close(rawIn);
 254     gras_rawsock_close(rawOut);
 255     return 1;
 256   }
 257   gras_msg_free(msg);
 258   gras_rawsock_close(rawIn);
 259   gras_rawsock_close(rawOut);
 260   return 1;
 261 }
 262
 263 /* function to request a BW test between to external hosts */
 264 xbt_error_t grasbw_request(const char* from_name,unsigned int from_port,
 265                            const char* to_name,unsigned int to_port,
 266                            unsigned int bufSize,unsigned int expSize,unsigned int msgSize,
 267                            /*OUT*/ double *sec, double*bw) {
 268
 269   gras_sock_t *sock;
 270   gras_msg_t *answer;
 271   xbt_error_t errcode;
 272   /* The request */
 273   BwExp_t *request;
 274   msgHost_t *target;
 275
 276   if((errcode=gras_sock_client_open(from_name,from_port,&sock))) {
 277     fprintf(stderr,"grasbw_request(): Error %s encountered while contacting the actuator\n",
 278             xbt_error_name(errcode));
 279     return errcode;
 280   }
 281   if (!(request=(BwExp_t *)malloc(sizeof(BwExp_t))) ||
 282       !(target=(msgHost_t*)malloc(sizeof(msgHost_t)))) {
 283     fprintf(stderr,"grasbw_test(): Malloc error\n");
 284     gras_sock_close(sock);
 285     return malloc_error;
 286   }
 287
 288   request->bufSize=bufSize;
 289   request->expSize=expSize;
 290   request->msgSize=msgSize;
 291   strcpy(target->host,to_name);
 292   target->port=to_port;
 293
 294   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_BW_REQUEST, 2,
 295                               target,1,
 296                               request,1))) {
 297     fprintf(stderr,"grasbw_request(): Error %s encountered while sending the request.\n",
 298             xbt_error_name(errcode));
 299     gras_sock_close(sock);
 300     return errcode;
 301   }
 302   if ((errcode=gras_msg_wait(240,GRASMSG_BW_RESULT,&answer))) {
 303     fprintf(stderr,"grasbw_request(): Error %s encountered while waiting for the answer.\n",
 304             xbt_error_name(errcode));
 305     gras_sock_close(sock);
 306     return errcode;
 307   }
 308
 309   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
 310     fprintf(stderr,"grasbw_request(): Peer reported error %s (%s).\n",
 311             xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
 312     gras_msg_free(answer);
 313     gras_sock_close(sock);
 314     return errcode;
 315   }
 316
 317   /*  fprintf(stderr,"sec=%p",gras_msg_ctn(answer,1,0,msgResult_t)); */
 318   *sec=gras_msg_ctn(answer,1,0,msgResult_t).value;
 319   *bw=gras_msg_ctn(answer,1,1,msgResult_t).value;
 320
 321   gras_msg_free(answer);
 322   gras_sock_close(sock);
 323   return no_error;
 324 }
 325
 326 int grasbw_cbBWRequest(gras_msg_t *msg) {
 327   /* specification of the test to run */
 328   char* to_name=gras_msg_ctn(msg,0,0,msgHost_t).host;
 329   unsigned int to_port=gras_msg_ctn(msg,0,0,msgHost_t).port;
 330
 331   unsigned int bufSize=gras_msg_ctn(msg,1,0,BwExp_t).bufSize;
 332   unsigned int expSize=gras_msg_ctn(msg,1,0,BwExp_t).expSize;
 333   unsigned int msgSize=gras_msg_ctn(msg,1,0,BwExp_t).msgSize;
 334   /* our answer */
 335   msgError_t *error;
 336   msgResult_t *res;
 337
 338   if (!(error=(msgError_t *)malloc(sizeof(msgError_t))) ||
 339       !(res=(msgResult_t *)malloc(sizeof(msgResult_t) * 2))) {
 340     fprintf(stderr,"%s:%d:grasbw_cbRequest: Malloc error\n",__FILE__,__LINE__);
 341     return malloc_error;
 342   }
 343
 344   if ((error->errcode=grasbw_test(to_name,to_port,bufSize,expSize,msgSize,
 345                                   &(res[0].value),&(res[1].value) ))) {
 346     fprintf(stderr,
 347             "%s:%d:grasbw_cbRequest: Error %s encountered while doing the test\n",
 348             __FILE__,__LINE__,xbt_error_name(error->errcode));
 349     strncpy(error->errmsg,"Error within grasbw_test",ERRMSG_LEN);
 350     gras_msg_new_and_send(msg->sock,GRASMSG_BW_RESULT,2,
 351                    error,1,
 352                    res,2);
 353     return 1;
 354   }
 355   res[0].timestamp = (unsigned int) gras_time();
 356   res[1].timestamp = (unsigned int) gras_time();
 357   gras_msg_new_and_send(msg->sock,GRASMSG_BW_RESULT,2,
 358                  error,1,
 359                  res,2);
 360   gras_msg_free(msg);
 361   return 1;
 362 }
 363
 364 /* ***************************************************************************
 365  * Link saturation
 366  * ***************************************************************************/
 367
 368 xbt_error_t grasbw_saturate_start(const char* from_name,unsigned int from_port,
 369                                   const char* to_name,unsigned int to_port,
 370                                   unsigned int msgSize, unsigned int timeout) {
 371   gras_sock_t *sock;
 372   xbt_error_t errcode;
 373   /* The request */
 374   SatExp_t *request;
 375   msgHost_t *target;
 376   /* answer */
 377   gras_msg_t *answer;
 378
 379   if((errcode=gras_sock_client_open(from_name,from_port,&sock))) {
 380     fprintf(stderr,"%s:%d:saturate_start(): Error %s encountered while contacting peer\n",
 381             __FILE__,__LINE__,xbt_error_name(errcode));
 382     return errcode;
 383   }
 384   if (!(request=(SatExp_t *)malloc(sizeof(SatExp_t))) ||
 385       !(target=(msgHost_t*)malloc(sizeof(msgHost_t)))) {
 386     fprintf(stderr,"%s:%d:saturate_start(): Malloc error\n",__FILE__,__LINE__);
 387     gras_sock_close(sock);
 388     return malloc_error;
 389   }
 390
 391   request->timeout=timeout;
 392   request->msgSize=msgSize;
 393
 394   strcpy(target->host,to_name);
 395   target->port=to_port;
 396
 397   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_SAT_START, 2,
 398                               target,1,
 399                               request,1))) {
 400     fprintf(stderr,"%s:%d:saturate_start(): Error %s encountered while sending the request.\n",
 401             __FILE__,__LINE__,xbt_error_name(errcode));
 402     gras_sock_close(sock);
 403     return errcode;
 404   }
 405   if ((errcode=gras_msg_wait(120,GRASMSG_SAT_STARTED,&answer))) {
 406     fprintf(stderr,"%s:%d:saturate_start(): Error %s encountered while waiting for the ACK.\n",
 407             __FILE__,__LINE__,xbt_error_name(errcode));
 408     gras_sock_close(sock);
 409     return errcode;
 410   }
 411
 412   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
 413     fprintf(stderr,"%s:%d:saturate_start(): Peer reported error %s (%s).\n",
 414             __FILE__,__LINE__,xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
 415     gras_msg_free(answer);
 416     gras_sock_close(sock);
 417     return errcode;
 418   }
 419
 420   gras_msg_free(answer);
 421   gras_sock_close(sock);
 422   return no_error;
 423 }
 424
 425 int grasbw_cbSatStart(gras_msg_t *msg) {
 426   gras_rawsock_t *raw;
 427   gras_sock_t *sock;
 428   xbt_error_t errcode;
 429   double start; /* time to timeout */
 430
 431   /* specification of the test to run */
 432   char* to_name=gras_msg_ctn(msg,0,0,msgHost_t).host;
 433   unsigned int to_port=gras_msg_ctn(msg,0,0,msgHost_t).port;
 434
 435   unsigned int msgSize=gras_msg_ctn(msg,1,0,SatExp_t).msgSize;
 436   unsigned int timeout=gras_msg_ctn(msg,1,0,SatExp_t).timeout;
 437   unsigned int raw_port;
 438
 439   /* The request */
 440   SatExp_t *request;
 441   /* answer */
 442   gras_msg_t *answer;
 443
 444   /*
 445   fprintf(stderr,"grasbw_cbSatStart(sd=%p)\n",msg->sock);
 446   fprintf(stderr,"(server=%d,raw=%d,fromPID=%d,toPID=%d,toHost=%p,toPort=%d,toChan=%d)\n",
 447           msg->sock->server_sock,msg->sock->raw_sock,msg->sock->from_PID,
 448           msg->sock->to_PID,msg->sock->to_host,msg->sock->to_port,msg->sock->to_chan);
 449   */
 450
 451   /* Negociate the saturation with the peer */
 452   if((errcode=gras_sock_client_open(to_name,to_port,&sock))) {
 453     fprintf(stderr,"cbSatStart(): Error %s encountered while contacting peer\n",
 454             xbt_error_name(errcode));
 455     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 456                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 457                      errcode,"Cannot contact peer.\n");
 458     return 1;
 459   }
 460   if (!(request=(SatExp_t *)malloc(sizeof(SatExp_t)))) {
 461     fprintf(stderr,"cbSatStart(): Malloc error\n");
 462     gras_sock_close(sock);
 463     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 464                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 465                      malloc_error,"Cannot build request.\n");
 466     return 1;
 467   }
 468
 469   request->timeout=gras_msg_ctn(msg,1,0,SatExp_t).timeout;
 470   request->msgSize=gras_msg_ctn(msg,1,0,SatExp_t).msgSize;
 471
 472   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_SAT_BEGIN, 1,
 473                               request,1))) {
 474     fprintf(stderr,"cbSatStart(): Error %s encountered while sending the request.\n",
 475             xbt_error_name(errcode));
 476     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 477                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 478                      errcode,"Cannot send request.\n");
 479     gras_sock_close(sock);
 480     return 1;
 481   }
 482
 483   if ((errcode=gras_msg_wait(120,GRASMSG_SAT_BEGUN,&answer))) {
 484     fprintf(stderr,"cbSatStart(): Error %s encountered while waiting for the ACK.\n",
 485             xbt_error_name(errcode));
 486     gras_sock_close(sock);
 487
 488     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 489                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 490                      errcode,
 491                      "Cannot receive the ACK.\n");
 492     return 1;
 493   }
 494
 495   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
 496     fprintf(stderr,"cbSatStart(): Peer reported error %s (%s).\n",
 497             xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
 498
 499     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 500                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 501                      errcode,
 502                      "Peer repported '%s'.\n",gras_msg_ctn(answer,0,0,msgError_t).errmsg);
 503     gras_msg_free(answer);
 504     gras_sock_close(sock);
 505     return 1;
 506   }
 507
 508   raw_port=gras_msg_ctn(answer,1,0,SatExp_t).port;
 509
 510   if ((errcode=gras_rawsock_client_open(to_name,raw_port,msgSize,&raw))) {
 511     fprintf(stderr,"cbSatStart(): Error %s while opening raw socket to %s:%d.\n",
 512             xbt_error_name(errcode),to_name,gras_msg_ctn(answer,1,0,SatExp_t).port);
 513
 514     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 515                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 516                      errcode,"Cannot open raw socket.\n");
 517     gras_sock_close(sock);
 518     return 1;
 519   }
 520
 521   /* send a train of data before repporting that XP is started */
 522   if ((errcode=gras_rawsock_send(raw,msgSize,msgSize))) {
 523     fprintf(stderr,"cbSatStart: Failure %s during raw send\n",xbt_error_name(errcode));
 524     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 525                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 526                      errcode,"Cannot raw send.\n");
 527     gras_sock_close(sock);
 528     gras_rawsock_close(raw);
 529     return 1;
 530   }
 531
 532   grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
 533                    "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 534                    no_error,"Saturation started");
 535   gras_msg_free(answer);
 536   gras_msg_free(msg);
 537
 538   /* Do the saturation until we get a SAT_STOP message or until we timeout the whole XP*/
 539   start=gras_time();
 540   while (gras_msg_wait(0,GRASMSG_SAT_STOP,&msg)==timeout_error &&
 541          gras_time()-start < timeout) {
 542     if ((errcode=gras_rawsock_send(raw,msgSize,msgSize))) {
 543       fprintf(stderr,"cbSatStart: Failure %s during raw send\n",xbt_error_name(errcode));
 544       /* our error message do not interess anyone. SAT_STOP will do nothing. */
 545       gras_sock_close(sock);
 546       gras_rawsock_close(raw);
 547       return 1;
 548     }
 549   }
 550   if (gras_time()-start > timeout) {
 551     fprintf(stderr,"The saturation experiment did timeout. Stop it NOW\n");
 552     gras_sock_close(sock);
 553     gras_rawsock_close(raw);
 554     return 1;
 555   }
 556
 557   /* Handle the SAT_STOP which broke the previous while */
 558
 559   if ((errcode=gras_msg_new_and_send(sock, GRASMSG_SAT_END,0))) {
 560     fprintf(stderr,"cbSatStart(): Cannot tell peer to stop saturation\n");
 561
 562     grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
 563                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 564                      errcode,"Sending SAT_END to peer failed.\n");
 565     gras_sock_close(sock);
 566     gras_rawsock_close(raw);
 567     return 1;
 568   }
 569
 570   if ((errcode=gras_msg_wait(60,GRASMSG_SAT_ENDED,&answer))) {
 571     fprintf(stderr,"cbSatStart(): Peer didn't ACK the end\n");
 572
 573     grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
 574                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 575                      errcode,"Receiving SAT_ENDED from peer failed.\n");
 576     gras_sock_close(sock);
 577     gras_rawsock_close(raw);
 578     return 1;
 579   }
 580   grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
 581                    "cbSatStart: Severe error: Cannot send error status to requester!!\n",
 582                    no_error,"");
 583
 584   gras_sock_close(sock);
 585   gras_rawsock_close(raw);
 586   gras_msg_free(answer);
 587   gras_msg_free(msg);
 588
 589   return 1;
 590 }
 591
 592 int grasbw_cbSatBegin(gras_msg_t *msg) {
 593   gras_rawsock_t *raw;
 594   xbt_error_t errcode;
 595   double start; /* timer */
 596   /* request */
 597   unsigned int msgSize=gras_msg_ctn(msg,0,0,SatExp_t).msgSize;
 598   unsigned int timeout=gras_msg_ctn(msg,0,0,SatExp_t).timeout;
 599   /* answer */
 600   SatExp_t *request;
 601   msgError_t *error;
 602
 603   if (!(request=(SatExp_t*)malloc(sizeof(SatExp_t))) ||
 604       !(error=(msgError_t *)malloc(sizeof(msgError_t)))) {
 605     fprintf(stderr,"cbSatBegin(): Malloc error\n");
 606     grasRepportError(msg->sock,GRASMSG_SAT_BEGUN,2,
 607                      "cbSatBegin: Severe error: Cannot send error status to requester!!\n",
 608                      malloc_error,"Malloc error");
 609     return 1;
 610   }
 611
 612   if ((errcode=gras_rawsock_server_open(6666,8000,msgSize,&raw))) {
 613     fprintf(stderr,"cbSatBegin(): Error %s encountered while opening a raw socket\n",
 614             xbt_error_name(errcode));
 615     grasRepportError(msg->sock,GRASMSG_SAT_BEGUN,2,
 616                      "cbSatBegin: Severe error: Cannot send error status to requester!!\n",
 617                      errcode,"Cannot open raw socket");
 618     return 1;
 619   }
 620   request->port=gras_rawsock_get_peer_port(raw);
 621   request->msgSize=msgSize;
 622   error->errcode=no_error;
 623   error->errmsg[0]='\0';
 624   if ((errcode=gras_msg_new_and_send(msg->sock,GRASMSG_SAT_BEGUN,2,
 625                               error,1,
 626                               request,1))) {
 627     fprintf(stderr,"cbSatBegin(): Error %s encountered while send ACK to peer\n",
 628             xbt_error_name(errcode));
 629     return 1;
 630   }
 631   gras_msg_free(msg);
 632
 633   start=gras_time();
 634   while (gras_msg_wait(0,GRASMSG_SAT_END,&msg)==timeout_error &&
 635          gras_time() - start < timeout) {
 636     errcode=gras_rawsock_recv(raw,msgSize,msgSize,1);
 637     if (errcode != timeout_error && errcode != no_error) {
 638       fprintf(stderr,"cbSatBegin: Failure %s during raw receive\n",xbt_error_name(errcode));
 639       /* our error message do not interess anyone. SAT_END will do nothing. */
 640       /* (if timeout'ed, it may be because the sender stopped emission. so survive it) */
 641       return 1;
 642     }
 643   }
 644   if (gras_time()-start > timeout) {
 645     fprintf(stderr,"The saturation experiment did timeout. Stop it NOW.\n");
 646     gras_rawsock_close(raw);
 647     return 1;
 648   }
 649
 650   grasRepportError(msg->sock,GRASMSG_SAT_ENDED,1,
 651                    "cbSatBegin: Cannot send SAT_ENDED.\n",
 652                    no_error,"");
 653   gras_rawsock_close(raw);
 654   gras_msg_free(msg);
 655   return 1;
 656 }
 657
 658 xbt_error_t grasbw_saturate_stop(const char* from_name,unsigned int from_port,
 659                                  const char* to_name,unsigned int to_port) {
 660   xbt_error_t errcode;
 661   gras_sock_t *sock;
 662   gras_msg_t *answer;
 663
 664   if((errcode=gras_sock_client_open(from_name,from_port,&sock))) {
 665     fprintf(stderr,"saturate_stop(): Error %s encountered while contacting peer\n",
 666             xbt_error_name(errcode));
 667     return errcode;
 668   }
 669
 670   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_SAT_STOP,0))) {
 671     fprintf(stderr,"saturate_stop(): Error %s encountered while sending request\n",
 672             xbt_error_name(errcode));
 673     gras_sock_close(sock);
 674     return errcode;
 675   }
 676
 677   if ((errcode=gras_msg_wait(120,GRASMSG_SAT_STOPPED,&answer))) {
 678     fprintf(stderr,"saturate_stop(): Error %s encountered while receiving ACK\n",
 679             xbt_error_name(errcode));
 680     gras_sock_close(sock);
 681     return errcode;
 682   }
 683
 684   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
 685     fprintf(stderr,"saturate_stop(): Peer reported error %s (%s).\n",
 686             xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
 687     gras_msg_free(answer);
 688     gras_sock_close(sock);
 689     return errcode;
 690   }
 691
 692   gras_msg_free(answer);
 693   gras_sock_close(sock);
 694
 695   return no_error;
 696 }
 697 #endif