Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Plug a memleak and shut it up
[simgrid.git] / src / amok / Bandwidth / saturate.c
1 /* $Id$ */
2
3 /* amok_saturate - Link saturating facilities (for ALNeM's BW testing)      */
4
5 /* Copyright (c) 2003, 2004 Martin Quinson. All rights reserved.            */
6
7 /* This program is free software; you can redistribute it and/or modify it
8  * under the terms of the license (GNU LGPL) which comes with this package. */
9
10 #include "amok/Bandwidth/bandwidth_private.h"
11 #include "gras/Msg/msg_private.h" /* FIXME: This mucks with contextes to answer RPC directly */
12
13 XBT_LOG_EXTERNAL_CATEGORY(bw);
14 XBT_LOG_DEFAULT_CATEGORY(bw);
15
16 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload);
17 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload);
18
19
20 void amok_bw_sat_init(void) {
21   gras_datadesc_type_t sat_request_desc;
22   /* Build the saturation datatype descriptions */ 
23   
24   sat_request_desc = gras_datadesc_struct("s_sat_request_desc_t");
25   gras_datadesc_struct_append(sat_request_desc,"host",gras_datadesc_by_name("s_xbt_host_t"));
26   gras_datadesc_struct_append(sat_request_desc,"msg_size",gras_datadesc_by_name("unsigned int"));
27   gras_datadesc_struct_append(sat_request_desc,"duration",gras_datadesc_by_name("unsigned int"));
28   gras_datadesc_struct_close(sat_request_desc);
29   sat_request_desc = gras_datadesc_ref("sat_request_t",sat_request_desc);
30   
31   /* Register the saturation messages */
32   gras_msgtype_declare_rpc("amok_bw_sat start", sat_request_desc, NULL);
33   gras_msgtype_declare_rpc("amok_bw_sat begin", sat_request_desc, sat_request_desc);
34   gras_msgtype_declare_rpc("amok_bw_sat stop",  NULL,             NULL);
35
36 }
37 void amok_bw_sat_join(void) {
38   gras_cb_register(gras_msgtype_by_name("amok_bw_sat start"),
39                    &amok_bw_cb_sat_start);
40   gras_cb_register(gras_msgtype_by_name("amok_bw_sat begin"),
41                    &amok_bw_cb_sat_begin);
42 }
43 void amok_bw_sat_leave(void) {
44   gras_cb_unregister(gras_msgtype_by_name("amok_bw_sat start"),
45                      &amok_bw_cb_sat_start);
46   gras_cb_unregister(gras_msgtype_by_name("amok_bw_sat begin"),
47                      &amok_bw_cb_sat_begin);
48 }
49
50 /* ***************************************************************************
51  * Link saturation
52  * ***************************************************************************/
53
54 /**
55  * @brief Ask 'from_name:from_port' to stop saturating going to to_name:to_name.
56  *
57  * @param from_name: Name of the host we are asking to do a experiment with (to_name:to_port)
58  * @param from_port: port on which the process we are asking for an experiment is listening
59  * (for message, do not give a raw socket here. The needed raw socket will be negociated 
60  * between the peers)
61  * @param to_name: Name of the host with which we should conduct the experiment
62  * @param to_port: port on which the peer process is listening for message
63  * @param msg_size: Size of each message sent.
64  * @param duration: How long in maximum should be the saturation.
65  *
66  * Ask the process 'from_name:from_port' to start to saturate the link between itself
67  * and to_name:to_name.
68  */
69 void amok_bw_saturate_start(const char* from_name,unsigned int from_port,
70                             const char* to_name,unsigned int to_port,
71                             unsigned int msg_size, double duration) {
72   gras_socket_t sock;
73
74   sat_request_t request = xbt_new(s_sat_request_t,1);
75
76   sock = gras_socket_client(from_name,from_port);
77
78   request->host.name = (char*)to_name;
79   request->host.port = to_port;
80   
81   request->duration=duration;
82   request->msg_size=msg_size;
83
84
85   gras_msg_rpccall(sock,60,gras_msgtype_by_name("amok_bw_sat start"),&request, NULL);
86
87   free(request);
88   gras_socket_close(sock);
89 }
90
91 /* Asked to begin a saturation */
92 static int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx, void *payload){
93   sat_request_t request = *(sat_request_t*)payload;
94   gras_msg_rpcreturn(60,ctx, NULL);
95   amok_bw_saturate_begin(request->host.name,request->host.port,
96                          request->msg_size, request->duration,
97                          NULL,NULL);
98   free(request->host.name);
99   free(request);
100   return 1;
101 }
102
103 /**
104  * @brief Start saturating between the current process and the designated peer
105  *
106  * Note that the only way to break this function before the end of the timeout
107  * is to have a remote host calling amok_bw_saturate_stop to this process.
108  */
109 void amok_bw_saturate_begin(const char* to_name,unsigned int to_port,
110                             unsigned int msg_size, double duration,
111                             /*out*/ double *elapsed_res, double *bw_res) {
112  
113   xbt_ex_t e;
114
115   gras_socket_t peer_cmd = gras_socket_client(to_name, to_port);
116   gras_msg_cb_ctx_t ctx;
117
118   gras_socket_t meas;
119
120   s_gras_msg_t msg_got;
121
122   unsigned int packet_sent=0;
123   double start,elapsed=-1; /* timer */
124   double bw;
125
126   volatile int saturate_further; /* boolean in the main loop */ 
127
128   /* Negociate the saturation with the peer */
129   sat_request_t request = xbt_new(s_sat_request_t,1);
130
131   DEBUG2("Begin to saturate to %s:%d",to_name,to_port);
132   memset(&msg_got,0,sizeof(msg_got));
133
134   request->msg_size = msg_size;
135   request->duration = duration;
136   request->host.name = NULL;
137   request->host.port = 0;
138
139   ctx = gras_msg_rpc_async_call(peer_cmd, 60, 
140                                 gras_msgtype_by_name("amok_bw_sat begin"),
141                                                   &request);
142   free(request);
143   gras_msg_rpc_async_wait(ctx,&request);
144   meas=gras_socket_client_ext( to_name, request->host.port,
145                                0 /*bufsize: auto*/,
146                                1 /*meas: true*/);
147   free(request);
148
149   gras_socket_close(peer_cmd);
150   INFO2("Saturation from %s to %s started",gras_os_myname(),to_name);
151
152   /* Start experiment */
153   start=gras_os_time();
154
155   do {
156     /* do send it */
157     gras_socket_meas_send(meas,120,msg_size,msg_size);
158     packet_sent++;
159
160     /* Check whether someone asked us to stop saturation */
161     saturate_further = 0;
162     TRY {
163       gras_msg_wait_ext(0/*no wait*/,gras_msgtype_by_name("amok_bw_sat stop"),
164                         NULL /* accept any sender */,
165                         NULL, NULL, /* No specific filter */
166                         &msg_got);
167     } CATCH(e) {
168       if (e.category == timeout_error) {
169         saturate_further=1;
170         memset(&msg_got,0,sizeof(msg_got)); /* may be overprotectiv here */
171       }
172       xbt_ex_free(&e);
173     }
174
175     /* Check whether the experiment has to be terminated by now */
176     elapsed=gras_os_time()-start;
177     VERB2("elapsed %f duration %f",elapsed, duration);
178
179   } while (saturate_further && elapsed < duration);
180
181   INFO2("Saturation from %s to %s stopped",gras_os_myname(),to_name);
182   bw = ((double)(packet_sent*msg_size)) / elapsed;
183
184   if (elapsed_res)
185     *elapsed_res = elapsed;
186   if (bw_res)
187     *bw_res = bw;
188
189   if (elapsed >= duration) {
190     INFO2("Saturation experiment terminated. Took %f sec (achieving %f kb/s)",
191           elapsed, bw/1024.0);
192   }
193
194   /* If someone stopped us, inform him about the achieved bandwidth */
195   if (msg_got.expe) {
196     bw_res_t answer = xbt_new(s_bw_res_t,1);
197     s_gras_msg_cb_ctx_t ctx;
198
199     answer->timestamp=gras_os_time();
200     answer->sec=elapsed;
201     answer->bw=bw;
202
203     ctx.expeditor = msg_got.expe;
204     ctx.ID = msg_got.ID;
205     ctx.msgtype = msg_got.type;
206
207     gras_msg_rpcreturn(60,&ctx,&answer);
208     free(answer);
209   }
210   
211   gras_socket_close(meas);
212 }
213
214 /* Sender will saturate link to us */
215 static int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx, void *payload){
216   sat_request_t request=*(sat_request_t*)payload;
217   sat_request_t answer=xbt_new0(s_sat_request_t,1);
218   volatile int saturate_further = 1;
219   xbt_ex_t e;
220   gras_socket_t measMaster=NULL,meas=NULL;
221
222   int port=6000;
223   while (port <= 10000 && measMaster == NULL) {
224     TRY {
225       measMaster = gras_socket_server_ext(port,
226                                           0 /*bufsize: auto*/,
227                                           1 /*meas: true*/);
228     } CATCH(e) {
229       measMaster = NULL;
230       if (port < 10000)
231         xbt_ex_free(&e);
232       else
233         RETHROW0("Error encountered while opening a measurement server socket: %s");
234     }
235     if (measMaster == NULL) 
236       port++; /* prepare for a new loop */
237   }
238   answer->host.port=port;
239
240   gras_msg_rpcreturn(60, ctx, &answer);
241   free(answer);
242
243   gras_os_sleep(5); /* Wait for the accept */
244
245   TRY {
246     meas = gras_socket_meas_accept(measMaster);
247     DEBUG0("saturation handshake answered");
248   } CATCH(e) {
249     gras_socket_close(measMaster);
250     RETHROW0("Error during saturation handshake: %s");
251   }  
252
253   while (saturate_further) {
254     TRY {
255       gras_socket_meas_recv(meas,120,request->msg_size,request->msg_size);
256     } CATCH(e) {
257       saturate_further = 0;
258       xbt_ex_free(&e);
259     }
260   }
261   INFO1("Saturation stopped on %s",gras_os_myname());
262   gras_socket_close(meas);
263   gras_socket_close(measMaster);
264   free(request);
265   return 1;
266 }
267
268 /**
269  * @brief Ask 'from_name:from_port' to stop any saturation experiments
270  * @param from_name: Name of the host we are asking to do a experiment with (to_name:to_port)
271  * @param from_port: port on which the process we are asking for an experiment is listening
272  * @param time: the duration of the experiment
273  * @param bw: the achieved bandwidth
274  *
275  */
276 void amok_bw_saturate_stop(const char* from_name,unsigned int from_port,
277                            /*out*/ unsigned int *time, unsigned int *bw) {
278
279   gras_socket_t sock = gras_socket_client(from_name,from_port);
280   gras_msg_rpccall(sock,60,gras_msgtype_by_name("amok_bw_sat stop"),NULL,NULL);
281   gras_socket_close(sock);
282 }
283
284
285 #if 0
286 int grasbw_cbSatStart(gras_msg_t *msg) {
287   gras_rawsock_t *raw;
288   gras_sock_t *sock;
289   xbt_error_t errcode;
290   double start; /* time to timeout */
291
292   /* specification of the test to run */
293   char* to_name=gras_msg_ctn(msg,0,0,msgHost_t).host;
294   unsigned int to_port=gras_msg_ctn(msg,0,0,msgHost_t).port;
295
296   unsigned int msgSize=gras_msg_ctn(msg,1,0,SatExp_t).msgSize;
297   unsigned int timeout=gras_msg_ctn(msg,1,0,SatExp_t).timeout;
298   unsigned int raw_port;
299
300   /* The request */
301   SatExp_t *request;
302   /* answer */
303   gras_msg_t *answer;
304
305
306   /* Negociate the saturation with the peer */
307   if((errcode=gras_sock_client_open(to_name,to_port,&sock))) {
308     fprintf(stderr,"cbSatStart(): Error %s encountered while contacting peer\n",
309             xbt_error_name(errcode));
310     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
311                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
312                      errcode,"Cannot contact peer.\n");
313     return 1;
314   }
315   if (!(request=(SatExp_t *)malloc(sizeof(SatExp_t)))) {
316     fprintf(stderr,"cbSatStart(): Malloc error\n");
317     gras_sock_close(sock);
318     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
319                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
320                      malloc_error,"Cannot build request.\n");
321     return 1;    
322   }
323
324   request->timeout=gras_msg_ctn(msg,1,0,SatExp_t).timeout;
325   request->msgSize=gras_msg_ctn(msg,1,0,SatExp_t).msgSize;
326
327   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_SAT_BEGIN, 1, 
328                               request,1))) {
329     fprintf(stderr,"cbSatStart(): Error %s encountered while sending the request.\n",
330             xbt_error_name(errcode));
331     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
332                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
333                      errcode,"Cannot send request.\n");
334     gras_sock_close(sock);
335     return 1;
336   }
337
338   if ((errcode=gras_msg_wait(120,GRASMSG_SAT_BEGUN,&answer))) {
339     fprintf(stderr,"cbSatStart(): Error %s encountered while waiting for the ACK.\n",
340             xbt_error_name(errcode));
341     gras_sock_close(sock);
342
343     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
344                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
345                      errcode,
346                      "Cannot receive the ACK.\n");
347     return 1;
348   }
349
350   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
351     fprintf(stderr,"cbSatStart(): Peer reported error %s (%s).\n",
352             xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
353
354     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
355                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
356                      errcode,
357                      "Peer repported '%s'.\n",gras_msg_ctn(answer,0,0,msgError_t).errmsg);
358     gras_msg_free(answer);
359     gras_sock_close(sock);
360     return 1;
361   }
362
363   raw_port=gras_msg_ctn(answer,1,0,SatExp_t).port;
364
365   if ((errcode=gras_rawsock_client_open(to_name,raw_port,msgSize,&raw))) {
366     fprintf(stderr,"cbSatStart(): Error %s while opening raw socket to %s:%d.\n",
367             xbt_error_name(errcode),to_name,gras_msg_ctn(answer,1,0,SatExp_t).port);
368
369     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
370                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
371                      errcode,"Cannot open raw socket.\n");
372     gras_sock_close(sock);
373     return 1;
374   }
375
376   /* send a train of data before repporting that XP is started */
377   if ((errcode=gras_rawsock_send(raw,msgSize,msgSize))) {
378     fprintf(stderr,"cbSatStart: Failure %s during raw send\n",xbt_error_name(errcode));
379     grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
380                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
381                      errcode,"Cannot raw send.\n");
382     gras_sock_close(sock);
383     gras_rawsock_close(raw);
384     return 1;
385   }
386   
387   grasRepportError(msg->sock,GRASMSG_SAT_STARTED,1,
388                    "cbSatStart: Severe error: Cannot send error status to requester!!\n",
389                    no_error,"Saturation started");
390   gras_msg_free(answer);
391   gras_msg_free(msg);
392   
393   /* Do the saturation until we get a SAT_STOP message or until we timeout the whole XP*/
394   start=gras_time();
395   while (gras_msg_wait(0,GRASMSG_SAT_STOP,&msg)==timeout_error && 
396          gras_time()-start < timeout) {
397     if ((errcode=gras_rawsock_send(raw,msgSize,msgSize))) {
398       fprintf(stderr,"cbSatStart: Failure %s during raw send\n",xbt_error_name(errcode));
399       /* our error message do not interess anyone. SAT_STOP will do nothing. */
400       gras_sock_close(sock);
401       gras_rawsock_close(raw);
402       return 1;
403     } 
404   }
405   if (gras_time()-start > timeout) {
406     fprintf(stderr,"The saturation experiment did timeout. Stop it NOW\n");
407     gras_sock_close(sock);
408     gras_rawsock_close(raw);
409     return 1;
410   }
411
412   /* Handle the SAT_STOP which broke the previous while */
413   
414   if ((errcode=gras_msg_new_and_send(sock, GRASMSG_SAT_END,0))) {
415     fprintf(stderr,"cbSatStart(): Cannot tell peer to stop saturation\n");
416
417     grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
418                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
419                      errcode,"Sending SAT_END to peer failed.\n");
420     gras_sock_close(sock);
421     gras_rawsock_close(raw);
422     return 1;
423   }
424   
425   if ((errcode=gras_msg_wait(60,GRASMSG_SAT_ENDED,&answer))) {
426     fprintf(stderr,"cbSatStart(): Peer didn't ACK the end\n");
427
428     grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
429                      "cbSatStart: Severe error: Cannot send error status to requester!!\n",
430                      errcode,"Receiving SAT_ENDED from peer failed.\n");
431     gras_sock_close(sock);
432     gras_rawsock_close(raw);
433     return 1;
434   }
435   grasRepportError(msg->sock,GRASMSG_SAT_STOPPED,1,
436                    "cbSatStart: Severe error: Cannot send error status to requester!!\n",
437                    no_error,"");
438
439   gras_sock_close(sock);
440   gras_rawsock_close(raw);
441   gras_msg_free(answer);
442   gras_msg_free(msg);
443
444   return 1;  
445 }
446
447 int grasbw_cbSatBegin(gras_msg_t *msg) {
448   gras_rawsock_t *raw;
449   xbt_error_t errcode;
450   double start; /* timer */
451   /* request */
452   unsigned int msgSize=gras_msg_ctn(msg,0,0,SatExp_t).msgSize;
453   unsigned int timeout=gras_msg_ctn(msg,0,0,SatExp_t).timeout;
454   /* answer */
455   SatExp_t *request;
456   msgError_t *error;
457
458   if (!(request=(SatExp_t*)malloc(sizeof(SatExp_t))) ||
459       !(error=(msgError_t *)malloc(sizeof(msgError_t)))) {
460     fprintf(stderr,"cbSatBegin(): Malloc error\n");
461     grasRepportError(msg->sock,GRASMSG_SAT_BEGUN,2,
462                      "cbSatBegin: Severe error: Cannot send error status to requester!!\n",
463                      malloc_error,"Malloc error");
464     return 1;
465   }
466
467   if ((errcode=gras_rawsock_server_open(6666,8000,msgSize,&raw))) { 
468     fprintf(stderr,"cbSatBegin(): Error %s encountered while opening a raw socket\n",
469             xbt_error_name(errcode));
470     grasRepportError(msg->sock,GRASMSG_SAT_BEGUN,2,
471                      "cbSatBegin: Severe error: Cannot send error status to requester!!\n",
472                      errcode,"Cannot open raw socket");
473     return 1;
474   }
475   request->port=gras_rawsock_get_peer_port(raw);
476   request->msgSize=msgSize;
477   error->errcode=no_error;
478   error->errmsg[0]='\0';
479   if ((errcode=gras_msg_new_and_send(msg->sock,GRASMSG_SAT_BEGUN,2,
480                               error,1,
481                               request,1))) {
482     fprintf(stderr,"cbSatBegin(): Error %s encountered while send ACK to peer\n",
483             xbt_error_name(errcode));
484     return 1;
485   }
486   gras_msg_free(msg);
487
488   start=gras_time();
489   while (gras_msg_wait(0,GRASMSG_SAT_END,&msg)==timeout_error &&
490          gras_time() - start < timeout) {
491     errcode=gras_rawsock_recv(raw,msgSize,msgSize,1);
492     if (errcode != timeout_error && errcode != no_error) {
493       fprintf(stderr,"cbSatBegin: Failure %s during raw receive\n",xbt_error_name(errcode));
494       /* our error message do not interess anyone. SAT_END will do nothing. */
495       /* (if timeout'ed, it may be because the sender stopped emission. so survive it) */
496       return 1;
497     } 
498   }
499   if (gras_time()-start > timeout) {
500     fprintf(stderr,"The saturation experiment did timeout. Stop it NOW.\n");
501     gras_rawsock_close(raw);
502     return 1;
503   }
504
505   grasRepportError(msg->sock,GRASMSG_SAT_ENDED,1,
506                    "cbSatBegin: Cannot send SAT_ENDED.\n",
507                    no_error,"");
508   gras_rawsock_close(raw);
509   gras_msg_free(msg);
510   return 1;
511 }
512
513 xbt_error_t grasbw_saturate_stop(const char* from_name,unsigned int from_port,
514                                  const char* to_name,unsigned int to_port) {
515   xbt_error_t errcode;
516   gras_sock_t *sock;
517   gras_msg_t *answer;
518
519   if((errcode=gras_sock_client_open(from_name,from_port,&sock))) {
520     fprintf(stderr,"saturate_stop(): Error %s encountered while contacting peer\n",
521             xbt_error_name(errcode));
522     return errcode;
523   }
524
525   if ((errcode=gras_msg_new_and_send(sock,GRASMSG_SAT_STOP,0))) {
526     fprintf(stderr,"saturate_stop(): Error %s encountered while sending request\n",
527             xbt_error_name(errcode));
528     gras_sock_close(sock);
529     return errcode;
530   }
531
532   if ((errcode=gras_msg_wait(120,GRASMSG_SAT_STOPPED,&answer))) {
533     fprintf(stderr,"saturate_stop(): Error %s encountered while receiving ACK\n",
534             xbt_error_name(errcode));
535     gras_sock_close(sock);
536     return errcode;
537   }
538
539   if((errcode=gras_msg_ctn(answer,0,0,msgError_t).errcode)) {
540     fprintf(stderr,"saturate_stop(): Peer reported error %s (%s).\n",
541             xbt_error_name(errcode),gras_msg_ctn(answer,0,0,msgError_t).errmsg);
542     gras_msg_free(answer);
543     gras_sock_close(sock);
544     return errcode;
545   }
546
547   gras_msg_free(answer);
548   gras_sock_close(sock);
549
550   return no_error;
551 }
552 #endif