/*
 * 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_jobobit.c
 *
 * @brief
 * 		req_jobobit.c - functions dealing with a Job Obituary Request (Notice)
 *		and the associated post execution job clean up.
 *
 * Functions included are:
 * 	setup_from()
 * 	setup_cpyfiles()
 * 	is_joined()
 * 	cpy_stdfile()
 * 	cpy_stage()
 * 	mom_comm()
 * 	rel_resc()
 * 	on_exitrerun_msg()
 * 	conn_to_mom_failed()
 * 	on_job_exit()
 * 	on_job_rerun()
 * 	setrerun()
 * 	job_obit()
 */

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

#include <sys/types.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#ifdef WIN32
#include <windows.h>
#include "win.h"
#include <time.h>
#include <sys/timeb.h>
#endif

#include "libpbs.h"
#include "server_limits.h"
#include "list_link.h"
#include "attribute.h"
#include "resource.h"
#include "server.h"
#include "job.h"
#include "credential.h"
#include "ticket.h"
#include "batch_request.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 "mom_server.h"
#include "dis.h"
#include "rpp.h"
#include "libutil.h"
#include "pbs_sched.h"


/* External Global Data Items */

extern unsigned int   pbs_mom_port;
extern char *path_spool;
extern int   server_init_type;
extern pbs_net_t pbs_server_addr;
extern char *msg_init_abt;
extern char *msg_job_end;
extern char *msg_job_end_sig;
extern char *msg_job_end_stat;
extern char *msg_momnoexec1;
extern char *msg_momnoexec2;
extern char *msg_baduser;
extern char *msg_job_globfail1;
extern char *msg_obitnojob;
extern char *msg_obitnocpy;
extern char *msg_obitnodel;
extern char *msg_bad_password;
extern char *msg_hook_reject_deletejob;
extern char *msg_hook_reject_rerunjob;
extern struct connect_handle connection[];
extern time_t time_now;

/* External Functions called */

extern void set_resc_assigned(void *, int,  enum batch_op);
extern long get_walltime(job *, int);

/* Local public functions  */

void	job_obit(struct resc_used_update *, int s);
void	on_job_rerun(struct work_task *ptask);
void	on_job_exit(struct work_task *ptask);
extern void set_admin_suspend(job *pjob, int set_remove_nstate);

/* Local private functions */

static struct batch_request *setup_cpyfiles
(struct batch_request *, job *, char *,  char *, int, int);

static char *msg_obitnotrun = "job not running, may have been requeued on node failure";
/**
 * @brief
 * 		setup_from - setup the "from" name for a standard job file:
 *		output, error, or chkpt
 *
 * @param[in]	pjob	- job structure
 * @param[in]	suffix	- suffix for the "from" name
 *
 * @return	"from" name
 */

static char *
setup_from(job  *pjob, char *suffix)
{
	char *from;

	from = malloc(strlen(pjob->ji_qs.ji_jobid) + strlen(suffix) + 1);
	if (from) {
		(void)strcpy(from, pjob->ji_qs.ji_jobid);
		(void)strcat(from, suffix);
	}
	return (from);
}

/**
 * @brief
 * 		setup_cpyfiles - if need be, allocate and initialize a Copy Files
 *		batch request, then append the file pairs
 *
 * @param[in]	preq	- batch request
 * @param[in]	pjob	- job structure
 * @param[in]	from	- local (to mom) name
 * @param[in]	to		- remote (destination) name
 * @param[in]	direction	- copy direction
 * @param[in]	tflag	- 1 if stdout or stderr , 2 if stage out or in
 *
 * @return	modified batch request.
 * @retval	NULL	- failure
 */

static
struct batch_request *
setup_cpyfiles(struct batch_request *preq, job  *pjob, char *from, char *to, int  direction, int  tflag)
{
	struct rq_cpyfile *pcf;
	struct rq_cpyfile_cred *pcfc;
	struct rqfpair    *pair;
	size_t		  cred_len;
	char		  *cred = NULL;
	char		  *prq_jobid;
	char		  *prq_owner;
	char		  *prq_user;
	char		  *prq_group;
	int 		  *prq_dir;
	pbs_list_head	  *prq_pair;

	/* if this is a sub job of an array job, then check to see if the */
	/* index needs to be substituted in the paths			  */

	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
		to   = subst_array_index(pjob, to);
		from = subst_array_index(pjob, from);
	}

	if (preq == NULL) {
		char		*momname;

		/* check that certain required attributues are valid */

		if ((pjob->ji_wattr[(int)JOB_ATR_job_owner].at_val.at_str==NULL) ||
			(pjob->ji_wattr[(int)JOB_ATR_euser].at_val.at_str == NULL)) {
			/* this case shouldn't happen, log it and don't do copy     */
			/* use null jobid, if attr missing, jobid is likely bad too */

			log_event(PBSEVENT_ERROR|PBSEVENT_JOB, PBS_EVENTCLASS_FILE,
				LOG_INFO,  "",
				"cannot copy files for job, owner/euser missing");
			if (from)
				free(from);
			if (to)
				free(to);
			if (cred)
				free(cred);
			return NULL;
		}
		/* allocate and initialize the batch request struct */

		momname = parse_servername(pjob->ji_wattr[(int)JOB_ATR_exec_vnode].
			at_val.at_str, NULL);
		if (get_credential(momname, pjob, PBS_GC_CPYFILE,
			&cred, &cred_len) == 0) {
			preq = alloc_br(PBS_BATCH_CopyFiles_Cred);
		} else {
			preq = alloc_br(PBS_BATCH_CopyFiles);
		}

		if (preq == NULL) {
			if (from)
				free(from);
			if (to)
				free(to);
			if (cred)
				free(cred);
			return (preq);
		}

		if (preq->rq_type == PBS_BATCH_CopyFiles_Cred) {
			preq->rq_ind.rq_cpyfile_cred.rq_credtype =
				pjob->ji_extended.ji_ext.ji_credtype;
			preq->rq_ind.rq_cpyfile_cred.rq_pcred = cred;
			preq->rq_ind.rq_cpyfile_cred.rq_credlen = cred_len;
			pcfc = &preq->rq_ind.rq_cpyfile_cred;
			prq_jobid = pcfc->rq_copyfile.rq_jobid;
			prq_owner = pcfc->rq_copyfile.rq_owner;
			prq_user = pcfc->rq_copyfile.rq_user;
			prq_group = pcfc->rq_copyfile.rq_group;
			prq_dir = &pcfc->rq_copyfile.rq_dir;
			prq_pair = &pcfc->rq_copyfile.rq_pair;
		} else {
			pcf = &preq->rq_ind.rq_cpyfile;
			prq_jobid = pcf->rq_jobid;
			prq_owner = pcf->rq_owner;
			prq_user = pcf->rq_user;
			prq_group = pcf->rq_group;
			prq_dir = &pcf->rq_dir;
			prq_pair = &pcf->rq_pair;
		}
		CLEAR_HEAD((*prq_pair));

		/* copy jobid, owner, exec-user, group names, upto the @host part */

		(void)strcpy(prq_jobid, pjob->ji_qs.ji_jobid);
		get_jobowner(pjob->ji_wattr[(int)JOB_ATR_job_owner].at_val.at_str,
			prq_owner);
		get_jobowner(pjob->ji_wattr[(int)JOB_ATR_euser].at_val.at_str,
			prq_user);
		if (((pjob->ji_wattr[(int)JOB_ATR_egroup].at_flags &
			ATR_VFLAG_DEFLT) ==0) &&
			(pjob->ji_wattr[(int)JOB_ATR_egroup].at_val.at_str != 0)) {
			strcpy(prq_group,
				pjob->ji_wattr[(int)JOB_ATR_egroup].at_val.at_str);
		}
		else
			prq_group[0] = '\0';	/* default: use login group */

		*prq_dir = direction;

		/* set "sandbox=PRIVATE" mode */
		if (pjob->ji_wattr[(int)JOB_ATR_sandbox].at_flags & ATR_VFLAG_SET) {
			/* set STAGE_JOBDIR mode based on job settings */
			if (strcasecmp(pjob->ji_wattr[(int)JOB_ATR_sandbox].at_val.at_str, "PRIVATE") == 0) {
				*prq_dir |= STAGE_JOBDIR;
			}
		} /* O_WORKDIR check would go here */

	} else {

		/* use the existing request structure */

		if (preq->rq_type == PBS_BATCH_CopyFiles_Cred) {
			pcfc = &preq->rq_ind.rq_cpyfile_cred;
			prq_pair = &pcfc->rq_copyfile.rq_pair;
		} else {
			pcf = &preq->rq_ind.rq_cpyfile;
			prq_pair = &pcf->rq_pair;
		}
	}

	pair = (struct rqfpair *)malloc(sizeof(struct rqfpair));
	if (pair == NULL) {
		free_br(preq);
		return NULL;
	}

	CLEAR_LINK(pair->fp_link);
	pair->fp_local  = from;
	pair->fp_rmt    = to;
	pair->fp_flag   = tflag;
	append_link(prq_pair, &pair->fp_link, pair);
	return (preq);
}
/**
 * @brief
 * 		is_joined - Is the file joined to another.
 *
 * @param[in]	pjob	- job structure
 * @param[in]	ati	- job attribute, output/error path.
 *
 * @return	joined or not
 * @retval	0	- either the first or not in list.
 * @retval	1	- being joined.
 */
static int
is_joined(pjob, ati)
job		*pjob;
enum job_atr	 ati;
{
	char       key;
	attribute *pattr;
	char	  *pd;

	if (ati == JOB_ATR_outpath)
		key = 'o';
	else if (ati == JOB_ATR_errpath)
		key = 'e';
	else
		return (0);
	pattr = &pjob->ji_wattr[(int)JOB_ATR_join];
	if (pattr->at_flags & ATR_VFLAG_SET) {
		pd = pattr->at_val.at_str;
		if (pd && *pd && (*pd != 'n')) {
			/* if not the first letter, and in list - is joined */
			if ((*pd != key) && (strchr(pd+1, (int)key)))
				return (1);	/* being joined */
		}
	}
	return (0);	/* either the first or not in list */
}

/**
 * @brief
 * 		cpy_stdfile - determine if one of the job's standard files (output or error)
 *		is to be copied, if so set up the Copy Files request.
 *
 * @param[in]	preq	- batch request
 * @param[in]	pjob	- job structure
 * @param[in]	ati	- JOB_ATR_, output/error path.
 *
 * @return	modified batch request.
 * @retval	NULL	failure
 */

static struct batch_request *cpy_stdfile(struct batch_request *preq, job *pjob, enum job_atr ati)
{
	char *from;
	char  key;
	attribute *jkpattr;
	attribute *pathattr = &pjob->ji_wattr[(int)ati];
	char *suffix;
	char *to = NULL;

	/* if the job is interactive, don't bother to return output file */

	if (pjob->ji_wattr[(int)JOB_ATR_interactive].at_flags &&
		pjob->ji_wattr[(int)JOB_ATR_interactive].at_val.at_long)
		return NULL;

	/* set up depending on which file */

	if (ati == JOB_ATR_errpath) {
		key    = 'e';
		suffix = JOB_STDERR_SUFFIX;
	} else {
		key    = 'o';
		suffix = JOB_STDOUT_SUFFIX;
	}

	if ((pathattr->at_flags & ATR_VFLAG_SET) == 0) { /* This shouldn't be */

		(void)sprintf(log_buffer, "%c file missing", key);
		log_event(PBSEVENT_ERROR|PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
			LOG_INFO,  pjob->ji_qs.ji_jobid, log_buffer);
		return NULL;
	}

	/* Is the file joined to another, if so don't copy it */

	if (is_joined(pjob, ati))
		return (preq);

	/*
	 * If the job has a keep file attribute, and the specified file is in
	 * the keep list, MOM has already placed the file in the user's HOME
	 * directory.  It don't need to be copied.
	 */

	jkpattr = &pjob->ji_wattr[(int)JOB_ATR_keep];
	if ((jkpattr->at_flags & ATR_VFLAG_SET) &&
		strchr(jkpattr->at_val.at_str, key) && !strchr(jkpattr->at_val.at_str, 'd'))

		return (preq);

	/*
	 * If the job has a remove file attribute and the job has succeeded,
	 * std_files doesn't has to be copied.
	 */
	if ((pjob->ji_wattr[(int)JOB_ATR_exit_status].at_flags) &
					ATR_VFLAG_SET) {
		if (pjob->ji_wattr[(int) JOB_ATR_exit_status].at_val.at_long
				== JOB_EXEC_OK) {
			jkpattr = &pjob->ji_wattr[(int) JOB_ATR_remove];
			if ((jkpattr->at_flags & ATR_VFLAG_SET)
					&& (strchr(jkpattr->at_val.at_str, key)))
				return (preq);
		}
	}

	/* else go with the supplied name */

	to = malloc(strlen(pathattr->at_val.at_str) + 1);
	if (to) {
		(void)strcpy(to, pathattr->at_val.at_str);

	} else
		return (preq);	/* cannot continue with this one */

	/* build up the name used by MOM as the from name */

	from = setup_from(pjob, suffix);
	if (from == NULL) {
		(void)free(to);
		return (preq);
	}

	/* now set names into the batch request */

	return (setup_cpyfiles(preq, pjob, from, to, STAGE_DIR_OUT, STDJOBFILE));
}

/**
 * @brief
 * 		cpy_stage - set up a Copy Files request to include files specified by the
 *		user to be staged out (also used for stage-in).
 *		"stage_out" is a resource that may or may not *	exist on a host.
 *		If such exists, the files are listed one per string as
 *		"local_name@remote_host:remote_name".
 *
 * @param[in]	preq	- batch request
 * @param[in]	pjob	- job structure
 * @param[in]	ati	- JOB_ATR_stageout
 * @param[in]	direction	-  1 = , 2 =
 *
 * @return	batch_request *
 */

struct batch_request *cpy_stage(struct batch_request *preq, job *pjob, enum job_atr ati, int direction)
{
	int		      i;
	char		     *from;
	attribute 	     *pattr;
	struct array_strings *parst;
	char 		     *plocal;
	char		     *prmt;
	char		     *to;

	pattr = &pjob->ji_wattr[(int)ati];
	if (pattr->at_flags & ATR_VFLAG_SET) {

		/* at last, we know we have files to stage out/in */

		parst = pattr->at_val.at_arst;
		for (i = 0; i<parst->as_usedptr; ++i) {
			plocal = parst->as_string[i];
			prmt   = strchr(plocal, (int)'@');
			if (prmt) {
				*prmt = '\0';
				from = malloc(strlen(plocal)+1);
				if (from) {
					(void)strcpy(from, plocal);
					*prmt = '@';	/* restore the @ */
				} else {
					return (preq);
				}
				to   = malloc(strlen(prmt+1) + 1);
				if (to) {
					(void)strcpy(to, prmt+1);
				} else {
					(void)free(from);
					return (preq);
				}
				preq = setup_cpyfiles(preq, pjob, from, to,
					direction, STAGEFILE);
			}
		}
	}

	return (preq);
}


/**
 * @brief
 * 		mom_comm - if needed, open a connection with the MOM under which
 *		the job was running.  The connection is typically set up by
 *		req_jobobit() using the connection already established by MOM.
 *		However, on server recovery there will be no pre-established connection.
 *
 *		If a connection is needed and cannot be setup, set up a work-task
 *		entry and try again later.
 *
 * @param[in]	pjob	- job structure
 * @param[in]	func	- function pointer which accepts a work task structure and returns void
 * 							here it can calls on_job_exit and on_job_rerun
 *
 * @return	open connection handle to MOM
 * @retval	-1	- failure
 */

int
mom_comm(job *pjob, void (*func)(struct work_task *))
{
	unsigned int dum;
	long t;
	attribute *peh;
	struct work_task *pwt;
	int prot = PROT_TCP;

	if (pbs_conf.pbs_use_tcp == 1)
		prot = PROT_RPP;

	if (pjob->ji_momhandle < 0) {

		/* need to make connection, called from pbsd_init() */

		if (pjob->ji_qs.ji_un.ji_exect.ji_momaddr == 0) {

			peh = &pjob->ji_wattr[(int)JOB_ATR_exec_vnode];
			if (((peh->at_flags & ATR_VFLAG_SET) == 0) ||
				(peh->at_val.at_str == 0))
				return (-1);
			pjob->ji_qs.ji_un.ji_exect.ji_momaddr =
				get_addr_of_nodebyname(peh->at_val.at_str, &dum);
			if (pjob->ji_qs.ji_un.ji_exect.ji_momaddr == 0)
				return (-1);

			pjob->ji_qs.ji_un.ji_exect.ji_momport = dum;
		}
		pjob->ji_momhandle = svr_connect(
			pjob->ji_qs.ji_un.ji_exect.ji_momaddr,
			pjob->ji_qs.ji_un.ji_exect.ji_momport,
			process_Dreply,
			ToServerDIS,
			prot);
		pjob->ji_mom_prot = prot;

		if (pjob->ji_momhandle < 0) {
			char	*operation;

			t = pjob->ji_retryok++;
			t = PBS_NET_RETRY_TIME + t * t;

			if (func == on_job_exit)
				operation = "exit";
			else if (func == on_job_rerun)
				operation = "rerun";
			else
				operation = "UNKNOWN";

			snprintf(log_buffer, sizeof(log_buffer),
				"cannot connect to MOM, reschedule job %s "
				"in %ld seconds", operation, t);
			log_err(-1, pjob->ji_qs.ji_jobid, log_buffer);

			t += time_now;
			pwt = set_task(WORK_Timed, t, func, (void *)pjob);
			append_link(&pjob->ji_svrtask, &pwt->wt_linkobj, pwt);
			return (-1);
		}
	}
	return (pjob->ji_momhandle);
}

/**
 * @brief
 * 		rel_resc - release resources assigned to the job
 *
 * @param[in]	pjob	- job structure
 */
void
rel_resc(job *pjob)
{
	conn_t *conn = NULL;
	pbs_sched *psched;

	free_nodes(pjob);

	/* removed the resources used by the job from the used svr/que attr  */

	set_resc_assigned((void *)pjob, 0,  DECR);

	/* is there a rerun request waiting for acknowledgement that        */
	/* resources (including licenses) are indeed released? Then ack it. */
	if (pjob->ji_rerun_preq != NULL) { /* set only in req_rerun() */
		if (pjob->ji_rerun_preq->rq_conn != PBS_LOCAL_CONNECTION)
			conn = get_conn(pjob->ji_rerun_preq->rq_conn);

		reply_ack(pjob->ji_rerun_preq);

		/* clear no-timeout flag on connection to prevent stale connections */
		if (conn)
			conn->cn_authen &= ~PBS_NET_CONN_NOTIMEOUT;

		pjob->ji_rerun_preq = NULL;
	}
	if(pjob->ji_qs.ji_svrflags & JOB_SVFLG_AdmSuspd)
		set_admin_suspend(pjob, 0);

	/* Mark that scheduler should be called */

	if (find_assoc_sched_jid(pjob->ji_qs.ji_jobid, &psched))
		set_scheduler_flag(SCH_SCHEDULE_TERM, psched);
	else {
		pbs_queue *pq;
		pq = find_queuebyname( pjob->ji_qs.ji_queue);
		sprintf(log_buffer, "Unable to reach scheduler associated with partition %s", pq->qu_attr[QA_ATR_partition].at_val.at_str);
		log_err(-1, __func__, log_buffer);
	}
}
/**
 * @brief
 * 		on_exitrerun_msg	- log message on exit rerun fails, used with
 * 								on_job_rerun() and conn_to_mom_failed()
 *
 * @param[in]	pjob	- job which has failed
 * @param[in]	fmt	- failure message
 */
static void
on_exitrerun_msg(job *pjob, char *fmt)
{
	char	       *hostname = " ? ";

	if (pjob->ji_qs.ji_destin[0] != '\0')
		hostname = pjob->ji_qs.ji_destin;

	sprintf(log_buffer, fmt, pjob->ji_qs.ji_jobid, hostname);
	log_event(PBSEVENT_ERROR|PBSEVENT_ADMIN|PBSEVENT_JOB,
		PBS_EVENTCLASS_JOB, LOG_INFO,
		pjob->ji_qs.ji_jobid, log_buffer);
}

/**
 * @brief
 * 		conn_to_mom_failed - called when the connection to Mom for end of job
 *		processing is broken (Mom gone?).  Log it and close and attempt to
 *		open a new one by going around again.
 *
 * @param[in]	pjob	- job structure
 * @param[in]	func	- function pointer which accepts a work task structure and returns void
 * 							here it can calls on_job_exit and on_job_rerun
 */

static void
conn_to_mom_failed(job *pjob, void(*func)(struct work_task *))
{
	struct work_task *ptask;

	on_exitrerun_msg(pjob, "end of job processing for %s, connection to Mom on host %s was broken");
	if (pjob->ji_mom_prot == PROT_TCP) {
		svr_disconnect(pjob->ji_momhandle);
	} else {
		rpp_close(pjob->ji_momhandle);
	}
	pjob->ji_momhandle = -1;
	ptask = set_task(WORK_Immed, 0, func, pjob);
	append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
	return;
}

/**
 * @brief
 * 		continue post-execution processing of a job that terminated.
 *
 *		This function is called by pbsd_init() on recovery, by job_obit()
 *		on job termination and by itself (via a work task).  The clue to where
 *		we are is the job substate and the type of the work task entry it is
 *		called with.  If the work task entry type is Work_Immed, then this is
 *		the first time in for the job substate.  Otherwise it is with the reply
 *		given by MOM.
 *
 *		NOTE:
 *		On the initial work task (WORK_Immed), the wt_parm1 is a job pointer.
 *		On a call-back work task (WORK_Deferred_Reply) generated by
 *		send_request(), the wt_parm1 is pointing to the request; and the
 *		rq_extra field in the request points to the job.
 *
 * @param[in,out]	ptask	- work task
 */
void
on_job_exit(struct work_task *ptask)
{
	int    handle;
	job   *pjob;
	struct batch_request *preq;
	struct work_task *pt;
	char  *rec;
	int    rc;
	int    stageout_status = 1; /* success */
	long   t;
	pbs_list_head	*mom_tasklist_ptr = NULL;
	mominfo_t *pmom = 0;
	int	release_nodes_on_stageout = 0;

	if (ptask->wt_type != WORK_Deferred_Reply) {
		preq = NULL;
		pjob = (job *)ptask->wt_parm1;
	} else {
		preq = (struct batch_request *)ptask->wt_parm1;
		pjob = (job *)preq->rq_extra;
	}

	if ((pjob->ji_wattr[(int)JOB_ATR_relnodes_on_stageout].at_flags & ATR_VFLAG_SET) &&
	   (pjob->ji_wattr[(int)JOB_ATR_relnodes_on_stageout].at_val.at_long != 0)) {
		release_nodes_on_stageout = 1;
	}

	/* minor check on validity of pjob */

	if (isdigit((int)pjob->ji_qs.ji_jobid[0]) == 0)
		return;		/* not pointing to currently valid job */

	if ((handle = mom_comm(pjob, on_job_exit)) < 0)
		return;

	if (pjob->ji_mom_prot == PROT_RPP) {
		pmom = tfind2((unsigned long) pjob->ji_qs.ji_un.ji_exect.ji_momaddr,
			pjob->ji_qs.ji_un.ji_exect.ji_momport,
			&ipaddrs);
		if (!pmom || (((mom_svrinfo_t *)(pmom->mi_data))->msr_state & INUSE_DOWN))
			return;
		mom_tasklist_ptr = &(((mom_svrinfo_t *)(pmom->mi_data))->msr_deferred_cmds);
	}

	switch (pjob->ji_qs.ji_substate) {

		case JOB_SUBSTATE_EXITING:
		case JOB_SUBSTATE_ABORT:


			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_STAGEOUT);
			ptask->wt_type = WORK_Immed;

			/* Initialize retryok */
			pjob->ji_retryok = 0;

			/* NO BREAK, fall into stage out processing */

		case JOB_SUBSTATE_STAGEOUT:

			if (ptask->wt_type != WORK_Deferred_Reply) {

				/* this is the very first call, have mom copy files */
				/* first check the standard files: output & error   */

				preq = cpy_stdfile(preq, pjob, JOB_ATR_outpath);
				preq = cpy_stdfile(preq, pjob, JOB_ATR_errpath);

				/* are there any stage-out files ?		 	*/

				preq = cpy_stage(preq, pjob, JOB_ATR_stageout, STAGE_DIR_OUT);

				if (preq) {		/* have files to copy 		*/
					if (release_nodes_on_stageout) {
						if (free_sister_vnodes(pjob, NULL, log_buffer, LOG_BUF_SIZE, NULL) != 0) {
							log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_WARNING, pjob->ji_qs.ji_jobid, log_buffer);
						}
					}
					preq->rq_extra = (void *)pjob;
					rc = issue_Drequest(handle, preq, on_job_exit, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;	/* come back when mom replies */
					} else {
						/* set up as if mom returned error */

						preq->rq_reply.brp_code   = rc;
						preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;
						preq->rq_reply.brp_un.brp_txt.brp_txtlen = 0;
						/* we will "fall" into the post reply side */
					}

				} else {		/* no files to copy, any to delete? */
					(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
						JOB_SUBSTATE_STAGEDEL);
					ptask = set_task(WORK_Immed, 0, on_job_exit, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
					return;
				}
			}

			/* here we have a reply (maybe faked) from MOM about the copy */

			if (preq->rq_reply.brp_code != 0) {	/* error from MOM */

				if ((preq->rq_reply.brp_code == DIS_EOF) ||
					(preq->rq_reply.brp_code == DIS_EOD)) {
					/* connection to Mom broken */
					conn_to_mom_failed(pjob, on_job_exit);
					return;
				}

				if (preq->rq_reply.brp_code == PBSE_NOCOPYFILE)
					stageout_status = 0;

				on_exitrerun_msg(pjob, msg_obitnocpy);
				if (preq->rq_reply.brp_choice == BATCH_REPLY_CHOICE_Text) {
					int	len = strlen(log_buffer);

					if (len < LOG_BUF_SIZE+2) {
						log_buffer[len++] = '\n';
						strncpy(&log_buffer[len],
							preq->rq_reply.brp_un.brp_txt.brp_str,
							LOG_BUF_SIZE - len);
					}
				}
				svr_mailowner(pjob, MAIL_OTHER, MAIL_FORCE, log_buffer);
			}

			pjob->ji_wattr[(int)JOB_ATR_stageout_status].at_val.at_long = stageout_status;
			pjob->ji_wattr[(int)JOB_ATR_stageout_status].at_flags = ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;

			/*
			 * files (generally) copied ok, move on to the next phase by
			 * "faking" the immediate work task.
			 */

			free_br(preq);
			preq = 0;
			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_STAGEDEL);
			ptask->wt_type = WORK_Immed;

			/* NO BREAK - FALL INTO THE NEXT CASE */

		case JOB_SUBSTATE_STAGEDEL:

			if (ptask->wt_type != WORK_Deferred_Reply) { /* first time in */

				/* Build list of files which were staged-in so they can
				 * can be deleted.
				 */

				preq = cpy_stage(preq, pjob, JOB_ATR_stagein, 0);

				if (preq) {		/* have files to delete		*/

					/* change the request type from copy to delete  */

					if (preq->rq_type == PBS_BATCH_CopyFiles_Cred)
						preq->rq_type = PBS_BATCH_DelFiles_Cred;
					else
						preq->rq_type = PBS_BATCH_DelFiles;
					preq->rq_extra = (void *)pjob;

					rc = issue_Drequest(handle, preq, on_job_exit, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;	/* come back when mom replies */
					} else {
						/* set up as if mom returned error */

						preq->rq_reply.brp_code   = rc;
						preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;

						/* we will "fall" into the post reply side */
					}

				} else {		/* preq == 0, no files to delete   */

					(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
						JOB_SUBSTATE_EXITED);
					ptask = set_task(WORK_Immed, 0, on_job_exit, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
					return;
				}
			}

			/* After MOM replied (maybe faked) to Delete Files request */

			if (preq->rq_reply.brp_code != 0) {	/* an error occurred */

				if ((preq->rq_reply.brp_code == DIS_EOF) ||
					(preq->rq_reply.brp_code == DIS_EOD)) {
					/* tcp connection to Mom broken */
					conn_to_mom_failed(pjob, on_job_exit);
					return;
				}
				if (preq->rq_reply.brp_code == PBSE_TRYAGAIN) {
					/* Mom hasn't finished her post processing yet,
					 * send the delete request again later.
					 */
					t = pjob->ji_retryok++;
					t = time_now + (t * t);
					ptask = set_task(WORK_Timed, t, on_job_exit, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask
						);
					return;
				}

				on_exitrerun_msg(pjob, msg_obitnodel);
				if (preq->rq_reply.brp_choice == BATCH_REPLY_CHOICE_Text) {
					int	len = strlen(log_buffer);

					if (len < LOG_BUF_SIZE+2) {
						log_buffer[len++] = '\n';
						strncpy(&log_buffer[len],
							preq->rq_reply.brp_un.brp_txt.brp_str,
							LOG_BUF_SIZE - len);
					}
				}
				svr_mailowner(pjob, MAIL_OTHER, MAIL_FORCE, log_buffer);
			}
			free_br(preq);
			preq = 0;
			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_EXITED);

			ptask->wt_type = WORK_Immed;

			/* NO BREAK, FALL INTO NEXT CASE */


		case JOB_SUBSTATE_EXITED:

			if (ptask->wt_type != WORK_Deferred_Reply) { /* first time in */

				/* see if have any dependencys */

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

				/* tell mom to delete the job */

				preq = alloc_br(PBS_BATCH_DeleteJob);
				if (preq) {
					(void)strcpy(preq->rq_ind.rq_delete.rq_objname,
						pjob->ji_qs.ji_jobid);
					preq->rq_extra = (void *)pjob;
					rc = issue_Drequest(handle, preq, on_job_exit, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;	/* come back when mom replies */
					} else {
						/* set up as if mom returned error */

						preq->rq_reply.brp_code   = rc;
						preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;

						/* we will "fall" into the post reply side */
					}
				} else {
					log_err(-1, pjob->ji_qs.ji_jobid,
						"Unable to malloc memory for deletejob");
					return;
				}
			}

			/* Set job's exec_vnodes with current time for last_used_time. */
			set_last_used_time_node(pjob, 0);

			/* here we have a reply from MOM about the delete */
			/* if delete ok, send final track and purge the job */

			if (preq->rq_reply.brp_code == PBSE_SISCOMM) {

				/* some sister Mom apparently failed to delete the job and
				 * free resoures, keep job until discard_job() does its job
				 */
				free_br(preq);
				preq = NULL;
				if (handle != -1 && pjob->ji_mom_prot == PROT_TCP)
					svr_disconnect(handle);

				discard_job(pjob, "A sister Mom failed to delete job", 0);
				return;
			} else if ((preq->rq_reply.brp_code == DIS_EOF) ||
				(preq->rq_reply.brp_code == DIS_EOD)) {
				/* tcp connection to Mom broken */
				conn_to_mom_failed(pjob, on_job_exit);
				return;
			} else if (preq->rq_reply.brp_code == PBSE_TRYAGAIN) {
				/* Mom hasn't finished her post processing yet,
				 * send the delete request again later.
				 */
				t = pjob->ji_retryok++;
				t = time_now + (t * t);
				ptask = set_task(WORK_Timed, t, on_job_exit, pjob);
				append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask
					);
				return;
			} else {
				/* all went ok with the delete by Mom(s) */
				free_br(preq);
				preq = NULL;
				if (handle != -1 && pjob->ji_mom_prot == PROT_TCP)
					svr_disconnect(handle);
				rel_resc(pjob); /* free any resc assigned to the job */

				account_job_update(pjob, PBS_ACCT_LAST);
				account_jobend(pjob, pjob->ji_acctrec, PBS_ACCT_END);

				if (pjob->ji_acctrec)
					rec = pjob->ji_acctrec;
				else
					rec = "";

				if (server.sv_attr[(int)SRV_ATR_log_events].at_val.at_long &
					PBSEVENT_JOB_USAGE) {
					/* log events set to record usage */
					log_event(PBSEVENT_JOB_USAGE | PBSEVENT_JOB_USAGE,
						PBS_EVENTCLASS_JOB, LOG_INFO,
						pjob->ji_qs.ji_jobid, rec);
				} else {
					char *pc;

					/* no usage in log, truncate messge */

					if ((pc = strchr(rec, (int)' ')) != NULL)
						*pc = '\0';
					log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO,
						pjob->ji_qs.ji_jobid, rec);
				}

				if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0)
					issue_track(pjob);
				/*
				 * Check if the history of the finished job can be saved or it needs to be purged .
				 */
				svr_saveorpurge_finjobhist(pjob);
			}
			break;
		case JOB_SUBSTATE_TERMINATED:
			set_last_used_time_node(pjob, 0);

	}
}

/**
 * @brief
 *	Unset values of various attributes of 'pjob'
 *	specifically for node ramp down feature.
 *
 * @param[in]	pjob - job in question
 *
 * @return void
 *
 */
void
unset_extra_attributes(job *pjob)
{
	if (pjob == NULL)
		return;

	if (pjob->ji_wattr[(int) JOB_ATR_resource_orig].at_flags & ATR_VFLAG_SET) {
		job_attr_def[(int) JOB_ATR_resource].at_free( &pjob->ji_wattr[(int) JOB_ATR_resource]);
		pjob->ji_wattr[(int) JOB_ATR_resource].at_flags &= ~ATR_VFLAG_SET;
		job_attr_def[(int) JOB_ATR_resource].at_set( &pjob->ji_wattr[(int) JOB_ATR_resource], &pjob->ji_wattr[(int) JOB_ATR_resource_orig], INCR);

		job_attr_def[(int) JOB_ATR_resource_orig].at_free( &pjob->ji_wattr[(int) JOB_ATR_resource_orig]);
		pjob->ji_wattr[(int) JOB_ATR_resource_orig].at_flags &= ~ATR_VFLAG_SET;
		pjob->ji_modified = 1;
	}

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


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

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

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

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

	if (pjob->ji_wattr[JOB_ATR_SchedSelect_orig].at_flags & ATR_VFLAG_SET) {

		(void)decode_str(
		     &pjob->ji_wattr[(int)JOB_ATR_SchedSelect], NULL, NULL, pjob->ji_wattr[JOB_ATR_SchedSelect_orig].at_val.at_str);
		job_attr_def[(int) JOB_ATR_SchedSelect_orig].at_free( &pjob->ji_wattr[(int) JOB_ATR_SchedSelect_orig]);
		pjob->ji_wattr[(int) JOB_ATR_SchedSelect_orig].at_flags &= ~ATR_VFLAG_SET;
	}

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

/**
 * @brief
 * 		on_job_rerun - Handle the clean up of jobs being rerun.  This gets
 *		messy if the job is being executed on another host.  Then the
 *		"standard" files must be copied to the server for safe keeping.
 *
 *		The basic flow is very much like that of on_job_exit().
 *		The substate will already set to JOB_SUBSTATE_RERUN and the
 *		JOB_SVFLG_HASRUN bit set in ji_svrflags.
 *
 * @param[in,out]	ptask	- work task structure
 */
void
on_job_rerun(struct work_task *ptask)
{
	int		      handle;
	int		      newstate;
	int		      newsubst;
	job		     *pjob;
	struct batch_request *preq;
	struct work_task     *pt;
	int		      rc;
	pbs_list_head	*mom_tasklist_ptr = NULL;
	mominfo_t *pmom = 0;

	if (ptask->wt_type != WORK_Deferred_Reply) {
		preq = NULL;
		pjob = (job *)ptask->wt_parm1;
	} else {
		preq = (struct batch_request *)ptask->wt_parm1;
		pjob = (job *)preq->rq_extra;
	}

	/* minor check on validatity of pjob */

	if (isdigit((int)pjob->ji_qs.ji_jobid[0]) == 0)
		return;		/* not pointing to currently valid job */

	if ((handle = mom_comm(pjob, on_job_rerun)) < 0)
		return;

	if (pjob->ji_mom_prot == PROT_RPP) {
		pmom = tfind2((unsigned long) pjob->ji_qs.ji_un.ji_exect.ji_momaddr,
			pjob->ji_qs.ji_un.ji_exect.ji_momport,
			&ipaddrs);
		if (!pmom || (((mom_svrinfo_t *)(pmom->mi_data))->msr_state & INUSE_DOWN))
			return;
		mom_tasklist_ptr = &(((mom_svrinfo_t *)(pmom->mi_data))->msr_deferred_cmds);
	}

	switch (pjob->ji_qs.ji_substate) {


		case JOB_SUBSTATE_RERUN:

			if (ptask->wt_type != WORK_Deferred_Reply) {
				if (pjob->ji_qs.ji_un.ji_exect.ji_momaddr==pbs_server_addr) {

					/* files don`t need to be moved, go to next step */

					(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
						JOB_SUBSTATE_RERUN1);
					ptask = set_task(WORK_Immed, 0, on_job_rerun, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
					return;
				}

				/* here is where we have to save the files	*/
				/* ask mom to send them back to the server	*/
				/* mom deletes her copy if returned ok	*/

				preq = alloc_br(PBS_BATCH_Rerun);
				if (preq == NULL) {
					return;
				}
				(void)strcpy(preq->rq_ind.rq_rerun, pjob->ji_qs.ji_jobid);
				preq->rq_extra = (void *)pjob;

				rc = issue_Drequest(handle, preq, on_job_rerun, &pt, pjob->ji_mom_prot);
				if (rc == 0) {
					/* request ok, will come back when its done */
					append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
					if (pjob->ji_mom_prot == PROT_RPP)
						if (mom_tasklist_ptr)
							append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
					return;
				} else {
					/* set up as if mom returned error */

					preq->rq_reply.brp_code = rc;
					preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;
					/* we will "fall" into the post reply side */
				}


			}

			/* We get here if MOM replied (may be faked above)  */
			/* to the rerun (return files) request issued above */

			if (preq->rq_reply.brp_code != 0) {	/* error */
				/* for now, just log it */
				if ((preq->rq_reply.brp_code == DIS_EOF) ||
					(preq->rq_reply.brp_code == DIS_EOD)) {
					/* tcp connection to Mom broken */
					conn_to_mom_failed(pjob, on_job_rerun);
					return;
				}
				on_exitrerun_msg(pjob, msg_obitnocpy);
			}
			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_RERUN1);
			ptask->wt_type = WORK_Immed;
			free_br(preq);
			preq = 0;

			/* NO BREAK, FALL THROUGH TO NEXT CASE, including the request */


		case JOB_SUBSTATE_RERUN1:

			if (ptask->wt_type != WORK_Deferred_Reply) {

				/* this is the very first call, have mom copy files */
				/* are there any stage-out files to process? 	*/

				preq = cpy_stage(preq, pjob, JOB_ATR_stageout, STAGE_DIR_OUT);

				if (preq) {		/* have files to copy 		*/
					preq->rq_extra = (void *)pjob;
					rc =  issue_Drequest(handle, preq, on_job_rerun, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;	/* come back when mom replies */
					} else
						/* set up as if mom returned error */

						preq->rq_reply.brp_code   = rc;
					preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;
					preq->rq_reply.brp_un.brp_txt.brp_txtlen = 0;
					/* we will "fall" into the post reply side */

				} else {		/* no files to copy, any to delete? */
					(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
						JOB_SUBSTATE_RERUN2);
					ptask = set_task(WORK_Immed, 0, on_job_rerun, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
					return;
				}
			}

			/* here we have a reply (maybe faked) from MOM about the copy */

			if (preq->rq_reply.brp_code != 0) {	/* error from MOM */

				if ((preq->rq_reply.brp_code == DIS_EOF) ||
					(preq->rq_reply.brp_code == DIS_EOD)) {
					/* tcp connection to Mom broken */
					conn_to_mom_failed(pjob, on_job_rerun);
					return;
				}
				on_exitrerun_msg(pjob, msg_obitnocpy);
				if (preq->rq_reply.brp_choice == BATCH_REPLY_CHOICE_Text) {
					int	len = strlen(log_buffer);

					if (len < LOG_BUF_SIZE+2) {
						log_buffer[len++] = '\n';
						strncpy(&log_buffer[len],
							preq->rq_reply.brp_un.brp_txt.brp_str,
							LOG_BUF_SIZE - len);
					}
				}
				svr_mailowner(pjob, MAIL_OTHER, MAIL_FORCE, log_buffer);
			}

			/*
			 * files (generally) copied ok, move on to the next phase by
			 * "faking" the immediate work task.
			 */

			free_br(preq);
			preq = 0;
			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_RERUN2);
			ptask->wt_type = WORK_Immed;

			/* NO BREAK - FALL INTO THE NEXT CASE */

		case JOB_SUBSTATE_RERUN2:

			if (ptask->wt_type != WORK_Deferred_Reply) {

				/* here is where we delete  any stage-in files	   */

				preq = cpy_stage(preq, pjob, JOB_ATR_stagein,  0);
				if (preq) {
					preq->rq_type = PBS_BATCH_DelFiles;
					preq->rq_extra = (void *)pjob;
					rc =  issue_Drequest(handle, preq, on_job_rerun, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;
					} else {	/* error on sending request */
						preq->rq_reply.brp_code = rc;
						preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;
						/* we will "fall" into the post reply side */
					}
				} else {
					(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
						JOB_SUBSTATE_RERUN3);
					ptask = set_task(WORK_Immed, 0, on_job_rerun, pjob);
					append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
					return;
				}
			}

			/* post reply side for delete file request to MOM */
			if (preq->rq_reply.brp_code != 0) {	/* error */
				if ((preq->rq_reply.brp_code == DIS_EOF) ||
					(preq->rq_reply.brp_code == DIS_EOD)) {
					/* tcp connection to Mom broken */
					conn_to_mom_failed(pjob, on_job_rerun);
					return;
				}
				/* for other errors, just log it */
				on_exitrerun_msg(pjob, msg_obitnocpy);
			}
			free_br(preq);
			preq = 0;
			(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
				JOB_SUBSTATE_RERUN3);
			ptask->wt_type = WORK_Immed;

			/* NO BREAK, FALL THROUGH TO NEXT CASE */


		case JOB_SUBSTATE_RERUN3:

			if (ptask->wt_type != WORK_Deferred_Reply) {
				/* need to have MOM delete her copy of the job */
				preq = alloc_br(PBS_BATCH_DeleteJob);
				if (preq) {
					(void)strcpy(preq->rq_ind.rq_delete.rq_objname,
						pjob->ji_qs.ji_jobid);
					preq->rq_extra = (void *)pjob;
					rc = issue_Drequest(handle, preq, on_job_rerun, &pt, pjob->ji_mom_prot);
					if (rc == 0) {
						append_link(&pjob->ji_svrtask, &pt->wt_linkobj, pt);
						if (pjob->ji_mom_prot == PROT_RPP)
							if (mom_tasklist_ptr)
								append_link(mom_tasklist_ptr, &pt->wt_linkobj2, pt); /* if rpp, link to mom list as well */
						return;	/* come back when Mom replies */
					} else {
						/* set up as if mom returned error */
						preq->rq_reply.brp_code   = rc;
						preq->rq_reply.brp_choice = BATCH_REPLY_CHOICE_NULL;
						/* fall into next section */
					}
				} else {
					log_err(-1, pjob->ji_qs.ji_jobid,
						"Unable to malloc memory for rerun");
					return;
				}
			}

			/* here we have a reply from MOM about the delete */
			/* if delete ok, send final track and purge the job */

			if (preq->rq_reply.brp_code == PBSE_SISCOMM) {

				/*
				 * some sister Mom apparently failed to delete the job and
				 * free resoures, keep job until discard_job() does its job
				 */
				free_br(preq);
				preq = NULL;
				if (handle != -1 && pjob->ji_mom_prot == PROT_TCP)
					svr_disconnect(handle);

				discard_job(pjob, "A sister Mom failed to delete job", 0);
				return;
			} else if ((preq->rq_reply.brp_code == DIS_EOF) ||
				(preq->rq_reply.brp_code == DIS_EOD)) {
				/* tcp connection to Mom broken */
				conn_to_mom_failed(pjob, on_job_rerun);
				return;
			} else {
				/* all went ok with the delete by Mom(s) */
				free_br(preq);
				preq = NULL;
				if (handle != -1 && pjob->ji_mom_prot == PROT_TCP)
					svr_disconnect(handle);

				account_jobend(pjob, pjob->ji_acctrec, PBS_ACCT_RERUN);
				if (pjob->ji_acctrec) {
					free(pjob->ji_acctrec);	/* logged, so clear it */
					pjob->ji_acctrec = NULL;
				}
				if ((pjob->ji_wattr[(int) JOB_ATR_resc_released].at_flags & ATR_VFLAG_SET)) {
					/* If JOB_ATR_resc_released attribute is set and we are trying
					 * to rerun a job then we need to reassign resources first because
					 * when we suspend a job we don't decrement all of the resources.
					 * So we need to set partially released resources
					 * back again to release all other resources
					 */
					set_resc_assigned(pjob, 0, INCR);
					job_attr_def[(int) JOB_ATR_resc_released].at_free(
						&pjob->ji_wattr[(int) JOB_ATR_resc_released]);
					pjob->ji_wattr[(int) JOB_ATR_resc_released].at_flags &= ~ATR_VFLAG_SET;
					if (pjob->ji_wattr[(int) JOB_ATR_resc_released_list].at_flags & ATR_VFLAG_SET) {
						job_attr_def[(int) JOB_ATR_resc_released_list].at_free(
							&pjob->ji_wattr[(int) JOB_ATR_resc_released_list]);
						pjob->ji_wattr[(int) JOB_ATR_resc_released_list].at_flags &= ~ATR_VFLAG_SET;
					}

				}
				rel_resc(pjob);		/* free resc assigned to job */
				unset_extra_attributes(pjob);


				/* Now  if not a Sub Job, then re-queue the job */

				if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {
					/* for a sub job, just purge it */
					/* note: substate already JOB_SUBSTATE_RERUN3 */
					job_purge(pjob);
					return;
				}
				if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HOTSTART) == 0) {
					/* in case of server shutdown, don't clear exec_vnode */
					/* will use it on hotstart when next comes up	      */
					job_attr_def[(int)JOB_ATR_exec_vnode].at_free(
						&pjob->ji_wattr[(int)JOB_ATR_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_pset].at_free(
						&pjob->ji_wattr[(int)JOB_ATR_pset]);
				}
				pjob->ji_modified = 1;	/* force full job save */
				pjob->ji_momhandle = -1;
				pjob->ji_mom_prot = PROT_INVALID;
				/* job dir has no meaning for re-queued jobs, so unset it */
				job_attr_def[(int)JOB_ATR_jobdir].at_free(&pjob->
					ji_wattr[(int)JOB_ATR_jobdir]);

				pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_StagedIn;
				svr_evaljobstate(pjob, &newstate, &newsubst, 0);
				(void)svr_setjobstate(pjob, newstate, newsubst);
			}
	}
}
/**
 * @brief
 * 		setrerun	- job is to be retried on start failure or
 * 		job is rerunnable and should set for rerun
 *
 * @param[in]	pjob	- job which needs to be set for rerun.
 *
 * @return	exit code
 * @retval	0	- substate set to rerun.
 * @retval	1	- substate left as it
 */
static int
setrerun(job *pjob)
{
	if ((pjob->ji_qs.ji_un.ji_exect.ji_exitstat == JOB_EXEC_RETRY) ||
		(pjob->ji_wattr[(int)JOB_ATR_rerunable].at_val.at_long != 0)) {
		pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN;
		return 0;
	} else {
		svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE, msg_init_abt);
		return 1;
	}
}

/**
 * @brief
 *		Concatenate the resources used to the buffer provided.
 *
 * @param[in,out]buffer - pointer to buffer to add info to.  May grow/change due to pbs_strcat() (realloc)
 * @param[in,out]buffer_size - size of buffer - may increase through pbs_strcat()
 * @param[in]		delim - a pointer to the delimiter to use
 * @param[in]		pjob - job structure for additional info
 */
int
concat_rescused_to_buffer(char **buffer, int *buffer_size, svrattrl *patlist, char *delim, job *pjob)
{
	int val_len;

	if(buffer == NULL || buffer_size == NULL || patlist == NULL || delim == NULL)
		return 1;
	/*
	 * To calculate length of the string of the form "resources_used.<resource>=<value>".
	 * Additional length of 3 is required to accommodate the characters '.', '=' and '\n'.
	 */
	val_len = strlen(patlist->al_value);
	/* log to accounting_logs only if there's a value */
	if (val_len > 0) {
		if(pbs_strcat(buffer, buffer_size, delim) == NULL) {
			log_err(errno, __func__, "Failed to allocate memory.");
			return 1;
		}
		if(pbs_strcat(buffer, buffer_size, patlist->al_name) == NULL) {
			log_err(errno, __func__, "Failed to allocate memory.");
			return 1;
		}
		if (patlist->al_resc) {
			if(pbs_strcat(buffer, buffer_size, ".") == NULL) {
				log_err(errno, __func__, "Failed to allocate memory.");
				return 1;
			}
			if(pbs_strcat(buffer, buffer_size, patlist->al_resc) == NULL) {
				log_err(errno, __func__, "Failed to allocate memory.");
				return 1;
			}
		}
		if(pbs_strcat(buffer, buffer_size, "=") == NULL) {
			log_err(errno, __func__, "Failed to allocate memory.");
			return 1;
		}
		if ( (pjob != NULL) &&
			patlist->al_resc && (strcmp(patlist->al_resc,
							WALLTIME) == 0) ) {
			long	j, k;

			k = get_walltime(pjob, JOB_ATR_resc_used_acct);
			j = get_walltime(pjob, JOB_ATR_resc_used);
			if ((k >= 0) && (j >= k)) {
				char timebuf[TIMEBUF_SIZE] = {0};

				convert_duration_to_str(j-k, timebuf, TIMEBUF_SIZE);
				if(pbs_strcat(buffer, buffer_size, timebuf) == NULL) {
					log_err(errno, __func__,
						"Failed to allocate memory.");
					return 1;
				}
			} else {
				if(pbs_strcat(buffer, buffer_size,
						patlist->al_value) == NULL) {
					log_err(errno, __func__,
						"Failed to allocate memory.");
					return 1;
				}

			}
		} else if(pbs_strcat(buffer, buffer_size,
						patlist->al_value) == NULL) {
			log_err(errno, __func__, "Failed to allocate memory.");
			return 1;
		}
	}
	return 0;
}


/**
 * @brief
 *		Process the Job Obituary Notice (request) from MOM for a job which has.
 *		terminated.  The Obit contains the exit status and final resource
 *		usage for the job.
 * @par
 *		If the job cannot be found, the Server tells Mom to discard her copy.
 *		This may be the case if the job was forcefully deleted while Mom was
 *		down or the Server was restarted cold/clean discarding the jobs.
 *
 *		Depending on the state of the job:
 *		- Not RUNNING and not EXITING - tell Mom to discard the job.
 *		- Also not in substate _TERM - Mom wishes to restart the end of job
 *	  	 processing; likely because she hasn't heard from the Server.
 *		- If the "run count" in the obit does not match the Server's, Mom has
 *	 	 an old copy and she is told to discard it.
 * @par
 *		Normally, the Obit is received when the job is in substate RUNNING.
 *		The job is moved into that substate when Mom sends the session id of
 *		the job,  see stat_update().  However, it is possible that  the Obit
 *		is received before that and the job is in substate _PRERUN (or very
 *		unlikely _PROVISION).  If this is the case, call complete_running()
 *		to update the job to _RUNNING and write the "S" accounting record before
 *		we write the "E" record.
 * @par
 *		There are special job exit values (negative nunbers which cannot be
 *		actual exits status of the job).  These are typically because Mom
 *		could not complete starting the job or Mom is being restarted without
 *		the "-p" option.
 *		- JOB_EXEC_FAIL1: Mom could not start job, the standard out/err files
 *	  	were not created.
 *		- JOB_EXEC_FAIL2: Mom could not start the job, but had created the
 *	  	files so there is useful info in them.
 *		- JOB_EXEC_INITABT: Mom aborted the running job on her initialization.
 *		- JOB_EXEC_FAILUID: Mom aborted the job because of an invalid uid/gid.
 *		- JOB_EXEC_FAIL_PASSWORD: Mom aborted the job because she needed the
 *	  	user's password (Windows) and the password didn't work.
 *		- JOB_EXEC_RETRY: Mom couldn't start the job, but it might work later,
 *	  	so requeue it.
 *		- JOB_EXEC_BADRESRT: The job could not be started from the checkpoint
 *	  	restart file.
 *		- JOB_EXEC_INITRST: Mom aborted a checkpointed job which should be
 *	  	requeued for a later "restart".
 *		- JOB_EXEC_QUERST: The Epilogue told Mom to requeue the job which can
 *	  	be restarted from a checkpoint.
 *		- JOB_EXEC_RERUN: or JOB_EXEC_RERUN_SIS_FAIL: requeue the job if it is
 *	  	rerunable (not submitted with "-r n").
 *		- JOB_EXEC_FAILHOOK_RERUN: returned by a job rejected by a mom hook
 *	  	and the next action is to requeue/rerun the job.
 *		- JOB_EXEC_FAILHOOK_DELETE: returned by a job rejected by a mom hook
 *	  	and the next action is to just delete the job.
 *		- JOB_EXEC_HOOK_RERUN - returned by a job that ran a mom hook that
 *	  	instructed the server to requeue the job once reaching the end.
 *		- JOB_EXEC_HOOK_DELETE - returned by a job that ran a mom hook that
 *	  	instructed the server to delete the job  once reaching the end.
 *		- JOB_EXEC_HOOKERROR: returned by a job rejected by a mom hook
 *		due to an exception, or hook alarm was raised,
 *		and the next action is to requeue/rerun the job.
 * @par
 *		Otherwise record the accounting information to be recorded later in the
 *		processing.  Now, the job is moved into "exiting" processing or "rerun"
 *		porcessiong (qrerun) via a work-task entry invoking either on_job_exit()
 *		or on_job_rerun().
 *
 * param[in] - pruu - the structure containing the resource usage info
 * param[in] - stream - the RPP stream connecting to the Mom
 *		The Server will send back either a rejection or an acceptance
 *		of  the Obit.
 */

void
job_obit(struct resc_used_update *pruu, int stream)
{
	int		  alreadymailed = 0;
	char		 *acctbuf = NULL;
	int  		  acctbuf_size = 0;
	int		  dummy;
	int		  num;
	int		  exitstatus;
	int		  local_exitstatus = 0;
	char		 *mailbuf = NULL;
	int		  mailbuf_size = 0;
	int		  newstate;
	int		  newsubst;
	job		 *pjob;
	svrattrl	 *patlist;
	struct work_task *ptask;
	void		(*eojproc)();
	char		 *mailmsg = NULL;
	char		 *msg = NULL;

	time_now = time(0);

	DBPRT(("%s: Obit received for job %s status=%d hop=%d\n", __func__,
		pruu->ru_pjobid, pruu->ru_status, pruu->ru_hop))
	pjob = find_job(pruu->ru_pjobid);
	if (pjob == NULL) {		/* not found */
		DBPRT(("%s: job %s not found!\n", __func__, pruu->ru_pjobid))
		if ((server_init_type == RECOV_COLD) ||
			(server_init_type == RECOV_CREATE)) {
			(void)sprintf(log_buffer, msg_obitnojob, PBSE_CLEANEDOUT);
		} else if (is_job_array(pruu->ru_pjobid) == IS_ARRAY_Single) {
			(void)sprintf(log_buffer, "%s", msg_obitnotrun);
		} else {
			(void)sprintf(log_buffer, msg_obitnojob, PBSE_UNKJOBID);
		}
		log_event(PBSEVENT_ERROR|PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
			LOG_NOTICE, pruu->ru_pjobid, log_buffer);

		/* tell MOM the job was blown away */

		reject_obit(stream, pruu->ru_pjobid);

		/* free pruu, see mom_server.h */
		FREE_RUU(pruu)
		return;
	}
	sprintf(log_buffer, "Obit received momhop:%d serverhop:%ld state:%d substate:%d",
		pruu->ru_hop,
		pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long,
		pjob->ji_qs.ji_state,
		pjob->ji_qs.ji_substate);
	log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, LOG_INFO,
		pruu->ru_pjobid, log_buffer);

	if (pjob->ji_qs.ji_state != JOB_STATE_RUNNING) {
		DBPRT(("%s: job %s not in running state!\n",
			__func__, pruu->ru_pjobid))
		if (pjob->ji_qs.ji_state != JOB_STATE_EXITING) {

			/* not running and not exiting - bad news   */
			/* may be from old Mom and job was requeued */
			/* tell mom to trash job		    */
			DBPRT(("%s: job %s not in exiting state!\n",
				__func__, pruu->ru_pjobid))
			reject_obit(stream, pruu->ru_pjobid);

			(void)sprintf(log_buffer, "%s", msg_obitnotrun);
			log_event(PBSEVENT_ERROR|PBSEVENT_JOB,
				PBS_EVENTCLASS_JOB, LOG_INFO,
				pruu->ru_pjobid, log_buffer);
			FREE_RUU(pruu)
			return;
		} else if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_TERM) {
			/*
			 * not in special site script substate, Mom must have
			 * had a problem and wants to have the post job
			 * processing restarted.
			 */
			FREE_RUU(pruu)
			/*
			 * If there is an open connection to Mom for this job,
			 * find the associate work task, remove and free it and
			 * any outstanding batch_request to Mom.  Then close
			 * the connection so we start fresh and stay in sync.
			 */
			if (pjob->ji_momhandle != -1) {
				struct batch_request *prequest;
				extern pbs_list_head task_list_event;

				ptask = (struct work_task *)
					GET_NEXT(task_list_event);
				while (ptask) {
					if ((ptask->wt_type == WORK_Deferred_Reply) &&
						(ptask->wt_event == pjob->ji_momhandle))
						break;
					ptask = (struct work_task *)
						GET_NEXT(ptask->wt_linkall);
				}
				if (ptask) {
					if ((prequest = ptask->wt_parm1) != NULL)
						free_br(prequest);
					delete_task(ptask);
				}
				if (pjob->ji_mom_prot == PROT_TCP)
					svr_force_disconnect(pjob->ji_momhandle);

				pjob->ji_momhandle = -1;
				pjob->ji_mom_prot = PROT_INVALID;
			}
			if (pjob->ji_qs.ji_substate < JOB_SUBSTATE_RERUN)
				eojproc = on_job_exit;
			else
				eojproc = on_job_rerun;
			ptask = set_task(WORK_Immed, 0, eojproc, (void *)pjob);
			append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);
			return;
		}
		/* State EXITING and substate TERM, this is the real obit */
		/* so fall throught and start real end of job processing  */
	}

	if (pruu->ru_hop<pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long) {
		/* Obit is for an older run version,  likely a Mom coming back */
		/* alive after being down awhile and job was requeue and run */
		/* somewhere else.   Just tell Mom to junk job.		     */
		DBPRT(("%s: job %s run count too low\n",
			__func__, pruu->ru_pjobid))
		reject_obit(stream, pruu->ru_pjobid);
		FREE_RUU(pruu) /* free pruu, see mom_server.h */
		return;
	} else if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {

		/* Won't have a valid hop count in the job structure        */
		/* look at where job is running and who is sending the obit */

		int		 ivndx;
		mominfo_t 	*psendmom;
		struct pbsnode  *sendvnp;
		char		*runningnode;
		extern struct tree *streams;

		psendmom = tfind2(stream, 0, &streams);
		runningnode = parse_servername(pjob->ji_wattr[(int)JOB_ATR_exec_vnode].at_val.at_str, NULL);
		if (psendmom && runningnode) {
			for (ivndx = 0; ivndx < ((mom_svrinfo_t *)(psendmom->mi_data))->msr_numvnds; ++ivndx) {
				sendvnp = ((mom_svrinfo_t *)(psendmom->mi_data))->msr_children[ivndx];
				if (strcasecmp(runningnode, sendvnp->nd_name) == 0) {
					break;
				}
			}
			if (ivndx == ((mom_svrinfo_t *)(psendmom->mi_data))->msr_numvnds) {
				/* not the same node,  reject the obit */
				reject_obit(stream, pruu->ru_pjobid);
				FREE_RUU(pruu) /* free pruu, see mom_server.h */
				return;
			}
		}
	}

	/*
	 * have hit a race condition where the send_job child's process
	 * may not yet have been reaped.  Update accounting for job start
	 */

	if ((pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN) ||
		(pjob->ji_qs.ji_substate == JOB_SUBSTATE_PROVISION)) {

		DBPRT(("%s: job %s in prerun state.\n", __func__, pruu->ru_pjobid))
		complete_running(pjob);
	}
	if (pjob->ji_prunreq) {
		reply_ack(pjob->ji_prunreq);
		pjob->ji_prunreq = NULL;
	}



	/*
	 * save exit state, update the resources used
	 */

	exitstatus = pruu->ru_status;

	pjob->ji_qs.ji_un.ji_exect.ji_exitstat = exitstatus;

	/* set the Exit_status job attribute */

	if (pjob->ji_wattr[(int)JOB_ATR_exit_status].at_flags & ATR_VFLAG_SET)
	local_exitstatus = \
			pjob->ji_wattr[(int)JOB_ATR_exit_status].at_val.at_long;

	if (((local_exitstatus == JOB_EXEC_HOOK_RERUN) ||
		(local_exitstatus == JOB_EXEC_HOOK_DELETE)) &&
		(exitstatus != JOB_EXEC_FAILHOOK_RERUN)  &&
		(exitstatus != JOB_EXEC_FAILHOOK_DELETE)) {
		exitstatus = local_exitstatus;
	} else {

		pjob->ji_wattr[(int)JOB_ATR_exit_status].at_val.at_long = \
								exitstatus;
		pjob->ji_wattr[(int)JOB_ATR_exit_status].at_flags |=
			(ATR_VFLAG_SET | ATR_VFLAG_MODCACHE);
	}

	patlist = (svrattrl *)GET_NEXT(pruu->ru_attr);

	/* record usage attribute to job for history */
	dummy = 0;
	if (modify_job_attr(pjob, patlist,
		ATR_DFLAG_MGWR | ATR_DFLAG_SvWR, &dummy) != 0) {
		for (num=1; num<dummy; num++)
			patlist = (struct svrattrl *)GET_NEXT(patlist->al_link);
		sprintf(log_buffer, "unable to update attribute %s.%s in job_obit", patlist->al_name, patlist->al_resc);
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB,
			LOG_NOTICE, pjob->ji_qs.ji_jobid, log_buffer);
	}

	/* Allocate initial space for acctbuf/mailbuf.  Future space will be allocated by pbs_strcat(). */
	acctbuf = malloc(RESC_USED_BUF_SIZE);
	mailbuf = malloc(RESC_USED_BUF_SIZE);

	if (acctbuf == NULL || mailbuf == NULL) {
		log_err(errno, __func__, "Failed to allocate memory");
		/* Just incase one of the buffers got allocated */
		free(acctbuf);
		acctbuf = NULL;
		free(mailbuf);
		mailbuf = NULL;
	}
	else {
		acctbuf_size = RESC_USED_BUF_SIZE;
		mailbuf_size = RESC_USED_BUF_SIZE;

		(void) snprintf(acctbuf,acctbuf_size, msg_job_end_stat,
				pjob->ji_qs.ji_un.ji_exect.ji_exitstat);
		if (exitstatus < 10000) {
			 (void) strncpy(mailbuf, acctbuf, mailbuf_size);
		} else {
			(void) snprintf(mailbuf, mailbuf_size, msg_job_end_sig, exitstatus - 10000);
		}
		/*
		 * NOTE:
		 * Following code for constructing resources used information is same as account_jobend()
		 * with minor difference that to traverse patlist in this code
		 * we have to use GET_NEXT(patlist->al_link) since it is part of batch request
		 * and in account_jobend() we are using patlist->al_sister which is encoded
		 * information in job struct.
		 * This collects all resources_used information returned from the mom.
		 */
		for (; patlist; patlist = (svrattrl *) GET_NEXT(patlist->al_link)) {

			resource_def *tmpdef;

			if (strcmp(patlist->al_name, ATTR_used_update) == 0)
				continue;

			tmpdef = find_resc_def(svr_resc_def, patlist->al_resc, svr_resc_size);

			if (tmpdef == NULL)
				continue;

			/*
			 * Copy all resources to the accounting buffer.
			 * Copy all but invisible resources into the mail buffer.
			 * The ATR_DFLAG_USRD flag will not be set on invisible resources.
			 */
			if (concat_rescused_to_buffer(&acctbuf, &acctbuf_size, patlist, " ", pjob) != 0)
				break;

			if (tmpdef->rs_flags & ATR_DFLAG_USRD) {
				if (concat_rescused_to_buffer(&mailbuf, &mailbuf_size, patlist, "\n", pjob) != 0)
					break;
			}
		}
	}


	/* make sure ji_momhandle is -1 to force new connection to mom */

	pjob->ji_momhandle = -1;
	pjob->ji_mom_prot = PROT_INVALID;
	pjob->ji_retryok = 0;	/* for retry if Mom down */

	/* clear suspended flag if it was set */
	/* clear suspended-workstation busy flag if set */

	pjob->ji_qs.ji_svrflags &= ~(JOB_SVFLG_Suspend | JOB_SVFLG_Actsuspd);

	/* Was there a special exit status from MOM ? */

	if ((exitstatus < 0) && (exitstatus != JOB_EXEC_CHKP)) {
		/* negative exit status is special */

		switch (exitstatus) {
			case JOB_EXEC_FAILHOOK_DELETE:
				/* this is a reject */
				log_event(PBSEVENT_ERROR|PBSEVENT_JOB,
					PBS_EVENTCLASS_JOB, LOG_INFO,
					pjob->ji_qs.ji_jobid,
					msg_hook_reject_deletejob);
				DBPRT(("%s: MOM rejected job %s due to a hook.\n",
					__func__, pruu->ru_pjobid))
				svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE,
					msg_hook_reject_deletejob);
				alreadymailed = 1;
				break;

			case JOB_EXEC_HOOK_DELETE:
				/* more likely an accept with a hook delete option */
				log_event(PBSEVENT_ADMIN|PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
					LOG_INFO, pjob->ji_qs.ji_jobid,
					"a hook requested for job to be deleted");
				DBPRT(("%s: a hook requested for job %s to be deleted.\n",
					__func__, pruu->ru_pjobid))
				svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE,
					"a hook requested for job to be deleted");
				alreadymailed = 1;
				break;

			case JOB_EXEC_FAIL1:
			default:

				/* MOM rejected job with fatal error, abort job */

				DBPRT(("%s: MOM rejected job %s with fatal error.\n",
					__func__, pruu->ru_pjobid))
				svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE, msg_momnoexec1);
				alreadymailed = 1;
				break;

			case JOB_EXEC_FAIL2:

				/* MOM reject job after files setup, abort job */

				DBPRT(("%s: MOM rejected job %s after setup.\n",
					__func__, pruu->ru_pjobid))
				svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE, msg_momnoexec2);
				alreadymailed = 1;
				break;

			case JOB_EXEC_INITABT:

				/* MOM aborted job on her initialization */

				DBPRT(("%s: MOM aborted job %s on init, no requeue.\n",
					__func__, pruu->ru_pjobid))
				alreadymailed = setrerun(pjob);
				pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HASRUN;
				break;

			case JOB_EXEC_FAILUID:

				/* MOM abort job because uid or gid was invalid */

				DBPRT(("%s: MOM rejected job %s with invaild uid/gid.\n",
					__func__, pruu->ru_pjobid))
				svr_mailowner(pjob, MAIL_ABORT, MAIL_FORCE, msg_baduser);
				alreadymailed = 1;
				/* go to the retry case */
				goto RetryJob;

			case JOB_EXEC_FAIL_PASSWORD:

				/* put job on password hold */
				pjob->ji_wattr[(int)JOB_ATR_hold].at_val.at_long |= HOLD_bad_password;
				pjob->ji_wattr[(int)JOB_ATR_hold].at_flags |= ATR_VFLAG_SET | ATR_VFLAG_MODCACHE;

				pjob->ji_qs.ji_substate = JOB_SUBSTATE_HELD;
				pjob->ji_modified = 1;
				svr_evaljobstate(pjob, &newstate, &newsubst, 0);
				(void)svr_setjobstate(pjob, newstate, newsubst);

				msg = (pruu->ru_comment ? pruu->ru_comment:"");
				mailmsg = (char *)malloc( strlen(msg) + 1 + \
							  strlen(msg_bad_password) + 1 );
				if (mailmsg) {
					sprintf(mailmsg, "%s:%s", msg, msg_bad_password);

					svr_mailowner(pjob, MAIL_BEGIN, MAIL_FORCE,
						mailmsg);
					job_attr_def[(int)JOB_ATR_Comment].at_decode(&pjob->ji_wattr[(int)JOB_ATR_Comment], NULL, NULL, mailmsg);

					log_event(PBSEVENT_ERROR|PBSEVENT_JOB,
						PBS_EVENTCLASS_JOB, LOG_INFO,
						pjob->ji_qs.ji_jobid, mailmsg);
					(void)free(mailmsg);
				} else {
					svr_mailowner(pjob, MAIL_BEGIN, MAIL_FORCE,
						msg_bad_password);
					job_attr_def[(int)JOB_ATR_Comment].at_decode(&pjob->ji_wattr[(int)JOB_ATR_Comment], NULL, NULL, msg_bad_password);
				}

			case JOB_EXEC_RETRY:
			case JOB_EXEC_FAILHOOK_RERUN:
			case JOB_EXEC_HOOK_RERUN:
			case JOB_EXEC_HOOKERROR:
				if ((exitstatus == JOB_EXEC_FAILHOOK_RERUN) ||
					   (exitstatus == JOB_EXEC_HOOKERROR)) {
					log_event(PBSEVENT_ERROR|PBSEVENT_JOB,
						PBS_EVENTCLASS_JOB, LOG_INFO,
						pjob->ji_qs.ji_jobid,
						msg_hook_reject_rerunjob);
					DBPRT(("%s: MOM rejected job %s due to a hook.\n",
						__func__, pruu->ru_pjobid))
				}
RetryJob:
				/* MOM rejected job, but said retry it */

				DBPRT(("%s: MOM rejected job %s but will retry.\n",
					__func__, pruu->ru_pjobid))
				if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_HASRUN) {
					/* has run before, treat this as another rerun */
					alreadymailed = setrerun(pjob);
				} else {
					/* have mom remove job files, not saving them,	*/
					/* and requeue job				*/
					pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN2;
				}
				if (pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long >
#ifdef NAS /* localmod 083 */
					PBS_MAX_HOPCOUNT
#else
					PBS_MAX_HOPCOUNT + PBS_MAX_HOPCOUNT
#endif /* localmod 083 */
				) {
					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;
					job_attr_def[(int)JOB_ATR_Comment].at_decode(
						&pjob->ji_wattr[(int)JOB_ATR_Comment],
						NULL, NULL,
						"job held, too many failed attempts to run");
				}
				break;

			case JOB_EXEC_BADRESRT:

				/* MOM could not restart job, setup for rerun */

				DBPRT(("%s: MOM could not restart job %s, will rerun.\n",
					__func__, pruu->ru_pjobid))
				alreadymailed = setrerun(pjob);
				pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHKPT;
				break;

			case JOB_EXEC_INITRST:
				/* Mom aborted job on Mom being restarted, job has been  */
				/* checkpointed and can be "restarted" rather than rerun */
			case JOB_EXEC_QUERST:
				/* Epilogue requested requeue of a checkpointed job    */
				/* it can be restarted later from restart file         */

				/* In both cases, job has checkpoint/restart file,     */
				/* requeue job and leave all information on execution  */
				/* host for a later restart                            */

				DBPRT(("%s: MOM request requeue of job for restart.\n", __func__))
				if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob)
					goto RetryJob;

				rel_resc(pjob);
				ack_obit(stream, pjob->ji_qs.ji_jobid);
				pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HASRUN|JOB_SVFLG_CHKPT;

					svr_evaljobstate(pjob, &newstate, &newsubst, 1);
					(void)svr_setjobstate(pjob, newstate, newsubst);
					if (pjob->ji_mom_prot == PROT_TCP)
						svr_disconnect(pjob->ji_momhandle);

					pjob->ji_momhandle = -1;
					pjob->ji_mom_prot = PROT_INVALID;

				FREE_RUU(pruu)
				free(mailbuf);
				return;

			case JOB_EXEC_INITRMG:

				/* MOM abort job on init, job has migratable checkpoint */
				/* Must recover output and checkpoint file, do eoj	    */

				DBPRT(("%s: MOM aborted migratable job %s on init,"
					" will requeue.\n", __func__, pruu->ru_pjobid))

				if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob)
					goto RetryJob;

				alreadymailed = setrerun(pjob);
				pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HASRUN |
					JOB_SVFLG_ChkptMig;
				break;

				case JOB_EXEC_RERUN:
				case JOB_EXEC_RERUN_SIS_FAIL:
					if (pjob->ji_wattr[(int)JOB_ATR_rerunable].at_val.at_long) {
						pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN;
					} else {
						pjob->ji_qs.ji_substate = JOB_SUBSTATE_EXITING;
						svr_mailowner(pjob, MAIL_ABORT, MAIL_NORMAL,
							"Non-rerunable job deleted on requeue");
					}
					break;
				case JOB_EXEC_FAIL_SECURITY:
					/* MOM rejected job with security breach fatal error, abort job */
					DBPRT(("%s: MOM rejected job %s with security breach fatal error.\n",
						__func__, pruu->ru_pjobid))
					pjob->ji_modified = 1;
					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 | ATR_VFLAG_MODIFY;
					job_attr_def[(int)JOB_ATR_Comment].at_decode(&pjob->ji_wattr[(int)JOB_ATR_Comment],NULL,
						NULL,"job held due to possible security breach of job tmpdir, failed to start");
					rel_resc(pjob);
					ack_obit(stream, pjob->ji_qs.ji_jobid);
					svr_setjobstate(pjob, JOB_STATE_HELD, JOB_SUBSTATE_HELD);
					FREE_RUU(pruu)
					free(mailbuf);
					return;
		}
	}

	/* can now free the resc_used_update structure */

	FREE_RUU(pruu)

	/* Set the following variable to make full save of the job. This is useful to retrieve the
	 * attributes of the job if in case pbs_server is restarted during the end of job processing.
	 */
	pjob->ji_modified = 1;

	/* Send email if exiting (not rerun) */

	if ((exitstatus == JOB_EXEC_FAILHOOK_DELETE) ||
		(exitstatus == JOB_EXEC_HOOK_DELETE) ||
		((pjob->ji_qs.ji_substate != JOB_SUBSTATE_RERUN) &&
		(pjob->ji_qs.ji_substate != JOB_SUBSTATE_RERUN2))) {

		DBPRT(("%s: Job %s is terminating and not rerun.\n",
			__func__, pjob->ji_qs.ji_jobid))

		(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
			JOB_SUBSTATE_EXITING);
		if (alreadymailed == 0 && mailbuf != NULL)
			svr_mailowner(pjob, MAIL_END, MAIL_NORMAL, mailbuf);

	}
        /* can free this now since no need to use it */
	free(mailbuf);

	/* save record accounting for later */

	free(pjob->ji_acctrec);
	pjob->ji_acctrec = acctbuf;

	/* Now, what do we do with the job... */

	if ((exitstatus == JOB_EXEC_FAILHOOK_DELETE) ||
		(exitstatus == JOB_EXEC_HOOK_DELETE) ||
		((pjob->ji_qs.ji_substate != JOB_SUBSTATE_RERUN) &&
		(pjob->ji_qs.ji_substate != JOB_SUBSTATE_RERUN2))) {

		if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) &&
			((pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) == 0) &&
			(pjob->ji_qs.ji_svrflags & JOB_SVFLG_HASHOLD)) {

			/* non-migratable checkpoint, leave there
			 * and just requeue the job.
			 */

			rel_resc(pjob);
			ack_obit(stream, pjob->ji_qs.ji_jobid);
			pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HASRUN;
			pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_HASHOLD;
			svr_evaljobstate(pjob, &newstate, &newsubst, 1);
			(void)svr_setjobstate(pjob, newstate, newsubst);
			if (pjob->ji_mom_prot == PROT_TCP)
				svr_disconnect(pjob->ji_momhandle);

			pjob->ji_momhandle = -1;
			pjob->ji_mom_prot = PROT_INVALID;
			return;
		}

		check_block(pjob, "");		/* if block set, send word */

		ptask = set_task(WORK_Immed, 0, on_job_exit, (void *)pjob);
		append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);

		/* "on_job_exit()" will be dispatched out of the main loop */

	} else {

		/*
		 * Rerunning job ...
		 * If not checkpointed, clear "resources_used"
		 * Requeue job
		 */
		DBPRT(("%s: Rerunning job %s\n", __func__, pjob->ji_qs.ji_jobid))

		if ((pjob->ji_qs.ji_svrflags &
			(JOB_SVFLG_CHKPT | JOB_SVFLG_ChkptMig)) == 0) {
			job_attr_def[(int)JOB_ATR_resc_used].at_free(
				&pjob->ji_wattr[(int)JOB_ATR_resc_used]);
		}

		(void)svr_setjobstate(pjob, JOB_STATE_EXITING,
			pjob->ji_qs.ji_substate);

		ptask = set_task(WORK_Immed, 0, on_job_rerun, (void *)pjob);
		append_link(&pjob->ji_svrtask, &ptask->wt_linkobj, ptask);

		/* "on_job_rerun()" will be dispatched out of the main loop */
	}

	/* ok,  if this is a sub job, we ack the obit to keep Mom happy  */
	/* otherwise, we might hear from her again, since subjobs are    */
	/* short and there might be several running there                */

	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob)
		ack_obit(stream, pjob->ji_qs.ji_jobid);

	DBPRT(("%s: Returning from end of function.\n", __func__))
	return;
}
