/*
*         OpenPBS (Portable Batch System) v2.3 Software License
*
* Copyright (c) 1999-2000 Veridian Information Solutions, Inc.
* All rights reserved.
*
* ---------------------------------------------------------------------------
* For a license to use or redistribute the OpenPBS software under conditions
* other than those described below, or to purchase support for this software,
* please contact Veridian Systems, PBS Products Department ("Licensor") at:
*
*    www.OpenPBS.org  +1 650 967-4675                  sales@OpenPBS.org
*                        877 902-4PBS (US toll-free)
* ---------------------------------------------------------------------------
*
* This license covers use of the OpenPBS v2.3 software (the "Software") at
* your site or location, and, for certain users, redistribution of the
* Software to other sites and locations.  Use and redistribution of
* OpenPBS v2.3 in source and binary forms, with or without modification,
* are permitted provided that all of the following conditions are met.
* After December 31, 2001, only conditions 3-6 must be met:
*
* 1. Commercial and/or non-commercial use of the Software is permitted
*    provided a current software registration is on file at www.OpenPBS.org.
*    If use of this software contributes to a publication, product, or
*    service, proper attribution must be given; see www.OpenPBS.org/credit.html
*
* 2. Redistribution in any form is only permitted for non-commercial,
*    non-profit purposes.  There can be no charge for the Software or any
*    software incorporating the Software.  Further, there can be no
*    expectation of revenue generated as a consequence of redistributing
*    the Software.
*
* 3. Any Redistribution of source code must retain the above copyright notice
*    and the acknowledgment contained in paragraph 6, this list of conditions
*    and the disclaimer contained in paragraph 7.
*
* 4. Any Redistribution in binary form must reproduce the above copyright
*    notice and the acknowledgment contained in paragraph 6, this list of
*    conditions and the disclaimer contained in paragraph 7 in the
*    documentation and/or other materials provided with the distribution.
*
* 5. Redistributions in any form must be accompanied by information on how to
*    obtain complete source code for the OpenPBS software and any
*    modifications and/or additions to the OpenPBS software.  The source code
*    must either be included in the distribution or be available for no more
*    than the cost of distribution plus a nominal fee, and all modifications
*    and additions to the Software must be freely redistributable by any party
*    (including Licensor) without restriction.
*
* 6. All advertising materials mentioning features or use of the Software must
*    display the following acknowledgment:
*
*     "This product includes software developed by NASA Ames Research Center,
*     Lawrence Livermore National Laboratory, and Veridian Information
*     Solutions, Inc.
*     Visit www.OpenPBS.org for OpenPBS software support,
*     products, and information."
*
* 7. DISCLAIMER OF WARRANTY
*
* THIS SOFTWARE IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND. ANY EXPRESS
* OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT
* ARE EXPRESSLY DISCLAIMED.
*
* IN NO EVENT SHALL VERIDIAN CORPORATION, ITS AFFILIATED COMPANIES, OR THE
* U.S. GOVERNMENT OR ANY OF ITS AGENCIES BE LIABLE FOR ANY DIRECT OR INDIRECT,
* INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
* OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
* NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
* EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
* This license will be governed by the laws of the Commonwealth of Virginia,
* without reference to its choice of law rules.
*/
/*
 * Functions which provide basic operation on the job structure
 *
 * Included public functions are:
 *
 *   job_abt   abort (remove from server) a job
 *   job_alloc    allocate job struct and initialize defaults
 *   job_free   free space allocated to the job structure and its
 *    childern structures.
 *   svr_job_purge   purge job from server
 *
 *   job_clone    clones a job (for use with job_arrays)
 *   job_clone_wt work task for cloning a job
 *
 * NOTE: for multi-threaded TORQUE, all functions in here except svr_find_job assume that
 * the caller holds any relevant mutexes
 */

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

#include <sys/param.h>
#include <sys/stat.h>
#include <ctype.h>
#include <errno.h>
#include <assert.h>

#ifndef SIGKILL
#include <signal.h>
#endif
#if __STDC__ != 1
#include <memory.h>
#endif

#include <fcntl.h>
#include <stdio.h>
#include <unistd.h>
#include <dirent.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <string>
#include <semaphore.h>

#include "pbs_ifl.h"
#include "list_link.h"
#include "work_task.h"
#include "attribute.h"
#include "resource.h"
#include "server_limits.h"
#include "server.h"
#include "queue.h"
#include "batch_request.h"
#include "pbs_job.h"
#include "log.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "lib_ifl.h"
#include "pbs_error.h"
#include "svrfunc.h"
#include "acct.h"
#include "net_connect.h"
#include "portability.h"
#include "array.h"
#include "pbs_job.h"
#include "svr_func.h" /* get_svr_attr_* */
#include "issue_request.h" /* release_req */
#include "ji_mutex.h"
#include "user_info.h"
#include "mutex_mgr.hpp"
#include "job_route.h" /* job_route */
#include "id_map.hpp"
#include "completed_jobs_map.h"
#include "utils.h"

#ifndef TRUE
#define TRUE 1
#define FALSE 0
#endif

#define MAXLINE 1024
extern int LOGLEVEL;

extern completed_jobs_map_class completed_jobs_map;

int conn_qsub(const char *, long, char *);

/* External functions */

extern void cleanup_restart_file(job *);
extern struct batch_request *setup_cpyfiles(struct batch_request *,job *,char*,char *,int,int);
extern int job_log_open(char *, char *);
extern int log_job_record(const char *buf);
extern void check_job_log(struct work_task *ptask);
int issue_signal(job **, const char *, void(*)(batch_request *), void *, char *);
void handle_complete_second_time(struct work_task *ptask);

/* Local Private Functions */

void free_all_of_job(job *pjob);

/* Global Data items */
all_jobs        alljobs;
extern all_jobs array_summary;

int check_job_log_started = 0;

extern struct server   server;
extern int queue_rank;
extern char *path_arrays;
extern char *msg_abt_err;
extern char *path_jobs;
extern char *path_spool;
extern char *path_aux;
extern char  server_name[];
extern int   LOGLEVEL;

extern char *path_checkpoint;
extern char *path_jobinfo_log;
extern char *log_file;
extern char *job_log_file;

extern sem_t *job_clone_semaphore;

static void send_qsub_delmsg(

  job  *pjob,  /* I */
  const char *text)  /* I */

  {
  char          *phost;
  pbs_attribute *pattri;
  int            qsub_sock;

  phost = arst_string((char *)"PBS_O_HOST", &pjob->ji_wattr[JOB_ATR_variables]);

  if ((phost == NULL) || ((phost = strchr(phost, '=')) == NULL))
    {
    return;
    }

  pattri = &pjob->ji_wattr[JOB_ATR_interactive];

  qsub_sock = conn_qsub(phost + 1, pattri->at_val.at_long, NULL);

  if (qsub_sock < 0)
    {
    return;
    }

  if (write_ac_socket(qsub_sock, "PBS: ", 5) == -1)
    {
    return;
    }

  if (write_ac_socket(qsub_sock, text, strlen(text)) == -1)
    {
    return;
    }

  close(qsub_sock);

  return;
  }  /* END send_qsub_delmsg() */



/*
 * remtree - remove a tree (or single file)
 *
 * returns  0 on success
 *  -1 on failure
 */

static int remtree(

  const char *dirname)

  {
  DIR           *dir;

  struct dirent *pdir;
  char           namebuf[MAXPATHLEN];
  char           log_buf[LOCAL_LOG_BUF_SIZE];
  char          *filnam;
  int            i;
  int            rtnv = 0;

#if defined(HAVE_STRUCT_STAT64) && defined(HAVE_STAT64) && defined(LARGEFILE_WORKS)
  struct stat64  sb;
#else
  struct stat    sb;
#endif

#if defined(HAVE_STRUCT_STAT64) && defined(HAVE_STAT64) && defined(LARGEFILE_WORKS)

  if (lstat64(dirname, &sb) == -1)
#else
  if (lstat(dirname, &sb) == -1)
#endif
    {

    if (errno != ENOENT)
      log_err(errno, __func__, "stat");

    return(-1);
    }

  if (S_ISDIR(sb.st_mode))
    {
    if ((dir = opendir(dirname)) == NULL)
      {
      if (errno != ENOENT)
        log_err(errno, __func__, "opendir");

      return(-1);
      }

    snprintf(namebuf, sizeof(namebuf), "%s/", dirname);

    i = strlen(namebuf);

    filnam = &namebuf[i];

    while ((pdir = readdir(dir)) != NULL)
      {
      if ((pdir->d_name[0] == '.') &&
          ((pdir->d_name[1] == '\0') || (pdir->d_name[1] == '.')))
        continue;

      strcpy(filnam, pdir->d_name);

#if defined(HAVE_STRUCT_STAT64) && defined(HAVE_STAT64) && defined(LARGEFILE_WORKS)
      if (lstat64(namebuf, &sb) == -1)
#else
      if (lstat(namebuf, &sb) == -1)
#endif
        {
        log_err(errno, __func__, "stat");

        rtnv = -1;

        continue;
        }

      if (S_ISDIR(sb.st_mode))
        {
        rtnv = remtree(namebuf);
        }
      else if (unlink(namebuf) < 0)
        {
        if (errno != ENOENT)
          {
          sprintf(log_buf, "unlink failed on %s", namebuf);

          log_err(errno, __func__, log_buf);

          rtnv = -1;
          }
        }
      else if (LOGLEVEL >= 7)
        {
        sprintf(log_buf, "unlink(1) succeeded on %s", namebuf);

        log_ext(-1, __func__, log_buf, LOG_DEBUG);
        }
      }    /* END while ((pdir = readdir(dir)) != NULL) */

    closedir(dir);

    if (rmdir(dirname) < 0)
      {
      if ((errno != ENOENT) && (errno != EINVAL))
        {
        sprintf(log_buf, "rmdir failed on %s", dirname);

        log_err(errno, __func__, log_buf);

        rtnv = -1;
        }
      }
    else if (LOGLEVEL >= 7)
      {
      sprintf(log_buf, "rmdir succeeded on %s", dirname);

      log_ext(-1, __func__, log_buf, LOG_DEBUG);
      }
    }
  else if (unlink(dirname) < 0)
    {
    sprintf(log_buf, "unlink failed on %s", dirname);

    log_err(errno, __func__, log_buf);

    rtnv = -1;
    }
  else if (LOGLEVEL >= 7)
    {
    sprintf(log_buf, "unlink(2) succeeded on %s", dirname);

    log_ext(-1, __func__, log_buf, LOG_DEBUG);
    }

  return(rtnv);
  }  /* END remtree() */



/*
 * handle_aborted_job determines whether to purge the job right away or
 * respect the KeepCompleted time.
 *
 * @param job_ptr - a pointer to the job pointer
 * @param dependentjob - true if the job is a dependent job being deleted because
 *                       the dependency can't be satisfied.
 * @param KeepSeconds - the number of seconds keep_completed is set to
 * @param text - the reason this job is being aborted
*/

void handle_aborted_job(

  job        **job_ptr, /* M */
  bool         dependentjob, /* I */
  long         KeepSeconds, /* I */
  const char  *text) /* I */

  {
  job *pjob = *job_ptr;
  int  rc = svr_setjobstate(pjob, JOB_STATE_COMPLETE, JOB_SUBSTATE_ABORT, FALSE);
    
  if (rc == PBSE_UNKQUE)
    {
    *job_ptr = NULL;
    return;
    }
  else if (rc == PBSE_BAD_JOB_STATE_TRANSITION)
    {
    // This is likely because something else is also cleaning up the job. Just
    // stop and let the other thread do it.
    return;
    }
   
  if ((!dependentjob) ||
      (!KeepSeconds))
    {
    svr_job_purge(pjob);
    *job_ptr = NULL;
    }
  else
    {
    /* Add a comment to say the reason it was aborted */
    if (text != NULL)
      {
      if ((pjob->ji_wattr[JOB_ATR_Comment].at_flags & ATR_VFLAG_SET) &&
          (pjob->ji_wattr[JOB_ATR_Comment].at_val.at_str != NULL))
        {
        std::string new_comment(pjob->ji_wattr[JOB_ATR_Comment].at_val.at_str);
        new_comment += ". ";
        new_comment += text;
        
        free(pjob->ji_wattr[JOB_ATR_Comment].at_val.at_str);
        pjob->ji_wattr[JOB_ATR_Comment].at_val.at_str = strdup(new_comment.c_str());
        }
      else
        {
        pjob->ji_wattr[JOB_ATR_Comment].at_flags |= ATR_VFLAG_SET;
        pjob->ji_wattr[JOB_ATR_Comment].at_val.at_str = strdup(text);
        }
      }

    pjob->ji_wattr[JOB_ATR_exitstat].at_val.at_long = 271;
    pjob->ji_wattr[JOB_ATR_exitstat].at_flags |= ATR_VFLAG_SET;
    set_task(WORK_Immed, KeepSeconds, add_to_completed_jobs, strdup(pjob->ji_qs.ji_jobid), FALSE);
    }
  } /* handle_aborted_job */



/*
 * job_abt - abort a job
 *
 * The job is deleted and a mail message is sent to the job owner.
 * NOTE: The job is always unlocked when this function exits.
 */

/* NOTE:  this routine is called under the following conditions:
 * 1) by req_deletejob whenever deleting a job that is not running,
 *    not transitting, not exiting and does not have a checkpoint
 *    file on the mom.
 * 2) by req_deletearray whenever deleting a job that is not running,
 *    not transitting, not in prerun, not exiting and does not have a
 *    checkpoint file on the mom.
 * 3) by close_quejob when the server fails to enqueue the job.
 * 4) by array_delete_wt for prerun jobs that hang around too long and
 *    do not have a checkpoint file on the mom.
 * 5) by pbsd_init when recovering jobs.
 * 6) by svr_movejob when done routing jobs around.
 * 7) by queue_route when trying toroute any "ready" jobs in a specific queue.
 * 8) by req_shutdown when trying to shutdown.
 * 9) by req_register when the request oparation is JOB_DEPEND_OP_DELETE.
 */

int job_abt(

  struct job **pjobp, /* I (modified/freed) */
  const char  *text,  /* I (optional) */
  bool         dependentjob) /* I */

  {
  char  log_buf[LOCAL_LOG_BUF_SIZE];
  int   old_state;
  int   old_substate;
  int   rc = 0;
  char  job_id[PBS_MAXSVRJOBID+1];
  int   job_exit_status;
  job  *pjob;

  if (pjobp == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input pointer to pointer to job");
    return(rc);
    }

  pjob = *pjobp;

  if (pjob == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input pointer to job");
    return(rc);
    }

  mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);

  long KeepSeconds = 0;
  int rc2 = get_svr_attr_l(SRV_ATR_KeepCompleted, &KeepSeconds);
  if ((rc2 != PBSE_NONE) || (KeepSeconds < 0))
    KeepSeconds = KEEP_COMPLETED_DEFAULT;

  strcpy(job_id, pjob->ji_qs.ji_jobid);

  /* save old state and update state to Exiting */
  old_state = pjob->ji_qs.ji_state;
  old_substate = pjob->ji_qs.ji_substate;

  if (LOGLEVEL >= 6)
    LOG_EVENT(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, pjob->ji_qs.ji_jobid);
  /* notify user of abort if notification was requested */

  if (text != NULL)
    {
    /* req_delete sends own mail and acct record */

    account_record(PBS_ACCT_ABT, pjob, "");
    svr_mailowner(pjob, MAIL_ABORT, MAIL_NORMAL, text);

    if ((pjob->ji_qs.ji_state == JOB_STATE_QUEUED) &&
        ((pjob->ji_wattr[JOB_ATR_interactive].at_flags & ATR_VFLAG_SET) &&
         pjob->ji_wattr[JOB_ATR_interactive].at_val.at_long))
      {
      /* interactive and not yet running... send a note to qsub */

      send_qsub_delmsg(pjob, text);
      }
    }

  if (old_state == JOB_STATE_RUNNING)
    {
    svr_setjobstate(pjob, JOB_STATE_RUNNING, JOB_SUBSTATE_ABORT, FALSE);

    if ((rc = issue_signal(&pjob, "SIGKILL", free_br, NULL, NULL)) != 0)
      {
      if (pjob != NULL)
        {
        sprintf(log_buf, msg_abt_err, pjob->ji_qs.ji_jobid, old_substate);
        
        log_err(-1, __func__, log_buf);
        
        if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0)
          {
          /* notify creator that job is exited */
          
          pjob->ji_wattr[JOB_ATR_state].at_val.at_char = 'E';
          
          issue_track(pjob);
          }
        
        if (pjob->ji_wattr[JOB_ATR_depend].at_flags & ATR_VFLAG_SET)
          {
          if (depend_on_term(pjob) == PBSE_JOBNOTFOUND)
            {
            pjob = NULL;
            pjob_mutex.set_unlock_on_exit(false);
            }
          }
        
        /* update internal array bookeeping values */
        if ((pjob != NULL) &&
            (pjob->ji_arraystructid[0] != '\0') &&
            (pjob->ji_is_array_template == false))
          {
          job_array *pa = get_jobs_array(&pjob);
          
          if (pjob != NULL)
            {
            job_exit_status = pjob->ji_qs.ji_un.ji_exect.ji_exitstat;
            pjob_mutex.unlock();
             
            if (pa)
              {
              pa->update_array_values(old_state,aeTerminate, job_id, job_exit_status);
            
              unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
              }

            if ((pjob = svr_find_job(job_id, TRUE)) != NULL)
              pjob_mutex.mark_as_locked();
            }
          }
      
        if (pjob != NULL)
          handle_aborted_job(&pjob, dependentjob, KeepSeconds, text);
        }
      }
    }
  else if ((old_state == JOB_STATE_TRANSIT) &&
           (old_substate == JOB_SUBSTATE_TRNOUT))
    {
    /* I don't know of a case where this could happen */

    sprintf(log_buf, msg_abt_err,
      pjob->ji_qs.ji_jobid,
      old_substate);
    log_err(-1, __func__, log_buf);
    }
  else
    {
    svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_ABORT, FALSE);

    if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_HERE) == 0)
      {
      /* notify creator that job is exited */

      issue_track(pjob);
      }

    if (pjob->ji_wattr[JOB_ATR_depend].at_flags & ATR_VFLAG_SET)
      {
      strcpy(job_id, pjob->ji_qs.ji_jobid);
      if (depend_on_term(pjob) == PBSE_JOBNOTFOUND)
        {
        pjob = NULL;
        pjob_mutex.set_unlock_on_exit(false);
        }
      /* pjob_mutex managed mutex already points to pjob->ji_mutex. Nothing to do */
      }

    /* update internal array bookeeping values */
    if ((pjob != NULL) &&
        (pjob->ji_arraystructid[0] != '\0') &&
        (pjob->ji_is_array_template == false))
      {
      job_array *pa = get_jobs_array(&pjob);

      if (pjob != NULL)
        {
        job_exit_status = pjob->ji_qs.ji_un.ji_exect.ji_exitstat;
        pjob_mutex.unlock();
        if (pa)
          {
          pa->update_array_values(old_state, aeTerminate, job_id, job_exit_status);
        
          unlock_ai_mutex(pa, __func__,(char *) "1", LOGLEVEL);
          }
        
        if ((pjob = svr_find_job(job_id, TRUE)) != NULL)
          pjob_mutex.mark_as_locked();
        }
      }

    if (pjob != NULL)
      handle_aborted_job(&pjob, dependentjob, KeepSeconds, text);
    }

  if (pjob == NULL)
    pjob_mutex.set_unlock_on_exit(false);

  return(rc);
  }  /* END job_abt() */



/*
 * conn_qsub - connect to the qsub that submitted this interactive job
 * return >= 0 on SUCCESS, < 0 on FAILURE
 * (this was moved from resmom/mom_inter.c)
 */

int conn_qsub(

  const char *hostname,  /* I */
  long        port,      /* I */
  char       *EMsg)      /* O (optional,minsize=1024) */

  {
  pbs_net_t hostaddr;
  int s;

  int flags;
  int local_errno = 0;

  if (hostname == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input hostname pointer");
    return(PBSE_BAD_PARAMETER);
    }

  if (EMsg != NULL)
    EMsg[0] = '\0';

  if ((hostaddr = get_hostaddr(&local_errno, hostname)) == (pbs_net_t)0)
    {
#if !defined(H_ERRNO_DECLARED) && !defined(_AIX)
    extern int h_errno;
#endif

    /* FAILURE */

    if (EMsg != NULL)
      {
      snprintf(EMsg, 1024, "cannot get address for host '%s', h_errno=%d",
               hostname,
               h_errno);
      }

    return(-1);
    }

  s = client_to_svr(hostaddr, (unsigned int)port, 0, EMsg);

  /* NOTE:  client_to_svr() can return 0 for SUCCESS */

  /* assume SUCCESS requires s > 0 (USC) was 'if (s >= 0)' */
  /* above comment not enabled */

  if (s < 0)
    {
    /* FAILURE */

    return(-1);
    }

  /* SUCCESS */

  /* this socket should be blocking */

  flags = fcntl(s, F_GETFL);

  flags &= ~O_NONBLOCK;

  fcntl(s, F_SETFL, flags);

  return(s);
  }  /* END conn_qsub() */



/*
 * job_alloc - allocate space for a job structure and initialize working
 * pbs_attribute to "unset"
 *
 * Returns: pointer to structure or null is space not available.
 *
 * @see job_init_wattr() - child
 */

job *job_alloc(void)

  {
  return(new job());
  }  /* END job_alloc() */



void free_all_of_job(
    
  job *pjob)

  {
  delete pjob;
  } /* END free_all_of_job() */



/*
 * job_free - free job structure and its various sub-structures
 */

void job_free(

  job *pj,
  int  use_recycle)  /* I (modified) */

  {
  char              log_buf[LOCAL_LOG_BUF_SIZE];

  if (pj == NULL)
    {
    return;
    }

  if (LOGLEVEL >= 8)
    {
    sprintf(log_buf, "freeing job");

    log_record(PBSEVENT_DEBUG,PBS_EVENTCLASS_JOB,pj->ji_qs.ji_jobid,log_buf);
    }

  remove_job(&alljobs, pj, true);

  /* move to the recycling structure - deleting right away can cause a race
   * condition where two threads are pending on the same job. Thread 1 gets 
   * the lock and then deletes the job, but thread 2 gets the job's lock as
   * the job is freed, causing segfaults. We use the recycler and the 
   * ji_being_recycled flag to solve this problem --dbeer */
  if (use_recycle)
    {
    insert_into_recycler(pj);
    unlock_ji_mutex(pj, __func__, log_buf, LOGLEVEL);
    }
  else
    {
    unlock_ji_mutex(pj, __func__, log_buf, LOGLEVEL);
    free_all_of_job(pj);
    }

  return;
  }  /* END job_free() */



job *copy_job(

  job       *parent)

  {
  job           *pnewjob;

  int            i;

  if (parent == NULL)
    {
    log_err(errno, __func__, "null parent to copy");

    return(NULL);
    }

  if ((pnewjob = new job()) == NULL)
    {
    log_err(errno, __func__, "no memory");

    return(NULL);
    }

  /* new job structure is allocated,
     now we need to copy the old job, but modify based on taskid */
  pnewjob->ji_modified = 1;   /* struct changed, needs to be saved */

  /* copy the fixed size quick save information */
  memcpy(&pnewjob->ji_qs, &parent->ji_qs, sizeof(struct jobfix));

  /* copy job attributes. some of these are going to have to be modified */
  for (i = 0; i < JOB_ATR_LAST; i++)
    {
    if (parent->ji_wattr[i].at_flags & ATR_VFLAG_SET)
      {
      job_attr_def[i].at_set(
        &(pnewjob->ji_wattr[i]),
        &(parent->ji_wattr[i]),
        SET);
      }
    }

  return(pnewjob);
  } /* END copy_job() */



/*
 * job_clone - create a clone of a job for use with job arrays
 */

job *job_clone(

  job       *template_job, /* I */  /* job to clone */
  job_array *pa,           /* I */  /* array which the job is a part of */
  int        taskid,       /* I */
  bool       place_hold)   // I

  {
  char           log_buf[LOCAL_LOG_BUF_SIZE];

  job           *pnewjob;
  pbs_attribute  tempattr;

  char          *oldid;
  char          *hostname;
  char          *bracket;
  char          *tmpstr;
  char           basename[PBS_JOBBASE+1];
  char           namebuf[MAXPATHLEN + 1];
  char           buf[256];
  int            fds;

  int            i;
  int            slen;
  int            release_mutex = FALSE;
  std::string	 adjusted_path_jobs;

  if (LOGLEVEL >= 7)
    {
    sprintf(log_buf, "taskid %d jobid %s ", taskid, template_job->ji_qs.ji_jobid);
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);
    }

  if (taskid > PBS_MAXJOBARRAY)
    {
    log_err(-1, __func__, "taskid out of range");

    return(NULL);
    }

  if (template_job == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "template_job* is NULL");
    return(NULL);
    }

  if ((pnewjob = new job()) == NULL)
    {
    log_err(errno, __func__, "no memory");

    return(NULL);
    }

  /* new job structure is allocated,
     now we need to copy the old job, but modify based on taskid */
  pnewjob->ji_modified = 1;   /* struct changed, needs to be saved */

  /* copy the fixed size quick save information */
  memcpy(&pnewjob->ji_qs, &template_job->ji_qs, sizeof(struct jobfix));

  /* find the job id for the cloned job */
  if ((oldid = strdup(template_job->ji_qs.ji_jobid)) == NULL)
    {
    log_err(ENOMEM, __func__, "no memory");
    job_free(pnewjob, FALSE);

    return(NULL);
    }

  bracket = index(oldid,'[');
  hostname = index(oldid, '.');

  if (bracket != NULL)
    {
    *bracket = '\0';
    }

  if (hostname != NULL)
    {
    hostname++;
    snprintf(pnewjob->ji_qs.ji_jobid, sizeof(pnewjob->ji_qs.ji_jobid), 
      "%s[%d].%s",
      oldid, taskid, hostname);
    }
  else
    {
    snprintf(pnewjob->ji_qs.ji_jobid, sizeof(pnewjob->ji_qs.ji_jobid),
      "%s[%d]",
      oldid, taskid);
    }

  /* update the job filename
   * We could optimize the sub-jobs to all use the same file. We would need a
   * way to track the number of tasks still using the job file so we know when
   * to delete it.
   */

  /*
   * make up new job file name, it is based on the new jobid
   */

  if (hostname != NULL)
    snprintf(basename, sizeof(basename), "%s-%d.%s", oldid, taskid, hostname);
  else
    snprintf(basename, sizeof(basename), "%s-%d", oldid, taskid);

  free(oldid);

  // get the adjusted path_jobs
  adjusted_path_jobs = get_path_jobdata(pnewjob->ji_qs.ji_jobid, path_jobs);

  do
    {
    snprintf(namebuf, sizeof(namebuf), "%s%s%s",
      adjusted_path_jobs.c_str(), basename, JOB_FILE_SUFFIX);

    fds = open(namebuf, O_CREAT | O_EXCL | O_WRONLY, 0600);

    if (fds < 0)
      {
      if (errno == EEXIST)
        {
        job_free(pnewjob, FALSE);
        return((job *)1); /* TODO: what is this magic for */
        }
      else
        {
        /* FAILURE */

        log_err(errno, __func__, "cannot create job file");
        job_free(pnewjob, FALSE);

        return(NULL);
        }
      }
    }
  while (fds < 0);

  close(fds);

  strcpy(pnewjob->ji_qs.ji_fileprefix, basename);

  /* copy job attributes. some of these are going to have to be modified */

  for (i = 0; i < JOB_ATR_LAST; i++)
    {
    if ((template_job->ji_wattr[i].at_flags & ATR_VFLAG_SET) &&
        (i != JOB_ATR_job_array_request))
      {
      if ((i == JOB_ATR_errpath) || (i == JOB_ATR_outpath) || (i == JOB_ATR_jobname))
        {
        /* modify the errpath and outpath */

        slen = strlen(template_job->ji_wattr[i].at_val.at_str);

        tmpstr = (char*)calloc(sizeof(char), (slen + PBS_MAXJOBARRAYLEN + 1));

        sprintf(tmpstr, "%s-%d",
                template_job->ji_wattr[i].at_val.at_str,
                taskid);

        clear_attr(&tempattr, &job_attr_def[i]);

        job_attr_def[i].at_decode(
          &tempattr,
          NULL,
          NULL,
          tmpstr,
          ATR_DFLAG_ACCESS);

        job_attr_def[i].at_set(
          &pnewjob->ji_wattr[i],
          &tempattr,
          SET);

        job_attr_def[i].at_free(&tempattr);

        free(tmpstr);
        }
      else
        {
        job_attr_def[i].at_set(
          &(pnewjob->ji_wattr[i]),
          &(template_job->ji_wattr[i]),
          SET);
        }
      }
    }

  if (place_hold)
    {
    /* put a system hold on the job.  we'll take the hold off once the
     * entire array is cloned. We don't want any of the jobs to run and
     * complete before the whole thing is cloned. This is in case we run into
     * a problem during setting up the array and want to abort before any of
     * the jobs run */
    pnewjob->ji_wattr[JOB_ATR_hold].at_val.at_long |= HOLD_a;
    pnewjob->ji_wattr[JOB_ATR_hold].at_flags |= ATR_VFLAG_SET;
    }

  /* set JOB_ATR_job_array_id */
  pnewjob->ji_wattr[JOB_ATR_job_array_id].at_val.at_long = taskid;
  pnewjob->ji_wattr[JOB_ATR_job_array_id].at_flags |= ATR_VFLAG_SET;

  /* set PBS_ARRAYID enironment variable */
  clear_attr(&tempattr, &job_attr_def[JOB_ATR_variables]);

  sprintf(buf, "PBS_ARRAYID=%d", taskid);

  job_attr_def[JOB_ATR_variables].at_decode(&tempattr,
      NULL,
      NULL,
      buf,
      0);

  job_attr_def[JOB_ATR_variables].at_set(
    &pnewjob->ji_wattr[JOB_ATR_variables],
    &tempattr,
    INCR);

  job_attr_def[JOB_ATR_variables].at_free(&tempattr);

  /* we need to put the cloned job into the array */
  if (pa == NULL)
    {
    release_mutex = TRUE;
    
    pa = get_array(template_job->ji_qs.ji_jobid);
    if (pa == NULL)
      {
      job_free(pnewjob, FALSE);
      return(NULL);
      }
    }

  pa->job_ids[taskid] = strdup(pnewjob->ji_qs.ji_jobid);
  strcpy(pnewjob->ji_arraystructid, pa->ai_qs.parent_id);

  pnewjob->ji_internal_id = job_mapper.get_new_id(pnewjob->ji_qs.ji_jobid);

  if (release_mutex == TRUE)
    {
    unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
    }

  return(pnewjob);
  } /* END job_clone() */



/*
 * Creates and queues an array subjob for this array
 *
 * @param pa - the array whose subjob we're creating
 * @param array_mgr - the mutex manager for the array
 * @param template_job - the template job for this array
 * @param template_job_mgr - the mutex manager for the template job
 * @param index - the index for the new array sub job
 * @param prev_job_id - we store the new job's id here
 * @return FATAL_ERROR if the array disappears while it was unlocked.
 *         NONFATAL_ERROR if something happens to the subjob
 *         PBSE_NONE on sucess
 */

int create_and_queue_array_subjob(
    
  job_array   *pa,
  mutex_mgr   &array_mgr,
  job         *template_job,
  mutex_mgr   &template_job_mgr,
  int          index,
  std::string &prev_job_id,
  bool         place_hold)

  {
  job         *pjobclone;
  int          newstate;
  int          newsub;
  int          rc = PBSE_NONE;
  std::string  arrayid = pa->ai_qs.parent_id;

  template_job_mgr.lock();
  pjobclone = job_clone(template_job, pa, index, place_hold);
  template_job_mgr.unlock();

  if (pjobclone == NULL)
    {
    log_err(-1, __func__, "unable to clone job in job_clone_wt");
    return(NONFATAL_ERROR);
    }
  else if (pjobclone == (job *)1)
    {
    /* this happens if we attempted to clone an existing job */
    return(PBSE_NONE);
    }

  mutex_mgr clone_mgr(pjobclone->ji_mutex, true);

  svr_evaljobstate(*pjobclone, newstate, newsub, 1);

  /* do this so that  svr_setjobstate() doesn't alter sv_jobstates,
   * these are set later in svr_enquejob() */
  pjobclone->ji_qs.ji_state = newstate;
  pjobclone->ji_qs.ji_substate = newsub;

  svr_setjobstate(pjobclone, newstate, newsub, FALSE);

  pjobclone->ji_wattr[JOB_ATR_qrank].at_val.at_long = ++queue_rank;
  pjobclone->ji_wattr[JOB_ATR_qrank].at_flags |= ATR_VFLAG_SET;

  // Clear this so that the jobs get a queued entry in the accounting file
  pjobclone->ji_wattr[JOB_ATR_qtime].at_flags &= ~ATR_VFLAG_SET;

  array_mgr.unlock();

  if ((rc = svr_enquejob(pjobclone, FALSE, prev_job_id.c_str(), false, false)))
    {
    /* XXX need more robust error handling */
    clone_mgr.set_unlock_on_exit(false);

    if (rc != PBSE_JOB_RECYCLED)
      {
      svr_job_purge(pjobclone);
      }

    if ((pa = get_array(arrayid.c_str())) == NULL)
      {
      sem_wait(job_clone_semaphore);
      return(FATAL_ERROR);
      }

    array_mgr.mark_as_locked();

    return(NONFATAL_ERROR);
    }

  if ((pa = get_jobs_array(&pjobclone)) == NULL)
    {
    if (pjobclone == NULL)
      {
      /* pjobclone has been released. No mutex left to unlock */
      clone_mgr.set_unlock_on_exit(false);
      }

    sem_wait(job_clone_semaphore);
    return(FATAL_ERROR);
    }
    
  array_mgr.mark_as_locked();

  if (job_save(pjobclone, SAVEJOB_FULL, 0) != 0)
    {
    /* XXX need more robust error handling */
    array_mgr.unlock();
    svr_job_purge(pjobclone);
    
    if ((pa = get_array(arrayid.c_str())) == NULL)
      {
      sem_wait(job_clone_semaphore);
      return(FATAL_ERROR);
      }

    array_mgr.mark_as_locked();
    
    return(NONFATAL_ERROR);
    }
    
  prev_job_id = pjobclone->ji_qs.ji_jobid;
  
  pa->ai_qs.num_idle++;
  pa->ai_qs.num_cloned++;

  return(PBSE_NONE);
  } // END create_and_queue_array_subjob()



/*
 * perform_array_postprocessing()
 *
 * Removes the creation hold on array subjobs and applies the slot limit
 *
 * @param pa - the array that needs post processing
 * @return PBSE_NONE
 */

int perform_array_postprocessing(

  job_array *pa)

  {
  int        rc = PBSE_NONE;
  int        newstate;
  int        newsub;
  int        actual_job_count = 0;
  pbs_queue *pque;

  /* scan over all the jobs in the array and unset the hold */
  for (int i = 0; i < pa->ai_qs.array_size; i++)
    {
    if (pa->job_ids[i] == NULL)
      continue;
    
    job *pjob = svr_find_job(pa->job_ids[i], TRUE);

    if (pjob == NULL)
      {
      free(pa->job_ids[i]);
      pa->job_ids[i] = NULL;
      }
    else
      {
      mutex_mgr job_mutex(pjob->ji_mutex,true);
      bool moab_compatible = false;
     
      actual_job_count++;

      get_svr_attr_b(SRV_ATR_MoabArrayCompatible, &moab_compatible);
      pjob->ji_wattr[JOB_ATR_hold].at_val.at_long &= ~HOLD_a;
      
      if (moab_compatible == true)
        {
        /* if configured and necessary, apply a slot limit hold to all
         * jobs above the slot limit threshold */
        if ((pa->ai_qs.slot_limit != NO_SLOT_LIMIT) &&
            (actual_job_count > pa->ai_qs.slot_limit))
          {
          pjob->ji_wattr[JOB_ATR_hold].at_val.at_long |= HOLD_l;
          }
        }
      
      if (pjob->ji_wattr[JOB_ATR_hold].at_val.at_long == 0)
        {
        pjob->ji_wattr[JOB_ATR_hold].at_flags &= ~ATR_VFLAG_SET;
        }
      else
        {
        pjob->ji_wattr[JOB_ATR_hold].at_flags |= ATR_VFLAG_SET;
        }
      
      pjob->ji_modified = TRUE;
      svr_evaljobstate(*pjob, newstate, newsub, 1);
      svr_setjobstate(pjob, newstate, newsub, FALSE);

       /*
       * if the job went into a Route (push) queue that has been started,
       * try once to route it to give immediate feedback as a courtesy
       * to the user.
       */

     if ((pque = get_jobs_queue(&pjob)) != NULL)
       {
       mutex_mgr pque_mutex = mutex_mgr(pque->qu_mutex,true);
       if ((pque->qu_qs.qu_type == QTYPE_RoutePush) &&
           (pque->qu_attr[QA_ATR_Started].at_val.at_long != 0))
         { 
         /* job_route expects the queue to be unlocked */
         pque_mutex.unlock();
         if ((rc = job_route(pjob)))
           {
           if (LOGLEVEL >= 6)
             {
             char log_buf[LOCAL_LOG_BUF_SIZE];
             snprintf(log_buf, sizeof(log_buf), "cannot route job %s", pjob->ji_qs.ji_jobid);
             log_record(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);
             }
           svr_job_purge(pjob);
           pque_mutex.unlock();
           continue;
           }
         }
       }

      if (pjob != NULL) //The call to get_jobs_queue call set pjob to NULL.
        {
        pjob->ji_commit_done = 1;
        }
      }
    }

  return(rc);
  } // END perform_array_postprocessing()



#ifndef CLONE_BATCH_SIZE
#define CLONE_BATCH_SIZE 256
#endif /* CLONE_BATCH_SIZE */



/*
 * job_clone_wt - worktask to clone jobs for job array
 */

void *job_clone_wt(

  void *cloned_id)

  {
  job         *template_job;
  char        *jobid;
  int          rc;
  std::string  prev_job_id;
  char         namebuf[MAXPATHLEN];
  job_array   *pa;

  std::string	 adjusted_path_jobs;

  jobid = (char *)cloned_id;

  if (jobid == NULL)
    {
    log_err(ENOMEM, __func__, "Can't malloc");
    return(NULL);
    }

  /* increment the job_clone_semaphore so people 
     know we are making jobs for this array */
  rc = sem_post(job_clone_semaphore);
  if (rc)
    {
    log_err(-1, __func__, "failed to post job_clone_semaphore");
    return(NULL);
    }

  /* don't call get_jobs_array because the template job isn't part of the array */
  if (((template_job = svr_find_job(jobid, TRUE)) == NULL) ||
      ((pa = get_jobs_array(&template_job)) == NULL))
    {
    free(jobid);

    if (template_job != NULL)
      unlock_ji_mutex(template_job, __func__, "1", LOGLEVEL);
    sem_wait(job_clone_semaphore);
    return(NULL);
    }

  mutex_mgr array_mgr(pa->ai_mutex, true);
  mutex_mgr template_job_mgr(template_job->ji_mutex,true);

  free(jobid);

  // get the adjusted path_jobs path
  adjusted_path_jobs = get_path_jobdata(template_job->ji_qs.ji_jobid, path_jobs);

  snprintf(namebuf, sizeof(namebuf), "%s%s%s",
    adjusted_path_jobs.c_str(), template_job->ji_qs.ji_fileprefix, ARRAY_FILE_SUFFIX);

  template_job_mgr.unlock();

  for (size_t i = 0; i < pa->uncreated_ids.size(); i++)
    {
    if (pa->uncreated_ids[i] == -1)
      continue;

    int index = pa->uncreated_ids[i];
    pa->uncreated_ids[i] = -1;
    pa->ai_qs.highest_id_created = index;

    /* This job already exists. This can happen when trying to recover a job
     * array that wasn't fully cloned. */
    if (pa->job_ids[index] != NULL)
      continue;

    rc = create_and_queue_array_subjob(pa, array_mgr, template_job, template_job_mgr, index,
                                       prev_job_id, true);

    if (rc == FATAL_ERROR)
      {
      return(NULL);
      }

    // If we are over the idle job limit, then stop cloning
    if ((pa->ai_qs.idle_slot_limit != NO_SLOT_LIMIT) &&
        (pa->ai_qs.idle_slot_limit <= pa->ai_qs.num_idle))
      break;
    }  /* END for (i) */

  array_save(pa);

  perform_array_postprocessing(pa);

  unlock_ai_mutex(pa, __func__, "3", LOGLEVEL);
  sem_wait(job_clone_semaphore);
  return(NULL);
  }  /* END job_clone_wt */



/*
 * cpy_checkpoint - set up a Copy Files request to transfer checkpoint files
 */

struct batch_request *cpy_checkpoint(

  struct batch_request *preq,
  job                  *pjob,
  enum job_atr          ati,  /* JOB_ATR_checkpoint_name or JOB_ATR_restart_name */
  int                   direction)

  {
  char            momfile[MAXPATHLEN+1];
  char            serverfile[MAXPATHLEN+1];
  char           *from = NULL;
  char           *to = NULL;
  char            log_buf[LOCAL_LOG_BUF_SIZE];
  pbs_attribute  *pattr;
  mode_t          saveumask = 0;
  
  if (pjob == NULL)
  {
  log_err(PBSE_BAD_PARAMETER, __func__, "null input job pointer");
  return(NULL);
  }

  pattr = &pjob->ji_wattr[ati];

  if ((pattr->at_flags & ATR_VFLAG_SET) == 0)
    {
    /* no file to transfer */
    
    return(preq);
    }
    
  /* build up the name used for SERVER file */
  snprintf(serverfile, sizeof(serverfile), "%s%s%s",
    path_checkpoint, pjob->ji_qs.ji_fileprefix, JOB_CHECKPOINT_SUFFIX);
  
  /*
   * We need to make sure the jobs checkpoint directory exists.  If it does
   * not we need to add it since this is the first time we are copying a
   * checkpoint file for this job
   */

  saveumask = umask(0000);
  if ((mkdir(serverfile, 01777) == -1) && (errno != EEXIST))
    {
    log_err(errno,"cpy_checkpoint", (char *)"Failed to create jobs checkpoint directory");
    }
  umask(saveumask);

  if (sizeof(serverfile) - strlen(serverfile) > 
        strlen(pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str) + 1)
    {
    strcat(serverfile, "/");
    strcat(serverfile, pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
    }

  /* build up the name used for MOM file */

  if (pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_flags & ATR_VFLAG_SET)
    {
    snprintf(momfile, sizeof(momfile), "%s/%s%s/%s",
      pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str,
      pjob->ji_qs.ji_fileprefix,
      JOB_CHECKPOINT_SUFFIX,
      pattr->at_val.at_str);
    }
  else
    {
    /* if not specified, moms path may not be the same */
    snprintf(momfile, sizeof(momfile), "%s/%s%s/%s",
      MOM_DEFAULT_CHECKPOINT_DIR,
      pjob->ji_qs.ji_fileprefix,
      JOB_CHECKPOINT_SUFFIX,
      pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf, "Job has NO checkpoint dir specified, using file %s", momfile);

      log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);
      }
    }

  if (direction == CKPT_DIR_OUT)
    {
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf,"Requesting checkpoint copy from MOM (%s) to SERVER (%s)",
        momfile,
        serverfile);

      log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);
      }
    }
  else
    {
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf,"Requesting checkpoint copy from SERVER (%s) to MOM (%s)",
        serverfile,
        momfile);

      log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);
      }
    }

  to = (char *)calloc(1, strlen(serverfile) + strlen(server_name) + 2);

  if (to == NULL)
    {
    /* FAILURE */

    /* cannot allocate memory for request this one */

    log_event(
      PBSEVENT_ERROR | PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      pjob->ji_qs.ji_jobid,
      (char *)"ERROR:  cannot allocate 'to' memory in cpy_checkpoint");

    return(preq);
    }

  strcpy(to, server_name);
  strcat(to, (char *)":");
  strcat(to, serverfile);

  from = (char *)calloc(1, strlen(momfile) + 1);

  if (from == NULL)
    {
    /* FAILURE */

    log_event(
      PBSEVENT_ERROR | PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      pjob->ji_qs.ji_jobid,
      (char *)"ERROR:  cannot allocate 'from' memory for from in cpy_checkpoint");

    free(to);

    return(preq);
    }

  strcpy(from, momfile);

  if (LOGLEVEL >= 7)
    {
    sprintf(log_buf,"Checkpoint copy from (%s) to (%s)", from, to);

    log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);
    }

  preq = setup_cpyfiles(preq, pjob, from, to, direction, JOBCKPFILE);

  return(preq);
  }  /* END cpy_checkpoint() */




/*
 * remove_checkpoint() - request that mom delete checkpoint file for a job
 * used when the job is to be purged after file has been transferred
 */

void remove_checkpoint(

  job **pjob_ptr)  /* I */

  {
  struct batch_request *preq = NULL;
  char                  log_buf[LOCAL_LOG_BUF_SIZE];
  job                  *pjob = NULL;

  if (pjob_ptr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "null input pointer to job pointer");
    return;
    }

  pjob = *pjob_ptr;

  if (pjob == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "null input job pointer");
    return;
    }

  preq = cpy_checkpoint(preq, pjob, JOB_ATR_checkpoint_name, CKPT_DIR_IN);

  if (preq != NULL)
    {
    /* have files to delete  */
    sprintf(log_buf,"Removing checkpoint file (%s/%s)",
      (*pjob_ptr)->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str,
      (*pjob_ptr)->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);

    log_ext(-1, __func__, log_buf, LOG_DEBUG);

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

    preq->rq_type = PBS_BATCH_DelFiles;

    preq->rq_extra = NULL;
    /* The preq is freed in relay_to_mom (failure)
     * or in issue_Drequest (success) */
    if (relay_to_mom(&pjob, preq, NULL) == PBSE_NONE)
      {
      if (pjob != NULL)
        pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHECKPOINT_COPIED;
      }
    else
      {
      /* log that we were unable to remove the files */

      log_event(
        PBSEVENT_JOB,
        PBS_EVENTCLASS_FILE,
        pjob->ji_qs.ji_jobid,
        (char *)"unable to remove checkpoint file for job");
      }

    free_br(preq);
    }

  return;
  }  /* END remove_checkpoint() */




/*
 * cleanup_restart_file() - request that mom cleanup checkpoint restart file for
 * a job. used when the job has completed or put on hold or deleted
 */

void cleanup_restart_file(

  job *pjob)  /* I */

  {
  if (pjob == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "null input job pointer");
    return;
    }
  
  /* checkpoint restart file cleanup was successful */

/*    pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_COPIED; */

  /* clear restart_name pbs_attribute since purging job will clean it up */

  pjob->ji_wattr[JOB_ATR_restart_name].at_flags &= ~ATR_VFLAG_SET;
  pjob->ji_modified = 1;
  
  job_save(pjob, SAVEJOB_FULL, 0);

  return;
  }  /* END cleanup_restart_file() */





int record_jobinfo(
    
  job *pjob)

  {
  pbs_attribute          *pattr;
  int                     i;
  int                     rc;
  std::string             bf("");
  char                    job_script_buf[(MAXPATHLEN << 4) + 1];
  char                    namebuf[MAXPATHLEN + 1];
  int                     fd;
  size_t                  bytes_read = 0;
  extern pthread_mutex_t  job_log_mutex;
  bool                    record_job_script = false;
  std::string		  adjusted_path_jobs;
  
  if (pjob == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "null input job pointer");
    return(rc);
    }

  pthread_mutex_lock(&job_log_mutex);
  if ((rc = job_log_open(job_log_file, path_jobinfo_log)) < 0)
    {
    pthread_mutex_unlock(&job_log_mutex);
    log_err(rc, __func__, "Could not open job log ");
    return(rc);
    }
  pthread_mutex_unlock(&job_log_mutex);

  bf += "<Jobinfo>\n";
  bf += "\t<Job_Id>";
  bf += pjob->ji_qs.ji_jobid;
  bf += "</Job_Id>\n";

  for (i = 0; i < JOB_ATR_LAST; i++)
    {
    pattr = &(pjob->ji_wattr[i]);

    if (pattr->at_flags & ATR_VFLAG_SET)
      {
      if (!strcmp(job_attr_def[i].at_name, "depend"))
        {
        /* we don't want this pbs_attribute in our log -
           The dependecies will show on the submit_args pbs_attribute */
        continue;
        }

      bf += "\t<";
      bf += job_attr_def[i].at_name;
      bf += ">";

      if (pattr->at_type == ATR_TYPE_RESC)
        bf += "\n";

      attr_to_str(bf, job_attr_def+i, pjob->ji_wattr[i], true);
      
      if (pattr->at_type == ATR_TYPE_RESC)
        bf += "\t";

      bf += "</";
      bf += job_attr_def[i].at_name;
      bf += ">\n";
      }
    }
  
  get_svr_attr_b(SRV_ATR_RecordJobScript, &record_job_script);
  if (record_job_script)
    {
    /* This is for Baylor. We will make it a server parameter eventually
     * Write the contents of the script to our log file*/
    
    bf += "\t<job_script>";

    // get the adjusted path_jobs path  
    adjusted_path_jobs = get_path_jobdata(pjob->ji_qs.ji_jobid, path_jobs);

    snprintf(namebuf, sizeof(namebuf), "%s%s%s",
      adjusted_path_jobs.c_str(), pjob->ji_qs.ji_fileprefix, JOB_SCRIPT_SUFFIX);
    
    if ((fd = open(namebuf, O_RDONLY)) >= 0)
      {
      memset(job_script_buf, 0, sizeof(job_script_buf));

      while ((bytes_read = read_ac_socket(fd, job_script_buf, sizeof(job_script_buf) - 1)) > 0)
        {
        bf += job_script_buf;
        memset(job_script_buf, 0, sizeof(job_script_buf));
        }

      close(fd);
      }
    else
      {
      bf += "unable to open script file\n";
      }
   
    bf += "\t</job_script>\n";
    }

  bf += "</Jobinfo>\n";

  rc = log_job_record(bf.c_str());
      
  return(rc);
  } /* END record_jobinfo() */



/*
 * svr_job_purge - purge job from system
 *
 * pjob->ji_mutex comes into this function locked but 
 * will always exit unlocked.
 * The job is dequeued; the job control file, script file and any spooled
 * output files are unlinked, and the job structure is freed.
 * @param pjob - the job that will be deleted and unlocked
 * @param leaveSpoolFiles - 
 */

int svr_job_purge(

  job *pjob,  /* I (modified) */
  int leaveSpoolFiles) /* I */

  {
  int           rc = PBSE_NONE;
  char          log_buf[LOCAL_LOG_BUF_SIZE];
  char          namebuf[MAXPATHLEN + 1];
  extern char  *msg_err_purgejob;
  time_t        time_now = time(NULL);
  bool          record_job_info = false;
  char          job_id[PBS_MAXSVRJOBID+1];
  char          job_fileprefix[PBS_JOBBASE+1];
  int           job_substate;
  bool          job_is_array_template;
  unsigned int  job_has_checkpoint_file;
  bool          job_has_arraystruct;
  bool          do_delete_array = false;
  job_array     *pa = NULL;
  char          array_id[PBS_MAXSVRJOBID+1];
  std::string	adjusted_path_jobs;
  
  if (pjob == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "null input job pointer fail");
    return(rc);
    }

  mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
 
  strcpy(job_id, pjob->ji_qs.ji_jobid);
  strcpy(job_fileprefix, pjob->ji_qs.ji_fileprefix);
  job_substate = pjob->ji_qs.ji_substate;
  job_is_array_template = pjob->ji_is_array_template;
  job_has_arraystruct = (pjob->ji_arraystructid[0] != '\0');
  job_has_checkpoint_file = pjob->ji_wattr[JOB_ATR_checkpoint_name].at_flags;

  if (LOGLEVEL >= 10)
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, pjob->ji_qs.ji_jobid);

  /* check to see if we are keeping a log of all jobs completed */
  get_svr_attr_b(SRV_ATR_RecordJobInfo, &record_job_info);
  if (record_job_info)
    {
    record_jobinfo(pjob);

    /* Start a task to monitor job log roll over if it is not already started */
    if (check_job_log_started == 0)
      {
      set_task(WORK_Timed, time_now + 10, check_job_log, NULL, FALSE);

      check_job_log_started = 1;
      }
    }

  if ((job_has_arraystruct == false) ||
      (job_is_array_template == true))
    {
    int rc2 = 0;
    if ((rc2 = remove_job(&array_summary, pjob)) == PBSE_JOBNOTFOUND)
      {
      /* PBSE_JOBNOTFOUND means the job is gone.
       * remove_job alreadly unlocked pjob->ji_mutex */
      pjob_mutex.set_unlock_on_exit(false); 
      return(PBSE_NONE);
      }
    else if (rc2 == THING_NOT_FOUND && (LOGLEVEL >= 7))
      {
      snprintf(log_buf,sizeof(log_buf),
          "Could not remove job %s from array_summary\n",
          pjob->ji_qs.ji_jobid);
      log_ext(-1, __func__, log_buf, LOG_WARNING);
      }
    }

  /* if part of job array then remove from array's job list */
  if ((job_has_arraystruct == true) &&
      (job_is_array_template == false))
    {
    /* pa->ai_mutex will come out locked after 
       the call to get_jobs_array */
    pa = get_jobs_array(&pjob);

    if (pjob != NULL)
      {
      if (pa != NULL)
        {
        if (pa->job_ids != NULL)
          {
          free(pa->job_ids[pjob->ji_wattr[JOB_ATR_job_array_id].at_val.at_long]);
          pa->job_ids[pjob->ji_wattr[JOB_ATR_job_array_id].at_val.at_long] = NULL;
          }
        
        /* if there are no more jobs in the array,
         * then we can clean that up too */
        pa->ai_qs.num_purged++;
        if ((pa->ai_qs.num_purged == pa->ai_qs.num_jobs) ||
            ((pa->is_deleted() == true) &&
             (pa->ai_qs.num_idle == 0)))
          {
          /* array_delete will unlock pa->ai_mutex */
          strcpy(array_id, pjob->ji_arraystructid);
          do_delete_array = true;
          }
        else
          array_save(pa);
        
        unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
        }
      }
    else
      {
      pjob_mutex.set_unlock_on_exit(false);
      return(PBSE_JOBNOTFOUND);
      }
    }


  if ((job_substate != JOB_SUBSTATE_TRANSIN) &&
      (job_substate != JOB_SUBSTATE_TRANSICM))
    {
    /* set the state to complete so that svr_dequejob() will function properly */
    pjob->ji_qs.ji_state = JOB_STATE_COMPLETE;
    rc = svr_dequejob(pjob, FALSE);

    if (rc != PBSE_JOBNOTFOUND)
      {
      /* we came out of svr_dequejob with pjob locked. Our pointer is still good */
      /* job_free will unlock the mutex for us.
       * If rc == PBSE_JOB_NOT_IN_QUEUE, it is because another thread is simultaneously
       * deleting this job. We'll quit for them. */
      if ((pjob->ji_being_recycled == false) &&
          (rc != PBSE_JOB_NOT_IN_QUEUE))
        {
        job_free(pjob, TRUE);
        pjob_mutex.set_unlock_on_exit(false);  /* job_free will release lock */
        }
      else
        pjob_mutex.unlock();
      }
    }
  else
    {
    job_free(pjob, TRUE);
    pjob_mutex.set_unlock_on_exit(false); /* job_free will release lock */
    }

  // get the adjusted path_jobs
  //  using the preserved job id in job_id
  adjusted_path_jobs = get_path_jobdata(job_id, path_jobs);

  /* pjob->ji_mutex is unlocked at this point */
  /* delete the script file */
  if ((job_has_arraystruct == false) || 
      (job_is_array_template == true))
    {
    /* delete script file */        
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", adjusted_path_jobs.c_str(),
      job_fileprefix, JOB_SCRIPT_SUFFIX);

    if (unlink(namebuf) < 0)
      {
      if (errno != ENOENT)
        log_err(errno, __func__, msg_err_purgejob);
      }
    else if (LOGLEVEL >= 6)
      {
      sprintf(log_buf, "removed job script");

      log_record(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, job_id, log_buf);
      }
    }

  if (!leaveSpoolFiles)
    {
    /* delete any spooled stdout */
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", path_spool, job_fileprefix, JOB_STDOUT_SUFFIX);

    if (unlink(namebuf) < 0)
      {
      if (errno != ENOENT)
        log_err(errno, __func__, msg_err_purgejob);
      }
    else if (LOGLEVEL >= 6)
      {
      sprintf(log_buf, "removed job stdout");

      log_record(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, job_id, log_buf);
      }

    /* delete any spooled stderr ($TRQ_HOME/spool) */
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", path_spool, job_fileprefix, JOB_STDERR_SUFFIX);

    if (unlink(namebuf) < 0)
      {
      if (errno != ENOENT)
        log_err(errno, __func__, msg_err_purgejob);
      }
    else if (LOGLEVEL >= 6)
      {
      sprintf(log_buf, "removed job stderr");

      log_record(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, job_id, log_buf);
      }
    }

  /* delete checkpoint file directory if there is one */
  if (job_has_checkpoint_file & ATR_VFLAG_SET)
    {
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", 
      path_checkpoint, job_fileprefix, JOB_CHECKPOINT_SUFFIX);

    if (remtree(namebuf) < 0)
      {
      if (errno != ENOENT)
        log_err(errno, __func__, msg_err_purgejob);
      }
    else if (LOGLEVEL >= 6)
      {
      sprintf(log_buf, "removed job checkpoint");

      log_record(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, job_id, log_buf);
      }
    }

  if (job_is_array_template == true)
    {
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", 
      adjusted_path_jobs.c_str(), job_fileprefix, JOB_FILE_TMP_SUFFIX);
    }
  else
    {
    snprintf(namebuf, sizeof(namebuf), "%s%s%s", 
      adjusted_path_jobs.c_str(), job_fileprefix, JOB_FILE_SUFFIX);
    }

  if (unlink(namebuf) < 0)
    {
    if (errno != ENOENT)
      log_err(errno, __func__, msg_err_purgejob);
    }
  else if (LOGLEVEL >= 6)
    {
    sprintf(log_buf, "removed job file");

    log_record(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, job_id, log_buf);
    }

  if (do_delete_array == true)
    array_delete(array_id);

  return(PBSE_NONE);
  }  /* END svr_job_purge() */



/* 
 * Always access the array in this way in order to avoid deadlock 
 *
 * @return this job's array struct with it's mutex locked 
 */

job_array *get_jobs_array(

  job **pjob_ptr)

  {
  char       log_buf[LOCAL_LOG_BUF_SIZE];
  char       jobid[PBS_MAXSVRJOBID + 1];
  char       arrayid[PBS_MAXSVRJOBID + 1];
  job       *pjob = NULL;
  job_array *pa = NULL;

  if (pjob_ptr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input pointer to pointer");
    return(NULL);
    }
  
  pjob = *pjob_ptr;

  if (pjob == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input job pointer");
    return(NULL);
    }

  mutex_mgr job_mutex(pjob->ji_mutex,true);
  job_mutex.set_unlock_on_exit(false);

  strcpy(jobid, pjob->ji_qs.ji_jobid);

  if (pjob->ji_arraystructid[0] != '\0')
    {
    strcpy(arrayid, pjob->ji_arraystructid);
    job_mutex.unlock();
  
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf, "Locking ai_mutex, %s", pjob->ji_qs.ji_jobid);
      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);
      }

    pa = get_array(arrayid);

    pjob = svr_find_job(jobid, TRUE);
    if (pjob == NULL)
      {
      if (pa != NULL)
        {
        unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
        pa = NULL;
        }

      *pjob_ptr = NULL;
      }
    }

  return(pa);
  } /* END get_jobs_array() */



struct pbs_queue *get_jobs_queue(

  job **pjob_ptr)

  {
  job       *pjob = NULL;
  pbs_queue *pque;

  if (LOGLEVEL >= 10)
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, "enter");

  if (pjob_ptr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input pointer to pointer");
    return(NULL);
    }
  pjob = *pjob_ptr;

  if (pjob == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input job pointer");
    return(NULL);
    }

  pque = lock_queue_with_job_held(pjob->ji_qhdr, pjob_ptr);

  if (LOGLEVEL >= 10)
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, "exit");

  return(pque);
  } /* END get_jobs_queue() */



/*
 * Checks if this hostname is in an exec_host-style list
 *
 * @param hostname - the hostname we're looking for
 * @param externals - the exec_host-style list we're searching
 * @return true if found, false otherwise
 */

bool hostname_in_externals(

  const char *hostname,
  const char *externals)

  {
  const char *ptr = strstr(externals, hostname);
  bool        found = false;
  const char *end_word;

  if (ptr != NULL)
    {
    end_word = ptr + strlen(hostname);

    if ((ptr > externals) &&
        (*(ptr - 1) != '+'))
      found = FALSE;
    else if ((*end_word != '\0') &&
             (*end_word != '+'))
      found = false;
    else
      found = true;
    }

  return(found);
  } /* END hostname_in_externals() */




int fix_external_exec_hosts(

  job *pjob) /* the external sub-job */

  {
  std::string     external_execs = "";
  char           *exec_host = pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str;
  char           *externals = pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str;
  char           *exec_ptr;
  char           *plus;
  char           *slash;

  if ((exec_host == NULL) ||
      (externals == NULL))
    return(PBSE_BAD_PARAMETER);

  /* wipe out the current destination */
  pjob->ji_qs.ji_destin[0] = '\0';

  exec_ptr = exec_host;

  while (exec_ptr != NULL)
    {
    /* remove the extra parts after the hostname and get the point we'll advance to.
     * exec_host strings are in the format hostname/index_range[hostname/index_range[+...]] */
    if ((plus = strchr(exec_ptr, '+')) != NULL)
      {
      *plus = '\0';
      plus++;
      }

    if ((slash = strchr(exec_ptr, '/')) != NULL)
      *slash = '\0';

    /* if we find a match, copy in that exec host entry */
    if (hostname_in_externals(exec_ptr, externals) == true)
      {
      /* capture the first external as my mother superior */
      if (pjob->ji_qs.ji_destin[0] == '\0')
        snprintf(pjob->ji_qs.ji_destin, sizeof(pjob->ji_qs.ji_destin), "%s", exec_ptr);

      if (slash != NULL)
        *slash = '/';

      /* delimit with + */
      if (external_execs.length() != 0)
        external_execs += "+";

      external_execs += exec_ptr;
      }

    exec_ptr = plus;
    }

  /* also remove the login attributes */
  if (pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str != NULL)
    {
    free(pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str);
    pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str = NULL;
    pjob->ji_wattr[JOB_ATR_login_node_id].at_flags &= ~ATR_VFLAG_SET;
    }

  if (pjob->ji_wattr[JOB_ATR_login_prop].at_val.at_str != NULL)
    {
    free(pjob->ji_wattr[JOB_ATR_login_prop].at_val.at_str);
    pjob->ji_wattr[JOB_ATR_login_prop].at_val.at_str = NULL;
    pjob->ji_wattr[JOB_ATR_login_prop].at_flags &= ~ATR_VFLAG_SET;
    }

  free(exec_host);
  pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str = strdup(external_execs.c_str());

  return(PBSE_NONE);
  } /* END fix_external_exec_hosts() */




int fix_cray_exec_hosts(

  job *pjob)

  {
  char *external = pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str;
  char *exec = pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str;
  char *new_exec = (char *)calloc(1, strlen(exec) + 1);
  char *exec_ptr = exec;
  char *plus;
  char *slash;

  while (exec_ptr != NULL)
    {
    if ((plus = strchr(exec_ptr, '+')) != NULL)
      {
      *plus = '\0';
      plus++;
      }

    if ((slash = strchr(exec_ptr, '/')) != NULL)
      *slash = '\0';

    /* if the hostname isn't in the externals, copy it in */
    if (hostname_in_externals(exec_ptr, external) == false)
      {
      if (slash != NULL)
        *slash = '/';

      if (*new_exec != '\0')
        strcat(new_exec, "+");

      strcat(new_exec, exec_ptr);
      }

    exec_ptr = plus;
    }

  free(exec);
  pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str = new_exec;

  return(PBSE_NONE);
  } /* END fix_cray_exec_hosts() */




int change_external_job_name(

  job *pjob)

  {
  char  tmp_jobid[PBS_MAXSVRJOBID + 1];
  char *dot = strchr(pjob->ji_qs.ji_jobid, '.');

  if (dot != NULL)
    *dot = '\0';

  snprintf(tmp_jobid, sizeof(tmp_jobid), "%s-0.%s",
    pjob->ji_qs.ji_jobid, dot + 1);

  strcpy(pjob->ji_qs.ji_jobid, tmp_jobid);

  return(PBSE_NONE);
  } /* END change_external_job_name() */




int split_job(

  job *pjob)

  {
  job            *external;
  job            *cray;

  if (pjob->ji_external_clone == NULL)
    {
    external = copy_job(pjob);
    fix_external_exec_hosts(external);
    change_external_job_name(external);
    external->ji_internal_id = job_mapper.get_new_id(external->ji_qs.ji_jobid);
    external->ji_parent_job = pjob;
    pjob->ji_external_clone = external;
    unlock_ji_mutex(external, __func__, NULL, LOGLEVEL);
    }

  if (pjob->ji_cray_clone == NULL)
    {
    cray = copy_job(pjob);
    fix_cray_exec_hosts(cray);
    cray->ji_internal_id    = pjob->ji_internal_id;
    cray->ji_parent_job     = pjob;
    pjob->ji_cray_clone     = cray;
    unlock_ji_mutex(cray, __func__, NULL, LOGLEVEL);
    }

  return(PBSE_NONE);
  } /* END split_job() */



bool job_id_exists(

  const  std::string &job_id_string,
  int   *rcode)

  {
  int ret;
  bool rc = false;

  ret = alljobs.trylock();
  if (ret != 0)
    {
    *rcode = ret;
    return(false);
    }

  *rcode = ret;

  if (alljobs.find(job_id_string) != NULL)
    {
    rc = true;
    }

  alljobs.unlock();

  return(rc);
  } // END job_id_exists()



/*
 * internal_job_id_exists()
 *
 * Checks if a job exists based on the internal id supplied.
 * @return true if a job exists with that internal id, false otherwise
 */

bool internal_job_id_exists(

  int internal_id)

  {
  bool        exists = false;
  const char *job_id = job_mapper.get_name(internal_id);

  if (job_id != NULL)
    {
    alljobs.lock();
    
    if (alljobs.find(job_id) != NULL)
      exists = true;

    alljobs.unlock();
    }

  return(exists);
  } // END internal_job_id_exists()
  

/* END job_func.c */

