if (! _amok_bw_initialized) {
- /* Build the datatype descriptions */
+ /* Build the Bandwidth datatype descriptions */
bw_request_desc = gras_datadesc_struct("s_bw_request_t");
gras_datadesc_struct_append(bw_request_desc,"host",
gras_datadesc_by_name("xbt_host_t"));
bw_request_desc = gras_datadesc_ref("bw_request_t",bw_request_desc);
bw_res_desc = gras_datadesc_struct("s_bw_res_t");
- gras_datadesc_struct_append(bw_res_desc,"err",gras_datadesc_by_name("s_amok_remoterr_t"));
gras_datadesc_struct_append(bw_res_desc,"timestamp",gras_datadesc_by_name("unsigned int"));
gras_datadesc_struct_append(bw_res_desc,"seconds",gras_datadesc_by_name("double"));
gras_datadesc_struct_append(bw_res_desc,"bw",gras_datadesc_by_name("double"));
gras_datadesc_struct_close(bw_res_desc);
bw_res_desc = gras_datadesc_ref("bw_res_t",bw_res_desc);
+ gras_msgtype_declare_rpc("BW handshake",bw_request_desc,bw_request_desc);
+ gras_msgtype_declare_rpc("BW request", bw_request_desc,bw_res_desc);
+
+ /* Build the saturation datatype descriptions */
sat_request_desc = gras_datadesc_struct("s_sat_request_desc_t");
gras_datadesc_struct_append(sat_request_desc,"host",gras_datadesc_by_name("xbt_host_t"));
gras_datadesc_struct_append(sat_request_desc,"msg_size",gras_datadesc_by_name("unsigned int"));
gras_datadesc_struct_close(sat_request_desc);
sat_request_desc = gras_datadesc_ref("sat_request_t",sat_request_desc);
- /* Register the bandwidth messages */
- gras_msgtype_declare("BW handshake", bw_request_desc);
- gras_msgtype_declare("BW handshake ACK", bw_request_desc);
- gras_msgtype_declare("BW request", bw_request_desc);
- gras_msgtype_declare("BW result", bw_res_desc);
-
/* Register the saturation messages */
gras_msgtype_declare("SAT start", sat_request_desc);
- gras_msgtype_declare("SAT started", gras_datadesc_by_name("amok_remoterr_t"));
+ gras_msgtype_declare("SAT started", NULL);
gras_msgtype_declare("SAT begin", sat_request_desc);
- gras_msgtype_declare("SAT begun", gras_datadesc_by_name("amok_remoterr_t"));
+ gras_msgtype_declare("SAT begun", NULL);
gras_msgtype_declare("SAT end", NULL);
- gras_msgtype_declare("SAT ended", gras_datadesc_by_name("amok_remoterr_t"));
+ gras_msgtype_declare("SAT ended", NULL);
gras_msgtype_declare("SAT stop", NULL);
- gras_msgtype_declare("SAT stopped", gras_datadesc_by_name("amok_remoterr_t"));
+ gras_msgtype_declare("SAT stopped", NULL);
}
/* Register the callbacks */
* \arg exp_size: Total size of data sent across the network
* \arg msg_size: Size of each message sent. Ie, (\e expSize % \e msgSize) messages will be sent.
* \arg sec: where the result (in seconds) should be stored.
- * \arg bw: observed Bandwidth (in kb/s)
+ * \arg bw: observed Bandwidth (in byte/s)
*
* Conduct a bandwidth test from the local process to the given peer.
* This call is blocking until the end of the experiment.
*
- * Results are reported in last args, and sizes are in kb.
+ * Results are reported in last args, and sizes are in byte.
*/
void amok_bw_test(gras_socket_t peer,
unsigned long int buf_size,
/*OUT*/ double *sec, double *bw) {
/* Measurement sockets for the experiments */
- gras_socket_t measMasterIn=NULL,measIn,measOut;
+ gras_socket_t measMasterIn=NULL,measIn,measOut=NULL;
int port;
bw_request_t request,request_ack;
xbt_ex_t e;
measMasterIn = gras_socket_server_ext(++port,buf_size,1);
} CATCH(e) {
measMasterIn = NULL;
- if (port < 10000) {
- xbt_ex_free(e);
- } else {
+ if (port == 10000 -1) {
RETHROW0("Error caught while opening a measurement socket: %s");
+ } else {
+ xbt_ex_free(e);
}
}
}
request=xbt_new0(s_bw_request_t,1);
- request->buf_size=buf_size*1024;
- request->exp_size=exp_size*1024;
- request->msg_size=msg_size*1024;
+ request->buf_size=buf_size;
+ request->exp_size=exp_size;
+ request->msg_size=msg_size;
request->host.name = NULL;
request->host.port = gras_socket_my_port(measMasterIn);
- VERB5("Handshaking with %s:%d to connect it back on my %d (expsize=%ld kb= %ld b)",
+ VERB5("Handshaking with %s:%d to connect it back on my %d (expsize=%ld byte= %ld b)",
gras_socket_peer_name(peer),gras_socket_peer_port(peer), request->host.port,
buf_size,request->buf_size);
TRY {
- gras_msg_send(peer,gras_msgtype_by_name("BW handshake"),&request);
+ gras_msg_rpccall(peer,60,
+ gras_msgtype_by_name("BW handshake"),&request, &request_ack);
} CATCH(e) {
RETHROW0("Error encountered while sending the BW request: %s");
}
measIn = gras_socket_meas_accept(measMasterIn);
-
- TRY {
- gras_msg_wait(60,gras_msgtype_by_name("BW handshake ACK"),NULL,&request_ack);
- } CATCH(e) {
- RETHROW0("Error encountered while waiting for the answer to BW request: %s");
- }
-
- /* FIXME: What if there is a remote error? */
TRY {
measOut=gras_socket_client_ext(gras_socket_peer_name(peer),
receive the corresponding data on the measurement socket,
close the measurment socket
- sizes are in byte (got converted from kb my expeditor)
+ sizes are in byte
*/
-int amok_bw_cb_bw_handshake(gras_socket_t expeditor,
+int amok_bw_cb_bw_handshake(gras_msg_cb_ctx_t ctx,
void *payload) {
- gras_socket_t measMasterIn=NULL,measIn,measOut;
+ gras_socket_t expeditor = gras_msg_cb_ctx_from(ctx);
+ gras_socket_t measMasterIn=NULL,measIn=NULL,measOut=NULL;
bw_request_t request=*(bw_request_t*)payload;
bw_request_t answer;
xbt_ex_t e;
answer->msg_size=request->msg_size;
answer->host.port=gras_socket_my_port(measMasterIn);
+
+
+ TRY {
+ gras_msg_rpcreturn(60,ctx,&answer);
+ } CATCH(e) {
+ gras_socket_close(measMasterIn);
+ /* FIXME: tell error to remote */
+ RETHROW0("Error encountered while sending the answer: %s");
+ }
+
+
/* Don't connect asap to leave time to other side to enter the accept() */
TRY {
measOut = gras_socket_client_ext(gras_socket_peer_name(expeditor),
/* FIXME: tell error to remote */
}
- TRY {
- gras_msg_send(expeditor, gras_msgtype_by_name("BW handshake ACK"), &answer);
- } CATCH(e) {
- gras_socket_close(measMasterIn);
- gras_socket_close(measOut);
- /* FIXME: tell error to remote */
- RETHROW0("Error encountered while sending the answer: %s");
- }
-
TRY {
measIn = gras_socket_meas_accept(measMasterIn);
DEBUG4("BW handshake answered. buf_size=%lu exp_size=%lu msg_size=%lu port=%d",
* \arg exp_size: Total size of data sent across the network
* \arg msg_size: Size of each message sent. (\e expSize % \e msgSize) messages will be sent.
* \arg sec: where the result (in seconds) should be stored.
- * \arg bw: observed Bandwidth (in kb/s)
+ * \arg bw: observed Bandwidth (in byte/s)
*
* Conduct a bandwidth test from the process from_host:from_port to to_host:to_port.
* This call is blocking until the end of the experiment.
*
- * Results are reported in last args, and sizes are in kb.
+ * Results are reported in last args, and sizes are in bytes.
*/
void amok_bw_request(const char* from_name,unsigned int from_port,
const char* to_name,unsigned int to_port,
request->host.port = to_port;
sock = gras_socket_client(from_name,from_port);
- gras_msg_send(sock,gras_msgtype_by_name("BW request"),&request);
- free(request);
-
- gras_msg_wait(240,gras_msgtype_by_name("BW result"),NULL, &result);
+ gras_msg_rpccall(sock,240,gras_msgtype_by_name("BW request"),&request, &result);
*sec=result->sec;
*bw =result->bw;
free(result);
}
-int amok_bw_cb_bw_request(gras_socket_t expeditor,
+int amok_bw_cb_bw_request(gras_msg_cb_ctx_t ctx,
void *payload) {
/* specification of the test to run, and our answer */
request->buf_size,request->exp_size,request->msg_size,
&(result->sec),&(result->bw));
- gras_msg_send(expeditor,gras_msgtype_by_name("BW result"),&result);
+ gras_msg_rpcreturn(240,ctx,&result);
gras_os_sleep(1);
gras_socket_close(peer);
return 1;
}
-int amok_bw_cb_sat_start(gras_socket_t expeditor,
+int amok_bw_cb_sat_start(gras_msg_cb_ctx_t ctx,
void *payload) {
CRITICAL0("amok_bw_cb_sat_start; not implemented");
return 1;
}
-int amok_bw_cb_sat_begin(gras_socket_t expeditor,
+int amok_bw_cb_sat_begin(gras_msg_cb_ctx_t ctx,
void *payload) {
CRITICAL0("amok_bw_cb_sat_begin: not implemented");
return 1;