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

#include <sstream>
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <grp.h>
#include <limits.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <time.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include "dis.h"
#include "libpbs.h"
#include "pbs_error.h"
#include "server_limits.h"
#include "list_link.h"
#include "credential.h"
#include "attribute.h"
#include "resource.h"
#include "pbs_job.h"
#include "batch_request.h"
#include "mom_comm.h" /* im_compose */
#include "mom_mach.h"
#include "mom_func.h"
#include "mom_hierarchy.h"
#include "log.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "lib_ifl.h"
#include "../lib/Libutils/lib_utils.h"
#include "checkpoint.h" /* start_checkpoint */
#include "resmon.h"
#include "net_connect.h"
#include "utils.h"
#include "alps_functions.h"
#include "tcp.h" /* tcp_chan */
#include "mom_config.h"
#include "power_state.hpp"
#ifdef USE_RESOURCE_PLUGIN
#include "plugin_internal.h"
#include "json/json.h"
#endif

#ifdef _CRAY
#include <sys/category.h>
#endif

#ifdef HAVE_WORDEXP
#include <wordexp.h>

extern int terminate_sisters(job *, int);

extern struct var_table vtable;      /* see start_exec.c */
extern char           **environ;

int reply_send_mom(struct batch_request *request);

extern int InitUserEnv(

    job            *pjob,   /* I */
    task           *ptask,  /* I (optional) */
    char          **envp,   /* I (optional) */
    struct passwd  *pwdp,   /* I (optional) */
    char           *shell);  /* I (optional) */

extern int mkdirtree(

    char *dirpath, /* I */
    mode_t mode);

extern int TMOMJobGetStartInfo(job *, pjobexec_t **) ;

#endif /* HAVE_WORDEXP */

#ifdef NVIDIA_GPUS
int setgpumode(char *, int);
int resetgpuecc(char *, int, int);
#endif /* NVIDIA_GPUS */

/* External Global Data Items */

extern unsigned int alarm_time;
extern unsigned int default_server_port;
extern int  exiting_tasks;
extern char            *msg_err_unlink;
extern char            *path_spool;
extern char            *path_undeliv;
extern attribute_def job_attr_def[];
extern char            *msg_jobmod;
extern char            *msg_manager;
extern time_t  time_now;
extern int  multi_mom;
#ifdef NVIDIA_GPUS
extern int  use_nvidia_gpu;
#endif /* NVIDIA_GPUS */

/* in attr_fn_resc.c */

extern char             MOMUNameMissing[];
extern int              pbs_rm_port;
extern char             path_checkpoint[];

/* Local Data Items */

static uid_t  useruid;
static gid_t  usergid;
static int    ngroup;
static int   *groups;
#if NO_SPOOL_OUTPUT == 0
static char  *output_retained = (char *)"Output retained on that host in: ";
#endif /* !NO_SPOOL_OUTPUT */

static char   rcperr[MAXPATHLEN]; /* file to contain rcp error */

extern int  LOGLEVEL;
extern char checkpoint_run_exe_name[]; 

/* prototypes */

char *get_job_envvar(job *, const char *);
int replace_checkpoint_path(char *);
int in_remote_checkpoint_dir(char *);
int post_stageout(job *pjob, int exit_code);
int send_job_obit(job *pjob, int exit_code);

/* loaded in mom_mach.h */

/* END prototypes */


char *get_job_envvar(

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

  {
  char *pc;

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

  pc = arst_string(

         variable,
         &pjob->ji_wattr[JOB_ATR_variables]);

  if (pc != NULL)
    {
    if ((pc = strchr(pc, (int)'=')) != NULL)
      pc++;
    }

  return(pc);
  }  /* END get_job_envvar() */




/*
 * fork_to_user - fork mom and go to user's home directory
 *    also sets up the global useruid and usergid in the child
 *
 * WARNING: valid only if called when preq points to a cpyfiles structure
 */

static pid_t fork_to_user(

  struct batch_request *preq,   /* I */
  int                   SetUID, /* I (boolean) */
  char                 *HDir,   /* O (job/user home directory) */
  char                 *EMsg)   /* I (optional,minsize=1024) */

  {
  struct group   *grpp;
  pid_t           pid;
  job            *pjob;
  struct passwd  *pwdp = NULL;
  static int      fgrp[NGROUPS_MAX];
  char           *idir;
  char           *pw_buf = NULL;
  char           *gr_buf = NULL;
  std::string     hdir;

  struct stat     sb;

  /* initialize */

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

  if ((pjob = mom_find_job(preq->rq_ind.rq_cpyfile.rq_jobid)) &&
      (pjob->ji_grpcache != 0) &&
      (preq->rq_ind.rq_cpyfile.rq_dir != CKPT_DIR_IN) &&
      (preq->rq_ind.rq_cpyfile.rq_dir != CKPT_DIR_OUT))
    {
    /* use information cached in the job structure */

    useruid = pjob->ji_qs.ji_un.ji_momt.ji_exuid;
    usergid = pjob->ji_qs.ji_un.ji_momt.ji_exgid;
    ngroup  = pjob->ji_grpcache->gc_ngroup;
    groups  = pjob->ji_grpcache->gc_groups;

    if ((idir = get_job_envvar(pjob, (char *)"PBS_O_INITDIR")) != NULL)
      {
      hdir = idir;
      }
    else
      {
      hdir = pjob->ji_grpcache->gc_homedir;
      }
    }
  else
    {

    if ((pwdp = getpwnam_ext(&pw_buf, preq->rq_ind.rq_cpyfile.rq_user)) == NULL)
      {
      if (MOMUNameMissing[0] == '\0')
        snprintf(MOMUNameMissing, 64, "%s", preq->rq_ind.rq_cpyfile.rq_user);

      sprintf(log_buffer, "cannot find user '%s' in password file",
              preq->rq_ind.rq_cpyfile.rq_user);

      if (EMsg != NULL)
        snprintf(EMsg, 1024, "%s", log_buffer);

      log_err(errno, __func__, log_buffer);

      return(-PBSE_BADUSER);
      }

    useruid = pwdp->pw_uid;

    if (preq->rq_ind.rq_cpyfile.rq_group[0] == '\0')
      {
      usergid = pwdp->pw_gid;   /* default to login group */
      }
    else if ((grpp = getgrnam_ext(&gr_buf, preq->rq_ind.rq_cpyfile.rq_group)) != NULL)
      {
      usergid = grpp->gr_gid;
      free_grname(grpp, gr_buf);
      }
    else
      {
      sprintf(log_buffer, "cannot find group '%s' for user '%s' in password file",
              preq->rq_ind.rq_cpyfile.rq_group,
              preq->rq_ind.rq_cpyfile.rq_user);

      if (EMsg != NULL)
        snprintf(EMsg, 1024, "%s", log_buffer);

      log_err(errno, __func__, log_buffer);

      free_pwnam(pwdp, pw_buf);
      return(-PBSE_BADUSER);
      }

    ngroup = init_groups(pwdp->pw_name, usergid, NGROUPS_MAX, fgrp);

    if (ngroup < 0)
      ngroup = 0;

    groups = fgrp;

    /* determine user`s home directory */

    if ((pjob != NULL) && ((idir = get_job_envvar(pjob, (char *)"PBS_O_INITDIR")) != NULL))
      {
      hdir = idir;
      }
    else
      {
      hdir = pwdp->pw_dir;
      }

    free_pwnam(pwdp, pw_buf);
    }    /* END if ((pjob = mom_find_job(preq->rq_ind.rq_cpyfile.rq_jobid)) && ...) */

  if (hdir.size() == 0)
    {
    /* FAILURE */

    log_err(PBSE_UNKRESC, __func__, (char *)"cannot determine home directory");

    if (EMsg != NULL)
      snprintf(EMsg, 1024, "%s", "cannot determine home directory");

    return(-PBSE_UNKRESC);
    }

  /* check user home directory as root and log failures */

  /* NOTE: root may not be able to see user home directory due to root_squash
           permissions but routine will test and log what is detected. Some
           issues will be logged by the parent but TORQUE will only fail if the
           problems persist in the child after the setuid() call */

  if (stat(hdir.c_str(), &sb) != 0)
    {
    sprintf(log_buffer, "Root cannot open home directory '%s' specified, errno=%d (%s) -- Ignore if root squashing is enabled",
            hdir.c_str(),
            errno,
            strerror(errno));

    if (LOGLEVEL >= 2)
      {
      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
      }

    if (EMsg != NULL)
      snprintf(EMsg, 1024, "%s", log_buffer);

    /* NOTE:  warn only, root may not be able to stat directory */

    /* return(-PBSE_UNKRESC); */
    }
  else if (!S_ISDIR(sb.st_mode))
    {
    sprintf(log_buffer, "invalid home directory '%s' specified, not a directory",
      hdir.c_str());

    log_err(PBSE_UNKRESC, __func__, log_buffer);

    if (EMsg != NULL)
      snprintf(EMsg, 1024, "%s", log_buffer);

    return(-PBSE_UNKRESC);
    }

  if (LOGLEVEL >= 1)
    {
    sprintf(log_buffer, "forking to user, uid: %ld  gid: %ld  homedir: '%s'",
            (long)useruid,
            (long)usergid,
            hdir.c_str());

    log_record(
      PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      (pjob != NULL) ? pjob->ji_qs.ji_jobid : "N/A",
      log_buffer);
    }

  if (HDir != NULL)
    strcpy(HDir, hdir.c_str());

  pid = fork_me(preq->rq_conn);

  if (pid < 0)
    {
    log_err(errno, __func__, "Fork failed.");

    if (EMsg != NULL)
      snprintf(EMsg, 1024, "%s", log_buffer);

    return(-PBSE_SYSTEM);
    }

  if (pid > 0)
    {
    /* parent - note leave connection open */

    free_br(preq);

    return(pid);
    }

  /* the child */

  /* NOTE:  writing to log as child may cause corruption */

#ifdef _CRAY

  if ((pjob = mom_find_job(preq->rq_ind.rq_cpyfile.rq_jobid)) &&
      (pjob->ji_grpcache != 0))
    {
    /* set account id */

    if (pjob->ji_wattr[JOB_ATR_account].at_flags & ATR_VFLAG_SET)
      {
      acctid(0, nam2acid(pjob->ji_wattr[JOB_ATR_account].at_val.at_str));
      }
    }

#endif /* _CRAY */

  /* NOTE:  only chdir now if SetUID is TRUE */

  if (SetUID == TRUE)
    {
    if (setgroups(ngroup,(gid_t *)groups) != 0 ||
        setgid(usergid) != 0)
      {
      /* unable to set user groups */

      return(-PBSE_BADGRP);
      }

    if (setuid_ext(useruid, FALSE) == -1)
      {
      /* cannot run as the user */

      return(-PBSE_BADUSER);
      }

    if (chdir(hdir.c_str()) == -1)
      {
      /* cannot change directory to user home dir (or 'INITDIR' if specified) */

      return(-PBSE_UNKRESC);
      }
    }

#ifdef HAVE_WORDEXP
    {
    int rc;

    /* set some useful env variables */

    rc = put_env_var("HOME", hdir.c_str());
    if (rc)
      {
      sprintf(log_buffer, "put_env_var failed with %d", rc);

      log_err(errno, __func__, log_buffer);

      if (EMsg != NULL)
        snprintf(EMsg, 1024, "%s", log_buffer);

      return(-PBSE_SYSTEM);
      }

    rc = put_env_var("PBS_JOBID", preq->rq_ind.rq_cpyfile.rq_jobid);
    if (rc)
      {
      sprintf(log_buffer, "put_env_var failed with %d", rc);

      log_err(errno, __func__, log_buffer);

      if (EMsg != NULL)
        snprintf(EMsg, 1024, "%s", log_buffer);

      return(-PBSE_SYSTEM);
      }
    }
#endif /* END HAVE_WORDEXP */

  return(pid);
  }  /* END fork_to_user() */





/*
 * add_bad_list -  add bad file message to bad file list
 */

static void add_bad_list(

  char **pbl,
  char  *newtext,
  int    nl)

  {
  int   needed = 0;
  char *pnew;

  if (*pbl != NULL)
    {
    needed += strlen(*pbl) + strlen(newtext) + nl + 1;

    pnew = (char *)realloc(*pbl, needed);
    }
  else
    {
    needed += strlen(newtext) + nl + 1;

    pnew = (char *)calloc(1, needed);

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

  if (pnew == NULL)
    {
    /* cannot allocate memory - FAILURE */

    return;
    }

  *pbl = pnew;

  while (nl--)    /* prefix new-lines */
    strcat(*pbl, "\n");

  strcat(*pbl, newtext);

  return;
  }  /* END add_bad_list() */





#define RT_BLK_SZ 4096

int return_file(

  job           *pjob,
  enum job_file  which,
  int            sock,
  int            remove_file)

  {
  int                   amt;
  char                  buf[RT_BLK_SZ];
  int                   fds;
  char                 *filename;

  struct batch_request *prq;
  int                   rc = 0;
  int                   seq = 0;
  struct tcp_chan      *chan = NULL;

  filename = std_file_name(pjob, which, &amt); /* amt is place holder */

  /* We need to check for NULL which may be returned */

  if (filename == NULL)
    {
    return(-1);
    }

  if (strcmp(filename, "/dev/null") == 0)
    {
    return(0);
    }

  prq = alloc_br(PBS_BATCH_MvJobFile);

  if (prq == NULL)
    {
    /* no memory */

    return(PBSE_SYSTEM);
    }

  fds = open(filename, O_RDONLY, 0);

  if (fds < 0)
    {
    return(errno);
    }

  strcpy(prq->rq_host, mom_host);

  strcpy(prq->rq_ind.rq_jobfile.rq_jobid, pjob->ji_qs.ji_jobid);

  while ((amt = read_ac_socket(fds, buf, RT_BLK_SZ)) > 0)
    {
    /* prq->rq_ind.rq_jobfile.rq_sequence = seq++; */
    /* prq->rq_ind.rq_jobfile.rq_type = (int)which; */
    /* prq->rq_ind.rq_jobfile.rq_size = amt; */
    /* prq->rq_ind.rq_jobfile.rq_data = buf; */

    if ((chan = DIS_tcp_setup(sock)) == NULL)
      {
      break;
      }
    else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_MvJobFile, pbs_current_user)) ||
             (rc = encode_DIS_JobFile(chan, seq++, buf, amt, pjob->ji_qs.ji_jobid, which)) ||
             (rc = encode_DIS_ReqExtend(chan, NULL)))
      {
      DIS_tcp_cleanup(chan);
      chan = NULL;

      break;
      }

    DIS_tcp_wflush(chan);

    if ((DIS_reply_read(chan, &prq->rq_reply) != 0) ||
        (prq->rq_reply.brp_code != 0))
      {
      rc = -1;

      DIS_tcp_cleanup(chan);
      chan = NULL;

      break;
      }

    DIS_tcp_cleanup(chan);
    chan = NULL;
    }    /* END while ((amt = read()) */

  free_br(prq);

  close(fds);

  if (remove_file == TRUE && rc == 0)
    job_unlink_file(pjob, filename);

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





/*
 * wchost_match - wild card host name match
 *
 * return 1 if can"idate" matches master name
 *  0 if not a match
 * master name may be wild carded at beginning
 */

static int wchost_match(

  const char *can,    /* I candidate */
  const char *master) /* I */

  {
  const char *pc;
  const char *pm;

  /* FORMAT:   master = [*]<SUFFIX>, candidate = <STRING><SUFFIX> */

  /* start at end and work backwards */

  pc = can + strlen(can) - 1;

  pm = master + strlen(master) - 1;

  while ((pc > can) && (pm > master))
    {
    if (*pc != *pm)
      {
      /* no match */

      return(0);
      }

    pc--;

    pm--;
    }

  /* comparison of one or both reached the start of the string */

  if (pm == master)
    {
    if (*pm == '*')
      {
      /* reached master wild card, all candidates match */

      return(1);
      }

    if ((pc == can) && (*pc == *pm))
      {
      /* start of candidate reached, always match */

      /* NOTE: this seems wrong (CRI)  ie, can=dog.com master=*.bigdog.com */

      return(1);
      }
    }

  /* no match */

  return(0);
  }  /* END wchost_match() */




/* Determine which local path to copy to if specified using $usecp */

static int told_to_cp(

  char  *host,    /* I */
  char  *oldpath, /* I */
  char **newpath) /* O */

  {
  int    i;
  unsigned int    nh;

  static char newp[MAXPATHLEN + 1];
  char linkpath[MAXPATHLEN + 1];
  int max_links;

  for (max_links = 16;max_links > 0;max_links--)
    {
    for (nh = 0;nh < pcphosts.size(); nh++)
      {
      if (wchost_match(host, pcphosts[nh].cph_hosts.c_str()))
        {

        if (LOGLEVEL >= 5)
          {
          sprintf(log_buffer, "host '%s' pcphosts[%d].cph_hosts: %s",
                  host,
                  nh,
                  pcphosts[nh].cph_hosts.c_str());

          log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, log_buffer);
          }

        i = pcphosts[nh].cph_from.size();

        if (strncmp(pcphosts[nh].cph_from.c_str(), oldpath, i) == 0)
          {
          int nchars, link_size;
          nchars = snprintf(newp, sizeof(newp), "%s%s",
                            pcphosts[nh].cph_to.c_str(), oldpath + i);

          if (nchars >= (int)sizeof(newp))
            {
            snprintf(log_buffer, sizeof(log_buffer),
                     "too long string when transforming path '%s' to '%s%s'\n",
                     oldpath, pcphosts[nh].cph_to.c_str(), oldpath + i);
            log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, log_buffer);
            return(0);
            }

          link_size = readlink((const char *)newp,

                               linkpath, sizeof(linkpath) - 1);

          if (link_size == -1)
            {
            /*
             * Catching only too many symbolic links, bad buffer
             * location and insufficient kernel memory cases.
             */
            if (errno == ELOOP || errno == EFAULT || errno == ENOMEM)
              {
              snprintf(log_buffer, sizeof(log_buffer),
                       "translation of symbolic link '%s' failed: %s\n",
                       newp, strerror(errno));
              log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, log_buffer);
              return(0);
              }

            /*
             * We're done.  All other errors (if any)  will be
             * reported in the respective routines.
             *
             * At least ENOENT and EINVAL are good error codes:
             * they correspond to non-existent object or to
             * object that is not a symbolic link.
             */
            else
              {
              *newpath = newp;
              /* success */
              return(1);
              }
            }
          else
            {
            linkpath[link_size] = '\0';
              {
              snprintf(log_buffer, sizeof(log_buffer),
                       "translated symbolic link '%s:%s' to '%s:%s'; "
                       "restarting $usecp search\n",
                       host, newp, host, linkpath);
              log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, log_buffer);
              }

            oldpath = linkpath;
            }

          break;
          }
        }

      if (LOGLEVEL >= 5)
        {
        sprintf(log_buffer, "host '%s' and path '%s' does not match usecp[%d]  (host '%s' path '%s')\n",
                host,
                oldpath,
                nh,
                pcphosts[nh].cph_hosts.c_str(),
                pcphosts[nh].cph_from.c_str());

        log_record(
          PBSEVENT_SYSTEM,
          PBS_EVENTCLASS_SERVER,
          __func__,
          log_buffer);
        }
      }    /* END for (nh) */
    }      /* END for (max_links) */

  /* failure */

  if (LOGLEVEL >= 3)
    {
    sprintf(log_buffer, "no local path matches host '%s' and path '%s' (%d paths checked)\n",
            host,
            oldpath,
            nh);

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

  return(0);
  }  /* END told_to_cp() */





/*
 * is_remote() - is the specified path to a local or remote file
 * checks to see if there is a hostname which matches this host
 *
 * @return true if path is remote or false if local
 */

bool is_remote(

  char **path)  /* I */

  {
  int   len;
  char *pcolon;

  pcolon = strchr(*path, (int)':');

  if (pcolon == NULL)
    {
    /* local file */

    return(false);
    }

  *pcolon = '\0';

  len = strlen(*path);

  if ((strncmp("localhost", *path, 9) == 0) ||
      ((strncmp(mom_host, *path, len) == 0) &&
       ((mom_host[len] == '\0') || (mom_host[len] == '.'))))
    {
    /* we have a host match, file is local */

    *pcolon = ':';

    *path = pcolon + 1;

    /* local file */

    return(false);
    }
  else if (told_to_cp(*path, pcolon + 1, path))
    {
    /* path updated in told_to_cp() */

    /* local file */

    return(false);
    }

  /* remote file */

  *pcolon = ':';

  return(true);
  }  /* END is_remote() */






/*
 * is_file_same() - are two paths pointing to the same file
 * returns: 1 if are the same
 *   0 if not the same (or cannot tell)
 */

static int is_file_same(

  char *file1,
  char *file2)

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

  struct stat64 sb1, sb2;
#else

  struct stat sb1, sb2;
#endif

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

  if ((stat64(file1, &sb1) == 0) && (stat64(file2, &sb2) == 0))
#else
  if ((stat(file1, &sb1) == 0) && (stat(file2, &sb2) == 0))
#endif
    {

    if (!memcmp(&sb1.st_dev, &sb2.st_dev, sizeof(dev_t)) &&
    !memcmp(&sb1.st_ino, &sb2.st_ino, sizeof(ino_t)))
      {
      /* files are same */

      return(1);
      }
    }

  return(0);
  } /* End of is_file_same() */




/*
 * is_file_going_to_dir() - is file going to destination directory
 * returns: 1 if file is going to directory
 *   0 if not going to the directory or destiation is not a directory (or cannot tell)
 */

static int is_file_going_to_dir(

  char *file,
  char *destdir)

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

  struct stat64 sb1;
#else

  struct stat sb1;
#endif

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

  if (stat64(destdir, &sb1) == 0)
#else

  if (stat(destdir, &sb1) == 0)
#endif
    {
    char *ptr1;
    char filename[MAXPATHLEN+1];
    int complen = 0;

    /* Make sure the destination is a directory */

    if (!S_ISDIR(sb1.st_mode))
      {
      /* destination is not a directory */
      return(0);
      }

    snprintf(filename, sizeof(filename), "%s", file);

    /* Does directory match the files path? */

    ptr1 = strrchr(filename, '/');
    if (ptr1 != NULL)
      {
      ptr1[0] = '\0';

      complen = strlen(destdir);
      if (destdir[complen - 1] == '/')
        {
        /* don't include trailing slash (if any) in comparision */
        complen--;
        }

      if (memcmp(filename, destdir, complen) == 0)
        {
        /* file is going to directory*/
        return(1);
        }
      }
    }
  else if (errno == 2)
    {
    /*
     * This is okay. Probably a file that does not yet exist because
     * we have not copied it yet
     */
    }
  else
    {
    sprintf(log_buffer, "File %s stat failed, errno = %d",
            destdir,
            errno);
    log_err(errno, __func__, log_buffer);
    }

  return(0);
  } /* End of is_file_going_to_dir() */





void req_deletejob(

  struct batch_request *preq)  /* I */

  {
  job        *pjob;
  pjobexec_t *TJE = NULL;
  char        log_buf[LOCAL_LOG_BUF_SIZE];

  pjob = mom_find_job(preq->rq_ind.rq_delete.rq_objname);

  if (pjob != NULL)
    {
    snprintf(log_buf, sizeof(log_buf), "%s", pjob->ji_qs.ji_jobid);
    if (LOGLEVEL >= 3)
      {
      log_record(
        PBSEVENT_JOB,
        PBS_EVENTCLASS_JOB,
        __func__,
        log_buf);
      }

    /*
     * We need to clear out the TJE starter slot if this job is in it.  This
     * can occur if we qdel the job while the prologue is running.  If we
     * don't remove ourself here then the TJE slot may remain with data that is
     * no longer valid, yet we will reuse the slot since it only matches on the
     * address of the pjob not on a jobid.  This can lead to unexpected crashes.
     */
    if (TMOMJobGetStartInfo(pjob, &TJE) == SUCCESS)
      {
      free_pwnam(static_cast<struct passwd *>(TJE->pwdp), TJE->buf);
      memset(TJE, 0, sizeof(pjobexec_t));
      }

    /* assume success? */

    mom_deljob(pjob);

    reply_ack(preq);
    }
  else
    {
    req_reject(PBSE_UNKJOBID, 0, preq, mom_host, (char *)"cannot locate job to delete");
    }

  return;
  }  /* END req_deletejob() */





/*
 * req_holdjob - checkpoint and terminate job
 */

void mom_req_holdjob(

  struct batch_request *preq)

  {
  int  rc;
  job *pjob;
  svrattrl *pal;
  pbs_attribute tmph;

  /* If checkpoint supported, do it and terminate the job */
  /* otherwise, return PBSE_NOSUP    */

  if ((pjob = mom_find_job(preq->rq_ind.rq_hold.rq_orig.rq_objname)) == NULL)
    {
    rc = PBSE_UNKJOBID;
    }
  else
    {
    /* propagate servers hold state to job */

    clear_attr(&tmph, &job_attr_def[JOB_ATR_hold]);

    if ((pal = (svrattrl *)GET_NEXT(preq->rq_ind.rq_hold.rq_orig.rq_attr)) != NULL)
      {
      job_attr_def[JOB_ATR_hold].at_decode(
        &tmph,
        pal->al_name,
        NULL,
        pal->al_value,
        0);
      }

    if ((rc = start_checkpoint(pjob,1,preq)) != PBSE_NONE)
      req_reject(rc,0,preq,mom_host,(char *)"cannot checkpoint job");    /* unable to start checkpoint */
    }

  return;
  }  /* END req_holdjob() */





/*
 * req_checkpointjob - checkpoint and continue job
 */

void req_checkpointjob(

  struct batch_request *preq)

  {
  int  rc;
  job *pjob;

  /* If checkpoint supported, do it and terminate the job */
  /* otherwise, return PBSE_NOSUP    */

  pjob = mom_find_job(preq->rq_ind.rq_manager.rq_objname);

  if (pjob == NULL)
    {
    rc = PBSE_UNKJOBID;
    req_reject(rc, 0, preq, mom_host, (char *)"job does not exist on mom");
    }
  else
    {
    if ((rc = start_checkpoint(pjob, 0, preq)) != PBSE_NONE)
      req_reject(rc, 0, preq, mom_host, (char *)"cannot checkpoint job");    /* unable to start checkpoint */
    }

  /* note, normally the reply to the server is in start_checkpoint() */
  }


/*
 *  Write text into a job's output file,
 *  Return a PBS error code.
 */

int message_job(

  job            *pjob,
  enum job_file   jft,  /* I */
  char           *text)

  {
  char  *pstr = NULL;
  int    len;
  int    fds;
  int    rc;

  if (pjob == NULL)
    {
    return(PBSE_UNKJOBID);
    }

  /* must be Mother Superior for this to make sense */
  if (am_i_mother_superior(*pjob) == false)
    {
    log_err(errno, __func__, "cannot message job, not mother superior");

    return(PBSE_MOMREJECT);
    }

  /* to prevent a security issue and also handle root-squashing slightly better,
   * fork and become the user before doing this. */
  if ((rc = fork_me(-1)) == -1)
    {
    /* ERROR */
    log_err(errno, __func__, "cannot fork child");
    return(PBSE_SYSTEM);
    }
  else if (rc > 0)
    {
    /* parent */
    return(PBSE_NONE);
    }

  /* only the child reaches here, become the user for root-squashing as well */
  if (become_the_user(pjob) != PBSE_NONE)
    {
    /* log_buffer is populated by become_the_user */
    log_err(errno, __func__, log_buffer);
    exit(PBSE_BADUSER);
    }

  len = is_joined(pjob);

  if (len == -1)
    jft = StdErr; /* only have stderr open */
  else if (len == 1)
    jft = StdOut; /* only have stdout open */

  if ((fds = open_std_file(
               pjob,
               jft,
               O_WRONLY | O_APPEND,
               pjob->ji_qs.ji_un.ji_momt.ji_exgid)) < 0)
    {
    sprintf(log_buffer, "cannot open %s file for job '%s' (msg: '%.64s')",
            (jft == StdErr) ? "stderr" : "stdout",
            pjob->ji_qs.ji_jobid,
            text);

    log_err(errno, __func__, log_buffer);

    exit(PBSE_MOMREJECT);
    }

  len = strlen(text);

  if (text[len - 1] != '\n')
    {
    if ((pstr = (char *)calloc(1, len + 2)) == NULL)
      {
      close(fds);

      exit(PBSE_INTERNAL);
      }

    strcpy(pstr, text);

    pstr[len++] = '\n'; /* append new-line */

    text = pstr;
    }

  rc = PBSE_NONE;

  alarm(alarm_time);
  if (write_ac_socket(fds, text, len) != len)
    {
    log_err(errno, __func__, "unable to write message to job");

    rc = PBSE_INTERNAL;
    }
  alarm(0);

  if (close(fds) != 0)
    {
    log_err(errno, __func__, "unable to write message to job");

    rc = PBSE_INTERNAL;
    }

  if (pstr != NULL)
    free(pstr);

  /* SUCCESS */
  exit(EXIT_SUCCESS);
  }  /* END message_job() */





/*
 * req_messagejob - Append message to job's output/error file
 */

void req_messagejob(

  struct batch_request *preq)

  {
  int   ret = 0;
  job  *pjob;

  pjob = mom_find_job(preq->rq_ind.rq_message.rq_jid);

  if ((preq->rq_ind.rq_message.rq_file == PBS_BATCH_FileOpt_Default) ||
      (preq->rq_ind.rq_message.rq_file & PBS_BATCH_FileOpt_OFlg))
    {
    ret = message_job(pjob, StdOut, preq->rq_ind.rq_message.rq_text);
    }

  if ((preq->rq_ind.rq_message.rq_file & PBS_BATCH_FileOpt_EFlg) &&
      (ret == 0))
    {
    ret = message_job(pjob, StdErr, preq->rq_ind.rq_message.rq_text);
    }

  if (ret == PBSE_NONE)
    {
    /* message request successful */

    reply_ack(preq);
    }
  else
    {
    req_reject(ret, 0, preq, mom_host, "cannot add message to job output/error buffer");
    }
  }  /* END req_messagejob() */




const char *TJobAttr[] =
  {
  "jobname",        /* this set appears first as they show */
  "job_owner",      /* in a basic job status display       */
  "resc_used",
  "state",
  "in_queue",
  "at_server",      /* 5 */
  "account",        /* the bulk of the attributes are in   */
  "chkpnt",         /* alphabetic order for no good reason */
  "ctime",
  "depend",
  "errpath",
  "exec_host",
  "exectime",
  "grouplst",
  "hold",
  "interactive",
  "join",
  "keep",
  "mailpnts",
  "mailuser",
  "mtime",          /* 20 */
  "outpath",
  "priority",
  "qtime",
  "rerunable",
  "resource",
  "session_id",
  "shell",
  "stagein",
  "stageout",
  "substate",
  "userlst",
  "variables",      /* 32 */
  "euser",          /* execution user name for MOM            */
  "egroup",         /* execution group name for MOM           */
  "hashname",       /* job name hashed into 14 characters     */
  "hopcount",
  "qrank",
  "queuetype",
  "schedhint",      /* 39 */
  "security",
  "comment",
  "cookie",
  "altid",
  "etime",
  "exitstat",
  "forwardx11",
  "submitargs",
  "jobarrayid",
  "jobarrayreq",
  "umask",          /* 50 */
  "start_time",
  "start_count",
  "chkptdir",
  "chkptname",
  "chkpttime",
  "restartstat",
  "restartname",
  "faulttol",
  "comp_time",
  "reported",       /* 60 */
  "job_type",
  "inter_cmd",
  "proxy_user",
  "exclusive_access",
#ifdef USEJOBCREATE
  "pagg_id",
#endif /* USEJOBCREATE */
#ifdef NVIDIA_GPUS
  "gpu_flags",
#endif  /* NVIDIA_GPUS */
  "job_id",
  NULL
  };




/*
 * req_modifyjob - service the Modify Job Request
 *
 * This request modifies a job's attributes.
 */

void req_modifyjob(

  struct batch_request *preq)  /* I */

  {
  int            bad = 0;
  int            i;
  pbs_attribute  newattr[JOB_ATR_LAST];
  pbs_attribute *pattr;
  job           *pjob;
  svrattrl      *plist;
  int            rc;
  unsigned int   momport = 0;

  char           tmpLine[1024];

  pjob = mom_find_job(preq->rq_ind.rq_modify.rq_objname);

  if (pjob == NULL)
    {
    sprintf(tmpLine, "modify job failed, unknown job %s",
            preq->rq_ind.rq_modify.rq_objname);

    req_reject(PBSE_UNKJOBID, 0, preq, mom_host, tmpLine);

    return;
    }

  if (LOGLEVEL >= 3)
    {
    log_record(
      PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      pjob->ji_qs.ji_jobid,
      (char *)"modifying job");
    }

  plist = (svrattrl *)GET_NEXT(preq->rq_ind.rq_modify.rq_attr);

  if (plist == NULL)
    {
    /* nothing to do */

    reply_ack(preq);

    /* SUCCESS */

    return;
    }

  /* modify the jobs attributes */

  bad = 0;

  pattr = pjob->ji_wattr;

  /* call attr_atomic_set to decode and set a copy of the attributes */

  rc = attr_atomic_set(
         plist,
         pattr,
         newattr,
         job_attr_def,
         JOB_ATR_LAST,
         -1,
         ATR_DFLAG_MGWR | ATR_DFLAG_MOM,
         &bad);

  if (rc != 0)
    {
    /* FAILURE - leave old values, free the new ones */

    for (i = 0;i < JOB_ATR_LAST;i++)
      {
      job_attr_def[i].at_free(newattr + i);
      }

    /* cannot set attributes, return FAILURE */

    req_reject(rc, 0, preq, mom_host, (char *)"cannot set attributes");

    return;
    }

  /* OK, now copy the new values into the job pbs_attribute array */

  for (i = 0;i < JOB_ATR_LAST;i++)
    {
    if (!(newattr[i].at_flags & ATR_VFLAG_MODIFY))
      continue;

    if (LOGLEVEL >= 5)
      {
      char tmpLine[1024];

      strcpy(tmpLine, "???");

      if (newattr[i].at_type == ATR_TYPE_STR)
        {
        if (newattr[i].at_val.at_str != NULL)
          snprintf(tmpLine, sizeof(tmpLine), "%s", newattr[i].at_val.at_str);
        }
      else if (newattr[i].at_type == ATR_TYPE_BOOL)
        {
        sprintf(tmpLine, newattr[i].at_type ? 
                "true" : "false");
        }
      else if (newattr[i].at_type == ATR_TYPE_LONG)
        {
        sprintf(tmpLine, "%ld",
                newattr[i].at_val.at_long);
        }
      else if (newattr[i].at_type == ATR_TYPE_CHAR)
        {
        sprintf(tmpLine, "%c",
                newattr[i].at_val.at_char);
        }
      else if (newattr[i].at_type == ATR_TYPE_RESC)
        {
        sprintf(tmpLine, "%s",
                "RESC");
        }
      else if (newattr[i].at_type == ATR_TYPE_ARST)
        {
        if (i == JOB_ATR_variables)
          {
          char *tmpPtr;

          tmpLine[0] = '\0';

          tmpPtr = arst_string(
              (char *)"",
                     &newattr[i]);

          if (tmpPtr != NULL)
            snprintf(tmpLine, sizeof(tmpLine), "%s", tmpPtr);
          }
        else
          {
          sprintf(tmpLine, "%s",
                  "ARST");
          }
        }

      sprintf(log_buffer, "modifying type %d attribute %s of job (value: '%s')",

              newattr[i].at_type,
              (i <= JOB_ATR_checkpoint_name) ? TJobAttr[i] : "Unkn",
              tmpLine);

      log_record(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
      }  /* END if (LOGLEVEL >= 5) */

    if (job_attr_def[i].at_action != NULL)
      job_attr_def[i].at_action(&newattr[i], pjob, ATR_ACTION_ALTER);

    job_attr_def[i].at_free(pattr + i);

    if (newattr[i].at_type == ATR_TYPE_LIST)
      list_move(&newattr[i].at_val.at_list, &(pattr + i)->at_val.at_list);
    else if (newattr[i].at_type == ATR_TYPE_RESC)
      {
      void *old_ptr = pattr[i].at_val.at_ptr;
      pattr[i].at_val.at_ptr = newattr[i].at_val.at_ptr;
      newattr[i].at_val.at_ptr = old_ptr;
      }

#ifdef TNOT
    else if (newattr[i].at_type == ATR_TYPE_ARST)
      {
      set_arst(&(pattr + i)->at_val, newattr[i], SET);

      /* set_arst(patr,&temp,INCR) */
      }

#endif /* TNOT */
    else
      {
      *(pattr + i) = newattr[i];
      }

    (pattr + i)->at_flags = newattr[i].at_flags;
    }    /* END for (i) */

  /* note, the newattr[] attributes are on the stack, they go away auto */

  if (rc == 0)
    {
    rc = mom_set_limits(pjob, SET_LIMIT_ALTER);
    }

  if (rc != 0)
    {
    req_reject(rc, bad, preq, mom_host, (char *)"cannot set limits");

    return;
    }

  if (multi_mom)
    {
    momport = pbs_rm_port;
    }
  
  job_save(pjob, SAVEJOB_FULL, momport);

  sprintf(log_buffer, msg_manager,
          msg_jobmod,
          preq->rq_user,
          preq->rq_host);

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

  /* SUCCESS */

  reply_ack(preq);

  return;
  }  /* END req_modifyjob() */





void req_shutdown(

  struct batch_request *preq)  /* I */

  {
  req_reject(PBSE_NOSUP, 0, preq, NULL, NULL);

  return;
  }




#ifdef _CRAY

/*
 * cray_susp_resum - special cray suspend/resume function
 */

static void cray_susp_resum(

  job                  *pjob,
  int                   which,
  struct batch_request *preq)

  {
  int   i;
  int  ct;
  pid_t  pid;
  long  sess;
  int  sock;

  sock = preq->rq_conn;

  pid = fork_me(sock);

  if (pid > 0)
    {
    /* record pid in job for when child terminates */

    pjob->ji_momsubt = pid;

    if (which == 1)
      {
      pjob->ji_mompost = post_suspend;

      /* save stop time for adjusting walltime */

      pjob->ji_momstat = time_now;
      }
    else
      {
      pjob->ji_mompost = post_resume;
      }

    free_br(preq);

    return;
    }
  else if (pid == -1)
    {
    /* fork failed - still the main mom */

    log_err(-1, __func__, (char *)"cannot fork child for cray suspend");

    req_reject(PBSE_SYSTEM, errno, preq, NULL, NULL);

    return;
    }

  /* child of MOM, cannot update job struct */

  for (unsigned int i = 0; i < pjob->ji_tasks->size(); i++)
    {
    task *ptask = pjob->ji_tasks->at(i);
    sess = ptask->ti_qs.ti_sid;

    for (ct = 0;ct < 3;ct++)
      {
      i = (which == 1) ?  suspend(C_JOB, sess) : resume(C_JOB, sess);

      if (i == 0)
        break;

      if ((errno != EAGAIN) && (errno != EINTR))
        break;
      }

    if (i == -1)
      {
      /* error */

      req_reject(PBSE_SYSTEM, errno, preq, NULL, NULL);

      exit(1);
      }
    }

  reply_ack(preq);

  exit(0);
  }  /* END cray_susp_resum() */

#endif /* _CRAY */





/* send a signal to all tasks on sisters */

int sigalltasks_sisters(

  job *pjob,
  int  signum)

  {
  char            *cookie;
  eventent        *ep;
  int              i;
  int              stream;
  struct tcp_chan *chan = NULL;

  cookie = pjob->ji_wattr[JOB_ATR_Cookie].at_val.at_str;

  for (i = 0;i < pjob->ji_numnodes;i++)
    {
    int      ret = PBSE_NONE;
    hnodent *np = &pjob->ji_hosts[i];

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

    DBPRT(("%s: sending sig%d to all tasks on sister %s\n", __func__, signum, np->hn_host));

    ep = event_alloc(IM_SIGNAL_TASK, np, TM_NULL_EVENT, TM_NULL_TASK);

    if ((stream = tcp_connect_sockaddr((struct sockaddr *)&np->sock_addr,sizeof(np->sock_addr), false)) < 0)
      return(-1);

    if ((chan = DIS_tcp_setup(stream)) == NULL)
      {
      ret = ENOMEM;
      }
    else if ((ret = im_compose(chan, 
            pjob->ji_qs.ji_jobid,
            cookie,
            IM_SIGNAL_TASK,
            ep->ee_event,
            TM_NULL_TASK)) == DIS_SUCCESS)
      {
      if ((ret = diswui(chan, pjob->ji_nodeid)) == DIS_SUCCESS)
        {
        if ((ret = diswsi(chan, TM_NULL_TASK)) == DIS_SUCCESS)
          {
          if ((ret = diswsi(chan, signum)) == DIS_SUCCESS)
            {
            ret = DIS_tcp_wflush(chan);

            /* NYI: add code to read reply from sister */
            }
          }
        }
      }

    if (chan != NULL)
      DIS_tcp_cleanup(chan);

    close(stream);

    if (ret != DIS_SUCCESS)
      return(ret);
    } /* END for each node in ji_hosts */

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




static void resume_suspend(

  job                  *pjob,
  int                   susp,  /* I (0=FALSE, 1=TRUE) */
  struct batch_request *preq)

  {
  int   stat = 0;
  int   savederr = 0;

  int   signum;

  signum = (susp == 1) ? SIGSTOP : SIGCONT;

  if((pjob == NULL)||(preq == NULL))
    return;

  if (LOGLEVEL >= 2)
    {
    sprintf(log_buffer, "%s: %s job", __func__,
      (susp == 1) ? "suspending" : "resuming");

    log_record(
      PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      pjob->ji_qs.ji_jobid,
      log_buffer);
    }

  /* Once upon a time, suspend/resume signals weren't propagated to
     sisters.  Verily, the parallel jobs weren't correctly suspended
     or resumed.

     Then it was decided that SIGTSTP would be used instead of SIGSTOP
     so that specially patched MPI launchers could suspend the sisters.
     To keep serial jobs suspending, a delayed SIGSTOP was added after
     the SIGTSTP.

     Of course, non-MPI parallel jobs were still not suspended properly.

     Now we are sending SIGTSTP only to the top-level task on MS, then
     SIGSTOP to all tasks and given MOM the ability to propagate
     suspend/resume signals to sisters.

                             The End.
   */


  /* NOTE:  format {suspend[:X]|resume[:X]} should be supported to allow
            job state change AND custom suspend/resume signal (NYI) */

  if (susp == 1)
    {
    if (pjob->ji_tasks->size() != 0)
      {
      task *tmpTask = pjob->ji_tasks->at(0);
      kill_task(pjob, tmpTask, SIGTSTP, 0);
      }

    sleep(5);
    }

  for (unsigned int i = 0; i < pjob->ji_tasks->size(); i++)
    {
    task *tp = pjob->ji_tasks->at(i);

    if (tp->ti_qs.ti_status != TI_STATE_RUNNING)
      continue;

    DBPRT(("%s: inspecting %d from node %d\n",
           __func__,
           tp->ti_qs.ti_task,
           tp->ti_qs.ti_parentnode));

    stat = kill_task(pjob, tp, signum, 0);

    if (stat < 0)
      {
      /* couldn't send signal, don't signal more tasks */

      savederr = errno;

      break;
      }  /* END if (stat < 0) */
    }    /* END for each task */

  if (stat >= 0)
    {
    if (pjob->ji_numnodes > 1)
      {
      stat = sigalltasks_sisters(pjob, signum);

      if (stat < 0)
        {
        savederr = errno;
        }
      }
    }

  if (stat < 0)
    {
    /* We couldn't signal all the tasks, signal them back to their old state */

    if (LOGLEVEL >= 1)
      {
      sprintf(log_buffer,
        "cannot send signal %s to tasks of job in %s (errno=%d %s) - attempt aborted",
        (susp == 1) ? "SIGSTOP" : "SIGCONT",
        __func__,
        savederr,
        pbs_strerror(savederr));

      log_record(
        PBSEVENT_ERROR,
        PBS_EVENTCLASS_JOB,
        pjob->ji_qs.ji_jobid,
        log_buffer);
      }

    signum = (susp == 1) ? SIGCONT : SIGSTOP;

    for (unsigned int i = 0; i < pjob->ji_tasks->size(); i++)
      {
      task *tp = pjob->ji_tasks->at(i);

      if (tp->ti_qs.ti_status != TI_STATE_RUNNING)
        continue;

      kill_task(pjob, tp, signum, 0);
      }

    if (pjob->ji_numnodes > 1)
      {
      sigalltasks_sisters(pjob, signum);
      }

    /* report suspend/resume failure */

    req_reject(PBSE_SYSTEM, savederr, preq, NULL, NULL);

    return;
    }  /* END if (stat < 0) */


  /* signals sent to all tasks, now adjust job state */

  if (susp == 1)
    {
    /* Successfully suspended, let's update status */
    /* This is needed for calculating correct walltime */

    pjob->ji_momstat = time_now;

    set_jobs_substate(pjob, JOB_SUBSTATE_SUSPEND);
    pjob->ji_qs.ji_svrflags |= JOB_SVFLG_Suspend;

    if (LOGLEVEL >= 1)
      {
      log_record(
        PBSEVENT_JOB,
        PBS_EVENTCLASS_JOB,
        pjob->ji_qs.ji_jobid,
        (char *)"job suspended - adjusted job state");
      }
    }
  else
    {
    set_jobs_substate(pjob, JOB_SUBSTATE_RUNNING);

    /* Ok, we resumed'em, we have set ji_momstat to the time we suspended the
       job.  We use this to compute a new start-time for the job, so that
       walltime is computed correctly elsewhere */

    if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_Suspend)
      {
      /* If it's suspended, update the start time */

      pjob->ji_qs.ji_stime = pjob->ji_qs.ji_stime - pjob->ji_momstat + time_now;
      }

    /* clear the Suspend bit */

    pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_Suspend;

    if (LOGLEVEL >= 1)
      {
      log_record(
        PBSEVENT_JOB,
        PBS_EVENTCLASS_JOB,
        pjob->ji_qs.ji_jobid,
        (char *)"job resumed - adjusted job state");
      }
    }    /* END else (susp != 0) */

  /* acknowledge change to pbs_server */

  reply_ack(preq);

  /* SUCCESS */

  return;
  }  /* END resume_suspend() */





/**
 * mom_req_signal_job - issue (kill) a specified signal to a job
 * Signal may be either a numeric string or a signal name
 * with or without the "SIG" prefix.
 *
 * NOTE:  mom_process_request() set up as request handler via 
 * accept_conn() 
 *
 * @see mom_process_request->dispatch_request() - parent
 * @see req_signaljob() in server/req_signal.c - peer
 */

void mom_req_signal_job(

  batch_request *preq) /* I */

  {
  job            *pjob;
  int             sig;
  int             sync_kill = FALSE;
  int             numprocs=0;
  char           *sname;
  unsigned int   momport = 0;

  struct sig_tbl *psigt;

  extern struct   sig_tbl sig_tbl[];

  pjob = mom_find_job(preq->rq_ind.rq_signal.rq_jid);

  if (pjob == NULL)
    {
    req_reject(PBSE_UNKJOBID, 0, preq, NULL, NULL);

    return;
    }

  sname = preq->rq_ind.rq_signal.rq_signame;

  if (LOGLEVEL >= 3)
    {
    sprintf(log_buffer, "signaling job with signal %s",
      sname);

    log_record(
      PBSEVENT_JOB,
      PBS_EVENTCLASS_JOB,
      pjob->ji_qs.ji_jobid,
      log_buffer);
    }

  if (preq->rq_extend != NULL)
    {
    if (!strcasecmp("rerun", preq->rq_extend))
      {
      pjob->ji_job_is_being_rerun = TRUE;
      }
    else if (!strcasecmp(preq->rq_extend, SYNC_KILL))
      sync_kill = TRUE;
    }

  if (!strcasecmp(sname, SIG_SUSPEND))
    {
    if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_RUNNING)
      {
      req_reject(PBSE_BADSTATE, 0, preq, NULL, NULL);
      }
    else
      {
#ifdef _CRAY /* suspend/resume on Cray only */
      cray_susp_resum(pjob, 1, preq);
#else
      resume_suspend(pjob, 1, preq);
#endif /* _CRAY */
      }

    return;
    }

  if (!strcasecmp(sname, SIG_RESUME))
    {
    if (pjob->ji_qs.ji_substate != JOB_SUBSTATE_SUSPEND)
      {
      log_event(
        PBSEVENT_JOB,
        PBS_EVENTCLASS_JOB,
        pjob->ji_qs.ji_jobid,
        (char *)"resume request on job that is not suspended");
      }

#ifdef _CRAY
    cray_susp_resum(pjob, 0, preq);

#else
    resume_suspend(pjob, 0, preq);

#endif /* _CRAY */

    return;
    }

  /* standard signal received - signal tasks but do not change job state */

  if (isdigit((int)*sname))
    {
    sig = atoi(sname);
    }
  else
    {
    if (!strncasecmp("SIG", sname, 3))
      sname += 3;

    psigt = sig_tbl;

    while (psigt->sig_name != NULL)
      {
      if (!strcasecmp(sname, psigt->sig_name))
        break;

      psigt++;
      }

    sig = psigt->sig_val;
    }

  if (sig < 0)
    {
    req_reject(PBSE_UNKSIG, 0, preq, NULL, NULL);

    return;
    }

  if ((sig == SIGTERM) && (pjob->ji_qs.ji_substate == JOB_SUBSTATE_SUSPEND))
    {
    /* if job is suspended, resume, and then kill - allow job to clean up on sigterm */

    kill_job(pjob, SIGCONT, __func__, "job is suspended, resume and kill");

    sleep(1);
    }

  if ((sync_kill == TRUE) &&
      ((pjob->ji_qs.ji_substate == JOB_SUBSTATE_EXITING) ||
       (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)))
    {
    snprintf(log_buffer, sizeof(log_buffer),
      "Job %s is being killed from pbs_server.", pjob->ji_qs.ji_jobid);
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
    mom_deljob(pjob);
    }
  else
    {
    int rc;

    /* We are killing the job. If this is a multi-node job contact the sister nodes first so they
     * are not hanging around after mother superior sends an obit indicating the job is already
     * finished
     */
    rc = terminate_sisters(pjob, sig);
    if (rc != PBSE_NONE)
      {
      sprintf(log_buffer, "%s - Could not terminate all sisters: %d", __func__, rc);
      log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
      }

    /*
     * When kill_job is launched, processes are killed and waitpid() should harvest the process
     * and takes action to send an obit. If no matching process exists, then an obit may never be
     * sent due to the current way that TORQUE's state machine works.
     */

    numprocs = kill_job(pjob, sig, __func__, "killing job");

    if ((numprocs == 0) && ((sig == 0)||(sig == SIGKILL)) &&
        (pjob->ji_qs.ji_substate != JOB_SUBSTATE_OBIT))
      {
      if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_EXITED)
        {
        if (pjob->ji_tasks->size() == 0)
          {
          snprintf(log_buffer, sizeof(log_buffer),
            "job recycled into exiting on SIGNULL/KILL from substate %d again. Terminating job now.",
            pjob->ji_qs.ji_substate);
          log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
          mom_deljob(pjob);
          }
        }
      else
        {
        /* SIGNUL and no procs found, force job to exiting */
        /* force issue of (another) job obit */

        sprintf(log_buffer, "job recycled into exiting on SIGNULL/KILL from substate %d",
          pjob->ji_qs.ji_substate);

        log_event(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);

        set_jobs_substate(pjob, JOB_SUBSTATE_EXITING);

        if (multi_mom)
          {
          momport = pbs_rm_port;
          }
      
        job_save(pjob, SAVEJOB_QUICK, momport);

        exiting_tasks = 1;
        }
      }
    }

  reply_ack(preq);

  return;
  }  /* END mom_req_signal_job() */



/**
 * Encodes the used resource information (cput, mem, walltime, etc.)
 * about the given job. (The data is encoded in preparation for
 * being sent to the pbs_server.)
 *
 */

void encode_used(

  job               *pjob,     /* I */
  int                perm,     /* I */
  Json::Value       *job_used, /* O */
  tlist_head        *phead)    /* O */

  {
  unsigned long  lnum;
  pbs_attribute *at;
  attribute_def *ad;
  char           valbuf[MAXLINE];

  at = &pjob->ji_wattr[JOB_ATR_resc_used];
  ad = &job_attr_def[JOB_ATR_resc_used];

  if (((at->at_flags & ATR_VFLAG_SET) == 0) ||
      (at->at_val.at_ptr == NULL))
    return;

  std::vector<resource> *resources = (std::vector<resource> *)at->at_val.at_ptr;

  for (size_t i = 0; i < resources->size(); i++)
    {
    resource     &r = resources->at(i);
    resource_def *rd = r.rs_defin;
    pbs_attribute val;
    int           rc = PBSE_NONE;
    bool          mem_val = false;

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

    val = r.rs_value; /* copy resource pbs_attribute */

    if ((!strcmp(rd->rs_name, "mem")) ||
        (!strcmp(rd->rs_name, "vmem")))
      mem_val = true;

    /* count up sisterhood too */

    if (pjob->ji_resources != NULL)
      {
      lnum = 0;

      if (!strcmp(rd->rs_name, "cput"))
        {
        for (int j = 0; j < pjob->ji_numnodes - 1; j++)
          {
          lnum += pjob->ji_resources[j].nr_cput;
          }
        }
      else if (!strcmp(rd->rs_name, "mem"))
        {
        for (int j = 0; j < pjob->ji_numnodes - 1; j++)
          {
          lnum += pjob->ji_resources[j].nr_mem;
          }
        }
      else if (!strcmp(rd->rs_name, "vmem"))
        {
        for (int j = 0; j < pjob->ji_numnodes - 1; j++)
          {
          lnum += pjob->ji_resources[j].nr_vmem;
          }
        }

      val.at_val.at_long += lnum;
      }

    if (phead != NULL)
      {
      rc = rd->rs_encode(
             &val,
             phead,
             ad->at_name,
             rd->rs_name,
             ATR_ENCODE_CLIENT,
             perm);
      }

    if (job_used != NULL)
      {
      if (mem_val == true)
        snprintf(valbuf, sizeof(valbuf), "%ldkb", val.at_val.at_long);
      else
        snprintf(valbuf, sizeof(valbuf), "%ld", val.at_val.at_long);
        
      (*job_used)[rd->rs_name] = valbuf;
      }

    if (rc < 0)
      break;
    }  /* END for each resource */

#ifdef USE_RESOURCE_PLUGIN

  if (pjob->ji_custom_usage_info != NULL)
    {
    Json::Value plugin_values;

    for (std::map<std::string, std::string>::iterator it = pjob->ji_custom_usage_info->begin();
         it != pjob->ji_custom_usage_info->end();
         it++)
      {
      pbs_attribute val;

      val.at_flags |= ATR_VFLAG_SET;

      if (phead != NULL)
        {
        svrattrl *pal = attrlist_create(ATTR_used, it->first.c_str(), it->second.size() + 1);
        strcpy(pal->al_value, it->second.c_str());
        pal->al_op = SET_PLUGIN;
        append_link(phead, &pal->al_link, pal);
        }

      if (job_used != NULL)
        plugin_values[it->first] = it->second;
      }
        
    if ((job_used != NULL) &&
        (pjob->ji_custom_usage_info->size() > 0))
      {
      (*job_used)[PLUGIN_RESC] = plugin_values;
      }
    }
#endif

  return;
  }  /* END encode_used() */



/*
 * encode_flagged_attrs()
 *
 * Packs the flagged attributes for sending to pbs_server
 *
 * @param pjob - the job whose attributes are being packed
 * @param perm - the permissions for looking at these attributes
 * @param job_info - the json object where these attributes should be added, if non-NULL
 * @param phead - the linked list where these attributes should be added, if non-NULL
 */

void encode_flagged_attrs(

  job               *pjob,     /* I */
  int                perm,     /* I */
  Json::Value       *job_info, /* O */
  tlist_head        *phead)    /* O */

  {
  int             index;
  pbs_attribute  *at;
  attribute_def  *ad;

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

    if (at->at_flags & ATR_VFLAG_SEND)
      {
      /* turn off "need to send" flag */

      at->at_flags &= ~ATR_VFLAG_SEND;

      if (LOGLEVEL >= 4)
        {
        sprintf(log_buffer, "encoding \"send flagged\" attr: %s",
                ad->at_name);

        log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buffer);
        }

      if (phead != NULL)
        ad->at_encode(at, phead, ad->at_name, NULL, ATR_ENCODE_CLIENT, perm);

      if (job_info != NULL)
        {
        std::string attr;
        attr_to_str(attr, ad, *at, false);
        (*job_info)[ad->at_name] = attr;
        }
      }
    }
  } // END encode_flagged_attrs()



void stat_single_job(

  job           *pjob,
  batch_request *preq,
  batch_reply   *preply)

  {
  static int         resc_access_perm = preq->rq_perm & ATR_DFLAG_RDACC;

  struct brp_status *pstat;

  if ((am_i_mother_superior(*pjob) == true) &&
      (pjob->ji_qs.ji_substate == JOB_SUBSTATE_RUNNING))
    {
    /* allocate reply structure and fill in header portion */
    pstat = (struct brp_status *)calloc(1, sizeof(struct brp_status));

    assert(pstat != NULL);

    CLEAR_LINK(pstat->brp_stlink);

    pstat->brp_objtype = MGR_OBJ_JOB;

    strcpy(pstat->brp_objname, pjob->ji_qs.ji_jobid);

    CLEAR_HEAD(pstat->brp_attr);

    append_link(&preply->brp_un.brp_status, &pstat->brp_stlink, pstat);

    encode_used(pjob, resc_access_perm, NULL, &pstat->brp_attr); /* adds resources_used attr */
    
    /* adds other flagged attrs */
    encode_flagged_attrs(pjob, resc_access_perm, NULL, &pstat->brp_attr);
    }

  } // END stat_single_job()



/*
 * req_stat_job - return the status of one (if id is specified) or all
 * jobs (if id is the null string).
 *
 * This is usually triggered due to a request from the pbs_server to learn
 * about this (or all) jobs. The server will query the MOM periodically
 * for this information. It is controlled by the pbs_server attributes
 * 'job_stat_rate' and 'poll_jobs'.
 *
 * This data is different than the occasional status update sent to
 * the server that tells the server the MOM's general stats (see mom_server_all_update_stat()).
 */

int req_stat_job(

  struct batch_request *preq)  /* I */

  {
  char    name[(PBS_MAXSVRJOBID > PBS_MAXDEST ? PBS_MAXSVRJOBID:PBS_MAXDEST)+1];
  job    *pjob;

  struct batch_reply *preply = &preq->rq_reply;

  /*
   * first, validate the name of the requested object, either
   * a single job or all jobs
   */

  snprintf(name, sizeof(name), "%s", preq->rq_ind.rq_status.rq_id);
  name[sizeof(name) - 1] = '\0';

  preply->brp_code = PBSE_NONE;
  preply->brp_choice = BATCH_REPLY_CHOICE_Status;

  CLEAR_HEAD(preply->brp_un.brp_status);

  if ((name[0] == '\0') || (name[0] == '@'))
    {
    std::list<job *>::iterator iter;

    for (iter = alljobs_list.begin(); iter != alljobs_list.end(); iter++)
      {
      stat_single_job(*iter, preq, preply);
      }
    }
  else
    {
    pjob = mom_find_job(name);

    if (pjob == NULL)
      {
      req_reject(PBSE_UNKJOBID, 0, preq, NULL, name);

      snprintf(log_buffer, sizeof(log_buffer),
        "Received job status request for job %s, unknown to me",
        name);
      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buffer);

      return(PBSE_UNKJOBID);
      }
    else
      stat_single_job(pjob, preq, preply);
    }

  reply_send_mom(preq);

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



/*
 * del_files - delete the files in a copy files or delete files request
 *
 * WARNING WARNING WARNING WARNING WARNING WARNING WARNING
 *
 * fork_to_user() must be called first so that useruid/gid is set up
 */

int del_files(

  struct batch_request *preq,      /* I */
  char                 *HDir,      /* I (home directory) */
  int                   setuserenv,/* I */
  char                **pbadfile)  /* O */

  {
  int   AsUser = FALSE;
  int		UID0 = TRUE;
  int		EUID0 = TRUE;

  struct rqfpair  *pair;
  int   rc = 0;
  char  *path;
  char  *pp;
  char  *prmt;

  struct stat  sb;
#if NO_SPOOL_OUTPUT == 1
  char   path_alt[MAXPATHLEN + 1];
  int   rcstat;

  struct stat  myspooldir;
#endif

#ifdef HAVE_WORDEXP
  job *pjob;
  wordexp_t pathexp;
#endif

  path = (char *)calloc((MAXPATHLEN + 1), sizeof(char));

  if (path==NULL)
    {
    add_bad_list(pbadfile,(char *)"calloc failed",1);

    return(-1);
    }

  /*
   * NOTE:  may be called as root in TORQUE home dir
   * or as user in user homedir.  Let's determine if we will
   * be permitted to run setXid()/setgroup calls.
   */

#ifndef __CYGWIN__
 if (getuid() != 0)
    {
#else
  if (IamUser() == 1)
    {
#endif  /* __CYGWIN__ */
    UID0 = FALSE;
    }

#ifndef __CYGWIN__
 if (geteuid() != 0)
    {
#else
  if (IamUser() == 1)
    {
#endif  /* __CYGWIN__ */
    EUID0 = FALSE;
    }

  /*
   * Build up path of file using local name only, then unlink it.
   * The first set of files may have the STDJOBFILE
   * flag set, which we need to unlink as root, the others as the user.
   */

  if (HDir != NULL)
    {
    /* ignore failure */

    if (chdir(HDir) == -1) {}
    }

  for (pair = (struct rqfpair *)GET_NEXT(preq->rq_ind.rq_cpyfile.rq_pair);
       pair != NULL;
       pair = (struct rqfpair *)GET_NEXT(pair->fp_link))
    {
    prmt = pair->fp_rmt;

    *path = '\0';

    if (pair->fp_flag == STDJOBFILE)
      {
      /* the job's stdout/stderr */

#if NO_SPOOL_OUTPUT == 0
      snprintf(path, MAXPATHLEN + 1, "%s", path_spool);
#endif /* !NO_SPOOL_OUTPUT */
      }
    else if (AsUser == FALSE)
      {
      if (setgroups(ngroup,(gid_t *)groups) != 0 && UID0 == TRUE)
        {
        snprintf(log_buffer,sizeof(log_buffer),
          "%s: setgroups() for UID = %lu failed: %s",
          __func__,
          (unsigned long)useruid,
          strerror(errno));

        add_bad_list(pbadfile,log_buffer,1);

        free(path);

        return(-1);
        }

      if (setgid(usergid) != 0 && EUID0 == TRUE)
        {
        snprintf(log_buffer,sizeof(log_buffer),
          "%s: setgid(%lu) for UID = %lu failed: %s",
          __func__,
          (unsigned long)usergid,
          (unsigned long)useruid,
          strerror(errno));

        add_bad_list(pbadfile,log_buffer,1);
        
        free(path);

        return(-1);
        }

      /* run as the user */
      if (setuid_ext(useruid, FALSE) != 0 && EUID0 == TRUE)
        {
        snprintf(log_buffer,sizeof(log_buffer),
          "%s: setuid(%lu) failed: %s",
          __func__,
          (unsigned long)useruid,
          strerror(errno));

        add_bad_list(pbadfile,log_buffer,1);
        
        free(path);

        return(-1);
        }

      EUID0 = FALSE;
      UID0 = FALSE;

      if (HDir != NULL)
        {
        /* ignore failure of chdir */

        if (chdir(HDir) == -1) {}
        }

#ifdef HAVE_WORDEXP
      if (setuserenv && 
          (pjob = mom_find_job(preq->rq_ind.rq_cpyfile.rq_jobid)) != NULL)
        {
        InitUserEnv(pjob, NULL, NULL, NULL, NULL);

        *(vtable.v_envp + vtable.v_used) = NULL;

        environ = vtable.v_envp;
        }

#endif
      AsUser = TRUE;
      }

    strcat(path, pair->fp_local);

    replace_checkpoint_path(path);

    if (is_remote(&prmt) == false)
      {
      /* local file, do source and destination match? */
      /* if so, don't delete it       */

      if (is_file_same(prmt, path) == 1)
        {
        continue;
        }
      }

#ifdef HAVE_WORDEXP
    switch (wordexp(path, &pathexp, WRDE_NOCMD | WRDE_UNDEF))
      {

      case 0:

        break; /* Successful */

      case WRDE_NOSPACE:

        wordfree(&pathexp);

        /* fall through */

      default:

        sprintf(log_buffer, "*** failed to delete files, expansion of %s failed",
                path);

        add_bad_list(pbadfile, log_buffer, 1);

        free(path);

        return(-1);

        /*NOTREACHED*/

        break;
      }

    strcpy(path, pathexp.we_wordv[0]);

    wordfree(&pathexp);
#endif

    if (stat(path, &sb) == 0)
      {
      if (S_ISDIR(sb.st_mode))
        {
        /* have a directory, must append last segment */
        /* of source name to it for the unlink       */

#if NO_SPOOL_OUTPUT == 1
        /* check for  ~/.pbs_spool */
        /* if it isn't a dir., use $HOME us usual */

        strcpy(path_alt, path);
        strcat(path_alt, "/.pbs_spool/");

        rcstat = stat(path_alt, &myspooldir);

        if ((rcstat == 0) &&
            (S_ISDIR(myspooldir.st_mode)) &&
            ((myspooldir.st_mode & S_IXOTH) == S_IXOTH))
          {
          strcpy(path, path_alt);
          }
        else
#endif /* NO_SPOOL_OUTPUT */
          {
          strcat(path, "/");
          }

        pp = strrchr(prmt, (int)'/');

        if (pp != NULL)
          {
          ++pp;
          }
        else if ((pp = strrchr(prmt, (int)':')))
          {
          ++pp;
          }
        else
          {
          pp = prmt;
          }

        strcat(path, pp);
        }
      }
    else
      {
      sprintf(log_buffer, "cannot stat %s",
              path);

      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_REQUEST, __func__, log_buffer);
      }

    /*
     * This should only be set it we are trying to delete a checkpoint restart
     * file that is in the moms default checkpoint directory.  We change the
     * path to remove the jobs checkpoint directory not just the checkpoint itself.
     * Do not remove if it is in the remote checkpoint directory list
     */

    if (pair->fp_flag == JOBCKPFILE)
      {
      char *ptr;
      ptr = strrchr(path,'/');
      if (ptr != NULL)
        {
        ptr[0] = '\0';
        }
      }

    if (!in_remote_checkpoint_dir(path))
      {
      if (remtree(path) == -1)
        {
        if (errno != ENOENT)
          {
          sprintf(log_buffer, "Unable to delete file %s for user %s, error = %d %s",
            path,
            preq->rq_ind.rq_cpyfile.rq_user,
            errno,
            pbs_strerror(errno));

          log_event(PBSEVENT_JOB, PBS_EVENTCLASS_REQUEST, __func__, log_buffer);

          add_bad_list(pbadfile, log_buffer, 2);

          rc = errno;
          }

#ifdef DEBUG

        }
      else
        {
        sprintf(log_buffer, "Deleted file %s for user %s",
          path,
          preq->rq_ind.rq_cpyfile.rq_user);

        log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_FILE, __func__, log_buffer);

#endif  /* DEBUG */
        }
      }
    }

  free(path);

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




void req_rerunjob(

  struct batch_request *preq)  /* I */

  {
  job           *pjob;
  int            sock;
  int            rc;

  pjob = mom_find_job(preq->rq_ind.rq_rerun);

  if (pjob == NULL)
    {
    req_reject(PBSE_UNKJOBID, 0, preq, NULL, NULL);

    return;
    }

  if (LOGLEVEL >= 3)
    {
    log_record(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, (char *)"rerunning job");
    }

  /* fork to send files back */

  if ((rc = fork_me(preq->rq_conn)) > 0)
    {
    free_br(preq); /* parent - note leave connection open   */

    return;
    }
  else if (rc < 0)
    {
    req_reject(-rc, 0, preq, NULL, NULL);

    return;
    }

  /* Child process ...  for each standard file generate and send a Job Files request(s).
   * No message handler function is needed because return_file blocks and waits for reply.
   * This is acceptable because we are a child process, not pbs_mom.
   */
  sock = mom_open_socket_to_jobs_server_with_retries(pjob, __func__, NULL, 10);

  if (sock < 0)
    { 
    req_reject(PBSE_NOSERVER, 0, preq, NULL, NULL);
    
    exit(0);
    }

  bool checkfile_sent_ok = true;
  /* send checkpoint file to the server only when blcr was enabled */
  if (strlen(checkpoint_run_exe_name) > 0)
    {
    rc = return_file(pjob, Checkpoint, sock, TRUE);
    if (rc != PBSE_NONE)
      {
      /* FAILURE - cannot report file to server */

      log_event(
        PBSEVENT_ERROR,
        PBS_EVENTCLASS_REQUEST,
        __func__,
        (char *)"cannot move output files to server. checkpoint file not found");
      checkfile_sent_ok = false;
      }
   }

  int remove_file = 1;

  /* check to see if OU, ER files need to be copied back to user's dir.
   * If so, do not remove files as the pbs_server will be sending a
   * copy files request.
  */
  if ((pjob->ji_wattr[JOB_ATR_copystd_on_rerun].at_flags & ATR_VFLAG_SET) 
    && (pjob->ji_wattr[JOB_ATR_copystd_on_rerun].at_val.at_long == 1))
      remove_file = 0;

  rc = return_file(pjob, StdOut, sock, remove_file);
  int rc2 = return_file(pjob, StdErr, sock, remove_file);

  if (rc != 0 || rc2 != 0 || (!checkfile_sent_ok))
    {
    /* FAILURE - cannot report file to server */ 
    /* If the following condition is true, it means
     * the copying of output files to the server had a problem */
    if (checkfile_sent_ok)
      log_event(
        PBSEVENT_ERROR,
        PBS_EVENTCLASS_REQUEST,
        __func__,
        (char *)"cannot move output files to server");

    req_reject(rc, 0, preq, NULL, NULL);
    
    close(sock);
    }
  else
    {
    /* SUCCESS */

    reply_ack(preq);

    close(sock);

    exit(0);
    }

  /* make sure to call exit() and not return to avoid zombie process */
  exit(EXIT_SUCCESS);
  }  /* END req_rerunjob() */



void req_returnfiles(

  struct batch_request *preq)

  {
  struct job  *pjob;
  int          sock;

  pjob = mom_find_job(preq->rq_ind.rq_returnfiles.rq_jobid);

  if (pjob != NULL)
    {
    sock = mom_open_socket_to_jobs_server_with_retries(pjob, __func__, NULL, 10);

    if (sock < 0)
      {
      sprintf(log_buffer, "mom_open_socket_to_jobs_server FAILED to get socket: %d for job %s",
        sock,
        pjob->ji_qs.ji_jobid);
      
      log_err(errno, __func__, log_buffer);
    
      req_reject(PBSE_SOCKET_FAULT, 0, preq, mom_host, "Cannot open a socket to pbs_server");
      }
    else
      {
      if (preq->rq_ind.rq_returnfiles.rq_return_stdout)
        {
        return_file(pjob, StdOut, sock, FALSE);
        }
      
      if (preq->rq_ind.rq_returnfiles.rq_return_stderr)
        {
        return_file(pjob, StdErr, sock, FALSE);
        }
      
      reply_ack(preq);
      
      close(sock);
      }
    }
  else
    {
    req_reject(PBSE_UNKJOBID, 0, preq, mom_host, (char *)"cannot locate job");
    }

  return;
  } /* END req_returnfiles() */


/*
 * sys_copy - issue system call to copy file
 *
 * Check error and retry as required
 */

static int sys_copy(

  int   rmtflg,  /* I */
  char *ag2,     /* I (is this source or destination?) */
  char *ag3,     /* I (is this source or destination?) */
  int   conn)    /* I */

  {
  const char *ag0;
  const char *ag1;
  int i;
  int loop;
  int rc;

  sprintf(rcperr, "%srcperr.%ld",
          path_spool,
          (long)getpid());

  if (rmtflg == 0)
    {
    /* local copy */

    ag0 = "/bin/cp";
    ag1 = "-rp";
    }
  else
    {
    ag0 = rcp_path;
    ag1 = rcp_args;
    }

  if (LOGLEVEL >= 6)
    {
    sprintf(log_buffer, "executing copy command: %s %s %s %s",
            ag0,
            ag1,
            ag2,
            ag3);

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

  for (loop = 1;loop < 4;++loop)
    {
    
    pid_t pid_fork;
    if ((pid_fork = fork()) > 0)
      {
      /* Parent - wait for copy to complete */

      while (((i = waitpid(pid_fork, &rc, 0)) < 0) && (errno == EINTR))
        /* NO-OP, retry for EINTR */;

      if (i == -1)
        {
        rc = (20000 + errno);         /* 200xx is error on wait */
        }
      else if (WIFEXITED(rc))
        {
        if ((rc = WEXITSTATUS(rc)) == 0)
          {
          return(rc);          /* good,  stop now */
          }
        }
      else if (WIFSTOPPED(rc))
        {
        rc = (30000 + WSTOPSIG(rc)); /* 300xx is stopped */
        }
      else if (WIFSIGNALED(rc))
        {
        rc = (40000 + WTERMSIG(rc)); /* 400xx is signaled */
        }
      }
    else if (pid_fork < 0)
      {
      rc = errno + 10000; /* error on fork (100xx), retry */
      }
    else
      {
      int fd;

      /* child - exec the copy command */

      close(conn);

      /* redirect stderr to make error from rcp available to MOM */

      unlink(rcperr);  /* likely failing, ignore */

      if ((fd = open(rcperr, O_RDWR | O_CREAT | O_EXCL, 0644)) < 0)
        {
        sprintf(log_buffer, "can't open %s, error = %d %s",
                rcperr, errno, pbs_strerror(errno));

        log_err(errno, __func__, log_buffer);

        exit(12);
        };

      if (fd != 2)
        {
        dup2(fd, 2);

        close(fd);
        }

      /* NOTE:  arg2 should be source, arg3 should be destination */

      execl(ag0, ag0, ag1, ag2, ag3, NULL);

      /* reached only if execl() fails */

      sprintf(log_buffer, "exec of command '%s %s %s %s' failed, errno=%d %s",
              ag0,
              ag1,
              ag2,
              ag3,
              errno,
              pbs_strerror(errno));

      log_err(errno, __func__, log_buffer);

      exit(13); /* 13, an unlucky number */
      }    /* END else ((pid_fork = fork()) > 0) */

    /* copy did not work, try again */

    if ((loop % 2) == 0)
      sleep(loop / 2 * 3 + 1);
    }  /* END for (loop) */

  /* tried a bunch of times, just give up */

  sprintf(log_buffer, "command '%s %s %s %s' failed with status=%d, giving up after %d attempts",
          ag0,
          ag1,
          ag2,
          ag3,
          rc,
          loop);

  log_err(rc, __func__, log_buffer);

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

// replace characters in a string
//  char_from replaced with char_to
void string_replchar(

  const char *str,
  char        char_from,
  char        char_to)

  {
  char *p;

  if (str == NULL)
    return;

  p = (char *)str;
  while (*p)
    {
    if (*p == char_from)
      *p = char_to;
    p++;
    }
  }



/*
 * expand_words()
 *
 * Replaces the environment variables in source and destination with their actual values
 * @param source - the source pre-expansion
 * @param destination - the destination pre-expansion, modified by this function
 * @param sources - where all of the source expansions are placed
 * @param bad_list - where any failed arguments are placed
 */

int expand_words(

  char                      *source,
  char                      *destination,
  std::vector<std::string>  &sources,
  char                     **bad_list)

  {
  sources.clear();

#ifdef HAVE_WORDEXP
  wordexp_t       src_exp;
  wordexp_t       dest_exp;

  switch (wordexp(source, &src_exp, WRDE_NOCMD | WRDE_UNDEF))
    {

    case 0:

      break; /* Successful */

    case WRDE_NOSPACE:

      wordfree(&src_exp);

      /* fall through */

    default:

      sprintf(log_buffer, "Failed to expand source path in data staging: %s",
        source);

      add_bad_list(bad_list, log_buffer, 2);

      return(-1);

      /*NOTREACHED*/

      break;
    }  /* END switch () */

  // Expand and verify destination

  // translate spaces so wordexp() won't split things up
  //  on a path containing them
  string_replchar(destination, ' ', '\001');

  switch (wordexp(destination, &dest_exp, WRDE_NOCMD | WRDE_UNDEF))
    {

    case 0:

      /* success - allow if word count is 1 */

      if (dest_exp.we_wordc == 1)
        {
        snprintf(destination, MAXPATHLEN+1, "%s", dest_exp.we_wordv[0]);

        // restore spaces (if any)
        string_replchar(destination, '\001', ' ');

        wordfree(&dest_exp);

        break; /* Successful */
        }

      /* fall through */

    case WRDE_NOSPACE:

      wordfree(&dest_exp);

      /* fall through */

    default:

      sprintf(log_buffer, "Failed to expand destination path in data staging: %s",
        destination);

      add_bad_list(bad_list, log_buffer, 2);

      return(-1);

      break;
    }  /* END switch () */

  /* NOTE: more than one word is only allowed for source */
  for (int i = 0; i < src_exp.we_wordc; i++)
    sources.push_back(src_exp.we_wordv[i]);

  wordfree(&src_exp);

#else
  sources.push_back(source);
#endif

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



void copy_file_cleanup(
    
  int             dir,
  bool            from_spool,
  batch_request  *preq,
  struct rqfpair *pair,
  char           *localname,
  int             localname_size,
  char          **bad_list)

  {
  if ((dir == STAGE_DIR_IN) || (dir == CKPT_DIR_IN))
    {
    /* delete the stage_in files that were just copied in */

    /* NOTE:  running as user in user homedir */
    if (preq != NULL)
      del_files(preq, NULL, 1, bad_list);

    }
#if NO_SPOOL_OUTPUT == 0
  else if (from_spool == true)
    {
    char undelname[MAXPATHLEN + 1];
    /* copy out of spool */

    /* Copying out files and in spool area ... */
    /* move to "undelivered" directory         */
    snprintf(localname, localname_size, "%s", path_spool);
    strncat(localname, pair->fp_local, (localname_size - strlen(localname) - 1));
    snprintf(undelname, sizeof(undelname), "%s", path_undeliv);
    strncat(undelname, pair->fp_local, (sizeof(undelname) - strlen(undelname) - 1));

    if (rename(localname, undelname) == 0)
      {
      add_bad_list(bad_list, output_retained, 1);
      add_bad_list(bad_list, undelname, 0);
      }
    else
      {
      sprintf(log_buffer, "Unable to rename %s to %s",
              localname,
              undelname);

      log_err(errno, __func__, log_buffer);
      }
    }
#endif /* !NO_SPOOL_OUTPUT */

  if ((dir == STAGE_DIR_IN) || (dir == CKPT_DIR_IN))
    {
    unlink(rcperr);
    }
  } // END copy_file_cleanup()



/*
 * copy_and_process()
 *
 * We have a lot of error processing around the call to sys_copy(). As a result, we move
 * it all into this function
 */

int copy_and_process(
    
  bool  rmtflag,
  char  *src,
  char  *dest,
  int    conn,
  int    dir,
  char  *localname,
  char **bad_list,
  int   &bad_files)

  {
  int rc;

  if ((rc = sys_copy(rmtflag, src, dest, conn)) != 0)
    {
    FILE *fp;

    /* copy failed */

    bad_files = 1;

    sprintf(log_buffer, "Unable to copy file %s to %s, error %d",
      src,
      dest,
      rc);

    add_bad_list(bad_list, log_buffer, 2);

    log_err(-1, __func__, log_buffer);

    /* copy message from rcp as well */

    if ((fp = fopen(rcperr, "r")) != NULL)
      {
      add_bad_list(bad_list, (char *)"*** error from copy", 1);

      while (fgets(log_buffer, LOG_BUF_SIZE, fp) != NULL)
        {
        int len = strlen(log_buffer) - 1;

        if (log_buffer[len] == '\n')
          log_buffer[len] = '\0';

        add_bad_list(bad_list, log_buffer, 1);
        }

      fclose(fp);

      add_bad_list(bad_list, (char *)"*** end error output", 1);
      }
    } /* END if ((rc = sys_copy(...)) != 0) */
  else
    {
    /* Copy in/out succeeded */
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buffer,"copy succeeded (%s) from (%s) to (%s)\n",
        (dir == 0)? "In" : "Out", src, dest);
      log_ext(-1, __func__, log_buffer, LOG_DEBUG);
      }

    if (dir == STAGE_DIR_OUT)
      {
      /* have copied out, ok to remove local one */

      if (remtree(localname) < 0)
        {
        sprintf(log_buffer, msg_err_unlink,
                "stage out",
                localname);

        log_err(errno, __func__, log_buffer);


        add_bad_list(bad_list, log_buffer, 2);

        bad_files = 1;
        }
      }
    else if (dir == CKPT_DIR_OUT)
      {
      /*
       * we need to clean up the job checkpoint file
       * the job directory gets deleted when job is done
       */

      /*
       * If the checkpoint file
       * is in the the TRemChkptDirList then we do not delete since directory
       * is remotely mounted.
       */
      if (!in_remote_checkpoint_dir(localname))
        {
        if (LOGLEVEL >= 7)
          {
          sprintf(log_buffer,"removing checkpoint file (%s)\n", localname);
          log_ext(-1, __func__, log_buffer, LOG_DEBUG);
          }

        /* have copied out, ok to remove local one */

        if (remtree(localname) < 0)
          {
          sprintf(log_buffer, msg_err_unlink,
                  "checkpoint",
                  localname);

          log_err(errno, __func__, log_buffer);

          add_bad_list(bad_list, log_buffer, 2);

          bad_files = 1;
          }
        }
      }
    }

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



/*
 * determine_spooldir()
 *
 * Determines the spool directory used for this job
 * @param spooldir - populated with the job's spool directory
 * @param pjob - the job in question
 */

void determine_spooldir(

  std::string &spooldir,
  job         *pjob)

  {
#if NO_SPOOL_OUTPUT == 1
  int         rcstat = 1;
  struct stat myspooldir;

  if (pjob != NULL)
    {
    spooldir = pjob->ji_grpcache->gc_homedir;
    spooldir += "/.pbs_spool/";

    rcstat = stat(spooldir.c_str(), &myspooldir);
    }

  if ((rcstat != 0) || 
      (!S_ISDIR(myspooldir.st_mode)))
    spooldir.clear();

#else  /* NO_SPOOL_OUTPUT == 1 */
  spooldir.clear();

#endif /* END NO_SPOOL_OUTPUT == 1 */

  if ((spooldir.size() == 0) &&
      (TNoSpoolDirList[0] != NULL))
    {
    char *wdir = NULL;

    if (pjob != NULL)
      {
      wdir = get_job_envvar(pjob, (char *)"PBS_O_WORKDIR");
      }

    if (wdir != NULL)
      {
      /* check if job's work dir matches the no-spool directory list */

      for (int dindex = 0;dindex < TMAX_NSDCOUNT;dindex++)
        {
        if (TNoSpoolDirList[dindex] == NULL)
          break;

        if (!strcasecmp(TNoSpoolDirList[dindex], "$WORKDIR") ||
            !strcmp(TNoSpoolDirList[dindex], "*"))
          {
          spooldir = wdir;

          break;
          }

        if (!strncmp(TNoSpoolDirList[dindex], wdir, strlen(TNoSpoolDirList[dindex])))
          {
          spooldir = wdir;

          break;
          }
        }  // END for (dindex)
      }    // END if (wdir != NULL)
    }      // END if spooldir isn't set yet and we have a no spool dir list

  // If we haven't set it yet, then it should be set to the spool path
  if (spooldir.size() == 0)
    spooldir = path_spool;

  } // END determine_spooldir()



/*
 * req_cpyfile - process the Copy Files request from the server to dispose
 * of output from the job.  This is done by a child of MOM since it
 * might take time.
 *
 * The supplied PBS means of moving the file is by "rcp".
 *  A site may wish to change this.
 */


/* NOTE:

---------------------------------------------------------------------------
PBS_BATCH_RunJob (received from sched)
        req_runjob
                svr_startjob
                        svr_stagein ----------------------> req_cpyfile
                                                                fork_to_user[1]
                        svr_strtjob2
                                send_job
                                        PBSD_queuejob ----> req_quejob[2]
                                        PBSD_jscript -----> req_jobscript
                                        PBSD_rdytocmt ----> req_rdytocommit
                                        PBSD_commit ------> req_commit[3]

[1] job not in alljobs_list (pjob == NULL)
[2] added to svr_newjobs list
[3] deleted from svr_newjobs, added to srv_alljobs

  therefore it is normal for pjob to be NULL in req_cpyfile.

*/

void req_cpyfile(

  struct batch_request *preq)  /* I */

  {
  char           *arg2 = NULL;
  char           *arg3 = NULL;
  int             bad_files = 0;
  char           *bad_list = NULL;
  int             dir = 0;
  bool            from_spool = false;  /* set if file must be removed from spool after copy */
  char            localname[MAXPATHLEN + 1];  /* used only for in-bound */

  struct rqfpair *pair = NULL;
  char           *prmt;
  int             rc;
  int             exitcode = 0;
  bool            rmtflag = false;

#ifdef  _CRAY
  char            tmpdirname[MAXPATHLEN + 1];
#endif  /* _CRAY */

  struct stat     myspooldir;
  int             rcstat;
  std::string     spool_dir;

  char            EMsg[1024];
  char            HDir[1024];

  job            *pjob = NULL;

#ifdef HAVE_WORDEXP
  std::vector<std::string> sources;
  int             madefaketmpdir = 0;
  int             usedfaketmpdir = 0;
  char            faketmpdir[1024];
  int             wordexperr = 0;
#endif

  /* there is nothing to copy */
  if (spoolasfinalname == TRUE)
    {
    reply_ack(preq);

    return;
    }

  if (LOGLEVEL >= 3)
    {
    pair = (struct rqfpair *)GET_NEXT(preq->rq_ind.rq_cpyfile.rq_pair);

    if ((pair != NULL) && (pair->fp_rmt != NULL))
      {
      sprintf(log_buffer, "attempting to copy file '%s'",
              pair->fp_rmt);
      }
    else
      {
      sprintf(log_buffer, "copy file request is corrupt");
      }

    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, preq->rq_ind.rq_cpyfile.rq_jobid, log_buffer);
    }
  
  pjob = mom_find_job(preq->rq_ind.rq_cpyfile.rq_jobid);

  rc = (int)fork_to_user(preq, TRUE, HDir, EMsg);

  if (rc < 0)
    {
    char tmpLine[1024];

    /* FAILURE */

    req_reject(-rc, 0, preq, mom_host, EMsg);

    if ((rc != -PBSE_SYSTEM) &&
        (rc != -PBSE_BADUSER))
      {
      sprintf(tmpLine, "fork_to_user failed with rc=%d '%s' - exiting",
        rc,
        EMsg);

      log_err(errno, __func__, tmpLine);

      exit(rc);
      }

    sprintf(tmpLine, "fork_to_user failed with rc=%d '%s' - returning failure",
      rc,
      EMsg);

    log_err(errno, __func__, tmpLine);

    return;
    }  /* END if (rc < 0) */

  if (rc > 0)
    {
    // parent - continue with other tasks. 
    // In single transaction mode, mark this as a special subtask
    if (pjob != NULL)
      {
      if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_STAGEOUT)
        {
        pjob->ji_momsubt = rc;
        pjob->ji_mompost = send_job_obit;
        }
      }

    /* SUCCESS */

    return;
    }

  determine_spooldir(spool_dir, pjob);

  // CHILD - now running as user in the user's home directory

#ifdef HAVE_WORDEXP
  faketmpdir[0] = '\0';

  if (pjob == NULL)
    {
    /* This is a stagein which happens before the job struct to sent to MOM
     * or a checkpoint file coming in.
     * This limits the available variables we can use.  fork_to_user()
     * has already set PBS_JOBID and HOME for us.  Now just fake a TMPDIR
     * if we need it. */

    pjob = mom_job_alloc();

    if (pjob == NULL)
      {
      /* FAILURE - in child process */

      sprintf(log_buffer,"alloc failed with errno=%d - returning failure",
        errno);

      log_err(errno, __func__, log_buffer);

      bad_files = 1;

      goto error;
      }

    strcpy(pjob->ji_qs.ji_jobid, preq->rq_ind.rq_cpyfile.rq_jobid);

    if (TTmpDirName(pjob, faketmpdir, sizeof(faketmpdir)))
      {
      if (!mkdirtree(faketmpdir, 0755))
        {
        int rc;

        rc = put_env_var("TMPDIR", faketmpdir);
        if (rc)
          {
          /* FAILURE - in child process */

          sprintf(log_buffer,"put_env_var failed with %d - returning failure",
            rc);

          log_err(rc, __func__, log_buffer);

          bad_files = 1;

          goto error;
          }


        madefaketmpdir = 1;
        }
      }
    }
  else
    {
    InitUserEnv(pjob, NULL, NULL, NULL, NULL);

    *(vtable.v_envp + vtable.v_used) = NULL;

    environ = vtable.v_envp;
    }

#endif  /* END HAVE_WORDEXP */

  /* build up cp/rcp command(s), one per file pair */

  arg2 = (char *)calloc((MAXPATHLEN + 1), sizeof(char));
  arg3 = (char *)calloc((MAXPATHLEN + 1), sizeof(char));

  dir = preq->rq_ind.rq_cpyfile.rq_dir;

  if ((arg2 == NULL) ||
      (arg3 == NULL))
    {
    /* FAILURE - in child process */

    sprintf(log_buffer,"alloc failed with errno=%d - returning failure",
      errno);

    log_err(errno, __func__, log_buffer);

    bad_files = 1;

    copy_file_cleanup(dir, from_spool, preq, pair, localname, sizeof(localname), &bad_list);
    }

  for (pair = (struct rqfpair *)GET_NEXT(preq->rq_ind.rq_cpyfile.rq_pair);
       pair != NULL;
       pair = (struct rqfpair *)GET_NEXT(pair->fp_link))
    {
    if ((pair->fp_rmt != NULL) && (strstr(pair->fp_rmt, ":/dev/null")))
      {
      /* ignore copies to/from /dev/null */

      continue;
      }

    from_spool = false;

    prmt       = pair->fp_rmt;

    rmtflag = is_remote(&prmt);

    /* which way to copy, in or out? */

    if ((dir == STAGE_DIR_OUT) || (dir == CKPT_DIR_OUT))
      {
      /*
       * out bound copy ...
       * build "from" path name, local to this system
       */

      if (pair->fp_flag == STDJOBFILE)
        {
        snprintf(localname, sizeof(localname), "%s/%s", spool_dir.c_str(), pair->fp_local);

        // If we can't find it under the spool we determined, try path_spool
        rcstat = stat(localname, &myspooldir);

        if ((rcstat != 0) ||
            (!S_ISREG(myspooldir.st_mode)))
          {
          if (spool_dir != path_spool)
            {
            snprintf(localname, sizeof(localname), "%s/%s", path_spool, pair->fp_local);
            from_spool = true;
            }
          }
        
        if (from_spool == false)
          from_spool = (spool_dir == path_spool);
        
        }  /* END if (pair->fp_flag == STDJOBFILE) */
      else if (pair->fp_flag == JOBCKPFILE)
        {
        snprintf(localname, sizeof(localname), "%s", pair->fp_local);

        replace_checkpoint_path(localname);

        /*
         * If the checkpoint directory
         * is in the the TRemChkptDirList then we do not transfer since directory
         * is remotely mounted.
         */
        if (in_remote_checkpoint_dir(localname))
          {
          continue;
          }
        }  /* END if (pair->fp_flag == JOBCKPFILE) */
      else
        {
        /* user-supplied stage-out file */
        snprintf(localname, sizeof(localname), "%s", pair->fp_local);
        }

      snprintf(arg2, MAXPATHLEN+1, "%s", localname);

      /* take (remote) destination name from request */

      *arg3 = '\0';

      if (rmtflag)
        {
        /* using rcp, need to prepend the owner name */
        snprintf(arg3, MAXPATHLEN+1, "%s@%s",
          preq->rq_ind.rq_cpyfile.rq_owner, prmt);
        }
      else
        snprintf(arg3, MAXPATHLEN+1, "%s", prmt);
      }  /* END if (dir == STAGE_DIR_OUT) */
    else
      {
      /* in bound (stage-in) file */

      /* take (remote) source name from request */

      snprintf(arg3, MAXPATHLEN+1, "%s", pair->fp_local);

      if (pair->fp_flag == JOBCKPFILE)
        {
        char needdir[MAXPATHLEN + 1];
        int saveumask;
        char *ptr;

        replace_checkpoint_path(arg3);

        /*
         * If the checkpoint directory
         * is in the the TRemChkptDirList then we do not transfer since directory
         * is remotely mounted.
         */
        if (in_remote_checkpoint_dir(arg3))
          {
          continue;
          }

        /*
         * We may need to create the directory for this inbound checkpoint /
         * restart file.  If the last segment of the path does not exist then
         * create it.
         */

        snprintf(needdir, sizeof(needdir), "%s", arg3);
        ptr = strrchr(needdir,'/');
        if (ptr != NULL)
          {
          ptr[0] = '\0';
          }

        saveumask = umask(0000);

        if ((mkdir(needdir, 01777) == -1) && (errno != EEXIST))
          {
          log_err(errno, __func__, (char *)"Failed to create jobs checkpoint directory");
          }

        umask(saveumask);
        }  /* END if (pair->fp_flag == JOBCKPFILE) */

      if (rmtflag)
        {
        /* using rcp, need to prepend the owner name */
        snprintf(arg2, MAXPATHLEN+1, "%s@%s",
          preq->rq_ind.rq_cpyfile.rq_owner, prmt);
        }
      else
        snprintf(arg2, MAXPATHLEN+1, "%s", prmt);

      }  /* END else (dir == STAGE_DIR_OUT) */

#ifdef HAVE_WORDEXP
    if ((wordexperr = expand_words(arg2, arg3, sources, &bad_list)) != PBSE_NONE)
      {
      if (bad_list != NULL)
        bad_files = 1;

      copy_file_cleanup(dir, from_spool, preq, pair, localname, sizeof(localname), &bad_list);

      break;
      }

    /* if we made a fake TMPDIR, and we are using it, don't delete after stagein */

    if (madefaketmpdir &&
        (faketmpdir[0] != '\0') &&
        !strncmp(faketmpdir, arg3, strlen(faketmpdir)))
      {
      usedfaketmpdir = 1;
      }
    else
      {
      usedfaketmpdir = 0;
      }

#endif /* HAVE_WORDEXP */

    // Apparently we can have more that one source? This doesn't make sense to me, as we're 
    // copying all of them to the same destination, but I'll preserve the functionality
    for (size_t i = 0; i < sources.size(); i++)
      {
      if ((rmtflag == 0) &&
           ((is_file_same(arg2, arg3) == 1) || (is_file_going_to_dir(arg2, arg3) == 1)))
        {
        /*
         * If this is a local file then don't copy it
         * if source file and destination file are the same file or
         * if the destination (arg3) is a directory not a file name
         * and the source file (arg2) is in the destination directory (arg3)
         */

        continue;
        }

      if ((rc = copy_and_process(rmtflag,
                                 arg2,
                                 arg3,
                                 preq->rq_conn,
                                 dir,
                                 localname,
                                 &bad_list,
                                 bad_files)) != PBSE_NONE)
        {
        copy_file_cleanup(dir, from_spool, preq, pair, localname, sizeof(localname), &bad_list);
        exitcode = COPY_FILE_FAIL;

        break;
        }

      unlink(rcperr);
      } // END for each source
    }  /* END for (pair) */

error:
#ifdef HAVE_WORDEXP
  if (madefaketmpdir && !usedfaketmpdir)
    {
    remtree(faketmpdir);
    }

#endif

  // In single transaction mode, delete the staged in files next
  if ((pjob != NULL) && 
      (pjob->ji_qs.ji_substate == JOB_SUBSTATE_STAGEOUT))
    {
    if (bad_files)
      log_err(-1, __func__, bad_list);

    delete_staged_in_files(pjob, HDir, &bad_list);

    free_br(preq);
    }
  else
    {
    if (bad_files)
      {
      reply_text(preq, PBSE_NOCOPYFILE, bad_list);

      log_err(-1, __func__, bad_list);
      }
    else
      {
      reply_ack(preq);
      }
    }

  /* we are the child, exit not return */

  /* SUCCESS */

  exit(exitcode);
  }  /* END req_cpyfile() */



/*
 * req_delfile - delete the specifled output/staged files
 */

void req_delfile(

  struct batch_request *preq)  /* I */

  {
  int  rc;
  char *bad_list = NULL;

  char   HDir[1024];
  char   EMsg[1024];

  rc = (int)fork_to_user(preq, FALSE, HDir, EMsg);

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

    req_reject(-rc, 0, preq, mom_host, EMsg);

    return;
    }

  if (rc > 0)
    {

    // parent - continue with other tasks
    /* SUCCESS */

    return;
    }

  /* child */

  /* running as root in TORQUE homedir */

  /* delete the files */

  if ((rc = del_files(preq, HDir, 1, &bad_list)))
    {
    /* FAILURE */

    reply_text(preq, rc, bad_list);

    exit(0);
    }

  /* SUCCESS */

  reply_ack(preq);

  /* we are the child, exit not return */

  exit(0);
  }  /* END req_delfile() */



job *job_with_reservation_id(

  const char *rsv_id)

  {
  job *pjob;
  std::list<job *>::iterator iter;

  for (iter = alljobs_list.begin(); iter != alljobs_list.end(); iter++)
    {
    pjob = *iter;

    if ((pjob->ji_wattr[JOB_ATR_reservation_id].at_flags & ATR_VFLAG_SET) &&
        (pjob->ji_wattr[JOB_ATR_reservation_id].at_val.at_str != NULL))
      {
      if (!strcmp(rsv_id, pjob->ji_wattr[JOB_ATR_reservation_id].at_val.at_str))
        return(pjob);
      }
    }

  return(NULL);
  } // END job_with_reservation_id()



void req_delete_reservation(
    
  struct batch_request *request)

  {
  char *rsv_id = request->rq_extend;
  int   rc = PBSE_NONE;
  job *pjob = NULL;

  if (rsv_id != NULL)
    {
    if ((pjob=job_with_reservation_id(rsv_id)) == NULL) 
      {
      if ((rc = destroy_alps_reservation(rsv_id, apbasil_path, apbasil_protocol, 1)) != PBSE_NONE)
        {
        snprintf(log_buffer, sizeof(log_buffer), "Couldn't release reservation id %s",
          rsv_id);
        log_err(-1, __func__, log_buffer);
        }
      }
    else
      {
      snprintf(log_buffer, sizeof(log_buffer), "Ignored release reservation request from server for reservation id %s because of job %s", 
        rsv_id, pjob->ji_qs.ji_jobid);
      log_record(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buffer);
      }
    }

  if (rc == PBSE_NONE)
    reply_ack(request);
  else
    req_reject(-1, 0, request, NULL, log_buffer);
  } /* END req_delete_reservation() */



/*
 * This call will put the machine into the low power state requested if
 * the machine is set up to support that low power state. Otherwise it will
 * reject the request. The function will do all checking and if the outcome is to
 * put the machine into low power state it will send the ack and pause before executing
 * the low power state command.
 *
 * For documentation on how to manage low power states for linux see this link:
 *
 * https://www.kernel.org/doc/Documentation/power/states.txt
 */
void req_change_power_state(
    
  batch_request *request)

  {
  power_state pstate;
  int         requested_power_state = request->rq_ind.rq_powerstate;

  if (!pstate.is_valid())
    {
    req_reject(-PBSE_POWER_STATE_UNSUPPORTED,0,request,NULL,pstate.get_last_error_string());
    return;
    }

  if (!pstate.is_valid_power_state(requested_power_state))
    {
    req_reject(-PBSE_POWER_STATE_UNAVAILABLE,0,request,NULL,pstate.get_last_error_string());
    return;
    }

  reply_ack(request); //Reply first, won't be able to after.
  pstate.set_power_state(requested_power_state);
  sleep(10);
  } // END req_change_power_state()



/*
 * initialize_stageout_request()
 *
 * Sets up a stageout request for pjob to be used by req_cpyfile()
 */

batch_request *initialize_stageout_request(

  job *pjob)

  {
  batch_request     *preq = alloc_br(PBS_BATCH_CopyFiles);
  struct rq_cpyfile *pcf = &preq->rq_ind.rq_cpyfile;

  pcf->rq_dir = STAGE_DIR_OUT;
  CLEAR_HEAD(pcf->rq_pair);
  strcpy(pcf->rq_jobid, pjob->ji_qs.ji_jobid);
  snprintf(pcf->rq_owner, sizeof(pcf->rq_owner),
           "%s", pjob->ji_wattr[JOB_ATR_job_owner].at_val.at_str);
  snprintf(pcf->rq_user, sizeof(pcf->rq_user),
           "%s", pjob->ji_wattr[JOB_ATR_euser].at_val.at_str);

  char *at;

  if (pjob->ji_wattr[JOB_ATR_egroup].at_val.at_str != NULL)
    {
    snprintf(pcf->rq_group, sizeof(pcf->rq_group),
             "%s", pjob->ji_wattr[JOB_ATR_egroup].at_val.at_str);

    if ((at = strchr(pcf->rq_group, '@')) != NULL)
      *at = '\0';
    }

  if ((at = strchr(pcf->rq_owner, '@')) != NULL)
    *at = '\0';

  if ((at = strchr(pcf->rq_user, '@')) != NULL)
    *at = '\0';

  return(preq);
  } // END initialize_stageout_request()



/*
 * get_std_file_info()
 *
 * Returns a batch request with the std file information to copy back
 * @param pjob - the job whose std file info is wanted
 */

batch_request *get_std_file_info(

  job *pjob)

  {
  batch_request *preq;
  bool           copy_stdout = true;
  bool           copy_stderr = true;

  // Interactive jobs have no std files, and if spoolasfinalname is set the output is already
  // where it should be
  if ((spoolasfinalname == TRUE) ||
      ((pjob->ji_wattr[JOB_ATR_interactive].at_flags & ATR_VFLAG_SET) &&
       (pjob->ji_wattr[JOB_ATR_interactive].at_val.at_long != 0)))
    return(NULL);

  if (pjob->ji_wattr[JOB_ATR_join].at_val.at_str != NULL)
    {
    if (!strcmp(pjob->ji_wattr[JOB_ATR_join].at_val.at_str, "oe"))
      copy_stderr = false;
    else if (!strcmp(pjob->ji_wattr[JOB_ATR_join].at_val.at_str, "eo"))
      copy_stdout = false;
    }

  preq = initialize_stageout_request(pjob);
  struct rq_cpyfile *pcf = &preq->rq_ind.rq_cpyfile;
    
  if (copy_stdout)
    {
    struct rqfpair *pair = (struct rqfpair *)calloc(1, sizeof(struct rqfpair));
    std::string local(pjob->ji_qs.ji_fileprefix);
    local += JOB_STDOUT_SUFFIX;

    pair->fp_local = strdup(local.c_str());
    pair->fp_rmt = strdup(pjob->ji_wattr[JOB_ATR_outpath].at_val.at_str);
    pair->fp_flag = STAGE_DIR_OUT;
    
    CLEAR_LINK(pair->fp_link);
    append_link(&pcf->rq_pair, &pair->fp_link, pair);
    }

  if (copy_stderr)
    {
    struct rqfpair *pair = (struct rqfpair *)calloc(1, sizeof(struct rqfpair));
    std::string local(pjob->ji_qs.ji_fileprefix);
    local += JOB_STDERR_SUFFIX;

    pair->fp_local = strdup(local.c_str());
    pair->fp_rmt = strdup(pjob->ji_wattr[JOB_ATR_errpath].at_val.at_str);
    pair->fp_flag = STAGE_DIR_OUT;
    
    CLEAR_LINK(pair->fp_link);
    append_link(&pcf->rq_pair, &pair->fp_link, pair);
    }

  return(preq);
  } // END get_std_file_info()



/*
 * place_files_in_preq()
 *
 * Takes the files from arst and places them as pairs into preq in the format to be used by
 * req_cpyfile() or req_delfile()
 *
 * @param preq - the batch request we're populating
 * @param arst - the array_strings that we're pulling file names from
 */
void place_files_in_preq(

  batch_request        *preq,
  struct array_strings *arst)

  {
  struct rq_cpyfile *pcf = &preq->rq_ind.rq_cpyfile;

  for (int i = 0; i < arst->as_usedptr; i++)
    {
    std::string stage(arst->as_string[i]);
    std::size_t pos = stage.find("@");

    if (pos != std::string::npos)
      {
      struct rqfpair *pair = (struct rqfpair *)calloc(1, sizeof(struct rqfpair));
      std::string local(stage.substr(0, pos));

      pair->fp_local = strdup(local.c_str());
      pair->fp_rmt = strdup(stage.c_str() + pos + 1);
      pair->fp_flag = STAGEFILE;
    
      CLEAR_LINK(pair->fp_link);
      append_link(&pcf->rq_pair, &pair->fp_link, pair);
      }
    }
  } // END place_files_in_preq()



/*
 * get_stageout_info()
 *
 * Reads the list of stageout files for pjob, if any, and places them in a batch request
 * @param preq - the batch request to create or add to
 * @param pjob - the job whose stage out files we're looking for.
 * @return an appended or created batch request with the stage out information, or NULL if there is
 * no stage out information an the preq parameter is NULL
 */

batch_request *get_stageout_info(

  batch_request *preq,
  job           *pjob)

  {
  struct array_strings *arst = pjob->ji_wattr[JOB_ATR_stageout].at_val.at_arst;

  // If this isn't set we have nothing to add
  if (arst == NULL)
    return(preq);

  if (preq == NULL)
    preq = initialize_stageout_request(pjob);

  place_files_in_preq(preq, arst);

  return(preq);
  } // END get_stageout_info()



/*
 * send_back_std_and_staged_files()
 *
 * Looks at pjob and sends back it's std files and stage out files, if any
 * @param pjob - the job whose files we're sending back
 * @param exit_status - ignored
 */

int send_back_std_and_staged_files(

  job *pjob,
  int  exit_status)

  {
  // Only attempt to copy files back if we actually ran the job and aren't re-running it
  // An exit status of JOB_EXEC_FAIL1 (-1), JOB_EXEC_FAIL2 (-2), or JOB_EXEC_RETRY (-3) 
  // indicate that the job didn't actually run
  if ((pjob->ji_job_is_being_rerun == FALSE) &&
      ((pjob->ji_qs.ji_un.ji_momt.ji_exitstat > JOB_EXEC_FAIL1) ||
       (pjob->ji_qs.ji_un.ji_momt.ji_exitstat < JOB_EXEC_RETRY)))
    {
    batch_request *preq = get_std_file_info(pjob);

    preq = get_stageout_info(preq, pjob);

    set_jobs_substate(pjob, JOB_SUBSTATE_STAGEOUT);

    if (preq != NULL)
      req_cpyfile(preq);
    else
      delete_staged_in_files(pjob, NULL, NULL);
    }
  else
    delete_staged_in_files(pjob, NULL, NULL);

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



/*
 * delete_staged_in_files()
 *
 * Looks at pjob and deletes its staged in files, if any
 */

void delete_staged_in_files(

  job   *pjob,
  char  *home_dir,
  char **bad_list)

  {
  if (pjob->ji_job_is_being_rerun == FALSE)
    {
    struct array_strings *arst = pjob->ji_wattr[JOB_ATR_stagein].at_val.at_arst;

    set_jobs_substate(pjob, JOB_SUBSTATE_STAGEDEL);

    if (arst != NULL)
      {
      batch_request *preq = initialize_stageout_request(pjob);
      preq->rq_ind.rq_cpyfile.rq_dir = STAGE_DIR_IN;

      place_files_in_preq(preq, arst);

      if (home_dir == NULL)
        {
        char *bad = NULL;
        home_dir = get_job_envvar(pjob, (char *)"PBS_O_INITDIR");
        
        if (home_dir == NULL)
          home_dir = pjob->ji_grpcache->gc_homedir;

        *bad_list = bad;
        }

      del_files(preq, home_dir, 1, bad_list);
      }
    // If we have forked, we are the user and not root, and we don't want to send the obit.
    // The obit gets sent once this child process exits.
    else if (getuid() == 0)
      send_job_obit(pjob, 0);
    }
  // If we have forked, we are the user and not root, and we don't want to send the obit.
  // The obit gets sent once this child process exits.
  else if (getuid() == 0)
    send_job_obit(pjob, 0);

  } // END delete_staged_in_files()



/*
 * post_stageout()
 *
 * We have finished copying back any stdout/stderr files and stage out files. Deleting any
 * stage in files in the next step.
 */

int post_stageout(

  job *pjob,  
  int  exit_code)

  {
  delete_staged_in_files(pjob, NULL, NULL);
  return(PBSE_NONE);
  } // END post_stageout()



/* END requests.c */

