Skip to content

fixes for hierarchy #6

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

Open
alobano1 wants to merge 1 commit into
base: wip/pmi
Choose a base branch
from
Open

fixes for hierarchy #6

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
2 changes: 1 addition & 1 deletion src/pm/hydra/libhydra/bstrap/slurm/slurm_launch.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
15 changes: 13 additions & 2 deletions src/pm/hydra/libhydra/bstrap/src/hydra_bstrap.c.in
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
};

Expand All @@ -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();
Expand All @@ -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++;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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];

Expand Down
41 changes: 25 additions & 16 deletions src/pm/hydra/mpiexec/mpiexec.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
Expand Down Expand Up @@ -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);
Expand Down
40 changes: 27 additions & 13 deletions src/pm/hydra/proxy/proxy.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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);
Expand Down Expand Up @@ -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 */
Expand All @@ -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
Expand Down Expand Up @@ -712,19 +727,13 @@ 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);
exitcode_node_ids[0][hash->val] = proxy_params.root.node_id;
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 */
Expand Down Expand Up @@ -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);
Expand All @@ -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 */
Expand All @@ -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);

Expand Down
13 changes: 11 additions & 2 deletions src/pm/hydra/proxy/proxy_cb.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 */
Expand All @@ -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 */
Expand All @@ -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:
Expand Down