Google Git
Sign in
third-party-mirror / SchedMD / slurm / edf4092834f76748cd7c303ee1d9808758f46cf6 / . / src / stepmgr / stepmgr.c
blob: 280e8fef0b18d039baefab24092ca55930144d34 [file] [log] [blame]
/*****************************************************************************\
* stepmgr.c - manage the job step information of slurm
*****************************************************************************
* Copyright (C) 2002-2007 The Regents of the University of California.
* Copyright (C) 2008-2010 Lawrence Livermore National Security.
* Copyright (C) SchedMD LLC.
* Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
* Written by Morris Jette <jette1@llnl.gov>, et. al.
* 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 <ctype.h>
#include <errno.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/param.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include "stepmgr.h"
#include "slurm/slurm_errno.h"
#include "src/common/assoc_mgr.h"
#include "src/common/bitstring.h"
#include "src/common/forward.h"
#include "src/common/node_features.h"
#include "src/common/port_mgr.h"
#include "src/common/slurm_protocol_pack.h"
#include "src/common/slurm_protocol_socket.h"
#include "src/common/slurm_resource_info.h"
#include "src/common/stepd_proxy.h"
#include "src/common/tres_bind.h"
#include "src/common/tres_frequency.h"
#include "src/common/xstring.h"
#include "src/interfaces/accounting_storage.h"
#include "src/interfaces/gres.h"
#include "src/interfaces/jobacct_gather.h"
#include "src/interfaces/mcs.h"
#include "src/interfaces/select.h"
#include "src/interfaces/switch.h"
#include "src/stepmgr/gres_stepmgr.h"
#include "src/stepmgr/srun_comm.h"
typedef struct {
uint16_t flags;
bool found;
int rc_in;
uint16_t signal;
slurm_step_id_t step_id;
uid_t uid;
} step_signal_t;
typedef struct {
bitstr_t *all_gres_core_bitmap;
bitstr_t *any_gres_core_bitmap;
int core_end_bit;
int core_start_bit;
int job_node_inx;
list_t *node_gres_list;
} foreach_gres_filter_t;
static void _build_pending_step(job_record_t *job_ptr,
job_step_create_request_msg_t *step_specs);
static int _step_partial_comp(step_record_t *step_ptr,
step_complete_msg_t *req, bool finish,
int *rem, uint32_t *max_rc);
static int _count_cpus(job_record_t *job_ptr, bitstr_t *bitmap,
uint32_t *usable_cpu_cnt);
static void _dump_step_layout(step_record_t *step_ptr);
static bool _is_mem_resv(void);
static int _opt_cpu_cnt(uint32_t step_min_cpus, bitstr_t *node_bitmap,
uint32_t *usable_cpu_cnt);
static int _opt_node_cnt(uint32_t step_min_nodes, uint32_t step_max_nodes,
int nodes_avail, int nodes_picked_cnt);
static bitstr_t *_pick_step_nodes(job_record_t *job_ptr,
job_step_create_request_msg_t *step_spec,
list_t *step_gres_list, int cpus_per_task,
uint32_t node_count, int *return_code);
static bitstr_t *_pick_step_nodes_cpus(job_record_t *job_ptr,
bitstr_t *nodes_bitmap, int node_cnt,
int cpu_cnt, uint32_t *usable_cpu_cnt);
static void _step_dealloc_lps(step_record_t *step_ptr);
static step_record_t *_build_interactive_step(
job_record_t *job_ptr_in,
job_step_create_request_msg_t *step_specs,
uint16_t protocol_version);
static int _build_ext_launcher_step(step_record_t **new_step_record,
job_record_t *job_ptr,
job_step_create_request_msg_t *step_specs,
uint16_t protocol_version);
static void _wake_pending_steps(job_record_t *job_ptr);
stepmgr_ops_t *stepmgr_ops = NULL;
extern void stepmgr_init(stepmgr_ops_t *ops)
{
/*
* Just keep pointers so that the pointers can be assigned after being
* initialized init is called.
*/
stepmgr_ops = ops;
}
/* Determine how many more CPUs are required for a job step */
static int _opt_cpu_cnt(uint32_t step_min_cpus, bitstr_t *node_bitmap,
uint32_t *usable_cpu_cnt)
{
int rem_cpus = step_min_cpus;
if (!node_bitmap)
return rem_cpus;
xassert(usable_cpu_cnt);
for (int i = 0; next_node_bitmap(node_bitmap, &i); i++) {
if (usable_cpu_cnt[i] >= rem_cpus)
return 0;
rem_cpus -= usable_cpu_cnt[i];
}
return rem_cpus;
}
/* Select the optimal node count for a job step based upon it's min and
* max target, available resources, and nodes already picked */
static int _opt_node_cnt(uint32_t step_min_nodes, uint32_t step_max_nodes,
int nodes_avail, int nodes_picked_cnt)
{
int target_node_cnt;
if ((step_max_nodes > step_min_nodes) && (step_max_nodes != NO_VAL))
target_node_cnt = step_max_nodes;
else
target_node_cnt = step_min_nodes;
if (target_node_cnt > nodes_picked_cnt)
target_node_cnt -= nodes_picked_cnt;
else
target_node_cnt = 0;
if (nodes_avail < target_node_cnt)
target_node_cnt = nodes_avail;
return target_node_cnt;
}
/* Purge any duplicate job steps for this PID */
static int _purge_duplicate_steps(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
job_step_create_request_msg_t *step_specs =
(job_step_create_request_msg_t *) arg;
if ((step_ptr->step_id.step_id == SLURM_PENDING_STEP) &&
(step_ptr->state == JOB_PENDING) &&
(step_ptr->srun_pid == step_specs->srun_pid) &&
(!xstrcmp(step_ptr->host, step_specs->host))) {
return 1;
}
/*
* See if we have the same step id. If we do check to see if we
* have the same step_het_comp or if the step's is NO_VAL,
* meaning this step is not a het step.
*/
if ((step_specs->step_id.step_id == step_ptr->step_id.step_id) &&
((step_specs->step_id.step_het_comp ==
step_ptr->step_id.step_het_comp) ||
(step_ptr->step_id.step_het_comp == NO_VAL)))
return -1;
return 0;
}
/* The step with a state of PENDING is used as a placeholder for a host and
* port that can be used to wake a pending srun as soon another step ends */
static void _build_pending_step(job_record_t *job_ptr,
job_step_create_request_msg_t *step_specs)
{
step_record_t *step_ptr;
if ((step_specs->host == NULL) || (step_specs->port == 0))
return;
step_ptr = create_step_record(job_ptr, 0);
if (step_ptr == NULL)
return;
*stepmgr_ops->last_job_update = time(NULL);
step_ptr->cpu_count = step_specs->num_tasks;
step_ptr->port = step_specs->port;
step_ptr->srun_pid = step_specs->srun_pid;
step_ptr->host = xstrdup(step_specs->host);
step_ptr->state = JOB_PENDING;
step_ptr->step_id.job_id = job_ptr->job_id;
step_ptr->step_id.step_id = SLURM_PENDING_STEP;
step_ptr->step_id.step_het_comp = NO_VAL;
step_ptr->cwd = xstrdup(step_specs->cwd);
step_ptr->std_err = xstrdup(step_specs->std_err);
step_ptr->std_in = xstrdup(step_specs->std_in);
step_ptr->std_out = xstrdup(step_specs->std_out);
step_ptr->submit_line = xstrdup(step_specs->submit_line);
if (job_ptr->node_bitmap)
step_ptr->step_node_bitmap = bit_copy(job_ptr->node_bitmap);
step_ptr->time_last_active = time(NULL);
}
static void _internal_step_complete(step_record_t *step_ptr, int remaining)
{
struct jobacctinfo *jobacct = (struct jobacctinfo *)step_ptr->jobacct;
job_record_t *job_ptr = step_ptr->job_ptr;
bool add_energy = true;
if ((slurm_conf.prolog_flags & PROLOG_FLAG_CONTAIN) &&
(step_ptr->step_id.step_id != SLURM_EXTERN_CONT))
add_energy = false;
if (add_energy && jobacct && job_ptr->tres_alloc_cnt &&
(jobacct->energy.consumed_energy != NO_VAL64)) {
if (job_ptr->tres_alloc_cnt[TRES_ARRAY_ENERGY] == NO_VAL64)
job_ptr->tres_alloc_cnt[TRES_ARRAY_ENERGY] = 0;
job_ptr->tres_alloc_cnt[TRES_ARRAY_ENERGY] +=
jobacct->energy.consumed_energy;
}
if (IS_JOB_FINISHED(job_ptr) &&
job_ptr->tres_alloc_cnt &&
(job_ptr->tres_alloc_cnt[TRES_ENERGY] != NO_VAL64) &&
(remaining == 1)) {
assoc_mgr_set_job_tres_alloc_str(job_ptr, false);
/* This flag says we have processed the tres alloc including
* energy from all steps, so don't process or handle it again
* with the job. It also tells the slurmdbd plugin to send it
* to the DBD.
*/
job_ptr->bit_flags |= TRES_STR_CALC;
}
jobacct_storage_g_step_complete(stepmgr_ops->acct_db_conn, step_ptr);
if (step_ptr->step_id.step_id == SLURM_PENDING_STEP)
return;
/*
* Derived exit code is the highest exit code of srun steps, so we
* exclude the batch and extern steps.
*
* Sync with _get_derived_ec_update_str() for setting derived_ec on the
* dbd side.
*/
if ((step_ptr->step_id.step_id != SLURM_EXTERN_CONT) &&
(step_ptr->step_id.step_id != SLURM_BATCH_SCRIPT) &&
((step_ptr->exit_code == SIG_OOM) ||
(step_ptr->exit_code > job_ptr->derived_ec)))
job_ptr->derived_ec = step_ptr->exit_code;
step_ptr->state |= JOB_COMPLETING;
_step_dealloc_lps(step_ptr);
/* Don't need to set state. Will be destroyed in next steps. */
/* step_ptr->state = JOB_COMPLETE; */
}
static int _step_signal(void *object, void *arg)
{
step_record_t *step_ptr = (step_record_t *)object;
step_signal_t *step_signal = (step_signal_t *)arg;
uint16_t signal;
int rc;
if (!(step_signal->flags & KILL_FULL_JOB) &&
!find_step_id(step_ptr, &step_signal->step_id))
return SLURM_SUCCESS;
step_signal->found = true;
signal = step_signal->signal;
/*
* If step_het_comp is NO_VAL means it is a non-het step, so return
* SLURM_ERROR to break out of the list_for_each.
*/
rc = (step_ptr->step_id.step_het_comp == NO_VAL) ?
SLURM_ERROR : SLURM_SUCCESS;
if (step_signal->flags & KILL_OOM)
step_ptr->exit_code = SIG_OOM;
if (step_signal->flags & KILL_NO_SIG_FAIL) {
debug("%s: setting SSF_NO_SIG_FAIL for %pS",
__func__, step_ptr);
step_ptr->flags |= SSF_NO_SIG_FAIL;
}
/*
* If SIG_NODE_FAIL codes through it means we had nodes failed
* so handle that in the select plugin and switch the signal
* to KILL afterwards.
*/
if (signal == SIG_NODE_FAIL) {
if (step_signal->rc_in != SLURM_SUCCESS)
return rc;
signal = SIGKILL;
}
/* save user ID of the one who requested the job be cancelled */
if (signal == SIGKILL) {
step_ptr->requid = step_signal->uid;
srun_step_complete(step_ptr);
}
signal_step_tasks(step_ptr, signal, REQUEST_SIGNAL_TASKS);
return rc;
}
static int _step_not_cleaning(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
int *remaining = (int *) arg;
if (step_ptr->step_id.step_id == SLURM_PENDING_STEP)
srun_step_signal(step_ptr, 0);
_internal_step_complete(step_ptr, *remaining);
(*remaining)--;
return 1;
}
/*
* _finish_step_comp - Finish deallocating and delete a non-pending step.
*/
static int _finish_step_comp(void *x, void *arg)
{
int remaining;
step_record_t *step_ptr = x;
job_record_t *job_ptr = step_ptr->job_ptr;
if (step_ptr->step_id.step_id == SLURM_PENDING_STEP)
return 0;
remaining = list_count(job_ptr->step_list);
_internal_step_complete(step_ptr, remaining);
delete_step_record(job_ptr, step_ptr);
_wake_pending_steps(job_ptr);
*stepmgr_ops->last_job_update = time(NULL);
return 1;
}
/*
* delete_step_records - Delete step record for specified job_ptr.
* This function is called when a step fails to run to completion. For example,
* when the job is killed due to reaching its time limit or allocated nodes
* go DOWN.
* IN job_ptr - pointer to job table entry to have step records removed
*/
extern void delete_step_records(job_record_t *job_ptr)
{
int remaining;
xassert(job_ptr);
remaining = list_count(job_ptr->step_list);
*stepmgr_ops->last_job_update = time(NULL);
list_delete_all(job_ptr->step_list, _step_not_cleaning, &remaining);
}
/*
* delete_step_record - delete record for job step for specified job_ptr
* and step_id
* IN job_ptr - pointer to job table entry to have step record removed
* IN step_ptr - pointer to step table entry of the desired job step
*/
void delete_step_record(job_record_t *job_ptr, step_record_t *step_ptr)
{
xassert(job_ptr);
xassert(job_ptr->step_list);
xassert(step_ptr);
*stepmgr_ops->last_job_update = time(NULL);
list_delete_ptr(job_ptr->step_list, step_ptr);
}
/*
* dump_step_desc - dump the incoming step initiate request message
* IN step_spec - job step request specification from RPC
*/
void
dump_step_desc(job_step_create_request_msg_t *step_spec)
{
uint64_t mem_value = step_spec->pn_min_memory;
char *mem_type = "node";
if (mem_value & MEM_PER_CPU) {
mem_value &= (~MEM_PER_CPU);
mem_type = "cpu";
}
log_flag(CPU_FREQ, "StepDesc: user_id=%u JobId=%u cpu_freq_gov=%u cpu_freq_max=%u cpu_freq_min=%u",
step_spec->user_id, step_spec->step_id.job_id,
step_spec->cpu_freq_gov,
step_spec->cpu_freq_max, step_spec->cpu_freq_min);
debug3("StepDesc: user_id=%u %ps node_count=%u-%u cpu_count=%u num_tasks=%u",
step_spec->user_id, &step_spec->step_id,
step_spec->min_nodes, step_spec->max_nodes,
step_spec->cpu_count, step_spec->num_tasks);
debug3(" cpu_freq_gov=%u cpu_freq_max=%u cpu_freq_min=%u "
"relative=%u task_dist=0x%X plane=%u",
step_spec->cpu_freq_gov, step_spec->cpu_freq_max,
step_spec->cpu_freq_min, step_spec->relative,
step_spec->task_dist, step_spec->plane_size);
debug3(" node_list=%s constraints=%s",
step_spec->node_list, step_spec->features);
debug3(" host=%s port=%u srun_pid=%u name=%s network=%s exclusive=%s",
step_spec->host, step_spec->port, step_spec->srun_pid,
step_spec->name, step_spec->network,
(step_spec->flags & SSF_EXCLUSIVE) ? "yes" : "no");
debug3(" mem_per_%s=%"PRIu64" resv_port_cnt=%u immediate=%u no_kill=%s",
mem_type, mem_value, step_spec->resv_port_cnt,
step_spec->immediate,
(step_spec->flags & SSF_NO_KILL) ? "yes" : "no");
debug3(" overcommit=%s time_limit=%u",
(step_spec->flags & SSF_OVERCOMMIT) ? "yes" : "no",
step_spec->time_limit);
if (step_spec->cpus_per_tres)
debug3(" CPUs_per_TRES=%s", step_spec->cpus_per_tres);
if (step_spec->mem_per_tres)
debug3(" Mem_per_TRES=%s", step_spec->mem_per_tres);
if (step_spec->tres_bind)
debug3(" TRES_bind=%s", step_spec->tres_bind);
if (step_spec->tres_freq)
debug3(" TRES_freq=%s", step_spec->tres_freq);
if (step_spec->tres_per_step)
debug3(" TRES_per_step=%s", step_spec->tres_per_step);
if (step_spec->tres_per_node)
debug3(" TRES_per_node=%s", step_spec->tres_per_node);
if (step_spec->tres_per_socket)
debug3(" TRES_per_socket=%s", step_spec->tres_per_socket);
if (step_spec->tres_per_task)
debug3(" TRES_per_task=%s", step_spec->tres_per_task);
if (step_spec->container || step_spec->container_id)
debug3(" Container=%s ContainerID=%s",
step_spec->container, step_spec->container_id);
}
/*
* job_step_signal - signal the specified job step
* IN step_id - filled in slurm_step_id_t
* IN signal - user id of user issuing the RPC
* IN flags - RPC flags
* IN uid - user id of user issuing the RPC
* RET 0 on success, otherwise ESLURM error code
* global: job_list - pointer global job list
* last_job_update - time of last job table update
*/
extern int job_step_signal(slurm_step_id_t *step_id,
uint16_t signal, uint16_t flags, uid_t uid)
{
job_record_t *job_ptr;
step_signal_t step_signal = {
.flags = flags,
.found = false,
.rc_in = SLURM_SUCCESS,
.signal = signal,
.uid = uid,
};
memcpy(&step_signal.step_id, step_id, sizeof(step_signal.step_id));
job_ptr = stepmgr_ops->find_job_record(step_id->job_id);
if (job_ptr == NULL) {
error("job_step_signal: invalid JobId=%u", step_id->job_id);
return ESLURM_INVALID_JOB_ID;
}
if (IS_JOB_FINISHED(job_ptr)) {
step_signal.rc_in = ESLURM_ALREADY_DONE;
if (signal != SIG_NODE_FAIL)
return step_signal.rc_in;
} else if (!IS_JOB_RUNNING(job_ptr)) {
verbose("%s: %pJ is in state %s, cannot signal steps",
__func__, job_ptr,
job_state_string(job_ptr->job_state));
if (signal != SIG_NODE_FAIL)
return ESLURM_TRANSITION_STATE_NO_UPDATE;
}
list_for_each(job_ptr->step_list, _step_signal, &step_signal);
if (!step_signal.found && running_in_slurmctld() &&
(job_ptr->bit_flags & STEPMGR_ENABLED)) {
agent_arg_t *agent_args = NULL;
job_step_kill_msg_t *kill_msg = NULL;
node_record_t *node_ptr;
kill_msg = xmalloc(sizeof(*kill_msg));
kill_msg->signal = signal;
kill_msg->flags = flags;
kill_msg->step_id = *step_id;
agent_args = xmalloc(sizeof(agent_arg_t));
agent_args->msg_type = REQUEST_CANCEL_JOB_STEP;
agent_args->retry = 1;
agent_args->hostlist = hostlist_create(job_ptr->batch_host);
agent_args->node_count = 1;
if ((node_ptr = find_node_record(job_ptr->batch_host)))
agent_args->protocol_version =
node_ptr->protocol_version;
agent_args->msg_args = kill_msg;
set_agent_arg_r_uid(agent_args, slurm_conf.slurmd_user_id);
stepmgr_ops->agent_queue_request(agent_args);
step_signal.found = true;
step_signal.rc_in = SLURM_SUCCESS;
}
if (!step_signal.found) {
info("%s: %pJ StepId=%u not found",
__func__, job_ptr, step_id->step_id);
return ESLURM_INVALID_JOB_ID;
}
return step_signal.rc_in;
}
/*
* signal_step_tasks - send specific signal to specific job step
* IN step_ptr - step record pointer
* IN signal - signal to send
* IN msg_type - message type to send
*/
void signal_step_tasks(step_record_t *step_ptr, uint16_t signal,
slurm_msg_type_t msg_type)
{
node_record_t *node_ptr;
static bool cloud_dns = false;
static time_t last_update = 0;
signal_tasks_msg_t *signal_tasks_msg;
agent_arg_t *agent_args = NULL;
xassert(step_ptr);
agent_args = xmalloc(sizeof(agent_arg_t));
agent_args->msg_type = msg_type;
agent_args->retry = 1;
agent_args->hostlist = hostlist_create(NULL);
signal_tasks_msg = xmalloc(sizeof(signal_tasks_msg_t));
memcpy(&signal_tasks_msg->step_id, &step_ptr->step_id,
sizeof(signal_tasks_msg->step_id));
signal_tasks_msg->signal = signal;
if (step_ptr->flags & SSF_NO_SIG_FAIL)
signal_tasks_msg->flags |= KILL_NO_SIG_FAIL;
log_flag(STEPS, "%s: queueing signal %d with flags=0x%x for %pS",
__func__, signal, signal_tasks_msg->flags, step_ptr);
if (last_update != slurm_conf.last_update) {
if (xstrcasestr(slurm_conf.slurmctld_params, "cloud_dns"))
cloud_dns = true;
else
cloud_dns = false;
last_update = slurm_conf.last_update;
}
agent_args->protocol_version = SLURM_PROTOCOL_VERSION;
for (int i = 0;
(node_ptr = next_node_bitmap(step_ptr->step_node_bitmap, &i));
i++) {
if (agent_args->protocol_version > node_ptr->protocol_version)
agent_args->protocol_version =
node_ptr->protocol_version;
hostlist_push_host(agent_args->hostlist, node_ptr->name);
agent_args->node_count++;
if (PACK_FANOUT_ADDRS(node_ptr))
agent_args->msg_flags |= SLURM_PACK_ADDRS;
}
if (agent_args->node_count == 0) {
xfree(signal_tasks_msg);
hostlist_destroy(agent_args->hostlist);
xfree(agent_args);
return;
}
agent_args->msg_args = signal_tasks_msg;
set_agent_arg_r_uid(agent_args, SLURM_AUTH_UID_ANY);
stepmgr_ops->agent_queue_request(agent_args);
}
/*
* signal_step_tasks_on_node - send specific signal to specific job step
* on a specific node.
* IN node_name - name of node on which to signal tasks
* IN step_ptr - step record pointer
* IN signal - signal to send
* IN msg_type - message type to send
*/
void signal_step_tasks_on_node(char* node_name, step_record_t *step_ptr,
uint16_t signal, slurm_msg_type_t msg_type)
{
signal_tasks_msg_t *signal_tasks_msg;
agent_arg_t *agent_args = NULL;
node_record_t *node_ptr;
xassert(step_ptr);
agent_args = xmalloc(sizeof(agent_arg_t));
agent_args->msg_type = msg_type;
agent_args->retry = 1;
if ((node_ptr = find_node_record(node_name)))
agent_args->protocol_version = node_ptr->protocol_version;
agent_args->node_count++;
agent_args->hostlist = hostlist_create(node_name);
if (!agent_args->hostlist)
fatal("Invalid node_name: %s", node_name);
signal_tasks_msg = xmalloc(sizeof(signal_tasks_msg_t));
memcpy(&signal_tasks_msg->step_id, &step_ptr->step_id,
sizeof(signal_tasks_msg->step_id));
signal_tasks_msg->signal = signal;
agent_args->msg_args = signal_tasks_msg;
set_agent_arg_r_uid(agent_args, SLURM_AUTH_UID_ANY);
stepmgr_ops->agent_queue_request(agent_args);
}
typedef struct {
int config_start_count;
int start_count;
time_t max_age;
} wake_steps_args_t;
static int _wake_steps(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
wake_steps_args_t *args = (wake_steps_args_t *) arg;
if (step_ptr->state != JOB_PENDING)
return 0;
if ((args->start_count < args->config_start_count) ||
(step_ptr->time_last_active <= args->max_age)) {
srun_step_signal(step_ptr, 0);
args->start_count++;
return 1;
}
return 0;
}
/* A step just completed, signal srun processes with pending steps to retry */
static void _wake_pending_steps(job_record_t *job_ptr)
{
static int config_start_count = -1, config_max_age = -1;
wake_steps_args_t args;
if (!IS_JOB_RUNNING(job_ptr))
return;
if (!job_ptr->step_list)
return;
if (config_start_count == -1) {
char *tmp_ptr;
long int param;
config_start_count = 8;
config_max_age = 60;
if ((tmp_ptr = xstrcasestr(slurm_conf.sched_params,
"step_retry_count="))) {
param = strtol(tmp_ptr + 17, NULL, 10);
if ((param >= 1) && (param != LONG_MIN) &&
(param != LONG_MAX))
config_start_count = param;
}
if ((tmp_ptr = xstrcasestr(slurm_conf.sched_params,
"step_retry_time="))) {
param = strtol(tmp_ptr + 16, NULL, 10);
if ((param >= 1) && (param != LONG_MIN) &&
(param != LONG_MAX))
config_max_age = param;
}
}
args.max_age = time(NULL) - config_max_age;
/* We do not know which steps can use currently available resources.
* Try to start a bit more based upon step sizes. Effectiveness
* varies with step sizes, constraints and order. */
args.config_start_count = config_start_count;
args.start_count = 0;
list_delete_all(job_ptr->step_list, _wake_steps, &args);
}
/* Set cur_inx to the next round-robin node index */
static int _next_node_inx(int *cur_inx, int *check_cnt, int len, int node_cnt,
bitstr_t *nodes_bitmap, bitstr_t **picked_node_bitmap,
int start_inx)
{
bool wrapped = false;
xassert(cur_inx);
xassert(check_cnt);
xassert(nodes_bitmap);
xassert(picked_node_bitmap);
if (*check_cnt == 0) {
*cur_inx = start_inx;
} else {
*cur_inx = (*cur_inx + 1) % len;
wrapped = *cur_inx <= start_inx;
if (*cur_inx == start_inx)
return SLURM_ERROR; /* Normal break case */
}
if (*check_cnt >= node_cnt)
return SLURM_ERROR; /* Normal break case */
*cur_inx = bit_ffs_from_bit(nodes_bitmap, *cur_inx);
if (wrapped && (*cur_inx >= start_inx))
return SLURM_ERROR; /* Normal break case */
if (*cur_inx < 0) {
/* This should never happen */
xassert(false);
FREE_NULL_BITMAP(*picked_node_bitmap);
return SLURM_ERROR;
}
(*check_cnt)++;
return SLURM_SUCCESS;
}
/* Pick nodes to be allocated to a job step. If a CPU count is also specified,
* then select nodes with a sufficient CPU count.
* IN job_ptr - job to contain step allocation
* IN/OUT node_bitmap - nodes available (IN), selectect for use (OUT)
* IN node_cnt - step node count specification
* IN cpu_cnt - step CPU count specification
* IN usable_cpu_cnt - count of usable CPUs on each node in node_bitmap
*/
static bitstr_t *_pick_step_nodes_cpus(job_record_t *job_ptr,
bitstr_t *nodes_bitmap, int node_cnt,
int cpu_cnt, uint32_t *usable_cpu_cnt)
{
bitstr_t *picked_node_bitmap = NULL;
int *usable_cpu_array;
int cpu_target; /* Target number of CPUs per allocated node */
int rem_nodes, rem_cpus, save_rem_nodes, save_rem_cpus;
int i;
int start_inx, bit_len, check_cnt;
xassert(node_cnt > 0);
xassert(nodes_bitmap);
xassert(usable_cpu_cnt);
picked_node_bitmap = bit_alloc(node_record_count);
start_inx = job_ptr->job_resrcs->next_step_node_inx;
bit_len = bit_fls(nodes_bitmap) + 1;
if (start_inx >= bit_len)
start_inx = 0;
cpu_target = ROUNDUP(cpu_cnt, node_cnt);
if (cpu_target > 1024)
info("%s: high cpu_target (%d)", __func__, cpu_target);
if ((cpu_cnt <= node_cnt) || (cpu_target > 1024)) {
check_cnt = 0;
while (_next_node_inx(&i, &check_cnt, bit_len, node_cnt,
nodes_bitmap, &picked_node_bitmap,
start_inx) == SLURM_SUCCESS)
bit_set(picked_node_bitmap, i);
return picked_node_bitmap;
}
/* Need to satisfy both a node count and a cpu count */
usable_cpu_array = xcalloc(cpu_target, sizeof(int));
rem_nodes = node_cnt;
rem_cpus = cpu_cnt;
check_cnt = 0;
while (_next_node_inx(&i, &check_cnt, bit_len, bit_len, nodes_bitmap,
&picked_node_bitmap, start_inx) ==
SLURM_SUCCESS) {
if (usable_cpu_cnt[i] < cpu_target) {
usable_cpu_array[usable_cpu_cnt[i]]++;
continue;
}
bit_set(picked_node_bitmap, i);
rem_cpus -= usable_cpu_cnt[i];
rem_nodes--;
if ((rem_cpus <= 0) && (rem_nodes <= 0)) {
/* Satisfied request */
xfree(usable_cpu_array);
return picked_node_bitmap;
}
if (rem_nodes == 0) { /* Reached node limit, not CPU limit */
xfree(usable_cpu_array);
FREE_NULL_BITMAP(picked_node_bitmap);
return NULL;
}
}
if (!picked_node_bitmap) {
xfree(usable_cpu_array);
return NULL;
}
/* Need more resources. Determine what CPU counts per node to use */
save_rem_nodes = rem_nodes;
save_rem_cpus = rem_cpus;
usable_cpu_array[0] = 0;
for (i = (cpu_target - 1); i > 0; i--) {
if (usable_cpu_array[i] == 0)
continue;
if (usable_cpu_array[i] > rem_nodes)
usable_cpu_array[i] = rem_nodes;
if (rem_nodes > 0) {
rem_nodes -= usable_cpu_array[i];
rem_cpus -= (usable_cpu_array[i] * i);
}
}
if ((rem_cpus > 0) || (rem_nodes > 0)){ /* Can not satisfy request */
xfree(usable_cpu_array);
FREE_NULL_BITMAP(picked_node_bitmap);
return NULL;
}
rem_nodes = save_rem_nodes;
rem_cpus = save_rem_cpus;
/* Pick nodes with CPU counts below original target */
check_cnt = 0;
while (_next_node_inx(&i, &check_cnt, bit_len, bit_len, nodes_bitmap,
&picked_node_bitmap, start_inx) ==
SLURM_SUCCESS) {
if (usable_cpu_cnt[i] >= cpu_target)
continue; /* already picked */
if (usable_cpu_array[usable_cpu_cnt[i]] == 0)
continue;
usable_cpu_array[usable_cpu_cnt[i]]--;
bit_set(picked_node_bitmap, i);
rem_cpus -= usable_cpu_cnt[i];
rem_nodes--;
if ((rem_cpus <= 0) && (rem_nodes <= 0)) {
/* Satisfied request */
xfree(usable_cpu_array);
return picked_node_bitmap;
}
if (rem_nodes == 0) /* Reached node limit */
break;
}
/* Can not satisfy request */
xfree(usable_cpu_array);
FREE_NULL_BITMAP(picked_node_bitmap);
return NULL;
}
static int _mark_busy_nodes(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
bitstr_t *busy = (bitstr_t *) arg;
if (step_ptr->state < JOB_RUNNING)
return 0;
/*
* Don't consider the batch and extern steps when
* looking for "idle" nodes.
*/
if ((step_ptr->step_id.step_id == SLURM_BATCH_SCRIPT) ||
(step_ptr->step_id.step_id == SLURM_EXTERN_CONT) ||
(step_ptr->step_id.step_id == SLURM_INTERACTIVE_STEP) ||
(step_ptr->flags & SSF_EXT_LAUNCHER))
return 0;
if (!step_ptr->step_node_bitmap) {
error("%s: %pS has no step_node_bitmap",
__func__, step_ptr);
return 0;
}
bit_or(busy, step_ptr->step_node_bitmap);
if (slurm_conf.debug_flags & DEBUG_FLAG_STEPS) {
char *temp;
temp = bitmap2node_name(step_ptr->step_node_bitmap);
log_flag(STEPS, "%s: %pS has nodes %s",
__func__, step_ptr, temp);
xfree(temp);
}
return 0;
}
static void _step_test_gres(job_step_create_request_msg_t *step_spec,
gres_stepmgr_step_test_args_t *gres_test_args,
job_record_t *job_ptr,
uint32_t *node_usable_cpu_cnt,
uint32_t *total_cpus,
uint32_t *avail_cpus,
int *gres_invalid_nodes,
int *fail_mode)
{
uint64_t gres_cpus;
int err_code = SLURM_SUCCESS;
gres_test_args->err_code = &err_code;
/* ignore current step allocations */
gres_test_args->ignore_alloc = true;
gres_cpus = gres_stepmgr_step_test(gres_test_args);
*total_cpus = MIN(*total_cpus, gres_cpus);
/*
* consider current step allocations if
* not --overlap=force
*/
if (!(step_spec->flags & SSF_OVERLAP_FORCE)) {
gres_test_args->ignore_alloc = false;
gres_cpus = gres_stepmgr_step_test(gres_test_args);
}
if (gres_cpus < *avail_cpus) {
log_flag(STEPS, "%s: %pJ Usable CPUs for GRES %"PRIu64" from %d previously available",
__func__, job_ptr, gres_cpus,
*avail_cpus);
*avail_cpus = gres_cpus;
*node_usable_cpu_cnt = *avail_cpus;
if (err_code != SLURM_SUCCESS)
*fail_mode = err_code;
else
*fail_mode = ESLURM_INVALID_GRES;
if (*total_cpus == 0) {
/*
* total_cpus == 0 is set from this:
* MIN(*total_cpus, gres_cpus);
* This means that it is impossible to run this step on
* this node due to GRES.
*/
*gres_invalid_nodes = *gres_invalid_nodes + 1;
}
}
}
/* Returns threads_per_core required by the step or NO_VAL16 if not specified */
static uint16_t _get_threads_per_core(uint16_t step_threads_per_core,
job_record_t *job_ptr)
{
uint16_t tpc = NO_VAL16;
if (step_threads_per_core &&
(step_threads_per_core != NO_VAL16)) {
tpc = step_threads_per_core;
} else if (job_ptr->details->mc_ptr->threads_per_core &&
(job_ptr->details->mc_ptr->threads_per_core != NO_VAL16))
tpc = job_ptr->details->mc_ptr->threads_per_core;
return tpc;
}
static int _cmp_cpu_counts(const void *num1, const void *num2) {
uint16_t cpu1 = *(uint16_t *) num1;
uint16_t cpu2 = *(uint16_t *) num2;
if (cpu1 > cpu2)
return -1;
else if (cpu1 < cpu2)
return 1;
return 0;
}
static void _set_max_num_tasks(job_step_create_request_msg_t *step_spec,
job_record_t *job_ptr,
bitstr_t *node_bitmap,
int cpus_per_task)
{
int j = 0;
int k = 0;
uint32_t avail_cnt, num_nodes;
uint16_t *cpus;
uint32_t num_tasks = 0;
uint16_t tpc = _get_threads_per_core(step_spec->threads_per_core,
job_ptr);
xassert(node_bitmap);
xassert(cpus_per_task);
avail_cnt = bit_set_count(node_bitmap);
num_nodes = MIN(avail_cnt, step_spec->max_nodes);
cpus = xcalloc(avail_cnt, sizeof(*cpus));
for (int i = 0; i < job_ptr->job_resrcs->nhosts; i++) {
j = bit_ffs_from_bit(job_ptr->job_resrcs->node_bitmap, j);
if (j < 0)
break;
if (!bit_test(node_bitmap, j)) {
j++;
continue;
}
if (tpc != NO_VAL16) {
cpus[k] = ROUNDUP(job_ptr->job_resrcs->cpus[i],
node_record_table_ptr[j]->tpc);
cpus[k] *= tpc;
} else
cpus[k] = job_ptr->job_resrcs->cpus[i];
j++;
k++;
}
if (num_nodes < avail_cnt)
qsort(cpus, avail_cnt, sizeof(*cpus), _cmp_cpu_counts);
for (int i = 0; i < num_nodes; i++) {
num_tasks += cpus[i] / cpus_per_task;
}
step_spec->num_tasks = num_tasks;
step_spec->cpu_count = num_tasks * cpus_per_task;
xfree(cpus);
}
/*
* _pick_step_nodes - select nodes for a job step that satisfy its requirements
* we satisfy the super-set of constraints.
* IN job_ptr - pointer to job to have new step started
* IN step_spec - job step specification
* IN step_gres_list - job step's gres requirement details
* IN cpus_per_task - NOTE could be zero
* IN node_count - How many real nodes a select plugin should be looking for
* OUT return_code - exit code or SLURM_SUCCESS
* global: node_record_table_ptr - pointer to global node table
* NOTE: returns all of a job's nodes if step_spec->node_count == INFINITE
* NOTE: returned bitmap must be freed by the caller using FREE_NULL_BITMAP()
*/
static bitstr_t *_pick_step_nodes(job_record_t *job_ptr,
job_step_create_request_msg_t *step_spec,
list_t *step_gres_list, int cpus_per_task,
uint32_t node_count, int *return_code)
{
node_record_t *node_ptr;
bitstr_t *nodes_avail = NULL, *nodes_idle = NULL;
bitstr_t *select_nodes_avail = NULL;
bitstr_t *nodes_picked = NULL, *node_tmp = NULL;
int error_code, nodes_picked_cnt = 0, cpus_picked_cnt = 0;
int cpu_cnt, i;
int mem_blocked_nodes = 0, mem_blocked_cpus = 0;
int job_blocked_nodes = 0, job_blocked_cpus = 0;
int gres_invalid_nodes = 0;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
uint32_t *usable_cpu_cnt = NULL;
gres_stepmgr_step_test_args_t gres_test_args = {
.cpus_per_task = cpus_per_task,
.first_step_node = true,
.job_gres_list = job_ptr->gres_list_alloc,
.job_id = job_ptr->job_id,
.job_resrcs_ptr = job_resrcs_ptr,
.max_rem_nodes = step_spec->max_nodes,
.step_gres_list = step_gres_list,
.step_id = NO_VAL,
.test_mem = false,
};
xassert(job_resrcs_ptr);
xassert(job_resrcs_ptr->cpus);
xassert(job_resrcs_ptr->cpus_used);
*return_code = SLURM_SUCCESS;
if (job_ptr->node_bitmap == NULL) {
*return_code = ESLURM_REQUESTED_NODE_CONFIG_UNAVAILABLE;
return NULL;
}
if (step_spec->max_nodes == 0)
step_spec->max_nodes = job_ptr->node_cnt;
if (step_spec->max_nodes < step_spec->min_nodes) {
*return_code = ESLURM_INVALID_NODE_COUNT;
return NULL;
}
if (!nodes_avail)
nodes_avail = bit_copy (job_ptr->node_bitmap);
bit_and(nodes_avail, stepmgr_ops->up_node_bitmap);
if (step_spec->exc_nodes) {
bitstr_t *exc_bitmap = NULL;
error_code = node_name2bitmap(step_spec->exc_nodes, false,
&exc_bitmap, NULL);
if (error_code)
debug("Assuming invalid nodename requested for exclusion from step is excluded");
bit_and_not(nodes_avail, exc_bitmap);
if (step_spec->node_list) {
bitstr_t *req_nodes = NULL;
error_code = node_name2bitmap(
step_spec->node_list, false,
&req_nodes, NULL);
if (error_code) {
info("%s: invalid requested node list %s",
__func__,
step_spec->node_list);
FREE_NULL_BITMAP(exc_bitmap);
FREE_NULL_BITMAP(req_nodes);
goto cleanup;
}
if (bit_overlap_any(req_nodes, exc_bitmap)) {
info("%s: %ps requested nodes %s is also excluded %s",
__func__, &step_spec->step_id,
step_spec->node_list,
step_spec->exc_nodes);
FREE_NULL_BITMAP(exc_bitmap);
FREE_NULL_BITMAP(req_nodes);
goto cleanup;
}
FREE_NULL_BITMAP(req_nodes);
}
FREE_NULL_BITMAP(exc_bitmap);
}
if (step_spec->features &&
(!job_ptr->details ||
xstrcmp(step_spec->features, job_ptr->details->features_use))) {
/*
* We only select for a single feature name here.
* Ignore step features if equal to job features.
* FIXME: Add support for AND, OR, etc. here if desired
*/
node_feature_t *feat_ptr;
feat_ptr =
list_find_first(active_feature_list, list_find_feature,
(void *) step_spec->features);
if (feat_ptr && feat_ptr->node_bitmap)
bit_and(nodes_avail, feat_ptr->node_bitmap);
else {
bit_clear_all(nodes_avail);
*return_code = ESLURM_INVALID_FEATURE;
goto cleanup;
}
}
if (step_spec->pn_min_memory &&
((job_resrcs_ptr->memory_allocated == NULL) ||
(job_resrcs_ptr->memory_used == NULL))) {
error("%s: job lacks memory allocation details to enforce memory limits for %pJ",
__func__, job_ptr);
step_spec->pn_min_memory = 0;
} else if (step_spec->pn_min_memory == MEM_PER_CPU)
step_spec->pn_min_memory = 0; /* clear MEM_PER_CPU flag */
if (job_ptr->next_step_id == 0) {
for (int i = 0;
(node_ptr = next_node_bitmap(job_ptr->node_bitmap, &i));
i++) {
if (IS_NODE_POWERED_DOWN(node_ptr) ||
IS_NODE_FUTURE(node_ptr) ||
IS_NODE_NO_RESPOND(node_ptr)) {
/*
* Node is/was powered down. Need to wait
* for it to start responding again.
*/
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
*return_code = ESLURM_NODES_BUSY;
return NULL;
}
}
if (IS_JOB_CONFIGURING(job_ptr)) {
info("%s: Configuration for %pJ is complete",
__func__, job_ptr);
stepmgr_ops->job_config_fini(job_ptr);
}
}
if (_is_mem_resv() && step_spec->pn_min_memory &&
((step_spec->pn_min_memory & MEM_PER_CPU) == 0) &&
job_ptr->details && job_ptr->details->pn_min_memory &&
((job_ptr->details->pn_min_memory & MEM_PER_CPU) == 0) &&
(step_spec->pn_min_memory >
job_ptr->details->pn_min_memory)) {
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
*return_code = ESLURM_INVALID_TASK_MEMORY;
return NULL;
}
usable_cpu_cnt = xcalloc(node_record_count, sizeof(uint32_t));
for (int i = 0, node_inx = -1;
(node_ptr = next_node_bitmap(job_resrcs_ptr->node_bitmap, &i));
i++) {
node_inx++;
if (!bit_test(nodes_avail, i))
continue; /* node now DOWN */
usable_cpu_cnt[i] = job_resrcs_ptr->cpus[node_inx];
log_flag(STEPS, "%s: %pJ Currently running steps use %d of allocated %d CPUs on node %s",
__func__, job_ptr,
job_resrcs_ptr->cpus_used[node_inx],
usable_cpu_cnt[i], node_record_table_ptr[i]->name);
/*
* Don't do this test if --overlap=force or
* --external-launcher
*/
if ((!(step_spec->flags & SSF_OVERLAP_FORCE)) &&
(!(step_spec->flags & SSF_EXT_LAUNCHER))) {
/*
* If whole is given and
* job_resrcs_ptr->cpus_used[node_inx]
* we can't use this node.
*/
if ((step_spec->flags & SSF_WHOLE) &&
job_resrcs_ptr->cpus_used[node_inx]) {
log_flag(STEPS, "%s: %pJ Node requested --whole node while other step running here.",
__func__, job_ptr);
job_blocked_cpus +=
job_resrcs_ptr->cpus_used[node_inx];
job_blocked_nodes++;
usable_cpu_cnt[i] = 0;
} else {
usable_cpu_cnt[i] -=
job_resrcs_ptr->cpus_used[node_inx];
job_blocked_cpus +=
job_resrcs_ptr->cpus_used[node_inx];
if (!usable_cpu_cnt[i]) {
job_blocked_nodes++;
log_flag(STEPS, "%s: %pJ Skipping node %s. Not enough CPUs to run step here.",
__func__,
job_ptr,
node_record_table_ptr[i]->name);
}
}
}
if (!usable_cpu_cnt[i]) {
bit_clear(nodes_avail, i);
continue;
}
if ((step_spec->pn_min_memory && _is_mem_resv()) ||
step_gres_list) {
int fail_mode = ESLURM_NODES_BUSY;
uint64_t tmp_mem;
uint32_t tmp_cpus, avail_cpus, total_cpus;
uint32_t avail_tasks, total_tasks;
gres_test_args.node_offset = node_inx;
gres_test_args.test_mem = false;
avail_cpus = total_cpus = usable_cpu_cnt[i];;
if (_is_mem_resv() &&
step_spec->pn_min_memory & MEM_PER_CPU) {
uint64_t mem_use = step_spec->pn_min_memory;
mem_use &= (~MEM_PER_CPU);
/* ignore current step allocations */
tmp_mem = job_resrcs_ptr->
memory_allocated[node_inx];
tmp_cpus = tmp_mem / mem_use;
total_cpus = MIN(total_cpus, tmp_cpus);
/*
* consider current step allocations if
* not --overlap=force
*/
if (!(step_spec->flags & SSF_OVERLAP_FORCE)) {
tmp_mem -= job_resrcs_ptr->
memory_used[node_inx];
tmp_cpus = tmp_mem / mem_use;
}
if (tmp_cpus < avail_cpus) {
avail_cpus = tmp_cpus;
usable_cpu_cnt[i] = avail_cpus;
fail_mode = ESLURM_INVALID_TASK_MEMORY;
}
log_flag(STEPS, "%s: %pJ Based on --mem-per-cpu=%"PRIu64" we have %d/%d usable of available cpus on node %s, usable memory was: %"PRIu64,
__func__, job_ptr, mem_use, tmp_cpus,
avail_cpus, node_ptr->name, tmp_mem);
} else if (_is_mem_resv() && step_spec->pn_min_memory) {
uint64_t mem_use = step_spec->pn_min_memory;
/* ignore current step allocations */
tmp_mem = job_resrcs_ptr->
memory_allocated[node_inx];
if (tmp_mem < mem_use)
total_cpus = 0;
/*
* consider current step allocations if
* not --overlap=force
*/
if (!(step_spec->flags & SSF_OVERLAP_FORCE)) {
tmp_mem -= job_resrcs_ptr->
memory_used[node_inx];
}
if ((tmp_mem < mem_use) && (avail_cpus > 0)) {
log_flag(STEPS, "%s: %pJ Usable memory on node %s: %"PRIu64" is less than requested %"PRIu64" skipping the node",
__func__, job_ptr,
node_ptr->name,
tmp_mem,
mem_use);
avail_cpus = 0;
usable_cpu_cnt[i] = avail_cpus;
fail_mode = ESLURM_INVALID_TASK_MEMORY;
}
} else if (_is_mem_resv())
gres_test_args.test_mem = true;
_step_test_gres(step_spec, &gres_test_args, job_ptr,
&usable_cpu_cnt[i],
&total_cpus, &avail_cpus,
&gres_invalid_nodes,
&fail_mode);
avail_tasks = avail_cpus;
total_tasks = total_cpus;
if (cpus_per_task > 0) {
avail_tasks /= cpus_per_task;
total_tasks /= cpus_per_task;
}
if (avail_tasks == 0) {
log_flag(STEPS, "%s: %pJ No task can start on node %s",
__func__, job_ptr, node_ptr->name);
if ((step_spec->min_nodes == INFINITE) ||
(step_spec->min_nodes ==
job_ptr->node_cnt)) {
log_flag(STEPS, "%s: %pJ All nodes in allocation required, but can't use them now",
__func__, job_ptr);
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
xfree(usable_cpu_cnt);
*return_code = ESLURM_NODES_BUSY;
if (total_tasks == 0 &&
(fail_mode !=
ESLURM_INVALID_TASK_MEMORY)) {
*return_code = fail_mode;
log_flag(STEPS, "%s: %pJ Step cannot ever run in the allocation: %s",
__func__,
job_ptr,
slurm_strerror(
fail_mode));
}
return NULL;
}
bit_clear(nodes_avail, i);
mem_blocked_nodes++;
mem_blocked_cpus += (total_cpus - avail_cpus);
} else {
mem_blocked_cpus += (total_cpus - avail_cpus);
gres_test_args.first_step_node = false;
}
}
}
if (gres_invalid_nodes >
(job_resrcs_ptr->nhosts - step_spec->min_nodes)) {
*return_code = ESLURM_INVALID_GRES;
log_flag(STEPS, "%s: Never able to satisfy the GRES request for this step",
__func__);
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
xfree(usable_cpu_cnt);
return NULL;
}
if (step_spec->min_nodes == INFINITE) { /* use all nodes */
if ((step_spec->num_tasks == NO_VAL) && nodes_avail &&
!(step_spec->flags & SSF_EXT_LAUNCHER)) {
_set_max_num_tasks(step_spec, job_ptr, nodes_avail,
cpus_per_task);
if (step_spec->num_tasks == 0) {
log_flag(STEPS, "%s: Step requested more processors per task (%d) than can be satisfied.",
__func__, cpus_per_task);
*return_code = ESLURM_TOO_MANY_REQUESTED_CPUS;
goto cleanup;
}
}
job_resrcs_ptr->next_step_node_inx = 0;
xfree(usable_cpu_cnt);
FREE_NULL_BITMAP(select_nodes_avail);
return nodes_avail;
}
if (select_nodes_avail) {
/*
* The select plugin told us these were the only ones we could
* choose from. If it doesn't fit here then defer request
*/
bit_and(nodes_avail, select_nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
}
/*
* An allocating srun will send in the same node_list that was already
* used to construct the job allocation. In that case, we can assume
* that the job allocation already satisfies those requirements.
*/
if (step_spec->node_list && xstrcmp(step_spec->node_list,
job_ptr->details->req_nodes)) {
bitstr_t *selected_nodes = NULL;
log_flag(STEPS, "%s: selected nodelist is %s",
__func__, step_spec->node_list);
error_code = node_name2bitmap(step_spec->node_list, false,
&selected_nodes, NULL);
if (error_code) {
log_flag(STEPS, "%s: invalid node list %s", __func__,
step_spec->node_list);
FREE_NULL_BITMAP(selected_nodes);
goto cleanup;
}
if (!bit_super_set(selected_nodes, job_ptr->node_bitmap)) {
log_flag(STEPS, "%s: requested nodes %s not part of %pJ",
__func__, step_spec->node_list, job_ptr);
FREE_NULL_BITMAP(selected_nodes);
goto cleanup;
}
if (!bit_super_set(selected_nodes, nodes_avail)) {
/*
* If some nodes still have some memory or CPUs
* allocated to other steps, just defer the execution
* of the step
*/
if (job_blocked_nodes) {
*return_code = ESLURM_NODES_BUSY;
log_flag(STEPS, "%s: some requested nodes %s still have CPUs used by other steps",
__func__, step_spec->node_list);
} else if (mem_blocked_nodes == 0) {
*return_code = ESLURM_INVALID_TASK_MEMORY;
log_flag(STEPS, "%s: requested nodes %s have inadequate memory",
__func__, step_spec->node_list);
} else {
*return_code = ESLURM_NODES_BUSY;
log_flag(STEPS, "%s: some requested nodes %s still have memory used by other steps",
__func__, step_spec->node_list);
}
FREE_NULL_BITMAP(selected_nodes);
goto cleanup;
}
if ((step_spec->task_dist & SLURM_DIST_STATE_BASE) ==
SLURM_DIST_ARBITRARY) {
step_spec->min_nodes = bit_set_count(selected_nodes);
}
if (selected_nodes) {
int node_cnt = 0;
/*
* Use selected nodes to run the step and
* mark them unavailable for future use
*/
/*
* If we have selected more than we requested
* make the available nodes equal to the
* selected nodes and we will pick from that
* list later on in the function.
* Other than that copy the nodes selected as
* the nodes we want.
*/
node_cnt = bit_set_count(selected_nodes);
if (node_cnt > step_spec->max_nodes) {
log_flag(STEPS, "%s: requested nodes %s exceed max node count for %pJ (%d > %u)",
__func__, step_spec->node_list,
job_ptr, node_cnt,
step_spec->max_nodes);
FREE_NULL_BITMAP(selected_nodes);
goto cleanup;
} else if (step_spec->min_nodes &&
(node_cnt > step_spec->min_nodes)) {
nodes_picked = bit_alloc(bit_size(nodes_avail));
FREE_NULL_BITMAP(nodes_avail);
nodes_avail = selected_nodes;
selected_nodes = NULL;
} else {
nodes_picked = bit_copy(selected_nodes);
bit_and_not(nodes_avail, selected_nodes);
FREE_NULL_BITMAP(selected_nodes);
}
}
} else {
nodes_picked = bit_alloc(bit_size(nodes_avail));
}
/* If gres_per_step then filter nodes_avail to nodes that fill req */
gres_stepmgr_step_test_per_step(step_gres_list, job_ptr,
nodes_avail, step_spec->min_nodes);
/*
* In case we are in relative mode, do not look for idle nodes
* as we will not try to get idle nodes first but try to get
* the relative node first
*/
if (step_spec->relative != NO_VAL16) {
/*
* Remove first (step_spec->relative) nodes from
* available list
*/
bitstr_t *relative_nodes = NULL;
relative_nodes = bit_pick_cnt(job_ptr->node_bitmap,
step_spec->relative);
if (relative_nodes == NULL) {
log_flag(STEPS, "%s: Invalid relative value (%u) for %pJ",
__func__, step_spec->relative, job_ptr);
goto cleanup;
}
bit_and_not(nodes_avail, relative_nodes);
FREE_NULL_BITMAP(relative_nodes);
} else {
nodes_idle = bit_alloc (bit_size (nodes_avail) );
list_for_each(job_ptr->step_list, _mark_busy_nodes, nodes_idle);
bit_not(nodes_idle);
bit_and(nodes_idle, nodes_avail);
}
if (slurm_conf.debug_flags & DEBUG_FLAG_STEPS) {
char *temp1, *temp2, *temp3;
temp1 = bitmap2node_name(nodes_avail);
temp2 = bitmap2node_name(nodes_idle);
if (step_spec->node_list)
temp3 = step_spec->node_list;
else
temp3 = "NONE";
log_flag(STEPS, "%s: step pick %u-%u nodes, avail:%s idle:%s picked:%s",
__func__, step_spec->min_nodes, step_spec->max_nodes,
temp1, temp2, temp3);
xfree(temp1);
xfree(temp2);
}
if ((step_spec->num_tasks == NO_VAL) &&
!(step_spec->flags & SSF_EXT_LAUNCHER)) {
uint32_t cnt = 0;
bitstr_t *node_bitmap = NULL;
if ((step_spec->flags & SSF_OVERLAP_FORCE) && nodes_avail) {
cnt = bit_set_count(nodes_avail);
node_bitmap = nodes_avail;
} else if (nodes_idle) {
cnt = bit_set_count(nodes_idle);
node_bitmap = nodes_idle;
}
if (cnt < step_spec->min_nodes) {
log_flag(STEPS, "%s: Step requested more nodes (%u) than are available (%d), deferring step until enough nodes are available.",
__func__, step_spec->min_nodes, cnt);
*return_code = ESLURM_NODES_BUSY;
goto cleanup;
}
_set_max_num_tasks(step_spec, job_ptr, node_bitmap,
cpus_per_task);
if (step_spec->num_tasks == 0) {
log_flag(STEPS, "%s: Step requested more processors per task (%d) than can be satisfied.",
__func__, cpus_per_task);
*return_code = ESLURM_TOO_MANY_REQUESTED_CPUS;
goto cleanup;
}
}
/*
* If user specifies step needs a specific processor count and
* all nodes have the same processor count, just translate this to
* a node count
*/
if (step_spec->cpu_count && job_resrcs_ptr &&
(job_resrcs_ptr->cpu_array_cnt == 1) &&
(job_resrcs_ptr->cpu_array_value)) {
uint32_t cpu_count = step_spec->cpu_count;
uint16_t req_tpc;
/*
* Expand cpu account to account for blocked/used threads when
* using threads-per-core. See _step_[de]alloc_lps() for similar
* code.
*/
req_tpc = _get_threads_per_core(step_spec->threads_per_core,
job_ptr);
/*
* Only process this differently if the allocation requested
* more threads per core than the step is requesting as
* job_resrcs->cpu_array_value is already processed with the
* threads per core the allocation requested so you don't need
* to do this again. See src/common/job_resources.c
* build_job_resources_cpu_array().
*/
if ((req_tpc != NO_VAL16) &&
(req_tpc < job_resrcs_ptr->threads_per_core)) {
int first_inx = bit_ffs(job_resrcs_ptr->node_bitmap);
if (first_inx == -1) {
error("%s: Job %pJ doesn't have any nodes in it! This should never happen",
__func__, job_ptr);
*return_code = ESLURM_INVALID_NODE_COUNT;
goto cleanup;
}
if (req_tpc < node_record_table_ptr[first_inx]->tpc) {
cpu_count = ROUNDUP(cpu_count, req_tpc);
cpu_count *=
node_record_table_ptr[first_inx]->tpc;
} else if (req_tpc >
node_record_table_ptr[first_inx]->tpc) {
log_flag(STEPS, "%s: requested more threads per core than possible in allocation (%u > %u) for %pJ",
__func__,
req_tpc,
node_record_table_ptr[first_inx]->tpc,
job_ptr);
*return_code = ESLURM_BAD_THREAD_PER_CORE;
goto cleanup;
}
}
i = ROUNDUP(cpu_count, job_resrcs_ptr->cpu_array_value[0]);
step_spec->min_nodes = (i > step_spec->min_nodes) ?
i : step_spec->min_nodes ;
/*
* If we are trying to pack the nodes we only want the minimum
* it takes to satisfy the request.
*/
if (step_spec->task_dist & SLURM_DIST_PACK_NODES)
step_spec->max_nodes = step_spec->min_nodes;
if (step_spec->max_nodes < step_spec->min_nodes) {
log_flag(STEPS, "%s: %pJ max node less than min node count (%u < %u)",
__func__, job_ptr, step_spec->max_nodes,
step_spec->min_nodes);
*return_code = ESLURM_TOO_MANY_REQUESTED_CPUS;
goto cleanup;
}
}
if (step_spec->min_nodes) {
int cpus_needed, node_avail_cnt, nodes_needed;
nodes_picked_cnt = bit_set_count(nodes_picked);
log_flag(STEPS, "%s: step picked %d of %u nodes",
__func__, nodes_picked_cnt, step_spec->min_nodes);
/*
* First do a basic test - if there aren't enough nodes for
* this step to run on then we need to defer execution of this
* step. As long as there aren't enough nodes for this
* step we can never test if the step requested too
* many CPUs, too much memory, etc. so we just bail right here.
*/
if (nodes_avail)
node_avail_cnt = bit_set_count(nodes_avail);
else
node_avail_cnt = 0;
if ((node_avail_cnt + nodes_picked_cnt) <
step_spec->min_nodes) {
log_flag(STEPS, "%s: Step requested more nodes (%u) than are available (%d), deferring step until enough nodes are available.",
__func__, step_spec->min_nodes,
node_avail_cnt);
*return_code = ESLURM_NODES_BUSY;
goto cleanup;
}
if (nodes_idle)
node_avail_cnt = bit_set_count(nodes_idle);
else
node_avail_cnt = 0;
nodes_needed = step_spec->min_nodes - nodes_picked_cnt;
if ((nodes_needed > 0) &&
(node_avail_cnt >= nodes_needed)) {
cpus_needed = _opt_cpu_cnt(step_spec->cpu_count,
nodes_picked,
usable_cpu_cnt);
nodes_needed = _opt_node_cnt(step_spec->min_nodes,
step_spec->max_nodes,
node_avail_cnt,
nodes_picked_cnt);
node_tmp = _pick_step_nodes_cpus(job_ptr, nodes_idle,
nodes_needed,
cpus_needed,
usable_cpu_cnt);
if (node_tmp) {
bit_or(nodes_picked, node_tmp);
bit_and_not(nodes_idle, node_tmp);
bit_and_not(nodes_avail, node_tmp);
FREE_NULL_BITMAP(node_tmp);
nodes_picked_cnt = step_spec->min_nodes;
nodes_needed = 0;
}
}
if (nodes_avail)
node_avail_cnt = bit_set_count(nodes_avail);
else
node_avail_cnt = 0;
if ((nodes_needed > 0) &&
(node_avail_cnt >= nodes_needed)) {
cpus_needed = _opt_cpu_cnt(step_spec->cpu_count,
nodes_picked,
usable_cpu_cnt);
nodes_needed = _opt_node_cnt(step_spec->min_nodes,
step_spec->max_nodes,
node_avail_cnt,
nodes_picked_cnt);
node_tmp = _pick_step_nodes_cpus(job_ptr, nodes_avail,
nodes_needed,
cpus_needed,
usable_cpu_cnt);
if (node_tmp == NULL) {
/* Count of nodes already picked for step */
int pick_node_cnt = bit_set_count(nodes_avail);
pick_node_cnt += nodes_picked_cnt;
if ((step_spec->max_nodes <= pick_node_cnt) &&
(mem_blocked_cpus == 0) &&
(job_blocked_cpus == 0)) {
*return_code =
ESLURM_TOO_MANY_REQUESTED_CPUS;
} else if ((mem_blocked_cpus > 0) ||
(step_spec->min_nodes <=
(pick_node_cnt + mem_blocked_nodes +
job_blocked_nodes))) {
*return_code = ESLURM_NODES_BUSY;
} else if (!bit_super_set(job_ptr->node_bitmap,
stepmgr_ops->up_node_bitmap)) {
*return_code = ESLURM_NODE_NOT_AVAIL;
}
goto cleanup;
}
bit_or(nodes_picked, node_tmp);
bit_and_not(nodes_avail, node_tmp);
FREE_NULL_BITMAP(node_tmp);
nodes_picked_cnt = step_spec->min_nodes;
} else if (nodes_needed > 0) {
if ((step_spec->max_nodes <= nodes_picked_cnt) &&
(mem_blocked_cpus == 0) &&
(job_blocked_cpus == 0)) {
*return_code = ESLURM_TOO_MANY_REQUESTED_CPUS;
} else if ((mem_blocked_cpus > 0) ||
(step_spec->min_nodes <=
(nodes_picked_cnt + mem_blocked_nodes +
job_blocked_nodes))) {
*return_code = ESLURM_NODES_BUSY;
} else if (!bit_super_set(job_ptr->node_bitmap,
stepmgr_ops->up_node_bitmap)) {
*return_code = ESLURM_NODE_NOT_AVAIL;
}
goto cleanup;
}
}
if (step_spec->cpu_count) {
/* make sure the selected nodes have enough cpus */
cpus_picked_cnt = _count_cpus(job_ptr, nodes_picked,
usable_cpu_cnt);
if ((step_spec->cpu_count > cpus_picked_cnt) &&
(step_spec->max_nodes > nodes_picked_cnt)) {
/* Attempt to add more nodes to allocation */
nodes_picked_cnt = bit_set_count(nodes_picked);
while (step_spec->cpu_count > cpus_picked_cnt) {
node_tmp = bit_pick_cnt(nodes_avail, 1);
if (node_tmp == NULL)
break;
cpu_cnt = _count_cpus(job_ptr, node_tmp,
usable_cpu_cnt);
if (cpu_cnt == 0) {
/*
* Node not usable (memory insufficient
* to allocate any CPUs, etc.)
*/
bit_and_not(nodes_avail, node_tmp);
FREE_NULL_BITMAP(node_tmp);
continue;
}
bit_or(nodes_picked, node_tmp);
bit_and_not(nodes_avail, node_tmp);
FREE_NULL_BITMAP(node_tmp);
nodes_picked_cnt += 1;
if (step_spec->min_nodes)
step_spec->min_nodes = nodes_picked_cnt;
cpus_picked_cnt += cpu_cnt;
if (nodes_picked_cnt >= step_spec->max_nodes)
break;
}
}
/*
* User is requesting more cpus than we got from the
* picked nodes. We should return with an error
*/
if (step_spec->cpu_count > cpus_picked_cnt) {
if (step_spec->cpu_count &&
(step_spec->cpu_count <=
(cpus_picked_cnt + mem_blocked_cpus +
job_blocked_cpus))) {
*return_code = ESLURM_NODES_BUSY;
} else if (!bit_super_set(job_ptr->node_bitmap,
stepmgr_ops->up_node_bitmap)) {
*return_code = ESLURM_NODE_NOT_AVAIL;
}
log_flag(STEPS, "Have %d nodes with %d cpus which is less than what the user is asking for (%d cpus) aborting.",
nodes_picked_cnt,
cpus_picked_cnt,
step_spec->cpu_count);
goto cleanup;
}
}
job_resrcs_ptr->next_step_node_inx = bit_fls(nodes_picked) + 1;
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
FREE_NULL_BITMAP(nodes_idle);
xfree(usable_cpu_cnt);
return nodes_picked;
cleanup:
FREE_NULL_BITMAP(nodes_avail);
FREE_NULL_BITMAP(select_nodes_avail);
FREE_NULL_BITMAP(nodes_idle);
FREE_NULL_BITMAP(nodes_picked);
xfree(usable_cpu_cnt);
if (*return_code == SLURM_SUCCESS) {
*return_code = ESLURM_REQUESTED_NODE_CONFIG_UNAVAILABLE;
} else if (*return_code == ESLURM_NODE_NOT_AVAIL) {
/*
* Return ESLURM_NODES_BUSY if the node is not responding.
* The node will eventually either come back UP or go DOWN.
*/
nodes_picked = bit_copy(stepmgr_ops->up_node_bitmap);
bit_not(nodes_picked);
bit_and(nodes_picked, job_ptr->node_bitmap);
for (i = 0; (node_ptr = next_node_bitmap(
job_resrcs_ptr->node_bitmap, &i));
i++) {
if (!IS_NODE_NO_RESPOND(node_ptr)) {
*return_code = ESLURM_NODES_BUSY;
break;
}
}
FREE_NULL_BITMAP(nodes_picked);
}
return NULL;
}
/*
* _count_cpus - report how many cpus are allocated to this job for the
* identified nodes
* IN job_ptr - point to job
* IN bitmap - map of nodes to tally
* IN usable_cpu_cnt - count of usable CPUs based upon memory or gres specs
* NULL if not available
* RET cpu count
*/
static int _count_cpus(job_record_t *job_ptr, bitstr_t *bitmap,
uint32_t *usable_cpu_cnt)
{
int i, sum = 0;
node_record_t *node_ptr;
if (job_ptr->job_resrcs && job_ptr->job_resrcs->cpus &&
job_ptr->job_resrcs->node_bitmap) {
int node_inx = -1;
for (i = 0;
(node_ptr = next_node_bitmap(
job_ptr->job_resrcs->node_bitmap, &i));
i++) {
node_inx++;
if (!bit_test(job_ptr->node_bitmap, node_ptr->index) ||
!bit_test(bitmap, node_ptr->index)) {
/* absent from current job or step bitmap */
continue;
}
if (usable_cpu_cnt)
sum += usable_cpu_cnt[node_ptr->index];
else
sum += job_ptr->job_resrcs->cpus[node_inx];
}
} else {
error("%pJ lacks cpus array", job_ptr);
for (i = 0; (node_ptr = next_node_bitmap(bitmap, &i)); i++) {
sum += node_ptr->config_ptr->cpus;
}
}
return sum;
}
/* Clear avail_core_bitmap cores which are not bound to the allocated gres */
static int _gres_filter_avail_cores(void *x, void *arg)
{
gres_state_t *gres_state_step = x;
foreach_gres_filter_t *args = arg;
gres_step_state_t *gres_ss = gres_state_step->gres_data;
bitstr_t *filter_core_bitmap;
gres_state_t *gres_state_node;
gres_node_state_t *gres_ns;
/* Bail early if this GRES isn't used on the node */
if (!gres_ss->gres_cnt_node_alloc ||
!gres_ss->gres_cnt_node_alloc[args->job_node_inx])
return 0;
if (!(gres_state_node = list_find_first(args->node_gres_list,
gres_find_id,
&gres_state_step->plugin_id))) {
error("No node gres when step gres is allocated. This should never happen.");
return 0;
}
gres_ns = gres_state_node->gres_data;
if (!gres_ns->topo_cnt) /* No topology info */
return 0;
filter_core_bitmap = bit_copy(args->all_gres_core_bitmap);
/* Determine which specific cores can be used */
for (int i = 0; i < gres_ns->topo_cnt; i++) {
/* Is this gres allocated to the step? */
if (gres_ss->gres_bit_alloc &&
!bit_overlap_any(
gres_ss->gres_bit_alloc[args->job_node_inx],
gres_ns->topo_gres_bitmap[i]))
continue;
/* Does it specify which cores which can use it */
if (!gres_ns->topo_core_bitmap[i]) {
bit_nset(args->any_gres_core_bitmap,
args->core_start_bit, args->core_end_bit);
continue;
}
bit_nclear(filter_core_bitmap, args->core_start_bit,
args->core_end_bit);
for (int j = 0;
j < bit_size(gres_ns->topo_core_bitmap[i]);
j++) {
if (bit_test(gres_ns->topo_core_bitmap[i], j)) {
bit_set(filter_core_bitmap,
args->core_start_bit + j);
}
}
bit_or(args->any_gres_core_bitmap, filter_core_bitmap);
bit_and(args->all_gres_core_bitmap, filter_core_bitmap);
}
FREE_NULL_BITMAP(filter_core_bitmap);
return 0;
}
/* Return true if a core was picked, false if not */
static bool _pick_step_core(step_record_t *step_ptr,
job_resources_t *job_resrcs_ptr,
bitstr_t *avail_core_bitmap, int job_node_inx,
int sock_inx, int core_inx, bool use_all_cores)
{
int bit_offset;
bit_offset = get_job_resources_offset(job_resrcs_ptr,
job_node_inx,
sock_inx,
core_inx);
if (bit_offset < 0)
fatal("get_job_resources_offset");
if (!bit_test(avail_core_bitmap, bit_offset))
return false;
if (bit_test(step_ptr->core_bitmap_job, bit_offset))
return false; /* already taken by this step */
/* Check and set the job's used cores. */
if (!(step_ptr->flags & SSF_OVERLAP_FORCE)) {
if ((use_all_cores == false) &&
bit_test(job_resrcs_ptr->core_bitmap_used, bit_offset))
return false;
bit_set(job_resrcs_ptr->core_bitmap_used, bit_offset);
}
log_flag(STEPS, "%s: alloc Node:%d Socket:%d Core:%d",
__func__, job_node_inx, sock_inx, core_inx);
bit_set(step_ptr->core_bitmap_job, bit_offset);
return true;
}
static bool _handle_core_select(step_record_t *step_ptr,
job_resources_t *job_resrcs_ptr,
bitstr_t *avail_core_bitmap,
int job_node_inx, uint16_t sockets,
uint16_t cores, bool use_all_cores,
int *core_cnt, uint16_t cores_per_task)
{
int core_inx, sock_inx;
xassert(core_cnt);
if (*core_cnt <= 0)
return true;
/*
* Figure out the task distribution. The default is to cyclically
* distribute to sockets.
*/
if (step_ptr->step_layout &&
((step_ptr->step_layout->task_dist & SLURM_DIST_SOCKMASK) ==
SLURM_DIST_SOCKBLOCK)) {
/* Fill sockets before allocating to the next socket */
for (sock_inx=0; sock_inx < sockets; sock_inx++) {
for (core_inx = 0; core_inx < cores; core_inx++) {
if (!_pick_step_core(step_ptr, job_resrcs_ptr,
avail_core_bitmap,
job_node_inx, sock_inx,
core_inx, use_all_cores))
continue;
if (--(*core_cnt) == 0)
return true;
}
}
} else if (step_ptr->step_layout &&
((step_ptr->step_layout->task_dist & SLURM_DIST_SOCKMASK) ==
SLURM_DIST_SOCKCFULL)) {
for (core_inx = 0; core_inx < cores; core_inx++) {
for (sock_inx = 0; sock_inx < sockets; sock_inx++) {
if (!_pick_step_core(step_ptr, job_resrcs_ptr,
avail_core_bitmap,
job_node_inx, sock_inx,
core_inx, use_all_cores)) {
if (sock_inx == sockets)
sock_inx = 0;
continue;
}
if (--(*core_cnt) == 0)
return true;
}
}
} else { /* SLURM_DIST_SOCKCYCLIC */
int task_alloc_cores = 0;
int *next_core = xcalloc(sockets, sizeof(int));
bool nothing_allocated = false;
while (!nothing_allocated) {
nothing_allocated = true;
for (sock_inx = 0; sock_inx < sockets; sock_inx++) {
for (core_inx = next_core[sock_inx];
core_inx < cores; core_inx++) {
next_core[sock_inx] = core_inx + 1;
if (!_pick_step_core(
step_ptr,
job_resrcs_ptr,
avail_core_bitmap,
job_node_inx,
sock_inx,
core_inx,
use_all_cores))
continue;
nothing_allocated = false;
if (--(*core_cnt) == 0) {
xfree(next_core);
return true;
}
if (++task_alloc_cores ==
cores_per_task) {
task_alloc_cores = 0;
break;
}
}
}
}
xfree(next_core);
}
return false;
}
/* Update the step's core bitmaps, create as needed.
* Add the specified task count for a specific node in the job's
* and step's allocation */
static int _pick_step_cores(step_record_t *step_ptr,
job_resources_t *job_resrcs_ptr, int job_node_inx,
uint16_t task_cnt, uint16_t cpus_per_core,
int node_inx, int ntasks_per_core,
int gres_cpus_alloc)
{
uint16_t sockets, cores, cores_per_task, tasks_per_node;
int core_cnt = (int) task_cnt;
bool use_all_cores;
bitstr_t *all_gres_core_bitmap = NULL, *any_gres_core_bitmap = NULL;
xassert(task_cnt);
if (!step_ptr->core_bitmap_job)
step_ptr->core_bitmap_job =
bit_alloc(bit_size(job_resrcs_ptr->core_bitmap));
if (get_job_resources_cnt(job_resrcs_ptr, job_node_inx,
&sockets, &cores))
fatal("get_job_resources_cnt");
if (ntasks_per_core != INFINITE16)
tasks_per_node = cores * ntasks_per_core * sockets;
else
tasks_per_node = cores * cpus_per_core * sockets;
if (((step_ptr->flags & SSF_WHOLE) || task_cnt == (cores * sockets)) &&
(task_cnt <= tasks_per_node || (step_ptr->flags & SSF_OVERCOMMIT)))
{
use_all_cores = true;
core_cnt = ROUNDUP(job_resrcs_ptr->cpus[job_node_inx],
cpus_per_core);
} else {
use_all_cores = false;
if (gres_cpus_alloc) {
core_cnt = ROUNDUP(gres_cpus_alloc, cpus_per_core);
} else if (step_ptr->cpus_per_task > 0) {
core_cnt *= step_ptr->cpus_per_task;
core_cnt = ROUNDUP(core_cnt, cpus_per_core);
}
log_flag(STEPS, "%s: step %pS requires %u cores on node %d with cpus_per_core=%u, available cpus from job: %u",
__func__, step_ptr, core_cnt, job_node_inx,
cpus_per_core, job_resrcs_ptr->cpus[job_node_inx]);
if (core_cnt > ROUNDUP(job_resrcs_ptr->cpus[job_node_inx],
cpus_per_core) &&
!(step_ptr->flags & SSF_OVERCOMMIT)) {
/* Node can never fulfill step request */
return ESLURM_TOO_MANY_REQUESTED_CPUS;
}
}
all_gres_core_bitmap = bit_copy(job_resrcs_ptr->core_bitmap);
any_gres_core_bitmap = bit_copy(job_resrcs_ptr->core_bitmap);
if (step_ptr->gres_list_alloc) {
foreach_gres_filter_t args = {
.all_gres_core_bitmap = all_gres_core_bitmap,
.any_gres_core_bitmap = any_gres_core_bitmap,
.core_start_bit = get_job_resources_offset(
job_resrcs_ptr, job_node_inx, 0, 0),
.core_end_bit = get_job_resources_offset(
job_resrcs_ptr, job_node_inx, sockets - 1,
cores - 1),
.job_node_inx = job_node_inx,
.node_gres_list =
node_record_table_ptr[node_inx]->gres_list,
};
if ((args.core_start_bit > bit_size(all_gres_core_bitmap)) ||
(args.core_end_bit > bit_size(all_gres_core_bitmap)))
error("coremap offsets fall outside core_bitmap size. This should never happen.");
else if (!args.node_gres_list)
error("No node gres when step gres is allocated. This should never happen.");
else {
bit_nclear(any_gres_core_bitmap, args.core_start_bit,
args.core_end_bit);
list_for_each(step_ptr->gres_list_alloc,
_gres_filter_avail_cores, &args);
bit_and(any_gres_core_bitmap,
job_resrcs_ptr->core_bitmap);
}
}
cores_per_task = ROUNDUP(core_cnt, task_cnt); /* Round up */
/* select idle cores that fit all gres binding first */
if (_handle_core_select(step_ptr, job_resrcs_ptr,
all_gres_core_bitmap, job_node_inx,
sockets, cores, use_all_cores, &core_cnt,
cores_per_task))
goto cleanup;
/* select idle cores that fit any gres binding second */
if (!bit_equal(all_gres_core_bitmap, any_gres_core_bitmap) &&
_handle_core_select(step_ptr, job_resrcs_ptr,
any_gres_core_bitmap, job_node_inx,
sockets, cores, use_all_cores, &core_cnt,
cores_per_task))
goto cleanup;
/* select any idle cores */
if (!(step_ptr->job_ptr->bit_flags & GRES_ENFORCE_BIND) &&
!bit_equal(any_gres_core_bitmap, job_resrcs_ptr->core_bitmap)) {
log_flag(STEPS, "gres topology sub-optimal for %ps",
&(step_ptr->step_id));
if (_handle_core_select(step_ptr, job_resrcs_ptr,
job_resrcs_ptr->core_bitmap,
job_node_inx, sockets, cores,
use_all_cores, &core_cnt,
cores_per_task))
goto cleanup;
}
/* The test for cores==0 is just to avoid CLANG errors.
* It should never happen */
if (use_all_cores || (cores == 0))
goto cleanup;
if (!(step_ptr->flags & SSF_OVERCOMMIT)) {
FREE_NULL_BITMAP(all_gres_core_bitmap);
FREE_NULL_BITMAP(any_gres_core_bitmap);
return ESLURM_NODES_BUSY;
}
/* We need to overcommit one or more cores. */
log_flag(STEPS, "%s: %pS needs to overcommit cores. Cores still needed:%u Cores assigned to step:%u exclusive:%c overlap:%c",
__func__, step_ptr, core_cnt,
bit_set_count(step_ptr->core_bitmap_job),
((step_ptr->flags & SSF_EXCLUSIVE) ? 'T' : 'F'),
((step_ptr->flags & SSF_OVERLAP_FORCE) ? 'T' : 'F'));
cleanup:
FREE_NULL_BITMAP(all_gres_core_bitmap);
FREE_NULL_BITMAP(any_gres_core_bitmap);
return SLURM_SUCCESS;
}
static bool _use_one_thread_per_core(step_record_t *step_ptr)
{
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
if ((step_ptr->threads_per_core == 1) ||
((step_ptr->threads_per_core == NO_VAL16) &&
(job_ptr->details->mc_ptr->threads_per_core == 1)) ||
(!(job_resrcs_ptr->whole_node & WHOLE_NODE_REQUIRED) &&
(slurm_conf.select_type_param & (SELECT_CORE | SELECT_SOCKET)) &&
(job_ptr->details &&
(job_ptr->details->cpu_bind_type != NO_VAL16) &&
(job_ptr->details->cpu_bind_type &
CPU_BIND_ONE_THREAD_PER_CORE))))
return true;
return false;
}
static void _modify_cpus_alloc_for_tpc(uint16_t cr_type, uint16_t req_tpc,
uint16_t vpus, int *cpus_alloc)
{
xassert(cpus_alloc);
if ((cr_type & (SELECT_CORE | SELECT_SOCKET | SELECT_LINEAR)) &&
(req_tpc != NO_VAL16) && (req_tpc < vpus)) {
*cpus_alloc = ROUNDUP(*cpus_alloc, req_tpc);
*cpus_alloc *= vpus;
}
}
/* Update a job's record of allocated CPUs when a job step gets scheduled */
static int _step_alloc_lps(step_record_t *step_ptr, char **err_msg)
{
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
node_record_t *node_ptr;
slurm_step_layout_t *step_layout = step_ptr->step_layout;
int cpus_alloc, cpus_alloc_mem, cpu_array_inx = 0;
int job_node_inx = -1, step_node_inx = -1, node_cnt = 0;
bool first_step_node = true, pick_step_cores = true;
bool all_job_mem = false;
uint32_t rem_nodes;
int rc = SLURM_SUCCESS, final_rc = SLURM_SUCCESS;
multi_core_data_t *mc_ptr = job_ptr->details->mc_ptr;
uint16_t orig_cpus_per_task = step_ptr->cpus_per_task;
uint16_t *cpus_per_task_array = NULL;
uint16_t *cpus_alloc_pn = NULL;
uint16_t ntasks_per_core = step_ptr->ntasks_per_core;
uint16_t req_tpc = _get_threads_per_core(step_ptr->threads_per_core,
job_ptr);
xassert(job_resrcs_ptr);
xassert(job_resrcs_ptr->cpus);
xassert(job_resrcs_ptr->cpus_used);
if (!step_layout) /* batch step */
return rc;
if (!bit_set_count(job_resrcs_ptr->node_bitmap))
return rc;
xfree(*err_msg);
xassert(job_resrcs_ptr->core_bitmap);
xassert(job_resrcs_ptr->core_bitmap_used);
if (step_ptr->core_bitmap_job) {
/* "scontrol reconfig" of live system */
pick_step_cores = false;
} else if (!(step_ptr->flags & SSF_OVERCOMMIT) &&
(step_ptr->cpu_count == job_ptr->total_cpus) &&
((ntasks_per_core == mc_ptr->threads_per_core) ||
(ntasks_per_core == INFINITE16))) {
/*
* If the step isn't overcommitting and uses all of job's cores
* Just copy the bitmap to save time
*/
step_ptr->core_bitmap_job = bit_copy(
job_resrcs_ptr->core_bitmap);
pick_step_cores = false;
}
if (step_ptr->pn_min_memory && _is_mem_resv() &&
((job_resrcs_ptr->memory_allocated == NULL) ||
(job_resrcs_ptr->memory_used == NULL))) {
error("%s: lack memory allocation details to enforce memory limits for %pJ",
__func__, job_ptr);
step_ptr->pn_min_memory = 0;
}
if (!step_ptr->pn_min_memory)
all_job_mem = true;
rem_nodes = bit_set_count(step_ptr->step_node_bitmap);
xassert(rem_nodes == step_layout->node_cnt);
cpus_alloc_pn = xcalloc(step_layout->node_cnt, sizeof(*cpus_alloc_pn));
step_ptr->memory_allocated = xcalloc(rem_nodes, sizeof(uint64_t));
for (int i = 0;
(node_ptr = next_node_bitmap(job_resrcs_ptr->node_bitmap, &i));
i++) {
/*
* gres_cpus_alloc - if cpus_per_gres is requested, this is
* cpus_per_gres * gres_alloc on this node
*/
int gres_cpus_alloc = 0;
uint16_t cpus_per_task = orig_cpus_per_task;
uint64_t gres_step_node_mem_alloc = 0;
uint16_t vpus, avail_cpus_per_core, alloc_cpus_per_core;
uint16_t task_cnt;
bitstr_t *unused_core_bitmap;
job_node_inx++;
if (!bit_test(step_ptr->step_node_bitmap, i))
continue;
step_node_inx++;
if (job_node_inx >= job_resrcs_ptr->nhosts)
fatal("%s: node index bad", __func__);
if (!(task_cnt = step_layout->tasks[step_node_inx])) {
/* This should have been caught earlier */
error("Bad step layout: no tasks placed on node %d (%s)",
job_node_inx,
node_ptr->name);
final_rc = ESLURM_BAD_TASK_COUNT;
/*
* Finish allocating resources to all nodes to avoid
* underflow errors in _step_alloc_lps
*/
continue;
}
/*
* NOTE: The --overcommit option can result in
* cpus_used[] having a higher value than cpus[]
*/
/*
* If whole allocate all cpus here instead of just the ones
* requested
*/
if (first_step_node)
step_ptr->cpu_count = 0;
if ((++node_cnt) >
job_resrcs_ptr->cpu_array_reps[cpu_array_inx]) {
cpu_array_inx++;
node_cnt = 0;
}
vpus = node_ptr->tpc;
if (req_tpc != NO_VAL16)
avail_cpus_per_core = req_tpc;
else
avail_cpus_per_core = vpus;
/*
* If the step requested cpus_per_gres, this is mutually
* exclusive with cpus_per_task. We need to calculate total
* gres times cpus_per_gres to get a total cpu count.
*/
unused_core_bitmap = bit_copy(job_resrcs_ptr->core_bitmap);
bit_and_not(unused_core_bitmap,
job_resrcs_ptr->core_bitmap_used);
rc = gres_stepmgr_step_alloc(step_ptr->gres_list_req,
&step_ptr->gres_list_alloc,
job_ptr->gres_list_alloc,
job_node_inx, first_step_node,
task_cnt,
rem_nodes, job_ptr->job_id,
step_ptr->step_id.step_id,
!(step_ptr->flags &
SSF_OVERLAP_FORCE),
&gres_step_node_mem_alloc,
node_ptr->gres_list,
unused_core_bitmap,
&gres_cpus_alloc);
FREE_NULL_BITMAP(unused_core_bitmap);
if (rc != SLURM_SUCCESS) {
log_flag(STEPS, "unable to allocate step GRES for job node %d (%s): %s",
job_node_inx,
node_ptr->name,
slurm_strerror(rc));
/*
* We need to set alloc resources before we continue to
* avoid underflow in _step_dealloc_lps()
*/
final_rc = rc;
}
first_step_node = false;
rem_nodes--;
if (gres_cpus_alloc) {
if (task_cnt > gres_cpus_alloc) {
/*
* Do not error here. If a job requests fewer
* cpus than tasks via cpus_per_gres,
* the job will be allocated one cpu per task.
* Do the same here.
* Use this same logic in _step_dealloc_lps.
*/
cpus_per_task = 1;
log_flag(STEPS, "%s: %pS node %d (%s) gres_cpus_alloc (%d) < tasks (%u), changing gres_cpus_alloc to tasks.",
__func__, step_ptr, job_node_inx,
node_ptr->name, gres_cpus_alloc,
task_cnt);
gres_cpus_alloc = task_cnt;
} else {
cpus_per_task = gres_cpus_alloc / task_cnt;
}
}
/*
* Modify cpus-per-task to request full cores if they can't
* be shared
*/
if ((ntasks_per_core != INFINITE16) && ntasks_per_core) {
alloc_cpus_per_core = avail_cpus_per_core /
ntasks_per_core;
if ((alloc_cpus_per_core > 1) &&
(cpus_per_task % alloc_cpus_per_core)) {
cpus_per_task += alloc_cpus_per_core -
(cpus_per_task % alloc_cpus_per_core);
/*
* Modify gres_cpus_alloc to account for
* ntasks_per_core. If this results in
* requesting more cores than are available,
* then _pick_step_cores() will fail.
*
* Make sure to use this same logic in
* _step_dealloc_lps() to know how many
* cpus were allocated to this step on this
* node.
*/
if (gres_cpus_alloc)
gres_cpus_alloc = task_cnt *
cpus_per_task;
}
}
step_ptr->cpus_per_task = cpus_per_task;
/*
* Only populate cpus_per_task_array if needed: if cpus_per_tres
* was requested, then cpus_per_task may not be the same on all
* nodes. Otherwise, cpus_per_task is the same on all nodes,
* and this per-node array isn't needed.
*/
if (gres_cpus_alloc) {
if (!cpus_per_task_array)
cpus_per_task_array =
xcalloc(step_layout->node_cnt,
sizeof(*cpus_per_task_array));
cpus_per_task_array[step_node_inx] = cpus_per_task;
}
log_flag(STEPS, "%s: %pS node %d (%s) gres_cpus_alloc=%d tasks=%u cpus_per_task=%u",
__func__, step_ptr, job_node_inx, node_ptr->name,
gres_cpus_alloc, task_cnt,
cpus_per_task);
if (step_ptr->flags & SSF_WHOLE) {
cpus_alloc_mem = cpus_alloc =
job_resrcs_ptr->cpus[job_node_inx];
/*
* If we are requesting all the memory in the job
* (--mem=0) we get it all, otherwise we use what was
* requested specifically for the step.
*
* Else factor in the tpc so we get the correct amount
* of memory.
*/
if (all_job_mem)
cpus_alloc_mem =
job_resrcs_ptr->
cpu_array_value[cpu_array_inx];
else if ((req_tpc != NO_VAL16) &&
(req_tpc < vpus)) {
cpus_alloc_mem = ROUNDUP(cpus_alloc_mem, vpus);
cpus_alloc_mem *= req_tpc;
}
} else {
if (gres_cpus_alloc)
cpus_alloc = gres_cpus_alloc;
else
cpus_alloc = task_cnt * cpus_per_task;
/*
* If we are requesting all the memory in the job
* (--mem=0) we get it all, otherwise we use what was
* requested specifically for the step.
*/
if (all_job_mem)
cpus_alloc_mem =
job_resrcs_ptr->
cpu_array_value[cpu_array_inx];
else
cpus_alloc_mem = cpus_alloc;
/*
* If we are doing threads per core we need the whole
* core allocated even though we are only using what was
* requested. Don't worry about cpus_alloc_mem, it's
* already correct.
*/
_modify_cpus_alloc_for_tpc(job_resrcs_ptr->cr_type,
req_tpc, vpus, &cpus_alloc);
/*
* TODO: We need ntasks-per-* sent to the ctld to make
* more decisions on allocation cores.
*/
}
step_ptr->cpu_count += cpus_alloc;
cpus_alloc_pn[step_node_inx] = cpus_alloc;
/*
* Don't count this step against the allocation if
* --overlap=force
*/
if (!(step_ptr->flags & SSF_OVERLAP_FORCE)) {
cpus_alloc = ROUNDUP(cpus_alloc, vpus);
cpus_alloc *= vpus;
if ((job_resrcs_ptr->cr_type & SELECT_CPU) &&
(vpus > 1) &&
(job_resrcs_ptr->cpus_used[job_node_inx] +
cpus_alloc) > job_resrcs_ptr->cpus[job_node_inx])
job_resrcs_ptr->cpus_used[job_node_inx] =
job_resrcs_ptr->cpus[job_node_inx];
else
job_resrcs_ptr->cpus_used[job_node_inx] +=
cpus_alloc;
}
if (!step_ptr->pn_min_memory && !gres_step_node_mem_alloc) {
/* If we aren't requesting memory get it from the job */
step_ptr->pn_min_memory =
job_resrcs_ptr->memory_allocated[job_node_inx];
step_ptr->flags |= SSF_MEM_ZERO;
}
if (step_ptr->pn_min_memory && _is_mem_resv()) {
uint64_t mem_use;
if (step_ptr->pn_min_memory & MEM_PER_CPU) {
mem_use = step_ptr->pn_min_memory;
mem_use &= (~MEM_PER_CPU);
mem_use *= cpus_alloc_mem;
} else if (step_ptr->flags & SSF_MEM_ZERO) {
mem_use = job_resrcs_ptr->
memory_allocated[job_node_inx];
} else {
mem_use = step_ptr->pn_min_memory;
}
step_ptr->memory_allocated[step_node_inx] = mem_use;
/*
* Do not count against the job's memory allocation if
* --mem=0 or --overlap=force were requested.
*/
if (!(step_ptr->flags & SSF_MEM_ZERO) &&
!(step_ptr->flags & SSF_OVERLAP_FORCE))
job_resrcs_ptr->memory_used[job_node_inx] +=
mem_use;
} else if (_is_mem_resv()) {
step_ptr->memory_allocated[step_node_inx] =
gres_step_node_mem_alloc;
/*
* Don't count this step against the allocation if
* --overlap=force
*/
if (!(step_ptr->flags & SSF_OVERLAP_FORCE))
job_resrcs_ptr->memory_used[job_node_inx] +=
gres_step_node_mem_alloc;
}
/*
* Now that we have set cpus and memory used for this node,
* we can check if there was an error, and continue to the
* next node. If any node had an error, we can also skip
* picking cores and skip to the next node.
*
*/
if (final_rc != SLURM_SUCCESS) {
continue;
}
if (pick_step_cores) {
uint16_t cpus_per_core = 1;
/*
* Here we're setting number of CPUs per core
* if we don't enforce 1 thread per core
*
* TODO: move cpus_per_core to slurm_step_layout_t
*/
if (!_use_one_thread_per_core(step_ptr) &&
(!(node_ptr->cpus == node_ptr->tot_cores))) {
if (step_ptr->threads_per_core != NO_VAL16)
cpus_per_core =
step_ptr->threads_per_core;
else if (mc_ptr->threads_per_core != NO_VAL16)
cpus_per_core =
mc_ptr->threads_per_core;
else {
cpus_per_core = node_ptr->threads;
}
}
if ((rc = _pick_step_cores(step_ptr, job_resrcs_ptr,
job_node_inx,
task_cnt,
cpus_per_core, i,
ntasks_per_core,
gres_cpus_alloc))) {
log_flag(STEPS, "unable to pick step cores for job node %d (%s): %s",
job_node_inx,
node_ptr->name,
slurm_strerror(rc));
final_rc = rc;
/* Finish allocating resources to all nodes */
continue;
}
}
if (slurm_conf.debug_flags & DEBUG_FLAG_CPU_BIND)
_dump_step_layout(step_ptr);
if (step_ptr->flags & SSF_OVERLAP_FORCE)
log_flag(STEPS, "step alloc on job node %d (%s); does not count against job allocation",
job_node_inx,
node_ptr->name);
else
log_flag(STEPS, "step alloc on job node %d (%s) used %u of %u CPUs",
job_node_inx,
node_ptr->name,
job_resrcs_ptr->cpus_used[job_node_inx],
job_resrcs_ptr->cpus[job_node_inx]);
if (step_node_inx == (step_layout->node_cnt - 1))
break;
}
slurm_array16_to_value_reps(cpus_per_task_array, step_layout->node_cnt,
&step_layout->cpt_compact_array,
&step_layout->cpt_compact_reps,
&step_layout->cpt_compact_cnt);
xfree(cpus_per_task_array);
slurm_array16_to_value_reps(cpus_alloc_pn, step_layout->node_cnt,
&step_ptr->cpu_alloc_values,
&step_ptr->cpu_alloc_reps,
&step_ptr->cpu_alloc_array_cnt);
xfree(cpus_alloc_pn);
gres_step_state_log(step_ptr->gres_list_req, job_ptr->job_id,
step_ptr->step_id.step_id);
if ((slurm_conf.debug_flags & DEBUG_FLAG_GRES) &&
step_ptr->gres_list_alloc)
info("Step Alloc GRES:");
gres_step_state_log(step_ptr->gres_list_alloc, job_ptr->job_id,
step_ptr->step_id.step_id);
/*
* If we failed to allocate resources on at least one of the nodes, we
* need to deallocate resources.
* Creating a backup of the resources then restoring in case of an
* error does not work - this method leaves cpus allocated to the node
* after the job completes. Instead, we try to allocate resources on
* all nodes in the job even if one of the nodes resulted in a failure.
*/
if (final_rc != SLURM_SUCCESS)
_step_dealloc_lps(step_ptr);
return final_rc;
}
/* Dump a job step's CPU binding information.
* NOTE: The core_bitmap_job and node index are based upon
* the _job_ allocation */
static void _dump_step_layout(step_record_t *step_ptr)
{
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
int i, bit_inx, core_inx, node_inx, rep, sock_inx;
if ((step_ptr->core_bitmap_job == NULL) ||
(job_resrcs_ptr == NULL) ||
(job_resrcs_ptr->cores_per_socket == NULL))
return;
info("====================");
info("%pS", step_ptr);
for (i=0, bit_inx=0, node_inx=0; node_inx<job_resrcs_ptr->nhosts; i++) {
for (rep=0; rep<job_resrcs_ptr->sock_core_rep_count[i]; rep++) {
for (sock_inx=0;
sock_inx<job_resrcs_ptr->sockets_per_node[i];
sock_inx++) {
for (core_inx=0;
core_inx<job_resrcs_ptr->cores_per_socket[i];
core_inx++) {
if (bit_test(step_ptr->
core_bitmap_job,
bit_inx++)) {
info("JobNode[%d] Socket[%d] "
"Core[%d] is allocated",
node_inx, sock_inx,
core_inx);
}
}
}
node_inx++;
}
}
info("====================");
}
static void _step_dealloc_lps(step_record_t *step_ptr)
{
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
int cpus_alloc;
int job_node_inx = -1, step_node_inx = -1;
uint32_t step_id = step_ptr->step_id.step_id;
node_record_t *node_ptr;
xassert(job_resrcs_ptr);
if (!job_resrcs_ptr) {
error("%s: job_resrcs is NULL for %pS; this should never happen",
__func__, step_ptr);
return;
}
xassert(job_resrcs_ptr->cpus);
xassert(job_resrcs_ptr->cpus_used);
/* These special steps do not allocate any resources */
if ((step_id == SLURM_EXTERN_CONT) ||
(step_id == SLURM_BATCH_SCRIPT) ||
(step_id == SLURM_INTERACTIVE_STEP) ||
(step_ptr->flags & SSF_EXT_LAUNCHER)) {
log_flag(STEPS, "Skip %s for %pS", __func__, step_ptr);
return;
}
if (!bit_set_count(job_resrcs_ptr->node_bitmap))
return;
if (step_ptr->memory_allocated && _is_mem_resv() &&
((job_resrcs_ptr->memory_allocated == NULL) ||
(job_resrcs_ptr->memory_used == NULL))) {
error("%s: lack memory allocation details to enforce memory limits for %pJ",
__func__, job_ptr);
}
for (int i = 0;
(node_ptr = next_node_bitmap(job_resrcs_ptr->node_bitmap, &i));
i++) {
int inx;
uint16_t vpus = node_ptr->tpc;
job_node_inx++;
if (!bit_test(step_ptr->step_node_bitmap, i))
continue;
step_node_inx++;
if (job_node_inx >= job_resrcs_ptr->nhosts)
fatal("_step_dealloc_lps: node index bad");
/*
* We need to free GRES structures regardless of overlap.
*/
gres_stepmgr_step_dealloc(step_ptr->gres_list_alloc,
job_ptr->gres_list_alloc, job_ptr->job_id,
step_ptr->step_id.step_id,
job_node_inx,
!(step_ptr->flags & SSF_OVERLAP_FORCE));
if (step_ptr->flags & SSF_OVERLAP_FORCE) {
log_flag(STEPS, "step dealloc on job node %d (%s); did not count against job allocation",
job_node_inx,
node_ptr->name);
continue; /* Next node */
}
/*
* If zero tasks, then _step_alloc_lps() error'd and did not
* allocate any resources, so we should not deallocate anything.
*/
if (!step_ptr->step_layout->tasks[step_node_inx])
continue;
xassert(step_ptr->cpu_alloc_array_cnt);
xassert(step_ptr->cpu_alloc_reps);
xassert(step_ptr->cpu_alloc_values);
inx = slurm_get_rep_count_inx(
step_ptr->cpu_alloc_reps,
step_ptr->cpu_alloc_array_cnt,
step_node_inx);
cpus_alloc = ROUNDUP(step_ptr->cpu_alloc_values[inx], vpus);
cpus_alloc *= vpus;
if ((job_resrcs_ptr->cr_type & SELECT_CPU) &&
(node_ptr->tpc > 1)) {
int core_alloc = ROUNDUP(cpus_alloc, vpus);
int used_cores =
ROUNDUP(job_resrcs_ptr->cpus_used[job_node_inx],
vpus);
/*
* If SELECT_CPU is used with a thread count > 1 the
* cpus recorded being allocated to a job don't have to
* be a multiple of threads per core. Make sure to
* dealloc full cores and not partial cores.
*/
if (used_cores >= core_alloc) {
used_cores -= core_alloc;
job_resrcs_ptr->cpus_used[job_node_inx] =
MIN(used_cores * vpus,
job_resrcs_ptr->cpus[job_node_inx]);
} else {
error("%s: CPU underflow for %pS (%u<%u on job node %d)",
__func__, step_ptr, used_cores * vpus,
core_alloc * vpus, job_node_inx);
job_resrcs_ptr->cpus_used[job_node_inx] = 0;
}
} else if (job_resrcs_ptr->cpus_used[job_node_inx] >=
cpus_alloc) {
job_resrcs_ptr->cpus_used[job_node_inx] -= cpus_alloc;
} else {
error("%s: CPU underflow for %pS (%u<%u on job node %d)",
__func__, step_ptr,
job_resrcs_ptr->cpus_used[job_node_inx],
cpus_alloc, job_node_inx);
job_resrcs_ptr->cpus_used[job_node_inx] = 0;
}
if (step_ptr->memory_allocated && _is_mem_resv() &&
!(step_ptr->flags & SSF_MEM_ZERO)) {
uint64_t mem_use =
step_ptr->memory_allocated[step_node_inx];
if (job_resrcs_ptr->memory_used[job_node_inx] >=
mem_use) {
job_resrcs_ptr->memory_used[job_node_inx] -=
mem_use;
log_flag(STEPS, "Deallocating %"PRIu64"MB of memory on node %d (%s) now used: %"PRIu64" of %"PRIu64,
mem_use,
job_node_inx,
node_ptr->name,
job_resrcs_ptr->
memory_used[job_node_inx],
job_resrcs_ptr->
memory_allocated[job_node_inx]);
} else {
error("%s: Allocated memory underflow for %pS (freed memory=%"PRIu64")",
__func__, step_ptr, mem_use);
job_resrcs_ptr->memory_used[job_node_inx] = 0;
}
}
log_flag(STEPS, "step dealloc on job node %d (%s) used: %u of %u CPUs",
job_node_inx, node_ptr->name,
job_resrcs_ptr->cpus_used[job_node_inx],
job_resrcs_ptr->cpus[job_node_inx]);
if (step_node_inx == (step_ptr->step_layout->node_cnt - 1))
break;
}
xassert(job_resrcs_ptr->core_bitmap);
xassert(job_resrcs_ptr->core_bitmap_used);
if (step_ptr->core_bitmap_job) {
/* Mark the job's cores as no longer in use */
int job_core_size, step_core_size;
job_core_size = bit_size(job_resrcs_ptr->core_bitmap_used);
step_core_size = bit_size(step_ptr->core_bitmap_job);
/*
* Don't remove step's used cores from job core_bitmap_used if
* SSF_OVERLAP_FORCE
*/
if (job_core_size == step_core_size) {
if (!(step_ptr->flags & SSF_OVERLAP_FORCE))
bit_and_not(job_resrcs_ptr->core_bitmap_used,
step_ptr->core_bitmap_job);
} else {
error("%s: %pS core_bitmap size mismatch (%d != %d)",
__func__, step_ptr, job_core_size,
step_core_size);
}
FREE_NULL_BITMAP(step_ptr->core_bitmap_job);
}
}
static int _test_strlen(char *test_str, char *str_name, int max_str_len)
{
int i = 0;
if (test_str)
i = strlen(test_str);
if (i > max_str_len) {
info("step_create_request: strlen(%s) too big (%d > %d)",
str_name, i, max_str_len);
return ESLURM_PATHNAME_TOO_LONG;
}
return SLURM_SUCCESS;
}
/* Calculate a step's cpus_per_task value. Set to zero if we can't distributed
* the tasks evenly over the nodes (heterogeneous job allocation). */
static int _calc_cpus_per_task(job_step_create_request_msg_t *step_specs,
job_record_t *job_ptr)
{
int cpus_per_task = 0, i;
int num_tasks;
char *cpt = NULL;
if ((cpt = xstrstr(step_specs->tres_per_task, "cpu:"))) {
cpus_per_task = slurm_atoul(cpt + 4);
if (cpus_per_task < 0)
cpus_per_task = 0;
return cpus_per_task;
}
if (step_specs->cpus_per_tres)
return 0;
if (step_specs->num_tasks == NO_VAL)
return 0;
if ((step_specs->cpu_count == 0) ||
(step_specs->cpu_count % step_specs->num_tasks))
return cpus_per_task;
cpus_per_task = step_specs->cpu_count / step_specs->num_tasks;
if (cpus_per_task < 1)
cpus_per_task = 1;
if (!job_ptr->job_resrcs)
return cpus_per_task;
num_tasks = step_specs->num_tasks;
for (i = 0; i < job_ptr->job_resrcs->cpu_array_cnt; i++) {
if (cpus_per_task > job_ptr->job_resrcs->cpu_array_value[i]) {
cpus_per_task = 0;
break;
}
num_tasks -= (job_ptr->job_resrcs->cpu_array_value[i] /
cpus_per_task) *
job_ptr->job_resrcs->cpu_array_reps[i];
}
if (num_tasks > 0)
return 0;
return cpus_per_task;
}
/*
* Set a job's default cpu_bind_type based upon configuration of allocated nodes,
* partition or global TaskPluginParams
*/
static void _set_def_cpu_bind(job_record_t *job_ptr)
{
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
node_record_t *node_ptr;
uint32_t bind_bits, bind_to_bits, node_bind = NO_VAL;
bool node_fail = false;
if (!job_ptr->details || !job_resrcs_ptr ||
!job_resrcs_ptr->node_bitmap)
return; /* No data structure */
bind_to_bits = CPU_BIND_TO_SOCKETS | CPU_BIND_TO_CORES |
CPU_BIND_TO_THREADS | CPU_BIND_TO_LDOMS;
if ((job_ptr->details->cpu_bind_type != NO_VAL16) &&
(job_ptr->details->cpu_bind_type & bind_to_bits)) {
if (slurm_conf.debug_flags & DEBUG_FLAG_CPU_BIND) {
char tmp_str[128];
slurm_sprint_cpu_bind_type(
tmp_str, job_ptr->details->cpu_bind_type);
log_flag(CPU_BIND, "%pJ CpuBind='%s' already set for job/allocation using it as a default for new step.",
job_ptr, tmp_str);
}
return; /* Already set */
}
bind_bits = job_ptr->details->cpu_bind_type & CPU_BIND_VERBOSE;
/*
* Set job's cpu_bind to the node's cpu_bind if all of the job's
* allocated nodes have the same cpu_bind (or it is not set)
*/
for (int i = 0;
(node_ptr = next_node_bitmap(job_resrcs_ptr->node_bitmap, &i));
i++) {
if (node_bind == NO_VAL) {
if (node_ptr->cpu_bind != 0)
node_bind = node_ptr->cpu_bind;
} else if ((node_ptr->cpu_bind != 0) &&
(node_bind != node_ptr->cpu_bind)) {
node_fail = true;
break;
}
}
if (!node_fail && (node_bind != NO_VAL)) {
job_ptr->details->cpu_bind_type = bind_bits | node_bind;
if (slurm_conf.debug_flags & DEBUG_FLAG_CPU_BIND) {
char tmp_str[128];
slurm_sprint_cpu_bind_type(
tmp_str, job_ptr->details->cpu_bind_type);
log_flag(CPU_BIND, "%pJ setting default CpuBind to nodes default '%s' for new step.",
job_ptr, tmp_str);
}
return;
}
/* Use partition's cpu_bind (if any) */
if (job_ptr->part_ptr && job_ptr->part_ptr->cpu_bind) {
job_ptr->details->cpu_bind_type = bind_bits |
job_ptr->part_ptr->cpu_bind;
if (slurm_conf.debug_flags & DEBUG_FLAG_CPU_BIND) {
char tmp_str[128];
slurm_sprint_cpu_bind_type(
tmp_str, job_ptr->details->cpu_bind_type);
log_flag(CPU_BIND, "%pJ setting default CpuBind to partition default '%s' for new step.",
job_ptr, tmp_str);
}
return;
}
/* Use global default from TaskPluginParams */
job_ptr->details->cpu_bind_type = bind_bits |
slurm_conf.task_plugin_param;
if (slurm_conf.debug_flags & DEBUG_FLAG_CPU_BIND) {
char tmp_str[128];
slurm_sprint_cpu_bind_type(tmp_str,
job_ptr->details->cpu_bind_type);
log_flag(CPU_BIND, "%pJ setting default CpuBind to TaskPluginParam '%s' for new step.",
job_ptr, tmp_str);
}
}
/*
* A step may explicitly set a TRES count to zero in order to avoid making use
* of the job's TRES specifications. At this point, clear the records with
* zero counts.
*/
static void _clear_zero_tres(char **tres_spec)
{
char *new_spec = NULL, *new_sep = "";
char *tmp, *tok, *sep, *end_ptr = NULL, *save_ptr = NULL;
long int cnt;
if (*tres_spec == NULL)
return;
tmp = xstrdup(*tres_spec);
tok = strtok_r(tmp, ",", &save_ptr);
while (tok) {
bool copy_rec = true;
sep = strrchr(tok, ':');
if (sep) {
cnt = strtoll(sep+1, &end_ptr, 10);
if ((cnt == 0) && (end_ptr[0] == '\0'))
copy_rec = false;
}
if (copy_rec) {
xstrfmtcat(new_spec, "%s%s", new_sep, tok);
new_sep = ",";
}
tok = strtok_r(NULL, ",", &save_ptr);
}
xfree(tmp);
xfree(*tres_spec);
*tres_spec = new_spec;
}
/*
* A step may explicitly request --gres=none in order to avoid making use
* of the job's TRES specifications. At this point, clear all GRES records.
*/
static void _clear_gres_tres(char **tres_spec)
{
char *new_spec = NULL, *new_sep = "";
char *tmp, *tok, *save_ptr = NULL;
if (*tres_spec == NULL)
return;
tmp = xstrdup(*tres_spec);
tok = strtok_r(tmp, ",", &save_ptr);
while (tok) {
if (xstrncmp(tok, "gres", 4)) {
xstrfmtcat(new_spec, "%s%s", new_sep, tok);
new_sep = ",";
}
tok = strtok_r(NULL, ",", &save_ptr);
}
xfree(tmp);
xfree(*tres_spec);
*tres_spec = new_spec;
}
/*
* If a job step specification does not include any GRES specification,
* then copy those values from the job record.
* Currently we only want to check if the step lacks a "gres" request.
* "tres_per_[step|task]" has "cpu:<count>" in it, so we need to search for
* "gres" in the strings.
*/
static void _copy_job_tres_to_step(job_step_create_request_msg_t *step_specs,
job_record_t *job_ptr)
{
if (!xstrcasecmp(step_specs->tres_per_node, "NONE")) {
xfree(step_specs->tres_per_node);
_clear_gres_tres(&step_specs->tres_per_step);
_clear_gres_tres(&step_specs->tres_per_socket);
_clear_gres_tres(&step_specs->tres_per_task);
} else if (xstrstr(step_specs->tres_per_step, "gres") ||
xstrstr(step_specs->tres_per_node, "gres") ||
xstrstr(step_specs->tres_per_socket, "gres") ||
xstrstr(step_specs->tres_per_task, "gres")) {
_clear_zero_tres(&step_specs->tres_per_step);
_clear_zero_tres(&step_specs->tres_per_node);
_clear_zero_tres(&step_specs->tres_per_socket);
_clear_zero_tres(&step_specs->tres_per_task);
} else {
xfree(step_specs->tres_per_step);
xfree(step_specs->tres_per_node);
xfree(step_specs->tres_per_socket);
xfree(step_specs->tres_per_task);
step_specs->tres_per_step = xstrdup(job_ptr->tres_per_job);
step_specs->tres_per_node = xstrdup(job_ptr->tres_per_node);
step_specs->tres_per_socket = xstrdup(job_ptr->tres_per_socket);
step_specs->tres_per_task = xstrdup(job_ptr->tres_per_task);
}
}
static int _test_step_desc_fields(job_step_create_request_msg_t *step_specs)
{
static time_t sched_update = 0;
static int max_submit_line = DEFAULT_MAX_SUBMIT_LINE_SIZE;
if (sched_update != slurm_conf.last_update) {
char *tmp_ptr;
sched_update = slurm_conf.last_update;
if ((tmp_ptr = xstrcasestr(slurm_conf.sched_params,
"max_submit_line_size="))) {
max_submit_line = atoi(tmp_ptr + 21);
} else {
max_submit_line = DEFAULT_MAX_SUBMIT_LINE_SIZE;
}
}
if (_test_strlen(step_specs->host, "host", 1024) ||
_test_strlen(step_specs->name, "name", 1024) ||
_test_strlen(step_specs->network, "network", 1024) ||
_test_strlen(step_specs->submit_line, "submit_line",
max_submit_line))
return ESLURM_PATHNAME_TOO_LONG;
return SLURM_SUCCESS;
}
static int _switch_setup(step_record_t *step_ptr)
{
xassert(step_ptr);
if (!step_ptr->step_layout)
return SLURM_SUCCESS;
errno = 0;
if (switch_g_build_stepinfo(&step_ptr->switch_step,
step_ptr->step_layout,
step_ptr) < 0) {
if (errno == ESLURM_INTERCONNECT_BUSY)
return errno;
return ESLURM_INTERCONNECT_FAILURE;
}
return SLURM_SUCCESS;
}
extern int step_create(job_record_t *job_ptr,
job_step_create_request_msg_t *step_specs,
step_record_t** new_step_record,
uint16_t protocol_version, char **err_msg)
{
step_record_t *step_ptr;
bitstr_t *nodeset;
int cpus_per_task, ret_code, i;
uint32_t node_count = 0;
time_t now = time(NULL);
char *step_node_list = NULL;
uint32_t orig_cpu_count;
list_t *step_gres_list = NULL;
uint32_t task_dist;
uint32_t max_tasks;
uint32_t over_time_limit;
bool resv_ports_present = false;
*new_step_record = NULL;
xassert(job_ptr);
/*
* NOTE: We have already confirmed the UID originating
* the request is identical with step_specs->user_id
*/
if (step_specs->user_id != job_ptr->user_id)
return ESLURM_ACCESS_DENIED ;
if (step_specs->step_id.step_id != NO_VAL) {
if (list_delete_first(job_ptr->step_list,
_purge_duplicate_steps,
step_specs) < 0)
return ESLURM_DUPLICATE_STEP_ID;
}
if ((job_ptr->details == NULL) || IS_JOB_SUSPENDED(job_ptr))
return ESLURM_DISABLED;
if (IS_JOB_PENDING(job_ptr)) {
/* NOTE: LSF creates a job allocation for batch jobs.
* After the allocation has been made, LSF submits a
* job to run in that allocation (sbatch --jobid= ...).
* If that job is pending either LSF messed up or LSF is
* not being used. We have seen this problem with Moab. */
return ESLURM_DUPLICATE_JOB_ID;
}
/* Get OverTimeLimit from job's partition if set, or globally. */
if (job_ptr->part_ptr &&
(job_ptr->part_ptr->over_time_limit != NO_VAL16))
over_time_limit = job_ptr->part_ptr->over_time_limit;
else
over_time_limit = slurm_conf.over_time_limit;
if (over_time_limit == INFINITE16)
over_time_limit = YEAR_MINUTES;
if (IS_JOB_FINISHED(job_ptr) ||
(((job_ptr->end_time + (over_time_limit * 60)) <= time(NULL)) &&
!IS_JOB_CONFIGURING(job_ptr)))
return ESLURM_ALREADY_DONE;
if (job_ptr->details->prolog_running)
return ESLURM_PROLOG_RUNNING;
if (step_specs->flags & SSF_INTERACTIVE) {
debug("%s: interactive step requested", __func__);
*new_step_record = _build_interactive_step(job_ptr, step_specs,
protocol_version);
if (*new_step_record)
return SLURM_SUCCESS;
else
return ESLURM_DUPLICATE_STEP_ID;
}
if (step_specs->flags & SSF_EXT_LAUNCHER) {
debug("%s: external launcher step requested", __func__);
return _build_ext_launcher_step(new_step_record, job_ptr,
step_specs, protocol_version);
}
/* A step cannot request more threads per core than its allocation. */
if ((step_specs->threads_per_core != NO_VAL16) &&
(step_specs->threads_per_core >
job_ptr->job_resrcs->threads_per_core))
return ESLURM_BAD_THREAD_PER_CORE;
task_dist = step_specs->task_dist & SLURM_DIST_STATE_BASE;
/* Set to block in the case that mem is 0. srun leaves the dist
* set to unknown if mem is 0. */
if ((task_dist == SLURM_DIST_UNKNOWN) &&
(!(step_specs->pn_min_memory &(~MEM_PER_CPU)))) {
step_specs->task_dist &= SLURM_DIST_STATE_FLAGS;
step_specs->task_dist |= SLURM_DIST_BLOCK;
task_dist = SLURM_DIST_BLOCK;
}
if ((task_dist != SLURM_DIST_CYCLIC) &&
(task_dist != SLURM_DIST_BLOCK) &&
(task_dist != SLURM_DIST_CYCLIC_CYCLIC) &&
(task_dist != SLURM_DIST_BLOCK_CYCLIC) &&
(task_dist != SLURM_DIST_CYCLIC_BLOCK) &&
(task_dist != SLURM_DIST_BLOCK_BLOCK) &&
(task_dist != SLURM_DIST_CYCLIC_CFULL) &&
(task_dist != SLURM_DIST_BLOCK_CFULL) &&
(task_dist != SLURM_DIST_CYCLIC_CYCLIC_CYCLIC) &&
(task_dist != SLURM_DIST_CYCLIC_CYCLIC_BLOCK) &&
(task_dist != SLURM_DIST_CYCLIC_CYCLIC_CFULL) &&
(task_dist != SLURM_DIST_CYCLIC_BLOCK_CYCLIC) &&
(task_dist != SLURM_DIST_CYCLIC_BLOCK_BLOCK) &&
(task_dist != SLURM_DIST_CYCLIC_BLOCK_CFULL) &&
(task_dist != SLURM_DIST_CYCLIC_CFULL_CYCLIC) &&
(task_dist != SLURM_DIST_CYCLIC_CFULL_BLOCK) &&
(task_dist != SLURM_DIST_CYCLIC_CFULL_CFULL) &&
(task_dist != SLURM_DIST_BLOCK_CYCLIC_CYCLIC) &&
(task_dist != SLURM_DIST_BLOCK_CYCLIC_BLOCK) &&
(task_dist != SLURM_DIST_BLOCK_CYCLIC_CFULL) &&
(task_dist != SLURM_DIST_BLOCK_BLOCK_CYCLIC) &&
(task_dist != SLURM_DIST_BLOCK_BLOCK_BLOCK) &&
(task_dist != SLURM_DIST_BLOCK_BLOCK_CFULL) &&
(task_dist != SLURM_DIST_BLOCK_CFULL_CYCLIC) &&
(task_dist != SLURM_DIST_BLOCK_CFULL_BLOCK) &&
(task_dist != SLURM_DIST_BLOCK_CFULL_CFULL) &&
(task_dist != SLURM_DIST_PLANE) &&
(task_dist != SLURM_DIST_ARBITRARY))
return ESLURM_BAD_DIST;
if (!assoc_mgr_valid_tres_cnt(step_specs->cpus_per_tres, 0) ||
!assoc_mgr_valid_tres_cnt(step_specs->mem_per_tres, 0) ||
tres_bind_verify_cmdline(step_specs->tres_bind) ||
tres_freq_verify_cmdline(step_specs->tres_freq) ||
!assoc_mgr_valid_tres_cnt(step_specs->tres_per_step, 0) ||
(!assoc_mgr_valid_tres_cnt(step_specs->tres_per_node, 0) &&
xstrcasecmp(step_specs->tres_per_node, "NONE")) ||
!assoc_mgr_valid_tres_cnt(step_specs->tres_per_socket, 0) ||
!assoc_mgr_valid_tres_cnt(step_specs->tres_per_task, 0))
return ESLURM_INVALID_TRES;
if ((ret_code = _test_step_desc_fields(step_specs)) != SLURM_SUCCESS)
return ret_code;
if (job_ptr->next_step_id >= slurm_conf.max_step_cnt)
return ESLURM_STEP_LIMIT;
/*
* If the overcommit flag is checked, we set cpu_count=0
* which makes it so we don't check to see the available cpus
*/
orig_cpu_count = step_specs->cpu_count;
if (step_specs->flags & SSF_OVERCOMMIT)
step_specs->cpu_count = 0;
if (!step_specs->ntasks_per_tres)
step_specs->ntasks_per_tres = NO_VAL16;
/* determine cpus_per_task value by reversing what srun does */
if (step_specs->num_tasks < 1)
return ESLURM_BAD_TASK_COUNT;
cpus_per_task = _calc_cpus_per_task(step_specs, job_ptr);
_copy_job_tres_to_step(step_specs, job_ptr);
/* If whole is given we probably need to copy tres_per_* from the job */
i = gres_step_state_validate(step_specs->cpus_per_tres,
step_specs->tres_per_step,
step_specs->tres_per_node,
step_specs->tres_per_socket,
step_specs->tres_per_task,
step_specs->mem_per_tres,
step_specs->ntasks_per_tres,
step_specs->min_nodes,
&step_gres_list,
job_ptr->job_id,
NO_VAL, &step_specs->num_tasks,
&step_specs->cpu_count, err_msg);
if (i != SLURM_SUCCESS) {
FREE_NULL_LIST(step_gres_list);
return i;
}
job_ptr->time_last_active = now;
nodeset = _pick_step_nodes(job_ptr, step_specs, step_gres_list,
cpus_per_task, node_count, &ret_code);
if (nodeset == NULL) {
FREE_NULL_LIST(step_gres_list);
if ((ret_code == ESLURM_NODES_BUSY) ||
(ret_code == ESLURM_PORTS_BUSY) ||
(ret_code == ESLURM_INTERCONNECT_BUSY))
_build_pending_step(job_ptr, step_specs);
return ret_code;
}
_set_def_cpu_bind(job_ptr);
node_count = bit_set_count(nodeset);
xassert(step_specs->num_tasks != NO_VAL);
max_tasks = node_count * slurm_conf.max_tasks_per_node;
if (step_specs->num_tasks > max_tasks) {
error("step has invalid task count: %u max is %u",
step_specs->num_tasks, max_tasks);
FREE_NULL_LIST(step_gres_list);
FREE_NULL_BITMAP(nodeset);
return ESLURM_BAD_TASK_COUNT;
}
step_ptr = create_step_record(job_ptr, protocol_version);
if (step_ptr == NULL) {
FREE_NULL_LIST(step_gres_list);
FREE_NULL_BITMAP(nodeset);
return ESLURMD_TOOMANYSTEPS;
}
*stepmgr_ops->last_job_update = time(NULL);
step_ptr->start_time = time(NULL);
step_ptr->state = JOB_RUNNING;
memcpy(&step_ptr->step_id, &step_specs->step_id,
sizeof(step_ptr->step_id));
if (step_specs->array_task_id != NO_VAL)
step_ptr->step_id.job_id = job_ptr->job_id;
if (step_specs->step_id.step_id != NO_VAL) {
if (step_specs->step_id.step_het_comp == NO_VAL) {
job_ptr->next_step_id =
MAX(job_ptr->next_step_id,
step_specs->step_id.step_id);
job_ptr->next_step_id++;
}
} else if (job_ptr->het_job_id &&
(job_ptr->het_job_id != job_ptr->job_id)) {
job_record_t *het_job;
het_job = stepmgr_ops->find_job_record(job_ptr->het_job_id);
if (het_job)
step_ptr->step_id.step_id = het_job->next_step_id++;
else
step_ptr->step_id.step_id = job_ptr->next_step_id++;
job_ptr->next_step_id = MAX(job_ptr->next_step_id,
step_ptr->step_id.step_id);
} else {
step_ptr->step_id.step_id = job_ptr->next_step_id++;
}
/* Here is where the node list is set for the step */
if (step_specs->node_list &&
((step_specs->task_dist & SLURM_DIST_STATE_BASE) ==
SLURM_DIST_ARBITRARY)) {
step_node_list = xstrdup(step_specs->node_list);
xfree(step_specs->node_list);
step_specs->node_list = bitmap2node_name(nodeset);
} else {
step_node_list = bitmap2node_name_sortable(nodeset, false);
xfree(step_specs->node_list);
step_specs->node_list = xstrdup(step_node_list);
}
log_flag(STEPS, "Picked nodes %s when accumulating from %s",
step_node_list, step_specs->node_list);
step_ptr->step_node_bitmap = nodeset;
switch (step_specs->task_dist & SLURM_DIST_NODESOCKMASK) {
case SLURM_DIST_CYCLIC:
case SLURM_DIST_CYCLIC_CYCLIC:
case SLURM_DIST_CYCLIC_CFULL:
case SLURM_DIST_CYCLIC_BLOCK:
step_ptr->cyclic_alloc = 1;
break;
default:
step_ptr->cyclic_alloc = 0;
break;
}
step_ptr->container = xstrdup(step_specs->container);
step_ptr->container_id = xstrdup(step_specs->container_id);
step_ptr->gres_list_req = step_gres_list;
step_gres_list = NULL;
gres_step_state_log(step_ptr->gres_list_req, job_ptr->job_id,
step_ptr->step_id.step_id);
if ((slurm_conf.debug_flags & DEBUG_FLAG_GRES) &&
step_ptr->gres_list_alloc)
info("Step Alloc GRES:");
gres_step_state_log(step_ptr->gres_list_alloc, job_ptr->job_id,
step_ptr->step_id.step_id);
step_ptr->port = step_specs->port;
step_ptr->srun_pid = step_specs->srun_pid;
step_ptr->host = xstrdup(step_specs->host);
if ((step_specs->cpu_freq_min == NO_VAL) &&
(step_specs->cpu_freq_max == NO_VAL) &&
(step_specs->cpu_freq_gov == NO_VAL)) {
step_ptr->cpu_freq_min = job_ptr->details->cpu_freq_min;
step_ptr->cpu_freq_max = job_ptr->details->cpu_freq_max;
step_ptr->cpu_freq_gov = job_ptr->details->cpu_freq_gov;
} else {
step_ptr->cpu_freq_min = step_specs->cpu_freq_min;
step_ptr->cpu_freq_max = step_specs->cpu_freq_max;
step_ptr->cpu_freq_gov = step_specs->cpu_freq_gov;
}
step_ptr->cpus_per_task = (uint16_t)cpus_per_task;
step_ptr->ntasks_per_core = step_specs->ntasks_per_core;
step_ptr->pn_min_memory = step_specs->pn_min_memory;
/*
* cpu_count can be updated by gres_step_state_validate() if OVERCOMMIT
* is not used. If so, use the updated value.
*/
if (step_specs->flags & SSF_OVERCOMMIT)
step_ptr->cpu_count = orig_cpu_count;
else
step_ptr->cpu_count = step_specs->cpu_count;
step_ptr->exit_code = NO_VAL;
step_ptr->flags = step_specs->flags;
step_ptr->cpus_per_tres = xstrdup(step_specs->cpus_per_tres);
step_ptr->mem_per_tres = xstrdup(step_specs->mem_per_tres);
step_ptr->cwd = xstrdup(step_specs->cwd);
step_ptr->std_err = xstrdup(step_specs->std_err);
step_ptr->std_in = xstrdup(step_specs->std_in);
step_ptr->std_out = xstrdup(step_specs->std_out);
step_ptr->submit_line = xstrdup(step_specs->submit_line);
step_ptr->tres_bind = xstrdup(step_specs->tres_bind);
step_ptr->tres_freq = xstrdup(step_specs->tres_freq);
step_ptr->tres_per_step = xstrdup(step_specs->tres_per_step);
step_ptr->tres_per_node = xstrdup(step_specs->tres_per_node);
step_ptr->tres_per_socket = xstrdup(step_specs->tres_per_socket);
step_ptr->tres_per_task = xstrdup(step_specs->tres_per_task);
step_ptr->threads_per_core = step_specs->threads_per_core;
/*
* step's name and network default to job's values if not
* specified in the step specification
*/
if (step_specs->name && step_specs->name[0])
step_ptr->name = xstrdup(step_specs->name);
else
step_ptr->name = xstrdup(job_ptr->name);
if (step_specs->network && step_specs->network[0])
step_ptr->network = xstrdup(step_specs->network);
else
step_ptr->network = xstrdup(job_ptr->network);
/*
* the step time_limit is recorded as submitted (INFINITE
* or partition->max_time by default), but the allocation
* time limits may cut it short
*/
if (step_specs->time_limit == NO_VAL || step_specs->time_limit == 0 ||
step_specs->time_limit == INFINITE) {
step_ptr->time_limit = INFINITE;
} else {
/* enforce partition limits if necessary */
if ((step_specs->time_limit > job_ptr->part_ptr->max_time) &&
slurm_conf.enforce_part_limits) {
info("%s: %pS time greater than partition's (%u > %u)",
__func__, step_ptr, step_specs->time_limit,
job_ptr->part_ptr->max_time);
delete_step_record(job_ptr, step_ptr);
xfree(step_node_list);
return ESLURM_INVALID_TIME_LIMIT;
}
step_ptr->time_limit = step_specs->time_limit;
}
step_ptr->step_layout =
step_layout_create(step_ptr,
step_node_list, node_count,
step_specs->num_tasks,
(uint16_t)cpus_per_task,
step_specs->task_dist,
step_specs->plane_size);
xfree(step_node_list);
if (!step_ptr->step_layout) {
delete_step_record(job_ptr, step_ptr);
if (step_specs->pn_min_memory)
return ESLURM_INVALID_TASK_MEMORY;
return SLURM_ERROR;
}
if (slurm_conf.mpi_params && xstrstr(slurm_conf.mpi_params, "ports="))
resv_ports_present = true;
if ((step_specs->resv_port_cnt == NO_VAL16) &&
(resv_ports_present || job_ptr->resv_ports)) {
step_specs->resv_port_cnt = 0;
/*
* reserved port count set to maximum task count on
* any node plus one
*/
for (i = 0; i < step_ptr->step_layout->node_cnt; i++) {
step_specs->resv_port_cnt =
MAX(step_specs->resv_port_cnt,
step_ptr->step_layout->tasks[i]);
}
step_specs->resv_port_cnt++;
}
if ((step_specs->resv_port_cnt != NO_VAL16) &&
(step_specs->resv_port_cnt != 0)) {
step_ptr->resv_port_cnt = step_specs->resv_port_cnt;
i = resv_port_step_alloc(step_ptr);
if (i != SLURM_SUCCESS) {
delete_step_record(job_ptr, step_ptr);
return i;
}
}
if ((ret_code = _switch_setup(step_ptr))) {
delete_step_record(job_ptr, step_ptr);
return ret_code;
}
if ((ret_code = _step_alloc_lps(step_ptr, err_msg))) {
delete_step_record(job_ptr, step_ptr);
return ret_code;
}
xassert(bit_set_count(step_ptr->core_bitmap_job) != 0);
*new_step_record = step_ptr;
step_set_alloc_tres(step_ptr, node_count, false, true);
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
return SLURM_SUCCESS;
}
extern slurm_step_layout_t *step_layout_create(step_record_t *step_ptr,
char *step_node_list,
uint32_t node_count,
uint32_t num_tasks,
uint16_t cpus_per_task,
uint32_t task_dist,
uint16_t plane_size)
{
slurm_step_layout_t *step_layout = NULL;
uint16_t cpus_per_node[node_count];
uint16_t cpus_per_task_array[node_count];
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
slurm_step_layout_req_t step_layout_req;
uint64_t gres_cpus;
int cpu_inx = -1, cpus_task_inx = -1;
int usable_cpus, usable_mem;
int set_nodes = 0/* , set_tasks = 0 */;
int pos = -1;
uint32_t cpu_count_reps[node_count];
uint32_t cpus_task_reps[node_count];
uint32_t cpus_task = 0;
uint16_t ntasks_per_core = step_ptr->ntasks_per_core;
uint16_t ntasks_per_socket = 0;
node_record_t *node_ptr;
gres_stepmgr_step_test_args_t gres_test_args = {
.cpus_per_task = step_ptr->cpus_per_task,
.first_step_node = true,
.job_gres_list = job_ptr->gres_list_alloc,
.job_id = job_ptr->job_id,
.job_resrcs_ptr = job_resrcs_ptr,
.node_offset = -1,
.step_gres_list = step_ptr->gres_list_req,
.step_id = step_ptr->step_id.step_id,
.test_mem = false,
};
xassert(job_resrcs_ptr);
xassert(job_resrcs_ptr->cpus);
xassert(job_resrcs_ptr->cpus_used);
if (step_ptr->pn_min_memory && _is_mem_resv() &&
((job_resrcs_ptr->memory_allocated == NULL) ||
(job_resrcs_ptr->memory_used == NULL))) {
error("%s: lack memory allocation details to enforce memory limits for %pJ",
__func__, job_ptr);
step_ptr->pn_min_memory = 0;
} else if (step_ptr->pn_min_memory == MEM_PER_CPU)
step_ptr->pn_min_memory = 0; /* clear MEM_PER_CPU flag */
/* build cpus-per-node arrays for the subset of nodes used by step */
gres_test_args.max_rem_nodes =
bit_set_count(step_ptr->step_node_bitmap);
for (int i = 0; (node_ptr = next_node_bitmap(job_ptr->node_bitmap, &i));
i++) {
uint16_t cpus, cpus_used;
int err_code = SLURM_SUCCESS;
node_record_t *node_ptr;
gres_test_args.test_mem = false;
gres_test_args.err_code = &err_code;
gres_test_args.node_offset++;
if (!bit_test(step_ptr->step_node_bitmap, i))
continue;
node_ptr = node_record_table_ptr[i];
if (step_ptr->start_protocol_ver > node_ptr->protocol_version)
step_ptr->start_protocol_ver =
node_ptr->protocol_version;
/* find out the position in the job */
if (!bit_test(job_resrcs_ptr->node_bitmap, i))
return NULL;
pos = bit_set_count_range(job_resrcs_ptr->node_bitmap, 0, i);
if (pos >= job_resrcs_ptr->nhosts)
fatal("%s: node index bad", __func__);
cpus = job_resrcs_ptr->cpus[pos];
cpus_used = job_resrcs_ptr->cpus_used[pos];
/*
* Here we are trying to figure out the number
* of cpus available if we only want to run 1
* thread per core.
*/
if (_use_one_thread_per_core(step_ptr)) {
uint16_t threads;
threads = node_ptr->config_ptr->threads;
cpus /= threads;
cpus_used /= threads;
cpus_per_task_array[0] = cpus_per_task;
cpus_task_reps[0] = node_count;
} else {
/*
* Here we are trying to figure out how many
* CPUs each task really needs. This really
* only becomes an issue if the job requested
* ntasks_per_core|socket=1. We just increase
* the number of cpus_per_task to the thread
* count. Since the system could be
* heterogeneous, we needed to make this an
* array.
*/
uint16_t threads_per_core;
multi_core_data_t *mc_ptr = NULL;
if (job_ptr->details)
mc_ptr = job_ptr->details->mc_ptr;
if (step_ptr->threads_per_core != NO_VAL16)
threads_per_core = step_ptr->threads_per_core;
else if (mc_ptr &&
(mc_ptr->threads_per_core != NO_VAL16))
threads_per_core = mc_ptr->threads_per_core;
else
threads_per_core =
node_ptr->config_ptr->threads;
if (ntasks_per_socket == 1) {
uint16_t threads_per_socket;
threads_per_socket =
node_ptr->config_ptr->cores;
threads_per_socket *= threads_per_core;
if (cpus_per_task < threads_per_socket)
cpus_task = threads_per_socket;
} else if ((ntasks_per_core == 1) &&
(cpus_per_task < threads_per_core))
cpus_task = threads_per_core;
else
cpus_task = cpus_per_task;
if ((cpus_task_inx == -1) ||
(cpus_per_task_array[cpus_task_inx] != cpus_task)) {
cpus_task_inx++;
cpus_per_task_array[cpus_task_inx] = cpus_task;
cpus_task_reps[cpus_task_inx] = 1;
} else
cpus_task_reps[cpus_task_inx]++;
}
if (step_ptr->flags & SSF_OVERLAP_FORCE)
usable_cpus = cpus;
else
usable_cpus = cpus - cpus_used;
if (usable_cpus <= 0)
continue;
if ((step_ptr->pn_min_memory & MEM_PER_CPU) && _is_mem_resv()) {
uint64_t mem_use = step_ptr->pn_min_memory;
mem_use &= (~MEM_PER_CPU);
usable_mem = job_resrcs_ptr->memory_allocated[pos] -
job_resrcs_ptr->memory_used[pos];
usable_mem /= mem_use;
usable_cpus = MIN(usable_cpus, usable_mem);
} else if ((!step_ptr->pn_min_memory) && _is_mem_resv()) {
gres_test_args.test_mem = true;
}
if (step_ptr->flags & SSF_OVERLAP_FORCE)
gres_test_args.ignore_alloc = true;
else
gres_test_args.ignore_alloc = false;
gres_cpus = gres_stepmgr_step_test(&gres_test_args);
if (usable_cpus > gres_cpus)
usable_cpus = gres_cpus;
if (usable_cpus <= 0) {
error("%s: no usable CPUs", __func__);
return NULL;
}
debug3("step_layout cpus = %d pos = %d", usable_cpus, pos);
if ((cpu_inx == -1) ||
(cpus_per_node[cpu_inx] != usable_cpus)) {
cpu_inx++;
cpus_per_node[cpu_inx] = usable_cpus;
cpu_count_reps[cpu_inx] = 1;
} else
cpu_count_reps[cpu_inx]++;
set_nodes++;
gres_test_args.first_step_node = false;
gres_test_args.max_rem_nodes--;
#if 0
/*
* FIXME: on a heterogeneous system running the
* select/linear plugin we could get a node that doesn't
* have as many CPUs as we decided we needed for each
* task. This would result in not getting a task for
* the node we selected. This is usually in error. This
* only happens when the person doesn't specify how many
* cpus_per_task they want, and we have to come up with
* a number, in this case it is wrong.
*/
if (cpus_per_task > 0) {
set_tasks +=
(uint16_t)usable_cpus / cpus_per_task;
} else {
/*
* Since cpus_per_task is 0, we just add the
* count of CPUs available for this job
*/
set_tasks += usable_cpus;
}
info("usable_cpus is %d and set_tasks %d %d",
usable_cpus, set_tasks, cpus_per_task);
#endif
if (set_nodes == node_count)
break;
}
/* if (set_tasks < num_tasks) { */
/* error("Resources only available for %u of %u tasks", */
/* set_tasks, num_tasks); */
/* return NULL; */
/* } */
/* layout the tasks on the nodes */
memset(&step_layout_req, 0, sizeof(slurm_step_layout_req_t));
step_layout_req.node_list = step_node_list;
step_layout_req.cpus_per_node = cpus_per_node;
step_layout_req.cpu_count_reps = cpu_count_reps;
step_layout_req.cpus_per_task = cpus_per_task_array;
step_layout_req.cpus_task_reps = cpus_task_reps;
step_layout_req.num_hosts = node_count;
step_layout_req.num_tasks = num_tasks;
step_layout_req.task_dist = task_dist;
step_layout_req.plane_size = plane_size;
if ((step_layout = slurm_step_layout_create(&step_layout_req))) {
step_layout->start_protocol_ver = step_ptr->start_protocol_ver;
if (job_ptr->node_addrs)
step_layout->alias_addrs = build_alias_addrs(job_ptr);
}
return step_layout;
}
typedef struct {
list_t *dealloc_steps;
node_record_t *node_ptr;
bool node_fail;
} kill_step_on_node_args_t;
static int _kill_step_on_node(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
kill_step_on_node_args_t *args = (kill_step_on_node_args_t *) arg;
int step_node_inx = 0;
int bit_position = args->node_ptr->index;
int rem = 0;
uint32_t step_rc = 0;
step_complete_msg_t req;
if (step_ptr->state != JOB_RUNNING)
return 0;
if (!bit_test(step_ptr->step_node_bitmap, bit_position))
return 0;
/* Remove step allocation from the job's allocation */
step_node_inx = bit_set_count_range(step_ptr->step_node_bitmap, 0,
bit_position);
memset(&req, 0, sizeof(step_complete_msg_t));
memcpy(&req.step_id, &step_ptr->step_id, sizeof(req.step_id));
req.range_first = step_node_inx;
req.range_last = step_node_inx;
req.step_rc = 9;
req.jobacct = NULL; /* No accounting */
(void) _step_partial_comp(step_ptr, &req, false, &rem, &step_rc);
/*
* Do not kill the extern step on all nodes, only on the nodes that
* failed. Otherwise things that rely on the extern step such as x11
* or job_container/tmpfs won't work on the remaining nodes in the
* allocation.
*/
if (args->node_fail && !(step_ptr->flags & SSF_NO_KILL) &&
(step_ptr->step_id.step_id != SLURM_EXTERN_CONT)) {
info("Killing %pS due to failed node %s",
step_ptr, args->node_ptr->name);
signal_step_tasks(step_ptr, SIGKILL, REQUEST_TERMINATE_TASKS);
} else {
info("Killing %pS on failed node %s",
step_ptr, args->node_ptr->name);
signal_step_tasks_on_node(args->node_ptr->name, step_ptr,
SIGKILL, REQUEST_TERMINATE_TASKS);
}
if (!rem) {
if (!args->dealloc_steps)
args->dealloc_steps = list_create(NULL);
list_append(args->dealloc_steps, step_ptr);
}
return 0;
}
/*
* kill_step_on_node - determine if the specified job has any job steps
* allocated to the specified node and kill them unless no_kill flag
* is set on the step
* IN job_ptr - pointer to an active job record
* IN node_ptr - pointer to a node record
* IN node_fail - true of removed node has failed
*/
extern void kill_step_on_node(job_record_t *job_ptr, node_record_t *node_ptr,
bool node_fail)
{
kill_step_on_node_args_t args = {
.dealloc_steps = NULL,
.node_ptr = node_ptr,
.node_fail = node_fail,
};
if (!job_ptr || !node_ptr)
return;
list_for_each(job_ptr->step_list, _kill_step_on_node, &args);
if (args.dealloc_steps) {
/*
* Because _finish_step_comp() may free the step_ptr, call
* list_delete_all() to delete the list-node when the step_ptr
* is free'd. It doesn't actually matter because we are
* deleting the list immediately afterward, but it is good
* practice to not leave invalid pointer references.
*/
list_delete_all(args.dealloc_steps, _finish_step_comp, NULL);
FREE_NULL_LIST(args.dealloc_steps);
}
}
/*
* step_partial_comp - Note the completion of a job step on at least
* some of its nodes
* IN req - step_completion_msg RPC from slurmstepd
* IN uid - UID issuing the request
* IN finish - If true, no error, and no rem is 0 finish the step.
* OUT rem - count of nodes for which responses are still pending
* OUT max_rc - highest return code for any step thus far
* RET 0 on success, otherwise ESLURM error code
*/
extern int step_partial_comp(step_complete_msg_t *req, uid_t uid, bool finish,
int *rem, uint32_t *max_rc)
{
job_record_t *job_ptr;
step_record_t *step_ptr;
xassert(rem);
/* find the job, step, and validate input */
job_ptr = stepmgr_ops->find_job_record(req->step_id.job_id);
if (job_ptr == NULL) {
info("%s: JobId=%u invalid", __func__, req->step_id.job_id);
return ESLURM_INVALID_JOB_ID;
}
/* If we are requeuing the job the completing flag will be set
* but the state will be Pending, so don't use IS_JOB_PENDING
* which won't see the completing flag.
*/
if (job_ptr->job_state == JOB_PENDING) {
info("%s: %pJ pending", __func__, job_ptr);
return ESLURM_JOB_PENDING;
}
if ((!validate_slurm_user(uid)) && (uid != job_ptr->user_id)) {
/* Normally from slurmstepd, from srun on some failures */
error("Security violation: REQUEST_STEP_COMPLETE RPC for %pJ from uid=%u",
job_ptr, (unsigned int) uid);
return ESLURM_USER_ID_MISSING;
}
step_ptr = find_step_record(job_ptr, &req->step_id);
if (step_ptr == NULL) {
info("step_partial_comp: %pJ StepID=%u invalid; this step may have already completed",
job_ptr, req->step_id.step_id);
return ESLURM_INVALID_JOB_ID;
}
if (req->range_last < req->range_first) {
error("%s: %pS range=%u-%u",
__func__, step_ptr, req->range_first, req->range_last);
return EINVAL;
}
return _step_partial_comp(step_ptr, req, finish, rem, max_rc);
}
static int _step_partial_comp(step_record_t *step_ptr,
step_complete_msg_t *req, bool finish,
int *rem, uint32_t *max_rc)
{
int nodes, rem_nodes;
int range_bits, set_bits;
if (step_ptr->step_id.step_id == SLURM_BATCH_SCRIPT) {
error("%s: batch step received for %pJ. This should never happen.",
__func__, step_ptr->job_ptr);
return ESLURM_INVALID_JOB_ID;
}
/* we have been adding task average frequencies for
* jobacct->act_cpufreq so we need to divide with the
* total number of tasks/cpus for the step average frequency */
if (step_ptr->cpu_count && step_ptr->jobacct)
step_ptr->jobacct->act_cpufreq /= step_ptr->cpu_count;
if (!step_ptr->exit_node_bitmap) {
/* initialize the node bitmap for exited nodes */
nodes = bit_set_count(step_ptr->step_node_bitmap);
step_ptr->exit_node_bitmap = bit_alloc(nodes);
step_ptr->exit_code = req->step_rc;
} else {
nodes = bit_size(step_ptr->exit_node_bitmap);
if ((req->step_rc == SIG_OOM) ||
(req->step_rc > step_ptr->exit_code))
step_ptr->exit_code = req->step_rc;
}
if ((req->range_first >= nodes) || (req->range_last >= nodes) ||
(req->range_first > req->range_last)) {
/* range is zero origin */
error("%s: %pS range=%u-%u nodes=%d",
__func__, step_ptr, req->range_first, req->range_last,
nodes);
return EINVAL;
}
if ((step_ptr->flags & SSF_NO_SIG_FAIL) && WIFSIGNALED(req->step_rc)) {
step_ptr->exit_code = 0;
}
range_bits = req->range_last + 1 - req->range_first;
set_bits = bit_set_count_range(step_ptr->exit_node_bitmap,
req->range_first,
req->range_last + 1);
/* Check if any stepd of the range was already received */
if (set_bits) {
/* If all are already received skip jobacctinfo_aggregate */
if (set_bits == range_bits) {
debug("Step complete from %d to %d was already processed. Probably a RPC was resent from a child.",
req->range_first, req->range_last);
goto no_aggregate;
}
/*
* If partially received, we cannot recover the right gathered
* information. If we don't gather the new one we'll miss some
* information, and if we gather it some of the info will be
* duplicated. We log that error and chose to partially
* duplicate because it's probably a smaller error.
*/
error("Step complete from %d to %d was already processed (%d of %d). Probably a RPC was resent from a child and gathered information is partially duplicated.",
req->range_first, req->range_last,
set_bits, range_bits);
}
bit_nset(step_ptr->exit_node_bitmap,
req->range_first, req->range_last);
jobacctinfo_aggregate(step_ptr->jobacct, req->jobacct);
no_aggregate:
rem_nodes = bit_clear_count(step_ptr->exit_node_bitmap);
*rem = rem_nodes;
if (rem_nodes == 0) {
/* release all switch windows */
if (step_ptr->switch_step) {
debug2("full switch release for %pS, nodes %s",
step_ptr, step_ptr->step_layout->node_list);
switch_g_job_step_complete(
step_ptr->switch_step,
step_ptr->step_layout->node_list);
switch_g_free_stepinfo(step_ptr->switch_step);
step_ptr->switch_step = NULL;
}
}
if (max_rc)
*max_rc = step_ptr->exit_code;
if (req->step_rc == ESLURMD_EXECVE_FAILED)
step_ptr->state = JOB_NODE_FAIL;
/* The step has finished, finish it completely */
if (!*rem && finish) {
(void) _finish_step_comp(step_ptr, NULL);
}
return SLURM_SUCCESS;
}
/*
* step_set_alloc_tres - set the tres up when allocating the step.
* Only set when job is running.
*/
extern void step_set_alloc_tres(step_record_t *step_ptr, uint32_t node_count,
bool assoc_mgr_locked, bool make_formatted)
{
uint64_t cpu_count = 1, mem_count = 0;
char *tmp_tres_str = NULL;
assoc_mgr_lock_t locks = { .tres = READ_LOCK };
job_record_t *job_ptr = step_ptr->job_ptr;
xassert(step_ptr);
xfree(step_ptr->tres_alloc_str);
xfree(step_ptr->tres_fmt_alloc_str);
if (((step_ptr->step_id.step_id == SLURM_EXTERN_CONT) ||
(step_ptr->flags & SSF_EXT_LAUNCHER)) &&
job_ptr->tres_alloc_str) {
/* get the tres from the whole job */
step_ptr->tres_alloc_str =
xstrdup(job_ptr->tres_alloc_str);
if (make_formatted)
step_ptr->tres_fmt_alloc_str =
xstrdup(job_ptr->tres_fmt_alloc_str);
return;
}
if (!assoc_mgr_locked)
assoc_mgr_lock(&locks);
if (((step_ptr->step_id.step_id == SLURM_BATCH_SCRIPT) ||
(step_ptr->step_id.step_id == SLURM_INTERACTIVE_STEP)) &&
job_ptr->job_resrcs) {
int batch_inx = 0;
/*
* Figure out the index for the batch_host in relation to the
* job specific job_resrcs structure.
*/
if (job_ptr->batch_host) {
batch_inx = job_get_node_inx(
job_ptr->batch_host, job_ptr->node_bitmap);
if (batch_inx == -1) {
error("%s: Invalid batch host %s for %pJ; this should never happen",
__func__, job_ptr->batch_host, job_ptr);
batch_inx = 0;
}
}
/* get the cpus and memory on the first node */
if (job_ptr->job_resrcs->cpus)
cpu_count = job_ptr->job_resrcs->cpus[batch_inx];
if (job_ptr->job_resrcs->memory_allocated)
mem_count = job_ptr->job_resrcs->
memory_allocated[batch_inx];
tmp_tres_str = gres_stepmgr_gres_on_node_as_tres(
job_ptr->gres_list_alloc, 0, true);
} else {
if (!step_ptr->step_layout || !step_ptr->step_layout->task_cnt)
cpu_count = (uint64_t)job_ptr->total_cpus;
else
cpu_count = (uint64_t)step_ptr->cpu_count;
for (int i = 0; i < bit_set_count(step_ptr->step_node_bitmap);
i++)
mem_count += step_ptr->memory_allocated[i];
tmp_tres_str = gres_stepmgr_gres_2_tres_str(
step_ptr->gres_list_alloc, true);
}
xstrfmtcat(step_ptr->tres_alloc_str,
"%s%u=%"PRIu64",%u=%"PRIu64",%u=%u",
step_ptr->tres_alloc_str ? "," : "",
TRES_CPU, cpu_count,
TRES_MEM, mem_count,
TRES_NODE, node_count);
if (tmp_tres_str) {
xstrfmtcat(step_ptr->tres_alloc_str, "%s%s",
step_ptr->tres_alloc_str ? "," : "",
tmp_tres_str);
xfree(tmp_tres_str);
}
if (make_formatted)
step_ptr->tres_fmt_alloc_str =
slurmdb_make_tres_string_from_simple(
step_ptr->tres_alloc_str, assoc_mgr_tres_list,
NO_VAL, CONVERT_NUM_UNIT_EXACT, 0, NULL);
if (!assoc_mgr_locked)
assoc_mgr_unlock(&locks);
}
static int _suspend_job_step(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
job_record_t *job_ptr = step_ptr->job_ptr;
time_t *now = (time_t *) arg;
if (step_ptr->state != JOB_RUNNING)
return 0;
if ((job_ptr->suspend_time) &&
(job_ptr->suspend_time > step_ptr->start_time)) {
step_ptr->pre_sus_time +=
difftime(*now, job_ptr->suspend_time);
} else {
step_ptr->pre_sus_time +=
difftime(*now, step_ptr->start_time);
}
return 0;
}
/* Update time stamps for job step suspend */
extern void suspend_job_step(job_record_t *job_ptr)
{
time_t now = time(NULL);
list_for_each(job_ptr->step_list, _suspend_job_step, &now);
}
static int _resume_job_step(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
job_record_t *job_ptr = (job_record_t *) step_ptr->job_ptr;
time_t *now = (time_t *) arg;
if (step_ptr->state != JOB_RUNNING)
return 0;
if ((job_ptr->suspend_time) &&
(job_ptr->suspend_time < step_ptr->start_time)) {
step_ptr->tot_sus_time +=
difftime(*now, step_ptr->start_time);
} else {
step_ptr->tot_sus_time +=
difftime(*now, job_ptr->suspend_time);
}
return 0;
}
/* Update time stamps for job step resume */
extern void resume_job_step(job_record_t *job_ptr)
{
time_t now = time(NULL);
list_for_each(job_ptr->step_list, _resume_job_step, &now);
}
static void _signal_step_timelimit(step_record_t *step_ptr, time_t now)
{
node_record_t *node_ptr;
static bool cloud_dns = false;
static time_t last_update = 0;
job_record_t *job_ptr = step_ptr->job_ptr;
kill_job_msg_t *kill_step;
agent_arg_t *agent_args = NULL;
step_ptr->state = JOB_TIMEOUT;
xassert(step_ptr);
agent_args = xmalloc(sizeof(agent_arg_t));
agent_args->msg_type = REQUEST_KILL_TIMELIMIT;
agent_args->retry = 1;
agent_args->hostlist = hostlist_create(NULL);
kill_step = xmalloc(sizeof(kill_job_msg_t));
memcpy(&kill_step->step_id, &step_ptr->step_id,
sizeof(kill_step->step_id));
kill_step->het_job_id = job_ptr->het_job_id;
kill_step->job_state = job_ptr->job_state;
kill_step->job_uid = job_ptr->user_id;
kill_step->job_gid = job_ptr->group_id;
kill_step->nodes = xstrdup(job_ptr->nodes);
kill_step->time = now;
kill_step->start_time = job_ptr->start_time;
kill_step->details = xstrdup(job_ptr->state_desc);
if (last_update != slurm_conf.last_update) {
if (xstrcasestr(slurm_conf.slurmctld_params, "cloud_dns"))
cloud_dns = true;
else
cloud_dns = false;
last_update = slurm_conf.last_update;
}
if (step_ptr->step_node_bitmap) {
agent_args->protocol_version = SLURM_PROTOCOL_VERSION;
for (int i = 0;
(node_ptr = next_node_bitmap(step_ptr->step_node_bitmap,
&i));
i++) {
if (agent_args->protocol_version >
node_ptr->protocol_version) {
agent_args->protocol_version =
node_ptr->protocol_version;
}
hostlist_push_host(agent_args->hostlist,
node_ptr->name);
agent_args->node_count++;
if (PACK_FANOUT_ADDRS(node_ptr))
agent_args->msg_flags |= SLURM_PACK_ADDRS;
}
} else {
/* Could happen on node failure */
info("%s: %pJ Step %u has NULL node_bitmap", __func__,
job_ptr, step_ptr->step_id.step_id);
}
if (agent_args->node_count == 0) {
hostlist_destroy(agent_args->hostlist);
xfree(agent_args);
slurm_free_kill_job_msg(kill_step);
return;
}
agent_args->msg_args = kill_step;
set_agent_arg_r_uid(agent_args, SLURM_AUTH_UID_ANY);
stepmgr_ops->agent_queue_request(agent_args);
}
extern int check_job_step_time_limit(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
time_t *now = (time_t *) arg;
uint32_t job_run_mins = 0;
if (step_ptr->state != JOB_RUNNING)
return 0;
if (step_ptr->time_limit == INFINITE || step_ptr->time_limit == NO_VAL)
return 0;
job_run_mins = (uint32_t) (((*now - step_ptr->start_time) -
step_ptr->tot_sus_time) / 60);
if (job_run_mins >= step_ptr->time_limit) {
/* this step has timed out */
info("%s: %pS has timed out (%u)",
__func__, step_ptr, step_ptr->time_limit);
_signal_step_timelimit(step_ptr, *now);
}
return 0;
}
/* Return true if memory is a reserved resources, false otherwise */
static bool _is_mem_resv(void)
{
static bool mem_resv_value = false;
static bool mem_resv_tested = false;
if (!mem_resv_tested) {
mem_resv_tested = true;
if (slurm_conf.select_type_param & SELECT_MEMORY)
mem_resv_value = true;
}
return mem_resv_value;
}
typedef struct {
int mod_cnt;
uint32_t time_limit;
} update_step_args_t;
static int _update_step(void *x, void *arg)
{
step_record_t *step_ptr = (step_record_t *) x;
update_step_args_t *args = (update_step_args_t *) arg;
if (step_ptr->state != JOB_RUNNING)
return 0;
step_ptr->time_limit = args->time_limit;
args->mod_cnt++;
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
info("Updating %pS time limit to %u", step_ptr, args->time_limit);
return 0;
}
/*
* Process job step update request from specified user,
* RET - 0 or error code
*/
extern int update_step(step_update_request_msg_t *req, uid_t uid)
{
job_record_t *job_ptr;
step_record_t *step_ptr = NULL;
update_step_args_t args = { .mod_cnt = 0 };
slurm_step_id_t step_id = { 0 };
job_ptr = stepmgr_ops->find_job_record(req->job_id);
if (job_ptr == NULL) {
error("%s: invalid JobId=%u", __func__, req->job_id);
return ESLURM_INVALID_JOB_ID;
}
step_id.job_id = job_ptr->job_id;
step_id.step_id = req->step_id;
step_id.step_het_comp = NO_VAL;
/*
* No need to limit step time limit as job time limit will kill
* any steps with any time limit
*/
if (req->step_id == NO_VAL) {
args.time_limit = req->time_limit;
list_for_each(job_ptr->step_list, _update_step, &args);
} else {
step_ptr = find_step_record(job_ptr, &step_id);
if (!step_ptr && (job_ptr->bit_flags & STEPMGR_ENABLED))
goto stepmgr;
if (!step_ptr)
return ESLURM_INVALID_JOB_ID;
if (req->time_limit) {
step_ptr->time_limit = req->time_limit;
args.mod_cnt++;
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn,
step_ptr);
info("Updating %pS time limit to %u",
step_ptr, req->time_limit);
}
}
stepmgr:
if (running_in_slurmctld() && !step_ptr &&
(job_ptr->bit_flags & STEPMGR_ENABLED)) {
agent_arg_t *agent_args = NULL;
step_update_request_msg_t *agent_update_msg = NULL;
agent_update_msg = xmalloc(sizeof(*agent_update_msg));
agent_update_msg->job_id = req->job_id;
agent_update_msg->step_id = req->step_id;
agent_update_msg->time_limit = req->time_limit;
agent_args = xmalloc(sizeof(*agent_args));
agent_args->msg_type = REQUEST_UPDATE_JOB_STEP;
agent_args->retry = 1;
agent_args->hostlist = hostlist_create(job_ptr->batch_host);
agent_args->node_count = 1;
agent_args->protocol_version = SLURM_PROTOCOL_VERSION;
agent_args->msg_args = agent_update_msg;
set_agent_arg_r_uid(agent_args, slurm_conf.slurmd_user_id);
stepmgr_ops->agent_queue_request(agent_args);
args.mod_cnt++;
}
if (args.mod_cnt)
*stepmgr_ops->last_job_update = time(NULL);
return SLURM_SUCCESS;
}
static int _rebuild_bitmaps(void *x, void *arg)
{
int i_first, i_last, i_size;
int old_core_offset = 0, new_core_offset = 0;
bool old_node_set, new_node_set;
uint32_t step_id;
bitstr_t *orig_step_core_bitmap;
step_record_t *step_ptr = (step_record_t *) x;
bitstr_t *orig_job_node_bitmap = (bitstr_t *) arg;
job_record_t *job_ptr = step_ptr->job_ptr;
if (step_ptr->state < JOB_RUNNING)
return 0;
gres_stepmgr_step_state_rebase(step_ptr->gres_list_alloc,
orig_job_node_bitmap,
job_ptr->job_resrcs->node_bitmap);
if (!step_ptr->core_bitmap_job)
return 0;
step_id = step_ptr->step_id.step_id;
orig_step_core_bitmap = step_ptr->core_bitmap_job;
i_size = bit_size(job_ptr->job_resrcs->core_bitmap);
step_ptr->core_bitmap_job = bit_alloc(i_size);
i_first = MIN(bit_ffs(orig_job_node_bitmap),
bit_ffs(job_ptr->job_resrcs->node_bitmap));
i_last = MAX(bit_fls(orig_job_node_bitmap),
bit_fls(job_ptr->job_resrcs->node_bitmap));
for (int i = i_first; i <= i_last; i++) {
old_node_set = bit_test(orig_job_node_bitmap, i);
new_node_set = bit_test(job_ptr->job_resrcs->node_bitmap, i);
if (!old_node_set && !new_node_set)
continue;
if (old_node_set && new_node_set) {
for (int j = 0; j < node_record_table_ptr[i]->tot_cores;
j++) {
if (!bit_test(orig_step_core_bitmap,
old_core_offset + j))
continue;
bit_set(step_ptr->core_bitmap_job,
new_core_offset + j);
/*
* Only regular, non-overlapping steps should
* set bits in core_bitmap_used
*/
if ((step_id != SLURM_INTERACTIVE_STEP) &&
(step_id != SLURM_EXTERN_CONT) &&
(step_id != SLURM_BATCH_SCRIPT) &&
!(step_ptr->flags & SSF_OVERLAP_FORCE) &&
!(step_ptr->flags & SSF_EXT_LAUNCHER))
bit_set(job_ptr->job_resrcs->
core_bitmap_used,
new_core_offset + j);
}
}
if (old_node_set)
old_core_offset += node_record_table_ptr[i]->tot_cores;
if (new_node_set)
new_core_offset += node_record_table_ptr[i]->tot_cores;
}
FREE_NULL_BITMAP(orig_step_core_bitmap);
return 0;
}
/*
* Rebuild a job step's core_bitmap_job after a job has just changed size
* job_ptr IN - job that was just re-sized
* orig_job_node_bitmap IN - The job's original node bitmap
*/
extern void rebuild_step_bitmaps(job_record_t *job_ptr,
bitstr_t *orig_job_node_bitmap)
{
if (job_ptr->step_list == NULL)
return;
log_flag(STEPS, "Resizing steps of %pJ", job_ptr);
list_for_each(job_ptr->step_list, _rebuild_bitmaps,
orig_job_node_bitmap);
}
extern step_record_t *build_extern_step(job_record_t *job_ptr)
{
step_record_t *step_ptr = create_step_record(job_ptr, 0);
char *node_list;
uint32_t node_cnt;
node_list = job_ptr->nodes;
node_cnt = job_ptr->node_cnt;
if (!step_ptr) {
error("%s: Can't create step_record! This should never happen",
__func__);
return NULL;
}
*stepmgr_ops->last_job_update = time(NULL);
step_ptr->step_layout = fake_slurm_step_layout_create(
node_list, NULL, NULL, node_cnt, node_cnt,
SLURM_PROTOCOL_VERSION);
step_ptr->name = xstrdup("extern");
step_ptr->state = JOB_RUNNING;
step_ptr->start_time = job_ptr->start_time;
step_ptr->step_id.job_id = job_ptr->job_id;
step_ptr->step_id.step_id = SLURM_EXTERN_CONT;
step_ptr->step_id.step_het_comp = NO_VAL;
if (job_ptr->node_bitmap)
step_ptr->step_node_bitmap =
bit_copy(job_ptr->node_bitmap);
step_ptr->time_last_active = time(NULL);
step_set_alloc_tres(step_ptr, 1, false, false);
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
return step_ptr;
}
extern step_record_t *build_batch_step(job_record_t *job_ptr_in)
{
job_record_t *job_ptr;
step_record_t *step_ptr;
char *host = NULL;
if (job_ptr_in->het_job_id) {
job_ptr = stepmgr_ops->find_job_record(job_ptr_in->het_job_id);
if (!job_ptr) {
error("%s: hetjob leader is corrupt! This should never happen",
__func__);
job_ptr = job_ptr_in;
}
} else
job_ptr = job_ptr_in;
step_ptr = create_step_record(job_ptr, 0);
if (!step_ptr) {
error("%s: Can't create step_record! This should never happen",
__func__);
return NULL;
}
*stepmgr_ops->last_job_update = time(NULL);
host = job_ptr->batch_host;
step_ptr->step_layout = fake_slurm_step_layout_create(
host, NULL, NULL, 1, 1, SLURM_PROTOCOL_VERSION);
step_ptr->name = xstrdup("batch");
step_ptr->state = JOB_RUNNING;
step_ptr->start_time = job_ptr->start_time;
step_ptr->step_id.job_id = job_ptr->job_id;
step_ptr->step_id.step_id = SLURM_BATCH_SCRIPT;
step_ptr->step_id.step_het_comp = NO_VAL;
step_ptr->container = xstrdup(job_ptr->container);
step_ptr->container_id = xstrdup(job_ptr->container_id);
if (node_name2bitmap(job_ptr->batch_host, false,
&step_ptr->step_node_bitmap, NULL)) {
error("%s: %pJ has invalid node list (%s)",
__func__, job_ptr, job_ptr->batch_host);
}
step_ptr->time_last_active = time(NULL);
step_set_alloc_tres(step_ptr, 1, false, false);
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
return step_ptr;
}
static step_record_t *_build_interactive_step(
job_record_t *job_ptr_in,
job_step_create_request_msg_t *step_specs,
uint16_t protocol_version)
{
job_record_t *job_ptr;
step_record_t *step_ptr;
char *host = NULL;
slurm_step_id_t step_id = {0};
if (job_ptr_in->het_job_id) {
job_ptr = stepmgr_ops->find_job_record(job_ptr_in->het_job_id);
if (!job_ptr) {
error("%s: hetjob leader is corrupt! This should never happen",
__func__);
job_ptr = job_ptr_in;
}
} else
job_ptr = job_ptr_in;
step_id.job_id = job_ptr->job_id;
step_id.step_id = SLURM_INTERACTIVE_STEP;
step_id.step_het_comp = NO_VAL;
step_ptr = find_step_record(job_ptr, &step_id);
if (step_ptr) {
debug("%s: interactive step for %pJ already exists",
__func__, job_ptr);
return NULL;
}
host = job_ptr->batch_host;
if (!host) {
error("%s: %pJ batch_host is NULL! This should never happen",
__func__, job_ptr);
return NULL;
}
step_ptr = create_step_record(job_ptr, protocol_version);
if (!step_ptr) {
error("%s: Can't create step_record! This should never happen",
__func__);
return NULL;
}
*stepmgr_ops->last_job_update = time(NULL);
step_ptr->step_layout = fake_slurm_step_layout_create(
host, NULL, NULL, 1, 1, protocol_version);
step_ptr->name = xstrdup("interactive");
step_ptr->state = JOB_RUNNING;
step_ptr->start_time = job_ptr->start_time;
step_ptr->step_id.job_id = job_ptr->job_id;
step_ptr->step_id.step_id = SLURM_INTERACTIVE_STEP;
step_ptr->step_id.step_het_comp = NO_VAL;
step_ptr->container = xstrdup(job_ptr->container);
step_ptr->container_id = xstrdup(job_ptr->container_id);
step_ptr->port = step_specs->port;
step_ptr->srun_pid = step_specs->srun_pid;
step_ptr->host = xstrdup(step_specs->host);
step_ptr->submit_line = xstrdup(step_specs->submit_line);
step_ptr->core_bitmap_job = bit_copy(job_ptr->job_resrcs->core_bitmap);
if (node_name2bitmap(job_ptr->batch_host, false,
&step_ptr->step_node_bitmap, NULL)) {
error("%s: %pJ has invalid node list (%s)",
__func__, job_ptr, job_ptr->batch_host);
delete_step_record(job_ptr, step_ptr);
return NULL;
}
step_ptr->time_last_active = time(NULL);
step_set_alloc_tres(step_ptr, 1, false, false);
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
return step_ptr;
}
/*
* Build a special step for mpi launchers.
*/
static int _build_ext_launcher_step(step_record_t **step_rec,
job_record_t *job_ptr,
job_step_create_request_msg_t *step_specs,
uint16_t protocol_version)
{
bitstr_t *nodeset;
uint32_t node_count;
int rc;
char *step_node_list;
step_record_t *step_ptr;
if (!step_rec)
return SLURM_ERROR;
if (job_ptr->next_step_id >= slurm_conf.max_step_cnt) {
error("%s: %pJ MaxStepCount limit reached", __func__, job_ptr);
return ESLURM_STEP_LIMIT;
}
/* Reset some fields we're going to ignore in _pick_step_nodes. */
step_specs->flags = SSF_EXT_LAUNCHER;
step_specs->cpu_count = 0;
xfree(step_specs->cpus_per_tres);
step_specs->ntasks_per_core = NO_VAL16;
step_specs->ntasks_per_tres = NO_VAL16;
step_specs->pn_min_memory = 0;
xfree(step_specs->mem_per_tres);
step_specs->threads_per_core = NO_VAL16;
xfree(step_specs->tres_bind);
xfree(step_specs->tres_per_step);
xfree(step_specs->tres_per_node);
xfree(step_specs->tres_per_socket);
xfree(step_specs->tres_per_task);
/* Select the nodes for this job */
nodeset = _pick_step_nodes(job_ptr, step_specs, NULL, 0, 0, &rc);
if (nodeset == NULL) {
return rc;
}
/* Here is where the node list is set for the step */
if (step_specs->node_list &&
((step_specs->task_dist & SLURM_DIST_STATE_BASE) ==
SLURM_DIST_ARBITRARY)) {
step_node_list = xstrdup(step_specs->node_list);
xfree(step_specs->node_list);
step_specs->node_list = bitmap2node_name(nodeset);
} else {
step_node_list = bitmap2node_name_sortable(nodeset, false);
xfree(step_specs->node_list);
step_specs->node_list = xstrdup(step_node_list);
}
log_flag(STEPS, "Picked nodes %s when accumulating from %s",
step_node_list, step_specs->node_list);
step_ptr = *step_rec = create_step_record(job_ptr, protocol_version);
if (!step_ptr) {
error("%s: Can't create step_record! This should never happen",
__func__);
return SLURM_ERROR;
}
*stepmgr_ops->last_job_update = time(NULL);
/* We want 1 task per node. */
step_ptr->step_node_bitmap = nodeset;
node_count = bit_set_count(nodeset);
step_specs->num_tasks = node_count;
/* Create the fake step layout with 1 task per node */
step_ptr->step_layout = fake_slurm_step_layout_create(
step_node_list, NULL, NULL, node_count, node_count,
SLURM_PROTOCOL_VERSION);
xfree(step_node_list);
if (!step_ptr->step_layout) {
delete_step_record(job_ptr, step_ptr);
return SLURM_ERROR;
}
/* Needed for not considering it in _mark_busy_nodes */
step_ptr->flags |= SSF_EXT_LAUNCHER;
/* Set the step id */
memcpy(&step_ptr->step_id, &step_specs->step_id,
sizeof(step_ptr->step_id));
if (step_specs->array_task_id != NO_VAL)
step_ptr->step_id.job_id = job_ptr->job_id;
if (step_specs->step_id.step_id != NO_VAL) {
if (step_specs->step_id.step_het_comp == NO_VAL) {
job_ptr->next_step_id =
MAX(job_ptr->next_step_id,
step_specs->step_id.step_id);
job_ptr->next_step_id++;
}
} else if (job_ptr->het_job_id &&
(job_ptr->het_job_id != job_ptr->job_id)) {
job_record_t *het_job;
het_job = stepmgr_ops->find_job_record(job_ptr->het_job_id);
if (het_job)
step_ptr->step_id.step_id = het_job->next_step_id++;
else
step_ptr->step_id.step_id = job_ptr->next_step_id++;
job_ptr->next_step_id = MAX(job_ptr->next_step_id,
step_ptr->step_id.step_id);
} else {
step_ptr->step_id.step_id = job_ptr->next_step_id++;
}
/* The step needs to run on all the cores. */
step_ptr->core_bitmap_job = bit_copy(job_ptr->job_resrcs->core_bitmap);
step_ptr->name = xstrdup(step_specs->name);
step_ptr->state = JOB_RUNNING;
step_ptr->start_time = job_ptr->start_time;
step_ptr->time_last_active = time(NULL);
step_set_alloc_tres(step_ptr, 1, false, false);
jobacct_storage_g_step_start(stepmgr_ops->acct_db_conn, step_ptr);
if ((rc = _switch_setup(step_ptr))) {
delete_step_record(job_ptr, step_ptr);
return rc;
}
return SLURM_SUCCESS;
}
extern slurm_node_alias_addrs_t *build_alias_addrs(job_record_t *job_ptr)
{
slurm_node_alias_addrs_t *alias_addrs;
if (!job_ptr || !job_ptr->node_addrs)
return NULL;
alias_addrs = xmalloc(sizeof(slurm_node_alias_addrs_t));
alias_addrs->node_cnt = job_ptr->node_cnt;
alias_addrs->node_addrs = xcalloc(job_ptr->node_cnt,
sizeof(slurm_addr_t));
memcpy(alias_addrs->node_addrs, job_ptr->node_addrs,
(sizeof(slurm_addr_t) * job_ptr->node_cnt));
alias_addrs->node_list = xstrdup(job_ptr->nodes);
return alias_addrs;
}
extern int job_get_node_inx(char *node_name, bitstr_t *node_bitmap)
{
int node_inx = -1;
if (!node_name)
return -1;
xassert(node_bitmap);
node_inx = node_name_get_inx(node_name);
if (node_inx == -1)
return -1;
if (!bit_test(node_bitmap, node_inx))
return -1;
return bit_set_count_range(node_bitmap, 0, node_inx);
}
static void _kill_step_on_msg_fail(step_complete_msg_t *req, slurm_msg_t *msg,
void (*lock_func)(bool lock))
{
int rc, rem;
uint32_t step_rc;
DEF_TIMERS;
/* init */
START_TIMER;
error("Step creation timed out: Deallocating %ps nodes %u-%u",
&req->step_id, req->range_first, req->range_last);
if (lock_func)
lock_func(true);
rc = step_partial_comp(req, msg->auth_uid, true, &rem, &step_rc);
if (lock_func)
lock_func(false);
END_TIMER2(__func__);
log_flag(STEPS, "%s: %ps rc:%s %s",
__func__, &req->step_id, slurm_strerror(rc), TIME_STR);
}
/* create a credential for a given job step, return error code */
static int _make_step_cred(step_record_t *step_ptr, slurm_cred_t **slurm_cred,
uint16_t protocol_version)
{
slurm_cred_arg_t cred_arg;
job_record_t *job_ptr = step_ptr->job_ptr;
job_resources_t *job_resrcs_ptr = job_ptr->job_resrcs;
xassert(job_resrcs_ptr && job_resrcs_ptr->cpus);
setup_cred_arg(&cred_arg, job_ptr);
memcpy(&cred_arg.step_id, &step_ptr->step_id, sizeof(cred_arg.step_id));
if (job_resrcs_ptr->memory_allocated) {
slurm_array64_to_value_reps(job_resrcs_ptr->memory_allocated,
job_resrcs_ptr->nhosts,
&cred_arg.job_mem_alloc,
&cred_arg.job_mem_alloc_rep_count,
&cred_arg.job_mem_alloc_size);
}
cred_arg.step_gres_list = step_ptr->gres_list_alloc;
cred_arg.step_core_bitmap = step_ptr->core_bitmap_job;
cred_arg.step_hostlist = step_ptr->step_layout->node_list;
if (step_ptr->memory_allocated) {
slurm_array64_to_value_reps(step_ptr->memory_allocated,
step_ptr->step_layout->node_cnt,
&cred_arg.step_mem_alloc,
&cred_arg.step_mem_alloc_rep_count,
&cred_arg.step_mem_alloc_size);
}
cred_arg.switch_step = step_ptr->switch_step;
*slurm_cred = slurm_cred_create(&cred_arg, true, protocol_version);
xfree(cred_arg.job_mem_alloc);
xfree(cred_arg.job_mem_alloc_rep_count);
xfree(cred_arg.step_mem_alloc);
xfree(cred_arg.step_mem_alloc_rep_count);
if (*slurm_cred == NULL) {
error("slurm_cred_create error");
return ESLURM_INVALID_JOB_CREDENTIAL;
}
return SLURM_SUCCESS;
}
static int _send_msg(slurm_msg_t *msg, int slurmd_fd, slurm_msg_type_t type,
void *data)
{
xassert(running_in_slurmctld() || running_in_slurmstepd());
int rc;
if (running_in_slurmctld()) {
if ((rc = send_msg_response(msg, type, data))) {
errno = rc;
return SLURM_ERROR;
}
return SLURM_SUCCESS;
}
if (running_in_slurmstepd()) {
if ((stepd_proxy_send_resp_to_slurmd(slurmd_fd, msg, type,
data))) {
return SLURM_ERROR;
}
return SLURM_SUCCESS;
}
return SLURM_ERROR;
}
extern int step_create_from_msg(slurm_msg_t *msg, int slurmd_fd,
void (*lock_func)(bool lock),
void (*fail_lock_func)(bool lock))
{
char *err_msg = NULL;
int error_code = SLURM_SUCCESS;
DEF_TIMERS;
step_record_t *step_rec;
job_step_create_response_msg_t job_step_resp;
job_step_create_request_msg_t *req_step_msg = msg->data;
slurm_cred_t *slurm_cred = NULL;
job_record_t *job_ptr = NULL;
START_TIMER;
xassert(msg->auth_ids_set);
if (req_step_msg->user_id == SLURM_AUTH_NOBODY) {
req_step_msg->user_id = msg->auth_uid;
if (get_log_level() >= LOG_LEVEL_DEBUG3) {
char *host = auth_g_get_host(msg);
debug3("%s: [%s] set RPC user_id to %d",
__func__, host, msg->auth_uid);
xfree(host);
}
} else if (msg->auth_uid != req_step_msg->user_id) {
return_code_msg_t rc_msg = {
.return_code = ESLURM_USER_ID_MISSING,
};
error("Security violation, JOB_STEP_CREATE RPC from uid=%u to run as uid %u",
msg->auth_uid, req_step_msg->user_id);
_send_msg(msg, slurmd_fd, RESPONSE_SLURM_RC, &rc_msg);
return ESLURM_USER_ID_MISSING;
}
dump_step_desc(req_step_msg);
if (lock_func) {
lock_func(true);
}
if (req_step_msg->array_task_id != NO_VAL)
job_ptr = stepmgr_ops->find_job_array_rec(
req_step_msg->step_id.job_id,
req_step_msg->array_task_id);
else
job_ptr = stepmgr_ops->find_job_record(
req_step_msg->step_id.job_id);
if (job_ptr == NULL) {
error_code = ESLURM_INVALID_JOB_ID ;
goto end_it;
}
if (job_ptr->bit_flags & EXTERNAL_JOB) {
error("%s: step creation disabled for external jobs", __func__);
error_code = ESLURM_NOT_SUPPORTED;
goto end_it;
}
if (running_in_slurmctld() &&
(job_ptr->bit_flags & STEPMGR_ENABLED)) {
reroute_msg_t reroute_msg = {
.stepmgr = job_ptr->batch_host,
};
_send_msg(msg, slurmd_fd, RESPONSE_SLURM_REROUTE_MSG,
&reroute_msg);
if (lock_func)
lock_func(false);
return SLURM_SUCCESS;
}
error_code = step_create(job_ptr, req_step_msg, &step_rec,
msg->protocol_version, &err_msg);
if (error_code == SLURM_SUCCESS) {
error_code = _make_step_cred(step_rec, &slurm_cred,
step_rec->start_protocol_ver);
}
END_TIMER2(__func__);
end_it:
/* return result */
if (error_code) {
if (lock_func)
lock_func(false);
if (error_code == ESLURM_PROLOG_RUNNING)
log_flag(STEPS, "%s for configuring %ps: %s",
__func__, &req_step_msg->step_id,
slurm_strerror(error_code));
else if (error_code == ESLURM_DISABLED)
log_flag(STEPS, "%s for suspended %ps: %s",
__func__, &req_step_msg->step_id,
slurm_strerror(error_code));
else
log_flag(STEPS, "%s for %ps: %s",
__func__, &req_step_msg->step_id,
slurm_strerror(error_code));
if (err_msg) {
return_code2_msg_t rc_msg = {
.return_code = error_code,
.err_msg = err_msg,
};
_send_msg(msg, slurmd_fd, RESPONSE_SLURM_RC_MSG,
&rc_msg);
} else {
return_code_msg_t rc_msg = {
.return_code = error_code,
};
_send_msg(msg, slurmd_fd, RESPONSE_SLURM_RC, &rc_msg);
}
} else {
slurm_step_layout_t *step_layout = NULL;
dynamic_plugin_data_t *switch_step = NULL;
log_flag(STEPS, "%s: %pS %s %s",
__func__, step_rec, req_step_msg->node_list, TIME_STR);
memset(&job_step_resp, 0, sizeof(job_step_resp));
job_step_resp.job_id = step_rec->step_id.job_id;
job_step_resp.job_step_id = step_rec->step_id.step_id;
job_step_resp.resv_ports = step_rec->resv_ports;
step_layout = slurm_step_layout_copy(step_rec->step_layout);
job_step_resp.step_layout = step_layout;
if (step_rec->job_ptr && step_rec->job_ptr->details &&
(step_rec->job_ptr->details->cpu_bind_type != NO_VAL16)) {
job_step_resp.def_cpu_bind_type =
step_rec->job_ptr->details->cpu_bind_type;
}
job_step_resp.cred = slurm_cred;
job_step_resp.use_protocol_ver = step_rec->start_protocol_ver;
if (step_rec->switch_step)
switch_g_duplicate_stepinfo(step_rec->switch_step,
&switch_step);
job_step_resp.switch_step = switch_step;
if (job_ptr->bit_flags & STEPMGR_ENABLED)
job_step_resp.stepmgr = job_ptr->batch_host;
if (lock_func)
lock_func(false);
if (msg->protocol_version != step_rec->start_protocol_ver) {
log_flag(NET, "%s: responding with non-matching msg 0x%x to step 0x%x protocol version",
__func__, msg->protocol_version,
step_rec->start_protocol_ver);
msg->protocol_version = step_rec->start_protocol_ver;
}
if (_send_msg(msg, slurmd_fd, RESPONSE_JOB_STEP_CREATE,
&job_step_resp)) {
step_complete_msg_t req;
memset(&req, 0, sizeof(req));
req.step_id = step_rec->step_id;
req.jobacct = step_rec->jobacct;
req.step_rc = SIGKILL;
req.range_first = 0;
req.range_last = step_layout->node_cnt - 1;
_kill_step_on_msg_fail(&req, msg, fail_lock_func);
}
slurm_cred_destroy(slurm_cred);
slurm_step_layout_destroy(step_layout);
switch_g_free_stepinfo(switch_step);
}
xfree(err_msg);
return error_code;
}
/*
* pack_job_step_info_response_msg - packs job step info
* IN step_id - specific id or NO_VAL/NO_VAL for all
* IN uid - user issuing request
* IN show_flags - job step filtering options
* OUT buffer - location to store data, pointers automatically advanced
* RET - 0 or error code
* NOTE: MUST free_buf buffer
*/
extern int pack_job_step_info_response_msg(pack_step_args_t *args)
{
int error_code = 0;
uint32_t tmp_offset;
time_t now = time(NULL);
if (args->proto_version >= SLURM_MIN_PROTOCOL_VERSION) {
/* steps_packed placeholder */
pack32(args->steps_packed, args->buffer);
pack_time(now, args->buffer);
list_for_each_ro(args->job_step_list,
args->pack_job_step_list_func, args);
if (list_count(job_list) && !args->valid_job &&
!args->steps_packed)
error_code = ESLURM_INVALID_JOB_ID;
slurm_pack_list(args->stepmgr_jobs,
slurm_pack_stepmgr_job_info, args->buffer,
args->proto_version);
/* put the real record count in the message body header */
tmp_offset = get_buf_offset(args->buffer);
set_buf_offset(args->buffer, 0);
pack32(args->steps_packed, args->buffer);
set_buf_offset(args->buffer, tmp_offset);
}
xfree(args->visible_parts);
return error_code;
}
extern int stepmgr_get_step_layouts(job_record_t *job_ptr,
slurm_step_id_t *step_id,
slurm_step_layout_t **out_step_layout)
{
list_itr_t *itr;
step_record_t *step_ptr = NULL;
slurm_step_layout_t *step_layout = NULL;
/* We can't call find_step_record here since we may need more than 1 */
itr = list_iterator_create(job_ptr->step_list);
while ((step_ptr = list_next(itr))) {
if (!verify_step_id(&step_ptr->step_id, step_id))
continue;
/*
* Rebuild alias_addrs if need after restart of slurmctld
*/
if (job_ptr->node_addrs &&
!step_ptr->step_layout->alias_addrs) {
step_ptr->step_layout->alias_addrs =
build_alias_addrs(job_ptr);
}
if (step_layout)
slurm_step_layout_merge(step_layout,
step_ptr->step_layout);
else
step_layout = slurm_step_layout_copy(
step_ptr->step_layout);
/* break if don't need to look for further het_steps */
if (step_ptr->step_id.step_het_comp == NO_VAL)
break;
/*
* If we are looking for a specific het step we can break here
* as well.
*/
if (step_id->step_het_comp != NO_VAL)
break;
}
list_iterator_destroy(itr);
if (!step_layout) {
log_flag(STEPS, "%s: %pJ StepId=%u Not Found",
__func__, job_ptr, step_id->step_id);
return ESLURM_INVALID_JOB_ID;
}
/*
* The cpt_compact* fields don't go to the client because they are not
* handled in slurm_step_layout_merge(). Free them so the client does
* not get bad data.
*/
xfree(step_layout->cpt_compact_array);
xfree(step_layout->cpt_compact_reps);
step_layout->cpt_compact_cnt = 0;
*out_step_layout = step_layout;
return SLURM_SUCCESS;
}
extern int stepmgr_get_job_sbcast_cred_msg(job_record_t *job_ptr,
slurm_step_id_t *step_id,
uint16_t protocol_version,
job_sbcast_cred_msg_t **out_sbcast_cred_msg)
{
sbcast_cred_t *sbcast_cred;
sbcast_cred_arg_t sbcast_arg;
step_record_t *step_ptr = NULL;
char *node_list = NULL;
job_sbcast_cred_msg_t *job_info_resp_msg;
xassert(job_ptr);
if (step_id->step_id != NO_VAL) {
step_ptr = find_step_record(job_ptr, step_id);
if (!step_ptr) {
return ESLURM_INVALID_JOB_ID;
} else if (step_ptr->step_layout &&
(step_ptr->step_layout->node_cnt !=
job_ptr->node_cnt)) {
node_list = step_ptr->step_layout->node_list;
}
}
if (!node_list)
node_list = job_ptr->nodes;
/*
* Note - using pointers to other xmalloc'd elements owned by other
* structures to avoid copy overhead. Do not free them!
*/
memset(&sbcast_arg, 0, sizeof(sbcast_arg));
sbcast_arg.job_id = job_ptr->job_id;
sbcast_arg.het_job_id = job_ptr->het_job_id;
if (step_ptr)
sbcast_arg.step_id = step_ptr->step_id.step_id;
else
sbcast_arg.step_id = job_ptr->next_step_id;
sbcast_arg.nodes = node_list; /* avoid extra copy */
sbcast_arg.expiration = job_ptr->end_time;
if (!(sbcast_cred = create_sbcast_cred(&sbcast_arg, job_ptr->user_id,
job_ptr->group_id,
protocol_version))) {
error("%s %pJ cred create error", __func__, job_ptr);
return SLURM_ERROR;
}
job_info_resp_msg = xmalloc(sizeof(*job_info_resp_msg));
job_info_resp_msg->job_id = job_ptr->job_id;
job_info_resp_msg->node_list = xstrdup(node_list);
job_info_resp_msg->sbcast_cred = sbcast_cred;
*out_sbcast_cred_msg = job_info_resp_msg;
return SLURM_SUCCESS;
}
/* Build structure with job allocation details */
extern resource_allocation_response_msg_t *build_job_info_resp(
job_record_t *job_ptr)
{
resource_allocation_response_msg_t *job_info_resp_msg;
int i, j;
job_info_resp_msg = xmalloc(sizeof(resource_allocation_response_msg_t));
if (!job_ptr->job_resrcs) {
;
} else if (bit_equal(job_ptr->node_bitmap,
job_ptr->job_resrcs->node_bitmap)) {
job_info_resp_msg->num_cpu_groups =
job_ptr->job_resrcs->cpu_array_cnt;
job_info_resp_msg->cpu_count_reps =
xcalloc(job_ptr->job_resrcs->cpu_array_cnt,
sizeof(uint32_t));
memcpy(job_info_resp_msg->cpu_count_reps,
job_ptr->job_resrcs->cpu_array_reps,
(sizeof(uint32_t) * job_ptr->job_resrcs->cpu_array_cnt));
job_info_resp_msg->cpus_per_node =
xcalloc(job_ptr->job_resrcs->cpu_array_cnt,
sizeof(uint16_t));
memcpy(job_info_resp_msg->cpus_per_node,
job_ptr->job_resrcs->cpu_array_value,
(sizeof(uint16_t) * job_ptr->job_resrcs->cpu_array_cnt));
} else {
/* Job has changed size, rebuild CPU count info */
job_info_resp_msg->num_cpu_groups = job_ptr->node_cnt;
job_info_resp_msg->cpu_count_reps = xcalloc(job_ptr->node_cnt,
sizeof(uint32_t));
job_info_resp_msg->cpus_per_node = xcalloc(job_ptr->node_cnt,
sizeof(uint32_t));
for (i = 0, j = -1; i < job_ptr->job_resrcs->nhosts; i++) {
if (job_ptr->job_resrcs->cpus[i] == 0)
continue;
if ((j == -1) ||
(job_info_resp_msg->cpus_per_node[j] !=
job_ptr->job_resrcs->cpus[i])) {
j++;
job_info_resp_msg->cpus_per_node[j] =
job_ptr->job_resrcs->cpus[i];
job_info_resp_msg->cpu_count_reps[j] = 1;
} else {
job_info_resp_msg->cpu_count_reps[j]++;
}
}
job_info_resp_msg->num_cpu_groups = j + 1;
}
job_info_resp_msg->account = xstrdup(job_ptr->account);
job_info_resp_msg->batch_host = xstrdup(job_ptr->batch_host);
job_info_resp_msg->job_id = job_ptr->job_id;
job_info_resp_msg->node_cnt = job_ptr->node_cnt;
job_info_resp_msg->node_list = xstrdup(job_ptr->nodes);
if (job_ptr->part_ptr)
job_info_resp_msg->partition = xstrdup(job_ptr->part_ptr->name);
else
job_info_resp_msg->partition = xstrdup(job_ptr->partition);
if (job_ptr->qos_ptr) {
slurmdb_qos_rec_t *qos;
qos = (slurmdb_qos_rec_t *)job_ptr->qos_ptr;
job_info_resp_msg->qos = xstrdup(qos->name);
}
job_info_resp_msg->resv_name = xstrdup(job_ptr->resv_name);
if (job_ptr->details) {
if (job_ptr->bit_flags & JOB_MEM_SET) {
job_info_resp_msg->pn_min_memory =
job_ptr->details->pn_min_memory;
}
if (job_ptr->details->mc_ptr) {
job_info_resp_msg->ntasks_per_board =
job_ptr->details->mc_ptr->ntasks_per_board;
job_info_resp_msg->ntasks_per_core =
job_ptr->details->mc_ptr->ntasks_per_core;
job_info_resp_msg->ntasks_per_socket =
job_ptr->details->mc_ptr->ntasks_per_socket;
}
} else {
/* job_info_resp_msg->pn_min_memory = 0; */
job_info_resp_msg->ntasks_per_board = NO_VAL16;
job_info_resp_msg->ntasks_per_core = NO_VAL16;
job_info_resp_msg->ntasks_per_socket = NO_VAL16;
}
if (job_ptr->details && job_ptr->details->env_cnt) {
job_info_resp_msg->env_size = job_ptr->details->env_cnt;
job_info_resp_msg->environment =
xcalloc(job_info_resp_msg->env_size + 1,
sizeof(char *));
for (i = 0; i < job_info_resp_msg->env_size; i++) {
job_info_resp_msg->environment[i] =
xstrdup(job_ptr->details->env_sup[i]);
}
job_info_resp_msg->environment[i] = NULL;
}
job_info_resp_msg->uid = job_ptr->user_id;
job_info_resp_msg->user_name = user_from_job(job_ptr);
job_info_resp_msg->gid = job_ptr->group_id;
job_info_resp_msg->group_name = group_from_job(job_ptr);
return job_info_resp_msg;
}