Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Last bits of convertion from xbt_error_t to exceptions. Some more cleanups (mainly...
[simgrid.git] / src / amok / Bandwidth / bandwidth.c
1 /* $Id$ */
2
3 /* amok_bandwidth - Bandwidth tests facilities                              */
4
5 /* Copyright (c) 2003-5 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 "xbt/ex.h"
11 #include "amok/Bandwidth/bandwidth_private.h"
12 #include "gras/messages.h"
13
14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(bw,amok,"Bandwidth testing");
15
16 static short _amok_bw_initialized = 0;
17
18 /** @brief module initialization; all participating nodes must run this */
19 void amok_bw_init(void) {
20   gras_datadesc_type_t bw_request_desc, bw_res_desc, sat_request_desc;
21
22   amok_base_init();
23
24   if (! _amok_bw_initialized) {
25         
26      /* Build the datatype descriptions */ 
27      bw_request_desc = gras_datadesc_struct("s_bw_request_t");
28      gras_datadesc_struct_append(bw_request_desc,"host",
29                                  gras_datadesc_by_name("xbt_host_t"));
30      gras_datadesc_struct_append(bw_request_desc,"buf_size",
31                                  gras_datadesc_by_name("unsigned long int"));
32      gras_datadesc_struct_append(bw_request_desc,"exp_size",
33                                  gras_datadesc_by_name("unsigned long int"));
34      gras_datadesc_struct_append(bw_request_desc,"msg_size",
35                                  gras_datadesc_by_name("unsigned long int"));
36      gras_datadesc_struct_close(bw_request_desc);
37      bw_request_desc = gras_datadesc_ref("bw_request_t",bw_request_desc);
38
39      bw_res_desc = gras_datadesc_struct("s_bw_res_t");
40      gras_datadesc_struct_append(bw_res_desc,"err",gras_datadesc_by_name("s_amok_remoterr_t"));
41      gras_datadesc_struct_append(bw_res_desc,"timestamp",gras_datadesc_by_name("unsigned int"));
42      gras_datadesc_struct_append(bw_res_desc,"seconds",gras_datadesc_by_name("double"));
43      gras_datadesc_struct_append(bw_res_desc,"bw",gras_datadesc_by_name("double"));
44      gras_datadesc_struct_close(bw_res_desc);
45      bw_res_desc = gras_datadesc_ref("bw_res_t",bw_res_desc);
46
47      sat_request_desc = gras_datadesc_struct("s_sat_request_desc_t");
48      gras_datadesc_struct_append(sat_request_desc,"host",gras_datadesc_by_name("xbt_host_t"));
49      gras_datadesc_struct_append(sat_request_desc,"msg_size",gras_datadesc_by_name("unsigned int"));
50      gras_datadesc_struct_append(sat_request_desc,"timeout",gras_datadesc_by_name("unsigned int"));
51      gras_datadesc_struct_close(sat_request_desc);
52      sat_request_desc = gras_datadesc_ref("sat_request_t",sat_request_desc);
53      
54      /* Register the bandwidth messages */
55      gras_msgtype_declare("BW handshake",     bw_request_desc);
56      gras_msgtype_declare("BW handshake ACK", bw_request_desc);
57      gras_msgtype_declare("BW request",       bw_request_desc);
58      gras_msgtype_declare("BW result",        bw_res_desc);
59      
60      /* Register the saturation messages */
61      gras_msgtype_declare("SAT start",   sat_request_desc);
62      gras_msgtype_declare("SAT started", gras_datadesc_by_name("amok_remoterr_t"));
63      gras_msgtype_declare("SAT begin",   sat_request_desc);
64      gras_msgtype_declare("SAT begun",   gras_datadesc_by_name("amok_remoterr_t"));
65      gras_msgtype_declare("SAT end",     NULL);
66      gras_msgtype_declare("SAT ended",   gras_datadesc_by_name("amok_remoterr_t"));
67      gras_msgtype_declare("SAT stop",    NULL);
68      gras_msgtype_declare("SAT stopped", gras_datadesc_by_name("amok_remoterr_t"));
69   }
70    
71   /* Register the callbacks */
72   gras_cb_register(gras_msgtype_by_name("BW request"),
73                    &amok_bw_cb_bw_request);
74   gras_cb_register(gras_msgtype_by_name("BW handshake"),
75                    &amok_bw_cb_bw_handshake);
76
77   gras_cb_register(gras_msgtype_by_name("SAT start"),
78                    &amok_bw_cb_sat_start);
79   gras_cb_register(gras_msgtype_by_name("SAT begin"),
80                    &amok_bw_cb_sat_begin);
81   
82   _amok_bw_initialized =1;
83 }
84
85 /** @brief module finalization */
86 void amok_bw_exit(void) {
87   if (! _amok_bw_initialized)
88     return;
89    
90   gras_cb_unregister(gras_msgtype_by_name("BW request"),
91                      &amok_bw_cb_bw_request);
92   gras_cb_unregister(gras_msgtype_by_name("BW handshake"),
93                      &amok_bw_cb_bw_handshake);
94
95   gras_cb_unregister(gras_msgtype_by_name("SAT start"),
96                      &amok_bw_cb_sat_start);
97   gras_cb_unregister(gras_msgtype_by_name("SAT begin"),
98                      &amok_bw_cb_sat_begin);
99
100   _amok_bw_initialized = 0;
101 }
102
103 /* ***************************************************************************
104  * Bandwidth tests
105  * ***************************************************************************/
106
107 /**
108  * \brief bandwidth measurement between localhost and \e peer
109  * 
110  * \arg peer: A (regular) socket at which the the host with which we should conduct the experiment can be contacted
111  * \arg buf_size: Size of the socket buffer
112  * \arg exp_size: Total size of data sent across the network
113  * \arg msg_size: Size of each message sent. Ie, (\e expSize % \e msgSize) messages will be sent.
114  * \arg sec: where the result (in seconds) should be stored.
115  * \arg bw: observed Bandwidth (in kb/s) 
116  *
117  * Conduct a bandwidth test from the local process to the given peer.
118  * This call is blocking until the end of the experiment.
119  *
120  * Results are reported in last args, and sizes are in kb.
121  */
122 void amok_bw_test(gras_socket_t peer,
123                   unsigned long int buf_size,
124                   unsigned long int exp_size,
125                   unsigned long int msg_size,
126           /*OUT*/ double *sec, double *bw) {
127
128   /* Measurement sockets for the experiments */
129   gras_socket_t measMasterIn=NULL,measIn,measOut;
130   int port;
131   bw_request_t request,request_ack;
132   xbt_ex_t e;
133   
134   for (port = 5000; port < 10000 && measMasterIn == NULL; port++) {
135     TRY {
136       measMasterIn = gras_socket_server_ext(++port,buf_size,1);
137     } CATCH(e) {
138       measMasterIn = NULL;
139       if (port < 10000) {
140         xbt_ex_free(e);
141       } else {
142         RETHROW0("Error caught while opening a measurement socket: %s");
143       }
144     }
145   }
146   
147   request=xbt_new0(s_bw_request_t,1);
148   request->buf_size=buf_size*1024;
149   request->exp_size=exp_size*1024;
150   request->msg_size=msg_size*1024;
151   request->host.name = NULL;
152   request->host.port = gras_socket_my_port(measMasterIn);
153   VERB5("Handshaking with %s:%d to connect it back on my %d (expsize=%ld kb= %ld b)", 
154         gras_socket_peer_name(peer),gras_socket_peer_port(peer), request->host.port,
155         buf_size,request->buf_size);
156
157   TRY {
158     gras_msg_send(peer,gras_msgtype_by_name("BW handshake"),&request);
159   } CATCH(e) {
160     RETHROW0("Error encountered while sending the BW request: %s");
161   }
162   measIn = gras_socket_meas_accept(measMasterIn);
163
164   TRY {
165     gras_msg_wait(60,gras_msgtype_by_name("BW handshake ACK"),NULL,&request_ack);
166   } CATCH(e) {
167     RETHROW0("Error encountered while waiting for the answer to BW request: %s");
168   }
169   
170   /* FIXME: What if there is a remote error? */
171    
172   TRY {
173     measOut=gras_socket_client_ext(gras_socket_peer_name(peer),
174                                    request_ack->host.port, 
175                                    request->buf_size,1);
176   } CATCH(e) {
177     RETHROW2("Error encountered while opening the measurement socket to %s:%d for BW test: %s",
178              gras_socket_peer_name(peer),request_ack->host.port);
179   }
180   DEBUG1("Got ACK; conduct the experiment (msg_size=%ld)",request->msg_size);
181
182   *sec=gras_os_time();
183   TRY {
184     gras_socket_meas_send(measOut,120,request->exp_size,request->msg_size);
185     gras_socket_meas_recv(measIn,120,1,1);
186   } CATCH(e) {
187     gras_socket_close(measOut);
188     gras_socket_close(measMasterIn);
189     gras_socket_close(measIn);
190     RETHROW0("Unable to conduct the experiment: %s");
191   }
192
193   *sec = gras_os_time() - *sec;
194   *bw = ((double)exp_size) / *sec;
195
196   free(request_ack);
197   free(request);
198   if (measIn != measMasterIn)
199     gras_socket_close(measIn);
200   gras_socket_close(measMasterIn);
201   gras_socket_close(measOut);
202 }
203
204
205 /* Callback to the "BW handshake" message: 
206    opens a server measurement socket,
207    indicate its port in an "BW handshaked" message,
208    receive the corresponding data on the measurement socket, 
209    close the measurment socket
210
211    sizes are in byte (got converted from kb my expeditor)
212 */
213 int amok_bw_cb_bw_handshake(gras_socket_t  expeditor,
214                             void          *payload) {
215   gras_socket_t measMasterIn=NULL,measIn,measOut;
216   bw_request_t request=*(bw_request_t*)payload;
217   bw_request_t answer;
218   xbt_ex_t e;
219   int port;
220   
221   VERB5("Handshaked to connect to %s:%d (sizes: buf=%lu exp=%lu msg=%lu)",
222         gras_socket_peer_name(expeditor),request->host.port,
223         request->buf_size,request->exp_size,request->msg_size);     
224
225   /* Build our answer */
226   answer = xbt_new0(s_bw_request_t,1);
227   
228   for (port = 6000; port < 10000 && measMasterIn == NULL; port++) {
229     TRY {
230       measMasterIn = gras_socket_server_ext(port,request->buf_size,1);
231     } CATCH(e) {
232       measMasterIn = NULL;
233       if (port < 10000)
234         xbt_ex_free(e);
235       else
236         /* FIXME: tell error to remote */
237         RETHROW0("Error encountered while opening a measurement server socket: %s");
238     }
239   }
240    
241   answer->buf_size=request->buf_size;
242   answer->exp_size=request->exp_size;
243   answer->msg_size=request->msg_size;
244   answer->host.port=gras_socket_my_port(measMasterIn);
245
246   /* Don't connect asap to leave time to other side to enter the accept() */
247   TRY {
248     measOut = gras_socket_client_ext(gras_socket_peer_name(expeditor),
249                                      request->host.port,
250                                      request->buf_size,1);
251   } CATCH(e) {
252     RETHROW2("Error encountered while opening a measurement socket back to %s:%d : %s", 
253              gras_socket_peer_name(expeditor),request->host.port);
254     /* FIXME: tell error to remote */
255   }
256
257   TRY {
258     gras_msg_send(expeditor, gras_msgtype_by_name("BW handshake ACK"), &answer);
259   } CATCH(e) { 
260     gras_socket_close(measMasterIn);
261     gras_socket_close(measOut);
262     /* FIXME: tell error to remote */
263     RETHROW0("Error encountered while sending the answer: %s");
264   }
265
266   TRY {
267     measIn = gras_socket_meas_accept(measMasterIn);
268     DEBUG4("BW handshake answered. buf_size=%lu exp_size=%lu msg_size=%lu port=%d",
269            answer->buf_size,answer->exp_size,answer->msg_size,answer->host.port);
270
271     gras_socket_meas_recv(measIn, 120,request->exp_size,request->msg_size);
272     gras_socket_meas_send(measOut,120,1,1);
273   } CATCH(e) {
274     gras_socket_close(measMasterIn);
275     gras_socket_close(measIn);
276     gras_socket_close(measOut);
277     /* FIXME: tell error to remote ? */
278     RETHROW0("Error encountered while receiving the experiment: %s");
279   }
280
281   if (measIn != measMasterIn)
282     gras_socket_close(measMasterIn);
283   gras_socket_close(measIn);
284   gras_socket_close(measOut);
285   free(answer);
286   free(request);
287   DEBUG0("BW experiment done.");
288   return 1;
289 }
290
291 /**
292  * \brief request a bandwidth measurement between two remote hosts
293  *
294  * \arg from_name: Name of the first host 
295  * \arg from_port: port on which the first process is listening for messages
296  * \arg to_name: Name of the second host 
297  * \arg to_port: port on which the second process is listening (for messages, do not 
298  * give a measurement socket here. The needed measurement sockets will be created 
299  * automatically and negociated between the peers)
300  * \arg buf_size: Size of the socket buffer
301  * \arg exp_size: Total size of data sent across the network
302  * \arg msg_size: Size of each message sent. (\e expSize % \e msgSize) messages will be sent.
303  * \arg sec: where the result (in seconds) should be stored.
304  * \arg bw: observed Bandwidth (in kb/s)
305  *
306  * Conduct a bandwidth test from the process from_host:from_port to to_host:to_port.
307  * This call is blocking until the end of the experiment.
308  *
309  * Results are reported in last args, and sizes are in kb.
310  */
311 void amok_bw_request(const char* from_name,unsigned int from_port,
312                      const char* to_name,unsigned int to_port,
313                      unsigned long int buf_size,
314                      unsigned long int exp_size,
315                      unsigned long int msg_size,
316              /*OUT*/ double *sec, double*bw) {
317   
318   gras_socket_t sock;
319   /* The request */
320   bw_request_t request;
321   bw_res_t result;
322
323   request=xbt_new0(s_bw_request_t,1);
324   request->buf_size=buf_size;
325   request->exp_size=exp_size;
326   request->msg_size=msg_size;
327
328   request->host.name = (char*)to_name;
329   request->host.port = to_port;
330
331   sock = gras_socket_client(from_name,from_port);
332   gras_msg_send(sock,gras_msgtype_by_name("BW request"),&request);
333   free(request);
334
335   gras_msg_wait(240,gras_msgtype_by_name("BW result"),NULL, &result);
336   
337   *sec=result->sec;
338   *bw =result->bw;
339
340   VERB6("BW test between %s:%d and %s:%d took %f sec, achieving %f kb/s",
341         from_name,from_port, to_name,to_port,
342         *sec,*bw);
343
344   gras_socket_close(sock);
345   free(result);
346 }
347
348 int amok_bw_cb_bw_request(gras_socket_t    expeditor,
349                           void            *payload) {
350                           
351   /* specification of the test to run, and our answer */
352   bw_request_t request = *(bw_request_t*)payload;
353   bw_res_t result = xbt_new0(s_bw_res,1);
354   gras_socket_t peer;
355
356   peer = gras_socket_client(request->host.name,request->host.port);
357   amok_bw_test(peer,
358                request->buf_size,request->exp_size,request->msg_size,
359                &(result->sec),&(result->bw));
360
361   gras_msg_send(expeditor,gras_msgtype_by_name("BW result"),&result);
362
363   gras_os_sleep(1);
364   gras_socket_close(peer);
365   free(request);
366   free(result);
367   
368   return 1;
369 }
370
371 int amok_bw_cb_sat_start(gras_socket_t     expeditor,
372                          void             *payload) {
373    CRITICAL0("amok_bw_cb_sat_start; not implemented");
374    return 1;
375
376 int amok_bw_cb_sat_begin(gras_socket_t     expeditor,
377                          void             *payload) {
378    CRITICAL0("amok_bw_cb_sat_begin: not implemented");
379    return 1;
380 }