/*
 * 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.
 *
 */
#include <pbs_config.h>   /* the master config generated by configure */
#ifdef PYTHON
#include <Python.h>
#endif

#ifdef WIN32
#include <direct.h>
#else
#include <unistd.h>
#include <dirent.h>
#include <pwd.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
#include <assert.h>
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include "dis.h"
#include "libpbs.h"
#include "list_link.h"
#include "server_limits.h"
#include "attribute.h"
#include "resource.h"
#include "job.h"
#include "log.h"
#include "work_task.h"
#include "credential.h"
#include "batch_request.h"
#include "net_connect.h"
#include "pbs_nodes.h"
#include "svrfunc.h"
#include "mom_mach.h"
#include "mom_func.h"
#include "mom_server.h"
#include "mom_vnode.h"
#include "pbs_error.h"
#include "rpp.h"
#include "mom_hook_func.h"
#include "placementsets.h"
#include "hook.h"
/**
 * @file	catch_child.c
 */
/* External Functions */
void			(*free_job_CPUs)(job *) = NULL;

/* External Globals */

extern  char            mom_host[];
extern char		*path_epilog;
extern char		*path_jobs;
extern unsigned int	default_server_port;
extern pbs_list_head	svr_alljobs;
extern int		exiting_tasks;
extern char		*msg_daemonname;
extern int		svr_hook_resend_job_attrs;
#ifdef	WIN32
extern char		*mom_home;
#endif
#ifndef	WIN32
extern int		termin_child;
#endif
extern int		resc_access_perm;
extern int		server_stream;
extern time_t		time_now;
extern pbs_list_head	mom_polljobs;
extern unsigned int	pbs_mom_port;
#if MOM_ALPS
extern useconds_t	alps_release_wait_time;
extern int		alps_release_timeout;
extern useconds_t	alps_release_jitter;
#endif

extern char		*path_hooks_workdir;

#ifndef WIN32
/**
 * @brief
 *	catch_child() - the signal handler for  SIGCHLD.
 *
 * @param[in] sig - signal number
 *
 * @par To keep the signal handler simple for
 *	SIGCHLD  - just indicate there was one.
 *
 * @return Void
 *
 */
void
catch_child(int sig)
{
	termin_child = 1;
}
#endif

/**
 * @brief
 *	returns execution node info for job pjob
 *
 * @param[in] pjob - job pointer to job
 * @param[in] nodeid - nodeid on which  job execs
 *
 * @return hnodent *
 * @retval  hostdetails  SUCCESS
 * @retval  NULL  	 Failure
 *
 */
hnodent	*
get_node(job *pjob, tm_node_id nodeid)
{
	int		i;
	vmpiprocs	*vp = pjob->ji_vnods;

	for (i=0; i<pjob->ji_numvnod; i++, vp++) {
		if (vp->vn_node == nodeid)
			return vp->vn_host;
	}
	return NULL;
}

/**
 * @brief
 *	Restart each task which has exited and has TI_FLAGS_CHKPT turned on.
 *	If all tasks have been restarted, turn off MOM_CHKPT_POST.
 *
 * @param[in] pjob - pointer to job structure
 *
 * @return Void
 *
 */
void
chkpt_partial(job *pjob)
{
	int		i;
	char		namebuf[MAXPATHLEN+1];
	char		*filnam;
	pbs_task		*ptask;
	int		texit = 0;
	extern	char	task_fmt[];
	extern	char	*path_checkpoint;

	assert(pjob != NULL);

	strcpy(namebuf, path_checkpoint);
	if (*pjob->ji_qs.ji_fileprefix != '\0')
		strcat(namebuf, pjob->ji_qs.ji_fileprefix);
	else
		strcat(namebuf, pjob->ji_qs.ji_jobid);
	strcat(namebuf, JOB_CKPT_SUFFIX);

	i = strlen(namebuf);
	filnam = &namebuf[i];

	pjob->ji_sampletim = 0; /* reset sampletime for cpupercent */

	for (ptask = (pbs_task *)GET_NEXT(pjob->ji_tasks);
		ptask != NULL;
		ptask = (pbs_task *)GET_NEXT(ptask->ti_jobtask)) {
		/*
		 ** See if the task was marked as one of those that did
		 ** actually checkpoint.
		 */
		if ((ptask->ti_flags & TI_FLAGS_CHKPT) == 0)
			continue;
		texit++;
		/*
		 ** Now see if it was reaped.  We don't want to
		 ** fool with it until we see it die.
		 */
		if (ptask->ti_qs.ti_status != TI_STATE_EXITED)
			continue;
		texit--;

		sprintf(filnam, task_fmt, ptask->ti_qs.ti_task);

		/*
		 **	Try action script with no post function.
		 */
		i = do_mom_action_script(RestartAction, pjob, ptask,
			namebuf, NULL);
		if (i != 0) {	/* script failed */
			/* if there is no script, try native support */
			if (i == -2)
				i = mach_restart(ptask, namebuf);
			if (i != 0)	/* everything failed */
				goto fail;
		}

		ptask->ti_qs.ti_status = TI_STATE_RUNNING;
		/*
		 ** Turn off TI_FLAGS_CHKPT if TI_FLAGS_SAVECKP is off.
		 ** Turn off TI_FLAGS_SAVECKP if it is on.
		 */
		if ((ptask->ti_flags & TI_FLAGS_SAVECKP) == 0)
			ptask->ti_flags &= ~TI_FLAGS_CHKPT;
		else
			ptask->ti_flags &= ~TI_FLAGS_SAVECKP;
		(void)task_save(ptask);
	}

	if (texit == 0) {
		char	oldname[MAXPATHLEN];
		struct	stat	statbuf;

		/*
		 ** All tasks should now be running.
		 ** Turn off MOM_CHKPT_POST and MOM_CHKPT_ACTIVE flags.
		 ** Job is back to where it was before the bad checkpoint
		 ** attempt.
		 */
		pjob->ji_flags &= ~MOM_CHKPT_POST;
		pjob->ji_flags &= ~MOM_CHKPT_ACTIVE;
		/*
		 ** Get rid of incomplete checkpoint directory and
		 ** move old chkpt dir back to regular if it exists.
		 */
		*filnam = '\0';
		(void)remtree(namebuf);
		strcpy(oldname, namebuf);
		strcat(oldname, ".old");
		if (stat(oldname, &statbuf) == 0) {
			if (rename(oldname, namebuf) == -1)
				pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHKPT;
		}
	}
	return;

fail:
	/*
	 ** If we cannot restart a task from a partially failed checkpoint,
	 ** the job will be killed.
	 */
	log_joberr(errno, __func__, "failed to restart", pjob->ji_qs.ji_jobid);
	pjob->ji_flags &= ~MOM_CHKPT_POST;
	(void)kill_job(pjob, SIGKILL);
	return;
}

/*
 * the following is a list of attributes to be returned to the server
 * for a newly executing job.   They are returned only if they have
 * been modified by Mom.  Note that JOB_ATR_session_id and JOB_ATR_resc_used
 * are always returned; see encode_used(), update_ajob_status_using_cmd(),
 * and update_jobs_status().
 */

static enum job_atr mom_rtn_list[] = {
	JOB_ATR_errpath,
	JOB_ATR_outpath,
	JOB_ATR_altid,
	JOB_ATR_acct_id,
	JOB_ATR_jobdir,
	JOB_ATR_exectime,
	JOB_ATR_hold,
	JOB_ATR_variables,
	JOB_ATR_runcount,
	JOB_ATR_LAST
};

static enum job_atr mom_rtn_list_ext[] = {
	JOB_ATR_exec_vnode,
	JOB_ATR_SchedSelect,
	JOB_ATR_LAST
};

#ifdef PYTHON
/**
 * @brief
 * 	Returns the Python dictionary representation of a string
 *	specyfing a JSON object.
 *
 * @param[in]		value	-	string of JSON-object format
 * @param[out]		msg 	-	error message buffer
 * @param[in]		msg_len -	size of 'msg' buffer
 *
 * @return	PyObject *
 * @retval	<python  object> -	dictionary representation of 'value'
 * @retval	NULL		 -	if not successful, filling out 'msg'
 *					with the actual error message.
 */
PyObject *
json_loads(char *value, char *msg, size_t msg_len)
{
	PyObject	*py_name = NULL;
	PyObject	*py_module = NULL;
	PyObject 	*py_dict = NULL;
	PyObject	*py_func_loads = NULL;
	PyObject	*py_value = NULL;
	PyObject	*py_result = NULL;

	if (value == NULL) {
		return NULL;
	}

	if (msg != NULL) {
		if (msg_len <= 0) {
			return NULL;
		}
		msg[0] = '\0';
	}

	py_name = PyString_FromString("json");
	if (py_name == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "failed to construct json name");
		}
		return NULL;
	}

	py_module = PyImport_Import(py_name);
	if (py_module == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "failed to import json");
		}
		goto json_loads_fail;
	}

	py_dict = PyModule_GetDict(py_module);
	if (py_dict == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"failed to get json module dictionary");
		}
		goto json_loads_fail;
	}

	py_func_loads = PyDict_GetItemString(py_dict, (char*)"loads");
	if ((py_func_loads == NULL) || !PyCallable_Check(py_func_loads)) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"did not find json.loads() function");
		}
		goto json_loads_fail;
	}

	py_value = Py_BuildValue("(z)", (char*)value);
	if (py_value == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"failed to build python arg %s", value);
		}
		goto json_loads_fail;
	}

	PyErr_Clear(); /* clear any exception */
	py_result = PyObject_CallObject(py_func_loads, py_value);

	if (PyErr_Occurred()) {
		if (msg != NULL) {
			PyObject *exc_string = NULL;
			PyObject *exc_type = NULL;
			PyObject *exc_value = NULL;
			PyObject *exc_traceback = NULL;

			PyErr_Fetch(&exc_type, &exc_value, &exc_traceback);

			/* get the exception */
			if ((exc_type != NULL) &&
				((exc_string = PyObject_Str(exc_type)) != NULL) &&
				(PyString_Check(exc_string))) {
				snprintf(msg, msg_len, "%s",
					PyString_AsString(exc_string));
			}
			Py_XDECREF(exc_string);
			Py_XDECREF(exc_type);
			Py_XDECREF(exc_value);
#if !defined(WIN32)
			Py_XDECREF(exc_traceback);
#elif !defined(_DEBUG)
			/* for some reason this crashes on Windows Debug */
			Py_XDECREF(exc_traceback);
#endif
		}
		goto json_loads_fail;
	} else if (!PyDict_Check(py_result)) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "value is not a dictionary");
		}
		goto json_loads_fail;

	}

	Py_XDECREF(py_name);
	Py_XDECREF(py_module);
	Py_XDECREF(py_value);
	return (py_result);

json_loads_fail:
	Py_XDECREF(py_name);
	Py_XDECREF(py_module);
	Py_XDECREF(py_value);
	Py_XDECREF(py_result);
	return NULL;

}

/**
 * @brief
 * 	Returns a JSON-formatted string representing the Python object 'py_val'.
 *
 * @param[in]		py_val	-	Python object
 * @param[out]		msg 	-	error message buffer
 * @param[in]		msg_len -	size of 'msg' buffer
 *
 * @return	char *
 * @retval	<string> 	-	the returned JSON-formatted string
 * @retval	NULL		-	if not successful, filling out 'msg'
 *					with the actual error message.
 * @note
 *		The returned string is malloced space that must be freed
 *		later when no longer needed.
 */
char *
json_dumps(PyObject *py_val, char *msg, size_t msg_len)
{
	PyObject	*py_name = NULL;
	PyObject	*py_module = NULL;
	PyObject 	*py_dict = NULL;
	PyObject	*py_func_dumps = NULL;
	PyObject	*py_value = NULL;
	PyObject	*py_result = NULL;
	char		*tmp_str = NULL;
	char		*ret_string = NULL;
	int		slen;

	if (py_val == NULL) {
		return NULL;
	}

	if (msg != NULL) {
		if (msg_len <= 0) {
			return NULL;
		}
		msg[0] = '\0';
	}

	py_name = PyString_FromString("json");
	if (py_name == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "failed to construct json name");
		}
		return NULL;
	}

	py_module = PyImport_Import(py_name);
	if (py_module == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "failed to import json");
		}
		goto json_dumps_fail;
	}

	py_dict = PyModule_GetDict(py_module);
	if (py_dict == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"failed to get json module dictionary");
		}
		goto json_dumps_fail;
	}

	py_func_dumps = PyDict_GetItemString(py_dict, (char*)"dumps");
	if ((py_func_dumps == NULL) || !PyCallable_Check(py_func_dumps)) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"did not find json.loads() function");
		}
		goto json_dumps_fail;
	}

	py_value = Py_BuildValue("(O)", py_val);
	if (py_value == NULL) {
		if (msg != NULL) {
			snprintf(msg, msg_len,
				"failed to build python arg %p", (void *)py_val);
		}
		goto json_dumps_fail;
	}
	PyErr_Clear(); /* clear any exception */
	py_result = PyObject_CallObject(py_func_dumps, py_value);

	if (PyErr_Occurred()) {
		if (msg != NULL) {
			PyObject *exc_string = NULL;
			PyObject *exc_type = NULL;
			PyObject *exc_value = NULL;
			PyObject *exc_traceback = NULL;

			PyErr_Fetch(&exc_type, &exc_value, &exc_traceback);

			/* get the exception */
			if ((exc_type != NULL) &&
				((exc_string = PyObject_Str(exc_type)) != NULL) &&
				(PyString_Check(exc_string))) {
				snprintf(msg, msg_len, "%s",
					PyString_AsString(exc_string));
			}
			Py_XDECREF(exc_string);
			Py_XDECREF(exc_type);
			Py_XDECREF(exc_value);
#if !defined(WIN32)
			Py_XDECREF(exc_traceback);
#elif !defined(_DEBUG)
			/* for some reason this crashes on Windows Debug */
			Py_XDECREF(exc_traceback);
#endif
		}
		goto json_dumps_fail;
	} else if (!PyString_Check(py_result)) {
		if (msg != NULL) {
			snprintf(msg, msg_len, "value is not a string");
		}
		goto json_dumps_fail;

	}

	Py_XDECREF(py_name);
	Py_XDECREF(py_module);

	tmp_str = PyString_AsString(py_result);
	/* returned tmp_str points to an internal buffer of 'py_result' */
	if (tmp_str == NULL) {
		snprintf(msg, msg_len, "PyString_AsString failed");
		Py_XDECREF(py_result);
		return NULL;
	}
	slen = strlen(tmp_str) + 3; /* for null character + 2 single quotes */
	ret_string = (char *)malloc(slen);
	if (ret_string == NULL) {
		snprintf(msg, msg_len, "malloc of ret_string failed");
		Py_XDECREF(py_result);
		return NULL;
	}
	snprintf(ret_string, slen, "'%s'", tmp_str);
	Py_XDECREF(py_result);
	return (ret_string);

json_dumps_fail:
	Py_XDECREF(py_name);
	Py_XDECREF(py_module);
	Py_XDECREF(py_result);
	return NULL;

}
#endif

/**
 * @brief
 * 	 encode_used - encode resources used by a job to be returned to the server
 *
 * @param[in] pjob - pointer to job structure
 * @param[in] phead - pointer to pbs_list_head structure
 *
 * @return Void
 *
 */
static void
encode_used(job *pjob, pbs_list_head *phead)
{
	unsigned long	 lnum;
	unsigned long	 lnum3;
	int		 i;
	attribute	*at;
	attribute	*at2;
	attribute_def	*ad;
	int		 rc;
	resource_def	*rd = NULL;
	resource_def	*rd2;
	resource	*rs;
	resource	*rs2;
	attribute	 val;	/* holds the final accumulated resources_used values from Moms including those released from the job */
	attribute	 val2;	/* temp variable for accumulating resources_used from sis Moms */
	attribute	 val3;	/* holds the final accumulated resources_used values from Moms, which does not include the released moms from job */
	struct  attribute tmpatr = {0};
	struct  attribute tmpatr3 = {0};
	char		*sval;
	char emsg[HOOK_BUF_SIZE];
	char *dumps = NULL;
	attribute_def	*ad3;
	PyObject *py_jvalue = NULL;
	PyObject *py_accum = NULL; /* holds accum resources_used values from all moms (including the released sister moms from job) */
	PyObject *py_accum3 = NULL; /* holds accum resources_used values from all moms (NOT including the released sister moms from job) */

	/* append resources_used */

	at = &pjob->ji_wattr[JOB_ATR_resc_used];
	ad = &job_attr_def[JOB_ATR_resc_used];
	if ((at->at_flags & ATR_VFLAG_SET) == 0)
		return;

	ad3 = &job_attr_def[JOB_ATR_resc_used_update];

	for (rs = (resource *)GET_NEXT(at->at_val.at_list);
		rs != NULL;
		rs = (resource *)GET_NEXT(rs->rs_link)) {

		rd = rs->rs_defin;

		if ((rd->rs_flags & resc_access_perm) == 0)
			continue;

		val = val3 = rs->rs_value;	/* copy resource attribute */

		/* count up sisterhood too */
		lnum = 0;
		lnum3 = 0;
		if (pjob->ji_resources != NULL) {
			/* NOTE: presence of pjob->ji_resources means a
			 *       multinode job (i.e. pjob->ji_numnodes > 1)
			 */
			if (strcmp(rd->rs_name, "cput") == 0) {
				for (i=0; i<pjob->ji_numrescs; i++) {
					noderes	*nr = &pjob->ji_resources[i];
					lnum += nr->nr_cput;
					if (nr->nr_status != PBS_NODERES_DELETE) {
						lnum3 += nr->nr_cput;
					}
				}
				val.at_val.at_long += lnum;
				val3.at_val.at_long += lnum3;
			}
			else if (strcmp(rd->rs_name, "mem") == 0) {
				for (i=0; i<pjob->ji_numrescs; i++) {
					noderes	*nr = &pjob->ji_resources[i];
					lnum += nr->nr_mem;
					if (nr->nr_status != PBS_NODERES_DELETE) {
						lnum3 += nr->nr_mem;
					}
				}
				val.at_val.at_long += lnum;
				val3.at_val.at_long += lnum3;
			}
			else if (strcmp(rd->rs_name, "cpupercent") == 0) {
				for (i=0; i<pjob->ji_numrescs; i++) {
					noderes	*nr = &pjob->ji_resources[i];
					lnum += nr->nr_cpupercent;
					if (nr->nr_status != PBS_NODERES_DELETE) {
						lnum3 += nr->nr_cpupercent;
					}
				}
				val.at_val.at_long += lnum;
				val3.at_val.at_long += lnum3;
#ifdef PYTHON
			} else if ((strcmp(rd->rs_name, RESOURCE_UNKNOWN) != 0) &&
				   ((val.at_type == ATR_TYPE_LONG)  ||
				(val.at_type == ATR_TYPE_FLOAT) ||
				(val.at_type == ATR_TYPE_SIZE)  ||
				(val.at_type == ATR_TYPE_STR))) {

				/* The following 2 temp variables will be set to 1
				 * if there's an error accuumlating resources_used
				 * values from all sister moms including those that
				 * have been released from the job (fail) or from
				 * all sister moms NOT including the released nodes
				 * from job (fail2).
				 */
				int	fail = 0;
				int	fail2 = 0;

				py_accum3 = NULL;
				py_jvalue = NULL;
				py_accum = NULL;

				(void)memset(&tmpatr, 0, sizeof(struct attribute));
				(void)memset(&tmpatr3, 0, sizeof(struct attribute));
				tmpatr.at_type  = tmpatr3.at_type  = val.at_type;

				if (val.at_type != ATR_TYPE_STR) {
					rd->rs_set(&tmpatr, &val, SET);
					rd->rs_set(&tmpatr3, &val, SET);
				} else {

					py_accum = PyDict_New();
					if (py_accum == NULL) {
						log_err(-1, __func__, "error creating accumulation dictionary");
						continue;
					}
					py_accum3 = PyDict_New();
					if (py_accum3 == NULL) {
						log_err(-1, __func__, "error creating accumulation dictionary 3");
						Py_CLEAR(py_accum);
						continue;
					}
				}

				/* accumulating resources_used values from sister
				 * moms into tmpatr (from all sisters including released
				 * moms) and tmpatr3 (from sisters that have not been
				 * released from the job).
				 */
				for (i=0; i < pjob->ji_numrescs; i++) {
					char mom_hname[PBS_MAXHOSTNAME+1];
					char *p = NULL;

					if (pjob->ji_resources[i].nodehost == NULL)
						continue;

					strncpy(mom_hname,
						pjob->ji_resources[i].nodehost,
							PBS_MAXHOSTNAME);
					mom_hname[PBS_MAXHOSTNAME] = '\0';
					p = strchr(mom_hname, '.');
					if (p != NULL)
						*p = '\0';

					at2 = &pjob->ji_resources[i].nr_used;
					if ((at2->at_flags & ATR_VFLAG_SET) == 0) {
						continue;
					}

					fail = fail2 = 0;
					for (rs2 = (resource *)GET_NEXT(at2->at_val.at_list);
						rs2 != NULL;
						rs2 = (resource *)GET_NEXT(rs2->rs_link)) {
						rd2 = rs2->rs_defin;
						val2 = rs2->rs_value;	/* copy resource attribute */
						if ((strcmp(rd2->rs_name, rd->rs_name) != 0) ||
							((val2.at_flags & ATR_VFLAG_SET) == 0)) {
							continue;
						}

						if (val2.at_type == ATR_TYPE_STR) {
							sval = val2.at_val.at_str;
							py_jvalue = json_loads(sval, emsg, HOOK_BUF_SIZE-1);
							if (py_jvalue == NULL) {

								snprintf(log_buffer, sizeof(log_buffer), "Job %s resources_used.%s cannot be accumulated: value '%s' from mom %s not JSON-format: %s", pjob->ji_qs.ji_jobid, rd2->rs_name, sval, mom_hname, emsg);
								log_err(-1, __func__, log_buffer);
								fail = 1;
							} else if (PyDict_Merge(py_accum, py_jvalue, 1) != 0) {
								snprintf(log_buffer, sizeof(log_buffer), "Job %s resources_used.%s cannot be accumulated: value '%s' from mom %s: error merging values", pjob->ji_qs.ji_jobid, rd2->rs_name, sval, mom_hname);
								log_err(-1, __func__, log_buffer);
								Py_CLEAR(py_jvalue);
								fail = 1;
							} else {
								if (pjob->ji_resources[i].nr_status != PBS_NODERES_DELETE) {
									if (PyDict_Merge(py_accum3, py_jvalue, 1) != 0) {
										snprintf(log_buffer, sizeof(log_buffer), "Job %s resources_used.%s cannot be accumulated: value '%s' from mom %s: error merging values", pjob->ji_qs.ji_jobid, rd2->rs_name, sval, mom_hname);
										log_err(-1, __func__, log_buffer);
										fail2 = 1;
									}
									Py_CLEAR(py_jvalue);
								} else {
									Py_CLEAR(py_jvalue);
								}
							}

						} else  {
							rd->rs_set(&tmpatr, &val2, INCR);
							if (pjob->ji_resources[i].nr_status != PBS_NODERES_DELETE) {
								rd->rs_set(&tmpatr3, &val2, INCR);
							}
						}
						break;
					}
				}

				/* accumulating the resources_used values from MS mom */

				if (val.at_type == ATR_TYPE_STR) {

					if (fail) {
						Py_CLEAR(py_accum);
						Py_CLEAR(py_accum3);
						/* unset resc */
						(void)add_to_svrattrl_list(
							phead,
							ad->at_name,
							rd->rs_name,
							"",
							SET, NULL);
						/* go to next resource to encode_used */
						continue;
					}

					if (fail2) {
						Py_CLEAR(py_accum);
						Py_CLEAR(py_accum3);
						/* unset resc */
						(void)add_to_svrattrl_list(
							phead,
							ad3->at_name,
							rd->rs_name,
							"",
							SET, NULL);
						/* go to next resource to encode_used */
						continue;
					}

					sval = val.at_val.at_str;
					if (PyDict_Size(py_accum) == 0) {
						/* no other values seen
						 * except from MS...use as is
						 * don't JSONify */
						rd->rs_decode(&tmpatr,
							ATTR_used, rd->rs_name,
							sval);
						Py_CLEAR(py_accum);
						Py_CLEAR(py_accum3);
					} else if ((py_jvalue = json_loads(sval, emsg, HOOK_BUF_SIZE - 1)) == NULL) {
						snprintf(log_buffer,
							sizeof(log_buffer),
							"Job %s resources_used.%s cannot be accumulated: value '%s' from mom %s not JSON-format: %s",
							pjob->ji_qs.ji_jobid,
							rd->rs_name, sval,
							mom_short_name, emsg);
						log_err(-1, __func__, log_buffer);
						Py_CLEAR(py_accum);
						Py_CLEAR(py_accum3);
						/* unset resc */
						(void)add_to_svrattrl_list(
							phead,
							ad->at_name,
							rd->rs_name,
							"",
							SET, NULL);
						/* go to next resource to encode */
						continue;
					} else if (PyDict_Merge(py_accum, py_jvalue, 1) != 0) {
						snprintf(log_buffer,
							sizeof(log_buffer),
							"Job %s resources_used.%s cannot be accumulated: value '%s' from mom %s: error merging values",
							pjob->ji_qs.ji_jobid,
							rd->rs_name, sval,
							mom_short_name);
						log_err(-1, __func__, log_buffer);
						Py_CLEAR(py_jvalue);
						Py_CLEAR(py_accum);
						Py_CLEAR(py_accum3);
						/* unset resc */
						(void)add_to_svrattrl_list(
							phead,
							ad->at_name,
							rd->rs_name,
							"",
							SET, NULL);
						/* go to next resource to encode */
						continue;
					} else {
						dumps = json_dumps(py_accum, emsg, HOOK_BUF_SIZE-1);
						if (dumps == NULL) {
							snprintf(log_buffer,
								sizeof(log_buffer),
								"Job %s resources_used.%s cannot be accumulated: %s",
								pjob->ji_qs.ji_jobid,
								rd->rs_name, emsg);
							log_err(-1, __func__, log_buffer);
							Py_CLEAR(py_jvalue);
							Py_CLEAR(py_accum);
							Py_CLEAR(py_accum3);
							/* unset resc */
							(void)add_to_svrattrl_list(
								phead,
								ad->at_name,
								rd->rs_name,
								"",
								SET, NULL);
							continue;
						}

						rd->rs_decode(&tmpatr, ATTR_used, rd->rs_name, dumps);
						Py_CLEAR(py_accum);
						free(dumps);

						if (PyDict_Merge(py_accum3, py_jvalue, 1) != 0) {
							snprintf(log_buffer,
								sizeof(log_buffer),
								"Job %s resources_used_update.%s cannot be accumulated: value '%s' from mom %s: error merging values",
								pjob->ji_qs.ji_jobid,
								rd->rs_name, sval,
								mom_short_name);
								log_err(-1, __func__, log_buffer);
								Py_CLEAR(py_jvalue);
								Py_CLEAR(py_accum3);
								/* unset resc */
								(void)add_to_svrattrl_list(
									phead,
									ad3->at_name,
									rd->rs_name,
									"",
									SET, NULL);
								/* go to next resource to encode */
								continue;
						} else if ((dumps=json_dumps(py_accum3, emsg, HOOK_BUF_SIZE-1)) == NULL) {
							snprintf(log_buffer,
								sizeof(log_buffer),
								"Job %s resources_used_update.%s cannot be accumulated: %s",
								pjob->ji_qs.ji_jobid,
								rd->rs_name, emsg);
							log_err(-1, __func__, log_buffer);
							Py_CLEAR(py_jvalue);
							Py_CLEAR(py_accum3);
							/* unset resc */
							(void)add_to_svrattrl_list(
								phead,
								ad3->at_name,
								rd->rs_name,
								"",
								SET, NULL);
							continue;
						}   else {
							rd->rs_decode(&tmpatr3, ATTR_used_update, rd->rs_name, dumps);
							Py_CLEAR(py_jvalue);
							Py_CLEAR(py_accum3);
							free(dumps);
						}
					}
				}
				val = tmpatr;
				val3 = tmpatr3;
#endif
			}
			/* no resource to accumulate and yet a multinode job */
		}

		if ((val.at_type != ATR_TYPE_STR) ||
			(pjob->ji_numnodes == 1)  ||
			(pjob->ji_resources != NULL)) {
			/* for string values, set value if single node job
			 * (i.e. pjob->ji_numnodes == 1), or
			 * if the value is accumulated from the various
			 * values obtained from sister nodes
			 * (i.e. pjob->ji_resources != NULL).
			 */
			if (val.at_type == ATR_TYPE_STR && pjob->ji_numnodes == 1) {
				/* check if string value is a valid json string,
				 * if it is then set the resource string within
				 * single quotes.
				 */

				sval = val.at_val.at_str;
				if ((py_jvalue = json_loads(sval, emsg, HOOK_BUF_SIZE - 1)) != NULL) {
					dumps = json_dumps(py_jvalue, emsg, HOOK_BUF_SIZE - 1);
					if (dumps == NULL)
						Py_CLEAR(py_jvalue);
					else {
						rd->rs_decode(&tmpatr, ATTR_used, rd->rs_name, dumps);
						val = tmpatr;
						Py_CLEAR(py_jvalue);
						free(dumps);
						dumps = NULL;
					}
				}
			}
			rc = rd->rs_encode(&val, phead,
				ad->at_name, rd->rs_name,
				ATR_ENCODE_CLIENT, NULL);
			if (rc < 0) {
				goto encode_used_exit;
			}

			rc = rd->rs_encode(&val3, phead,
				ad3->at_name, rd->rs_name,
				ATR_ENCODE_CLIENT, NULL);
			if (rc < 0) {
				goto encode_used_exit;
			}
		}

		if (((tmpatr.at_flags & ATR_VFLAG_SET) != 0) &&
					(tmpatr.at_type == ATR_TYPE_STR)) {
			rd->rs_free(&tmpatr);
		}
		if (((tmpatr3.at_flags & ATR_VFLAG_SET) != 0) &&
					(tmpatr3.at_type == ATR_TYPE_STR)) {
			rd->rs_free(&tmpatr3);
		}

	}
encode_used_exit:
	if (((tmpatr.at_flags & ATR_VFLAG_SET) != 0) &&
				(tmpatr.at_type == ATR_TYPE_STR)) {
		if (rd)
			rd->rs_free(&tmpatr);
	}
	if (((tmpatr3.at_flags & ATR_VFLAG_SET) != 0) &&
				(tmpatr3.at_type == ATR_TYPE_STR)) {
		if (rd)
			rd->rs_free(&tmpatr3);
	}
}

/**
 * @brief
 * 	Communicates the status (updated attributes, resources) of a single job
 *	to the server via the given mom-to-server 'cmd'.
 *
 * @param[in]	pjob - pointer to the job whose status is being returned.
 * @param[in]	cmd - command message to use to communicate status.
 * @param[in]	use_rtn_list_ext - set to 1 to use mom_rtn_list_ext[];
 *				   otherwise, use mom_rtn_list[]
 *				
 *
 * @return Void
 *
 */

void
update_ajob_status_using_cmd(job *pjob, int cmd, int use_rtn_list_ext)
{
	attribute		 *at;
	attribute_def		 *ad;
	int			  index;
	int			  nth = 0;
	struct resc_used_update	  rused;
	enum job_atr		*rtn_list;	

	if (use_rtn_list_ext)
		rtn_list = mom_rtn_list_ext;
	else
		rtn_list = mom_rtn_list; 
		

	/* pass user-client privilege to encode_resc() */

	resc_access_perm = ATR_DFLAG_MGRD;

	rused.ru_pjobid = pjob->ji_qs.ji_jobid;
	rused.ru_comment = NULL;
	rused.ru_status = 0;

	if (pjob->ji_wattr[(int)JOB_ATR_run_version].at_flags & ATR_VFLAG_SET)
		rused.ru_hop = pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long;
	else
		rused.ru_hop = pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long;

	CLEAR_HEAD(rused.ru_attr);
	rused.ru_next = NULL;

	/* Add attributes to the status reply */
	/* First, add the session id          */
	(void)job_attr_def[(int)JOB_ATR_session_id].at_encode(&pjob->ji_wattr[(int)JOB_ATR_session_id], &rused.ru_attr, job_attr_def[(int)JOB_ATR_session_id].at_name, NULL, ATR_ENCODE_CLIENT, NULL);

	/* Now add certain others as required for updating at the Server */
	for (index = 0; (int)rtn_list[index] != JOB_ATR_LAST; ++index) {
		nth = (int)rtn_list[index];
		at = &pjob->ji_wattr[nth];
		ad = &job_attr_def[nth];

		if (at->at_flags & ATR_VFLAG_MODIFY) {
			(void)ad->at_encode(at, &rused.ru_attr,
				ad->at_name, NULL,
				ATR_ENCODE_CLIENT, NULL);

			/* turn off modify so only sent if changed */
			at->at_flags &= ~ATR_VFLAG_MODIFY;
		}
	}


	/* if cmd is IS_RESCUSED_FROM_HOOK, send resources_used info
	 * to the server if coming from mother superior of job.
	 */
	if ((cmd != IS_RESCUSED_FROM_HOOK) ||
		((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) != 0)) {
		/* now append resources used */

		encode_used(pjob, &rused.ru_attr);
	}

	/* now send info to server via rpp */

	send_resc_used(cmd, 1, &rused);

	/* free svrattrl list */

	free_attrlist(&rused.ru_attr);
}

/**
 * @brief
 *	Wrapper to: update_ajob_status_using_cmd(pjob, IS_RESCUSED, 0).
 *
 * @param[in]	pjob - job whose status is being returned.
 *
 * @note
 *	This is used when a job is first started.   It returns the special
 *	listed attributes as well as resources used.
 * @return Void
 *
 */
void
update_ajob_status(job *pjob)
{
	update_ajob_status_using_cmd(pjob, IS_RESCUSED, 0);
}

/**
 * @brief
 * 	update_jobs_status - return the status of jobs to the server
 *
 *	Returns the updated resources_used for all running jobs.
 *	The special listed attrbutes are not returned because they are only
 *	modified when a job is first started and that case is covered by
 *	update_ajob_status() above.
 *
 * @return Void
 *
 */

void
update_jobs_status(void)
{
	int			count = 0;
	job			*pjob;
	struct resc_used_update	*prused;
	struct resc_used_update	*prusedtop = NULL;
	struct resc_used_update	**prusednext;	/* keep jobs in order */

	/* pass user-client privilege to encode_resc() */

	resc_access_perm = ATR_DFLAG_MGRD;
	prusednext = &prusedtop;

	for (pjob = (job *)GET_NEXT(svr_alljobs);
		pjob; pjob = (job *)GET_NEXT(pjob->ji_alljobs)) {

		if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0)
			continue;	/* not Mother Superior */
		if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_RUNNING)
			continue;

		++count;
		/* allocate reply structure and fill in header portion */
		prused = (struct resc_used_update *)
			malloc(sizeof(struct resc_used_update));
		assert(prused != NULL);
		prused->ru_pjobid = pjob->ji_qs.ji_jobid;
		prused->ru_comment= NULL;
		prused->ru_status = 0;
		if (pjob->ji_wattr[(int)JOB_ATR_run_version].at_flags & ATR_VFLAG_SET) {
			prused->ru_hop    = pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long;
		} else {
			prused->ru_hop    = pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long;
		}
		CLEAR_HEAD(prused->ru_attr);
		*prusednext	  = prused;	/* make last on list */
		prused->ru_next   = NULL;	/* terminate list */
		prusednext	  = &prused->ru_next;	/* track last link */

		/* now append the session id and resources used */
		(void)job_attr_def[(int)JOB_ATR_session_id].at_encode(
			&pjob->ji_wattr[(int)JOB_ATR_session_id],
			&prused->ru_attr,
			job_attr_def[(int)JOB_ATR_session_id].at_name,
			NULL, ATR_ENCODE_CLIENT, NULL);
		encode_used(pjob, &prused->ru_attr);

		if (svr_hook_resend_job_attrs != 0) {
			int		 index;
			int		 nth;
			attribute	*at;
			attribute_def	*ad;

			/* resend any of the attributes modified by a hook */
			/* just incase the update didn't reach the server  */

			for (index=0; (int)mom_rtn_list[index] != JOB_ATR_LAST; ++index) {
				nth = (int)mom_rtn_list[index];
				at = &pjob->ji_wattr[nth];
				ad = &job_attr_def[nth];

				if (at->at_flags & ATR_VFLAG_HOOK) {
					(void)ad->at_encode(at,
						&prused->ru_attr,
						ad->at_name, NULL,
						ATR_ENCODE_CLIENT, NULL);

				}
			}

		}
	}

	/* now send info to server via rpp */

	send_resc_used(IS_RESCUSED, count, prusedtop);

	/* free each resc_used_update struct and associated svrattrl list  */
	/* DO NOT use the free macro, ru_pjobid points into the job struct */
	/* and MUST be kept.						   */

	while (prusedtop) {
		prused = prusedtop;
		if (prused->ru_comment)
			(void)free(prused->ru_comment);
		free_attrlist(&prused->ru_attr);
		prusedtop = prused->ru_next;
		(void)free(prused);
	}

	svr_hook_resend_job_attrs = 0;	/* clear the send hooked flag */
}

/**
 * @brief
 * 	send_obit - routine called following completion of epilogue process
 *	Job then moved into substate OBIT and Obit RPP message sent to server.
 *
 * @param[in] pjob - pointer to job structure
 * @param[in] exval - exit value
 *
 * @return Void
 *
 */

void
send_obit(job *pjob, int exval)
{
	struct resc_used_update rud;
	pbs_list_head vnl_changes;

#ifndef WIN32
	/* update pjob with values set from an epilogue hook */
	/* since these are hooks that are executing in a child process */
	/* and changes inside the child will not be reflected in main */
	/* mom */
	if (num_eligible_hooks(HOOK_EVENT_EXECJOB_EPILOGUE) > 0) {
		char 	hook_outfile[MAXPATHLEN+1];
		struct	stat stbuf;
		snprintf(hook_outfile, MAXPATHLEN, FMT_HOOK_JOB_OUTFILE,
			path_hooks_workdir, pjob->ji_qs.ji_jobid);

		if (stat(hook_outfile, &stbuf) == 0) {
			int	      	reject_deletejob = 0;
			int	      	reject_rerunjob = 0;
			int	      	accept_flag = 1;

			CLEAR_HEAD(vnl_changes);

			if (get_hook_results(hook_outfile, &accept_flag,
				NULL, NULL, 0,
				&reject_rerunjob, &reject_deletejob, NULL,
				NULL, 0, &vnl_changes, pjob, NULL, 0,
								NULL) != 0) {
				log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_HOOK,
					LOG_ERR, "",
					"Failed to get epilogue hook results");
				vna_list_free(vnl_changes);
			} else {
				/* Delete job or reject job actions */
				/* NOTE: Must appear here before vnode changes, */
				/* since this action will be sent whether hook  */
				/* script executed by PBSADMIN or PBSUSER.      */
				if (reject_deletejob) {
					/* deletejob takes precedence */
#ifdef NAS /* localmod 005 */
					new_job_action_req(pjob, HOOK_PBSADMIN, 1);
#else
					new_job_action_req(pjob, 0, 1);
#endif /* localmod 005 */
				} else if (reject_rerunjob) {
#ifdef NAS /* localmod 005 */
					new_job_action_req(pjob, HOOK_PBSADMIN, 0);
#else
					new_job_action_req(pjob, 0, 0);
#endif /* localmod 005 */
				} else if (!accept_flag) {
					/* Per EDD on a pbs.event().reject() from an */
					/* epilogue hook, must delete the job. */
#ifdef NAS /* localmod 005 */
					new_job_action_req(pjob, HOOK_PBSADMIN, 1);
#else
					new_job_action_req(pjob, 0, 1);
#endif /* localmod 005 */
				}

				/* Whether or not we accept or reject, we'll make */
				/* job changes, vnode changes, job actions */


				update_ajob_status_using_cmd(pjob,
					IS_RESCUSED_FROM_HOOK, 0);

				/* Push vnl  hook changes to server */
				hook_requests_to_server(&vnl_changes);

			}
			/* need to clear out hook_outfile, */
			/* as epilogue hook processing  */
			/* in mom_process_hooks() will end up appending to */
			/* this same file when job is rerun, resulting in */
			/* duplicate actions. */
			unlink(hook_outfile);
		}
	}
#endif

	if (pjob->ji_wattr[(int)JOB_ATR_run_version].at_flags & ATR_VFLAG_SET) {
		DBPRT(("send_obit: job %s run_version %ld exval %d\n", pjob->ji_qs.ji_jobid, pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long, exval))
	} else {
		DBPRT(("send_obit: job %s runcount %ld exval %d\n", pjob->ji_qs.ji_jobid, pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long, exval))
	}

	pjob->ji_mompost = NULL;
	if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_OBIT) {
		pjob->ji_qs.ji_substate = JOB_SUBSTATE_OBIT;
		job_save(pjob, SAVEJOB_QUICK);
	}
	if (server_stream >= 0) {
		pjob->ji_sampletim = time_now;		/* when obit sent
								 to server */
		rud.ru_next   = NULL;
		rud.ru_pjobid = pjob->ji_qs.ji_jobid;	/* DO NOT free
							 this later */
		rud.ru_comment= NULL;
		/* epilogue script exit of 2 means requeue for	*/
		/* chkpt/restart if job was checkpointed	*/
		if ((exval == 2) &&
			(pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT))
			pjob->ji_qs.ji_un.ji_momt.ji_exitstat = JOB_EXEC_QUERST;
		rud.ru_status = pjob->ji_qs.ji_un.ji_momt.ji_exitstat;
		if (pjob->ji_wattr[(int)JOB_ATR_run_version].at_flags & ATR_VFLAG_SET) {
			rud.ru_hop = pjob->ji_wattr[(int)JOB_ATR_run_version].at_val.at_long;
		} else {
			rud.ru_hop = pjob->ji_wattr[(int)JOB_ATR_runcount].at_val.at_long;
		}
		CLEAR_HEAD(rud.ru_attr);
		encode_used(pjob, &rud.ru_attr);
#ifdef	WIN32
		if( pjob->ji_wattr[(int)JOB_ATR_Comment].at_flags & \
							ATR_VFLAG_SET) {
			rud.ru_comment = \
			    pjob->ji_wattr[(int)JOB_ATR_Comment].at_val.at_str;
		}
#endif
		/* now send info to server via rpp */
		send_resc_used(IS_JOBOBIT, 1, &rud);
		log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_JOB, LOG_DEBUG,
			pjob->ji_qs.ji_jobid, "Obit sent");

		/* free svrattrl list only */
		free_attrlist(&rud.ru_attr);
	} else {
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB, LOG_WARNING,
			pjob->ji_qs.ji_jobid, "Cannot Send Obit");

	}

	/*
	 **	Here, we reply to any checkpoint request that had
	 **	an abort set.  We need to send the obit before the
	 **	reply goes back.
	 */
	if (pjob->ji_preq) {
		reply_ack(pjob->ji_preq);
		pjob->ji_preq = NULL;
	}
}

/**
 * @brief
 * 	Look for job tasks that have terminated (see scan_for_terminating),
 *	and for each task, find which job the task was part, and if the top
 *	shell, start end of job processing by running the epilogue.
 *
 * @return Void
 *
 */

void
scan_for_exiting(void)
{

#ifndef WIN32
	pid_t			cpid;
#endif
	int			i;
	int			extval;
	int			found_one = 0;
	u_long			hours, mins, secs;
	job			*nxjob;
	job			*pjob;
	pbs_task		*ptask;
	obitent			*pobit;
	char			*cookie;
	u_long	gettime(resource *pres);
	u_long	getsize(resource *pres);
	int	im_compose(int, char *, char *, int, tm_event_t, tm_task_id, int);
	mom_hook_input_t hook_input;

#ifdef WIN32
	/* update the latest intelligence about the running jobs; */
	time_now = time(NULL);
	if (mom_get_sample() == PBSE_NONE) {
		pjob = (job *)GET_NEXT(svr_alljobs);
		while (pjob) {
			mom_set_use(pjob);
			pjob = (job *)GET_NEXT(pjob->ji_alljobs);
		}
	}
#endif

	/*
	 ** Look through the jobs.  Each one has it's tasks examined
	 ** and if the job is EXITING, it meets it's fate depending
	 ** on whether this is the Mother Superior or not.
	 */
	for (pjob = (job *)GET_NEXT(svr_alljobs); pjob; pjob = nxjob) {
		nxjob = (job *)GET_NEXT(pjob->ji_alljobs);

		/*
		 ** If a restart is active, skip this job since
		 ** not all of the tasks may have started yet.
		 */
		if (pjob->ji_flags & MOM_RESTART_ACTIVE) {
			continue;
		}
		/*
		 ** If a checkpoint with aborts is active,
		 ** skip it.  We don't want to report any obits
		 ** until we know that the whole thing worked.
		 */
		if ((pjob->ji_flags & MOM_CHKPT_ACTIVE) &&
			(pjob->ji_mompost != NULL)) {
			continue;
		}
		/*
		 ** If the job has had an error doing a checkpoint with
		 ** abort, the MOM_CHKPT_POST flag will be on.
		 */
		if (pjob->ji_flags & MOM_CHKPT_POST) {
			chkpt_partial(pjob);
			continue;
		}

		if (pjob->ji_wattr[(int)JOB_ATR_Cookie].at_flags &
			ATR_VFLAG_SET) {
			cookie = pjob->ji_wattr[(int)JOB_ATR_Cookie].
				at_val.at_str;
		}
		else
			cookie = NULL;

		/*
		 ** Check each EXITED task.  They transistion to DEAD here.
		 */
		for (ptask = (pbs_task *)GET_NEXT(pjob->ji_tasks);
			ptask != NULL;
			ptask = (pbs_task *)GET_NEXT(ptask->ti_jobtask)) {
			if (ptask->ti_qs.ti_status != TI_STATE_EXITED)
				continue;
			/*
			 ** Check if it is the top shell.
			 */
			if (ptask->ti_qs.ti_parenttask == TM_NULL_TASK) {
				int	*exitstat =
					&pjob->ji_qs.ji_un.ji_momt.ji_exitstat;

				pjob->ji_qs.ji_state    = JOB_STATE_EXITING;
				pjob->ji_qs.ji_substate = JOB_SUBSTATE_KILLSIS;
				if (*exitstat >= 0)
					*exitstat = ptask->ti_qs.ti_exitstat;
				log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
					LOG_INFO,
					pjob->ji_qs.ji_jobid, "Terminated");
#if IRIX6_CPUSET == 1
				/* KLUDGE - we need this fix when a job */
				/* was restarted from a restart (checkpoint) */
				/* MOM won't own the recreated processes */
				/* so it cannot be tracked by */
				/* scan_for_terminated() which calls */
				/* clear_cpuset(), but only here */

				/* THIS IS NO LONGER SUPPORTED and should */
				/* be removed */

				clear_cpuset(pjob);
#endif
#if	MOM_BGL
				(void)job_bgl_delete(pjob);
#endif	/* MOM_BGL */

				/*
				 ** Other places where clear_cpuset or
				 ** job_bgl_delete are called, the
				 ** function job_clean_extra would
				 ** also be called.  That is not done
				 ** here because the cleanup would
				 ** be done twice -- once here and
				 ** again when the job is deleted in
				 ** del_job_resc.
				 */

				if (send_sisters(pjob, IM_KILL_JOB, NULL)
					== 0) {
					pjob->ji_qs.ji_substate =
						JOB_SUBSTATE_EXITING;
					/*
					 ** if the job was checkpointed ok,
					 ** reset ji_nodekill to prevent mom_comm
					 ** error on restart resulting in job
					 ** being killed.
					 */
					if ((pjob->ji_flags & MOM_CHKPT_ACTIVE) &&
						!(pjob->ji_flags & MOM_CHKPT_POST)  &&
						(pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT))
						pjob->ji_nodekill = TM_ERROR_NODE;
				}
			}
			/*
			 ** Go through any TM client obits waiting.
			 */
			for (pobit = (obitent *)GET_NEXT(ptask->ti_obits);
				pobit != NULL;
				pobit = (obitent *)
				GET_NEXT(ptask->ti_obits)) {
				hnodent	*pnode;

				/* see if this is a batch request */
				if (pobit->oe_type == OBIT_TYPE_BREVENT) {
					pobit->oe_u.oe_preq->rq_reply.brp_code =
						PBSE_NONE;
					pobit->oe_u.oe_preq->rq_reply.brp_auxcode =
						ptask->ti_qs.ti_exitstat;
					pobit->oe_u.oe_preq->rq_reply.brp_choice  =
						BATCH_REPLY_CHOICE_NULL;
					(void)reply_send(pobit->oe_u.oe_preq);
					goto end_loop;
				}

				pnode = get_node(pjob, pobit->oe_u.oe_tm.oe_node);

				/* see if this is me or another MOM */
				if (pjob->ji_nodeid == pnode->hn_node) {
					pbs_task		*tp;

					/*
					 ** Send event to local kid.
					 */
					tp = task_find(pjob,
						pobit->oe_u.oe_tm.oe_taskid);
					if (pobit->oe_u.oe_tm.oe_fd != -1) {
						assert(tp != NULL);
						(void)tm_reply(pobit->oe_u.oe_tm.oe_fd,
							tp->ti_protover, IM_ALL_OKAY,
							pobit->oe_u.oe_tm.oe_event);
						(void)diswsi(pobit->oe_u.oe_tm.oe_fd,
							ptask->ti_qs.ti_exitstat);
						(void)DIS_tcp_wflush(pobit->oe_u.oe_tm.oe_fd);
					}
				}
				else if (pnode->hn_stream != -1 &&
					cookie != NULL) {
					/*
					 ** Send a response over to MOM
					 ** whose brat sent the request.
					 */
					(void)im_compose(pnode->hn_stream,
						pjob->ji_qs.ji_jobid,
						cookie, IM_ALL_OKAY,
						pobit->oe_u.oe_tm.oe_event,
						pobit->oe_u.oe_tm.oe_taskid, IM_OLD_PROTOCOL_VER);
					(void)diswsi(pnode->hn_stream,
						ptask->ti_qs.ti_exitstat);
					(void)rpp_flush(pnode->hn_stream);
				}

end_loop:
				delete_link(&pobit->oe_next);
				free(pobit);
			}
			ptask->ti_qs.ti_status = TI_STATE_DEAD;
			/*
			 ** KLUDGE
			 ** We need to save the value of the sid here just
			 ** in case it is exiting from a checkpoint/abort
			 ** and it will be restarted later.  Just set it
			 ** to the negative of itself.
			 */
			if (ptask->ti_qs.ti_sid <= 1) {
				ptask->ti_qs.ti_sid = 0;
#ifdef	_SX
				ptask->ti_qs.ti_u.ti_ext.ti_jid = 0;
#endif
			}
			else
				ptask->ti_qs.ti_sid = -ptask->ti_qs.ti_sid;
			task_save(ptask);
		}

		/*
		 ** Look to see if the job has terminated.  If it is
		 ** in any state other than EXITING continue on.
		 */
		if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_EXITING)
			continue;

		/*
		 ** This job is exiting.  If MOM_CHKPT_ACTIVE is on, it
		 ** is time to turn if off.
		 */
		pjob->ji_flags &= ~MOM_CHKPT_ACTIVE;

		/*
		 ** Once a job is exiting each task that is done running
		 ** gets a log message for the cpu and mem usage.
		 */
		ptask = (pbs_task *)GET_NEXT(pjob->ji_tasks);
		while (ptask != NULL) {
			secs = ptask->ti_cput;
			hours = secs/3600;
			secs -= hours*3600;
			mins = secs/60;
			secs -= mins*60;
			sprintf(log_buffer,
				"task %8.8X cput=%2lu:%2.2lu:%2.2lu",
				ptask->ti_qs.ti_task,
				hours, mins, secs);
			log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_JOB,
				LOG_DEBUG, pjob->ji_qs.ji_jobid, log_buffer);
			ptask = (pbs_task *)GET_NEXT(ptask->ti_jobtask);
		}

		/*
		 ** Look to see if I am a regular sister.  If so,
		 ** check to see if there is a obit event to
		 ** send back to mother superior.
		 ** Otherwise, I need to wait for her to send a KILL_JOB
		 ** so I can send the obit (unless she died).
		 */
		if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0) {
			int	stream = (pjob->ji_hosts == NULL) ? -1 :
				pjob->ji_hosts[0].hn_stream;

			/*
			 ** Check to see if I'm still in touch with
			 ** the head office.  If not, I'm just going to
			 ** get rid of this job.
			 */
			if (stream == -1) {
				(void)kill_job(pjob, SIGKILL);
				if ((pjob->ji_qs.ji_svrflags &
					(JOB_SVFLG_CHKPT | JOB_SVFLG_ChkptMig)) == 0) {
					mom_deljob(pjob);
				}
				continue;
			}

			/*
			 * No event waiting for sending info to MS
			 * so I'll just sit tight.
			 */
			if (pjob->ji_obit == TM_NULL_EVENT)
				continue;

			/* Check to see if any tasks are running */
			ptask = (pbs_task *)GET_NEXT(pjob->ji_tasks);
			while (ptask != NULL) {
				if (ptask->ti_qs.ti_status == TI_STATE_RUNNING)
					break;
				ptask = (pbs_task *)GET_NEXT(ptask->ti_jobtask);
			}
			/* Still somebody there so don't send it yet. */
			if (ptask != NULL)
				continue;
			/* No tasks running. Format and send a reply to the mother superior */
			if (cookie != NULL) {
				(void)im_compose(stream, pjob->ji_qs.ji_jobid,
					cookie, IM_ALL_OKAY,
					pjob->ji_obit, TM_NULL_TASK, IM_OLD_PROTOCOL_VER);
				(void)diswul(stream,
					resc_used(pjob, "cput", gettime));
				(void)diswul(stream,
					resc_used(pjob, "mem", getsize));
				(void)diswul(stream,
					resc_used(pjob, "cpupercent", gettime));
				(void)send_resc_used_to_ms(stream,
							pjob->ji_qs.ji_jobid);
				(void)rpp_flush(stream);
				pjob->ji_obit = TM_NULL_EVENT;
			}
			continue;
		}

		/*
		 * At this point, we know we are Mother Superior for this
		 * job which is EXITING.  Time for it to die.
		 */
		pjob->ji_qs.ji_svrflags &= ~(JOB_SVFLG_Suspend|
			JOB_SVFLG_Actsuspd);
		if (pjob->ji_qs.ji_un.ji_momt.ji_exitstat != JOB_EXEC_INITABT)
			(void)kill_job(pjob, SIGKILL);
		delete_link(&pjob->ji_jobque);	/* unlink from poll list */

		/*
		 * The SISTER_KILLDONE flag needs to be reset so
		 * we can talk to the sisterhood.
		 */
		for (i=0; i<pjob->ji_numnodes; i++) {
			hnodent		*np = &pjob->ji_hosts[i];

			if (np->hn_node == pjob->ji_nodeid)	/* me */
				continue;

			if (np->hn_sister == SISTER_KILLDONE)
				np->hn_sister = SISTER_OKAY;
		}

		/* Job termination begins */

		/* stop counting walltime */
		stop_walltime(pjob);

		/* summary for MS */
		secs = resc_used(pjob, "cput", gettime);
		hours = secs/3600;
		secs -= hours*3600;
		mins = secs/60;
		secs -= mins*60;
		sprintf(log_buffer,
			"%s cput=%2lu:%2.2lu:%2.2lu mem=%lukb",
			mom_short_name, hours, mins, secs,
			resc_used(pjob, "mem", getsize));
		log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_JOB,
			LOG_DEBUG, pjob->ji_qs.ji_jobid, log_buffer);

		/* summary for other nodes */
		for (i=0; i<pjob->ji_numrescs; i++) {
			noderes	*nr = &pjob->ji_resources[i];
			secs = nr->nr_cput;

			hours = secs/3600;
			secs -= hours*3600;
			mins = secs/60;
			secs -= mins*60;

			/*
			 ** ji_hosts starts with node 0 (MS)
			 ** ji_resource starts with node 1
			 */
			sprintf(log_buffer,
				"%s cput=%2lu:%2.2lu:%2.2lu mem=%lukb",
				pjob->ji_resources[i].nodehost?
				pjob->ji_resources[i].nodehost:"",
				hours, mins, secs, nr->nr_mem);
			log_event(PBSEVENT_DEBUG2, PBS_EVENTCLASS_JOB,
				LOG_DEBUG, pjob->ji_qs.ji_jobid, log_buffer);
		}

#ifdef WIN32 /* WIN32 --------------------------------------------- */
		/* change to the user's home directory and */
		/* run the epilogue script */
		/* if sandbox=PRIVATE, then run the epilogue script */
		/* in staging and execution directory */
		if (pjob->ji_grpcache) {
			if ((pjob->ji_wattr[(int)JOB_ATR_sandbox].at_flags & ATR_VFLAG_SET) && (strcasecmp(pjob->ji_wattr[JOB_ATR_sandbox].at_val.at_str, "PRIVATE") == 0)) {
				/* in "sandbox=PRIVATE" mode, so run epilogue */
				/* in PBS_JOBDIR */
				(void)chdir(jobdirname(pjob->ji_qs.ji_jobid,
					pjob->ji_grpcache->gc_homedir));
			} else {
				(void)chdir(pjob->ji_grpcache->gc_homedir);
			}
		}
		i = 0;
		mom_hook_input_init(&hook_input);
		hook_input.pjob = pjob;

		if (mom_process_hooks(HOOK_EVENT_EXECJOB_EPILOGUE,
			PBS_MOM_SERVICE_NAME, mom_host, &hook_input, NULL,
			NULL, 0, 1) == 2) {
			i = run_pelog(PE_EPILOGUE, path_epilog, pjob,
				PE_IO_TYPE_STD);
			if ((i == 2) &&
				(pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT))
			pjob->ji_qs.ji_un.ji_momt.ji_exitstat = \
								JOB_EXEC_QUERST;
			else
				i = 0;
		}
		send_obit(pjob, i);

		/* restore MOM's home since in Windows, we're in main mom */
		(void)chdir(mom_home);
#else	     /* UNIX  --------------------------------------------- */

		/*
		 ** Do dependent end of job processing if it needs to be
		 ** done.
		 */
		if (job_end_final != NULL)
			job_end_final(pjob);
#if MOM_CSA
		/*
		 ** if capability present, cause a workload management
		 ** record to be created for this phase of the job
		 */

		write_wkmg_record(WM_TERM, WM_TERM_EXIT, pjob);
#endif	/* MOM_CSA */

		/*
		 * Parent:
		 *  +  fork child process to run epilogue,
		 *  +  look for more terminated jobs.
		 * Child:
		 *  +  Run the epilogue script (if one)
		 */

		cpid = fork_me(-1);
		if (cpid > 0) {
			/* parent = mark that it is being sent */
			pjob->ji_sampletim = 0;	/* will be set in send_obit */
			pjob->ji_momsubt = cpid;
			pjob->ji_actalarm = 0;
			pjob->ji_mompost = send_obit;
			pjob->ji_qs.ji_substate = JOB_SUBSTATE_RUNEPILOG;

			if (found_one++ < 5) {
				continue;	/* look for more */
			} else {
				break;	/* five at a time is our limit */
			}
		} else if (cpid < 0)
			continue;	/* curses, foiled again */

		/* child: change to the user's home directory or PBS_JOBDIR */
		/* and run the epilogue script				    */

		if (pjob->ji_grpcache) {
			if ((pjob->ji_wattr[(int)JOB_ATR_sandbox].at_flags & ATR_VFLAG_SET) && (strcasecmp(pjob->ji_wattr[JOB_ATR_sandbox].at_val.at_str, "PRIVATE") == 0)) {
				/* in "sandbox=PRIVATE" mode so run epilogue */
				/* in PBS_JOBDIR */
				(void)chdir(jobdirname(pjob->ji_qs.ji_jobid,
					pjob->ji_grpcache->gc_homedir));

			} else {

				/* else run in usr's home */
				(void)chdir(pjob->ji_grpcache->gc_homedir);
			}
		}

		extval = 0;

		if (num_eligible_hooks(HOOK_EVENT_EXECJOB_EPILOGUE) > 0) {
			mom_hook_input_init(&hook_input);
			hook_input.pjob = pjob;
			(void)mom_process_hooks(HOOK_EVENT_EXECJOB_EPILOGUE,
				PBS_MOM_SERVICE_NAME, mom_host, &hook_input,
				NULL, NULL, 0, 0);
		} else {

			if ((pjob->ji_wattr[(int)JOB_ATR_interactive].at_flags & ATR_VFLAG_SET) && pjob->ji_wattr[(int)JOB_ATR_interactive].at_val.at_long) {

				extval = run_pelog(PE_EPILOGUE, path_epilog,
					pjob, PE_IO_TYPE_NULL);
			} else {
				extval = run_pelog(PE_EPILOGUE, path_epilog,
					pjob, PE_IO_TYPE_STD);
			}
		}

		if (extval != 2)
			extval = 0;
		exit(extval);
#endif	/* WIN32/UNIX */

	}
	if (pjob == NULL)
		exiting_tasks = 0;	/* went through all jobs */
}

/**
 * @brief
 * 	send old style IS_RESTART message to Server.
 *	Used when Server is older & does not recognize the TCP Restart message.
 *
 * @par
 *	Open an RPP stream to the named server/port, compose the IS_RESTART,
 *	flush the stream and then close it.
 *
 * @param[in]	svr  - name of Server to which to send the restart
 * @param[in]	port - port Server would be expecting to receive IM messages
 *
 * @return	void
 *
 */

static void
send_restart_rpp(char *svr, unsigned int port)
{
	int		j;

	j = rpp_open(svr, port);

	if (j < 0) {
		(void)sprintf(log_buffer, "rpp_open(%s, %d) failed", svr, port);
		log_err(errno, msg_daemonname, log_buffer);
		return;
	}

	if (is_compose(j, IS_RESTART) != DIS_SUCCESS) {
		(void)sprintf(log_buffer, "Failed to compose restart message");
		log_err(errno, msg_daemonname, log_buffer);
		rpp_close(j);
		return;
	}

	(void)diswui(j, pbs_mom_port);
	rpp_flush(j);
	(void)sprintf(log_buffer, "Restart sent to server at %s:%d", svr, port);
	log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, LOG_NOTICE,
		msg_daemonname, log_buffer);
	rpp_close(j);
}

/**
 * @brief
 * 	send PBS_BATCH_MomRestart message to Server via tcp.
 * @par
 *	Open an TCP connection to the Server/port specified.
 *	Build the batch request and send to the Server.
 *	Wait to read the reply back from the Server.
 *	If accepted, return 0, if explicitly rejected with PBSE_UNKREQ, then
 *	return 1, else return -1 on other errors.
 *
 * @param[in]	svr  - name of Server to which to send the restart
 * @param[in]	port - port Server would be expecting to receive IM messages
 *
 * @return	int
 * @rtnval  0 - Restart was sent to Server and acknowledged
 * @rtnval  1 - Restart was sent to Server but rejected, fall back to RPP
 * @rtnval -1 - Error in looking up host, connecting, or sending message
 *
 */
int
send_restart_tcp(char *svr, unsigned int port)
{
	pbs_net_t hostaddr;
	int rtn;
	struct batch_reply *reply;
	int sock;
	int mode;

	/* first, make sure we have a valid server (host), and ports */
	if ((hostaddr = get_hostaddr(svr)) == (pbs_net_t)0) {
		return (-1);
	}

	mode = B_RESERVED;
	if (pbs_conf.auth_method == AUTH_MUNGE)
		mode = B_EXTERNAL|B_SVR;

	sock = client_to_svr(hostaddr, port, mode);
	if (sock < 0) {
		return (-1);
	}

	DIS_tcp_setup(sock);

	/* send authentication information */

	if (encode_DIS_ReqHdr(sock, PBS_BATCH_MomRestart, "root") ||
		diswst(sock, mom_host)      ||
		diswui(sock, pbs_mom_port)  ||
		encode_DIS_ReqExtend(sock, NULL)) {
		return (-1);
	}
	if (DIS_tcp_wflush(sock)) {
		return (-1);
	}

	/* read back the response */

	reply = (struct batch_reply *)malloc(sizeof(struct batch_reply));
	if (reply == NULL)
		return (-1);
	(void)memset(reply, 0, sizeof(struct batch_reply));
	DIS_tcp_setup(sock);
	if (decode_DIS_replyCmd(sock, reply) != 0) {
		(void)free(reply);
		return (-1);
	}
	DIS_tcp_reset(sock, 0);

	CLOSESOCKET(sock);
	rtn = reply->brp_code;
	PBSD_FreeReply(reply);

	if (rtn == PBSE_NONE) {
		return (0);	/* restart sent via tcp */
	} else if (rtn == PBSE_UNKREQ) {
		return (1);	/* need to fall back to RPP IS_RESTART */
	}
	return (-1);
}

/**
 * @brief	Send a restart message to the Server.
 *
 * @par
 *	Close any existing rpp streams to the server, it is unlikely that
 *	there is one.  Parse the server name from pbs.conf;
 *	Use PBS_SERVER_HOST_NAME if defined, else use PBS_SERVER.
 *	Try sending message via TCP first
 *
 * @see send_restart_tcp()
 *	If that returns 1 or -1, fall back to useing rpp
 * @see send_restart()
 *
 * @return void
 */
void
send_restart(void)
{
	unsigned int	port = default_server_port;
	char	       *svr;

	if (server_stream >= 0) {
		sprintf(log_buffer, "Closing existing server stream %d",
			server_stream);
		log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, LOG_NOTICE,
			msg_daemonname, log_buffer);
		rpp_flush(server_stream);
		rpp_close(server_stream);
		server_stream = -1;
	}


	svr = get_servername(&port);
	if (pbs_conf.pbs_use_tcp==0 && send_restart_tcp(svr, port) == 0) {
		(void)sprintf(log_buffer, "Restart sent to server at %s:%d", svr, port);
		log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, LOG_NOTICE,
			msg_daemonname, log_buffer);
	} else {
		/* since sending via TCP didn't work, fall back to old rpp */
		send_restart_rpp(svr, port);
	}

}

/**
 * @brief
 *	On mom initialization, recover all running jobs.
 *
 *	Called on initialization
 *	   If the -p option was given (recover = 2), Mom will allow the jobs
 *	   to continue to run.   She depends on detecting when they terminate
 *	   via the slow poll method rather than SIGCHLD.
 *
 *	   If the -r option was given (recover = 1), MOM is recovering on a
 *  	   running system and the session id of the jobs should be valid;
 *	   the jobs are killed.
 *
 *	   If -r was not given (recover = 0), it is assumed that the whole
 *	   system, not just MOM, is comming up, the session ids are not valid;
 *	   so no attempt is made to kill the job processes.  But the jobs are
 *	   terminated and requeued.
 *
 * @param [in]	recover - Specify recovering mode for MoM.
 *
 */

void
init_abort_jobs(int recover)
{
	DIR		*dir;
	int		i, sisters;
	struct dirent	*pdirent;
	job		*pj = NULL;
	char		*job_suffix = JOB_FILE_SUFFIX;
	int		job_suf_len = strlen(job_suffix);
	char		*psuffix;
	char		path[MAXPATHLEN+1];
	char		oldp[MAXPATHLEN+1];
	char		rcperr[] = "rcperr.";
	struct	stat	statbuf;
	extern	char	*path_checkpoint;
	extern	char	*path_spool;

	dir = opendir(path_jobs);
	if (dir == NULL) {
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, LOG_ALERT,
			msg_daemonname, "Jobs directory not found");
		exit(1);
	}
	while (errno = 0, (pdirent = readdir(dir)) != NULL) {
		if ((i = strlen(pdirent->d_name)) <= job_suf_len)
			continue;

		psuffix = pdirent->d_name + i - job_suf_len;
		if (strcmp(psuffix, job_suffix))
			continue;
		pj = job_recov(pdirent->d_name);
		if (pj == NULL) {
			(void)strcpy(path, path_jobs);
			(void)strcat(path, pdirent->d_name);
			(void)unlink(path);
			psuffix = path + strlen(path) - job_suf_len;
			strcpy(psuffix, JOB_TASKDIR_SUFFIX);
			(void)remtree(path);
			continue;
		}

		/* To get homedir info */
		pj->ji_grpcache = NULL;
		check_pwd(pj);
		append_link(&svr_alljobs, &pj->ji_alljobs, pj);
		job_nodes(pj);
		task_recov(pj);

		/*
		 ** Check to see if a checkpoint.old dir exists.
		 ** If so, remove the regular checkpoint dir
		 ** and rename the old to the regular name.
		 */
		strcpy(path, path_checkpoint);
		if (*pj->ji_qs.ji_fileprefix != '\0')
			strcat(path, pj->ji_qs.ji_fileprefix);
		else
			strcat(path, pj->ji_qs.ji_jobid);
		strcat(path, JOB_CKPT_SUFFIX);
		strcpy(oldp, path);
		strcat(oldp, ".old");

		if (stat(oldp, &statbuf) == 0) {
			(void)remtree(path);
			if (rename(oldp, path) == -1)
				(void)remtree(oldp);
		}

		/*
		 ** Check to see if I am Mother Superior.  The
		 ** JOB_SVFLG_HERE flag is overloaded for MOM
		 ** for this purpose.
		 */
		if ((pj->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0) {
			/* I am sister, junk the job files */
			mom_deljob(pj);
			continue;
		}

		sisters = pj->ji_numnodes - 1;
		if (sisters > 0) {
			pj->ji_resources = (noderes *)calloc(sisters,
				sizeof(noderes));
			if (pj->ji_resources == NULL) {
				log_err(ENOMEM, "init_abort_jobs", "out of memory");
				continue;
			}
			pj->ji_numrescs = sisters;
		}

		/*
		 **	If mom went down during file stage ops,
		 **	the substate should be EXITED.  Set it
		 **	back to OBIT so the server can verify that
		 **	it still has the job or not.
		 */
		if (pj->ji_qs.ji_substate == JOB_SUBSTATE_EXITED) {
			/*
			 ** We don't want to change the state if the
			 ** job is checkpointed.
			 */
			if ((pj->ji_qs.ji_svrflags &
				(JOB_SVFLG_CHKPT|
				JOB_SVFLG_ChkptMig)) == 0) {
				pj->ji_qs.ji_substate = JOB_SUBSTATE_OBIT;
				job_save(pj, SAVEJOB_QUICK);
			}
		} else if (pj->ji_qs.ji_substate == JOB_SUBSTATE_TERM) {
			/*
			 * Mom went down while terminate action script was
			 * running, don't know if it finished or not;  force
			 * Mom to send/resend OBIT and lets end it
			 */
			if (recover)
				(void)kill_job(pj, SIGKILL);
			pj->ji_qs.ji_substate = JOB_SUBSTATE_OBIT;
			job_save(pj, SAVEJOB_QUICK);
		} else if ((recover != 2) &&
			((pj->ji_qs.ji_substate == JOB_SUBSTATE_RUNNING) ||
			(pj->ji_qs.ji_substate == JOB_SUBSTATE_SUSPEND) ||
			(pj->ji_qs.ji_substate == JOB_SUBSTATE_KILLSIS)   ||
			(pj->ji_qs.ji_substate == JOB_SUBSTATE_RUNEPILOG) ||
			(pj->ji_qs.ji_substate == JOB_SUBSTATE_EXITING))) {

			if (recover)
				(void)kill_job(pj, SIGKILL);

			/* set exit status to:
			 *   JOB_EXEC_INITABT - init abort and no chkpnt
			 *   JOB_EXEC_INITRST - init and chkpt, no mig
			 *   JOB_EXEC_INITRMG - init and chkpt, migrate
			 * to indicate recovery abort
			 */
			if (pj->ji_qs.ji_svrflags &
				(JOB_SVFLG_CHKPT |
				JOB_SVFLG_ChkptMig)) {
#if PBS_CHKPT_MIGRATE
				pj->ji_qs.ji_un.ji_momt.ji_exitstat =
					JOB_EXEC_INITRMG;
#else
				pj->ji_qs.ji_un.ji_momt.ji_exitstat =
					JOB_EXEC_INITRST;
#endif
			} else {
				pj->ji_qs.ji_un.ji_momt.ji_exitstat =
					JOB_EXEC_INITABT;
			}

			/*
			 ** I am MS, send a DELETE_JOB request to any
			 ** sisters that happen to still be alive.
			 */
			if (sisters > 0) {
				(void)send_sisters(pj, IM_DELETE_JOB, NULL);
			}

			pj->ji_qs.ji_substate = JOB_SUBSTATE_EXITING;
			job_save(pj, SAVEJOB_QUICK);
			exiting_tasks = 1;
		} else if (recover == 2) {
			pbs_task	*ptask;

			for (ptask = (pbs_task *)GET_NEXT(pj->ji_tasks);
				ptask != NULL;
				ptask = (pbs_task *)GET_NEXT(ptask->
				ti_jobtask)) {
				ptask->ti_flags |= TI_FLAGS_ORPHAN;
			}

			if (pj->ji_qs.ji_substate == JOB_SUBSTATE_RUNNING) {
				recover_walltime(pj);
				start_walltime(pj);
			}

			if (mom_do_poll(pj))
				append_link(&mom_polljobs, &pj->ji_jobque, pj);
		}
	}
	if (errno != 0 && errno != ENOENT) {
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, LOG_ALERT,
			msg_daemonname, "Jobs directory cannot be read");
		(void)closedir(dir);
		exit(1);
	}
	(void)closedir(dir);

	/*
	 ** Go through spool dir and remove files that match
	 ** "rcperr.<pid>".  These would be leftover from file
	 ** stage operations that were interrupted.
	 */
	dir = opendir(path_spool);
	if (dir == NULL) {
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, LOG_ALERT,
			msg_daemonname, "spool directory not found");
		return;
	}

	while (errno = 0, (pdirent = readdir(dir)) != NULL) {
		if (strncmp(pdirent->d_name, rcperr, sizeof(rcperr)-1) != 0)
			continue;

		(void)strcpy(path, path_spool);
		(void)strcat(path, pdirent->d_name);
		(void)unlink(path);
	}
	if (errno != 0 && errno != ENOENT)
		log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, LOG_ALERT,
			msg_daemonname, "spool directory cannot be read");
	(void)closedir(dir);
}

/**
 * @brief
 * 	static handler function to be called by deferred child exit work task
 * 	for alps cancel reservation child of mom
 *
 * 	The forked child process cannot send a req_reject or reply_ack since
 * 	transmission of data via rpp is not supported from child processes
 * 	(rpp streams are automatically closed when proccess forks).
 * 	Thus this child exit handler is added to send the reply from the
 * 	parent process after reaping the exit status from child
 *
 * @param[in] ptask - Pointer to the task structure
 *
 */
#if MOM_ALPS
static void
post_alps_cancel_resv(struct work_task *ptask)
{
	struct batch_request *preq = ptask->wt_parm1;
	int j;

	if (preq == NULL)
		return;

	j = ptask->wt_aux;
	if (j > 0) {
		/* Tell the server we failed */
		req_reject(PBSE_ALPSRELERR, j, preq);
	} else if (j < 0) {
		/* Fatal error, log message was logged in
		 * alps_cancel_request
		 */
		req_reject(PBSE_ALPSRELERR, j, preq);
	} else {
		/* The job will have been purged in mom_deljob_wait at this point
		 * so just do the reply.
		 */
		reply_ack(preq);
	}
}
#endif


/**
 * @brief
 * 	del_job_hw	delete job/hardware related resources such as cpusets, ...
 *
 *	Used by del_job_resc() and exec_bail()
 *	Most items here are platform dependent.
 *
 * @param pjob  - pointer to job structure
 *
 * @return void
 *
 */
void
del_job_hw(job *pjob)
{
#if MOM_ALPS
	int		i;
	int		j;
	int 	sleeptime = 0;
	time_t 	total_time = 0;
	time_t 	begin_time = 0;
	time_t 	end_time = 0;
	long 	jitter = 0;
	pid_t	parent_pid=0;
	pid_t	pid;
	int	sconn = -1;
	struct work_task *wtask = NULL;
#endif

#if	MOM_CPUSET
	clear_cpuset(pjob);
#endif	/* MOM_CPUSET */

#if	MOM_BGL
	(void)job_bgl_delete(pjob);
#endif	/* MOM_BGL */

#if MOM_ALPS

	/*
	 * Try to cancel the reservation once as 'main MOM'.
	 * If we got an acknowledgment from ALPS that the reservation
	 * is actually gone, then send ACK to server.
	 * Else, fork a child process that will continue to try to cancel
	 * the reservation until the remaining processes count is zero.
	 * Or until the ALPS reservation no longer exists.
	 */
	if ((j = alps_cancel_reservation(pjob)) > 0) {
		/*
		 * alps reservation cancel failed with "temporary" error
		 * This could be due to one of more of the following:
		 * 	- the reservation still has claims on it
		 * 	- ALPS is down
		 * Retry in child until success, or a hard error is returned
		 * Or alps_release_timeout is reached.
		 * Once the ALPS reservation is successfully canceled,
		 * respond to the server's delete job request.
		 * The job will remain in the 'E' state until then.
		 */
		if (pjob->ji_preq != NULL)
			sconn = pjob->ji_preq->rq_conn;

		if ((pid = fork_me(sconn)) == 0) {
			/* We are the child */
			begin_time = time(NULL);
			end_time = begin_time;
			/* add jobid to the seed */
			srandom((unsigned)(atoi(pjob->ji_qs.ji_jobid) + begin_time));
			for (i = 1; (total_time = end_time - begin_time) < alps_release_timeout; ++i, end_time = time(NULL)) {
				/* calculate time to sleep */
				sleeptime = alps_release_wait_time;
				/* Add randomness of 0 to 0.12 seconds to the
				 * sleeptime so we don't overwhelm ALPS with
				 * multiple ALPS release requests when jobs end
				 * at the same time.
				 */
				jitter = random() % alps_release_jitter;
				sleeptime += jitter;
				usleep(sleeptime);
				if ((j = alps_cancel_reservation(pjob)) <= 0)
					break;
			}
			if (j > 0) {
				sprintf(log_buffer,
					"Timed out after %d attempts over "
					"%ld seconds of attempting "
					"to cancel ALPS reservation %ld",
					i, total_time,
					pjob->ji_extended.ji_ext.ji_reservation);
				log_joberr(-1, __func__, log_buffer,
					pjob->ji_qs.ji_jobid);
				/* send a HUP to main MOM so she re-reads
				 * the ALPS inventory
				 */
				parent_pid = getppid();
				kill(parent_pid, SIGHUP);

			} else if (j == 0) {
				sprintf(log_buffer,
					"Cancelled ALPS reservation %ld after a "
					"total of %d tries",
					pjob->ji_extended.ji_ext.ji_reservation, i + 1);
				log_event(PBSEVENT_DEBUG3, PBS_EVENTCLASS_JOB,
					LOG_DEBUG, pjob->ji_qs.ji_jobid,
					log_buffer);
			}

			/* exit with the respective error code to the parent process
			 * Parents (moms) post handler will handle this
			 */
			exit(j);

		} else if (pid > 0) {
			/* we are the parent, the reply happens after the child exits */
			if ((wtask = set_task(WORK_Deferred_Child, pid,
					post_alps_cancel_resv, pjob->ji_preq)) == NULL) {
				log_err(errno, NULL, "Failed to create deferred work task, Out of memory");
				req_reject(PBSE_SYSTEM, 0, pjob->ji_preq);
			}
		} else if (pid < 0) {
			/* fork failed, reply to the server so the job
			 * doesn't stay in the "E" state
			 */
			req_reject(PBSE_ALPSRELERR, j, pjob->ji_preq);
		}
	} else if (j < 0) {
		/* ALPS returned a PERMANENT error */
		req_reject(PBSE_ALPSRELERR, j, pjob->ji_preq);
	} else {
		/* The reservation was canceled, let server know */
		reply_ack(pjob->ji_preq);
	}
	pjob->ji_preq = NULL;
#endif
}

/**
 * @brief
 * 	del_job_resc - delete job related resources, files, etc
 *	Used by mom_deljob() and mom_deljob_wait()
 *
 *	Items which are kept until the very bitter end of the job, just
 *	before the job structure is freed, are released/freed/cleared here.
 *
 * @param[in] pjob - pointer to job structure
 *
 * @return Void
 *
 */
static void
del_job_resc(job *pjob)
{
	/*
	 * WARNING - the following is for QA automated testing to induce
	 * certain failures modes
	 */

	if (QA_testing != 0) {
		if (QA_testing & PBSQA_DELJOB_SLEEP)
			sleep(90);	/* 90 second delay */
		else if (QA_testing & PBSQA_DELJOB_SLEEPLONG)
			sleep(900);	/* 900 second long delay */
		else if (QA_testing & PBSQA_DELJOB_CRASH)
			exit(99);	/* simulate crash */
	}

	/* remove PBS_NODEFILE - Mother Superior only has one */

	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) {
		char	file[MAXPATHLEN+1];
#ifdef WIN32
		(void)sprintf(file, "%s/auxiliary/%s",
			pbs_conf.pbs_home_path, pjob->ji_qs.ji_jobid);
#else
		(void)sprintf(file, "%s/aux/%s",
			pbs_conf.pbs_home_path, pjob->ji_qs.ji_jobid);
#endif
		(void)unlink(file);
	}

	/* TMPDIR removed in job_purge so files are available for staging */

	if (job_clean_extra != NULL) {
		(void)job_clean_extra(pjob);
	}

	/* delete the hardware related items */

	del_job_hw(pjob);

}

/**
 * @brief
 * 	mom_deljob - delete the job entry, MOM no longer knows about the job
 *	This version does NOT wait for the Sisters to reply
 *
 * @param[in] pjob - pointer to job structure
 *
 * @return Void
 *
 */
void
mom_deljob(job *pjob)
{

	del_job_resc(pjob);	/* rm tmpdir, cpusets, etc */

	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE)	/* MS */
		(void)send_sisters(pjob, IM_DELETE_JOB, NULL);
	job_purge(pjob);

	/*
	 ** after job is gone, check to make sure no rogue user
	 ** procs are still hanging about
	 */
	dorestrict_user();
}

/**
 * @brief
 * 	mom_deljob_wait - deletes most of the job stuff, job entry not deleted
 *	untill the sisters have rplied or are down
 *	This version DOES wait for the Sisters to reply, see processing of
 *	IM_DELETE_JOB_REPLY in mom_comm.c
 *	IT should only be called for a job for which this is Mother Superior.
 *
 * @param[in] pjob - pointer to job structure
 *
 * @return int
 * @retval the number of sisters to whom the request was sent
 *
 */
int
mom_deljob_wait(job *pjob)
{
	int	i;

	del_job_resc(pjob);	/* rm tmpdir, cpusets, etc */

	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) {	/* MS */
		pjob->ji_qs.ji_substate = JOB_SUBSTATE_DELJOB;
		pjob->ji_sampletim      = time_now;
		/*
		 * The SISTER_KILLDONE flag needs to be reset so
		 * we can talk to the sisterhood and know when they reply.
		 */
		for (i=0; i<pjob->ji_numnodes; i++) {
			hnodent		*np = &pjob->ji_hosts[i];

			if (np->hn_node == pjob->ji_nodeid)	/* me */
				continue;

			if (np->hn_sister == SISTER_KILLDONE)
				np->hn_sister = SISTER_OKAY;
		}
		i = send_sisters(pjob, IM_DELETE_JOB_REPLY, NULL);
		if (i == 0) {
			if (pjob->ji_numnodes > 1) {
				sprintf(log_buffer, "Unable to send delete job "
					"request to one or more sisters");
				log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB,
					LOG_ERR, pjob->ji_qs.ji_jobid, log_buffer);
			}
			/* job is purged here first, discard job happens later
			 * and IM_DISCARD_JOB does not find pjob to kill
			 * job process in case of a mom restart
			 *
			 * Fixing by killing job here, should not hurt in any
			 * case (since we are purging job anyway)
			 */
			(void) kill_job(pjob, SIGKILL);
			job_purge(pjob);
			dorestrict_user();
		}
		/*
		 * otherwise, job_purge() and dorestrict_user() are called in
		 * mom_comm when all the sisters have replied.  The reply to
		 * the Server is also done there
		 */
		return (i);
	} else
		return 0;

}

/**
 * @brief
 * 	rid_job - rid mom of a job that the server says no longer exists
 *
 * @param[in] jobid - char pointer holding jobid
 *
 * @return Void
 *
 */
void
rid_job(char *jobid)
{
	job	*pjob;

	pjob = find_job(jobid);
	if (pjob)
		mom_deljob(pjob);
}

/**
 * @brief
 * 	set_job_toexited - set job substate to exited
 *
 *	Called when a checkpointed job's obit is acknowledged by the server,
 *	prevents a second obit from being sent.
 *
 * @param[in] jobid - char pointer holding jobid
 *
 * @return Void
 *
 */
void
set_job_toexited(char *jobid)
{
	job *pjob;

	pjob = find_job(jobid);
	if (pjob) {
		pjob->ji_qs.ji_substate = JOB_SUBSTATE_EXITED;
		if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHKPT) {
			/* if checkpointed, save state to disk, otherwise  */
			/* leave unchanges on disk so recovery will resend */
			/* obit to server                                  */
			(void)job_save(pjob, SAVEJOB_QUICK);
		}
	}
}
