Skip to content

UCP/HELLO-WORLD: added error simulations #5901

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yosefe merged 1 commit into from
Dec 13, 2020
Merged

UCP/HELLO-WORLD: added error simulations #5901

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions contrib/test_jenkins.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -802,14 +802,14 @@ run_hello() {
fi

# set smaller timeouts so the test will complete faster
if [[ ${test_args} == *"-e"* ]]
if [[ ${test_args} =~ "-e" ]]
then
export UCX_UD_TIMEOUT=15s
export UCX_RC_TIMEOUT=1ms
export UCX_RC_RETRY_COUNT=4
fi

if [[ ${test_args} == *"-e"* ]]
if [[ ${test_args} =~ "-e" ]]
then
error_emulation=1
else
Expand Down Expand Up @@ -842,7 +842,10 @@ run_ucp_hello() {
mem_types_list+="cuda cuda-managed "
fi

for test_mode in -w -f -b -e
export UCX_KEEPALIVE_TIMEOUT=1s
export UCX_KEEPALIVE_NUM_EPS=10

for test_mode in -w -f -b -erecv -esend -ekeepalive
do
for mem_type in $mem_types_list
do
Expand All @@ -851,6 +854,9 @@ run_ucp_hello() {
done
done
rm -f ./ucp_hello_world

unset UCX_KEEPALIVE_TIMEOUT
unset UCX_KEEPALIVE_NUM_EPS
}

#
Expand Down
136 changes: 95 additions & 41 deletions examples/ucp_hello_world.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,18 +67,25 @@ enum ucp_test_mode_t {
TEST_MODE_EVENTFD
} ucp_test_mode = TEST_MODE_PROBE;

typedef enum {
FAILURE_MODE_NONE,
FAILURE_MODE_SEND, /* fail send operation on server */
FAILURE_MODE_RECV, /* fail receive operation on client */
FAILURE_MODE_KEEPALIVE /* fail without communication on client */
} failure_mode_t;

static struct err_handling {
ucp_err_handling_mode_t ucp_err_mode;
int failure;
failure_mode_t failure_mode;
} err_handling_opt;

static ucs_status_t client_status = UCS_OK;
static uint16_t server_port = 13337;
static long test_string_length = 16;
static const ucp_tag_t tag = 0x1337a880u;
static const ucp_tag_t tag_mask = UINT64_MAX;
static const char *addr_msg_str = "UCX address message";
static const char *data_msg_str = "UCX data message";
static ucs_status_t ep_status = UCS_OK;
static uint16_t server_port = 13337;
static long test_string_length = 16;
static const ucp_tag_t tag = 0x1337a880u;
static const ucp_tag_t tag_mask = UINT64_MAX;
static const char *addr_msg_str = "UCX address message";
static const char *data_msg_str = "UCX data message";
static ucp_address_t *local_addr;
static ucp_address_t *peer_addr;

Expand Down Expand Up @@ -220,9 +227,14 @@ static int run_ucx_client(ucp_worker_h ucp_worker)

/* Send client UCX address to server */
ep_params.field_mask = UCP_EP_PARAM_FIELD_REMOTE_ADDRESS |
UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE;
UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE |
Comment thread
hoopoepg marked this conversation as resolved.
Resolved
UCP_EP_PARAM_FIELD_ERR_HANDLER |
UCP_EP_PARAM_FIELD_USER_DATA;
ep_params.address = peer_addr;
ep_params.err_mode = err_handling_opt.ucp_err_mode;
ep_params.err_handler.cb = failure_handler;
ep_params.err_handler.arg = NULL;
ep_params.user_data = &ep_status;

status = ucp_ep_create(ucp_worker, &ep_params, &server_ep);
CHKERR_JUMP(status != UCS_OK, "ucp_ep_create\n", err);
Expand Down Expand Up @@ -250,14 +262,14 @@ static int run_ucx_client(ucp_worker_h ucp_worker)

free(msg);

if (err_handling_opt.failure) {
fprintf(stderr, "Emulating unexpected failure on client side\n");
if (err_handling_opt.failure_mode == FAILURE_MODE_RECV) {
fprintf(stderr, "Emulating failure before receive operation on client side\n");
raise(SIGKILL);
}

/* Receive test string from server */
for (;;) {

CHKERR_JUMP(ep_status != UCS_OK, "receive data: EP disconnected\n", err_ep);
/* Probing incoming events in non-block mode */
msg_tag = ucp_tag_probe_nb(ucp_worker, tag, tag_mask, 1, &info_tag);
if (msg_tag != NULL) {
Expand All @@ -281,6 +293,13 @@ static int run_ucx_client(ucp_worker_h ucp_worker)
CHKERR_JUMP(status != UCS_OK, "test_poll_wait\n", err_ep);
}
}

if (err_handling_opt.failure_mode == FAILURE_MODE_KEEPALIVE) {
fprintf(stderr, "Emulating unexpected failure after receive completion "
"on client side, server should detect error by "
"keepalive mechanism\n");
raise(SIGKILL);
}

msg = mem_type_malloc(info_tag.length);
CHKERR_JUMP(msg == NULL, "allocate memory\n", err_ep);
Expand Down Expand Up @@ -378,6 +397,13 @@ static int run_ucx_server(ucp_worker_h ucp_worker)
goto err;
}

if (err_handling_opt.failure_mode == FAILURE_MODE_SEND) {
fprintf(stderr, "Emulating unexpected failure on server side, client "
"should detect error by keepalive mechanism\n");
free(msg);
raise(SIGKILL);
}

peer_addr_len = msg->data_len;
peer_addr = malloc(peer_addr_len);
if (peer_addr == NULL) {
Expand All @@ -400,13 +426,13 @@ static int run_ucx_server(ucp_worker_h ucp_worker)
ep_params.err_mode = err_handling_opt.ucp_err_mode;
ep_params.err_handler.cb = failure_handler;
ep_params.err_handler.arg = NULL;
ep_params.user_data = &client_status;
ep_params.user_data = &ep_status;

status = ucp_ep_create(ucp_worker, &ep_params, &client_ep);
/* If peer failure testing was requested, it could be possible that UCP EP
* couldn't be created; in this case set `ret = 0` to report success */
CHKERR_ACTION(status != UCS_OK, "ucp_ep_create\n",
ret = (err_handling_opt.failure) ? 0 : -1; goto err);
ret = (err_handling_opt.failure_mode != FAILURE_MODE_NONE) ? 0 : -1;
CHKERR_ACTION(status != UCS_OK, "ucp_ep_create\n", goto err);

msg_len = sizeof(*msg) + test_string_length;
msg = mem_type_malloc(msg_len);
Expand All @@ -417,12 +443,14 @@ static int run_ucx_server(ucp_worker_h ucp_worker)
ret = generate_test_string((char *)(msg + 1), test_string_length);
CHKERR_JUMP(ret < 0, "generate test string", err_free_mem_type_msg);

if (err_handling_opt.failure) {
/* Sleep for small amount of time to ensure that server was killed
if (err_handling_opt.failure_mode == FAILURE_MODE_RECV) {
/* Sleep for small amount of time to ensure that client was killed
* and peer failure handling is covered */
sleep(5);
}

ucp_worker_progress(ucp_worker);

send_param.op_attr_mask = UCP_OP_ATTR_FIELD_CALLBACK |
UCP_OP_ATTR_FIELD_USER_DATA |
UCP_OP_ATTR_FIELD_MEMORY_TYPE;
Expand All @@ -434,21 +462,28 @@ static int run_ucx_server(ucp_worker_h ucp_worker)
status = ucx_wait(ucp_worker, request, "send",
data_msg_str);
if (status != UCS_OK) {
if (!err_handling_opt.failure) {
if (err_handling_opt.failure_mode != FAILURE_MODE_NONE) {
ret = -1;
} else {
/* If peer failure testing was requested, set `ret = 0` to report
* success from the application */
ret = 0;

/* Make sure that failure_handler was called */
while (client_status == UCS_OK) {
while (ep_status == UCS_OK) {
ucp_worker_progress(ucp_worker);
}
}
goto err_free_mem_type_msg;
}

if (err_handling_opt.failure_mode == FAILURE_MODE_KEEPALIVE) {
fprintf(stderr, "Waiting for client is terminated\n");
while (ep_status == UCS_OK) {
ucp_worker_progress(ucp_worker);
}
}

status = flush_ep(ucp_worker, client_ep);
printf("flush_ep completed with status %d (%s)\n",
status, ucs_status_string(status));
Expand Down Expand Up @@ -564,7 +599,7 @@ int main(int argc, char **argv)

ret = run_test(client_target_name, ucp_worker);

if (!ret && !err_handling_opt.failure) {
if (!ret && (err_handling_opt.failure_mode != FAILURE_MODE_NONE)) {
/* Make sure remote is disconnected before destroying local worker */
ret = barrier(oob_sock);
}
Expand All @@ -586,15 +621,38 @@ int main(int argc, char **argv)
return ret;
}

static void print_usage()
{
fprintf(stderr, "Usage: ucp_hello_world [parameters]\n");
fprintf(stderr, "UCP hello world client/server example utility\n");
fprintf(stderr, "\nParameters are:\n");
fprintf(stderr, " -w Select test mode \"wait\" to test "
"ucp_worker_wait function\n");
fprintf(stderr, " -f Select test mode \"event fd\" to test "
"ucp_worker_get_efd function with later poll\n");
fprintf(stderr, " -b Select test mode \"busy polling\" to test "
"ucp_tag_probe_nb and ucp_worker_progress (default)\n");
fprintf(stderr, " -e <type> Emulate unexpected failure and handle an "
"error with enabled UCP_ERR_HANDLING_MODE_PEER\n");
fprintf(stderr, " send - send failure on server side "
"before send initiated\n");
fprintf(stderr, " recv - receive failure on client side "
"before receive completed\n");
fprintf(stderr, " keepalive - keepalive failure on client side "
"after communication completed\n");
print_common_help();
fprintf(stderr, "\n");
}

ucs_status_t parse_cmd(int argc, char * const argv[], char **server_name)
{
int c = 0, idx = 0;
opterr = 0;

err_handling_opt.ucp_err_mode = UCP_ERR_HANDLING_MODE_NONE;
err_handling_opt.failure = 0;
err_handling_opt.ucp_err_mode = UCP_ERR_HANDLING_MODE_NONE;
err_handling_opt.failure_mode = FAILURE_MODE_NONE;

while ((c = getopt(argc, argv, "wfben:p:s:m:h")) != -1) {
while ((c = getopt(argc, argv, "wfbe:n:p:s:m:h")) != -1) {
switch (c) {
case 'w':
ucp_test_mode = TEST_MODE_WAIT;
Expand All @@ -606,8 +664,17 @@ ucs_status_t parse_cmd(int argc, char * const argv[], char **server_name)
ucp_test_mode = TEST_MODE_PROBE;
break;
case 'e':
err_handling_opt.ucp_err_mode = UCP_ERR_HANDLING_MODE_PEER;
err_handling_opt.failure = 1;
err_handling_opt.ucp_err_mode = UCP_ERR_HANDLING_MODE_PEER;
if (!strcmp(optarg, "recv")) {
err_handling_opt.failure_mode = FAILURE_MODE_RECV;
} else if (!strcmp(optarg, "send")) {
err_handling_opt.failure_mode = FAILURE_MODE_SEND;
} else if (!strcmp(optarg, "keepalive")) {
err_handling_opt.failure_mode = FAILURE_MODE_KEEPALIVE;
} else {
print_usage();
return UCS_ERR_UNSUPPORTED;
}
break;
case 'n':
*server_name = optarg;
Expand Down Expand Up @@ -643,25 +710,12 @@ ucs_status_t parse_cmd(int argc, char * const argv[], char **server_name)
/* Fall through */
case 'h':
default:
fprintf(stderr, "Usage: ucp_hello_world [parameters]\n");
fprintf(stderr, "UCP hello world client/server example utility\n");
fprintf(stderr, "\nParameters are:\n");
fprintf(stderr, " -w Select test mode \"wait\" to test "
"ucp_worker_wait function\n");
fprintf(stderr, " -f Select test mode \"event fd\" to test "
"ucp_worker_get_efd function with later poll\n");
fprintf(stderr, " -b Select test mode \"busy polling\" to test "
"ucp_tag_probe_nb and ucp_worker_progress (default)\n");
fprintf(stderr, " -e Emulate unexpected failure on server side"
"and handle an error on client side with enabled "
"UCP_ERR_HANDLING_MODE_PEER\n");
print_common_help();
fprintf(stderr, "\n");
print_usage();
return UCS_ERR_UNSUPPORTED;
}
}
fprintf(stderr, "INFO: UCP_HELLO_WORLD mode = %d server = %s port = %d\n",
ucp_test_mode, *server_name, server_port);
fprintf(stderr, "INFO: UCP_HELLO_WORLD mode = %d server = %s port = %d, pid = %d\n",
ucp_test_mode, *server_name, server_port, getpid());

for (idx = optind; idx < argc; idx++) {
fprintf(stderr, "WARNING: Non-option argument %s\n", argv[idx]);
Expand Down