/*
 * Copyright (C) 1994-2018 Altair Engineering, Inc.
 * For more information, contact Altair at www.altair.com.
 *
 * This file is part of the PBS Professional ("PBS Pro") software.
 *
 * Open Source License Information:
 *
 * PBS Pro 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.
 *
 * PBS Pro 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 <http://www.gnu.org/licenses/>.
 *
 * Commercial License Information:
 *
 * For a copy of the commercial license terms and conditions,
 * go to: (http://www.pbspro.com/UserArea/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 PBS Pro 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™",
 * "PBS Professional®", and "PBS Pro™" and Altair’s logos is subject to Altair's
 * trademark licensing policies.
 *
 */
/**
 * @file	req_runjob.c
 *
 * @brief
 * 		req_runjob.c - functions dealing with a Run Job Request
 *
 * Included functions are:
 *	check_and_provision_job()
 *	clear_from_defr()
 *	req_runjob()
 *	req_runjob2()
 *	clear_exec_on_run_fail()
 *	req_stagein()
 *	post_stagein()
 *	svr_stagein()
 *	svr_startjob()
 *	svr_strtjob2()
 *	complete_running()
 *	parse_hook_rejectmsg()
 *	post_sendmom()
 *	chk_job_torun()
 *	where_to_runjob()
 *	assign_hosts()
 *	req_defschedreply()
 *
 */

#include <pbs_config.h>   /* the master config generated by configure */

#include <ctype.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>

#ifdef WIN32
#include <windows.h>
#include "win.h"
#else
#include <strings.h>
#include <sys/wait.h>
#endif

#include <signal.h>
#include <stdlib.h>
#include "libpbs.h"
#include "server_limits.h"
#include "list_link.h"
#include "attribute.h"
#include "resource.h"
#include "server.h"
#include "credential.h"
#include "batch_request.h"
#include "job.h"
#include "reservation.h"
#include "queue.h"
#include "work_task.h"
#include "pbs_error.h"
#include "log.h"
#include "acct.h"
#include "net_connect.h"
#include "pbs_nodes.h"
#include "svrfunc.h"
#include "sched_cmds.h"
#include "pbs_license.h"
#include "hook.h"
#include "provision.h"
#include "pbs_share.h"
#include "pbs_sched.h"


/* External Functions Called: */

extern struct batch_request *cpy_stage(struct batch_request *, job *,
	enum job_atr, int);
extern struct batch_request *cpy_stage(struct batch_request *, job *, enum job_atr, int);

/* Public Functions in this file */

int  svr_startjob(job *, struct batch_request *);
extern char *msg_daemonname;
extern	char	*path_hooks_workdir;
extern char	*msg_hook_reject_deletejob;

/* Private Function local to this file */

void post_sendmom(struct work_task *);
static int  svr_stagein(job *, struct batch_request *, int, int);
static int  svr_strtjob2(job *, struct batch_request *);
static job *chk_job_torun(struct batch_request *preq, job *);
static void req_runjob2(struct batch_request *preq, job *pjob);
static job *where_to_runjob(struct batch_request *preq, job *);

/* Global Data Items: */

extern int       license_expired;
extern pbs_net_t pbs_mom_addr;
extern int	 pbs_mom_port;
extern struct server server;
extern char *msg_badexit;
extern char *msg_jobrun;
extern char *msg_job_end_sig;
extern char *msg_init_substate;
extern char *msg_manager;
extern char *msg_stageinfail;
extern char *msg_job_abort;
extern pbs_list_head svr_deferred_req;
extern time_t time_now;
extern int   svr_totnodes;	/* non-zero if using nodes */
extern job  *chk_job_request(char *, struct batch_request *, int *);


/* private data */


/**
 * @brief
 *		Take a batch_request and job pointer as arguments
 *	 	Enque provisioning by calling check_and_engue_provisioning
 *	 	if Enque is successful, sets the job substate to provisioning
 *	 	else returns an error, caller sends a req_reject to scheduler
 *
 * @see
 *		req_runjob2
 *
 * @param[in]	preq	-	batch_request
 * @param[in,out]	pjob	-	job pointer
 * @param[out]	need_prov	-	boolean value, whether job will provision
 *
 * @return	int
 * @retval	0	: no provisioning required
 * @retval	-1	: provisioning required
 * @retval	>0	: PBS error codes
 *
 * @par Side Effects:
 *	Unknown
 *
 * @par MT-safe: No
 *
 */
static int
check_and_provision_job(struct batch_request *preq, job *pjob, int *need_prov)
{
	int rc=0;

	/* prov node is part of exec_vnodes, */
	/* cut and update exec_vnode and prov_vnode */
	if (!preq || !pjob || !need_prov)
		return (PBSE_IVALREQ);

	rc=check_and_enqueue_provisioning(pjob, need_prov);
	if (rc) {
		/* log message about failure to start provisioning for a job */
		log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, LOG_INFO,
			pjob->ji_qs.ji_jobid,
			"Job failed to start provisioning");

		/* put system hold and move to held state */
		pjob->ji_wattr[(int)JOB_ATR_hold].at_val.at_long |= HOLD_s;
		pjob->ji_wattr[(int)JOB_ATR_hold].at_flags |=
			ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;
		(void)svr_setjobstate(pjob, JOB_STATE_HELD, JOB_SUBSTATE_HELD);
		job_attr_def[(int)JOB_ATR_Comment].at_decode(
			&pjob->ji_wattr[(int)JOB_ATR_Comment],
			NULL, NULL,
			"job held, provisioning failed to start");

		/* not offlining vnodes, since its not a vnode's fault. vnode */
		/* is good to run other jobs, so why waste resource. */
		return rc;
	}

	if (*need_prov == 0)
		return PBSE_NONE;

	/* provisioning was needed and enqueued */

	(void)svr_setjobstate(pjob, JOB_STATE_RUNNING, JOB_SUBSTATE_PROVISION);
	DBPRT(("%s: Sucessfully enqueued provisioning for job %s\n", __func__, pjob->ji_qs.ji_jobid))

	/* log accounting line for start of prov for a job */
	set_job_ProvAcctRcd(pjob, time_now, PROVISIONING_STARTED);

	return PBSE_NONE;
}

/**
 * @brief
 *		Search for a deferred reply element in the list whose client
 *		request came in on the socket which was just closed.  If found,
 *		clear the pointer to the original request which has been freed
 *		when the connection was closed.
 *
 * @param[in]	sd	-	socket which was just closed
 *
 * @return	none
 *
 * @par MT-safe: not really
 */
static void
clear_from_defr(int sd)
{
	struct deferred_request *pdefr;

	for (pdefr = (struct deferred_request *)GET_NEXT(svr_deferred_req);
		pdefr;
		pdefr = (struct deferred_request *)GET_NEXT(pdefr->dr_link)) {
		if (pdefr->dr_preq != NULL) {
			if (pdefr->dr_preq->rq_conn == sd) {
				/* found deferred run job request whose */
				/* connection to the client has closed  */
				if (pdefr->dr_sent != 0) {
					/* request sent to scheduler, wait   */
					/* for it to respond before removing */
					/* this request, just null the qrun  */
					/* request pointer                   */
					pdefr->dr_preq = NULL;
				} else {
					/* unlink & free the deferred request */
					delete_link(&pdefr->dr_link);
					free(pdefr);
				}
				break;
			}
		}
	}
}

/**
 * @brief	Wrapper function that calls process_hooks()
 *
 * @see		req_runjob()
 *
 * @return	int
 *
 */
int
call_to_process_hooks(struct batch_request *preq, char *hook_msg, size_t msg_len,
	void	(*pyinter_func)(void))
{
	int rc;
	rc = process_hooks(preq, hook_msg, msg_len, pyinter_func);
	if (rc == -1)
		log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_HOOK,
				LOG_INFO, "", "runjob event: accept req by default");
	return rc;
}

/**
 * @brief
 * 		req_runjob - service the Run Job and Asyc Run Job Requests
 * @par
 *		This request forces a job into execution.  Client must be privileged.
 *
 * @param[in]	preq	-	Run Job Requests
 */

void
req_runjob(struct batch_request *preq)
{
	int		  anygood;
	int		  i;
	int		  j;
	char		 *jid;
	int		  jt;		/* job type */
	int		  offset = -1;
	char		 *pc;
	job		 *pjob = NULL;
	job		 *pjobsub = NULL;
	job		 *parent  = NULL;
	char		 *range;
	int		  x, y, z;
	struct deferred_request *pdefr;
	char		  hook_msg[HOOK_MSG_SIZE];
	pbs_sched	  *psched;

	if (license_expired) {
		req_reject(PBSE_LICENSEINV, 0, preq);
		return;
	}

	if ((preq->rq_perm & (ATR_DFLAG_MGWR | ATR_DFLAG_OPWR)) == 0) {
		req_reject(PBSE_PERM, 0, preq);
		return;
	}

	jid = preq->rq_ind.rq_run.rq_jid;
	parent = chk_job_request(jid, preq, &jt);
	if (parent == NULL)
		return;		/* note, req_reject already called */

	/* the job must be in an execution queue */

	if (parent->ji_qhdr->qu_qs.qu_type != QTYPE_Execution) {
		req_reject(PBSE_IVALREQ, 0, preq);
		return;
	}

	if (!find_assoc_sched_jid(jid, &psched)) {
		sprintf(log_buffer, "Unable to reach scheduler associated with job %s", jid);
		log_err(-1, __func__, log_buffer);
		req_reject(PBSE_NOSCHEDULER, 0, preq);
		return;
	}

#ifndef NAS /* localmod 133 */
	if ((psched->scheduler_sock != -1) && was_job_alteredmoved(parent)) {
		int index = find_attr(sched_attr_def, ATTR_throughput_mode, SCHED_ATR_LAST);
		/* do not blacklist altered/moved jobs when throughput_mode is enabled */
		if ((index == -1) ||
			((psched->sch_attr[index].at_flags & ATR_VFLAG_SET) &&
			 (psched->sch_attr[index].at_val.at_long == 0))) {
			req_reject(PBSE_NORUNALTEREDJOB, 0, preq);
			set_scheduler_flag(SCH_SCHEDULE_NEW, psched);
			return;
		}
	}
#endif /* localmod 133 */

	if (jt == IS_ARRAY_NO) {

		/* just a regular job, pass it on down the line and be done */

		pjob = chk_job_torun(preq, parent);
		if (pjob == NULL)
			return;

	} else if (jt == IS_ARRAY_Single) {

		/* single subjob, if queued, it can be run */

		offset = subjob_index_to_offset(parent, get_index_from_jid(jid));
		if (offset == -1) {
			req_reject(PBSE_UNKJOBID, 0, preq);
			return;
		}
		i = get_subjob_state(parent, offset);
		if (i == -1) {
			req_reject(PBSE_IVALREQ, 0, preq);
			return;
		} else if (i != JOB_STATE_QUEUED) {
			/* job already running */
			req_reject(PBSE_BADSTATE, 0, preq);
			return;
		}

	} else if (jt == IS_ARRAY_ArrayJob) {

		/* invalid to run the array itself */

		req_reject(PBSE_IVALREQ, 0, preq);
		return;

	} else {
		/* what's left to handle is a range of subjobs,	*/
		/* foreach subjob, if queued, run it		*/

		anygood = 0;
		range = get_index_from_jid(jid);
		if (range == NULL) {
			req_reject(PBSE_IVALREQ, 0, preq);
			return;
		}

		while (1) {
			if ((i = parse_subjob_index(range, &pc, &x, &y, &z, &j)) == -1) {
				req_reject(PBSE_IVALREQ, 0, preq);
				return;
			} else if (i == 1)
				break;	/* no more in the range */
			for (; x <= y; x += z) {
				if ((i = numindex_to_offset(parent, x)) >= 0) {
					if (get_subjob_state(parent, i) == JOB_STATE_QUEUED)
						anygood = 1;
				}
			}
			range = pc;
		}
		if (anygood == 0) {
			req_reject(PBSE_BADSTATE, 0, preq);
			return;
		}
	}

	/* At this point, we know the basic request to run the job */
	/* or jobs is valid.  So we can proceed farther ...        */

	/* If there is a specified list of execution vnodes and    */
	/* resources, then process the run request, else it has    */
	/* to go to the scheduler				   */

	if ((preq->rq_ind.rq_run.rq_destin == NULL)  ||
		(*preq->rq_ind.rq_run.rq_destin == '\0')) {
		char fixjid[PBS_MAXSVRJOBID+1];

		/* if runjob request is from the Scheduler, */
		/* it must have a destination specified     */
		if (preq->rq_conn == psched->scheduler_sock) {
			sprintf(log_buffer,
				"runjob request from scheduler with null destination");
			log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_JOB, LOG_INFO,
				jid, log_buffer);
			req_reject(PBSE_IVALREQ, 0, preq);
			return;
		}
		pdefr = (struct deferred_request *)malloc(sizeof(struct deferred_request));
		if (pdefr == NULL) {
			req_reject(PBSE_SYSTEM, 0, preq);
			return;
		}
		CLEAR_LINK(pdefr->dr_link);

		/* fix the job id so the suffix matches the real jobid's */
		/* suffix;  in case qrun 1.short vs 1.short.domain.com   */

		snprintf(fixjid, sizeof(fixjid), "%s", jid);
		pc = strchr(fixjid, (int)'.');
		if (pc)
			*pc = '\0';
		pc = strchr(parent->ji_qs.ji_jobid, (int)'.');
		if (pc)
			(void)strcat(fixjid, pc);

		(void)strncpy(pdefr->dr_id, fixjid, PBS_MAXSVRJOBID);
		pdefr->dr_id[PBS_MAXSVRJOBID] ='\0';
		pdefr->dr_preq = preq;
		pdefr->dr_sent = 0;
		append_link(&svr_deferred_req, &pdefr->dr_link, pdefr);
		/* ensure that request is removed if client connect is closed */
		net_add_close_func(preq->rq_conn, clear_from_defr);

		if (schedule_jobs(psched) == -1) {
			/* unable to contact the Scheduler, reject */
			req_reject(PBSE_NOSCHEDULER, 0, preq);
			/* unlink and free the deferred request entry */
			delete_link(&pdefr->dr_link);
			free(pdefr);
		}
		return;
	}


	DBPRT(("req_runjob: received command to run job on destin %s\n",
		preq->rq_ind.rq_run.rq_destin))

	/* OK - go back over the run job request, assign the vhosts */
	/* and finally run the job by calling req_runjob2()	    */

	if (jt == IS_ARRAY_NO) {

		/* just a regular job, pass it on down the line and be done */
		if (call_to_process_hooks(preq, hook_msg, sizeof(hook_msg),
				pbs_python_set_interrupt) == 0) {
			reply_text(preq, PBSE_HOOKERROR, hook_msg);
			return;
		}
		pjob = where_to_runjob(preq, parent);
		if (pjob) {
			/* free prov_vnode before use */
			job_attr_def[(int)JOB_ATR_prov_vnode].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_prov_vnode]);
			req_runjob2(preq, parent);
		}
		return;

	} else if (jt == IS_ARRAY_Single) {

		/* single subjob, if parent qeueud, it can be run */

		if ((pjobsub = parent->ji_ajtrk->tkm_tbl[offset].trk_psubjob) != NULL)
			job_purge(pjobsub);

		if ((pjobsub = create_subjob(parent, jid, &j)) == NULL) {
			req_reject(j, 0, preq);
			return;
		}
		if (call_to_process_hooks(preq, hook_msg, sizeof(hook_msg),
				pbs_python_set_interrupt) == 0) {
			/* subjob reject from hook*/
			job_purge(pjobsub);
			reply_text(preq, PBSE_HOOKERROR, hook_msg);
			return;
		}
		pjob = where_to_runjob(preq, pjobsub);
		if (pjob == NULL) {
			/* requeue subjob */
			pjobsub->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;
			job_purge(pjobsub);
			return;
		}

		/* free prov_vnode before use */
		job_attr_def[(int)JOB_ATR_prov_vnode].at_free(
			&pjob->ji_wattr[(int)JOB_ATR_prov_vnode]);

		req_runjob2(preq, pjob);
		return;

	}

	/* what's left to handle is a range of subjobs,	*/
	/* foreach subjob, if queued, run it		*/

	range = get_index_from_jid(jid);
	if (range == NULL) {
		req_reject(PBSE_IVALREQ, 0, preq);
		return;
	}

	++preq->rq_refct;

	while (1) {
		if ((i = parse_subjob_index(range, &pc, &x, &y, &z, &j)) == -1) {
			req_reject(PBSE_IVALREQ, 0, preq);
			break;
		} else if (i == 1)
			break;
		for (; x <= y; x += z) {
			i = numindex_to_offset(parent, x);
			if (i < 0) {
				continue;
			}

			if (get_subjob_state(parent, i) == JOB_STATE_QUEUED) {
				jid  = mk_subjob_id(parent, i);
				if ((pjobsub = parent->ji_ajtrk->tkm_tbl[i].trk_psubjob) != NULL)
					job_purge(pjobsub);

				if ((pjobsub = create_subjob(parent, jid, &j)) == NULL) {
					req_reject(j, 0, preq);
					continue;
				}
				if (call_to_process_hooks(preq, hook_msg, sizeof(hook_msg),
						pbs_python_set_interrupt) == 0) {
					/* subjob reject from hook*/
					job_purge(pjobsub);
					reply_text(preq, PBSE_HOOKERROR, hook_msg);
					return;
				}
				if ((pjob = where_to_runjob(preq, pjobsub)) == NULL) {
					/* requeue subjob */
					pjobsub->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;
					job_purge(pjobsub);
					continue;
				}
				dup_br_for_subjob(preq, pjob, req_runjob2);
			}
		}
		range = pc;
	}


	/* if not waiting on any running subjobs, can reply; else */
	/* it is taken care of when last running subjob responds  */
	if (--preq->rq_refct == 0)
		reply_send(preq);
	return;
}
/**
 * @brief
 * 		req_runjob - service the Run Job and Asyc Run Job Requests
 *
 * @param[in,out]	preq	-	Run Job Requests
 * @param[in,out]	pjob	-	job pointer
 */
static void
req_runjob2(struct batch_request *preq, job *pjob)
{
	int		  rc;
	int 		  prov_rc=0;
	int		  need_prov;
	char		 *dest;
	int		 rq_type = 0;

	/* Check if prov is required, if so, reply_ack and let prov finish */
	/* else follow normal flow */
	prov_rc = check_and_provision_job(preq, pjob, &need_prov);
	if (prov_rc) { /* problem with the request */
		free_nodes(pjob);
		req_reject(prov_rc, 0, preq);
		return;
	}
	else if (need_prov == 1) { /* prov required and request is fine */
		/* allocate resources right away */
		set_resc_assigned((void *)pjob, 0,  INCR);

		/* provisioning was needed and was qneueued successfully */
		/* Allways send ack for prov jobs, even if not async run */
		reply_ack(preq);
		return;
	}
	/* if need_prov ==0 then no prov required, so continue normal flow */

	dest = preq->rq_ind.rq_run.rq_destin;
	if ((dest == NULL) || (*dest == '\0') || ((*dest == '-') && (*(dest + 1) == '\0'))) {
		if (pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_flags & ATR_VFLAG_SET) {
			dest = pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_val.at_str;
		} else {
			dest = NULL;
		}
	}
	if ((dest == NULL) || (*dest == '\0')) {
		/* Neither the run request nor the job specified an execvnode. */
		free_nodes(pjob);
		req_reject(PBSE_IVALREQ, 0, preq);
		return;
	}
	(void)sprintf(log_buffer, msg_manager, msg_jobrun,
		preq->rq_user, preq->rq_host);
	(void)strcat(log_buffer, " on exec_vnode ");
	rc = LOG_BUF_SIZE - strlen(log_buffer) - 1;
	(void)strncat(log_buffer, dest, rc);
	*(log_buffer + LOG_BUF_SIZE - 1) = '\0';
	log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO,
		pjob->ji_qs.ji_jobid, log_buffer);

	/* If async run, reply now; otherwise reply is handled in */
	/* post_sendmom or post_stagein				  */
	rq_type = preq->rq_type;
	if (preq && (rq_type == PBS_BATCH_AsyrunJob)) {
		reply_ack(preq);
		preq = 0;	/* cleared so we don't try to reuse */
	}

	if (((rc = svr_startjob(pjob, preq)) != 0) &&
		((rq_type == PBS_BATCH_AsyrunJob) || preq)) {
		free_nodes(pjob);
		if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
			/* requeue subjob */
			pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;
			job_purge(pjob);
		}
		if (preq)
			req_reject(rc, 0, preq);
	}
}


/**
 * @brief
 *		clear_exec_on_run_fail - On failure of running a job,
 *		clear exec strings so job can be resecheduled anywhere.
 *
 * @par Functionality:
 *		If the job has been checkpointed or has files staged in, then
 *		the job must run where it ran before or where the files where staged.
 *		Otherwise it is free to run anywhere when re-scheduled.  In this case,
 *		clear the exec_hosts, exec_vnodes, etc.
 *
 * @param[in]	jobp	-	pointer to to job whose run failed
 *
 * @return	none
 *
 * @par MT-safe: yes
 */
void
clear_exec_on_run_fail(job *jobp)
{
	if ((jobp->ji_qs.ji_svrflags &
		(JOB_SVFLG_CHKPT | JOB_SVFLG_StagedIn)) == 0) {

		job_attr_def[(int)JOB_ATR_exec_host].at_free(
			&jobp->ji_wattr[(int)JOB_ATR_exec_host]);
		job_attr_def[(int)JOB_ATR_exec_host2].at_free(
			&jobp->ji_wattr[(int)JOB_ATR_exec_host2]);
		job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
			&jobp->ji_wattr[(int)JOB_ATR_exec_vnode]);
		job_attr_def[(int)JOB_ATR_pset].at_free(
			&jobp->ji_wattr[(int)JOB_ATR_pset]);
		jobp->ji_qs.ji_destin[0] = '\0';
	}
}

/*
 * @brief
 * 		req_stagein	-	service the Stage In Files for a Job Request
 *
 *		This request causes MOM to start stagin in files.
 *		Client must be privileged.
 *
 * @param[in]	preq	-	Job Request
 */

void
req_stagein(struct batch_request *preq)
{
	req_reject(PBSE_NOSUP, 0, preq);
}

/**
 * @brief
 * 		post_stagein - process reply from MOM to stage-in request
 *
 * @param[in]	pwt	-	pointer to work task structure which contains the request
 */

static void
post_stagein(struct work_task *pwt)
{
	int		      code;
	int		      newstate;
	int		      newsub;
	job		     *paltjob;
	job		     *pjob;
	struct batch_request *preq;
	attribute	     *pwait;

	preq = pwt->wt_parm1;
	code = preq->rq_reply.brp_code;
	pjob = find_job(preq->rq_extra);
	free(preq->rq_extra);

	if (pjob != NULL) {

		if (code != 0) {

			/* stage in failed - "wait" job */

			set_resc_assigned((void *)pjob, 0, DECR);
			free_nodes(pjob);
			job_attr_def[(int)JOB_ATR_exec_host].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host]);
			job_attr_def[(int)JOB_ATR_exec_host2].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host2]);
			job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_vnode]);
			job_attr_def[(int)JOB_ATR_pset].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_pset]);

			if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
				/* for subjob, "wait" the parent array */
				paltjob = pjob->ji_parentaj;
			} else {
				/* for regular job, "wait" that job */
				paltjob = pjob;
			}
			pwait = &paltjob->ji_wattr[(int)JOB_ATR_exectime];
			if ((pwait->at_flags & ATR_VFLAG_SET) == 0) {
				pwait->at_val.at_long = time_now + PBS_STAGEFAIL_WAIT;
				pwait->at_flags |= ATR_VFLAG_SET|ATR_VFLAG_MODCACHE;
				(void)job_set_wait(pwait, paltjob, 0);
			}
			(void)svr_setjobstate(paltjob, JOB_STATE_WAITING,
				JOB_SUBSTATE_STAGEFAIL);

			if (preq->rq_reply.brp_choice == BATCH_REPLY_CHOICE_Text)
				svr_mailowner(pjob, MAIL_STAGEIN, MAIL_FORCE,
					preq->rq_reply.brp_un.brp_txt.brp_str);
			if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
				/* requeue subjob */
				pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;
				job_purge(pjob);
			}
		} else {
			/* stage in was successful */
			pjob->ji_qs.ji_svrflags |= JOB_SVFLG_StagedIn;
			if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_STAGEGO) {
				/* continue to start job running */
				(void)svr_strtjob2(pjob, NULL);
			} else {
				svr_evaljobstate(pjob, &newstate, &newsub, 0);
				(void)svr_setjobstate(pjob, newstate, newsub);
			}
		}
	}
	release_req(pwt);	/* close connection and release request */

}

/**
 * @brief
 * 		svr_stagein - direct MOM to stage in the requested files for a job
 *
 * @param[in,out]	pjob	-	job structure
 * @param[in,out]	preq	-	request structure
 * @param[in]	state	-	job state
 * @param[in,out]	substate	-	job substate
 *
 * @return	int
 * @retval	0	- success
 * @retval	non-zero	- error code
 */

static int
svr_stagein(job *pjob, struct batch_request *preq, int state, int substate)
{
	struct batch_request *momreq = 0;
	int		      rc;

	momreq = cpy_stage(momreq, pjob, JOB_ATR_stagein, STAGE_DIR_IN);
	if (momreq) {	/* have files to stage in */

		/* save job id for post_stagein */

		momreq->rq_extra = malloc(PBS_MAXSVRJOBID+1);
		if (momreq->rq_extra == 0)
			return (PBSE_SYSTEM);
		(void)strcpy(momreq->rq_extra, pjob->ji_qs.ji_jobid);
		rc = relay_to_mom(pjob, momreq, post_stagein);
		if (rc == 0) {

			(void)svr_setjobstate(pjob, state, substate);
			/*
			 * show resources allocated as stage-in may take
			 * take sufficient time to run into another
			 * scheduling cycle
			 */
			set_resc_assigned((void *)pjob, 0, INCR);
			/*
			 * stage-in started ok - reply to client as copy may
			 * take too long to wait.
			 */

			if (preq)
				reply_ack(preq);
		} else {
			free(momreq->rq_extra);
		}
		return (rc);

	} else {

		/* no files to stage-in, go direct to sending job to mom */

		return (svr_strtjob2(pjob, preq));
	}
}

/**
 * @brief
 * 		form_attr_comment - Creates and return attribute comment in the given template
 * 		by appending time and execvnode
 *
 * @param[in]	template	-	template of the string
 * @param[in]	execvnode	-	execution node, NULL if this field is not required in the output
 *
 * @return	string
 * @retval	new attribute comment with time and execnode appended.
 *
 * @note
 * 		Do not copy the output of this function into log_buffer. It is used internally.
 */
char *
form_attr_comment(const char *template, const char *execvnode)
{
	char timebuf[128];
	strftime(timebuf, 128, "%a %b %d at %H:%M", localtime(&time_now));
	sprintf(log_buffer, template, timebuf);
	if (execvnode != NULL) {
		strcat(log_buffer, " on ");
		if (strlen(execvnode) > COMMENT_BUF_SIZE-strlen(log_buffer)-1) {
			strncat(log_buffer, execvnode, COMMENT_BUF_SIZE-strlen(log_buffer)-1-3);
			strcat(log_buffer, "...");
			log_buffer[COMMENT_BUF_SIZE-1] = '\0';
		} else
			strcat(log_buffer, execvnode);
	}
	return log_buffer;
}

/**
 * @brief
 * 		svr_startjob - place a job into running state by shipping it to MOM
 *
 * @param[in,out]	pjob	-	job to run
 * @param[in,out]	preq	-	 NULL or Run Job batch request
 *
 * @return	int
 * @retval	0	- success
 * @retval	non-zero	- error code
 */
int
svr_startjob(job *pjob, struct batch_request *preq)
{
	int   f;
	int   rc;
	char *nspec;
	pbs_queue *pque = pjob->ji_qhdr;
	long delay = 10; /* Default value for kill_delay */

	/* if not already setup, transfer the control/script file basename */
	/* into an attribute accessable to MOM				   */

	if (!(pjob->ji_wattr[(int)JOB_ATR_hashname].at_flags & ATR_VFLAG_SET))
		if (job_attr_def[(int)JOB_ATR_hashname].at_decode(
			&pjob->ji_wattr[(int)JOB_ATR_hashname],
			NULL, NULL,
			pjob->ji_qs.ji_jobid))
			return (PBSE_SYSTEM);

	/* clear Exit_status which may have been set in a hook and requeued */

	clear_attr(&pjob->ji_wattr[(int)JOB_ATR_exit_status],
		&job_attr_def[(int)JOB_ATR_exit_status]);

	/* if exec_vnode already set and either (hotstart or checkpoint) */
	/* then reuseuse the host(s) listed in the current exec_vnode	 */

	rc = 0;
	f = pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_flags & ATR_VFLAG_SET;
	if (f && ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HOTSTART) ||
		(pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT))
		&& ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HasNodes) == 0)) {

		nspec = pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_val.at_str;
		if (nspec == NULL)
			return (PBSE_SYSTEM);
		rc = assign_hosts(pjob, nspec, 0);

	} else if (f == 0) {
		/* exec_vnode not already set, use hosts from request   */
		if (preq == NULL)
			return (PBSE_INTERNAL);
		nspec = preq->rq_ind.rq_run.rq_destin;
		if (nspec == NULL)
			return (PBSE_IVALREQ);

		rc = assign_hosts(pjob, nspec, 1);
	}
	if (rc != 0)
		return rc;

	/* Move job_kill_delay attribute from Server to MOM */
	if (pque->qu_attr[(int)QE_ATR_KillDelay].at_flags & ATR_VFLAG_SET)
		delay = pque->qu_attr[(int)QE_ATR_KillDelay].at_val.at_long;
	pjob->ji_wattr[(int)JOB_ATR_job_kill_delay].at_val.at_long = delay;
	pjob->ji_wattr[(int)JOB_ATR_job_kill_delay].at_flags |=
		(ATR_VFLAG_SET | ATR_VFLAG_MODCACHE);


	/* Next, are there files to be staged-in? */

	if ((pjob->ji_wattr[(int)JOB_ATR_stagein].at_flags & ATR_VFLAG_SET) &&
		(pjob->ji_qs.ji_substate != JOB_SUBSTATE_STAGECMP)) {

		/* yes, we do that first; then start the job */

		rc = svr_stagein(pjob, preq, JOB_STATE_RUNNING,
			JOB_SUBSTATE_STAGEGO);

		/* note, the positive acknowledgment to the run job request */
		/* is done by svr_stagein if the stage-in is successful     */

		if (rc != 0) {
			/* If the stage-in failed and we aren't          */
			/* checkpointed, clear the exec_host/exec_vnode; */
			/* job can be run  elsewhere			 */
			if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) == 0) {
				/* clear StagedIn flag for good measure */
				pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_StagedIn;
				job_attr_def[(int)JOB_ATR_exec_host].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_host]);
				job_attr_def[(int)JOB_ATR_exec_host2].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_host2]);
				job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_vnode]);
				job_attr_def[(int)JOB_ATR_pset].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_pset]);
				pjob->ji_modified = 1;
			}
		}

	} else {

		/* No stage-in or already done, start job executing */

		rc = svr_strtjob2(pjob, preq);
	}
	return (rc);
}

/**
 * @brief
 * 		Continue the process of running a job by sending it to Mother Superior,
 *		and making sure it is in JOB_SUBSTATE_PRERUN.
 *
 * @param[in]	pjob - pointer to job to run
 * @param[in]	preq - the run job request from the scheduler or client
 *
 * @return	int
 * @retval	0	:  success, job is being sent to Mom
 * @retval	!0	:  error in trying to send to Mom
 */
static int
svr_strtjob2(job *pjob, struct batch_request *preq)
{
	int	old_state;
	int	old_subst;


	old_state = pjob->ji_qs.ji_state;
	old_subst = pjob->ji_qs.ji_substate;
	pjob->ji_qs.ji_stime = 0;	/* updated in complete_running() */

	/* if not restarting a checkpointed job, increment the run/hop count */

	if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) == 0) {
		++pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long;
		pjob->ji_wattr[(int)JOB_ATR_run_version].at_flags |=
			(ATR_VFLAG_SET | ATR_VFLAG_MODCACHE);
		++pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long;
		pjob->ji_wattr[(int)JOB_ATR_runcount].at_flags |=
			(ATR_VFLAG_SET | ATR_VFLAG_MODCACHE);
	}

	/* send the job to MOM */
	set_attr_svr(&(pjob->ji_wattr[(int) JOB_ATR_Comment]), &job_attr_def[(int) JOB_ATR_Comment],
		form_attr_comment("Job was sent for execution at %s", pjob->ji_wattr[(int) JOB_ATR_exec_vnode].at_val.at_str));

	if (old_subst != JOB_SUBSTATE_PROVISION)
		(void)svr_setjobstate(pjob, JOB_STATE_RUNNING,
			JOB_SUBSTATE_PRERUN);


	if (send_job(pjob, pjob->ji_qs.ji_un.ji_exect.ji_momaddr,
		pjob->ji_qs.ji_un.ji_exect.ji_momport, MOVE_TYPE_Exec,
		post_sendmom, (void *)preq) == 2) {
		pjob->ji_prunreq = preq;
		/* Clear the suspend server flag. */
		pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_Suspend;

		/* in case of async runjob, we need to assign resources
		 * since another scheduling cycle can happen before the
		 * mom responds to the req_commit message. This is the
		 * same logic that is done for jobs with files to stage
		 * in
		 */
		if (preq == NULL || (preq->rq_type == PBS_BATCH_AsyrunJob)) {
			if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)
				set_resc_assigned((void *)pjob, 0, INCR);
		}
		return (0);
	} else {
		log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_NOTICE,
			pjob->ji_qs.ji_jobid,
			"Unable to Run Job, send to Mom failed");

		if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION ||
				pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)
			rel_resc(pjob);
		else
			free_nodes(pjob);

		clear_exec_on_run_fail(pjob);
		svr_evaljobstate(pjob, &old_state, &old_subst, 1);
		(void)svr_setjobstate(pjob, old_state, old_subst);
		return (pbs_errno);
	}
}

/**
 * @brief
 *		Complete the process of placing a job into execution state
 * @par
 *		Records a bunch of information for accouting and resource management,
 *		and sets substate to PRERUN if it isn't already.
 *		The sub moves to SUBSTATE_RUNNING when the session id is received
 *		from Mom, meaning it is in fact running; see stat_update().
 * @par
 *		Note, if a job is in substate PROVISION,  the resources have already
 *		been allocated.
 *
 * @param[in]	jobp	-	pointer to job which is just starting to run.
 */

void
complete_running(job *jobp)
{
	job    *parent;
	time_t	wall;

	if (jobp->ji_qs.ji_stime != 0)
		return;		/* already called for this incarnation */

	/**
	 *	For a subjob, insure the parent array's state is set to 'B'
	 *	and deal with any dependency on the parent.
	 */
	if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
		/* if this is first subjob to run, mark */
		/* parent Array as state "Begun"	*/
		parent = jobp->ji_parentaj;
		if (parent->ji_qs.ji_state == JOB_STATE_QUEUED ||
			(parent->ji_qs.ji_state == JOB_STATE_BEGUN && parent->ji_qs.ji_stime == 0)) {
			svr_setjobstate(parent, JOB_STATE_BEGUN, JOB_SUBSTATE_BEGUN);

			/* Also set the parent job's stime */
			parent->ji_qs.ji_stime = time_now;
			parent->ji_wattr[(int)JOB_ATR_stime].at_val.at_long = time_now;
			parent->ji_wattr[(int)JOB_ATR_stime].at_flags |=
			ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;

			account_jobstr(parent);
			job_attr_def[(int) JOB_ATR_Comment].at_decode(
					&parent->ji_wattr[(int) JOB_ATR_Comment], NULL,
					NULL,
					form_attr_comment("Job Array Began at %s", NULL));
			/* if any dependencies, see if action required */
			if (parent->ji_wattr[(int)JOB_ATR_depend].at_flags&ATR_VFLAG_SET)
				(void)depend_on_exec(parent);

			svr_mailowner(parent, MAIL_BEGIN, MAIL_NORMAL, NULL);
		}
	}
	/* Job started ATR_Comment is set in server since scheduler cannot read	*/
	/* the reply in case of error in asynchronous communication.	*/
	job_attr_def[(int) JOB_ATR_Comment].at_decode(
		&jobp->ji_wattr[(int) JOB_ATR_Comment],
		NULL,
		NULL,
		form_attr_comment("Job run at %s",
			jobp->ji_wattr[(int) JOB_ATR_exec_vnode].at_val.at_str));

	jobp->ji_qs.ji_svrflags &= ~JOB_SVFLG_HOTSTART;

	/* record start time for accounting and for the Scheduler */
	/* setting ji_stime is also an indicator that we have done all this */

	jobp->ji_qs.ji_stime = time_now;
	jobp->ji_wattr[(int)JOB_ATR_stime].at_val.at_long = time_now;
	jobp->ji_wattr[(int)JOB_ATR_stime].at_flags |=
		ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;

	/* compute an upper bound on job end time if possible */
	if ((wall = get_wall(jobp)) != -1)
		jobp->ji_qs.ji_endtBdry = jobp->ji_qs.ji_stime + wall;

	/*
	 * if job is in substate PROVISION, set to PRERUN.
	 * It is possible that the job is in substate:
	 * - RUNNING if Mom sent the status update first before we get to
	 *   process the send_job SIGCHLD, see stat_update()
	 * - EXITING if the Obit was received before send_job's exit status.
	 */
	if (jobp->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION) {
		(void)svr_setjobstate(jobp, JOB_STATE_RUNNING,
			JOB_SUBSTATE_PRERUN);
		/* above saves job structure */
	}

	/* update resource usage attributes */
	/* may have already been done for provisioning, but    */
	/* that will be detected inside of set_resc_assigned() */
	set_resc_assigned((void *)jobp, 0, INCR);
	/* These attributes need to be cleared/freed now that the job has been resumed */
	if (jobp->ji_wattr[(int) JOB_ATR_resc_released].at_flags & ATR_VFLAG_SET) {
		job_attr_def[(int) JOB_ATR_resc_released].at_free(
			&jobp->ji_wattr[(int) JOB_ATR_resc_released]);
		jobp->ji_wattr[(int) JOB_ATR_resc_released].at_flags &= ~ATR_VFLAG_SET;
	}

	if (jobp->ji_wattr[(int) JOB_ATR_resc_released_list].at_flags & ATR_VFLAG_SET) {
		job_attr_def[(int) JOB_ATR_resc_released_list].at_free(
			&jobp->ji_wattr[(int) JOB_ATR_resc_released_list]);
		jobp->ji_wattr[(int) JOB_ATR_resc_released_list].at_flags &= ~ATR_VFLAG_SET;
	}

	/* accounting log for start or restart */
	if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT)
		account_record(PBS_ACCT_RESTRT, jobp, NULL);
	else
		account_jobstr(jobp);

	/* if any dependencies, see if action required */

	if (jobp->ji_wattr[(int)JOB_ATR_depend].at_flags&ATR_VFLAG_SET)
		(void)depend_on_exec(jobp);

	svr_mailowner(jobp, MAIL_BEGIN, MAIL_NORMAL, NULL);
	/*
	 * it is unfortunate, but while the job has gone into execution,
	 * there is no way of obtaining the session id except by making
	 * a status request of MOM.  (Even if the session id was passed
	 * back to the sending child, it couldn't get up to the parent.)
	 */
}

/**
 * @brief
 * 		Helper function to parse the hookname and hook_msg out of a hook rejection message
 *
 * @param[in]	reject_msg	-	The hooks rejection message
 * @param[out]	hook_name	-	pointer to buffer to fill parsed hook_name
 * @param[in]	hook_name_size	- The length of the hook name output buffer
 *
 * @return	The hook message
 * @retval	NULL	: Failed to parse out hook_name and hook_msg
 * @retval	!NULL	: The hook message
 */
static char *
parse_hook_rejectmsg(char *reject_msg, char *hook_name, int hook_name_size)
{
	char *p;
	if (reject_msg != NULL) {
		p = strchr(reject_msg, ',');
		if (p != NULL) {
			*p = '\0';
			p++;
			strncpy(hook_name, reject_msg, hook_name_size);
			return p;
		}
	}
	return NULL;
}

/**
 * @brief
 * 		post_sendmom - clean up action for child started in send_job
 *		which was sending a job "home" to MOM
 * @par
 * 		If send was successfull, mark job as executing.
 * 		See comments in complete_running() above about the possible substate changes.
 *
 * 		The job's session id will be updated with Mom first responds with
 * 		the resources_used.
 *
 * 		If send didn't work, requeue the job.
 *
 * 		If the work_task has a non-null wt_parm2, it is the address of a batch
 * 		request to which a reply must be sent.
 * @par
 * 		If the ji_prunreq (pointer to the run request) is null,  the run request
 * 		has already been replied to.  This might happen if the job's Obit is
 * 		received prior to reaping the send_job child.  In that case, we skip all
 * 		this because the job has already "run" and is now in Exiting state.
 *
 * @param[in,out]	pwt	-	work_task structure
 *
 * Returns: none.
 */
void
post_sendmom(struct work_task *pwt)
{
	int 	newstate;
	int 	newsub;
	int 	r;
	char	*reject_msg = NULL;
	int 	wstat = pwt->wt_aux;
	job 	*jobp = (job *) pwt->wt_parm2;
	struct 	batch_request *preq = (struct batch_request *) pwt->wt_parm1;
	int 	isrpp = pwt->wt_aux2;
	struct	batch_reply *reply = (struct batch_reply *) pwt->wt_parm3;
	char	dest_host[PBS_MAXROUTEDEST + 1];
	char	hook_name[PBS_HOOK_NAME_SIZE + 1] = {'\0'};
	char	*hook_msg = NULL;

	if (jobp == NULL) {
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB, LOG_INFO, "", "post_sendmom failed, jobp NULL");
		if (preq)
			req_reject(PBSE_SYSTEM, 0, preq);
		return;
	}

	DBPRT(("post_sendmom: %s substate is %d", jobp->ji_qs.ji_jobid, jobp->ji_qs.ji_substate))

	if (jobp->ji_prunreq)
		jobp->ji_prunreq = NULL;	/* set in svr_strtjob2() */
	else {
		if (pbs_conf.pbs_use_tcp == 0)
			return; /* reply must have already been handled, see job_obit */
	}

	if (!isrpp) {
		if (WIFEXITED(wstat)) {
			r = WEXITSTATUS(wstat);
		}
#ifndef WIN32
		 else if (WIFSIGNALED(wstat)) { /* Check if send_job child process has been signaled or not */
			r = SEND_JOB_SIGNAL;
			snprintf(log_buffer, LOG_BUF_SIZE, msg_job_end_sig, WTERMSIG(wstat));
			log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_JOB, LOG_INFO,
				jobp->ji_qs.ji_jobid, log_buffer);
		}
#endif
		else {
			r = SEND_JOB_RETRY;
			(void)sprintf(log_buffer, msg_badexit, wstat);
			(void)strcat(log_buffer, __func__);
			log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_JOB, LOG_INFO,
				jobp->ji_qs.ji_jobid, log_buffer);
		}

		/* if the return code says it was hook error read the hook error message
		 * from the file and parse out hookname and hook_msg
		 */
		if ((r == SEND_JOB_HOOKERR) ||
			(r == SEND_JOB_HOOK_REJECT) ||
			(r == SEND_JOB_HOOK_REJECT_RERUNJOB) ||
			(r == SEND_JOB_HOOK_REJECT_DELETEJOB)) {

			char name_buf[MAXPATHLEN + 1];
			int fd;
			struct stat sbuf;

			snprintf(name_buf, sizeof(name_buf), "%s%s%s", path_hooks_workdir, jobp->ji_qs.ji_jobid,
						HOOK_REJECT_SUFFIX);

			if ((stat(name_buf, &sbuf) != -1) && (sbuf.st_size > 0)) {

				if ((fd = open(name_buf, O_RDONLY)) != -1) {

					reject_msg = malloc(sbuf.st_size);
					if (reject_msg != NULL) {
#ifdef WIN32
						setmode(fd, O_BINARY);
#endif
						if (read(fd, reject_msg, sbuf.st_size) != sbuf.st_size) {
							sprintf(log_buffer, "read %s is incomplete", name_buf);
							log_err(errno, __func__, log_buffer);
							reject_msg[0] = '\0';
						}
					}
					close(fd);
					unlink(name_buf);
				}
			}
			hook_msg = parse_hook_rejectmsg(reject_msg, hook_name, PBS_HOOK_NAME_SIZE);
		}

	} else {
		/* in case of rpp, the pbs_errno is set in wstat, based
		 * on which we determine value of r
		 */
		switch (wstat) {
		case PBSE_NONE:
			r = SEND_JOB_OK;
			break;
		case PBSE_NORELYMOM:
			r = SEND_JOB_NODEDW;
			break;
		case PBSE_HOOKERROR:
			r = SEND_JOB_HOOKERR;
			break;
		case PBSE_HOOK_REJECT:
			r = SEND_JOB_HOOK_REJECT;
			break;
		case PBSE_HOOK_REJECT_RERUNJOB:
			r = SEND_JOB_HOOK_REJECT_RERUNJOB;
			break;
		case PBSE_HOOK_REJECT_DELETEJOB:
			r = SEND_JOB_HOOK_REJECT_DELETEJOB;
			break;
		default:
			r = SEND_JOB_FATAL;
			break;
		}

		/* also take note of the reject msg if any */
		if (reply && reply->brp_choice == BATCH_REPLY_CHOICE_Text)
			reject_msg = reply->brp_un.brp_txt.brp_str;

		/* the above reject_msg should never be freed within this function
		 * since it will be freed by the caller process_DreplyRPP() in the
		 * case of a RPP based job send
		 */

		if (r != SEND_JOB_OK) {
			if (reject_msg)
				sprintf(log_buffer,
				        "send of job to %s failed error = %d reject_msg=%s",
				        jobp->ji_qs.ji_destin, pbs_errno, reject_msg);
			else
				sprintf(log_buffer,
					"send of job to %s failed error = %d",
					jobp->ji_qs.ji_destin, pbs_errno);

			log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO, jobp->ji_qs.ji_jobid, log_buffer);
			snprintf(log_buffer, LOG_BUF_SIZE,
					"Not Running: PBS Error: %s", pbse_to_txt(PBSE_MOMREJECT));


			if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
				/*
				 * if the job is a subjob, set the comment of parent job array
				 * only if the job array is not in state begun. Once the job
				 * array starts its comment is set to a begun message and
				 * should not change after that
				 */
				if (jobp->ji_parentaj->ji_qs.ji_state != JOB_STATE_BEGUN) {
					job_attr_def[(int) JOB_ATR_Comment].at_decode(
						&jobp->ji_parentaj->ji_wattr[(int) JOB_ATR_Comment],
						NULL, NULL, log_buffer);
				}
			}

			/* if the job is a normal job or a subjob */
			set_attr_svr(&(jobp->ji_wattr[(int) JOB_ATR_Comment]), &job_attr_def[(int) JOB_ATR_Comment], log_buffer);

		}

		/* in the case of hook error we parse the hook_name and hook msg */
		if ((r == SEND_JOB_HOOKERR) ||
			(r == SEND_JOB_HOOK_REJECT) ||
			(r == SEND_JOB_HOOK_REJECT_RERUNJOB) ||
			(r == SEND_JOB_HOOK_REJECT_DELETEJOB)) {

			hook_msg = parse_hook_rejectmsg(reject_msg, hook_name, PBS_HOOK_NAME_SIZE);
		}
	}


	if (!(jobp->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN   ||
		jobp->ji_qs.ji_substate == JOB_SUBSTATE_RUNNING  ||
		jobp->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION)) {
		sprintf(log_buffer, "send_job returned with exit status = %d and job substate = %d",
		        r, jobp->ji_qs.ji_substate);

		log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_JOB, LOG_INFO,
			jobp->ji_qs.ji_jobid, log_buffer);
	}

	switch (r) {

		case SEND_JOB_OK:		/* send to MOM went ok */

			if (preq)
				reply_ack(preq);
			if ((jobp->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)	||
					(jobp->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION))
				complete_running(jobp);
			break;

#ifndef WIN32
		case SEND_JOB_SIGNAL:

			/* send_job child process has been signaled
			 * therefore kill the job if it is already
			 * running on the MOM and force requeue the job
			 */
			if (preq)
				req_reject(PBSE_SYSTEM, 0, preq);

			/* need to record log message before aborting and
			 * requeuing job both in server and accounting logs
			 */
			snprintf(log_buffer, LOG_BUF_SIZE, "%s", msg_job_abort);
			log_event(PBSEVENT_SYSTEM|PBSEVENT_JOB|PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, LOG_INFO, jobp->ji_qs.ji_jobid, log_buffer);

			/* abort job irrespective of its presence
			 * (may or may not be running) in a MOM
			 */
			(void)job_abt(jobp, log_buffer);

			snprintf(log_buffer, LOG_BUF_SIZE, msg_init_substate, jobp->ji_qs.ji_substate);
			log_event(PBSEVENT_SYSTEM|PBSEVENT_JOB|PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, LOG_INFO, jobp->ji_qs.ji_jobid, log_buffer);

			/* Force requeue the job since the job has been aborted by the server */
			(void)force_reque(jobp);
			break;
#endif
		case SEND_JOB_NODEDW:	/* node (mother superior) is down? */
			mark_node_down(jobp->ji_qs.ji_destin, "could not send job to mom");

			/* fall through to requeue job */

		default :		/* send failed, requeue the job */
			log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_NOTICE,
				jobp->ji_qs.ji_jobid,
				"Unable to Run Job, MOM rejected");

			/* release resources */
			if (jobp->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION ||
					jobp->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)
				rel_resc(jobp);
			else
				free_nodes(jobp);

			/*
			 * If there is a checkpoint, we leave exec_vnode
			 *    to force that it will run there again;
			 * Or if stagein was already successful, we leave
			 *    exec_vnode set as the files are there and
			 *    we dont want to copy them again;
			 * ELSE clear exec_vnode, exec_host, etc.
			 */
			snprintf(dest_host, sizeof(dest_host), "%s", jobp->ji_qs.ji_destin);
			clear_exec_on_run_fail(jobp);

			if (jobp->ji_qs.ji_substate != JOB_SUBSTATE_ABORT) {
				if (preq) {
					if ((r == SEND_JOB_HOOKERR) ||
						(r == SEND_JOB_HOOK_REJECT) ||
						(r == SEND_JOB_HOOK_REJECT_RERUNJOB) ||
						(r == SEND_JOB_HOOK_REJECT_DELETEJOB)) {
						int err;

						if (r == SEND_JOB_HOOK_REJECT)
							err = PBSE_HOOK_REJECT;
						else if (r == SEND_JOB_HOOK_REJECT_RERUNJOB)
							err = PBSE_HOOK_REJECT_RERUNJOB;
						else if (r == SEND_JOB_HOOK_REJECT_DELETEJOB)
							err = PBSE_HOOK_REJECT_DELETEJOB;
						else
							err = PBSE_HOOKERROR;

						log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
							LOG_NOTICE, jobp->ji_qs.ji_jobid, pbse_to_txt(err));

						reply_text(preq, err, hook_msg?hook_msg:"");
					} else {
						req_reject(PBSE_MOMREJECT, 0, preq);
					}
				}

				if (r == SEND_JOB_HOOK_REJECT_DELETEJOB) {
					log_event(PBSEVENT_DEBUG,
						PBS_EVENTCLASS_JOB, LOG_INFO,
						jobp->ji_qs.ji_jobid,
						"Job aborted per a hook rejection");

					/* Need to force queued state so */
					/* job_abt() call does not try   */
					/* to issue a kill job signal to mom */
					jobp->ji_qs.ji_state = JOB_STATE_QUEUED;
					jobp->ji_qs.ji_substate = JOB_SUBSTATE_QUEUED;
					(void)job_abt(jobp, msg_hook_reject_deletejob);
					break;
				} else if ((r == SEND_JOB_HOOKERR) ||
					(r == SEND_JOB_HOOK_REJECT) ||
					(r == SEND_JOB_HOOK_REJECT_RERUNJOB)) {

					if (jobp->ji_wattr[(int)JOB_ATR_runcount].\
				     at_val.at_long >
						PBS_MAX_HOPCOUNT + PBS_MAX_HOPCOUNT) {
						jobp->ji_wattr[(int)JOB_ATR_hold].\
						      at_val.at_long |= HOLD_s;
						jobp->ji_wattr[(int)JOB_ATR_hold].\
							at_flags |=
							ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;
						job_attr_def[(int)JOB_ATR_Comment].\
						at_decode(\
					  &jobp->ji_wattr[(int)JOB_ATR_Comment],
							NULL, NULL,
							"job held, too many failed attempts to run");
					}

					if (r == SEND_JOB_HOOKERR) {
						hook	*phook;
						phook = find_hook(hook_name);
						if (phook != NULL) {
							if ((phook->fail_action & HOOK_FAIL_ACTION_OFFLINE_VNODES) != 0) {
								/*
								 * hook_buf must be large enough
								 * to hold the hook_name and a
								 * small amount of text.
								 */
								char hook_buf[PBS_HOOK_NAME_SIZE + 64];

								snprintf(hook_buf, sizeof(hook_buf),
									"offlined by hook '%s' due to hook error",
									hook_name);
								mark_node_offline_by_mom(dest_host, hook_buf);
							}
							if ((phook->fail_action & HOOK_FAIL_ACTION_SCHEDULER_RESTART_CYCLE) != 0) {

								set_scheduler_flag(SCH_SCHEDULE_RESTART_CYCLE, dflt_scheduler);
								log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_HOOK, LOG_INFO, phook->hook_name, "requested for scheduler to restart cycle");
							}
						}
					}
				}

				svr_evaljobstate(jobp, &newstate, &newsub, 1);
				(void)svr_setjobstate(jobp, newstate, newsub);
				if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
					/* requeue subjob */
					jobp->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;
					job_purge(jobp);
				}
			} else {
				if (preq)
					req_reject(PBSE_BADSTATE, 0, preq);
			}

			break;
	}

	if (!isrpp && reject_msg != NULL)
		free(reject_msg); /* free this only in case of non-rpp since it was locally allocated */

	return;
}

/**
 * @brief
 * 		chk_job_torun - check state and past execution host of a job for which
 *		files are about to be staged in or the job is about to be run.
 * 		Returns pointer to job if all is ok, else returns null.
 *
 *		pjob must be to a existing job structure
 *
 * @param[in,out]	preq	-	Pointer to batch request
 * @param[in]	pjob	-	existing job structure
 *
 * @return	Pointer to job
 * @retval	null	: fail
 */

static job *
chk_job_torun(struct batch_request *preq, job *pjob)
{

	if (pjob == NULL)
		return pjob;


	if ((pjob->ji_qs.ji_state == JOB_STATE_TRANSIT)       ||
		(pjob->ji_qs.ji_state == JOB_STATE_EXITING)	      ||
		(pjob->ji_qs.ji_substate == JOB_SUBSTATE_STAGEGO) ||
		(pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)  ||
		(pjob->ji_qs.ji_substate == JOB_SUBSTATE_RUNNING)) {
		req_reject(PBSE_BADSTATE, 0, preq);
		return NULL;
	}

	if (preq->rq_type == PBS_BATCH_StageIn) {
		if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_STAGEIN) {
			req_reject(PBSE_BADSTATE, 0, preq);
			return NULL;
		}
	}
	return (pjob);
}
/**
 * @brief
 * 		where to execute the job
 *
 * @param[in,out]	preq	-	Pointer to batch request
 * @param[in,out]	pjob	-	existing job structure
 *
 * @return	Pointer to job
 * @retval	null	: fail
 */
static job *
where_to_runjob(struct batch_request *preq, job *pjob)
{
	char		 *nspec;
	struct rq_runjob *prun = &preq->rq_ind.rq_run;
	int 		  rc;

	if ((pjob->ji_qs.ji_svrflags & (JOB_SVFLG_CHKPT | JOB_SVFLG_StagedIn)) ||
		((prun->rq_destin != NULL) && (*prun->rq_destin == '-') && (*(prun->rq_destin + 1) == '\0'))) {
		/* Job has files staged, a checkpoint image, or "qrun -H -" was specified.	*/
		/* Reuse assigned resources.							*/
		if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HasNodes) == 0) {
			/* re-reserve nodes and leave exec_vnode as is */
			/* convert exec_vnode string into form like user spec */
			nspec = pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_val.at_str;
			if (nspec == NULL) {
				/* something's wrong, before we reject the */
				/* job let us clear the flags so the job can */
				/* run the next time around  */
				pjob->ji_qs.ji_svrflags &= ~(JOB_SVFLG_CHKPT |
					JOB_SVFLG_StagedIn);
				req_reject(PBSE_IVALREQ, 0, preq);
				return NULL;
			}
			if ((rc = assign_hosts(pjob, nspec, 0)) != 0) {
				free(nspec);
				req_reject(rc, 0, preq);
				return NULL;
			}
		}
	} else {

		/* job has not run before or need not run there again	*/
		/* reserve nodes and set exec_vnode anew		*/

		if ((prun->rq_destin == NULL) ||
			(*prun->rq_destin == '\0')) {
			req_reject(PBSE_IVALREQ, 0, preq);
			return NULL;
		}

		if ((pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_flags & ATR_VFLAG_SET) != 0) {
			/*
			 * Instruct MoM to discard the existing job before we assign new
			 * resources. This ensures previously assigned resources are cleaned
			 * up properly and prevents orphaned processes. If the job is not
			 * discarded, files and directories created for the job will linger.
			 */
			discard_job(pjob, "Force qrun", 1);
		}

		rc = assign_hosts(pjob, prun->rq_destin, 1);

		if (rc != 0) {
			req_reject(rc, 0, preq);
			return NULL;
		}
	}

	/* If the request did not come from the scheduler, update the comment. */
	if (find_sched_from_sock(preq->rq_conn) == NULL) {
		char comment[MAXCOMMENTLEN];
		nspec = pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_val.at_str;
		if ((nspec != NULL) && (*nspec != '\0')) {
			snprintf(comment, MAXCOMMENTLEN, "Job manually qrun on %s", nspec);
		} else {
			snprintf(comment, MAXCOMMENTLEN, "Job manually qrun.");
		}
		job_attr_def[(int)JOB_ATR_Comment].at_decode(
			&pjob->ji_wattr[(int)JOB_ATR_Comment],
			NULL, NULL, comment);
	}

	return (pjob);
}

/**
 * @brief
 * 		assign_hosts - assign hosts (vnodes) to job which are specified (given) by:
 *		1. the scheduler when it runs a job,
 *		2. the operator as the -H option to qrun
 *		3. from exec_vnode when required by checkpoint-restart or file stage-in
 *
 * @param[in,out]	pjob	-	pointer to a job object
 * @param[in]	given	-	original vnode list from scheduler/operator
 * @param[in]	set_exec_vnode	-	if True (non-zero), this function is to create
 *                              	a new hoststr including new job indicies,
 *                              	otherwise return existing exec_host unchanged.
 *
 * @return	int
 * @retval	0	: success
 * @retval	!0	: error code
 */

int
assign_hosts(job  *pjob, char *given, int set_exec_vnode)
{
	char		*hoststr;
	char            *hoststr2;
	char		*vnodestoalloc;
	pbs_net_t	 momaddr = 0;
	unsigned int	 port;
	int		 rc = 0;

	if (svr_totnodes == 0) 	/* Must have nodes file */
		return (PBSE_NONODES);

	if (given == NULL)
		return (PBSE_IVALREQ);

	/* allocate the execution nodes and resources */

	if ((set_exec_vnode == 0) &&
		(pjob->ji_wattr[(int) JOB_ATR_exec_host].at_flags & ATR_VFLAG_SET)) {
		hoststr = pjob->ji_wattr[(int) JOB_ATR_exec_host].at_val.at_str;
		hoststr2 = pjob->ji_wattr[(int) JOB_ATR_exec_host2].at_val.at_str;
	} else {
		hoststr = NULL;
		hoststr2 = NULL;
	}

	rc = set_nodes((void *)pjob, JOB_OBJECT, given, &vnodestoalloc, &hoststr, &hoststr2,
		set_exec_vnode, FALSE);

	if (rc == 0) {
		if (set_exec_vnode) {
			job_attr_def[(int)JOB_ATR_exec_host].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host]);
			job_attr_def[(int)JOB_ATR_exec_host2].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host2]);
			job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_exec_vnode]);
			(void)job_attr_def[(int)JOB_ATR_exec_vnode].at_decode(
				&pjob->ji_wattr[(int)JOB_ATR_exec_vnode],
				NULL,
				NULL,
				vnodestoalloc);
			(void)job_attr_def[(int)JOB_ATR_exec_host].at_decode(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host],
				NULL,
				NULL,
				hoststr);
			(void)job_attr_def[(int)JOB_ATR_exec_host2].at_decode(
				&pjob->ji_wattr[(int)JOB_ATR_exec_host2],
				NULL,
				NULL,
				hoststr2);
			pjob->ji_modified = 1;
		} else {
			/* leave exec_vnode alone and reuse old IP address */
			momaddr = pjob->ji_qs.ji_un.ji_exect.ji_momaddr;
			port    = pjob->ji_qs.ji_un.ji_exect.ji_momport;

		}
		(void)strncpy(pjob->ji_qs.ji_destin,
			parse_servername(hoststr, NULL),
			PBS_MAXROUTEDEST);
		if (momaddr == 0) {
			momaddr = get_addr_of_nodebyname(pjob->ji_qs.ji_destin,
				&port);
			if (momaddr == 0) {
				free_nodes(pjob);
				job_attr_def[(int)JOB_ATR_exec_host].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_host]);
				job_attr_def[(int)JOB_ATR_exec_host2].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_host2]);
				job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_exec_vnode]);
				job_attr_def[(int)JOB_ATR_pset].at_free(
					&pjob->ji_wattr[(int)JOB_ATR_pset]);
				return (PBSE_BADHOST);
			}
		}
		pjob->ji_qs.ji_un.ji_exect.ji_momaddr = momaddr;
		pjob->ji_qs.ji_un.ji_exect.ji_momport = port;
	}
	return (rc);
}

/**
 * @brief
 * 		req_defschedreply - handle the deferred scheduler reply call
 *
 * @param[in,out]	preq	-	Pointer to batch request
 */

void
req_defschedreply(struct batch_request *preq)
{
	struct deferred_request  *pdefr;

	if (preq->rq_ind.rq_defrpy.rq_cmd != SCH_SCHEDULE_AJOB) {
		req_reject(PBSE_IVALREQ, 0, preq);
		return;
	}

	for (pdefr = (struct deferred_request *)GET_NEXT(svr_deferred_req);
		pdefr;
		pdefr = (struct deferred_request *)GET_NEXT(pdefr->dr_link)) {
		if (strcmp(preq->rq_ind.rq_defrpy.rq_id, pdefr->dr_id) == 0)
			break;
	}

	if (pdefr == NULL) {
		req_reject(PBSE_UNKJOBID, 0, preq);
		return;
	}

	/* reply to the original (deferred) request */
	/* if the connection for the original request (qrun) was closed */
	/* the pointer to it will have been nulled */
	if (pdefr->dr_preq != NULL) {
		/* "preq" points to the deferred reply from the Scheduler  */
		/* "pdefr" points to the original qrun batch request, this */
		/* request structure will be freed on the reply            */

		if (preq->rq_ind.rq_defrpy.rq_txt) {
			/* have a text string from the Scheduler to send to qrun */
			reply_text(pdefr->dr_preq, preq->rq_ind.rq_defrpy.rq_err,
				preq->rq_ind.rq_defrpy.rq_txt);

		} else if (preq->rq_ind.rq_defrpy.rq_err == 0) {
			/* no error, acknowledge qrun */
			reply_send(pdefr->dr_preq);

		} else {
			/* was an error (without text string), send error to qrun */
			req_reject(preq->rq_ind.rq_defrpy.rq_err, 0, pdefr->dr_preq);
		}
	}

	/* unlink and free the deferred request entry */
	delete_link(&pdefr->dr_link);
	free(pdefr);

	reply_send(preq);
}
