src/amok/bandwidth.c

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