Google Git
Sign in
third-party-mirror / SchedMD / slurm / 5ba844f540900017f593e9caf70005e0e009ad30 / . / src / srun / srun_job.c
blob: a566af83a189a5ec7b74edfe45210a0808b6aa06 [file] [log] [blame]
/****************************************************************************\
* srun_job.c - job data structure creation functions
*****************************************************************************
* Copyright (C) 2002-2007 The Regents of the University of California.
* Copyright (C) 2008 Lawrence Livermore National Security.
* Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
* Written by Mark Grondona <grondona@llnl.gov>.
* CODE-OCEC-09-009. All rights reserved.
*
* This file is part of Slurm, a resource management program.
* For details, see <https://slurm.schedmd.com/>.
* Please also read the included file: DISCLAIMER.
*
* Slurm is free software; you can redistribute it and/or modify it under
* the terms of the GNU General Public License as published by the Free
* Software Foundation; either version 2 of the License, or (at your option)
* any later version.
*
* In addition, as a special exception, the copyright holders give permission
* to link the code of portions of this program with the OpenSSL library under
* certain conditions as described in each individual source file, and
* distribute linked combinations including the two. You must obey the GNU
* General Public License in all respects for all of the code used other than
* OpenSSL. If you modify file(s) with this exception, you may extend this
* exception to your version of the file(s), but you are not obligated to do
* so. If you do not wish to do so, delete this exception statement from your
* version. If you delete this exception statement from all source files in
* the program, then also delete it here.
*
* Slurm is distributed in the hope that it will be useful, but WITHOUT ANY
* WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along
* with Slurm; if not, write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
\*****************************************************************************/
#include "config.h"
#include <fcntl.h>
#include <grp.h>
#include <limits.h>
#include <netdb.h>
#include <signal.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "src/common/bitstring.h"
#include "src/interfaces/cli_filter.h"
#include "src/common/cbuf.h"
#include "src/common/fd.h"
#include "src/common/forward.h"
#include "src/common/hostlist.h"
#include "src/common/log.h"
#include "src/common/macros.h"
#include "src/common/proc_args.h"
#include "src/common/read_config.h"
#include "src/common/slurm_opt.h"
#include "src/common/slurm_protocol_api.h"
#include "src/common/slurm_rlimits_info.h"
#include "src/common/spank.h"
#include "src/common/uid.h"
#include "src/common/xmalloc.h"
#include "src/common/xsignal.h"
#include "src/common/xstring.h"
#include "src/api/step_launch.h"
#include "src/srun/allocate.h"
#include "src/srun/debugger.h"
#include "src/srun/fname.h"
#include "src/srun/launch.h"
#include "src/srun/opt.h"
#include "src/srun/multi_prog.h"
#include "src/srun/srun_job.h"
/*
* allocation information structure used to store general information
* about node allocation to be passed to _job_create_structure()
*/
typedef struct allocation_info {
uint16_t *cpus_per_node;
uint32_t *cpu_count_reps;
uint32_t nnodes;
char *nodelist;
uint16_t ntasks_per_board;/* number of tasks to invoke on each board */
uint16_t ntasks_per_core; /* number of tasks to invoke on each core */
uint16_t ntasks_per_tres; /* number of tasks that can access each gpu */
uint16_t ntasks_per_socket;/* number of tasks to invoke on
* each socket */
uint32_t num_cpu_groups;
char *partition;
slurm_step_id_t step_id;
uid_t uid; /* resolved user id of job */
char *user_name; /* resolved user name of job */
gid_t gid; /* resolved group id of job */
char *group_name; /* resolved group name of job */
} allocation_info_t;
typedef struct het_job_resp_struct {
uint16_t *cpu_cnt;
hostlist_t *host_list;
uint32_t node_cnt;
} het_job_resp_struct_t;
static int shepherd_fd = -1;
static pthread_t signal_thread = (pthread_t) 0;
static int pty_sigarray[] = { SIGWINCH, 0 };
extern char **environ;
/*
* Prototypes:
*/
static void _call_spank_fini(void);
static int _call_spank_local_user(srun_job_t *job, slurm_opt_t *opt_local);
static long _diff_tv_str(struct timeval *tv1, struct timeval *tv2);
static void _handle_intr(srun_job_t *job);
static void _handle_pipe(void);
static srun_job_t *_job_create_structure(allocation_info_t *ainfo,
slurm_opt_t *opt_local);
static char *_normalize_hostlist(const char *hostlist);
static void _print_job_information(resource_allocation_response_msg_t *resp);
static void _run_srun_epilog (srun_job_t *job);
static void _run_srun_prolog (srun_job_t *job);
static int _run_srun_script (srun_job_t *job, char *script);
static void _set_env_vars(resource_allocation_response_msg_t *resp,
int het_job_offset);
static void _set_env_vars2(resource_allocation_response_msg_t *resp,
int het_job_offset);
static void _set_ntasks(allocation_info_t *ai, slurm_opt_t *opt_local);
static int _set_rlimit_env(void);
static void _set_submit_dir_env(void);
static int _set_umask_env(void);
static void _shepherd_notify(int shepherd_fd);
static int _shepherd_spawn(srun_job_t *job, list_t *srun_job_list,
bool got_alloc);
static void *_srun_signal_mgr(void *no_data);
static void _srun_cli_filter_post_submit(uint32_t jobid, uint32_t stepid);
static int _validate_relative(resource_allocation_response_msg_t *resp,
slurm_opt_t *opt_local);
/*
* Create an srun job structure w/out an allocation response msg.
* (i.e. use the command line options)
*/
srun_job_t *
job_create_noalloc(void)
{
srun_job_t *job = NULL;
allocation_info_t *ai = xmalloc(sizeof(allocation_info_t));
uint16_t cpn[1];
uint32_t cpu_count_reps[1];
slurm_opt_t *opt_local = &opt;
hostlist_t *hl = hostlist_create(opt_local->nodelist);
if (!hl) {
error("Invalid node list `%s' specified", opt_local->nodelist);
goto error;
}
srand48(getpid());
ai->step_id.job_id = MIN_NOALLOC_JOBID +
((uint32_t) lrand48() %
(MAX_NOALLOC_JOBID - MIN_NOALLOC_JOBID + 1));
ai->step_id.step_id = (uint32_t) (lrand48());
ai->step_id.step_het_comp = NO_VAL;
ai->nodelist = opt_local->nodelist;
ai->nnodes = hostlist_count(hl);
ai->uid = getuid();
ai->user_name = uid_to_string_or_null(ai->uid);
ai->gid = getgid();
ai->group_name = gid_to_string_or_null(ai->gid);
hostlist_destroy(hl);
cpn[0] = ROUNDUP(opt_local->ntasks, ai->nnodes);
ai->cpus_per_node = cpn;
cpu_count_reps[0] = ai->nnodes;
ai->cpu_count_reps = cpu_count_reps;
ai->num_cpu_groups = 1;
/*
* Create job, then fill in host addresses
*/
job = _job_create_structure(ai, opt_local);
if (job != NULL)
job_update_io_fnames(job, opt_local);
if (job && (job->ntasks == NO_VAL)) {
job->ntasks = ai->nnodes;
job->cpu_count = opt_local->cpus_per_task * job->ntasks;
}
error:
xfree(ai);
return (job);
}
static void _set_min_node_count(allocation_info_t *ai,
resource_allocation_response_msg_t *resp,
slurm_opt_t *opt_local)
{
int num_tasks;
if (opt_local->nodes_set)
return;
opt_local->nodes_set = true;
if (!local_het_step) {
/*
* we don't want to set the number of nodes =
* to the number of requested processes unless we
* know it is less than the number of nodes
* in the allocation
*/
if (opt_local->ntasks_set &&
(opt_local->ntasks < ai->nnodes))
opt_local->min_nodes = opt_local->ntasks;
else
opt_local->min_nodes = ai->nnodes;
return;
}
/*
* Here we want to try to figure out what the minimum amount of nodes
* should be needed to put this step into the allocation.
*/
num_tasks = 0;
opt_local->min_nodes = 0;
for (int i = 0; ((i < resp->num_cpu_groups) &&
(opt_local->min_nodes < resp->node_cnt));
i++) {
for (int j = 0; j < resp->cpu_count_reps[i]; j++) {
/*
* Given this node, figure out how many tasks could fit
* on it.
*/
int ntasks_per_node = resp->cpus_per_node[i];
if (opt_local->cpus_per_task)
ntasks_per_node /=
opt_local->cpus_per_task;
if ((opt_local->ntasks_per_node != NO_VAL) &&
(ntasks_per_node >= opt_local->ntasks_per_node))
ntasks_per_node = opt_local->ntasks_per_node;
/* Then add it to the total task count */
num_tasks += ntasks_per_node;
opt_local->min_nodes++;
if (num_tasks >= opt_local->ntasks)
return;
}
}
}
/*
* Create an srun job structure for a step w/out an allocation response msg.
* (i.e. inside an allocation)
*/
extern srun_job_t *job_step_create_allocation(
resource_allocation_response_msg_t *resp,
slurm_opt_t *opt_local)
{
srun_opt_t *srun_opt = opt_local->srun_opt;
uint32_t job_id = resp->job_id;
srun_job_t *job = NULL;
allocation_info_t *ai = xmalloc(sizeof(allocation_info_t));
hostlist_t *hl = NULL;
char *buf = NULL;
int count = 0;
uint32_t alloc_count = 0;
char *step_nodelist = NULL;
xassert(srun_opt);
ai->step_id.job_id = job_id;
ai->step_id.step_id = NO_VAL;
ai->step_id.step_het_comp = NO_VAL;
if (srun_opt->alloc_nodelist)
ai->nodelist = xstrdup(srun_opt->alloc_nodelist);
else
ai->nodelist = xstrdup(resp->node_list);
hl = hostlist_create(ai->nodelist);
hostlist_uniq(hl);
alloc_count = hostlist_count(hl);
ai->nnodes = alloc_count;
hostlist_destroy(hl);
/* exclude is handled elsewhere for het steps */
if (!local_het_step && opt_local->exclude) {
hostlist_t *exc_hl = hostlist_create(opt_local->exclude);
hostlist_t *inc_hl = NULL;
char *node_name = NULL;
hl = hostlist_create(ai->nodelist);
if (opt_local->nodelist)
inc_hl = hostlist_create(opt_local->nodelist);
hostlist_uniq(hl);
//info("using %s or %s", opt_local->nodelist, ai->nodelist);
while ((node_name = hostlist_shift(exc_hl))) {
int inx = hostlist_find(hl, node_name);
if (inx >= 0) {
debug("excluding node %s", node_name);
hostlist_delete_nth(hl, inx);
ai->nnodes--; /* decrement node count */
}
if (inc_hl) {
inx = hostlist_find(inc_hl, node_name);
if (inx >= 0) {
error("Requested node %s is also "
"in the excluded list.",
node_name);
error("Job not submitted.");
hostlist_destroy(exc_hl);
hostlist_destroy(inc_hl);
goto error;
}
}
free(node_name);
}
hostlist_destroy(exc_hl);
/* we need to set this here so if there are more nodes
* available than we requested we can set it
* straight. If there is no exclude list then we set
* the vars then.
*/
if (!opt_local->nodes_set) {
/* we don't want to set the number of nodes =
* to the number of requested processes unless we
* know it is less than the number of nodes
* in the allocation
*/
if (opt_local->ntasks_set &&
(opt_local->ntasks < ai->nnodes))
opt_local->min_nodes = opt_local->ntasks;
else
opt_local->min_nodes = ai->nnodes;
opt_local->nodes_set = true;
}
if (!opt_local->max_nodes)
opt_local->max_nodes = opt_local->min_nodes;
if ((opt_local->max_nodes > 0) &&
(opt_local->max_nodes < ai->nnodes))
ai->nnodes = opt_local->max_nodes;
count = hostlist_count(hl);
if (!count) {
error("Hostlist is empty! Can't run job.");
hostlist_destroy(hl);
goto error;
}
if (inc_hl) {
count = hostlist_count(inc_hl);
if (count < ai->nnodes) {
/* add more nodes to get correct number for
allocation */
hostlist_t *tmp_hl = hostlist_copy(hl);
int i = 0;
int diff = ai->nnodes - count;
buf = hostlist_ranged_string_xmalloc(inc_hl);
hostlist_delete(tmp_hl, buf);
xfree(buf);
while ((i < diff) &&
(node_name = hostlist_shift(tmp_hl))) {
hostlist_push_host(inc_hl, node_name);
free(node_name);
i++;
}
hostlist_destroy(tmp_hl);
}
buf = hostlist_ranged_string_xmalloc(inc_hl);
hostlist_destroy(inc_hl);
xfree(opt_local->nodelist);
opt_local->nodelist = buf;
} else {
/* remove more nodes than needed for allocation */
for (int i = count; i > ai->nnodes; i--) {
hostlist_delete_nth(hl, i - 1);
}
xfree(opt_local->nodelist);
opt_local->nodelist = hostlist_ranged_string_xmalloc(hl);
}
hostlist_destroy(hl);
} else {
_set_min_node_count(ai, resp, opt_local);
if (!opt_local->max_nodes)
opt_local->max_nodes = opt_local->min_nodes;
if ((opt_local->max_nodes > 0) &&
(opt_local->max_nodes < ai->nnodes))
ai->nnodes = opt_local->max_nodes;
/* Don't reset the ai->nodelist because that is the
* nodelist we want to say the allocation is under
* opt_local->nodelist is what is used for the allocation.
*/
/* xfree(ai->nodelist); */
/* ai->nodelist = xstrdup(buf); */
}
/* get the correct number of hosts to run tasks on */
if (opt_local->nodelist)
step_nodelist = opt_local->nodelist;
else if (((opt_local->distribution & SLURM_DIST_STATE_BASE) ==
SLURM_DIST_ARBITRARY) && (count == 0))
step_nodelist = getenv("SLURM_ARBITRARY_NODELIST");
if (step_nodelist) {
hl = hostlist_create(step_nodelist);
if ((opt_local->distribution & SLURM_DIST_STATE_BASE) !=
SLURM_DIST_ARBITRARY)
hostlist_uniq(hl);
if (!hostlist_count(hl)) {
error("Hostlist is empty! Can not run job.");
hostlist_destroy(hl);
goto error;
}
buf = hostlist_ranged_string_xmalloc(hl);
count = hostlist_count(hl);
hostlist_destroy(hl);
/*
* Don't reset the ai->nodelist because that is the
* nodelist we want to say the allocation is under
* opt_local->nodelist is what is used for the allocation.
*/
/* xfree(ai->nodelist); */
/* ai->nodelist = xstrdup(buf); */
xfree(opt_local->nodelist);
opt_local->nodelist = buf;
}
if (((opt_local->distribution & SLURM_DIST_STATE_BASE) ==
SLURM_DIST_ARBITRARY) && (count != opt_local->ntasks)) {
error("You asked for %d tasks but hostlist specified %d nodes",
opt_local->ntasks, count);
goto error;
}
if (ai->nnodes == 0) {
error("No nodes in allocation, can't run job");
goto error;
}
ai->num_cpu_groups = resp->num_cpu_groups;
ai->cpus_per_node = resp->cpus_per_node;
ai->cpu_count_reps = resp->cpu_count_reps;
ai->ntasks_per_board = resp->ntasks_per_board;
/* Here let the srun options override the allocation resp */
ai->ntasks_per_core = (opt_local->ntasks_per_core != NO_VAL) ?
opt_local->ntasks_per_core : resp->ntasks_per_core;
ai->ntasks_per_socket = (opt_local->ntasks_per_socket != NO_VAL) ?
opt_local->ntasks_per_socket : resp->ntasks_per_socket;
ai->partition = resp->partition;
/* info("looking for %d nodes out of %s with a must list of %s", */
/* ai->nnodes, ai->nodelist, opt_local->nodelist); */
/*
* Create job
*/
job = _job_create_structure(ai, opt_local);
error:
xfree(ai->nodelist);
xfree(ai);
return (job);
}
/*
* Create an srun job structure from a resource allocation response msg
*/
extern srun_job_t *job_create_allocation(
resource_allocation_response_msg_t *resp,
slurm_opt_t *opt_local)
{
srun_job_t *job;
allocation_info_t *i = xmalloc(sizeof(allocation_info_t));
i->nodelist = _normalize_hostlist(resp->node_list);
i->nnodes = resp->node_cnt;
i->partition = resp->partition;
i->step_id.job_id = resp->job_id;
i->step_id.step_id = NO_VAL;
i->step_id.step_het_comp = NO_VAL;
i->num_cpu_groups = resp->num_cpu_groups;
i->cpus_per_node = resp->cpus_per_node;
i->cpu_count_reps = resp->cpu_count_reps;
i->ntasks_per_board = resp->ntasks_per_board;
i->ntasks_per_core = resp->ntasks_per_core;
i->ntasks_per_socket = resp->ntasks_per_socket;
i->uid = resp->uid;
i->user_name = xstrdup(resp->user_name);
i->gid = resp->gid;
i->group_name = xstrdup(resp->group_name);
job = _job_create_structure(i, opt_local);
if (job) {
job->account = xstrdup(resp->account);
job->qos = xstrdup(resp->qos);
job->resv_name = xstrdup(resp->resv_name);
}
xfree(i->nodelist);
xfree(i);
return (job);
}
static void _copy_args(list_t *missing_argc_list, slurm_opt_t *opt_master)
{
list_itr_t *iter;
slurm_opt_t *opt_local;
int i;
iter = list_iterator_create(missing_argc_list);
while ((opt_local = list_next(iter))) {
opt_local->argc = opt_master->argc;
opt_local->argv = xcalloc(sizeof(char *),
(opt_local->argc + 1));
for (i = 0; i < opt_local->argc; i++)
opt_local->argv[i] = xstrdup(opt_master->argv[i]);
list_remove(iter);
}
list_iterator_destroy(iter);
}
/*
* Build "het_group" string. If set on execute line, it may need to be
* rebuilt for multiple option structures ("--het-group=1,2" becomes two
* opt structures). Clear "het_grp_bits".if determined to not be a hetjob.
*/
static void _het_grp_test(list_t *opt_list)
{
list_itr_t *iter;
int het_job_offset;
bitstr_t *master_map = NULL;
list_t *missing_argv_list = NULL;
bool multi_comp = false, multi_prog = false;
if (opt_list) {
slurm_opt_t *opt_local;
missing_argv_list = list_create(NULL);
iter = list_iterator_create(opt_list);
while ((opt_local = list_next(iter))) {
srun_opt_t *srun_opt = opt_local->srun_opt;
xassert(srun_opt);
if (opt_local->argc == 0)
list_append(missing_argv_list, opt_local);
else
_copy_args(missing_argv_list, opt_local);
xfree(srun_opt->het_group);
if (srun_opt->het_grp_bits &&
((het_job_offset =
bit_ffs(srun_opt->het_grp_bits)) >= 0)) {
xstrfmtcat(srun_opt->het_group, "%d",
het_job_offset);
}
if (!srun_opt->het_grp_bits) {
error("%s: het_grp_bits is NULL", __func__);
} else if (!master_map) {
master_map
= bit_copy(srun_opt->het_grp_bits);
} else {
if (bit_overlap_any(master_map,
srun_opt->het_grp_bits)) {
fatal("Duplicate het groups in single srun not supported");
}
bit_or(master_map, srun_opt->het_grp_bits);
}
if (srun_opt->multi_prog)
multi_prog = true;
}
if (master_map && (bit_set_count(master_map) > 1))
multi_comp = true;
FREE_NULL_BITMAP(master_map);
list_iterator_destroy(iter);
FREE_NULL_LIST(missing_argv_list);
} else if (!sropt.het_group && !getenv("SLURM_HET_SIZE")) {
FREE_NULL_BITMAP(sropt.het_grp_bits);
/* het_group is already NULL */
} else if (!sropt.het_group && sropt.het_grp_bits) {
if ((het_job_offset = bit_ffs(sropt.het_grp_bits)) < 0)
het_job_offset = 0;
else if (bit_set_count(sropt.het_grp_bits) > 1)
multi_comp = true;
if (sropt.multi_prog)
multi_prog = true;
xstrfmtcat(sropt.het_group, "%d", het_job_offset);
}
if (multi_comp && multi_prog)
fatal("--multi-prog option not supported with multiple het groups");
}
/*
* Copy job name from last component to all hetjob components unless
* explicitly set.
*/
static void _match_job_name(list_t *opt_list)
{
int cnt;
list_itr_t *iter;
slurm_opt_t *opt_local;
if (!opt_list)
return;
cnt = list_count(opt_list);
if (cnt < 2)
return;
iter = list_iterator_create(opt_list);
while ((opt_local = list_next(iter))) {
if (!opt_local->job_name)
opt_local->job_name = xstrdup(opt.job_name);
if (opt_local->open_mode == 0) {
opt_local->open_mode = OPEN_MODE_APPEND;
}
}
list_iterator_destroy(iter);
}
static int _sort_by_offset(void *x, void *y)
{
slurm_opt_t *opt_local1 = *(slurm_opt_t **) x;
slurm_opt_t *opt_local2 = *(slurm_opt_t **) y;
int offset1 = -1, offset2 = -1;
if (opt_local1->srun_opt->het_grp_bits)
offset1 = bit_ffs(opt_local1->srun_opt->het_grp_bits);
if (opt_local2->srun_opt->het_grp_bits)
offset2 = bit_ffs(opt_local2->srun_opt->het_grp_bits);
if (offset1 < offset2)
return -1;
if (offset1 > offset2)
return 1;
return 0;
}
static void _post_opts(list_t *opt_list)
{
_het_grp_test(opt_list);
_match_job_name(opt_list);
if (opt_list)
list_sort(opt_list, _sort_by_offset);
}
extern void init_srun(int argc, char **argv, log_options_t *logopt,
bool handle_signals)
{
bool het_job_fini = false;
int i, het_job_argc, het_job_argc_off;
char **het_job_argv;
/*
* This must happen before we spawn any threads
* which are not designed to handle arbitrary signals
*/
if (handle_signals) {
if (xsignal_block(sig_array) < 0)
error("Unable to block signals");
}
xsignal_block(pty_sigarray);
/*
* Initialize plugin stack, read options from plugins, etc.
*/
init_spank_env();
if (spank_init(NULL)) {
error("Plug-in initialization failed");
exit(error_exit);
}
/*
* Be sure to call spank_fini when srun exits.
*/
if (atexit(_call_spank_fini) < 0)
error("Failed to register atexit handler for plugins: %m");
opt.submit_line = slurm_option_get_argv_str(argc, argv);
het_job_argc = argc;
het_job_argv = argv;
while (!het_job_fini) {
het_job_argc_off = -1;
if (initialize_and_process_args(het_job_argc, het_job_argv,
&het_job_argc_off) < 0) {
error("srun parameter parsing");
exit(1);
}
if ((het_job_argc_off >= 0) &&
(het_job_argc_off < het_job_argc)) {
for (i = het_job_argc_off; i < het_job_argc; i++) {
if (!xstrcmp(het_job_argv[i], ":")) {
het_job_argc_off = i;
break;
}
}
}
if ((het_job_argc_off >= 0) &&
(het_job_argc_off < het_job_argc) &&
!xstrcmp(het_job_argv[het_job_argc_off], ":")) {
/*
* move het_job_argv[0] from "srun" to ":"
*/
het_job_argc -= het_job_argc_off;
het_job_argv += het_job_argc_off;
colon_cnt++;
} else {
het_job_fini = true;
}
}
if (!mpi_g_client_init(&sropt.mpi_type)) {
error("Invalid MPI type '%s', --mpi=list for acceptable types",
sropt.mpi_type);
exit(error_exit);
}
_post_opts(opt_list);
/*
* reinit log with new verbosity (if changed by command line)
*/
if (logopt && (opt.verbose || opt.quiet)) {
/*
* If log level is already increased, only increment the
* level to the difference of opt.verbose an LOG_LEVEL_INFO
*/
if ((opt.verbose -= (logopt->stderr_level - LOG_LEVEL_INFO)) > 0)
logopt->stderr_level += opt.verbose;
logopt->stderr_level -= opt.quiet;
logopt->prefix_level = 1;
log_alter(*logopt, 0, NULL);
}
(void) _set_rlimit_env();
set_prio_process_env();
(void) _set_umask_env();
_set_submit_dir_env();
/*
* save process startup time to be used with -I<timeout>
*/
srun_begin_time = time(NULL);
}
/*
* Modify options for a job step (after job allocation is complete
*/
static void _set_step_opts(slurm_opt_t *opt_local,
resource_allocation_response_msg_t *resp)
{
srun_opt_t *srun_opt = opt_local->srun_opt;
int new_cpt;
xassert(srun_opt);
opt_local->time_limit = NO_VAL;/* not applicable for step, only job */
xfree(opt_local->constraint); /* not applicable for this step */
if ((srun_opt->core_spec_set || srun_opt->exclusive)
&& opt_local->cpus_set) {
/* Step gets specified CPU count, which may only part
* of the job allocation. */
srun_opt->exclusive = true;
} else {
/* Step gets all CPUs in the job allocation. */
srun_opt->exclusive = false;
}
new_cpt = slurm_opt_get_tres_per_task_cpu_cnt(resp->tres_per_task);
if (new_cpt)
opt_local->cpus_per_task = new_cpt;
if (resp->tres_per_task) {
xfree(opt_local->tres_per_task);
SWAP(opt_local->tres_per_task, resp->tres_per_task);
}
}
static int _handle_het_step_exclude(srun_job_t *job, slurm_opt_t *opt_local,
hostlist_t *exclude_hl_in)
{
hostlist_t *exclude_hl, *allocation_hl;
int rc = SLURM_SUCCESS;
if (!exclude_hl_in || !hostlist_count(exclude_hl_in))
return rc;
allocation_hl = hostlist_create(job->nodelist);
hostlist_uniq(allocation_hl);
exclude_hl = hostlist_copy(exclude_hl_in);
hostlist_push(exclude_hl, opt_local->exclude);
hostlist_uniq(exclude_hl);
hostlist_sort(exclude_hl);
xfree(opt_local->exclude);
opt_local->exclude = hostlist_ranged_string_xmalloc(exclude_hl);
if ((hostlist_count(allocation_hl) - hostlist_count(exclude_hl)) <
opt_local->min_nodes) {
error("Allocation failure of %d nodes: job size of %d, already allocated %d nodes to previous components.",
opt_local->min_nodes, hostlist_count(allocation_hl),
hostlist_count(exclude_hl));
rc = SLURM_ERROR;
goto end_it;
}
if (opt_local->nodelist) {
char *node_name = NULL;
hostlist_t *inc_hl = hostlist_create(opt_local->nodelist);
while ((node_name = hostlist_shift(exclude_hl))) {
if (hostlist_find(inc_hl, node_name) >= 0) {
error("Requested nodelist %s overlaps with excluded %s.",
opt_local->nodelist,
opt_local->exclude);
error("Job not submitted.");
rc = SLURM_ERROR;
break;
}
free(node_name);
}
FREE_NULL_HOSTLIST(inc_hl);
}
end_it:
FREE_NULL_HOSTLIST(allocation_hl);
FREE_NULL_HOSTLIST(exclude_hl);
return rc;
}
/*
* Create the job step(s). For a heterogeneous job, each step is requested in
* a separate RPC. create_job_step() references "opt", so we need to match up
* the job allocation request with its requested options.
*/
static int _create_job_step(srun_job_t *job, bool use_all_cpus,
list_t *srun_job_list, uint32_t het_job_id,
char *het_job_nodelist)
{
list_itr_t *opt_iter = NULL, *job_iter;
slurm_opt_t *opt_local = &opt;
uint32_t node_offset = 0, het_job_nnodes = 0, step_id = NO_VAL;
uint32_t het_job_ntasks = 0, task_offset = 0;
bool update_het_nnodes = false;
uint32_t updated_het_nnodes;
uint32_t updated_het_ntasks = 0;
job_step_create_response_msg_t *step_resp;
char *resv_ports = NULL;
int rc = 0;
if (srun_job_list) {
hostlist_t *exclude_hl = NULL;
if (local_het_step)
exclude_hl = hostlist_create(NULL);
if (opt_list)
opt_iter = list_iterator_create(opt_list);
job_iter = list_iterator_create(srun_job_list);
while ((job = list_next(job_iter))) {
if (het_job_id)
job->het_job_id = het_job_id;
job->step_id.step_id = NO_VAL;
/*
* Only set the step_het_comp if we are in a het step
* from a single allocation
*/
if (local_het_step)
job->step_id.step_het_comp =
job->het_job_offset;
else
job->step_id.step_het_comp = NO_VAL;
het_job_nnodes += job->nhosts;
if (job->ntasks == NO_VAL)
het_job_ntasks = NO_VAL;
else if (het_job_ntasks != NO_VAL)
het_job_ntasks += job->ntasks;
}
updated_het_nnodes = het_job_nnodes;
list_iterator_reset(job_iter);
while ((job = list_next(job_iter))) {
uint32_t old_nhosts = job->nhosts;
if (opt_list)
opt_local = list_next(opt_iter);
if (!opt_local)
fatal("%s: opt_list too short", __func__);
job->het_job_node_offset = node_offset;
job->het_job_nnodes = het_job_nnodes;
job->het_job_ntasks = het_job_ntasks;
job->het_job_task_offset = task_offset;
if (step_id != NO_VAL)
job->step_id.step_id = step_id;
if ((rc = _handle_het_step_exclude(
job, opt_local, exclude_hl)) !=
SLURM_SUCCESS)
break;
rc = create_job_step(job, use_all_cpus, opt_local);
if (rc < 0)
break;
if (step_id == NO_VAL)
step_id = job->step_id.step_id;
if (exclude_hl) {
slurm_step_layout_t *step_layout =
launch_common_get_slurm_step_layout(
job);
hostlist_push(exclude_hl,
step_layout->node_list);
}
step_resp = job->step_ctx->step_resp;
if (step_resp && step_resp->resv_ports &&
strcmp(step_resp->resv_ports, "(null)")) {
if (resv_ports)
xstrcat(resv_ports, ",");
xstrcat(resv_ports, step_resp->resv_ports);
}
node_offset += job->nhosts;
task_offset += job->ntasks;
/*
* If packing nodes (SELECT_PACK_NODES, -mpack), the step
* may have an updated layout. Need to update each
* component's het_job_nnodes with the updated counts.
*/
if (job->nhosts < old_nhosts) {
update_het_nnodes = true;
updated_het_nnodes -= old_nhosts - job->nhosts;
}
if (het_job_ntasks == NO_VAL)
updated_het_ntasks += job->ntasks;
}
if (update_het_nnodes) {
list_iterator_reset(job_iter);
while ((job = list_next(job_iter))) {
job->het_job_nnodes = updated_het_nnodes;
}
}
if (updated_het_ntasks) {
list_iterator_reset(job_iter);
while ((job = list_next(job_iter))) {
job->het_job_ntasks = updated_het_ntasks;
}
}
FREE_NULL_HOSTLIST(exclude_hl);
if (!rc && resv_ports) {
/*
* Merge numeric values into single range
* (e.g. "10-12,13-15,16-18" -> "10-18")
*/
hostset_t *hs;
char *tmp = NULL, *sep;
xstrfmtcat(tmp, "[%s]", resv_ports);
hs = hostset_create(tmp);
hostset_ranged_string(hs, strlen(tmp) + 1, tmp);
sep = strchr(tmp, ']');
if (sep)
sep[0] = '\0';
xfree(resv_ports);
resv_ports = xstrdup(tmp + 1);
xfree(tmp);
hostset_destroy(hs);
list_iterator_reset(job_iter);
while ((job = list_next(job_iter))) {
if (!job->step_ctx->step_resp)
continue;
xfree(job->step_ctx->step_resp->resv_ports);
job->step_ctx->step_resp->resv_ports =
xstrdup(resv_ports);
}
}
xfree(resv_ports);
list_iterator_destroy(job_iter);
if (opt_iter)
list_iterator_destroy(opt_iter);
return rc;
} else if (job) {
if (het_job_id) {
job->het_job_id = het_job_id;
job->het_job_nnodes = job->nhosts;
job->het_job_ntasks = job->ntasks;
job->het_job_task_offset = 0;
}
if ((rc = create_job_step(job, use_all_cpus, &opt)) < 0)
return rc;
if (het_job_id) {
/*
* If packing nodes (SELECT_PACK_NODES, -mpack), the step
* may have an updated layout.
*/
job->het_job_nnodes = job->nhosts;
/* The stepmgr logic can modify ntasks */
job->het_job_ntasks = job->ntasks;
}
return rc;
} else {
return -1;
}
}
static void _cancel_steps(list_t *srun_job_list)
{
srun_job_t *job;
list_itr_t *job_iter;
slurm_msg_t req;
step_complete_msg_t msg;
int rc = 0;
if (!srun_job_list)
return;
slurm_msg_t_init(&req);
req.msg_type = REQUEST_STEP_COMPLETE;
req.data = &msg;
memset(&msg, 0, sizeof(step_complete_msg_t));
msg.step_rc = 0;
job_iter = list_iterator_create(srun_job_list);
while ((job = list_next(job_iter))) {
if (job->step_id.step_id == NO_VAL)
continue;
memcpy(&msg.step_id, &job->step_id, sizeof(msg.step_id));
msg.range_first = 0;
msg.range_last = job->nhosts - 1;
(void) slurm_send_recv_controller_rc_msg(&req, &rc,
working_cluster_rec);
}
list_iterator_destroy(job_iter);
}
static void _het_job_struct_del(void *x)
{
het_job_resp_struct_t *het_job_resp = (het_job_resp_struct_t *) x;
xfree(het_job_resp->cpu_cnt);
if (het_job_resp->host_list)
hostlist_destroy(het_job_resp->host_list);
xfree(het_job_resp);
}
static char *_compress_het_job_nodelist(list_t *used_resp_list)
{
resource_allocation_response_msg_t *resp;
het_job_resp_struct_t *het_job_resp;
list_t *het_job_resp_list;
list_itr_t *resp_iter;
char *tmp;
char *het_job_nodelist = NULL, *node_name;
hostset_t *hs;
int cnt, i, j, k;
uint16_t *cpus;
uint32_t *reps, cpu_inx;
if (!used_resp_list)
return het_job_nodelist;
cnt = list_count(used_resp_list);
het_job_resp_list = list_create(_het_job_struct_del);
hs = hostset_create("");
resp_iter = list_iterator_create(used_resp_list);
while ((resp = list_next(resp_iter))) {
if (!resp->node_list)
continue;
hostset_insert(hs, resp->node_list);
het_job_resp = xmalloc(sizeof(het_job_resp_struct_t));
het_job_resp->node_cnt = resp->node_cnt;
het_job_resp->cpu_cnt =
xmalloc(sizeof(uint16_t) * resp->node_cnt);
het_job_resp->host_list = hostlist_create(resp->node_list);
for (i = 0, k = 0;
(i < resp->num_cpu_groups) && (k < resp->node_cnt); i++) {
for (j = 0; j < resp->cpu_count_reps[i]; j++) {
het_job_resp->cpu_cnt[k++] =
resp->cpus_per_node[i];
if (k >= resp->node_cnt)
break;
}
if (k >= resp->node_cnt)
break;
}
list_append(het_job_resp_list, het_job_resp);
}
list_iterator_destroy(resp_iter);
het_job_nodelist = hostset_ranged_string_xmalloc(hs);
cpu_inx = 0;
cnt = hostset_count(hs);
cpus = xmalloc(sizeof(uint16_t) * (cnt + 1));
reps = xmalloc(sizeof(uint32_t) * (cnt + 1));
for (i = 0; i < cnt; i++) {
node_name = hostset_nth(hs, i);
resp_iter = list_iterator_create(het_job_resp_list);
while ((het_job_resp = list_next(resp_iter))) {
j = hostlist_find(het_job_resp->host_list, node_name);
if ((j == -1) || !het_job_resp->cpu_cnt)
continue; /* node not in this hetjob */
if (cpus[cpu_inx] == het_job_resp->cpu_cnt[j]) {
reps[cpu_inx]++;
} else {
if (cpus[cpu_inx] != 0)
cpu_inx++;
cpus[cpu_inx] = het_job_resp->cpu_cnt[j];
reps[cpu_inx]++;
}
break;
}
list_iterator_destroy(resp_iter);
free(node_name);
}
cpu_inx++;
tmp = uint32_compressed_to_str(cpu_inx, cpus, reps);
if (setenv("SLURM_JOB_CPUS_PER_NODE", tmp, 1) < 0) {
error("%s: Unable to set SLURM_JOB_CPUS_PER_NODE in environment",
__func__);
}
xfree(tmp);
xfree(reps);
xfree(cpus);
hostset_destroy(hs);
FREE_NULL_LIST(het_job_resp_list);
return het_job_nodelist;
}
/*
* Here we have a regular job allocation, but we are requesting a het step in
* that allocation. So here we will copy the resp_list to the number of
* components we care about.
*/
static void _copy_job_resp(list_t *job_resp_list, int count)
{
resource_allocation_response_msg_t *new, *orig;
xassert(job_resp_list);
xassert(list_count(job_resp_list) == 1);
orig = list_peek(job_resp_list);
for (int i = 0; i < count; i++) {
new = slurm_copy_resource_allocation_response_msg(orig);
list_append(job_resp_list, new);
}
}
static void _check_gpus_per_socket(slurm_opt_t *opt_local)
{
static bool checked = false; /* Only log the warning once */
if (!opt_local->gpus_per_socket || checked)
return;
checked = true;
if (opt_local->gpus_per_socket &&
!slurm_option_set_by_env(opt_local, LONG_OPT_GPUS_PER_SOCKET)) {
/*
* gpus_per_socket does not work for steps.
* If it is set by env, it was likely inherited by the job.
*/
warning("Ignoring --gpus-per-socket because it can only be specified at job allocation time, not during step allocation.");
}
}
extern void create_srun_job(void **p_job, bool *got_alloc)
{
resource_allocation_response_msg_t *resp;
list_t *job_resp_list = NULL, *srun_job_list = NULL;
list_t *used_resp_list = NULL;
list_itr_t *opt_iter, *resp_iter;
srun_job_t *job = NULL;
int i, max_list_offset, max_het_job_offset, het_job_offset = -1,
het_step_offset = -1;
uint32_t my_job_id = 0, het_job_id = 0;
char *het_job_nodelist = NULL;
bool begin_error_logged = false;
bool core_spec_error_logged = false;
bool node_cnt_error_logged = false;
bool tres_license_error_logged = false;
bool x11_error_logged = false;
/*
* now global "opt" should be filled in and available,
* create a job from opt
*/
if (sropt.test_only) {
int rc = allocate_test();
if (rc) {
slurm_perror("allocation failure");
exit (1);
}
exit (0);
} else if (sropt.no_alloc) {
if (opt_list ||
(sropt.het_grp_bits && (bit_fls(sropt.het_grp_bits) > 0)))
fatal("--no-allocation option not supported for heterogeneous jobs");
info("do not allocate resources");
job = job_create_noalloc();
if (job == NULL) {
error("Job creation failure.");
exit(error_exit);
}
if (create_job_step(job, false, &opt) < 0)
exit(error_exit);
} else if ((job_resp_list = existing_allocation())) {
slurm_opt_t *opt_local;
max_list_offset = 0;
max_het_job_offset = list_count(job_resp_list) - 1;
if (opt_list) {
opt_iter = list_iterator_create(opt_list);
while ((opt_local = list_next(opt_iter))) {
srun_opt_t *srun_opt = opt_local->srun_opt;
xassert(srun_opt);
if (srun_opt->het_grp_bits) {
i = bit_fls(srun_opt->het_grp_bits);
max_list_offset = MAX(max_list_offset,
i);
}
}
list_iterator_destroy(opt_iter);
if (max_list_offset > max_het_job_offset) {
if (list_count(job_resp_list) != 1) {
error("Attempt to run a job step with het group value of %d, but the job allocation has maximum value of %d",
max_list_offset,
max_het_job_offset);
exit(1);
}
/*
* Here we have a regular job allocation, but we
* are requesting a het step in that
* allocation. So here we will copy the
* resp_list to the number of components we care
* about.
*/
_copy_job_resp(job_resp_list, max_list_offset);
max_het_job_offset = max_list_offset;
local_het_step = true;
}
if (list_count(opt_list) > 1)
het_step_offset = 0;
}
srun_job_list = list_create(NULL);
used_resp_list = list_create(NULL);
if (max_het_job_offset > 0)
het_job_offset = 0;
resp_iter = list_iterator_create(job_resp_list);
while ((resp = list_next(resp_iter))) {
bool merge_nodelist = true;
if (my_job_id == 0) {
my_job_id = resp->job_id;
if (resp->working_cluster_rec)
slurm_setup_remote_working_cluster(resp);
}
_print_job_information(resp);
(void) get_next_opt(-2);
/*
* Check using het_job_offset here, but we use
* het_step_offset for the job being added.
*/
while ((opt_local = get_next_opt(het_job_offset))) {
srun_opt_t *srun_opt = opt_local->srun_opt;
xassert(srun_opt);
if (local_het_step)
opt_local->step_het_comp_cnt =
max_het_job_offset;
if (merge_nodelist) {
merge_nodelist = false;
list_append(used_resp_list, resp);
}
if (slurm_option_set_by_env(opt_local, 'N') &&
(opt_local->min_nodes > resp->node_cnt)) {
/*
* This signifies the job used the
* --no-kill option and a node went DOWN
* or it used a node count range
* specification, was checkpointed from
* one size and restarted at a different
* size
*/
if (!node_cnt_error_logged) {
error("SLURM_JOB_NUM_NODES environment variable conflicts with allocated node count (%u != %u).",
opt_local->min_nodes,
resp->node_cnt);
node_cnt_error_logged = true;
}
/*
* Modify options to match resource
* allocation.
* NOTE: Some options are not supported
*/
opt_local->min_nodes = resp->node_cnt;
xfree(srun_opt->alloc_nodelist);
if (!opt_local->ntasks_set) {
opt_local->ntasks =
opt_local->min_nodes;
}
}
_check_gpus_per_socket(opt_local);
if (!tres_license_error_logged &&
!slurm_option_set_by_env(
opt_local,
LONG_OPT_TRES_PER_TASK) &&
xstrstr(opt_local->tres_per_task,
"license")) {
warning("Ignoring --tres-per-task license specification because licenses can only be specified at job allocation time, not during step allocation.");
tres_license_error_logged = true;
}
if (srun_opt->core_spec_set &&
!core_spec_error_logged) {
/*
* NOTE: Silently ignore specialized
* core count set with SLURM_CORE_SPEC
* environment variable
*/
error("Ignoring --core-spec value for a job step "
"within an existing job. Set specialized cores "
"at job allocation time.");
core_spec_error_logged = true;
}
/*
* Here we send the het job groups to the
* slurmctld to set up the interconnect
* correctly. We only ever need to send it to
* the first component of the step.
*
* FIXME - is this still needed post-Cray?
*/
if (g_het_grp_bits) {
xfree(opt_local->step_het_grps);
opt_local->step_het_grps =
bit_fmt_hexmask(g_het_grp_bits);
}
_set_env_vars(resp, het_step_offset);
if (_validate_relative(resp, opt_local))
exit(error_exit);
if (opt_local->begin && !begin_error_logged) {
error("--begin is ignored because nodes are already allocated.");
begin_error_logged = true;
}
if (opt_local->x11 && !x11_error_logged) {
error("Ignoring --x11 option for a job step within an "
"existing job. Set x11 options at job allocation time.");
x11_error_logged = true;
}
job = job_step_create_allocation(resp,
opt_local);
if (!job)
exit(error_exit);
if (max_het_job_offset > 0)
job->het_job_offset = het_step_offset;
list_append(srun_job_list, job);
het_step_offset++;
} /* While more option structures */
het_job_offset++;
} /* More hetjob components */
list_iterator_destroy(resp_iter);
max_het_job_offset = get_max_het_group();
het_job_offset = list_count(job_resp_list) - 1;
if (max_het_job_offset > het_job_offset) {
error("Requested het-group offset exceeds highest hetjob index (%d > %d)",
max_het_job_offset, het_job_offset);
exit(error_exit);
}
i = list_count(srun_job_list);
if (i == 0) {
error("No directives to start application on any available hetjob components");
exit(error_exit);
}
if (i == 1)
FREE_NULL_LIST(srun_job_list); /* Just use "job" */
if (list_count(job_resp_list) > 1) {
/* only set if actually a hetjob */
if (!local_het_step && my_job_id)
het_job_id = my_job_id;
het_job_nodelist =
_compress_het_job_nodelist(used_resp_list);
}
FREE_NULL_LIST(used_resp_list);
if (_create_job_step(job, false, srun_job_list, het_job_id,
het_job_nodelist) < 0) {
if (*got_alloc)
slurm_complete_job(my_job_id, 1);
else
_cancel_steps(srun_job_list);
exit(error_exit);
}
xfree(het_job_nodelist);
} else {
/* Combined job allocation and job step launch */
if (slurm_option_set_by_cli(&opt, 'J'))
setenvfs("SLURM_JOB_NAME=%s", opt.job_name);
else if (!slurm_option_set_by_env(&opt, 'J') && opt.argc)
setenvfs("SLURM_JOB_NAME=%s", opt.argv[0]);
if (opt_list) {
if (!colon_cnt) {
error("--het-group expected to be used within an HetJob allocation");
exit(error_exit);
}
job_resp_list = allocate_het_job_nodes();
if (!job_resp_list)
exit(error_exit);
srun_job_list = list_create(NULL);
opt_iter = list_iterator_create(opt_list);
resp_iter = list_iterator_create(job_resp_list);
while ((resp = list_next(resp_iter))) {
slurm_opt_t *opt_local;
if (my_job_id == 0) {
my_job_id = resp->job_id;
*got_alloc = true;
}
opt_local = list_next(opt_iter);
if (!opt_local)
break;
_print_job_information(resp);
_set_env_vars(resp, ++het_job_offset);
_set_env_vars2(resp, het_job_offset);
if (_validate_relative(resp, opt_local)) {
slurm_complete_job(my_job_id, 1);
exit(error_exit);
}
job = job_create_allocation(resp, opt_local);
job->het_job_offset = het_job_offset;
list_append(srun_job_list, job);
_set_step_opts(opt_local, resp);
}
list_iterator_destroy(opt_iter);
list_iterator_destroy(resp_iter);
if (!local_het_step) {
/* Continue support for old pack terminology. */
setenvfs("SLURM_PACK_SIZE=%d",
het_job_offset + 1);
setenvfs("SLURM_HET_SIZE=%d",
het_job_offset + 1);
}
} else {
if (sropt.het_grp_bits &&
(bit_fls(sropt.het_grp_bits) != -1)) {
error("--het-group expected to be used within an HetJob allocation");
exit(error_exit);
}
if (!(resp = allocate_nodes(&opt)))
exit(error_exit);
*got_alloc = true;
my_job_id = resp->job_id;
_print_job_information(resp);
_set_env_vars(resp, -1);
if (_validate_relative(resp, &opt)) {
slurm_complete_job(resp->job_id, 1);
exit(error_exit);
}
job = job_create_allocation(resp, &opt);
_set_step_opts(&opt, resp);
}
if (srun_job_list && (list_count(srun_job_list) > 1) &&
opt_list && (list_count(opt_list) > 1) && my_job_id) {
/* only set if actually a hetjob */
if (!local_het_step)
het_job_id = my_job_id;
het_job_nodelist =
_compress_het_job_nodelist(job_resp_list);
}
if (_create_job_step(job, true, srun_job_list, het_job_id,
het_job_nodelist) < 0) {
slurm_complete_job(my_job_id, 1);
exit(error_exit);
}
xfree(het_job_nodelist);
if (opt_list) {
resp_iter = list_iterator_create(job_resp_list);
while ((resp = list_next(resp_iter))) {
slurm_free_resource_allocation_response_msg(
resp);
}
list_iterator_destroy(resp_iter);
} else {
slurm_free_resource_allocation_response_msg(resp);
}
}
/*
* Spawn process to ensure clean-up of job and/or step
* on abnormal termination
*/
shepherd_fd = _shepherd_spawn(job, srun_job_list, *got_alloc);
if (opt_list)
*p_job = (void *) srun_job_list;
else
*p_job = (void *) job;
if (job)
_srun_cli_filter_post_submit(my_job_id, job->step_id.step_id);
}
extern void pre_launch_srun_job(srun_job_t *job, slurm_opt_t *opt_local)
{
if (!signal_thread)
slurm_thread_create(&signal_thread, _srun_signal_mgr, job);
_run_srun_prolog(job);
if (_call_spank_local_user(job, opt_local)) {
error("Failure in local plugin stack");
slurm_step_launch_abort(job->step_ctx);
exit(error_exit);
}
env_array_merge(&job->env, (const char **)environ);
}
extern void fini_srun(srun_job_t *job, bool got_alloc, uint32_t *global_rc)
{
if (got_alloc) {
cleanup_allocation();
/* Tell slurmctld that we were cancelled */
if (job->state >= SRUN_JOB_CANCELLED)
slurm_complete_job(job->step_id.job_id, NO_VAL);
else
slurm_complete_job(job->step_id.job_id, *global_rc);
}
_shepherd_notify(shepherd_fd);
if (signal_thread) {
srun_shutdown = true;
pthread_kill(signal_thread, SIGINT);
slurm_thread_join(signal_thread);
}
_run_srun_epilog(job);
step_ctx_destroy(job->step_ctx);
if (WIFEXITED(*global_rc))
*global_rc = WEXITSTATUS(*global_rc);
else if (WIFSIGNALED(*global_rc))
*global_rc = 128 + WTERMSIG(*global_rc);
mpir_cleanup();
}
void
update_job_state(srun_job_t *job, srun_job_state_t state)
{
slurm_mutex_lock(&job->state_mutex);
if (job->state < state) {
job->state = state;
slurm_cond_signal(&job->state_cond);
}
slurm_mutex_unlock(&job->state_mutex);
return;
}
srun_job_state_t
job_state(srun_job_t *job)
{
srun_job_state_t state;
slurm_mutex_lock(&job->state_mutex);
state = job->state;
slurm_mutex_unlock(&job->state_mutex);
return state;
}
void
job_force_termination(srun_job_t *job)
{
static int kill_sent = 0;
static time_t last_msg = 0;
if (kill_sent == 0) {
info("forcing job termination");
/* Send SIGKILL to tasks directly */
update_job_state(job, SRUN_JOB_CANCELLED);
launch_g_fwd_signal(SIGKILL);
} else {
time_t now = time(NULL);
if (last_msg != now) {
info("job abort in progress");
last_msg = now;
}
if (kill_sent == 1) {
/* Try sending SIGKILL through slurmctld */
slurm_kill_job_step(job->step_id.job_id,
job->step_id.step_id, SIGKILL, 0);
}
}
kill_sent++;
}
static void _set_ntasks(allocation_info_t *ai, slurm_opt_t *opt_local)
{
int cnt = 0;
/* Distinction between explicit or implicit set of ntasks */
if (opt_local->ntasks_opt_set ||
(opt_local->ntasks_set &&
(opt_local->ntasks_per_node == NO_VAL)))
return;
if (opt_local->ntasks_per_node != NO_VAL) {
cnt = ai->nnodes * opt_local->ntasks_per_node;
opt_local->ntasks_set = true; /* implicit */
} else if (opt_local->cpus_set) {
opt_local->ntasks = NO_VAL;
opt_local->ntasks_set = true; /* implicit */
return;
}
opt_local->ntasks = (cnt < ai->nnodes) ? ai->nnodes : cnt;
}
/*
* Create an srun job structure from a resource allocation response msg
*/
static srun_job_t *_job_create_structure(allocation_info_t *ainfo,
slurm_opt_t *opt_local)
{
srun_job_t *job = xmalloc(sizeof(srun_job_t));
int i;
_set_ntasks(ainfo, opt_local);
debug2("creating job with %d tasks", opt_local->ntasks);
slurm_mutex_init(&job->state_mutex);
slurm_cond_init(&job->state_cond, NULL);
job->state = SRUN_JOB_INIT;
job->container = xstrdup(opt_local->container);
job->nodelist = xstrdup(ainfo->nodelist);
job->partition = xstrdup(ainfo->partition);
memcpy(&job->step_id, &ainfo->step_id, sizeof(job->step_id));
job->het_job_id = NO_VAL;
job->het_job_nnodes = NO_VAL;
job->het_job_ntasks = NO_VAL;
job->het_job_offset = NO_VAL;
job->het_job_task_offset = NO_VAL;
job->nhosts = ainfo->nnodes;
if (opt_local->min_nodes > job->nhosts) {
error("Only allocated %d nodes asked for %d",
job->nhosts, opt_local->min_nodes);
if (opt_local->exclude) {
/* When resources are pre-allocated and some nodes
* are explicitly excluded, this error can occur. */
error("Are required nodes explicitly excluded?");
}
xfree(job);
return NULL;
}
if ((ainfo->cpus_per_node == NULL) ||
(ainfo->cpu_count_reps == NULL)) {
error("cpus_per_node array is not set");
xfree(job);
return NULL;
}
job->ntasks = opt_local->ntasks;
job->ntasks_per_board = ainfo->ntasks_per_board;
job->ntasks_per_core = ainfo->ntasks_per_core;
job->ntasks_per_socket = ainfo->ntasks_per_socket;
/*
* If cpus_per_task is set then get the exact count of cpus for the
* requested step (we might very well use less, especially if
* --exclusive is used). Else get the total for the allocation given.
*/
if (opt_local->cpus_set) {
if (opt_local->ntasks == NO_VAL)
job->cpu_count = NO_VAL;
else
job->cpu_count = opt_local->ntasks *
opt_local->cpus_per_task;
} else {
for (i = 0; i < ainfo->num_cpu_groups; i++) {
job->cpu_count += ainfo->cpus_per_node[i] *
ainfo->cpu_count_reps[i];
}
}
job->rc = -1;
job_update_io_fnames(job, opt_local);
job->uid = ainfo->uid;
job->user_name = xstrdup(ainfo->user_name);
job->gid = ainfo->gid;
job->group_name = xstrdup(ainfo->group_name);
return (job);
}
extern void job_update_io_fnames(srun_job_t *job, slurm_opt_t *opt_local)
{
job->ifname = fname_create(job, opt_local->ifname, opt_local->ntasks);
job->ofname = fname_create(job, opt_local->ofname, opt_local->ntasks);
job->efname = opt_local->efname ?
fname_create(job, opt_local->efname, opt_local->ntasks) :
job->ofname;
}
static char *
_normalize_hostlist(const char *hostlist)
{
char *buf = NULL;
hostlist_t *hl = hostlist_create(hostlist);
if (hl) {
buf = hostlist_ranged_string_xmalloc(hl);
hostlist_destroy(hl);
}
if (!buf)
return xstrdup(hostlist);
return buf;
}
static int _call_spank_local_user(srun_job_t *job, slurm_opt_t *opt_local)
{
struct spank_launcher_job_info info[1];
info->argc = opt_local->argc;
info->argv = opt_local->argv;
info->gid = opt_local->gid;
info->jobid = job->step_id.job_id;
info->stepid = job->step_id.step_id;
info->step_layout = launch_common_get_slurm_step_layout(job);
info->uid = opt_local->uid;
return spank_local_user(info);
}
/* Return the number of microseconds between tv1 and tv2 with a maximum
* a maximum value of 10,000,000 to prevent overflows */
static long _diff_tv_str(struct timeval *tv1, struct timeval *tv2)
{
long delta_t;
delta_t = MIN((tv2->tv_sec - tv1->tv_sec), 10);
delta_t *= USEC_IN_SEC;
delta_t += tv2->tv_usec - tv1->tv_usec;
return delta_t;
}
static void _handle_intr(srun_job_t *job)
{
static struct timeval last_intr = { 0, 0 };
struct timeval now;
gettimeofday(&now, NULL);
if (sropt.quit_on_intr || _diff_tv_str(&last_intr, &now) < 1000000) {
info("sending Ctrl-C to %ps", &job->step_id);
launch_g_fwd_signal(SIGINT);
job_force_termination(job);
} else {
if (sropt.disable_status) {
info("sending Ctrl-C to %ps", &job->step_id);
launch_g_fwd_signal(SIGINT);
} else if (job->state < SRUN_JOB_CANCELLED) {
info("interrupt (one more within 1 sec to abort)");
launch_g_print_status();
}
last_intr = now;
}
}
static void _handle_pipe(void)
{
static int ending = 0;
if (ending)
return;
ending = 1;
launch_g_fwd_signal(SIGKILL);
}
static void _print_job_information(resource_allocation_response_msg_t *resp)
{
int i;
char *str = NULL;
char *sep = "";
if (!opt.verbose)
return;
xstrfmtcat(str, "jobid %u: nodes(%u):`%s', cpu counts: ",
resp->job_id, resp->node_cnt, resp->node_list);
for (i = 0; i < resp->num_cpu_groups; i++) {
xstrfmtcat(str, "%s%u(x%u)",
sep, resp->cpus_per_node[i],
resp->cpu_count_reps[i]);
sep = ",";
}
verbose("%s", str);
xfree(str);
}
/* NOTE: Executed once for entire hetjob */
static void _run_srun_epilog (srun_job_t *job)
{
int rc;
if (sropt.epilog && xstrcasecmp(sropt.epilog, "none") != 0) {
if (setenvf(NULL, "SLURM_SCRIPT_CONTEXT", "epilog_srun") < 0)
error("unable to set SLURM_SCRIPT_CONTEXT in environment");
rc = _run_srun_script(job, sropt.epilog);
if (rc) {
error("srun epilog failed status=%d", rc);
}
}
}
static void _run_srun_prolog (srun_job_t *job)
{
int rc;
if (sropt.prolog && xstrcasecmp(sropt.prolog, "none") != 0) {
if (setenvf(NULL, "SLURM_SCRIPT_CONTEXT", "prolog_srun") < 0)
error("unable to set SLURM_SCRIPT_CONTEXT in environment");
rc = _run_srun_script(job, sropt.prolog);
if (rc) {
error("srun prolog failed rc = %d. Aborting step.", rc);
slurm_step_launch_abort(job->step_ctx);
}
}
}
/*
* Run srun prolog/epilog script.
*
* RET the exit status of the script or 1 on generic error and 0 on success
*/
static int _run_srun_script (srun_job_t *job, char *script)
{
int status;
pid_t cpid;
int i;
char **args = NULL;
if (script == NULL || script[0] == '\0')
return 0;
if (access(script, R_OK | X_OK) < 0) {
info("Access denied for %s: %m", script);
return 0;
}
if ((cpid = fork()) < 0) {
error ("run_srun_script: fork: %m");
return 1;
}
if (cpid == 0) {
/*
* set the prolog/epilog scripts command line arguments to the
* application arguments (for last hetjob component), but
* shifted one higher
*/
args = xmalloc(sizeof(char *) * 1024);
args[0] = script;
for (i = 0; i < opt.argc; i++) {
args[i + 1] = opt.argv[i];
}
args[i + 1] = NULL;
execv(script, args);
error("Failed to execute srun prolog/epilog script: %m");
_exit(127);
}
do {
if (waitpid(cpid, &status, 0) < 0) {
if (errno == EINTR)
continue;
error("waitpid: %m");
return 0;
} else if (WIFEXITED(status)) {
return WEXITSTATUS(status);
} else {
error("script did not exit normally");
return 1;
}
} while(1);
/* NOTREACHED */
}
static char *_build_key(char *base, int het_job_offset)
{
char *key = NULL;
/* If we are a local_het_step we treat it like a normal step */
if (local_het_step || (het_job_offset == -1))
key = xstrdup(base);
else
xstrfmtcat(key, "%s_PACK_GROUP_%d", base, het_job_offset);
return key;
}
static void _set_env_vars(resource_allocation_response_msg_t *resp,
int het_job_offset)
{
char *key, *value, *tmp;
int i;
key = _build_key("SLURM_JOB_CPUS_PER_NODE", het_job_offset);
if (!getenv(key)) {
tmp = uint32_compressed_to_str(resp->num_cpu_groups,
resp->cpus_per_node,
resp->cpu_count_reps);
if (setenvf(NULL, key, "%s", tmp) < 0)
error("unable to set %s in environment", key);
xfree(tmp);
}
xfree(key);
if (resp->env_size) { /* Used to set Burst Buffer environment */
for (i = 0; i < resp->env_size; i++) {
tmp = xstrdup(resp->environment[i]);
key = tmp;
value = strchr(tmp, '=');
if (value) {
value[0] = '\0';
value++;
setenv(key, value, 0);
}
xfree(tmp);
}
}
if (resp->pn_min_memory & MEM_PER_CPU) {
uint64_t tmp_mem = resp->pn_min_memory & (~MEM_PER_CPU);
key = _build_key("SLURM_MEM_PER_CPU", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%"PRIu64, tmp_mem) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
} else if (resp->pn_min_memory) {
uint64_t tmp_mem = resp->pn_min_memory;
key = _build_key("SLURM_MEM_PER_NODE", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%"PRIu64, tmp_mem) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
}
if (resp->segment_size) {
key = _build_key("SLURM_JOB_SEGMENT_SIZE", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%u", resp->segment_size) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
}
return;
}
/*
* Set some hetjob environment variables for combined job & step allocation
*/
static void _set_env_vars2(resource_allocation_response_msg_t *resp,
int het_job_offset)
{
char *key;
if (resp->account) {
key = _build_key("SLURM_JOB_ACCOUNT", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%s", resp->account) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
}
key = _build_key("SLURM_JOB_ID", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%u", resp->job_id) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
key = _build_key("SLURM_JOB_NODELIST", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%s", resp->node_list) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
key = _build_key("SLURM_JOB_PARTITION", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%s", resp->partition) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
if (resp->qos) {
key = _build_key("SLURM_JOB_QOS", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%s", resp->qos) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
}
if (resp->resv_name) {
key = _build_key("SLURM_JOB_RESERVATION", het_job_offset);
if (!getenv(key) &&
(setenvf(NULL, key, "%s", resp->resv_name) < 0)) {
error("unable to set %s in environment", key);
}
xfree(key);
}
}
/* Set SLURM_RLIMIT_* environment variables with current resource
* limit values, reset RLIMIT_NOFILE to maximum possible value */
static int _set_rlimit_env(void)
{
int rc = SLURM_SUCCESS;
struct rlimit rlim[1];
unsigned long cur;
char name[64], *format;
slurm_rlimits_info_t *rli;
/* Modify limits with any command-line options */
if (sropt.propagate
&& parse_rlimits(sropt.propagate, PROPAGATE_RLIMITS)) {
error( "--propagate=%s is not valid.", sropt.propagate );
exit(error_exit);
}
for (rli = get_slurm_rlimits_info(); rli->name != NULL; rli++ ) {
if (rli->propagate_flag != PROPAGATE_RLIMITS)
continue;
if (getrlimit (rli->resource, rlim) < 0) {
error ("getrlimit (RLIMIT_%s): %m", rli->name);
rc = SLURM_ERROR;
continue;
}
cur = (unsigned long) rlim->rlim_cur;
snprintf(name, sizeof(name), "SLURM_RLIMIT_%s", rli->name);
if (sropt.propagate && (rli->propagate_flag == PROPAGATE_RLIMITS))
/*
* Prepend 'U' to indicate user requested propagate
*/
format = "U%lu";
else
format = "%lu";
if (setenvf (NULL, name, format, cur) < 0) {
error ("unable to set %s in environment", name);
rc = SLURM_ERROR;
continue;
}
debug ("propagating RLIMIT_%s=%lu", rli->name, cur);
}
/*
* Now increase NOFILE to the max available for this srun
*/
rlimits_use_max_nofile();
return rc;
}
/* Set SLURM_CLUSTER_NAME< SLURM_SUBMIT_DIR and SLURM_SUBMIT_HOST environment
* variables within current state */
static void _set_submit_dir_env(void)
{
char buf[PATH_MAX], host[256];
/* Only set these environment variables in new allocations */
if (sropt.jobid != NO_VAL)
return;
if (setenvf(NULL, "SLURM_CLUSTER_NAME", "%s",
slurm_conf.cluster_name) < 0)
error("unable to set SLURM_CLUSTER_NAME in environment");
if ((getcwd(buf, PATH_MAX)) == NULL)
error("getcwd failed: %m");
else if (setenvf(NULL, "SLURM_SUBMIT_DIR", "%s", buf) < 0)
error("unable to set SLURM_SUBMIT_DIR in environment");
if ((gethostname(host, sizeof(host))))
error("gethostname_short failed: %m");
else if (setenvf(NULL, "SLURM_SUBMIT_HOST", "%s", host) < 0)
error("unable to set SLURM_SUBMIT_HOST in environment");
}
/* Set some environment variables with current state */
static int _set_umask_env(void)
{
if (!getenv("SRUN_DEBUG")) { /* do not change current value */
/* NOTE: Default debug level is 3 (info) */
int log_level = LOG_LEVEL_INFO + opt.verbose - opt.quiet;
if (setenvf(NULL, "SRUN_DEBUG", "%d", log_level) < 0)
error ("unable to set SRUN_DEBUG in environment");
}
if (!getenv("SLURM_UMASK")) { /* do not change current value */
char mask_char[5];
mode_t mask;
mask = (int)umask(0);
umask(mask);
sprintf(mask_char, "0%d%d%d",
((mask>>6)&07), ((mask>>3)&07), mask&07);
if (setenvf(NULL, "SLURM_UMASK", "%s", mask_char) < 0) {
error ("unable to set SLURM_UMASK in environment");
return SLURM_ERROR;
}
debug ("propagating UMASK=%s", mask_char);
}
return SLURM_SUCCESS;
}
static void _shepherd_notify(int shepherd_fd)
{
int rc;
while (1) {
rc = write(shepherd_fd, "", 1);
if (rc == -1) {
if ((errno == EAGAIN) || (errno == EINTR))
continue;
error("write(shepherd): %m");
}
break;
}
close(shepherd_fd);
}
static int _shepherd_spawn(srun_job_t *job, list_t *srun_job_list,
bool got_alloc)
{
int shepherd_pipe[2], rc;
pid_t shepherd_pid;
char buf[1];
if (pipe(shepherd_pipe)) {
error("pipe: %m");
return -1;
}
shepherd_pid = fork();
if (shepherd_pid == -1) {
error("fork: %m");
return -1;
}
if (shepherd_pid != 0) {
close(shepherd_pipe[0]);
return shepherd_pipe[1];
}
/* Wait for parent to notify of completion or I/O error on abort */
close(shepherd_pipe[1]);
while (1) {
rc = read(shepherd_pipe[0], buf, 1);
if (rc == 1) {
_exit(0);
} else if (rc == 0) {
break; /* EOF */
} else if (rc == -1) {
if ((errno == EAGAIN) || (errno == EINTR))
continue;
break;
}
}
if (srun_job_list) {
list_itr_t *job_iter;
job_iter = list_iterator_create(srun_job_list);
while ((job = list_next(job_iter))) {
(void) slurm_kill_job_step(job->step_id.job_id, job->step_id.step_id,
SIGKILL, 0);
if (got_alloc)
slurm_complete_job(job->step_id.job_id, NO_VAL);
}
list_iterator_destroy(job_iter);
} else {
(void) slurm_kill_job_step(job->step_id.job_id,
job->step_id.step_id, SIGKILL, 0);
if (got_alloc)
slurm_complete_job(job->step_id.job_id, NO_VAL);
}
_exit(0);
return -1;
}
/* _srun_signal_mgr - Process daemon-wide signals */
static void *_srun_signal_mgr(void *job_ptr)
{
int sig;
int i, rc;
sigset_t set;
srun_job_t *job = (srun_job_t *)job_ptr;
/* Make sure no required signals are ignored (possibly inherited) */
for (i = 0; sig_array[i]; i++)
xsignal_default(sig_array[i]);
while (!srun_shutdown) {
xsignal_sigset_create(sig_array, &set);
rc = sigwait(&set, &sig);
if (rc == EINTR)
continue;
switch (sig) {
case SIGINT:
if (!srun_shutdown)
_handle_intr(job);
break;
case SIGQUIT:
info("Quit");
/* continue with slurm_step_launch_abort */
case SIGTERM:
case SIGHUP:
/* No need to call job_force_termination here since we
* are ending the job now and we don't need to update
* the state. */
info("forcing job termination");
launch_g_fwd_signal(SIGKILL);
break;
case SIGCONT:
info("got SIGCONT");
break;
case SIGPIPE:
_handle_pipe();
break;
case SIGALRM:
if (srun_max_timer) {
info("First task exited %ds ago", sropt.max_wait);
launch_g_print_status();
launch_g_step_terminate();
}
break;
default:
launch_g_fwd_signal(sig);
break;
}
}
return NULL;
}
static int _validate_relative(resource_allocation_response_msg_t *resp,
slurm_opt_t *opt_local)
{
srun_opt_t *srun_opt = opt_local->srun_opt;
xassert(srun_opt);
if ((srun_opt->relative != NO_VAL) &&
((srun_opt->relative + opt_local->min_nodes)
> resp->node_cnt)) {
if (slurm_option_set_by_cli(opt_local, 'N')) {
/* -N command line option used */
error("--relative and --nodes option incompatible "
"with count of allocated nodes (%d+%d>%d)",
srun_opt->relative,
opt_local->min_nodes,
resp->node_cnt);
} else { /* SLURM_JOB_NUM_NODES option used */
error("--relative and SLURM_JOB_NUM_NODES option incompatible with count of allocated nodes (%d+%d>%d)",
srun_opt->relative,
opt_local->min_nodes,
resp->node_cnt);
}
return SLURM_ERROR;
}
return SLURM_SUCCESS;
}
static void _call_spank_fini(void)
{
if (-1 != shepherd_fd)
spank_fini(NULL);
}
/*
* Run cli_filter_post_submit on all opt structures
* Convenience function since this might need to run in two spots
*/
static void _srun_cli_filter_post_submit(uint32_t jobid, uint32_t stepid)
{
static bool post_submit_ran = false;
int idx = 0, components = 1;
if (post_submit_ran)
return;
if (opt_list)
components = list_count(opt_list);
for (idx = 0; idx < components; idx++)
cli_filter_g_post_submit(idx, jobid, stepid);
post_submit_ran = true;
}