/*
* Copyright (C) 1994-2021 Altair Engineering, Inc.
* For more information, contact Altair at www.altair.com.
*
* This file is part of both the OpenPBS software ("OpenPBS")
* and the PBS Professional ("PBS Pro") software.
*
* Open Source License Information:
*
* OpenPBS is free software. You can redistribute it and/or modify it under
* the terms of the GNU Affero General Public License as published by the
* Free Software Foundation, either version 3 of the License, or (at your
* option) any later version.
*
* OpenPBS 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 Affero General Public
* License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*
* Commercial License Information:
*
* PBS Pro is commercially licensed software that shares a common core with
* the OpenPBS software. For a copy of the commercial license terms and
* conditions, go to: (http://www.pbspro.com/agreement.html) or contact the
* Altair Legal Department.
*
* Altair's dual-license business model allows companies, individuals, and
* organizations to create proprietary derivative works of OpenPBS and
* distribute them - whether embedded or bundled with other software -
* under a commercial license agreement.
*
* Use of Altair's trademarks, including but not limited to "PBS™",
* "OpenPBS®", "PBS Professional®", and "PBS Pro™" and Altair's logos is
* subject to Altair's trademark licensing policies.
*/
/**
*
* @brief
* Functions relating to the Queue Job Batch Request sequence, including
* Queue Job, Job Script, Ready to Commit, and Commit.
*
*
*/
#include /* the master config generated by configure */
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include "libpbs.h"
#include "server_limits.h"
#include "list_link.h"
#include "attribute.h"
#include "resource.h"
#include "server.h"
#include "work_task.h"
#include "credential.h"
#include "ticket.h"
#include "batch_request.h"
#include "resv_node.h"
#include "queue.h"
#include "job.h"
#include "reservation.h"
#include "net_connect.h"
#include "pbs_error.h"
#include "pbs_nodes.h"
#include "svrfunc.h"
#include "sched_cmds.h"
#include "log.h"
#include "acct.h"
#include "tpp.h"
#include "user.h"
#include "hook.h"
#include "pbs_internal.h"
#include "pbs_sched.h"
#ifndef PBS_MOM
#include "pbs_db.h"
#define SEQ_WIN_INCR 1000 /*save jobid number to database in this increment*/
#endif
#include "libutil.h"
#ifdef PBS_MOM
#include "mom_hook_func.h"
#include "placementsets.h"
#include
#include "mom_func.h"
extern char mom_host[PBS_MAXHOSTNAME + 1];
#endif /* PBS_MOM */
#define RESV_INFINITY (60 * 60 * 24 * 365 * 5)
/* External Functions Called: */
extern struct connection *svr_conn;
#ifndef PBS_MOM
extern int remtree(char *);
#ifdef NAS /* localmod 005 */
extern int apply_aoe_inchunk_rules(resource *, attribute *, void *, int);
#endif /* localmod 005 */
void post_sendmom(struct work_task *);
void post_sendmom_inner(job *, struct batch_request *, int, int, char *);
#endif /* PBS_MOM */
/* Global Data Items: */
#ifndef PBS_MOM
extern char *path_spool;
extern struct server server;
extern struct attribute attr_jobscript_max_size;
extern char server_name[];
extern unsigned int pbs_server_port_dis;
extern char *resc_in_err;
#endif /* PBS_MOM */
extern int resc_access_perm;
extern pbs_list_head svr_alljobs;
extern pbs_list_head svr_newjobs;
extern attribute_def job_attr_def[];
extern char *path_jobs;
extern char *pbs_o_host;
extern char *msg_script_open;
extern char *msg_script_write;
extern char *msg_jobnew;
extern char *msg_resvQcreateFail;
extern char *msg_defproject;
extern char *msg_mom_reject_root_scripts;
extern int reject_root_scripts;
extern time_t time_now;
#ifndef PBS_MOM
extern void *svr_db_conn;
extern char *msg_max_no_minwt;
extern char *msg_min_gt_maxwt;
extern char *msg_nostf_resv;
extern char *msg_nostf_jobarray;
#endif
/* Private Functions in this file */
static job *locate_new_job(struct batch_request *preq, char *jobid);
#ifndef PBS_MOM /* SERVER only */
static void handle_qmgr_reply_to_resvQcreate(struct work_task *);
static int get_queue_for_reservation(resc_resv *);
static int ignore_attr(char *);
static int validate_place_req_of_job_in_reservation(job *pj);
/* To generate the job/resv id's locally */
void reset_svr_sequence_window(void);
long long next_svr_sequence_id = 0;
static char *pbs_o_que = "PBS_O_QUEUE=";
/**
* @brief
* validate_perm_res_in_select - checks to see if the resources
* appearing in select spec 'val' are valid based on
* "caller's" permission level (i.e. resc_access_perm).
* optionally checks if the resources exist based on val_exist parameter
*
* @param[in] val - select spec 'val'
* @param[in] val_exist - validate if res exists in server def
*
* @return error code
* @retval 0 : success
* @retval !0 : failure
* @retval PBSE_INVALSELECTRESC : 'resc_in_err' is also set to the name of the offending resource.
*
* @note
* NOTE: 'resc_in_err' is a malloced-string which is used and
* freed inside req_reject().
* So upon PBS_INVALSELECTRES or PBSE_UNKRESC return, be sure to
* issue req_reject().
*/
int
validate_perm_res_in_select(char *val, int val_exist)
{
char *chunk;
int nchk;
int nelem;
struct key_value_pair *pkvp;
int rc = 0;
int j;
resource_def *presc;
if (val == NULL)
return (0); /* nothing to validate */
chunk = parse_plus_spec(val, &rc); /* break '+' seperated substrings */
if (rc != 0)
return (rc);
while (chunk) {
if (parse_chunk(chunk, &nchk, &nelem, &pkvp, NULL) == 0) {
/* first check for any invalid resources in the select */
for (j = 0; j < nelem; ++j) {
presc = find_resc_def(svr_resc_def, pkvp[j].kv_keyw);
if (presc) {
if ((presc->rs_flags & resc_access_perm) == 0) {
if ((resc_in_err = strdup(pkvp[j].kv_keyw)) == NULL)
return PBSE_SYSTEM;
return PBSE_INVALSELECTRESC; /* for freeing resc_in_err please read "NOTE" above in function brief*/
}
} else if (val_exist) {
if ((resc_in_err = strdup(pkvp[j].kv_keyw)) == NULL)
return PBSE_SYSTEM;
return PBSE_UNKRESC; /* for freeing resc_in_err please read "NOTE" above in function brief*/
}
} /* for */
} /* if */
chunk = parse_plus_spec(NULL, &rc);
if (rc != 0)
return (rc);
} /* while */
return (0);
}
#endif
#define SET_RESC_SELECT 1
#define SET_RESC_PLACE 2
#ifndef PBS_MOM
/**
* @brief Generate and fill jobid of the new job
* Note: Consider directly modifying get_next_svr_sequence_id() if reservations
* also get sharded for multi-server
*
* @param[out] idbuf - buffer to fill job/resv id in
* @param[in] clusterid - cluster name (PBS_SERVER)
* @param[in] objtype - object type which specifies whether it is a normal/array job or reservation
* @param[in] resv_char - character representing type of reservation
*
* @return int
* @retval 0 for Success
* @retval 1 for Failure
*/
static int
generate_objid(char *idbuf, char *clusterid, int objtype, char resv_char)
{
if (idbuf == NULL || server_name == NULL || clusterid == NULL)
return 1;
if (objtype == MGR_OBJ_JOB)
sprintf(idbuf, "%lld.%s", next_svr_sequence_id, clusterid);
else if (objtype == MGR_OBJ_JOBARRAY_PARENT)
sprintf(idbuf, "%lld[].%s", next_svr_sequence_id, clusterid);
else if (objtype == MGR_OBJ_RESV)
sprintf(idbuf, "%c%lld.%s", resv_char, next_svr_sequence_id, clusterid);
return 0;
}
#endif /* #ifndef PBS_MOM */
/**
* @brief
* Queue Job Batch Request processing routine
*
* @param[in] - ptr to the decoded request
*
*/
void
req_quejob(struct batch_request *preq)
{
int created_here = 0;
int index;
char *jid;
attribute_def *pdef;
job *pj;
svrattrl *psatl;
int rc;
int sock = preq->rq_conn;
int resc_access_perm_save;
int implicit_commit = 0;
#ifndef PBS_MOM
int set_project = 0;
int i;
char jidbuf[PBS_MAXSVRJOBID + 1];
pbs_queue *pque;
char *qname;
char *result;
resource_def *prdefnod;
resource_def *prdefsel;
resource_def *prdefplc;
resource *presc;
conn_t *conn = NULL;
char *physhost = NULL;
#else
mom_hook_input_t hook_input;
mom_hook_output_t hook_output;
int hook_errcode = 0;
int hook_rc = 0;
char hook_buf[HOOK_MSG_SIZE];
hook *last_phook = NULL;
unsigned int hook_fail_action = 0;
#endif
char hook_msg[HOOK_MSG_SIZE];
/* set basic (user) level access permission */
resc_access_perm = ATR_DFLAG_USWR | ATR_DFLAG_Creat;
#ifndef PBS_MOM /* server server server server */
if (preq->prot != PROT_TPP) {
conn = get_conn(sock);
if (!conn) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
if (conn->cn_authen & PBS_NET_CONN_FORCE_QSUB_UPDATE) {
req_reject(PBSE_FORCE_QSUB_UPDATE, 0, preq);
conn->cn_authen &= ~PBS_NET_CONN_FORCE_QSUB_UPDATE;
return;
}
}
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
if (psatl->al_name == NULL || (!strcasecmp(psatl->al_name, ATTR_l) && psatl->al_resc == NULL)) {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
if (!strcasecmp(psatl->al_name, ATTR_l) &&
!strcasecmp(psatl->al_resc, "select") &&
((psatl->al_value != NULL) &&
(psatl->al_value[0] != '\0'))) {
if ((rc = validate_perm_res_in_select(psatl->al_value, 0)) != 0) {
req_reject(rc, 0, preq);
return;
}
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
switch (process_hooks(preq, hook_msg, sizeof(hook_msg),
pbs_python_set_interrupt)) {
case 0: /* explicit reject */
reply_text(preq, PBSE_HOOKERROR, hook_msg);
return;
case 1: /* explicit accept */
if (recreate_request(preq) == -1) { /* error */
/* we have to reject the request, as 'preq' */
/* may have been partly modified */
strcpy(hook_msg,
"queuejob event: rejected request");
log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_HOOK,
LOG_ERR, "", hook_msg);
reply_text(preq, PBSE_HOOKERROR, hook_msg);
return;
}
break;
case 2: /* no hook script executed - go ahead and accept event*/
break;
default:
log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_HOOK,
LOG_INFO, "", "queuejob event: accept req by default");
}
prdefsel = &svr_resc_def[RESC_SELECT];
prdefplc = &svr_resc_def[RESC_PLACE];
prdefnod = &svr_resc_def[RESC_NODES];
/*
* if the job id is supplied, the request had better be
* from another server
*/
if (preq->rq_fromsvr) {
/* from another server - accept the extra attributes */
resc_access_perm |= ATR_DFLAG_MGWR | ATR_DFLAG_SvWR;
jid = preq->rq_ind.rq_queuejob.rq_jid;
} else if (preq->rq_ind.rq_queuejob.rq_jid[0] != '\0') {
/* a job id is not allowed from a client */
req_reject(PBSE_IVALREQ, 0, preq);
return;
} else {
/* assign it a job id */
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
i = MGR_OBJ_JOB;
while (psatl) {
/* Ensure that array_indices_submitted has a proper */
/* value (non-"" and non-NULL) before asserting that */
/* current job is a job array. */
/* The Hook script can set value to "" meaning none */
/* was specified making it a normal (non job array) */
/* job. */
/* The value should never be NULL and if so, then */
/* it won't be a job array. */
if (!strcasecmp(psatl->al_name,
ATTR_array_indices_submitted) &&
((psatl->al_value != NULL) &&
(psatl->al_value[0] != '\0'))) {
i = MGR_OBJ_JOBARRAY_PARENT;
break;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
/* fetch job id locally*/
if ((next_svr_sequence_id = get_next_svr_sequence_id()) == -1) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
created_here = JOB_SVFLG_HERE;
if (generate_objid(jidbuf, server_name, i, '\0') != 0) {
req_reject(PBSE_INTERNAL, 0, preq);
return;
}
jid = jidbuf;
}
#else /* PBS_MOM mom mom mom mom mom mom*/
if (preq->rq_fromsvr) { /* must be from a server */
/* from another server - accept the extra attributes */
resc_access_perm |= ATR_DFLAG_MGWR | ATR_DFLAG_SvWR |
ATR_DFLAG_MOM;
jid = preq->rq_ind.rq_queuejob.rq_jid;
} else {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
#endif /* PBS_MOM all all all all all */
/* does job already exist, check both old and new jobs */
if ((pj = find_job(jid)) == NULL) {
pj = (job *) GET_NEXT(svr_newjobs);
while (pj) {
if (!strcasecmp(pj->ji_qs.ji_jobid, jid))
break;
pj = (job *) GET_NEXT(pj->ji_alljobs);
}
}
#ifndef PBS_MOM /* server server server server server server */
/*
* Check if the server is configured for history job info. If yes and
* server has the history job with same job id, then don't reject the
* queue request with PBSE_JOBEXIST but purge the history job(i.e. with
* job state 'M') from the server and accept queue request. If you have
* the real job, keeping its history info does not make any sense.
* Otherwise SERVER will continue to reject queue request if job already
* exists.
*/
if (pj != NULL) {
if ((svr_chk_history_conf()) &&
(check_job_state(pj, JOB_STATE_LTR_MOVED))) {
job_purge(pj);
} else {
/* server rejects the queue request */
req_reject(PBSE_JOBEXIST, 0, preq);
return;
}
}
/* find requested queue, is it there? */
qname = preq->rq_ind.rq_queuejob.rq_destin;
if ((*qname == '\0') || (*qname == '@')) { /* use default queue */
pque = get_dfltque();
rc = PBSE_QUENODFLT;
} else { /* else find the named queue */
pque = find_queuebyname(preq->rq_ind.rq_queuejob.rq_destin);
#ifdef NAS /* localmod 075 */
if (pque == NULL)
pque = find_resvqueuebyname(qname);
#endif /* localmod 075 */
rc = PBSE_UNKQUE;
}
if (pque == NULL) {
req_reject(rc, 0, preq); /* not there */
return;
}
/* create the job structure */
if ((pj = job_alloc()) == NULL) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
*pj->ji_qs.ji_fileprefix = '\0';
#else /* PBS_MOM mom mom mom mom*/
if (pj) {
/*
* An existing job - likely was checkpointed but in rare
* cases may result from a Server-Mom communication error
*
* Update run version from the new request so server will
* accept the obit when the job finishes.
*/
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
/* look for run count attribute */
index = find_attr(job_attr_idx, job_attr_def, psatl->al_name);
if (index == (int) JOB_ATR_run_version) {
set_jattr_str_slim(pj, index, psatl->al_value, psatl->al_resc);
break;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
/* if actually running, tell Server we already have it */
if (check_job_substate(pj, JOB_SUBSTATE_RUNNING)) {
req_reject(PBSE_JOBEXIST, 0, preq);
return;
}
/* if checkpointed, then keep old and skip rest of process */
if (pj->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) {
set_job_substate(pj, JOB_SUBSTATE_TRANSIN);
int prot = preq->prot;
if (reply_jobid(preq, pj->ji_qs.ji_jobid, BATCH_REPLY_CHOICE_Queue) == 0) {
delete_link(&pj->ji_alljobs);
if (pbs_idx_delete(jobs_idx, pj->ji_qs.ji_jobid) != PBS_IDX_RET_OK)
log_joberr(PBSE_INTERNAL, __func__, "Failed to remove checkpointed job from index", pj->ji_qs.ji_jobid);
append_link(&svr_newjobs, &pj->ji_alljobs, pj);
pj->ji_qs.ji_un_type = JOB_UNION_TYPE_NEW;
pj->ji_qs.ji_un.ji_newt.ji_fromsock = sock;
if (prot == PROT_TCP) {
pj->ji_qs.ji_un.ji_newt.ji_fromaddr = get_connectaddr(sock);
} else {
struct sockaddr_in *addr = tpp_getaddr(sock);
if (addr)
pj->ji_qs.ji_un.ji_newt.ji_fromaddr = (pbs_net_t) ntohl(addr->sin_addr.s_addr);
}
pj->ji_qs.ji_un.ji_newt.ji_scriptsz = 0;
} else {
close_client(sock); /* error on reply */
}
return;
}
/* unlink job from svr_alljobs since will be place on newjobs */
delete_link(&pj->ji_alljobs);
if (pbs_idx_delete(jobs_idx, pj->ji_qs.ji_jobid) != PBS_IDX_RET_OK)
log_joberr(PBSE_INTERNAL, __func__, "Failed to remove job from index", pj->ji_qs.ji_jobid);
} else {
char *namebuf;
char basename[MAXPATHLEN + 1] = {0};
/* if not already here, allocate job struct */
if ((pj = job_alloc()) == NULL) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
/*
*
* for MOM - rather than make up a hashname, we use the sent
* to us by the server as an attribute.
*/
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
if (!strcmp(psatl->al_name, ATTR_hashname)) {
strncpy(basename, psatl->al_value, MAXPATHLEN);
if (strlen(basename) <= PBS_JOBBASE)
strcpy(pj->ji_qs.ji_fileprefix, basename);
else
*pj->ji_qs.ji_fileprefix = '\0';
break;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
pbs_asprintf(&namebuf, "%s%s%s", path_jobs, basename, JOB_TASKDIR_SUFFIX);
if (mkdir(namebuf, 0700) == -1) {
pj->ji_qs.ji_un.ji_momt.ji_exitstat = -1;
if (*pj->ji_qs.ji_fileprefix == '\0' && *pj->ji_qs.ji_jobid == '\0') {
snprintf(pj->ji_qs.ji_jobid, sizeof(pj->ji_qs.ji_jobid), "%s", jid);
}
job_purge(pj);
req_reject(PBSE_SYSTEM, 0, preq);
free(namebuf);
return;
}
created_here = JOB_SVFLG_HERE;
free(namebuf);
}
#endif /* PBS_MOM */
(void) strcpy(pj->ji_qs.ji_jobid, jid);
pj->ji_qs.ji_svrflags = created_here;
pj->ji_qs.ji_un_type = JOB_UNION_TYPE_NEW;
/* decode attributes from request into job structure */
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
resc_access_perm_save = resc_access_perm; /* save perm */
while (psatl) {
/* identify the attribute by name */
index = find_attr(job_attr_idx, job_attr_def, psatl->al_name);
if (index < 0) {
/* didn`t recognize the name */
#ifndef PBS_MOM
index = JOB_ATR_UNKN; /* keep as "unknown" for now */
#else /* is PBS_MOM */
reply_badattr(PBSE_NOATTR, 1, psatl, preq);
return;
#endif /* PBS_MOM */
}
pdef = &job_attr_def[index];
#ifndef PBS_MOM
if (index == JOB_ATR_project) {
set_project = 1;
}
#endif
/* Is attribute not writeable by manager or by a server? */
/* Exempt attributes set by the hook script */
resc_access_perm = resc_access_perm_save; /* reset */
if ((psatl->al_flags & ATR_VFLAG_HOOK)) {
resc_access_perm = ATR_DFLAG_USWR |
ATR_DFLAG_OPWR |
ATR_DFLAG_MGWR |
ATR_DFLAG_SvWR |
ATR_DFLAG_Creat;
}
if (((pdef->at_flags & resc_access_perm) == 0)) {
job_purge(pj);
reply_badattr(PBSE_ATTRRO, 1, psatl, preq);
return;
}
/* decode attribute */
#ifndef PBS_MOM
if (index == JOB_ATR_create_resv_from_job) {
if (qname != NULL && *qname != '\0') {
resc_resv *presv;
presv = find_resv(qname);
if (presv) {
job_purge(pj);
req_reject(PBSE_RESV_FROM_RESVJOB, 0, preq);
return;
}
}
}
#endif
if ((index == JOB_ATR_resource) && (psatl->al_resc != NULL) && (strcmp(psatl->al_resc, "neednodes") == 0))
rc = 0;
else
rc = set_jattr_generic(pj, index, psatl->al_value, psatl->al_resc, INTERNAL);
#ifndef PBS_MOM
if (rc != 0) {
if (rc == PBSE_UNKRESC) {
/* unknown resources not allow in Exec queue */
if (pque->qu_qs.qu_type == QTYPE_Execution) {
job_purge(pj);
reply_badattr(rc, 1, psatl, preq);
return;
}
} else {
/* any other error is fatal */
job_purge(pj);
reply_badattr(rc, 1, psatl, preq);
return;
}
}
#else /* PBS_MOM MOM MOM MOM */
if (rc != 0) {
/* all errors are fatal for MOM */
job_purge(pj);
reply_badattr(rc, 1, psatl, preq);
return;
}
if (psatl->al_op == DFLT) {
attribute *attr = get_jattr(pj, index);
if (psatl->al_resc) {
resource *presc;
resource_def *prdef;
prdef = find_resc_def(svr_resc_def, psatl->al_resc);
if (prdef == 0) {
job_purge(pj);
reply_badattr(rc, 1, psatl, preq);
return;
}
presc = find_resc_entry(attr, prdef);
if (presc)
presc->rs_value.at_flags |= ATR_VFLAG_DEFLT;
} else {
attr->at_flags |= ATR_VFLAG_DEFLT;
}
}
#endif /* PBS_MOM */
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
#ifndef PBS_MOM
/* perform any at_action routine declared for the attributes */
for (i = 0; i < JOB_ATR_LAST; ++i) {
pdef = &job_attr_def[i];
if ((is_jattr_set(pj, i)) &&
(pdef->at_action)) {
rc = pdef->at_action(get_jattr(pj, i), pj, ATR_ACTION_NEW);
if (rc) {
job_purge(pj);
req_reject(rc, i, preq);
return;
}
}
}
#if defined(PBS_SECURITY) && (PBS_SECURITY == KRB5)
/* save gssapi/krb5 creds for this job */
if (conn && conn->cn_credid != NULL &&
conn->cn_auth_config != NULL &&
conn->cn_auth_config->auth_method != NULL && strcmp(conn->cn_auth_config->auth_method, AUTH_GSS_NAME) == 0) {
log_eventf(PBSEVENT_DEBUG, PBS_EVENTCLASS_SERVER, LOG_DEBUG, __func__,
"saving creds. conn is %d, cred id %s", preq->rq_conn, conn->cn_credid);
set_jattr_str_slim(pj, JOB_ATR_cred_id, conn->cn_credid, NULL);
if (is_sattr_set(SVR_ATR_acl_krb_submit_realms)) {
if (!acl_check(get_sattr(SVR_ATR_acl_krb_submit_realms), conn->cn_credid, ACL_Host)) {
job_purge(pj);
req_reject(PBSE_PERM, 0, preq);
return;
}
}
}
#endif
/*
* Now that the attributes have been decoded, we can setup some
* additional parameters and perform a few more checks.
*
* First, set some items based on who created the job...
*/
if (created_here) { /* created here */
int l;
char buf[256];
/* check that job has a jobname */
if ((is_jattr_set(pj, JOB_ATR_jobname)) == 0)
set_jattr_str_slim(pj, JOB_ATR_jobname, "none", NULL);
/* check resources in the Resource_List are valid job wide */
if (is_jattr_set(pj, JOB_ATR_resource)) {
int have_selectplace = 0;
resource_def *prdefbad;
presc = (resource *) GET_NEXT(get_jattr_list(pj, JOB_ATR_resource));
prdefbad = NULL;
while (presc) {
if (presc->rs_defin == prdefsel) {
have_selectplace |= SET_RESC_SELECT;
} else if (presc->rs_defin == prdefplc) {
have_selectplace |= SET_RESC_PLACE;
} else if ((presc->rs_defin == prdefnod) &&
(presc->rs_value.at_val.at_str != NULL)) {
/*
* if "nodes" is set and has non-NULL value,
* remember as potential bad resource
* if this appears along "select" or "place".
*/
prdefbad = presc->rs_defin;
} else if (
(presc->rs_defin->rs_flags & ATR_DFLAG_CVTSLT) &&
(is_attr_set(&presc->rs_value))) {
/*
* if this resource is not "select", "place",
* or "nodes", but is meant to appear inside
* a "select" line, then remember as potential
* bad resource if this appears along
* "select" or "place".
*/
prdefbad = presc->rs_defin;
}
presc = (resource *) GET_NEXT(presc->rs_link);
}
if (have_selectplace && prdefbad) {
if (prdefbad == prdefnod)
rc = PBSE_INVALNODEPLACE;
else
rc = PBSE_INVALJOBRESC;
if ((resc_in_err = strdup(prdefbad->rs_name)) == NULL)
rc = PBSE_SYSTEM;
job_purge(pj);
req_reject(rc, 0, preq);
return;
}
if (have_selectplace == SET_RESC_PLACE) {
/* cannot have place without select */
job_purge(pj);
req_reject(PBSE_PLACENOSELECT, 0, preq);
return;
}
}
/* check value of priority */
if (is_jattr_set(pj, JOB_ATR_priority)) {
if ((get_jattr_long(pj, JOB_ATR_priority) < -1024) ||
(get_jattr_long(pj, JOB_ATR_priority) > 1024)) {
job_purge(pj);
req_reject(PBSE_BADATVAL, 0, preq);
return;
}
}
/* set job owner attribute to user@host */
strcpy(buf, preq->rq_user);
strcat(buf, "@");
strcat(buf, preq->rq_host);
set_jattr_str_slim(pj, JOB_ATR_job_owner, buf, NULL);
if (conn) {
strcpy(buf, conn->cn_physhost);
set_jattr_str_slim(pj, JOB_ATR_submit_host, buf, NULL);
}
/* set create time */
set_jattr_l_slim(pj, JOB_ATR_ctime, time_now, SET);
/* set hop count = 1 */
set_jattr_l_slim(pj, JOB_ATR_hopcount, 1, SET);
/* need to set certain environmental variables per POSIX */
strcpy(buf, pbs_o_que);
strcat(buf, pque->qu_qs.qu_name);
if (conn && get_variable(pj, pbs_o_host) == NULL) {
strcat(buf, ",");
strcat(buf, pbs_o_host);
strcat(buf, "=");
strcat(buf, conn->cn_physhost);
}
set_jattr_generic(pj, JOB_ATR_variables, buf, NULL, INCR);
/* if JOB_ATR_outpath/JOB_ATR_errpath not set, set default */
if (!(is_jattr_set(pj, JOB_ATR_outpath))) {
set_jattr_str_slim(pj, JOB_ATR_outpath, prefix_std_file(pj, (int) 'o'), NULL);
} else {
l = strlen(get_jattr_str(pj, JOB_ATR_outpath));
if (l > 0) {
if (get_jattr_str(pj, JOB_ATR_outpath)[l - 1] == ':') {
cat_default_std(pj, (int) 'o', get_jattr_str(pj, JOB_ATR_outpath), &result);
if (result)
set_jattr_str_slim(pj, JOB_ATR_outpath, result, NULL);
}
}
}
if (!(is_jattr_set(pj, JOB_ATR_errpath))) {
set_jattr_str_slim(pj, JOB_ATR_errpath, prefix_std_file(pj, (int) 'e'), NULL);
} else {
l = strlen(get_jattr_str(pj, JOB_ATR_errpath));
if (l > 0) {
if (get_jattr_str(pj, JOB_ATR_errpath)[l - 1] == ':') {
cat_default_std(pj, (int) 'e', get_jattr_str(pj, JOB_ATR_errpath), &result);
if (result)
set_jattr_str_slim(pj, JOB_ATR_errpath, result, NULL);
}
}
}
if ((get_jattr_str(pj, JOB_ATR_outpath) == 0) ||
(get_jattr_str(pj, JOB_ATR_errpath) == 0)) {
job_purge(pj);
req_reject(PBSE_NOATTR, 0, preq);
return;
}
if ((is_jattr_set(pj, JOB_ATR_interactive)) && (is_jattr_set(pj, JOB_ATR_array_indices_submitted))) {
job_purge(pj);
req_reject(PBSE_NOSUP, 0, preq);
return;
}
/* Interactive jobs are not rerunable */
if ((is_jattr_set(pj, JOB_ATR_interactive)) && get_jattr_long(pj, JOB_ATR_interactive))
set_jattr_l_slim(pj, JOB_ATR_rerunable, 0, SET);
if (!is_jattr_set(pj, JOB_ATR_project))
set_jattr_str_slim(pj, JOB_ATR_project, PBS_DEFAULT_PROJECT, NULL);
} else { /* job was created elsewhere and moved here */
/* make sure job_owner is set, error if not */
if (!is_jattr_set(pj, JOB_ATR_job_owner)) {
job_purge(pj);
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
/* increment hop count */
set_jattr_l_slim(pj, JOB_ATR_hopcount, 1, INCR);
if (get_jattr_long(pj, JOB_ATR_hopcount) > PBS_MAX_HOPCOUNT) {
job_purge(pj);
req_reject(PBSE_HOPCOUNT, 0, preq);
return;
}
/* make sure that if job belonged to an advance reservation
* queue, that old information is wiped out. If being moved
* into an advance reservation queue, the reservation's ID
* gets attached later in the code
*/
set_attr_generic(get_jattr(pj, JOB_ATR_reserve_ID), &job_attr_def[(int) JOB_ATR_reserve_ID], NULL, NULL, INTERNAL);
}
/* set up at_server attribute for status */
set_jattr_str_slim(pj, JOB_ATR_at_server, pbs_server_name, NULL);
/* If enabled, check the server's required cred type */
if (is_sattr_set(SVR_ATR_ReqCredEnable) &&
get_sattr_long(SVR_ATR_ReqCredEnable) &&
is_sattr_set(SVR_ATR_ReqCred)) {
char *reqc = get_sattr_str(SVR_ATR_ReqCred);
char *jobc = get_jattr_str(pj, JOB_ATR_cred);
/*
** The server requires a cred, if job has none, or
** it is the wrong one, reject.
*/
if (!is_jattr_set(pj, JOB_ATR_cred) || strcmp(reqc, jobc) != 0) {
job_purge(pj);
req_reject(PBSE_BADCRED, 0, preq);
return;
}
}
if (conn) {
physhost = conn->cn_physhost;
}
/*
* See if the job is qualified to go into the requested queue.
* Note, if an execution queue, then ji_qs.ji_un.ji_exect is set up
*
* svr_chkque is called way down here because it needs to have the
* job structure and attributes already set up.
*/
rc = svr_chkque(pj, pque, get_jattr_str(pj, JOB_ATR_submit_host), physhost, MOVE_TYPE_Move);
if (rc) {
if (pj->ji_clterrmsg)
reply_text(preq, rc, pj->ji_clterrmsg);
else
req_reject(rc, 0, preq);
job_purge(pj);
return;
}
(void) strcpy(pj->ji_qs.ji_queue, pque->qu_qs.qu_name);
/* Is job being submitted to a reservation queue?
* If yes, have the job point to the resc_resv object and
* update the job attribute JOB_ATR_reservation.
*
* Also check for conflict for job and reservation place spec
*/
if (pque->qu_resvp) {
free_jattr(pj, JOB_ATR_reserve_ID);
set_jattr_str_slim(pj, JOB_ATR_reserve_ID, pque->qu_resvp->ri_qs.ri_resvID, NULL);
pj->ji_myResv = pque->qu_resvp;
if (!validate_place_req_of_job_in_reservation(pj)) {
job_purge(pj);
psatl = (svrattrl *) GET_NEXT(
preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
if (!strcasecmp(psatl->al_name, ATTR_l) &&
!strcasecmp(psatl->al_resc, "place")) {
reply_badattr(PBSE_JOBINRESV_CONFLICT, 1,
psatl, preq);
return;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
if (!psatl) {
req_reject(PBSE_JOBINRESV_CONFLICT, 0, preq);
return;
}
}
}
set_jattr_l_slim(pj, JOB_ATR_substate, JOB_SUBSTATE_TRANSIN, SET);
/* action routine for select does not have reservation data hence
* returns without doing checks. Checks are called now.
*/
presc = find_resc_entry(get_jattr(pj, JOB_ATR_resource), prdefsel);
if (presc) {
rc = apply_aoe_inchunk_rules(presc, get_jattr(pj, JOB_ATR_resource),
pj, PARENT_TYPE_JOB);
if (rc) {
job_purge(pj);
req_reject(rc, 0, preq);
return;
}
}
#endif /* not PBS_MOM */
/* set remaining job structure elements */
set_job_state(pj, JOB_STATE_LTR_TRANSIT);
set_job_substate(pj, JOB_SUBSTATE_TRANSIN);
set_jattr_l_slim(pj, JOB_ATR_mtime, time_now, SET);
pj->ji_qs.ji_un_type = JOB_UNION_TYPE_NEW;
pj->ji_qs.ji_un.ji_newt.ji_fromsock = sock;
pj->ji_qs.ji_un.ji_newt.ji_scriptsz = 0;
#ifdef PBS_MOM
if ((is_jattr_set(pj, JOB_ATR_executable)) &&
(reject_root_scripts == TRUE) &&
(is_jattr_set(pj, JOB_ATR_euser)) &&
(get_jattr_str(pj, JOB_ATR_euser) != NULL)) {
#ifdef WIN32
/* equivalent of root */
if (isAdminPrivilege(get_jattr_str(pj, JOB_ATR_euser)))
#else
struct passwd *pwdp;
pwdp = getpwnam(get_jattr_str(pj, JOB_ATR_euser));
if ((pwdp != NULL) && (pwdp->pw_uid == 0))
#endif
{
log_err(-1, __func__, msg_mom_reject_root_scripts);
reply_text(preq, PBSE_MOM_REJECT_ROOT_SCRIPTS, msg_mom_reject_root_scripts);
job_purge(pj);
return;
}
}
mom_hook_input_init(&hook_input);
hook_input.pjob = pj;
mom_hook_output_init(&hook_output);
hook_output.reject_errcode = &hook_errcode;
hook_output.last_phook = &last_phook;
hook_output.fail_action = &hook_fail_action;
switch ((hook_rc = mom_process_hooks(HOOK_EVENT_EXECJOB_BEGIN, PBS_MOM_SERVICE_NAME,
mom_host, &hook_input, &hook_output, hook_buf, sizeof(hook_buf), 1))) {
case 1: /* explicit accept */
break;
case 2: /* no hook script executed - go ahead and accept event*/
break;
default:
/* a value of '0' means explicit reject encountered. */
if (hook_rc != 0) {
/* we've hit an internal error (malloc error, full disk, etc...), so */
/* treat this now like a hook error so hook fail_action */
/* will be consulted. */
/* Before, behavior of an internal error was to ignore it! */
hook_errcode = PBSE_HOOKERROR;
}
if (hook_errcode == PBSE_HOOKERROR) { /* error */
/* piggy back the hook_name in the message */
/* to be stripped out by the server upon */
/* processing hook fail_action */
snprintf(hook_msg, sizeof(hook_msg), "%s,%.*s",
last_phook ? last_phook->hook_name : "",
(int) (sizeof(hook_msg) - (last_phook ? strlen(last_phook->hook_name) : 0) - 2),
(hook_rc == 0) ? hook_buf : "internal error");
} else {
snprintf(hook_msg, sizeof(hook_msg), ",%.*s",
(int) (sizeof(hook_msg) - 2), hook_buf);
}
job_purge(pj);
reply_text(preq, hook_errcode, hook_msg);
return;
}
#endif
/* check implicit commit only not blocking job */
if ((is_jattr_set(pj, JOB_ATR_block)) == 0)
implicit_commit = ((preq->rq_extend) && (strstr(preq->rq_extend, EXTEND_OPT_IMPLICIT_COMMIT)));
/* acknowledge the request with the job id */
if (!implicit_commit) {
if (preq->prot == PROT_TCP) {
pj->ji_qs.ji_un.ji_newt.ji_fromaddr = get_connectaddr(sock);
/* acknowledge the request with the job id */
if (reply_jobid(preq, pj->ji_qs.ji_jobid, BATCH_REPLY_CHOICE_Queue) != 0) {
/* reply failed, purge the job and close the connection */
close_client(sock);
job_purge(pj);
return;
}
} else {
struct sockaddr_in *addr = tpp_getaddr(sock);
if (addr)
pj->ji_qs.ji_un.ji_newt.ji_fromaddr = (pbs_net_t) ntohl(addr->sin_addr.s_addr);
free_br(preq);
/* No need of acknowledge for TPP */
}
}
#ifndef PBS_MOM
if (set_project && (is_jattr_set(pj, JOB_ATR_project)) &&
(strcmp(get_jattr_str(pj, JOB_ATR_project), PBS_DEFAULT_PROJECT) == 0))
log_eventf(PBSEVENT_DEBUG4, PBS_EVENTCLASS_JOB, LOG_INFO, pj->ji_qs.ji_jobid, msg_defproject, ATTR_project, PBS_DEFAULT_PROJECT);
#endif
if (implicit_commit) {
req_commit_now(preq, pj);
return;
}
/* link job into server's new jobs list request */
append_link(&svr_newjobs, &pj->ji_alljobs, pj);
#ifndef PBS_MOM
{
job *pjob;
int myport, port;
char *myhost, *host;
/*
** If the JOB_ATR_block attribute is set, look through the
** other jobs to make sure the host/port combo is unique.
*/
if ((is_jattr_set(pj, JOB_ATR_block)) == 0)
return;
myhost = get_jattr_str(pj, JOB_ATR_submit_host);
if (myhost == NULL)
return;
myport = (int) get_jattr_long(pj, JOB_ATR_block);
if (myport == 0)
return;
for (pjob = (job *) GET_NEXT(svr_alljobs);
pjob != NULL;
pjob = (job *) GET_NEXT(pjob->ji_alljobs)) {
if (pjob == pj)
continue;
if (!is_jattr_set(pjob, JOB_ATR_block))
continue;
port = (int) get_jattr_long(pjob, JOB_ATR_block);
if (port != myport)
continue;
host = get_jattr_str(pjob, JOB_ATR_submit_host);
if (host == NULL)
continue;
if (strcmp(host, myhost) != 0)
continue;
/* we found a job with the same host/port */
sprintf(log_buffer,
"job %s has duplicate BLOCK host %s port %d",
pjob->ji_qs.ji_jobid, host, port);
log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_ERR,
pj->ji_qs.ji_jobid, log_buffer);
/* unset the old job's JOB_ATR_block */
set_jattr_l_slim(pjob, JOB_ATR_block, 0, SET);
mark_jattr_not_set(pjob, JOB_ATR_block);
}
}
#endif /* not PBS_MOM */
}
/**
* @brief
* req_jobcredential - receive a set of credentials to be used by the job
*
* @param[in] preq - ptr to the decoded request
*/
void
req_jobcredential(struct batch_request *preq)
{
job *pj;
int type;
char *cred;
size_t len;
DBPRT(("%s: entered\n", __func__))
pj = locate_new_job(preq, NULL);
if (pj == NULL) {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
if (!check_job_substate(pj, JOB_SUBSTATE_TRANSIN)) {
delete_link(&pj->ji_alljobs);
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
#ifndef PBS_MOM
if (svr_authorize_jobreq(preq, pj) == -1) {
req_reject(PBSE_PERM, 0, preq);
return;
}
#endif /* PBS_MOM */
type = pj->ji_extended.ji_ext.ji_credtype =
preq->rq_ind.rq_jobcred.rq_type;
cred = preq->rq_ind.rq_jobcred.rq_data;
len = (size_t) preq->rq_ind.rq_jobcred.rq_size;
switch (type) {
default:
if (write_cred(pj, cred, len) == -1) {
delete_link(&pj->ji_alljobs);
req_reject(PBSE_SYSTEM, 0, preq);
} else
reply_ack(preq);
break;
}
return;
}
/**
* @brief
* Receive job script section
* @par Functionality:
* For Mom, each section is appended to the file
* For Server, its appended to the ji_script member
* of the job structure, to be later saved to the DB
*
* @param[in,out] preq - Pointer to batch request structure
*/
void
req_jobscript(struct batch_request *preq)
{
job *pj;
#ifdef PBS_MOM
int filemode = 0700;
int fds;
char namebuf[MAXPATHLEN];
#else
char *temp;
u_Long size;
#endif
pj = locate_new_job(preq, NULL);
if (pj == NULL) {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
if (!check_job_substate(pj, JOB_SUBSTATE_TRANSIN)) {
delete_link(&pj->ji_alljobs);
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
#ifndef PBS_MOM
if (svr_authorize_jobreq(preq, pj) == -1) {
req_reject(PBSE_PERM, 0, preq);
return;
}
#else
/* mom - if job has been checkpointed, discard script,already have it */
if (pj->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) {
/* do nothing, ignore script */
reply_ack(preq);
return;
}
#endif /* PBS_MOM */
#ifdef PBS_MOM
if (reject_root_scripts == TRUE) {
if (is_jattr_set(pj, JOB_ATR_euser) && get_jattr_str(pj, JOB_ATR_euser) != NULL) {
#ifdef WIN32
/* equivalent of root */
if (!isAdminPrivilege(get_jattr_str(pj, JOB_ATR_euser)))
#else
struct passwd *pwdp;
pwdp = getpwnam(get_jattr_str(pj, JOB_ATR_euser));
if ((pwdp != NULL) && (pwdp->pw_uid == 0))
#endif
{
log_err(-1, "req_jobscript",
msg_mom_reject_root_scripts);
delete_link(&pj->ji_alljobs);
req_reject(PBSE_MOM_REJECT_ROOT_SCRIPTS, 0, preq);
return;
}
}
}
(void) strcpy(namebuf, path_jobs);
if (*pj->ji_qs.ji_fileprefix != '\0')
(void) strcat(namebuf, pj->ji_qs.ji_fileprefix);
else
(void) strcat(namebuf, pj->ji_qs.ji_jobid);
(void) strcat(namebuf, JOB_SCRIPT_SUFFIX);
if (pj->ji_qs.ji_un.ji_newt.ji_scriptsz == 0) {
fds = open(namebuf, O_WRONLY | O_CREAT, filemode);
} else {
fds = open(namebuf, O_WRONLY | O_APPEND, filemode);
}
if (fds < 0) {
log_err(errno, "req_jobscript", msg_script_open);
delete_link(&pj->ji_alljobs);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
#ifdef WIN32
secure_file2(namebuf, "Administrators", READS_MASK | WRITES_MASK | STANDARD_RIGHTS_REQUIRED, "Everyone", READS_MASK | READ_CONTROL);
setmode(fds, O_BINARY);
#endif /* WIN32 */
if (write(fds, preq->rq_ind.rq_jobfile.rq_data,
(unsigned) preq->rq_ind.rq_jobfile.rq_size) !=
preq->rq_ind.rq_jobfile.rq_size) {
log_err(errno, "req_jobscript", msg_script_write);
delete_link(&pj->ji_alljobs);
req_reject(PBSE_SYSTEM, 0, preq);
(void) close(fds);
return;
}
(void) close(fds);
#else /* server - server - server - server */
/* add the script to the job */
size = get_bytes_from_attr(&attr_jobscript_max_size);
if (pj->ji_qs.ji_un.ji_newt.ji_scriptsz + preq->rq_ind.rq_jobfile.rq_size > size) {
job_purge(pj);
req_reject(PBSE_JOBSCRIPTMAXSIZE, 0, preq);
return;
}
temp = realloc(pj->ji_script, pj->ji_qs.ji_un.ji_newt.ji_scriptsz +
preq->rq_ind.rq_jobfile.rq_size + 1);
if (!temp) {
job_purge(pj);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
pj->ji_script = temp;
memmove(pj->ji_script + pj->ji_qs.ji_un.ji_newt.ji_scriptsz,
preq->rq_ind.rq_jobfile.rq_data,
(size_t) preq->rq_ind.rq_jobfile.rq_size);
#endif
pj->ji_qs.ji_un.ji_newt.ji_scriptsz += preq->rq_ind.rq_jobfile.rq_size;
#ifndef PBS_MOM
pj->ji_script[pj->ji_qs.ji_un.ji_newt.ji_scriptsz] = '\0';
#endif
pj->ji_qs.ji_svrflags = (pj->ji_qs.ji_svrflags & ~JOB_SVFLG_CHKPT) |
JOB_SVFLG_SCRIPT; /* has a script file */
reply_ack(preq);
}
#ifndef PBS_MOM
/* the following is for the server only, MOM has her own version below */
/**
* @brief
* req_mvjobfile - receive a job file
* This request is used to move a file associated with a job, typically
* the standard output or error, between a server and a server or from
* a mom back to a server. For a server, the destination is alway
* within the spool directory.
*
* @param[in,out] preq - ptr to the decoded request
*/
void
req_mvjobfile(struct batch_request *preq)
{
int fds;
char namebuf[MAXPATHLEN + 1];
job *pj;
mode_t cur_mask;
struct stat sb;
pj = locate_new_job(preq, NULL);
if (pj == NULL)
pj = find_job(preq->rq_ind.rq_jobfile.rq_jobid);
if ((preq->rq_fromsvr == 0) || (pj == NULL)) {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
(void) strcpy(namebuf, path_spool);
if (*pj->ji_qs.ji_fileprefix != '\0')
(void) strcat(namebuf, pj->ji_qs.ji_fileprefix);
else
(void) strcat(namebuf, pj->ji_qs.ji_jobid);
switch ((enum job_file) preq->rq_ind.rq_jobfile.rq_type) {
case StdOut:
(void) strcat(namebuf, JOB_STDOUT_SUFFIX);
break;
case StdErr:
(void) strcat(namebuf, JOB_STDERR_SUFFIX);
break;
case Chkpt:
(void) strcat(namebuf, JOB_CKPT_SUFFIX);
break;
default:
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
/* check symlinks */
if (lstat(namebuf, &sb) == 0) {
/* if it exists, the file must be a prior copy which means */
/* it must be a regular file and owned by me (root) */
if (((sb.st_mode & S_IFMT) != S_IFREG) ||
(sb.st_nlink != 1) ||
(sb.st_uid != 0)) {
/* this does not meet the above conditions */
/* someone may be trying to hack in a link to */
/* cause the link target to be overwritten */
/* lets log it and leave the file as evidence */
log_suspect_file(__func__, "wrong type or owner", namebuf, &sb);
}
}
if (preq->rq_ind.rq_jobfile.rq_sequence == 0) {
char ntmpbuf[MAXPATHLEN + 1];
/* receiving first piece, so create new file securely */
/* will discard any existing file (via rename) */
snprintf(ntmpbuf, sizeof(ntmpbuf), "%s", namebuf);
if (strlen(ntmpbuf) > (sizeof(ntmpbuf) - 8))
ntmpbuf[sizeof(ntmpbuf) - 8] = '\0';
strcat(ntmpbuf, "XXXXXX"); /* template for mkstemp() */
cur_mask = umask(077); /* force to create -rw------ */
fds = mkstemp(ntmpbuf);
(void) umask(cur_mask);
if (fds != -1) {
/* now rename to the filename we want */
if (rename(ntmpbuf, namebuf) == -1) {
close(fds);
unlink(ntmpbuf);
fds = -1;
}
}
} else {
/* receiving a follow-on chunk of data, file */
/* should already exist as regular file */
fds = open(namebuf, O_WRONLY | O_APPEND | O_Sync, 0600);
if (fds != -1) {
if (lstat(namebuf, &sb) == 0) {
/* if exists, file must be a regular file and be */
/* owned by me (root) */
if (((sb.st_mode & S_IFMT) != S_IFREG) ||
(sb.st_nlink != 1) ||
(sb.st_uid != 0)) {
/* this does not meet the above conditions */
log_suspect_file(__func__, "wrong type or owner",
namebuf, &sb);
close(fds);
unlink(namebuf);
fds = -1;
}
} else {
sprintf(log_buffer, "unable to lstat %s", namebuf);
log_err(errno, __func__, log_buffer);
close(fds);
fds = -1;
}
}
}
if (fds < 0) {
log_err(errno, "req_mvjobfile", msg_script_open);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
if (write(fds, preq->rq_ind.rq_jobfile.rq_data,
(unsigned) preq->rq_ind.rq_jobfile.rq_size) !=
preq->rq_ind.rq_jobfile.rq_size) {
log_err(errno, "req_jobfile", msg_script_write);
req_reject(PBSE_SYSTEM, 0, preq);
(void) close(fds);
return;
}
(void) close(fds);
reply_ack(preq);
}
#else /* PBS_MOM - MOM MOM MOM */
/**
* @brief
* req_mvjobfile - move the specifled job standard files
* This is MOM's version. The files are owned by the user and placed
* in either the spool area or the user's home directory depending
* on the compile option, see std_file_name().
*
* @param[in,out] preq - ptr to the decoded request
*/
void
req_mvjobfile(struct batch_request *preq)
{
int fds;
enum job_file jft;
int oflag;
job *pj;
struct passwd *check_pwd(job *);
jft = (enum job_file) preq->rq_ind.rq_jobfile.rq_type;
if (preq->rq_ind.rq_jobfile.rq_sequence == 0)
oflag = O_CREAT | O_WRONLY | O_TRUNC;
else
oflag = O_CREAT | O_WRONLY | O_APPEND;
pj = locate_new_job(preq, NULL);
if (pj == NULL)
pj = find_job(preq->rq_ind.rq_jobfile.rq_jobid);
if (pj == NULL) {
req_reject(PBSE_UNKJOBID, 0, preq);
return;
}
/* this call sets up home/uid/gid information for the job */
if (check_pwd(pj) == NULL) {
req_reject(PBSE_MOMREJECT, 0, preq);
return;
}
if ((is_jattr_set(pj, JOB_ATR_sandbox)) &&
(strcasecmp(get_jattr_str(pj, JOB_ATR_sandbox), "PRIVATE") == 0)) {
/* have a private sandbox which must be recreated */
/* prior to copying standard out and err back */
char *pbs_jobdir;
pbs_jobdir = jobdirname(pj->ji_qs.ji_jobid,
pj->ji_grpcache->gc_homedir);
/* call mkjobdir() with a NULL for the environment entry */
/* We are not at a point where we can setup the job's */
/* environment and mkjobdir() will be called again in */
/* start_exec where the permissions are reset to match */
/* the user's umask and the environment is built. */
#ifdef WIN32
if (mkjobdir(pj->ji_qs.ji_jobid,
pbs_jobdir,
(pj->ji_user != NULL) ? pj->ji_user->pw_name : NULL,
INVALID_HANDLE_VALUE) != 0) {
sprintf(log_buffer, "unable to create the job directory %s", pbs_jobdir);
log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO, pj->ji_qs.ji_jobid, log_buffer);
req_reject(PBSE_MOMREJECT, 0, preq);
return;
}
#else
if (mkjobdir(pj->ji_qs.ji_jobid,
pbs_jobdir,
pj->ji_grpcache->gc_uid,
pj->ji_grpcache->gc_gid) != 0) {
sprintf(log_buffer, "unable to create the job directory %s", pbs_jobdir);
log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO, pj->ji_qs.ji_jobid, log_buffer);
req_reject(PBSE_MOMREJECT, 0, preq);
return;
}
#endif /* WIN32 */
}
if ((fds = open_std_file(pj, jft, oflag,
pj->ji_grpcache->gc_gid)) < 0) {
req_reject(PBSE_MOMREJECT, 0, preq);
return;
}
if (write(fds, preq->rq_ind.rq_jobfile.rq_data,
preq->rq_ind.rq_jobfile.rq_size) !=
preq->rq_ind.rq_jobfile.rq_size)
req_reject(PBSE_SYSTEM, 0, preq);
else {
reply_ack(preq);
}
(void) close(fds);
}
#endif /* PBS_MOM */
/**
* @brief
* Parse the next message type and parameter from extend field
*
* @param[in] extend - extend field
* @param[out] msg_type - batch request type
* @param[out] msg_params - this array will be filled with parameter in the order they received
* parameters has to be freed by the caller.
* @param[in] param_arr_sz - size of the parameter array.
*
* @return int
* @return 0 success
* @retval -1 failure, memory allocation / input array is not of sufficient size
*/
int
parse_next_msg(char *extend, int *msg_type, char **msg_params, int param_arr_sz)
{
int i, j;
char **arr;
char *pc;
if ((arr = break_comma_list(extend)) == NULL)
return -1;
for (i = 0; arr[i]; i++) {
if ((pc = strstr(arr[i], EXTEND_OPT_NEXT_MSG_TYPE))) {
*msg_type = strtol(pc + strlen(EXTEND_OPT_NEXT_MSG_TYPE) + 1, NULL, 10);
break;
}
}
for (j = 0; arr[i]; i++) {
if ((pc = strstr(arr[i], EXTEND_OPT_NEXT_MSG_PARAM))) {
if (j >= param_arr_sz) {
free_str_array(arr);
return -1;
} else {
msg_params[j++] = strdup(pc + strlen(EXTEND_OPT_NEXT_MSG_PARAM) + 1);
}
}
}
if (j < param_arr_sz)
msg_params[j] = NULL;
free_str_array(arr);
return 0;
}
/**
* @brief
* Commit ownership of job
* @par Functionality:
* Set state of job to JOB_STATE_LTR_QUEUED (or Held or Waiting) and
* enqueue the job into its destination queue.
*
* @param[in] preq - The batch request structure
* @param[in] pj - Pointer to the job structure
*
*/
void
req_commit_now(struct batch_request *preq, job *pj)
{
#ifndef PBS_MOM
char newstate;
int newsub;
pbs_queue *pque;
int rc;
pbs_db_jobscr_info_t jobscr;
pbs_db_obj_info_t obj;
long long time_usec;
struct timeval tval;
void *conn = (void *) svr_db_conn;
char *runjob_extend = NULL;
struct batch_request *preq_runjob = NULL;
#endif
if (!check_job_substate(pj, JOB_SUBSTATE_TRANSIN)) {
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
set_job_state(pj, JOB_STATE_LTR_TRANSIT);
set_job_substate(pj, JOB_SUBSTATE_TRANSICM);
#ifdef PBS_MOM /* MOM only */
/* move job from new job list to "all" job list, set to running state */
delete_link(&pj->ji_alljobs);
if (pbs_idx_insert(jobs_idx, pj->ji_qs.ji_jobid, pj) != PBS_IDX_RET_OK) {
log_joberr(PBSE_INTERNAL, __func__, "Failed insert job in index", pj->ji_qs.ji_jobid);
req_reject(PBSE_INTERNAL, 0, preq);
job_purge(pj);
return;
}
append_link(&svr_alljobs, &pj->ji_alljobs, pj);
/*
** Set JOB_SVFLG_HERE to indicate that this is Mother Superior.
*/
pj->ji_qs.ji_svrflags |= JOB_SVFLG_HERE;
set_job_state(pj, JOB_STATE_LTR_RUNNING);
set_job_substate(pj, JOB_SUBSTATE_PRERUN);
pj->ji_qs.ji_un_type = JOB_UNION_TYPE_MOM;
if (preq->prot) {
struct sockaddr_in *addr = tpp_getaddr(preq->rq_conn);
if (addr)
pj->ji_qs.ji_un.ji_momt.ji_svraddr = (pbs_net_t) ntohl(addr->sin_addr.s_addr);
} else
pj->ji_qs.ji_un.ji_momt.ji_svraddr = get_connectaddr(preq->rq_conn);
pj->ji_qs.ji_un.ji_momt.ji_exitstat = 0;
if ((pj->ji_qs.ji_svrflags & (JOB_SVFLG_CHKPT | JOB_SVFLG_ChkptMig)) == 0) {
pj->ji_qs.ji_stime = time_now; /* start of walltime */
set_jattr_l_slim(pj, JOB_ATR_stime, time_now, SET);
}
/*
* For MOM - reply to the request and start up the job
* any errors will be dealt with via the mechanism
* used for a terminated job
*/
(void) reply_jobid(preq, pj->ji_qs.ji_jobid, BATCH_REPLY_CHOICE_Commit);
job_save(pj);
start_exec(pj);
/* The ATR_VFLAG_MODIFY bit for several attributes used to be
* set here. Now we rely on these bits to be set when and where
* an attribute is modified. Several of these are also set in
* record_finish_exec().
*/
#else /* PBS_SERVER */
if (svr_authorize_jobreq(preq, pj) == -1) {
req_reject(PBSE_PERM, 0, preq);
return;
}
/* Set Server level entity usage */
if ((rc = account_entity_limit_usages(pj, NULL, NULL, INCR, ETLIM_ACC_ALL)) != 0) {
job_purge(pj);
req_reject(rc, 0, preq);
return;
}
/* remove job for the server new job list, set state, and enqueue it */
delete_link(&pj->ji_alljobs);
svr_evaljobstate(pj, &newstate, &newsub, 1);
svr_setjobstate(pj, newstate, newsub);
gettimeofday(&tval, NULL);
time_usec = (tval.tv_sec * 1000000L) + tval.tv_usec;
/* set the queue rank attribute */
set_jattr_ll_slim(pj, JOB_ATR_qrank, time_usec, SET);
if (preq->rq_type == PBS_BATCH_Commit)
runjob_extend = preq->rq_extend;
if ((rc = svr_enquejob(pj, runjob_extend)) != 0) {
job_purge(pj);
req_reject(rc, 0, preq);
return;
}
account_jobstr(pj, PBS_ACCT_QUEUE);
/* Make things faster by writing job only once here - at commit time */
if (job_save_db(pj)) {
job_purge(pj);
req_reject(PBSE_SAVE_ERR, 0, preq);
return;
}
if (pj->ji_script) {
strcpy(jobscr.ji_jobid, pj->ji_qs.ji_jobid);
jobscr.script = pj->ji_script;
obj.pbs_db_obj_type = PBS_DB_JOBSCR;
obj.pbs_db_un.pbs_db_jobscr = &jobscr;
if (pbs_db_save_obj(conn, &obj, OBJ_SAVE_NEW) != 0) {
job_purge(pj);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
free(pj->ji_script);
pj->ji_script = NULL;
}
/* Now, no need to save server here because server
has already saved in the get_next_svr_sequence_id() */
/*
* if the job went into a Route (push) queue that has been started,
* try once to route it to give immediate feedback as a courtsey
* to the user.
*/
pque = pj->ji_qhdr;
if ((preq->rq_fromsvr == 0) &&
(pque->qu_qs.qu_type == QTYPE_RoutePush) &&
(get_qattr_long(pque, QA_ATR_Started) != 0)) {
if ((rc = job_route(pj)) != 0) {
job_purge(pj);
req_reject(rc, 0, preq);
return;
}
}
/* need to print message first, before request goes away */
log_eventf(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO, pj->ji_qs.ji_jobid,
msg_jobnew, preq->rq_user, preq->rq_host,
get_jattr_str(pj, JOB_ATR_job_owner),
get_jattr_str(pj, JOB_ATR_jobname),
pj->ji_qhdr->qu_qs.qu_name);
/* Allocate a new batch request to use for runjob as we will be freeing preq in reply_jobid() */
if (runjob_extend) {
char *param_arr[1];
if ((preq_runjob = copy_br(preq)) == NULL) {
job_purge(pj);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
if (parse_next_msg(runjob_extend, &preq_runjob->rq_type, param_arr, 1) == -1) {
job_purge(pj);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
strcpy(preq_runjob->rq_ind.rq_run.rq_jid, pj->ji_qs.ji_jobid);
preq_runjob->rq_ind.rq_run.rq_destin = param_arr[0];
preq_runjob->tpp_ack = 1;
preq_runjob->tppcmd_msgid = strdup(preq->tppcmd_msgid);
}
/* acknowledge the request with the job id */
if ((rc = reply_jobid(preq, pj->ji_qs.ji_jobid, BATCH_REPLY_CHOICE_Commit))) {
log_eventf(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_ERR, pj->ji_qs.ji_jobid, "Failed to reply with Job Id, error %d", rc);
job_purge(pj);
return;
}
if ((pj->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0)
issue_track(pj); /* notify creator where job is */
if (!preq_runjob)
return;
req_runjob(preq_runjob);
#endif /* PBS_SERVER */
}
/**
* @brief
* locate job and call req_commit_now
*
* @param[in] preq - The batch request structure
*
*/
void
req_commit(struct batch_request *preq)
{
job *pj;
pj = locate_new_job(preq, preq->rq_ind.rq_commit);
if (pj == NULL) {
req_reject(PBSE_UNKJOBID, 0, preq);
return;
}
req_commit_now(preq, pj);
}
/**
* @brief
* locate_new_job - locate a "new" job which has been set up req_quejob on
* the servers new job list.
*
* @par Functionality:
* This function is used by the sub-requests which make up the global
* "Queue Job Request" to locate the job structure.
*
* If the jobid is specified (will be for rdytocommit and commit, but not
* for script), we search for a matching jobid.
*
* The job must (also) match the socket specified and the host associated
* with the socket unless ji_fromsock == -1, then its a recovery situation.
*
* @param[in] preq - The batch request structure
* @param[in] jobid - Job Id which needs to be located
*
* @return job structure associated with jobid.
*/
static job *
locate_new_job(struct batch_request *preq, char *jobid)
{
job *pj;
int sock = -1;
pbs_net_t conn_addr = 0;
if (preq == NULL)
return NULL;
sock = preq->rq_conn;
if (!preq->prot) { /* Connection from TCP stream */
conn_addr = get_connectaddr(sock);
} else {
struct sockaddr_in *addr = tpp_getaddr(sock);
if (addr)
conn_addr = (pbs_net_t) ntohl(addr->sin_addr.s_addr);
}
pj = (job *) GET_NEXT(svr_newjobs);
while (pj) {
if ((pj->ji_qs.ji_un.ji_newt.ji_fromsock == -1) ||
((pj->ji_qs.ji_un.ji_newt.ji_fromsock == sock) &&
(pj->ji_qs.ji_un.ji_newt.ji_fromaddr == conn_addr))) {
if (jobid != NULL) {
if (!strncmp(pj->ji_qs.ji_jobid, jobid, PBS_MAXSVRJOBID))
break;
} else
break;
}
pj = (job *) GET_NEXT(pj->ji_alljobs);
}
return (pj);
}
#ifndef PBS_MOM /* SERVER only */
/**
* @brief Function to notify relevant scheduler of the command passed to this function
*
* @param[in] cmd - The command that is to be notified to the scheduler
* @param[in] resv - The reservation related to the command
*
* @return Number of schedulers notified.
*
*/
int
notify_scheds_about_resv(int cmd, resc_resv *resv)
{
pbs_sched *psched;
char *partition_name = NULL;
int num_scheds = 0;
if (resv != NULL) {
if (is_rattr_set(resv, RESV_ATR_partition))
partition_name = get_rattr_str(resv, RESV_ATR_partition);
else
/* for reservations without partitions, set request/reply count to 0
* because this is the only case when notification will be sent to multiple
* schedulers and server expects multiple replies. Once partition name is set,
* only relevant scheduler is notified and server will receive only one reply.
*/
resv->req_sched_count = resv->rep_sched_count = 0;
}
for (psched = (pbs_sched *) GET_NEXT(svr_allscheds); psched; psched = (pbs_sched *) GET_NEXT(psched->sc_link)) {
if (partition_name != NULL) {
if (strcmp(partition_name, DEFAULT_PARTITION) == 0) {
if (get_sched_attr_long(dflt_scheduler, SCHED_ATR_scheduling) == 1) {
set_scheduler_flag(cmd, dflt_scheduler);
num_scheds++;
}
break;
} else {
pbs_sched *tmp;
tmp = find_sched_from_partition(partition_name);
if (tmp != NULL && (get_sched_attr_long(tmp, SCHED_ATR_scheduling) == 1)) {
set_scheduler_flag(cmd, tmp);
num_scheds++;
break;
}
}
} else {
if (get_sched_attr_long(psched, SCHED_ATR_scheduling) == 1) {
set_scheduler_flag(cmd, psched);
if (resv != NULL)
resv->req_sched_count++;
num_scheds++;
}
}
}
return num_scheds;
}
/**
* @brief
* "resvSub" Batch Request processing routine
*
* @param[in] - ptr to the decoded request
*/
void
req_resvSub(struct batch_request *preq)
{
/*
* buf and buf1 are used to hold user@hostname strings together
* with a small amount (less than 64 characters) of text.
*/
char buf[PBS_MAXUSER + PBS_MAXHOSTNAME + 64] = {0};
char buf1[PBS_MAXUSER + PBS_MAXHOSTNAME + 64] = {0};
int created_here = 0;
int i = 0;
char *rid = NULL;
char ridbuf[PBS_MAXSVRRESVID + 1] = {0};
char qbuf[PBS_MAXSVRRESVID + 1] = {0};
char *pc = NULL;
attribute_def *pdef = NULL;
resc_resv *presv = NULL;
svrattrl *psatl = NULL;
int rc = 0;
int sock = preq->rq_conn;
char hook_msg[HOOK_MSG_SIZE] = {0};
int resc_access_perm_save = 0;
int qmove_requested = 0;
char *fmt = "%a %b %d %H:%M:%S %Y";
char tbuf1[256] = {0};
char tbuf2[256] = {0};
int is_maintenance = 0;
int is_resv_from_job = 0;
job *pjob;
int rc2 = 0;
char owner[PBS_MAXUSER + 1];
char *partition_name = NULL;
char *ptr = NULL;
if (preq->rq_extend && strchr(preq->rq_extend, 'm'))
is_maintenance = 1;
switch (process_hooks(preq, hook_msg, sizeof(hook_msg),
pbs_python_set_interrupt)) {
case 0: /* explicit reject */
reply_text(preq, PBSE_HOOKERROR, hook_msg);
return;
case 1: /* explicit accept */
if (recreate_request(preq) == -1) { /* error */
/* we have to reject the request, as 'preq' */
/* may have been partly modified */
strcpy(hook_msg,
"resvsub event: rejected request");
log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_HOOK,
LOG_ERR, "", hook_msg);
reply_text(preq, PBSE_HOOKERROR, hook_msg);
return;
}
break;
case 2: /* no hook script executed - go ahead and accept event*/
break;
default:
log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_HOOK,
LOG_INFO, "", "resvsub event: accept req by default");
}
/* Is the admin refusing to allow reservations on this server? */
if (is_sattr_set(SVR_ATR_ResvEnable) && !get_sattr_long(SVR_ATR_ResvEnable)) {
snprintf(buf, sizeof(buf), "reservations disallowed on %s", server_name);
if ((rc = reply_text(preq, PBSE_RESVAUTH_U, buf))) {
/* reply failed, close connection; purge resv */
close_client(sock);
}
return;
}
/* Are reservations from submitting host allowed? */
if (get_sattr_long(SVR_ATR_acl_Resvhost_enable)) {
/* acl enabled so need to check it */
if (acl_check(get_sattr(SVR_ATR_acl_Resvhosts),
preq->rq_host, ACL_Host) == 0) {
req_reject(PBSE_RESVAUTH_H, 0, preq);
return;
}
}
resc_access_perm = preq->rq_perm | ATR_DFLAG_Creat;
if (is_maintenance && !(preq->rq_perm & (ATR_DFLAG_OPWR | ATR_DFLAG_MGWR))) {
req_reject(PBSE_PERM, 0, preq);
return;
}
/* get reservation id/queue name locally */
if ((next_svr_sequence_id = get_next_svr_sequence_id()) == -1) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
/*
* if the reservation ID is supplied, the request had better be
* from another server
* Remark: would be the case if the reservation is being forwarded
* to another server - something to think about for the future
*/
if (preq->rq_fromsvr) {
/* from another server - accept the extra attributes */
resc_access_perm |= ATR_DFLAG_MGWR | ATR_DFLAG_SvWR;
rid = preq->rq_ind.rq_queuejob.rq_jid;
} else if (preq->rq_ind.rq_queuejob.rq_jid[0] != '\0') {
/* a reservation id is not allowed from a client */
req_reject(PBSE_IVALREQ, 0, preq);
return;
} else {
/* No reservation ID came with the request, create one */
/* Note: use server's job seq number generation mechanism */
created_here = RESV_SVFLG_HERE;
if (generate_objid(ridbuf, server_name, MGR_OBJ_RESV, PBS_RESV_ID_CHAR) != 0) {
req_reject(PBSE_INTERNAL, 0, preq);
return;
}
rid = ridbuf;
}
/* generate a queue name then update sv_jobidnumber and
* save serve struct
* generate a queue name then update sv_resvidnumber and
* save serve struct
* The second comment above is the one we really want,
* but the structure field would be an addition to the
* "quick save" area of the server - can't do
*/
ptr = strchr(rid, '.');
if (ptr == NULL) {
req_reject(PBSE_INTERNAL, 0, preq);
return;
}
*ptr = '\0';
pbs_strncpy(qbuf, rid, sizeof(qbuf));
*ptr = '.';
/* does reservation already exist, check both old
* and new reservations?
* This could happen if we are allowing reservations
* submitted to one server to be passed off to another
* server to fulfill or reject; we may or may not want
* this capability, but will have this code here
*/
presv = find_resv(rid);
if (presv != NULL) {
/* server rejects resvSub request if already exists */
req_reject(PBSE_RESVEXIST, 0, preq);
return;
}
/* OK, we have created a name for the local backing
* store file and a zero length file of that name
* is on the disk. Now, CREATE THE RESC_RESV STRUCTURE
* for managing the reservation and later on try and
* create a pbs_queue into which jobs submitted to the
* reservation get assigned
*/
if ((presv = resv_alloc(rid)) == NULL) {
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
/* Take a quick pass through the attribute list to see
* whether a qmove is being performed. If so, the operation
* is granted special permission to modify readonly
* resources.
*/
qmove_requested = 0;
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
if (strcmp(psatl->al_atopl.name, ATTR_convert) == 0) {
qmove_requested = 1;
break;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
/* decode attributes from resvSub request into
* the resc_resv structure's attributes
*/
resc_access_perm_save = resc_access_perm; /* save perm */
psatl = (svrattrl *) GET_NEXT(preq->rq_ind.rq_queuejob.rq_attr);
while (psatl) {
int index;
/* reservation does not support Shrink-to-fitness */
if (!(strcasecmp(psatl->al_name, ATTR_l)) &&
(!(strcasecmp(psatl->al_resc, MIN_WALLTIME)) ||
!(strcasecmp(psatl->al_resc, MAX_WALLTIME)))) {
req_reject(PBSE_NOSTF_RESV, 0, preq);
log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_REQUEST, LOG_ERR, "", msg_nostf_resv);
resv_free(presv);
return;
}
/* identify the attribute by name */
index = find_attr(resv_attr_idx, resv_attr_def, psatl->al_name);
if (index < 0) {
if (ignore_attr(psatl->al_name) >= 0) {
/*ignore some currently not handled options
*also helpful in debugging using qsub;
*remove later on
*/
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
continue;
}
/* didn`t recognize the name */
resv_free(presv);
reply_badattr(PBSE_NOATTR, 1, psatl, preq);
return;
}
pdef = &resv_attr_def[index];
/* Does attribute's definition flags indicate that
* we have sufficient permission to write the attribute?
*/
resc_access_perm = resc_access_perm_save; /* reset */
if ((psatl->al_flags & ATR_VFLAG_HOOK) || qmove_requested) {
resc_access_perm = ATR_DFLAG_USWR |
ATR_DFLAG_OPWR |
ATR_DFLAG_MGWR |
ATR_DFLAG_SvWR |
ATR_DFLAG_Creat;
}
if ((pdef->at_flags & resc_access_perm) == 0) {
resv_free(presv);
reply_badattr(PBSE_ATTRRO, 1, psatl, preq);
return;
}
/* decode attribute */
rc = set_rattr_generic(presv, index, psatl->al_value, psatl->al_resc, INTERNAL);
if (rc != 0) {
resv_free(presv);
reply_badattr(rc, 1, psatl, preq);
return;
}
if (!(strcasecmp(psatl->al_name, ATTR_resv_job))) {
if ((pjob = find_job(psatl->al_value)) == NULL) {
req_reject(PBSE_UNKJOBID, 0, preq);
resv_free(presv);
return;
}
if (pjob->ji_myResv) {
req_reject(PBSE_RESV_FROM_RESVJOB, 0, preq);
resv_free(presv);
return;
}
if (is_job_array(psatl->al_value) != IS_ARRAY_NO) {
req_reject(PBSE_RESV_FROM_ARRJOB, 0, preq);
resv_free(presv);
return;
}
get_jobowner(get_jattr_str(pjob, JOB_ATR_job_owner), owner);
if (strcmp(preq->rq_user, owner)) {
req_reject(PBSE_PERM, 0, preq);
resv_free(presv);
return;
}
rc2 = copy_params_from_job(psatl->al_value, presv);
if (rc2) {
req_reject(rc2, 0, preq);
resv_free(presv);
return;
}
if (is_qattr_set(pjob->ji_qhdr, QA_ATR_partition))
partition_name = get_qattr_str(pjob->ji_qhdr, QA_ATR_partition);
else
partition_name = DEFAULT_PARTITION;
is_resv_from_job = 1;
}
psatl = (svrattrl *) GET_NEXT(psatl->al_link);
}
resc_access_perm = resc_access_perm_save; /* restore perm */
/* invoke any defined attribute at_action routines */
for (i = 0; i < RESV_ATR_LAST; ++i) {
pdef = &resv_attr_def[i];
if (is_rattr_set(presv, i) && (pdef->at_action)) {
rc = pdef->at_action(get_rattr(presv, i), presv, ATR_ACTION_NEW);
if (rc) {
resv_free(presv);
req_reject(rc, i, preq);
return;
}
}
}
/*"start", "end","duration", and "wall"; derive and check*/
if (start_end_dur_wall(presv)) {
resv_free(presv);
req_reject(PBSE_BADTSPEC, 0, preq);
return;
}
/* If standing reservation check the recurrence rule
* and possibly change the queue and reservation id to start with
* 'S' instead of 'R'
*/
if (get_rattr_long(presv, RESV_ATR_resv_standing)) {
int resv_count;
/* Check the recurrence rule. If this fails, an error message
* is sent back to the requestor. Otherwise, check the number
* of occurrences requested by the recurrence rule. If 1 then
* it is treated as an advance reservation.
*/
resv_count = check_rrule(
get_rattr_str(presv, RESV_ATR_resv_rrule),
get_rattr_long(presv, RESV_ATR_start),
get_rattr_long(presv, RESV_ATR_end),
get_rattr_str(presv, RESV_ATR_resv_timezone),
&rc);
/* rc is set by check_rrule to report any possible icalendar
* syntax or time problem
*/
if (rc != 0) {
resv_free(presv);
req_reject(rc, 0, preq);
return;
}
set_rattr_l_slim(presv, RESV_ATR_resv_count, resv_count, SET);
/* If more than 1 occurrence are requested then alter the
* reservation and queue first character
*/
if (resv_count > 1) {
rid[0] = PBS_STDNG_RESV_ID_CHAR;
qbuf[0] = PBS_STDNG_RESV_ID_CHAR;
} else /* If only 1 occurrence, treat it as an advance reservation */
set_rattr_l_slim(presv, RESV_ATR_resv_standing, 0, SET);
} else
set_rattr_l_slim(presv, RESV_ATR_resv_count, 1, SET);
if (is_maintenance)
rid[0] = qbuf[0] = PBS_MNTNC_RESV_ID_CHAR;
(void) strcpy(presv->ri_qs.ri_resvID, rid);
if (created_here) {
presv->ri_qs.ri_svrflags = created_here;
}
/*
* for resources that are not specified in the request and
* for which default values can be determined, set these values
* as the values for those resources
*/
if (!is_resv_from_job && ((rc = set_resc_deflt((void *) presv, RESC_RESV_OBJECT, NULL)) != 0)) {
resv_free(presv);
req_reject(rc, 0, preq);
return;
}
/*
* Now that the attributes have been decoded, setup some
* additional parameters and perform a few more checks.
*/
/* set some items based on who created the reservation... */
if (created_here) {
/* reservation got created by this server */
/* ck priority value - in future, reservations
* may support the notion of priority
*/
if (is_rattr_set(presv, RESV_ATR_priority)) {
if (get_rattr_long(presv, RESV_ATR_priority) < -1024 || get_rattr_long(presv, RESV_ATR_priority) > 1024) {
resv_free(presv);
req_reject(PBSE_BADATVAL, 0, preq);
return;
}
}
/* set reservation name to "NULL" if not specified by user */
if (!is_rattr_set(presv, RESV_ATR_resv_name))
set_rattr_str_slim(presv, RESV_ATR_resv_name, "NULL", NULL);
if (!is_resv_from_job) {
/* set reservation owner attribute to user@host */
(void) strcpy(buf, preq->rq_user);
(void) strcat(buf, "@");
(void) strcat(buf, preq->rq_host);
set_rattr_str_slim(presv, RESV_ATR_resv_owner, buf, NULL);
}
/* make sure owner is in reservation's Authorized_Users */
if (act_resv_add_owner(get_rattr(presv, RESV_ATR_auth_u), presv, ATR_ACTION_NEW)) {
resv_free(presv);
req_reject(PBSE_BADATVAL, 0, preq);
return;
}
/* set create time */
set_rattr_l_slim(presv, RESV_ATR_ctime, (long) time_now, SET);
/* set hop count = 1 */
set_rattr_l_slim(presv, RESV_ATR_hopcount, 1, SET);
} else {
/* reservation created elsewhere and being moved here */
long hop;
/* make sure resv_owner is set, ERROR IF NOT */
if (!is_rattr_set(presv, RESV_ATR_resv_owner)) {
resv_purge(presv);
req_reject(PBSE_IVALREQ, 0, preq);
return;
}
/* increment hop count */
hop = get_rattr_long(presv, RESV_ATR_hopcount);
if (++hop > PBS_MAX_HOPCOUNT) {
resv_purge(presv);
req_reject(PBSE_HOPCOUNT, 0, preq);
return;
} else
set_rattr_l_slim(presv, RESV_ATR_hopcount, hop, SET);
}
/* determine values for the "euser" and "egroup" attributes */
if ((rc = set_objexid((void *) presv, RESC_RESV_OBJECT, presv->ri_wattr))) {
resv_free(presv);
req_reject(rc, 0, preq);
return;
}
/*
* Are reservation submissions being controlled by a group ACL?
* If yes, check if this one is allowed or denied
*/
if (is_sattr_set(SVR_ATR_acl_ResvGroup_enable) && get_sattr_long(SVR_ATR_acl_ResvGroup_enable)) {
if (acl_check(get_sattr(SVR_ATR_acl_ResvGroups), get_rattr_str(presv, RESV_ATR_euser), ACL_Group) == 0) {
resv_free(presv);
req_reject(PBSE_RESVAUTH_G, 0, preq);
return;
}
}
/* Is this user allowed to submit a reservation? */
if (is_sattr_set(SVR_ATR_AclResvUserEnabled) && get_sattr_long(SVR_ATR_AclResvUserEnabled)) {
if (NULL != preq->rq_host) {
snprintf(buf1, sizeof(buf1), "%s@%s", get_rattr_str(presv, RESV_ATR_euser), preq->rq_host);
}
if (acl_check(get_sattr(SVR_ATR_AclResvUsers), buf1, ACL_User) == 0) {
resv_free(presv);
req_reject(PBSE_RESVAUTH_U, 0, preq);
return;
}
}
/* set up at_server attribute for status */
set_rattr_str_slim(presv, RESV_ATR_at_server, server_name, NULL);
/* set what will be the name of the reservation's associated queue */
set_rattr_str_slim(presv, RESV_ATR_queue, qbuf, NULL);
/*
* Now that the resc_resv structure exists and and has been setup,
* try to acquire and setup a pbs_queue into which jobs submitted
* to the reservation get placed - actually, right now, the user
* directly does a "qsub" to this created queue but, at some point
* it's conceivable that the interface to user might change for
* submitting jobs to reservations and the user would specify the
* reservation ID string instead of the queue
*/
if ((rc = get_queue_for_reservation(presv)) != 0) {
/* couldn't acquire the queue */
if ((pc = pbse_to_txt(rc)) != 0)
log_event(PBSEVENT_RESV, PBS_EVENTCLASS_RESV, LOG_INFO,
presv->ri_qs.ri_resvID, pc);
log_event(PBSEVENT_RESV, PBS_EVENTCLASS_RESV, LOG_INFO,
presv->ri_qs.ri_resvID,
"error - reservation being deleted");
resv_free(presv);
/* Single out duplicate list entries to inform end-user about
* erroneous input. Other errors are internal and will fall under
* a generic "reservation failure" message.
*/
if (rc != PBSE_DUPLIST)
rc = PBSE_resvFail;
req_reject(rc, 0, preq);
return;
}
/* set remaining resc_resv structure elements */
presv->ri_qs.ri_state = RESV_UNCONFIRMED;
presv->ri_qs.ri_substate = RESV_UNCONFIRMED;
set_rattr_l_slim(presv, RESV_ATR_state, RESV_UNCONFIRMED, SET);
set_rattr_l_slim(presv, RESV_ATR_substate, RESV_UNCONFIRMED, SET);
set_rattr_l_slim(presv, RESV_ATR_mtime, (long) time_now, SET);
if (is_rattr_set(presv, RESV_ATR_convert) && !is_rattr_set(presv, RESV_ATR_del_idle_time))
set_rattr_l_slim(presv, RESV_ATR_del_idle_time, RESV_ASAP_IDLE_TIME, SET);
/* save resv and server structure */
if (resv_save_db(presv)) {
(void) resv_purge(presv);
req_reject(PBSE_SAVE_ERR, 0, preq);
return;
}
/* If not a standing reservation, put onto the "timed task" list a task
* that causes deletion of the reservation if the window passes
*/
if (!get_rattr_long(presv, RESV_ATR_resv_standing)) {
if (get_rattr_long(presv, RESV_ATR_start) != PBS_RESV_FUTURE_SCH) {
if (gen_task_EndResvWindow(presv)) {
resv_purge(presv);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
}
}
/* link reservation into server's reservation list */
append_link(&svr_allresvs, &presv->ri_allresvs, presv);
if ((is_resv_from_job) && (confirm_resv_locally(presv, preq, partition_name))) {
resv_purge(presv);
req_reject(PBSE_resvFail, 0, preq);
return;
}
/* acknowledge the request with the reservation id
* Remark: for reply we can use the function used for jobs
*/
if (is_rattr_set(presv, RESV_ATR_interactive) == 0) {
/*Not "interactive" so don't wait on scheduler, reply now*/
if (is_resv_from_job)
snprintf(buf, sizeof(buf), "%s CONFIRMED", presv->ri_qs.ri_resvID);
else
snprintf(buf, sizeof(buf), "%s UNCONFIRMED", presv->ri_qs.ri_resvID);
if (get_rattr_long(presv, RESV_ATR_resv_standing))
snprintf(buf1, sizeof(buf1), "requestor=%s@%s recurrence_rrule=%s timezone=%s",
preq->rq_user, preq->rq_host,
get_rattr_str(presv, RESV_ATR_resv_rrule),
get_rattr_str(presv, RESV_ATR_resv_timezone));
else
snprintf(buf1, sizeof(buf1), "requestor=%s@%s",
preq->rq_user, preq->rq_host);
if ((rc = reply_text(preq, PBSE_NONE, buf))) {
/* reply failed, close connection; purge resv */
close_client(sock);
resv_purge(presv);
return;
}
if (!is_resv_from_job)
account_recordResv(PBS_ACCT_UR, presv, buf1);
} else {
/*Don't reply back until scheduler decides*/
long dt;
presv->ri_brp = preq;
dt = get_rattr_long(presv, RESV_ATR_interactive);
if (dt >= 0) {
/*reply with id and state no decision in +dt secs*/
(void) gen_future_reply(presv, dt);
} else {
/*no decision in -dt seconds, delete with msg*/
(void) gen_negI_deleteResv(presv, -dt);
}
if (get_rattr_long(presv, RESV_ATR_resv_standing))
snprintf(buf, sizeof(buf), "requestor=%s@%s Interactive=%ld recurrence_rrule=%s timezone=%s",
preq->rq_user, preq->rq_host, dt,
get_rattr_str(presv, RESV_ATR_resv_rrule),
get_rattr_str(presv, RESV_ATR_resv_timezone));
else
snprintf(buf, sizeof(buf), "requestor=%s@%s Interactive=%ld",
preq->rq_user, preq->rq_host, dt);
account_recordResv(PBS_ACCT_UR, presv, buf);
}
{
long dt = get_rattr_long(presv, RESV_ATR_start);
strftime(tbuf1, sizeof(tbuf1), fmt, localtime((time_t *) &dt));
dt = get_rattr_long(presv, RESV_ATR_end);
strftime(tbuf2, sizeof(tbuf2), fmt, localtime((time_t *) &dt));
}
if (!get_rattr_long(presv, RESV_ATR_resv_standing)) {
snprintf(log_buffer, sizeof(log_buffer), "New reservation submitted start=%s end=%s", tbuf1, tbuf2);
} else {
snprintf(log_buffer, sizeof(log_buffer), "New reservation submitted start=%s end=%s "
"recurrence_rrule=%s timezone=%s",
tbuf1, tbuf2,
get_rattr_str(presv, RESV_ATR_resv_rrule),
get_rattr_str(presv, RESV_ATR_resv_timezone));
}
log_event(PBSEVENT_RESV, PBS_EVENTCLASS_RESV, LOG_INFO,
presv->ri_qs.ri_resvID, log_buffer);
/* let the scheduler know that something new
* is available for consideration
*/
if (!is_maintenance && !is_resv_from_job)
notify_scheds_about_resv(SCH_SCHEDULE_NEW, presv);
}
static struct dont_set_in_max {
char *ds_name; /* resource name */
resource *ds_rescp; /* ptr to resource entry */
} dont_set_in_max[] = {
{"nodes", NULL},
{"nodect", NULL},
{"select", NULL},
{"place", NULL},
{"walltime", NULL}};
/**
* @brief
* create a queue to bind to the reservation
*
* @par Functionality:
* get_queue_for_reservation - call this function to create and setup
* a queue that's associated with a general resources reservation.
*
* An internally generated request is built up and issued to the
* "batch manager" subsystem to create a queue having the desired
* queue attributes
*
* @param[in] presv - The reservation to which a queue is to be be associated
*
* @par
* Note that the queue is created by issuing a request to "ourselves"
* (the server) and that this request is fulfilled asynchronously. The queue
* may fail to be created and cause the reservation to be queue-less.
*
* @return error code
* @retval 0 - success
* @retval PBS error code - error
*
* @par MT-safe: No
*/
static int
get_queue_for_reservation(resc_resv *presv)
{
int i;
int j;
int rc = 0;
svrattrl *psatl;
attribute *pattr;
static const int lenF = 6; /*strlen("False") + 1*/
static const int lenT = 5; /*strlen("True") + 1*/
static const int lenE = 10; /*strlen("Execution") + 1*/
pbs_list_head *plhed;
struct work_task *pwt;
struct batch_request *newreq;
newreq = alloc_br(PBS_BATCH_Manager);
if (newreq == NULL) {
(void) sprintf(log_buffer, "batch request allocation failed");
log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_RESV, LOG_ERR,
presv->ri_qs.ri_resvID, log_buffer);
return (PBSE_SYSTEM);
}
newreq->rq_ind.rq_manager.rq_cmd = MGR_CMD_CREATE;
newreq->rq_ind.rq_manager.rq_objtype = MGR_OBJ_QUEUE;
newreq->rq_perm = ATR_DFLAG_MGWR | ATR_DFLAG_OPWR;
(void) strcpy(newreq->rq_user, "pbs_server");
(void) strcpy(newreq->rq_host, server_name);
strcpy(newreq->rq_ind.rq_manager.rq_objname, get_rattr_str(presv, RESV_ATR_queue));
pattr = get_rattr(presv, RESV_ATR_resource);
CLEAR_HEAD(newreq->rq_ind.rq_manager.rq_attr);
plhed = &newreq->rq_ind.rq_manager.rq_attr;
/* resources specified by the reservation become "resc_avail" for queue
* Have already (at_action processing for RESV_ATR_resource) checked
* that server has control of resources needed, in sufficient quantity
*
* Note: "resc_avail" and "resc_max" attributes on the queue should
* not include certain (string) resources from the reservation's
* "resource_list" attribute as specified in the array dont_set_in_max;
* so, that is why those links are deleted from the attribute
* and then appended back a few lines later
*/
j = sizeof(dont_set_in_max) / sizeof(struct dont_set_in_max);
for (i = 0; i < j; ++i) {
resource_def *prdef;
prdef = find_resc_def(svr_resc_def, dont_set_in_max[i].ds_name);
dont_set_in_max[i].ds_rescp = find_resc_entry(pattr, prdef);
if (dont_set_in_max[i].ds_rescp)
delete_link(&dont_set_in_max[i].ds_rescp->rs_link);
}
rc = resv_attr_def[RESV_ATR_resource].at_encode(pattr, plhed,
ATTR_rescavail, NULL,
ATR_ENCODE_CLIENT, NULL);
rc += resv_attr_def[RESV_ATR_resource].at_encode(pattr, plhed,
ATTR_rescmax, NULL,
ATR_ENCODE_CLIENT, NULL);
for (i = 0; i < j; ++i) {
if (dont_set_in_max[i].ds_rescp)
append_link(
&pattr->at_val.at_list,
&dont_set_in_max[i].ds_rescp->rs_link,
dont_set_in_max[i].ds_rescp);
}
if (rc < 0) {
free_br(newreq);
return (PBSE_genBatchReq);
}
if ((psatl = attrlist_create(ATTR_qtype, NULL, lenE)) != NULL) {
static char Execution[] = "Execution";
psatl->al_flags = que_attr_def[QA_ATR_QType].at_flags;
strcpy(psatl->al_value, Execution);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
/* Don't enable queue until reservation is RESV_CONFIRMED */
if ((psatl = attrlist_create(ATTR_enable, NULL, lenF)) !=
NULL) {
psatl->al_flags = que_attr_def[QA_ATR_Enabled].at_flags;
strcpy(psatl->al_value, ATR_FALSE);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
/* Don't start queue until reservation window here, RESV_TIME_TO_RUN */
if ((psatl = attrlist_create(ATTR_start, NULL, lenF)) != NULL) {
psatl->al_flags = que_attr_def[QA_ATR_Started].at_flags;
strcpy(psatl->al_value, ATR_FALSE);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
/* Generate a "user_acl" for PBS_BATCH_manager req, use
* reservation's "Authorized_Users" attribute
* Remark: "Authorized_Users" has, atleast, the reservation's owner
*/
if (is_rattr_set(presv, RESV_ATR_auth_u)) {
pattr = get_rattr(presv, RESV_ATR_auth_u);
plhed = &newreq->rq_ind.rq_manager.rq_attr;
rc = check_duplicates(pattr->at_val.at_arst);
if (rc == 1) {
free_br(newreq);
return (PBSE_DUPLIST);
}
rc = resv_attr_def[RESV_ATR_auth_u].at_encode(pattr, plhed,
ATTR_acluser, NULL,
ATR_ENCODE_SVR, NULL);
if (rc < 0) {
free_br(newreq);
return (PBSE_genBatchReq);
}
/*let the que know user acl is to be enforced*/
if ((psatl = attrlist_create(ATTR_acluren,
NULL, lenT)) != NULL) {
psatl->al_flags = que_attr_def[QA_ATR_AclUserEnabled].at_flags;
strcpy(psatl->al_value, ATR_TRUE);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
}
/* Generate a "group_acl" for PBS_BATCH_manager req, use
* reservation's "Authorized_Groups" attribute
*/
if (is_rattr_set(presv, RESV_ATR_auth_g)) {
pattr = get_rattr(presv, RESV_ATR_auth_g);
plhed = &newreq->rq_ind.rq_manager.rq_attr;
rc = check_duplicates(pattr->at_val.at_arst);
if (rc == 1) {
free_br(newreq);
return (PBSE_DUPLIST);
}
rc = resv_attr_def[RESV_ATR_auth_g].at_encode(pattr, plhed,
ATTR_aclgroup, NULL,
ATR_ENCODE_SVR, NULL);
if (rc < 0) {
free_br(newreq);
return (PBSE_genBatchReq);
}
/*let the que know user acl is to be enforced*/
if ((psatl = attrlist_create(ATTR_aclgren,
NULL, lenT)) != NULL) {
psatl->al_flags = que_attr_def[QE_ATR_AclGroupEnabled].at_flags;
strcpy(psatl->al_value, ATR_TRUE);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
}
/* Generate a "host_acl" for PBS_BATCH_manager req, use
* reservation's "Authorized_Hosts" attribute
*/
if (is_rattr_set(presv, RESV_ATR_auth_h)) {
pattr = get_rattr(presv, RESV_ATR_auth_h);
plhed = &newreq->rq_ind.rq_manager.rq_attr;
rc = check_duplicates(pattr->at_val.at_arst);
if (rc == 1) {
free_br(newreq);
return (PBSE_DUPLIST);
}
rc = resv_attr_def[RESV_ATR_auth_h].at_encode(pattr, plhed,
ATTR_aclhost, NULL,
ATR_ENCODE_SVR, NULL);
if (rc < 0) {
free_br(newreq);
return (PBSE_genBatchReq);
}
/*let the que know user acl is to be enforced*/
if ((psatl = attrlist_create(ATTR_aclhten,
NULL, lenT)) != NULL) {
psatl->al_flags = que_attr_def[QA_ATR_AclHostEnabled].at_flags;
strcpy(psatl->al_value, ATR_TRUE);
append_link(plhed, &psatl->al_link, psatl);
} else {
free_br(newreq);
return (PBSE_genBatchReq);
}
}
/* Ok, everything is successfully built up, issue the Batch_Request */
if (issue_Drequest(PBS_LOCAL_CONNECTION, newreq,
handle_qmgr_reply_to_resvQcreate, &pwt, 0) == -1) {
free_br(newreq);
(void) sprintf(log_buffer, "%s", msg_resvQcreateFail);
log_event(PBSEVENT_RESV, PBS_EVENTCLASS_RESV, LOG_ERR,
presv->ri_qs.ri_resvID, log_buffer);
return (PBSE_mgrBatchReq);
}
if (pwt)
pwt->wt_parm2 = presv; /*needed to handle qmgr's response*/
return (0);
}
/**
* @brief
* ignore_attr - wrapper function for find_attr.
*
* @param[in] name - attribute name to find
*
* @return return vlaue from find_attr()
*/
static int
ignore_attr(char *name)
{
return (find_attr(job_attr_idx, job_attr_def, name));
}
/**
* @brief
* act_resv_add_owner - This is a special action function
* for a reservation's "Authorized_Users" attribute - i.e. who is
* allowed to submit jobs to the reservation.
*
* @param[in] pattr - not used here
* @param[in] pobj - reservation structure
* @param[in] amode - "actmode" stands for the type of action,
* if ATR_ACTION_NEW, just returns from the function.
*
* @return error code
* @retval 0 : Success
* @retval !=0 : fails
*/
int
act_resv_add_owner(attribute *pattr, void *pobj, int amode)
{
attribute dummy, *ap;
struct array_strings dumarst;
enum batch_op op;
resc_resv *presv;
char *ps;
int len;
if (amode != ATR_ACTION_NEW)
return (0); /*success - nothing to do*/
presv = (resc_resv *) pobj;
if (is_rattr_set(presv, RESV_ATR_resv_owner) == 0)
return (0); /*success - nothing to do*/
ps = get_rattr_str(presv, RESV_ATR_resv_owner);
len = strlen(ps);
ap = get_rattr(presv, RESV_ATR_auth_u);
if (is_attr_set(ap)) {
int i;
for (i = 0; i < ap->at_val.at_arst->as_usedptr; ++i)
if (!strcmp(ps, ap->at_val.at_arst->as_string[i]))
return (0); /*resv owner in Authorized_Users*/
op = INCR;
} else
op = SET; /*Authorized_Users is NULL, must set*/
(void) memset(&dummy, 0, sizeof(dummy));
dummy.at_flags = NO_USER_SET | ATR_VFLAG_SET;
dummy.at_type = ATR_TYPE_ACL;
dummy.at_val.at_arst = &dumarst;
dumarst.as_npointers = 1;
dumarst.as_usedptr = 1;
dumarst.as_bufsize = strlen(ps) + len;
dumarst.as_buf = ps;
dumarst.as_next = ps + len;
dumarst.as_string[0] = ps;
/*"at_set" function returns 0 on success and NZ on failure*/
/*Remark: nice thing would be to have owner appear first */
return (resv_attr_def[RESV_ATR_auth_u].at_set(ap, &dummy, op));
}
/**
* @brief
* handle_qmgr_reply_to_resvQcreate - this is the function that's to be
* called to handle the qmgr's response to the earlier issued request
* for queue creation for a reservation. If the response from the
* qmgr is successful, copy the queue's name into the ri_queue field
* of the reservation and set the reservation's ri_qp field pointing
* to this newly established queue. If not successful log a message.
* @par Functionality:
* This function should only be called through an INTERNALLY GENERATED
* request to another server (including ourself).
* It frees the request structure and closes the connection (handle).
* @par
* In the work task entry, wt_event is the connection handle and
* wt_parm1 is a pointer to the request structure (containing the reply).
* wt_parm2 should have the address of the reservation structure
*
* @note
* THIS SHOULD NOT BE USED IF AN EXTERNAL (CLIENT) REQUEST IS "relayed",
* because the request/reply structure is still needed to reply back
* to the client.
*
* @param[in,out] pwt - earlier issued request for queue creation for a reservation.
*/
static void
handle_qmgr_reply_to_resvQcreate(struct work_task *pwt)
{
job *pjob;
pbs_queue *pque;
resc_resv *presv = pwt->wt_parm2;
struct batch_request *preq = pwt->wt_parm1;
if (preq->rq_reply.brp_code) {
(void) sprintf(log_buffer, msg_resvQcreateFail,
presv->ri_jbp->ji_qs.ji_jobid,
presv->ri_qs.ri_resvID);
log_event(PBSEVENT_RESV, PBS_EVENTCLASS_RESV, LOG_INFO,
presv->ri_qs.ri_resvID, log_buffer);
} else {
pque = find_queuebyname(preq->rq_ind.rq_manager.rq_objname);
if ((presv->ri_qp = pque) != 0)
pque->qu_resvp = presv;
if ((pjob = find_job(get_rattr_str(presv, RESV_ATR_job))))
pjob->ji_myResv = presv;
(void) strcpy(presv->ri_qs.ri_queue, get_rattr_str(presv, RESV_ATR_queue));
if (resv_save_db(presv)) {
(void) resv_purge(presv);
req_reject(PBSE_SYSTEM, 0, preq);
return;
}
}
free_br((struct batch_request *) pwt->wt_parm1);
if (pwt->wt_event != -1)
svr_disconnect(pwt->wt_event);
}
/**
* @brief
* Validate job and reservation place directives.
*
* @param[in] pj - The job to validate.
*
* @note
* The reservation associated to the job is obtained
* from the job structure.
*
* @return Whether the place directive between the job
* and the reservation are in conflict or not.
*
* @retval 1 : job and reservation place directives do not conflict
* @retval 0 : job and reservation place directives conflict
*
* @par MT-safe: No
*/
static int
validate_place_req_of_job_in_reservation(job *pj)
{
resource_def *prsdef;
resource *job_place;
resource *resv_place;
attribute *jattr;
attribute *rattr;
enum vnode_sharing job_sharetype;
enum vnode_sharing resv_sharetype;
/* A job not in reservation is implicitly considered valid */
if (pj->ji_myResv == NULL)
return 1;
prsdef = &svr_resc_def[RESC_PLACE];
jattr = get_jattr(pj, JOB_ATR_resource);
rattr = get_rattr(pj->ji_myResv, RESV_ATR_resource);
job_place = find_resc_entry(jattr, prsdef);
if (!job_place || !job_place->rs_value.at_val.at_str)
return 1;
resv_place = find_resc_entry(rattr, prsdef);
if (!resv_place || !resv_place->rs_value.at_val.at_str)
return 1;
/* Cehck for exclhost should come before excl because exclhost contains
* the excl prefix
*/
job_sharetype = place_sharing_type(
job_place->rs_value.at_val.at_str,
VNS_FORCE_EXCLHOST);
if (job_sharetype == VNS_UNSET) {
job_sharetype = place_sharing_type(
job_place->rs_value.at_val.at_str,
VNS_FORCE_EXCL);
}
resv_sharetype = place_sharing_type(
resv_place->rs_value.at_val.at_str,
VNS_FORCE_EXCLHOST);
if (resv_sharetype == VNS_UNSET) {
resv_sharetype = place_sharing_type(
resv_place->rs_value.at_val.at_str,
VNS_FORCE_EXCL);
}
/* Reject the request if job requests exclusive and reservation not */
if ((resv_sharetype == VNS_UNSET) &&
(job_sharetype != VNS_UNSET))
return 0;
/* Reject request if job requests exclhost and reservation doesn't */
if ((resv_sharetype != VNS_FORCE_EXCLHOST) &&
(job_sharetype == VNS_FORCE_EXCLHOST))
return 0;
return 1;
}
/**
* @brief
* Provides the next job id
*
* @param[in] void
*
* @return long long
* @retval (>=0 to <= max_job_sequence_id): Success
* @retval -1 : database error
*
*/
long long
get_next_svr_sequence_id(void)
{
static long long lastid = -1;
long long seq = server.sv_qs.sv_jobidnumber;
/* If server job limit is over, reset back to zero */
if (++server.sv_qs.sv_jobidnumber > svr_max_job_sequence_id) {
server.sv_qs.sv_jobidnumber = 0;
lastid = -1;
}
/* check if we should save jobid */
if (lastid == -1 || server.sv_qs.sv_jobidnumber == lastid) {
lastid = ((server.sv_qs.sv_jobidnumber / SEQ_WIN_INCR) + 1) * SEQ_WIN_INCR;
server.sv_qs.sv_lastid = lastid;
svr_save_db(&server);
}
return seq;
}
/**
* @brief
* Resets server sequence window count and server sv_jobidnumber
*
* @return None
*
*/
void
reset_svr_sequence_window(void)
{
server.sv_qs.sv_jobidnumber = 0;
}
/**
* @brief - Copy parameters from job to reservation
*
* @param[in] - jobid - id of the job from which the parameters will be copied.
* @param[in] - presv - reservation to which the parameters will be copied to.
*
* @return int
* @retval 0: Success
* @retval < 0: error
*/
int
copy_params_from_job(char *jobid, resc_resv *presv)
{
job *pjob;
int bufsize;
attribute temp;
resource *presc;
resource_def *prdefsl;
resource_def *resc_def;
resource *job_resc_entry;
resource *resv_resc_entry;
char buf[PBS_MAXUSER + PBS_MAXHOSTNAME + 64] = {0};
int walltime_copied = 0;
pjob = find_job(jobid);
if (pjob == NULL)
return PBSE_UNKJOBID;
if ((!check_job_substate(pjob, JOB_SUBSTATE_RUNNING)) &&
(get_jattr_str(pjob, JOB_ATR_exec_vnode) == NULL))
return PBSE_BADSTATE;
bufsize = PBS_MAXUSER + PBS_MAXHOSTNAME + 64 - 1;
if (strchr(get_jattr_str(pjob, JOB_ATR_job_owner), '@')) {
strncpy(buf, get_jattr_str(pjob, JOB_ATR_job_owner), bufsize);
} else
snprintf(buf, bufsize, "%s@%s", get_jattr_str(pjob, JOB_ATR_job_owner),
get_jattr_str(pjob, JOB_ATR_submit_host));
set_rattr_str_slim(presv, RESV_ATR_resv_owner, buf, NULL);
set_rattr_str_slim(presv, RESV_ATR_resv_nodes, get_jattr_str(pjob, JOB_ATR_exec_vnode), NULL);
if (is_jattr_set(pjob, JOB_ATR_stime))
set_rattr_l_slim(presv, RESV_ATR_start, get_jattr_long(pjob, JOB_ATR_stime), SET);
else
set_rattr_l_slim(presv, RESV_ATR_start, time_now, SET);
post_attr_set(get_rattr(presv, RESV_ATR_SchedSelect));
job_resc_entry = (resource *) GET_NEXT(get_jattr_list(pjob, JOB_ATR_resource));
for (; job_resc_entry; job_resc_entry = (resource *) GET_NEXT(job_resc_entry->rs_link)) {
resc_def = job_resc_entry->rs_defin;
resv_resc_entry = find_resc_entry(get_rattr(presv, RESV_ATR_resource), resc_def);
if (resv_resc_entry == NULL) {
if (!(resv_resc_entry = add_resource_entry(get_rattr(presv, RESV_ATR_resource), resc_def)))
return PBSE_SYSTEM;
}
if (is_attr_set(&job_resc_entry->rs_value)) {
(void) resc_def->rs_set(&resv_resc_entry->rs_value, &job_resc_entry->rs_value, SET);
}
if (!strcmp(resc_def->rs_name, WALLTIME))
walltime_copied = 1;
}
if (!walltime_copied) {
resc_def = &svr_resc_def[RESC_WALLTIME];
if (resc_def != NULL) {
resv_resc_entry = find_resc_entry(get_rattr(presv, RESV_ATR_resource), resc_def);
if (resv_resc_entry == NULL) {
if (!(resv_resc_entry = add_resource_entry(get_rattr(presv, RESV_ATR_resource), resc_def)))
return PBSE_SYSTEM;
}
temp.at_flags = ATR_VFLAG_SET;
temp.at_type = ATR_TYPE_LONG;
temp.at_val.at_long = RESV_INFINITY;
(void) resc_def->rs_set(&resv_resc_entry->rs_value, &temp, SET);
}
}
prdefsl = &svr_resc_def[RESC_SELECT];
presc = find_resc_entry(get_jattr(pjob, JOB_ATR_resource), prdefsl);
make_schedselect(get_jattr(pjob, JOB_ATR_resource), presc, NULL, get_rattr(presv, RESV_ATR_SchedSelect));
return 0;
}
/**
* @brief - Confirm reservation that is being created out of a job.
*
* @param[in] - presv - reservation that needs to be confirmed.
* @param[in] - orig_preq - batch request.
* @param[in] - partition_name - partition in which the reservation needs to be confirmed.
*
* @return int
* @retval 0: Success
* @retval != 0: error
*/
int
confirm_resv_locally(resc_resv *presv, struct batch_request *orig_preq, char *partition_name)
{
char *at;
job *pjob;
struct work_task *pwt;
struct batch_request *preq;
int extend_size = 0;
presv->resv_from_job = 1;
preq = alloc_br(PBS_BATCH_ConfirmResv);
preq->rq_ind.rq_run.rq_destin = strdup(get_rattr_str(presv, RESV_ATR_resv_nodes));
if (preq->rq_ind.rq_run.rq_destin == NULL) {
free_br(preq);
return 1;
}
/* extend field format is "PBS_RESV_CONFIRM_SUCCESS:partition="
* allocate enough memory to be able to support the format.
*/
extend_size = strlen(PBS_RESV_CONFIRM_SUCCESS) + strlen(partition_name) + 12;
preq->rq_extend = malloc(extend_size);
if (preq->rq_extend == NULL) {
free_br(preq);
return 1;
}
snprintf(preq->rq_extend, extend_size, "%s:partition=%s", PBS_RESV_CONFIRM_SUCCESS, partition_name);
(void) strcpy(preq->rq_ind.rq_run.rq_jid, presv->ri_qs.ri_resvID);
preq->rq_perm |= ATR_DFLAG_MGWR;
if (issue_Drequest(PBS_LOCAL_CONNECTION, preq, release_req, &pwt, 0) == -1) {
free_br(preq);
return 1;
}
preq = alloc_br(PBS_BATCH_MoveJob);
preq->rq_perm |= ATR_DFLAG_MGWR;
strcpy(preq->rq_user, orig_preq->rq_user);
strcpy(preq->rq_host, orig_preq->rq_host);
pjob = find_job(get_rattr_str(presv, RESV_ATR_job));
snprintf(preq->rq_ind.rq_move.rq_jid, sizeof(preq->rq_ind.rq_move.rq_jid), "%s", get_rattr_str(presv, RESV_ATR_job));
at = strchr(presv->ri_qs.ri_resvID, (int) '.');
if (at)
*at = '\0';
snprintf(preq->rq_ind.rq_move.rq_destin, sizeof(preq->rq_ind.rq_move.rq_destin), "%s", presv->ri_qs.ri_resvID);
if (at)
*at = '.';
snprintf(pjob->ji_qs.ji_destin, PBS_MAXROUTEDEST, "%s", preq->rq_ind.rq_move.rq_destin);
return (local_move(pjob, preq));
}
#endif /*SERVER ONLY*/