Logo AND Algorithmique Numérique Distribuée

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