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_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");
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,"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_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"));
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);
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);
/* Register the saturation messages */
gras_msgtype_declare("SAT start", sat_request_desc);
* \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 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.
*
* Conduct a bandwidth test from the local process to the given peer.
* This call is blocking until the end of the experiment.
*
*
* Conduct a bandwidth test from the local process to the given peer.
* This call is blocking until the end of the experiment.
*
*/
void amok_bw_test(gras_socket_t peer,
unsigned long int buf_size,
*/
void amok_bw_test(gras_socket_t peer,
unsigned long int buf_size,
- 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;
gras_socket_peer_name(peer),gras_socket_peer_port(peer), request->host.port,
buf_size,request->buf_size);
TRY {
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);
} 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? */
receive the corresponding data on the measurement socket,
close the measurment socket
receive the corresponding data on the measurement socket,
close the measurment socket
gras_socket_t measMasterIn=NULL,measIn=NULL,measOut=NULL;
bw_request_t request=*(bw_request_t*)payload;
bw_request_t answer;
gras_socket_t measMasterIn=NULL,measIn=NULL,measOut=NULL;
bw_request_t request=*(bw_request_t*)payload;
bw_request_t answer;
+
+
+ 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),
/* 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),
- 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",
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 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.
*
* 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.
*
*
* 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.
*
*/
void amok_bw_request(const char* from_name,unsigned int from_port,
const char* to_name,unsigned int to_port,
*/
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);
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);
request->buf_size,request->exp_size,request->msg_size,
&(result->sec),&(result->bw));
request->buf_size,request->exp_size,request->msg_size,
&(result->sec),&(result->bw));