From 834330ff13c7ad58c89065ab94a1fc1c7428067c Mon Sep 17 00:00:00 2001 From: Andrey Lobanov Date: Mon, 10 Jul 2017 11:50:57 +0300 Subject: [PATCH] fixes for hierarchy --- .../libhydra/bstrap/slurm/slurm_launch.c | 2 +- .../libhydra/bstrap/src/hydra_bstrap.c.in | 15 ++++++- src/pm/hydra/mpiexec/mpiexec.c | 41 +++++++++++-------- src/pm/hydra/proxy/proxy.c | 40 ++++++++++++------ src/pm/hydra/proxy/proxy_cb.c | 13 +++++- 5 files changed, 77 insertions(+), 34 deletions(-) diff --git a/src/pm/hydra/libhydra/bstrap/slurm/slurm_launch.c b/src/pm/hydra/libhydra/bstrap/slurm/slurm_launch.c index f7d2c4e43a1..a26c0178381 100644 --- a/src/pm/hydra/libhydra/bstrap/slurm/slurm_launch.c +++ b/src/pm/hydra/libhydra/bstrap/slurm/slurm_launch.c @@ -61,7 +61,7 @@ HYD_status HYDI_bstrap_slurm_launch(const char *hostname, const char *launch_exe } - status = HYD_spawn(targs, NULL, fd_stdin, fd_stdout, fd_stderr, pid, -1); + status = HYD_spawn(targs, 0, NULL, fd_stdin, fd_stdout, fd_stderr, pid, -1); HYD_ERR_POP(status, "create process returned error\n"); fn_exit: diff --git a/src/pm/hydra/libhydra/bstrap/src/hydra_bstrap.c.in b/src/pm/hydra/libhydra/bstrap/src/hydra_bstrap.c.in index 264156ad786..7a7362c780c 100644 --- a/src/pm/hydra/libhydra/bstrap/src/hydra_bstrap.c.in +++ b/src/pm/hydra/libhydra/bstrap/src/hydra_bstrap.c.in @@ -32,6 +32,7 @@ struct bstrap_control { int my_proxy_id; int total_downstreams; struct HYD_int_hash *fd_control_hash; + struct HYD_int_hash *real_proxy_id_hash; int current_downstreams; }; @@ -50,7 +51,7 @@ static HYD_status enqueue_control_fd(int fd_control, struct bstrap_control *bstr { struct HYDI_bstrap_cmd cmd; int recvd, closed; - struct HYD_int_hash *hash; + struct HYD_int_hash *hash, *real_proxy_hash; HYD_status status = HYD_SUCCESS; HYD_FUNC_ENTER(); @@ -63,9 +64,12 @@ static HYD_status enqueue_control_fd(int fd_control, struct bstrap_control *bstr /* the first command coming in should be the proxy ID */ HYD_ASSERT(cmd.type == HYDI_BSTRAP_CMD__PROXY_ID, status); + MPL_HASH_FIND_INT(bstrap_control->real_proxy_id_hash, &cmd.u.proxy_id.id, real_proxy_hash); + HYD_ASSERT(real_proxy_hash->val < bstrap_control->total_downstreams, status); + HYD_MALLOC(hash, struct HYD_int_hash *, sizeof(struct HYD_int_hash), status); hash->key = fd_control; - hash->val = cmd.u.proxy_id.id - bstrap_control->my_proxy_id - 1; + hash->val = real_proxy_hash->val; MPL_HASH_ADD_INT(bstrap_control->fd_control_hash, key, hash); bstrap_control->current_downstreams++; @@ -157,6 +161,7 @@ HYD_status HYD_bstrap_setup(const char *path, const char *launcher, const char * bstrap_control->my_proxy_id = my_proxy_id; bstrap_control->total_downstreams = (*num_downstream); bstrap_control->fd_control_hash = NULL; + bstrap_control->real_proxy_id_hash = NULL; bstrap_control->current_downstreams = 0; HYD_MALLOC(proxy_id, int *, (*num_downstream) * sizeof(int), status); @@ -196,11 +201,17 @@ HYD_status HYD_bstrap_setup(const char *path, const char *launcher, const char * /* ok, all the groundwork has been set. now loop over the set of * nodes that we need to launch on and get cracking */ for (m = 0; m < bstrap_control->total_downstreams; m++) { + struct HYD_int_hash *proxy_hash = NULL; k = HYD_tree_start(num_nodes, m, tree_width); proxy_id[m] = my_proxy_id + k + 1; subtree_size = HYD_tree_size(num_nodes, m, tree_width); + HYD_MALLOC(proxy_hash, struct HYD_int_hash *, sizeof(struct HYD_int_hash), status); + proxy_hash->key = proxy_id[m]; + proxy_hash->val = m; + MPL_HASH_ADD_INT(bstrap_control->real_proxy_id_hash, key, proxy_hash); + if (MPL_host_is_local(node_list[k].hostname)) { int sockpair[2]; diff --git a/src/pm/hydra/mpiexec/mpiexec.c b/src/pm/hydra/mpiexec/mpiexec.c index 90cfdd0789a..9611a6be55d 100644 --- a/src/pm/hydra/mpiexec/mpiexec.c +++ b/src/pm/hydra/mpiexec/mpiexec.c @@ -545,25 +545,27 @@ static HYD_status control_cb(int fd, HYD_dmx_event_t events, void *userp) } else if (cmd.type == MPX_CMD_TYPE__KVCACHE_IN) { struct HYD_int_hash *hash; + int rel_proxy_id; MPL_HASH_FIND_INT(mpiexec_pg_hash, &cmd.u.kvcache.pgid, pg); MPL_HASH_FIND_INT(pg->downstream.fd_control_hash, &fd, hash); - HYD_ASSERT(hash->val < pg->num_downstream, status); + rel_proxy_id = hash->val; + HYD_ASSERT(rel_proxy_id < pg->num_downstream, status); - HYD_ASSERT(pg->downstream.kvcache[hash->val] == NULL, status); - HYD_ASSERT(pg->downstream.kvcache_size[hash->val] == 0, status); - HYD_ASSERT(pg->downstream.kvcache_num_blocks[hash->val] == 0, status); + HYD_ASSERT(pg->downstream.kvcache[rel_proxy_id] == NULL, status); + HYD_ASSERT(pg->downstream.kvcache_size[rel_proxy_id] == 0, status); + HYD_ASSERT(pg->downstream.kvcache_num_blocks[rel_proxy_id] == 0, status); - pg->downstream.kvcache_num_blocks[hash->val] = cmd.u.kvcache.num_blocks; - pg->downstream.kvcache_size[hash->val] = cmd.data_len; + pg->downstream.kvcache_num_blocks[rel_proxy_id] = cmd.u.kvcache.num_blocks; + pg->downstream.kvcache_size[rel_proxy_id] = cmd.data_len; - HYD_MALLOC(pg->downstream.kvcache[hash->val], char *, - pg->downstream.kvcache_size[hash->val], status); + HYD_MALLOC(pg->downstream.kvcache[rel_proxy_id], char *, + pg->downstream.kvcache_size[rel_proxy_id], status); status = - HYD_sock_read(fd, pg->downstream.kvcache[hash->val], - pg->downstream.kvcache_size[hash->val], &recvd, &closed, + HYD_sock_read(fd, pg->downstream.kvcache[rel_proxy_id], + pg->downstream.kvcache_size[rel_proxy_id], &recvd, &closed, HYD_SOCK_COMM_TYPE__BLOCKING); HYD_ERR_POP(status, "error reading PMI command\n"); } @@ -623,33 +625,40 @@ static HYD_status control_cb(int fd, HYD_dmx_event_t events, void *userp) /* If we have all of the pids, post the list to the MPIR_PROCDESC struct so the debugger can find it */ if (mpiexec_params.pid_ref_count == pg->num_downstream) { HYD_dbg_setup_procdesc(pg->total_proc_count, pg->exec_list, contig_pids, pg->node_count, pg->node_list); - MPL_free(contig_pids); } } else if (cmd.type == MPX_CMD_TYPE__EXITCODE) { int *contig_data; + struct HYD_int_hash *hash; + int rel_proxy_id; MPL_HASH_FIND_INT(mpiexec_pg_hash, &cmd.u.exitcodes.pgid, pg); + + MPL_HASH_FIND_INT(pg->downstream.fd_control_hash, &fd, hash); + rel_proxy_id = hash->val; + HYD_ASSERT(rel_proxy_id < pg->num_downstream, status); + if (n_proxy_exitcodes == NULL) HYD_MALLOC(n_proxy_exitcodes, int *, pg->num_downstream * sizeof(int), status); - n_proxy_exitcodes[cmd.u.exitcodes.proxy_id] = cmd.data_len / (2 * sizeof(int)); + n_proxy_exitcodes[rel_proxy_id] = cmd.data_len / (2 * sizeof(int)); if (exitcodes == NULL) HYD_MALLOC(exitcodes, int **, pg->num_downstream * sizeof(int *), status); - HYD_MALLOC(exitcodes[cmd.u.exitcodes.proxy_id], int *, cmd.data_len / 2, status); + HYD_MALLOC(exitcodes[rel_proxy_id], int *, cmd.data_len / 2, status); if (exitcode_node_ids == NULL) HYD_MALLOC(exitcode_node_ids, int **, pg->num_downstream * sizeof(int *), status); - HYD_MALLOC(exitcode_node_ids[cmd.u.exitcodes.proxy_id], int *, cmd.data_len / 2, status); + HYD_MALLOC(exitcode_node_ids[rel_proxy_id], int *, cmd.data_len / 2, status); /* Read the data from the socket */ HYD_MALLOC(contig_data, int *, cmd.data_len, status); status = HYD_sock_read(fd, contig_data, cmd.data_len, &recvd, &closed, HYD_SOCK_COMM_TYPE__BLOCKING); + + memcpy(exitcodes[rel_proxy_id], contig_data, cmd.data_len / 2); + memcpy(exitcode_node_ids[rel_proxy_id], &contig_data[n_proxy_exitcodes[rel_proxy_id]], cmd.data_len / 2); MPL_free(contig_data); - memcpy(exitcodes[cmd.u.exitcodes.proxy_id], contig_data, cmd.data_len / 2); - memcpy(exitcode_node_ids[cmd.u.exitcodes.proxy_id], &contig_data[n_proxy_exitcodes[cmd.u.exitcodes.proxy_id]], cmd.data_len / 2); } else { HYD_ERR_SETANDJUMP(status, HYD_ERR_INTERNAL, "received unknown cmd %d\n", cmd.type); diff --git a/src/pm/hydra/proxy/proxy.c b/src/pm/hydra/proxy/proxy.c index ab64d816c65..cb0d847c919 100644 --- a/src/pm/hydra/proxy/proxy.c +++ b/src/pm/hydra/proxy/proxy.c @@ -197,7 +197,8 @@ static HYD_status get_bstrap_params(void) tmp = strtok((char *) str, ","); else tmp = strtok(NULL, ","); - HYD_ASSERT(tmp, status); + if(!tmp) + break; HYD_MALLOC(hash, struct HYD_int_hash *, sizeof(struct HYD_int_hash), status); hash->key = atoi(tmp); @@ -279,6 +280,11 @@ static HYD_status get_bstrap_params(void) proxy_params.immediate.proxy.num_children * sizeof(int), status); HYD_MALLOC(proxy_params.immediate.proxy.kvcache_num_blocks, int *, proxy_params.immediate.proxy.num_children * sizeof(int), status); + for (i = 0; i < proxy_params.immediate.proxy.num_children; i++) { + proxy_params.immediate.proxy.kvcache[i] = NULL; + proxy_params.immediate.proxy.kvcache_size[i] = 0; + proxy_params.immediate.proxy.kvcache_num_blocks[i] = 0; + } } fn_exit: @@ -585,10 +591,6 @@ int main(int argc, char **argv) status = HYD_print_set_prefix_str("proxy:unset"); HYD_ERR_POP(status, "unable to set dbg prefix\n"); - HYD_MALLOC(proxy_pids, int **, proxy_params.immediate.proxy.num_children * sizeof(int *), status); - HYD_MALLOC(n_proxy_pids, int *, proxy_params.immediate.proxy.num_children * sizeof(int), status); - HYD_MALLOC(proxy_pmi_ids, int **, proxy_params.immediate.proxy.num_children * sizeof(int *), status); - /* To launch the MPI processes, we follow a process: * (1) get parameters from the bstrap, as arguments or from * upstream, (2) make sure all the parameters we need are @@ -602,6 +604,10 @@ int main(int argc, char **argv) status = get_bstrap_params(); HYD_ERR_POP(status, "error getting parameters\n"); + HYD_MALLOC(proxy_pids, int **, (proxy_params.immediate.proxy.num_children + 1) * sizeof(int *), status); + HYD_MALLOC(n_proxy_pids, int *, (proxy_params.immediate.proxy.num_children + 1) * sizeof(int), status); + HYD_MALLOC(proxy_pmi_ids, int **, (proxy_params.immediate.proxy.num_children + 1) * sizeof(int *), status); + MPL_snprintf(dbg_prefix, 2 * HYD_MAX_HOSTNAME_LEN, "proxy:%d:%d", proxy_params.all.pgid, proxy_params.root.proxy_id); status = HYD_print_set_prefix_str((const char *) dbg_prefix); @@ -656,6 +662,8 @@ int main(int argc, char **argv) /* step 3: close all downstream stdin fds */ MPL_HASH_ITER(hh, proxy_params.immediate.proxy.fd_stdin_hash, hash, tmp) { close(hash->key); + MPL_HASH_DEL(proxy_params.immediate.proxy.fd_stdin_hash, hash); + MPL_free(hash); } /* step 4: launch processes */ @@ -675,6 +683,13 @@ int main(int argc, char **argv) } proxy_send_pids_upstream(); + HYD_MALLOC(n_proxy_exitcodes, int *, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int), status); + n_proxy_exitcodes[0] = proxy_params.immediate.process.num_children; + HYD_MALLOC(exitcodes, int **, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int *), status); + HYD_MALLOC(exitcodes[0], int *, (1 + proxy_params.immediate.process.num_children) * sizeof(int), status); + HYD_MALLOC(exitcode_node_ids, int **, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int *), status); + HYD_MALLOC(exitcode_node_ids[0], int *, (1 + proxy_params.immediate.process.num_children) * sizeof(int), status); + /* The launch is now complete: we wait for the processes to * complete their execution, in a five-step process: (1) wait for * events till the stdout/stderr/pmi-fd sockets of all MPI @@ -712,10 +727,6 @@ int main(int argc, char **argv) } /* step 2: wait for MPI process to terminate */ - HYD_MALLOC(exitcodes, int **, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int *), status); - HYD_MALLOC(exitcodes[0], int *, (1 + proxy_params.immediate.process.num_children) * sizeof(int), status); - HYD_MALLOC(exitcode_node_ids, int **, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int *), status); - HYD_MALLOC(exitcode_node_ids[0], int *, (1 + proxy_params.immediate.process.num_children) * sizeof(int), status); MPL_HASH_ITER(hh, proxy_params.immediate.process.pid_hash, hash, tmp) { waitpid(hash->key, &tmp_ret, 0); exitcodes[0][hash->val] = WEXITSTATUS(tmp_ret); @@ -723,8 +734,6 @@ int main(int argc, char **argv) MPL_HASH_DEL(proxy_params.immediate.process.pid_hash, hash); MPL_free(hash); } - HYD_MALLOC(n_proxy_exitcodes, int *, (1 + proxy_params.immediate.proxy.num_children) * sizeof(int), status); - n_proxy_exitcodes[0] = proxy_params.immediate.process.num_children; proxy_send_exitcodes_upstream(); /* step 3: wait for proxy stdout/stderr to close */ @@ -754,7 +763,7 @@ int main(int argc, char **argv) } /* step 4: wait for proxies to terminate */ - MPL_HASH_ITER(hh, proxy_params.immediate.process.pid_hash, hash, tmp) { + MPL_HASH_ITER(hh, proxy_params.immediate.proxy.pid_hash, hash, tmp) { int ret = 0; waitpid(hash->key, &ret, 0); @@ -771,7 +780,7 @@ int main(int argc, char **argv) } } - MPL_HASH_DEL(proxy_params.immediate.process.pid_hash, hash); + MPL_HASH_DEL(proxy_params.immediate.proxy.pid_hash, hash); } /* step 5: deregister upstream fd and stdin */ @@ -783,6 +792,11 @@ int main(int argc, char **argv) status = HYD_dmx_deregister_fd(proxy_params.root.upstream_fd); HYD_ERR_POP(status, "error deregistering upstream fd\n"); + MPL_HASH_ITER(hh, proxy_params.immediate.proxy.fd_control_hash, hash, tmp) { + MPL_HASH_DEL(proxy_params.immediate.proxy.fd_control_hash, hash); + MPL_free(hash); + } + if (proxy_params.cwd) MPL_free(proxy_params.cwd); diff --git a/src/pm/hydra/proxy/proxy_cb.c b/src/pm/hydra/proxy/proxy_cb.c index 16bdf5c5b57..f87234ce30e 100644 --- a/src/pm/hydra/proxy/proxy_cb.c +++ b/src/pm/hydra/proxy/proxy_cb.c @@ -285,9 +285,12 @@ HYD_status proxy_downstream_control_cb(int fd, HYD_dmx_event_t events, void *use { int rel_proxy_id; int *contig_data; + struct HYD_int_hash *hash; /* Find the proxy id of the proxy sending the data */ - rel_proxy_id = cmd.u.pids.proxy_id - proxy_params.root.proxy_id; + MPL_HASH_FIND_INT(proxy_params.immediate.proxy.fd_control_hash, &fd, hash); + rel_proxy_id = hash->val + 1; + n_proxy_pids[rel_proxy_id] = cmd.data_len / (2 * sizeof(int)); /* Read the data from the socket */ @@ -297,20 +300,24 @@ HYD_status proxy_downstream_control_cb(int fd, HYD_dmx_event_t events, void *use status = HYD_sock_read(fd, contig_data, cmd.data_len, &recvd, &closed, HYD_SOCK_COMM_TYPE__BLOCKING); + HYD_ERR_POP(status, "error reading pids from downstream\n"); memcpy(proxy_pids[rel_proxy_id], contig_data, cmd.data_len / 2); memcpy(proxy_pmi_ids[rel_proxy_id], &contig_data[n_proxy_pids[rel_proxy_id]], cmd.data_len / 2); /* Call the function to stitch it all together */ proxy_send_pids_upstream(); + break; } case MPX_CMD_TYPE__EXITCODE: { int rel_proxy_id; int *contig_data; + struct HYD_int_hash *hash; /* Find the proxy id of the proxy sending the data */ - rel_proxy_id = cmd.u.pids.proxy_id - proxy_params.root.proxy_id; + MPL_HASH_FIND_INT(proxy_params.immediate.proxy.fd_control_hash, &fd, hash); + rel_proxy_id = hash->val + 1; n_proxy_exitcodes[rel_proxy_id] = cmd.data_len / (2 * sizeof(int)); /* Read the data from the socket */ @@ -320,12 +327,14 @@ HYD_status proxy_downstream_control_cb(int fd, HYD_dmx_event_t events, void *use status = HYD_sock_read(fd, contig_data, cmd.data_len, &recvd, &closed, HYD_SOCK_COMM_TYPE__BLOCKING); + HYD_ERR_POP(status, "error reading exitcodes from downstream\n"); memcpy(exitcodes[rel_proxy_id], contig_data, cmd.data_len / 2); memcpy(exitcode_node_ids[rel_proxy_id], &contig_data[n_proxy_exitcodes[rel_proxy_id]], cmd.data_len / 2); /* Call the function to stitch it all together */ proxy_send_exitcodes_upstream(); + break; } default: