#include "license_pbs.h" /* See here for the software license */
/**
 * checkpoint.c
 *
 * Support for job checkpoint and restart.
 *
 * Note that there are two coexistent systems for implementing
 * checkpoint and restart.  The older one is based on machine
 * dependent code in the mom_mach.c module.  The newer one uses
 * the BLCR system modules.  The working of the code below is
 * based on the variable checkpoint_system_type which can have
 * the values CST_NONE, CST_MACH_DEP or CST_BLCR.  This variable
 * is set by calling mom_does_checkpoint at system startup time.
 *
 * The CST_MACH_DEP system uses a directory based scheme of
 * taking a checkpoint and produces a file for each task.  The
 * restart code iterates over the directory and does a restart
 * task for each file it finds.  The entire set of files must
 * be written for the checkpoint to be valid.  To ensure this,
 * an existing checkpoint directory is renamed with a .old
 * extension while writing a new checkpoint.  If something
 * fails during this process, recovery code will notice the
 * .old directory and deduce that the existing directory is
 * invalid and should be deleted at the same time renaming
 * the backup to the current.
 *
 * In the case of the CST_BLCR system, BLCR is MPI aware and
 * takes care of doing the checkpoints for each task.
 * By the way, a task as defined in Torque is a process group
 * executing on a node.
 * So the checkpoint and restore for BLCR is done on the head
 * node task.  A directory can be specified where the checkpoint
 * image will reside but there is only one file per checkpoint.
 * BLCR also differs in that multiple checkpoint images are
 * allowed to exist on the disk and when the restart is performed,
 * the name in the job is used to decide which one of these is
 * to be used.  This name may be altered to a previous image by
 * the operators use of the qalter command.
 *
 */


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

#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <dirent.h>
#include <time.h>
#include <sys/types.h>
#include <sys/stat.h>



#include "dis.h"
#include "libpbs.h"
#include "pbs_error.h"
#include "attribute.h"
#include "server_limits.h"
#include "pbs_job.h"
#include "batch_request.h"
#include "resmon.h"
#include "mom_mach.h"
#include "mom_func.h"
#include "log.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "../lib/Liblog/chk_file_sec.h"
#include "mcom.h"
#include "net_connect.h"
#include "resource.h"
#include "csv.h"
#include "svrfunc.h"
#include "pbs_ifl.h"
#include "alps_constants.h"
#include "alps_functions.h"
#include "../lib/Libifl/lib_ifl.h"

#define MAX_CONN_RETRY 3

extern int exiting_tasks;
extern int LOGLEVEL;
extern     int             lockfds;
extern char TORQUE_JData[];

extern int task_recov(job *pjob);
extern char *path_spool;
extern char  *path_jobs;
extern char  *TRemChkptDirList[];
extern char  *apbasil_path;
extern char  *apbasil_protocol;

extern int  multi_mom;
extern unsigned short pbs_rm_port;
extern int            is_login_node;

#ifdef USEJOBCREATE
extern uint64_t get_jobid(char *);
#endif /* USEJOBCREATE */

#ifdef ENABLE_CSA
extern void add_wkm_start(uint64_t, char *);
#endif /* ENABLE_CSA */

int     checkpoint_system_type = CST_NONE;
char    path_checkpoint[MAXPATHLEN + 1];

/* BLCR variables */
char       checkpoint_script_name[MAXPATHLEN + 1];
char       restart_script_name[MAXPATHLEN + 1];
char       checkpoint_run_exe_name[MAXPATHLEN + 1];
int        default_checkpoint_interval = 10; /* minutes */

extern char *mk_dirs(const char *);
extern "C"
{
extern void set_attr(struct attrl **, const char *, const char *);
}
extern int write_nodes_to_file(job *);
extern int write_attr_to_file(job *, int, const char *);

int create_missing_files(job *pjob);

/* The following is used for building command line args
 * and makes sure that at least something is generated
 * for each arg so that the script gets a consistent
 * command line.
 */
#define SET_ARG(x) (((x) == NULL) || (*(x) == 0))?(char *)"-":(x)


/**
 * mom_checkpoint_job_is_checkpointable
 *
 * @param pjob Pointer to job structure.
 * @see TMomFinalizeChild
 */


int mom_checkpoint_job_is_checkpointable(

  job *pjob)

  {
  pbs_attribute *pattr;
  int            rc;

  pattr = &pjob->ji_wattr[JOB_ATR_checkpoint];

  rc = checkpoint_system_type != CST_NONE &&
       checkpoint_script_name[0] != 0 &&
       (pattr->at_flags & ATR_VFLAG_SET) &&
       ((csv_find_string(pattr->at_val.at_str, "c") != NULL) ||
        (csv_find_string(pattr->at_val.at_str, "s") != NULL) ||
        (csv_find_string(pattr->at_val.at_str, "enabled") != NULL) ||
        (csv_find_string(pattr->at_val.at_str, "shutdown") != NULL) ||
        (csv_find_string(pattr->at_val.at_str, "periodic") != NULL));

  return(rc);
  }


/**
 * mom_checkpoint_execute_job
 *
 * This routine is called from the newly created child process.
 * It is required for the BLCR system because the job must run as
 * a child of the cr_run program.
 *
 * @param pjob Pointer to job structure.
 * @see TMomFinalizeChild
 */
int mom_checkpoint_execute_job(

  job              *pjob,
  char             *shell,
  char             *arg[],
  struct var_table *vtable)

  {

  /* Launch job executable with cr_run command so that cr_checkpoint command will work. */

  /* shuffle up the existing args */
  arg[5] = arg[4];
  arg[4] = arg[3];
  arg[3] = arg[2];
  arg[2] = arg[1];
  /* replace first arg with shell name
     note, this func is called from a child process that exits after the
     executable is launched, so we don't have to worry about freeing
     this calloc later */
  arg[1] = (char *)calloc(1, strlen(shell) + 1);

  if (arg[1] == NULL)
    {
    log_err(errno, __func__, (char *)"cannot alloc env");

    return(-1);
    }

  strcpy(arg[1], shell);
  arg[0] = checkpoint_run_exe_name;

  if (LOGLEVEL >= 10)
    {
    char cmd[MAXPATHLEN + 1];
    int i;
    int len;

    snprintf(cmd, sizeof(cmd), "%s", arg[0]);
    
    for (i = 1; arg[i] != NULL; i++)
      {
      len = strlen(cmd);
      snprintf(cmd + len, sizeof(cmd) - len, " %s", arg[i]);
      }

    len = strlen(cmd);
    snprintf(cmd + len, sizeof(cmd) - len, ")");

    log_buffer[0] = '\0';
    sprintf(log_buffer, "execing checkpoint command (%s)\n", cmd);
    log_ext(-1, __func__, log_buffer, LOG_DEBUG);
    }

  execve(checkpoint_run_exe_name, arg, vtable->v_envp);

  return (0);
  }


/**
 * mom_checkpoint_init
 *
 * This routine is called from the mom startup code.
 * @see setup_program_environment
 */
int mom_checkpoint_init(void)

  {
  int   c = 0;
  char *path_checkpt_tmp;

  checkpoint_system_type = mom_does_checkpoint(); /* {CST_NONE, CST_MACH_DEP, CST_BLCR} */

  if (strlen(path_checkpoint) == 0) /* if not -C option */
    {
    /* mk_dirs callocs the string it returns so this string must be freed */

    path_checkpt_tmp = mk_dirs("checkpoint/");
    snprintf(path_checkpoint,sizeof(path_checkpoint),"%s",path_checkpt_tmp);
    free(path_checkpt_tmp);
    }

  if (checkpoint_system_type == CST_BLCR)
    {
    /* set permissions on default checkpoint path, if needed */
    struct stat sb;
    
    if ((stat(path_checkpoint, &sb) == 0) && ((sb.st_mode & 01777) != 01777)) 
      {
      (void)chmod(path_checkpoint, 01777);
      }
    }

#if !defined(DEBUG) && !defined(NO_SECURITY_CHECK)

  if (checkpoint_system_type == CST_BLCR)
    {
    c = chk_file_sec(path_checkpoint, 1, 1, 0, 1, NULL);
    }
  else
    {
    c = chk_file_sec(path_checkpoint, 1, 1, S_IWGRP | S_IWOTH, 1, NULL);
    }

#endif  /* not DEBUG and not NO_SECURITY_CHECK */
  return(c);
  }



/*========================================================================*/
/* Routines called from the config file parsing to set various variables. */
/*========================================================================*/


void mom_checkpoint_set_directory_path(
    
  const char *str)

  {
  int   len = strlen(str);

  if (str[len - 1] == '/')
    snprintf(path_checkpoint, sizeof(path_checkpoint), "%s", str);
  else
    snprintf(path_checkpoint, sizeof(path_checkpoint), "%s/", str);
  }


unsigned long
mom_checkpoint_set_checkpoint_interval(const char *value)  /* I */

  {
  log_record(
    PBSEVENT_SYSTEM,
    PBS_EVENTCLASS_SERVER,
    "checkpoint_interval",
    value);

  default_checkpoint_interval = atoi(value);

  return(1);
  }  /* END set_checkpoint_script() */





unsigned long mom_checkpoint_set_checkpoint_script(
    
  const char *value)  /* I */

  {
  struct stat sbuf;

  log_record(
    PBSEVENT_SYSTEM,
    PBS_EVENTCLASS_SERVER,
    "checkpoint_script",
    value);

  if ((stat(value, &sbuf) == -1) || !(sbuf.st_mode & S_IXUSR))
    {
    /* file does not exist or is not executable */

    return(0);  /* error */
    }

  snprintf(checkpoint_script_name, sizeof(checkpoint_script_name), "%s", value);

  return(1);
  }  /* END set_checkpoint_script() */





unsigned long mom_checkpoint_set_restart_script(
    
  const char *value)  /* I */

  {
  struct stat sbuf;

  log_record(
    PBSEVENT_SYSTEM,
    PBS_EVENTCLASS_SERVER,
    "restart_script",
    value);

  if ((stat(value, &sbuf) == -1) || !(sbuf.st_mode & S_IXUSR))
    {
    /* file does not exist or is not executable */

    return(0);  /* error */
    }

  snprintf(restart_script_name, sizeof(restart_script_name), "%s", value);

  return(1);
  }  /* END set_restart_script() */





unsigned long mom_checkpoint_set_checkpoint_run_exe_name(
    
  const char *value)  /* I */

  {
  struct stat sbuf;

  log_record(
    PBSEVENT_SYSTEM,
    PBS_EVENTCLASS_SERVER,
    "checkpoint_run_exe",
    value);

  if ((stat(value, &sbuf) == -1) || !(sbuf.st_mode & S_IXUSR))
    {
    /* file does not exist or is not executable */

    return(0);  /* error */
    }

  snprintf(checkpoint_run_exe_name, sizeof(checkpoint_run_exe_name), "%s", value);

  return(1);
  }  /* END set_checkpoint_run_exe() */



/*
 * get_jobs_default_checkpoint_dir - Fills in jobs default checkpoint path.
 */

void get_jobs_default_checkpoint_dir(
    
  char *prefix,      /* I */
  char *defaultpath, /* O */
  int   size)        /* I - max writtable to defaultpath */

  {
  snprintf(defaultpath, size, "%s%s%s", 
    path_checkpoint, prefix, JOB_CHECKPOINT_SUFFIX);

  return;
  }  /* END get_jobs_default_checkpoint_dir() */



/*
** Get jobs checkpoint directory, add jobs directory if needed.
*/

void get_chkpt_dir_to_use(

  job  *pjob,
  char *chkpt_dir,
  int   chkpt_dir_size)

  {
  /*
   * Append jobs directory, if needed.
   */

  char job_dir[MAXPATHLEN+1];

  if ((!(pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_flags & ATR_VFLAG_SET)) ||
    (checkpoint_system_type != CST_BLCR))

    {
    /* No dir specified, use the default job checkpoint directory
       e.g.  /var/spool/torque/checkpoint/42.host.domain.CK */

    get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, chkpt_dir, chkpt_dir_size);
    }
  else
    {
    bool need_slash = false;
    int len;

    sprintf(job_dir,"%s%s",
      pjob->ji_qs.ji_fileprefix, JOB_CHECKPOINT_SUFFIX);

    len = strlen(pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str);
    if (pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str[len-1] != '/')
      need_slash = true;


    if ((strlen(pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str) <=
      strlen(job_dir)) ||
      (strcmp(job_dir, &pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str[strlen(pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str) - strlen(job_dir)])))
      {
      if (need_slash == true)
        {
        snprintf(chkpt_dir, chkpt_dir_size, "%s/%s",
          pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str, job_dir);
        }
      else
        snprintf(chkpt_dir, chkpt_dir_size, "%s%s",
          pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str, job_dir);
      }
    else
      snprintf(chkpt_dir, chkpt_dir_size, "%s", pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str);
    }

  return;
  } /* END get_chkpt_dir_to_use() */





/*
 * replace_checkpoint_path - Replaces MOM_DEFAULT_CHECKPOINT_DIR in string
 * with the default checkpoint path. Returns 1 if path was changed else 0.
 */

int replace_checkpoint_path(

  char *path) /* I */

  {
  char *ptr1;
  char *ptr2;
  char tmppath[MAXPATHLEN+1];
  int len;
  int rtnval = 0;

  snprintf(tmppath, sizeof(tmppath), "%s", path);

  ptr1 = strstr(path, MOM_DEFAULT_CHECKPOINT_DIR);
  ptr2 = strstr(tmppath, MOM_DEFAULT_CHECKPOINT_DIR);

  if (ptr1 != NULL)
    {
    ptr1++;
    ptr1 = strchr(ptr1,'$');
    ptr1++;
    len = strlen(path_checkpoint);
    memcpy(ptr2, path_checkpoint, len);
    ptr2 += len;
    if ((path_checkpoint[strlen(path_checkpoint) - 1] == '/') && (ptr1[0] == '/'))
      {
      ptr1++;
      }
    strcpy(ptr2, ptr1);
    strcpy(path, tmppath);
    sprintf(log_buffer,"Converted filename is (%s)\n",
        path);
    log_ext(-1, __func__, log_buffer, LOG_DEBUG);
    rtnval = 1;
    }

  return (rtnval);
  }  /* END replace_checkpoint_path() */



/*
 * in_remote_checkpoint_dir - Checks if path is in the remote checkpoint
 * directories list TRemChkptDirList. If it is then returns TRUE else FALSE
 *
 */

int in_remote_checkpoint_dir(

        char *ckpt_path) /* I */

  {
  int   dindex;

  /*
   * Is the checkpoint directory in the TRemChkptDirList
   */

  for (dindex = 0;dindex < TMAX_RCDCOUNT;dindex++)
    {
    if (TRemChkptDirList[dindex] == NULL)
      {
      if ((LOGLEVEL >= 10) && (dindex == 0))
        {
        sprintf(log_buffer, "NO remote checkpoint directories configured");
        log_ext(-1, __func__, log_buffer, LOG_DEBUG);
        }

      break;
      }

    if (!strncasecmp(TRemChkptDirList[dindex],
        ckpt_path,
        strlen(TRemChkptDirList[dindex])) ||
        !strcmp(TRemChkptDirList[dindex], "*"))
      {
      if (LOGLEVEL >= 10)
        {
        snprintf(log_buffer, sizeof(log_buffer),
          "Checkpoint file %s matched in remote directory %s\n",
          ckpt_path,
          TRemChkptDirList[dindex]);
        log_ext(-1, __func__, log_buffer, LOG_DEBUG);
        }

      return(TRUE);
      }
    }  /* END for (dindex) */

  return(FALSE);
  }  /* END in_remote_checkpoint_dir() */



/**
 * delete_blcr_checkpoint_files
 *
 * This routine is called to remove a checkpoint file / directory
 *
 * @param pjob Pointer to the job structure
 */

void delete_blcr_checkpoint_files(

  job *pjob)

  {
  char namebuf[MAXPATHLEN+1];

  if ((pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_flags & ATR_VFLAG_SET) == 0)
    {
    if (LOGLEVEL > 7)
      {
      sprintf(log_buffer,
        "No checkpoint directory specified for %s\n", pjob->ji_qs.ji_jobid);
      log_ext(-1, __func__, log_buffer, LOG_DEBUG);
      }

    return;
    }

  if (pjob->ji_wattr[JOB_ATR_checkpoint_name].at_flags & ATR_VFLAG_SET)
    {
    /*
    * Get jobs checkpoint directory.
    */
    get_chkpt_dir_to_use(pjob, namebuf, sizeof(namebuf));

    /*
     * we need to clean up the checkpoint job directory
     * but not if it is in a remotely mounted directory
     */

    if (!in_remote_checkpoint_dir(namebuf))
      {
      if (LOGLEVEL >= 7)
        {
        sprintf(log_buffer,
          "remtree for checkpoint %s\n",namebuf);
        log_ext(-1, __func__, log_buffer, LOG_DEBUG);
        }
      remtree(namebuf);
      }
    }

  return;
  }  /* END delete_blcr_checkpoint_files() */




void get_blcr_chkpt(

  job_file_delete_info *jfdi,      /* I */
  char                 *chkpt_dir) /* O */

  {
  char job_dir[MAXPATHLEN+1];

  sprintf(job_dir,"%s%s",
    jfdi->prefix,
    JOB_CHECKPOINT_SUFFIX);
  
  strcpy(chkpt_dir, jfdi->checkpoint_dir);
  
  if ((strlen(jfdi->checkpoint_dir) <= strlen(job_dir)) ||
      (strcmp(job_dir, &jfdi->checkpoint_dir[strlen(jfdi->checkpoint_dir) - strlen(job_dir)])))
    {
    if (chkpt_dir[strlen(chkpt_dir) - 1] != '/')
      {
      strcat(chkpt_dir, "/");
      }
    
    strcat(chkpt_dir, job_dir);
    }
  } /* END get_blcr_chkpt() */




void delete_blcr_files(

  job_file_delete_info *jfdi)

  {
  char namebuf[MAXPATHLEN+1];

  if (jfdi->checkpoint_dir == NULL)
    {
    if (LOGLEVEL > 7)
      {
      sprintf(log_buffer,
        "No checkpoint directory specified for %s\n", jfdi->jobid);
      log_ext(-1, __func__, log_buffer, LOG_DEBUG);
      }

    return;
    }

  /* Get jobs checkpoint directory. */
  get_blcr_chkpt(jfdi, namebuf);
  
  /*
   * we need to clean up the checkpoint job directory
   * but not if it is in a remotely mounted directory
   */
  if (!in_remote_checkpoint_dir(namebuf))
    {
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buffer, "remtree for checkpoint %s\n", namebuf);
      log_ext(-1, __func__, log_buffer, LOG_DEBUG);
      }
    remtree(namebuf);
    }

  } /* END delete_blcr_files() */




/**
 * mom_checkpoint_delete_files
 *
 * This routine is called from the mom_job_purge routine
 * which cleans up all files related to a job.
 *
 * @param pjob Pointer to the job structure
 * @see mom_job_purge
 */

void mom_checkpoint_delete_files(

  job_file_delete_info *jfdi)

  {
  char namebuf[MAXPATHLEN+1];

  if (checkpoint_system_type == CST_MACH_DEP)
    {
    /* delete any checkpoint file */

    get_jobs_default_checkpoint_dir(jfdi->prefix, namebuf, sizeof(namebuf));

    remtree(namebuf);
    }
  else if (checkpoint_system_type == CST_BLCR)
    {
    delete_blcr_files(jfdi);
    }
  return;
  }  /* END mom_checkpoint_delete_files() */





/**
 * mom_checkpoint_recover
 *
 * This routine is called from init_abort_jobs which in turn is called
 * on mom startup.  The purpose is to recover jobs listed in the mom_priv/jobs
 * directory.
 *
 * This routine does not actually start the job.  This happens in start_exec.c.
 * It's purpose is to remove a partially completed checkpoint directory,
 * signified by the name suffix of ".old".
 *
 * @param pjob Pointer to job data structure
 * @see init_abort_jobs
 */

void mom_checkpoint_recover(

  job *pjob)

  {
  char           path[MAXPATHLEN + 1];
  char           oldp[MAXPATHLEN + 1];

  struct stat statbuf;


  if (checkpoint_system_type == CST_MACH_DEP)
    {
    /*
    ** Check to see if a checkpoint.old dir exists.
    ** If so, remove the regular checkpoint dir
    ** and rename the old to the regular name.
    */

    get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, path, sizeof(path));
    snprintf(oldp, sizeof(oldp), "%s", path);
    strncat(oldp, ".old", sizeof(oldp) - strlen(oldp) - 1);

    if (stat(oldp, &statbuf) == 0)
      {
      remtree(path);

      if (rename(oldp, path) == -1)
        remtree(oldp);
      }
    }

  return;
  }





/**
 * mom_checkpoint_check_periodic_timer
 *
 * This routine is called from the main loop routine examine_all_running_jobs.
 * Each job that is checkpointable will have timer variables set up.
 * This routine checks the timer variables and if set and it is time
 * to do a checkpoint, fires the code that starts a checkpoint.
 *
 * @param pjob Pointer to the job structure
 * @see examine_all_running_jobs
 * @see main_loop
 */

void mom_checkpoint_check_periodic_timer(

  job *pjob)

  {
  resource *prwall;
  extern int start_checkpoint(job *pjob, int abort, struct batch_request *preq);
  int rc;
  static resource_def *rdwall;

  /* see if need to checkpoint any job */

  if (pjob->ji_checkpoint_time != 0)  /* ji_checkpoint_time gets set below */
    {
    if (rdwall == NULL)
      {
      rdwall = find_resc_def(svr_resc_def, "walltime", svr_resc_size);
      }

    if (rdwall != NULL)
      {
      prwall = find_resc_entry(
                 &pjob->ji_wattr[JOB_ATR_resc_used],
                 rdwall);  /* resource definition cput set in startup */

      if (prwall &&
          (prwall->rs_value.at_val.at_long >= pjob->ji_checkpoint_next))
        {
        pjob->ji_checkpoint_next =
          prwall->rs_value.at_val.at_long +
          pjob->ji_checkpoint_time;

        if ((rc = start_checkpoint(pjob, 0, 0)) != PBSE_NONE)
          {
          sprintf(log_buffer, "Checkpoint failed, error %d", rc);

          message_job(pjob, StdErr, log_buffer);

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

  return;
  }  /* END mom_checkpoint_check_periodic_timer() */



/*
 * establish_server_connection()
 *
 * establishes a conncection to pjob's server if possible.
 * @pre-cond pjob must have a server attribute and be a valid job
 * @param - pjob the job whose server we should connect to
 * @return - the index of the connection to pjob's server, or -1 on error
 */

int establish_server_connection(
    
  job *pjob)

  {
  int local_errno = 0;
  int connection  = -1;
  int sock;

  if ((pjob == NULL) ||
      (pjob->ji_wattr[JOB_ATR_at_server].at_val.at_str == NULL))
    return(connection);

  sock = mom_open_socket_to_jobs_server(pjob, __func__, NULL);

  if (sock >= 0)
    connection = socket_to_handle(sock, &local_errno);
  
  if (connection < 0)
    {
    sprintf(log_buffer,"Job %s failed %d times to get connection to %s",
      pjob->ji_qs.ji_jobid,
      MAX_CONN_RETRY,
      pjob->ji_wattr[JOB_ATR_at_server].at_val.at_str);
    log_err(-1, __func__, log_buffer);
    }

  return(connection);
  } /* END establish_server_connection() */



/**
 * blcr_checkpoint_job
 *
 * This routine lauches the checkpoint script for a BLCR
 * checkpoint system.
 * currently only supports single process job, so a BLCR job will
 * only have one task associated with the job.
 *
 * @see start_checkpoint() - parent
 * @pre-cond: this is a child routine, not the main pbs_mom process
 * @pre-cond: pjob is a valid job pointer
 * @pre-cond: pjob has a valid checkpoint directory in its attributes
 *
 * @returns PBSE_NONE if no error
 */


int blcr_checkpoint_job(

  job                  *pjob,  /* I */
  int                   abort, /* I */
  struct batch_request *preq)  /* may be null */

  {
  char             sid[20];
  char            *arg[20];
  char             buf[1024];
  int              len;
  char           **ap;
  FILE            *fs;
  char            *cmd;
  int              rc;
  int              request_type = 0;
  char             err_buf[4098];
  char             line[1028];
  char             namebuf[MAXPATHLEN+1];
  int              conn = -1;
  int              err;
  int              local_errno = 0;
  struct attrl    *attrib = NULL;
  time_t           epoch;
  unsigned short   momport = 0;

  if ((pjob == NULL) ||
      (pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str == NULL))
    exit(-1);

  err_buf[0] = '\0';
  /*
  * Get jobs checkpoint directory.
  */
  get_chkpt_dir_to_use(pjob, namebuf, sizeof(namebuf));

  /* Make sure that the specified directory exists. */

  if (mkdir(namebuf, 0755)
      == 0)
    {
    /* Change the owner of the checkpoint directory to be the user */
    if (chown(namebuf,
          pjob->ji_qs.ji_un.ji_momt.ji_exuid,
          pjob->ji_qs.ji_un.ji_momt.ji_exgid) == -1)
      {
      log_err(errno, __func__, (char *)"cannot change checkpoint directory owner");
      }
    }
  
  /* if a checkpoint script is defined launch it */

  if (checkpoint_script_name[0] == '\0')
    {
    log_err(PBSE_RMEXIST, __func__, (char *)"No checkpoint script defined");

    if (preq != NULL)
      {
      req_reject(PBSE_RMEXIST,PBS_CHECKPOINT_MIGRATE,preq,NULL,NULL);
      }
    exit(PBSE_RMEXIST);
    }

  /* Checkpoint successful (assumed) */
  pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_FILE;

  if (multi_mom)
    {
    momport = pbs_rm_port;
    }
  job_save(pjob,SAVEJOB_FULL, momport); /* to save resources_used so far */

  sprintf(log_buffer,"checkpointed to %s / %s at %ld",
    namebuf,
    pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str,
    pjob->ji_wattr[JOB_ATR_checkpoint_time].at_val.at_long);

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

  sprintf(sid,"%ld",
    pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long);

  arg[0] = checkpoint_script_name;
  arg[1] = sid;
  arg[2] = SET_ARG(pjob->ji_qs.ji_jobid);
  arg[3] = SET_ARG(pjob->ji_wattr[JOB_ATR_euser].at_val.at_str);
  arg[4] = SET_ARG(pjob->ji_wattr[JOB_ATR_egroup].at_val.at_str);
  arg[5] = SET_ARG(namebuf);
  arg[6] = SET_ARG(pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
  arg[7] = (abort) ? (char *)"15" /*abort*/ : (char *)"0" /*run/continue*/;
  arg[8] = SET_ARG(csv_find_value(pjob->ji_wattr[JOB_ATR_checkpoint].at_val.at_str, (char *)"depth"));
  arg[9] = NULL;

  /* XXX this should be fixed to make sure there is no chance of a buffer overrun */
  snprintf(buf, sizeof(buf), "checkpoint args:");
  cmd = buf + strlen("checkpoint args: "); /* this extra space compared to above is intentional... */

  for (ap = arg; *ap; ap++)
    {
    len = strlen(buf);
    snprintf(buf + len, sizeof(buf) - len, " %s", *ap);
    }

  len = strlen(buf);
  snprintf(buf + len, sizeof(buf) - len, "  2>&1 1>/dev/null");

  log_ext(-1, __func__, buf, LOG_DEBUG);
  if (preq != NULL)
    {
    request_type = preq->rq_type;
    reply_ack(preq);
    }

  /* execv(arg[0], arg); */
  /* change execv to popen so we can grab the stderr */

  fs = popen(cmd, "r"); /* create a read pipe for the command */
  rc = 0;
  if (fs == NULL)
    {
    sprintf(buf, "error executing checkpoint script");
    log_err(errno, __func__, buf);
    rc = -1;
    }
  else
    {
    err_buf[0] = '\0';
    while (fgets(line, 1024, fs) != NULL &&
            strlen(err_buf) + strlen(line) + 1 < 4098)
      {
      strcat(err_buf, line);
      }

    rc = pclose(fs);
    if (rc != -1)
      {
      rc = WEXITSTATUS(rc);
      }

    if (rc != 0)
      {
      sprintf(buf, "checkpoint script returned value %d\n", rc);
      log_err(-1, __func__, buf);
      }
    }

  if (rc != 0)
    {
    /*
     * If the checkpoint script did not return data for the err_buf,
     * fill it in so we can show that something went wrong
     */

    if (strlen(err_buf) == 0)
      {
      sprintf(err_buf,"Checkpoint script failed with return value of %d", rc);
      }

    /* checkpoint script returned a non-zero value.  We assume the checkpoint
        failed */

    /* remove checkpoint directory that was created for this checkpoint attempt */

    sprintf(buf, "Checkpoint failed for job %s, removing checkpoint directory\n",
        pjob->ji_qs.ji_jobid);
    log_ext(-1, __func__, buf, LOG_DEBUG);

    delete_blcr_checkpoint_files(pjob);

    conn = establish_server_connection(pjob);

    if (conn >= 0)
      {
      set_attr(&attrib, ATTR_comment, err_buf);
      
      err = pbs_alterjob_err(conn, pjob->ji_qs.ji_jobid, attrib, NULL, &local_errno);

      if (err != 0)
        {
        sprintf(buf, "pbs_alterjob requested on job %s failed (%d-%s)\n",
            pjob->ji_qs.ji_jobid, err, pbs_strerror(err));
        log_err(-1, __func__, buf);
        if (err == PBSE_UNKJOBID)
          {
          /* TODO: GB - can the job exit while waiting for the checkpoint
              script to exit?? call log_err */
          pbs_disconnect(conn);
          exit(err);
          }
        }

      if (abort != 0)
        {
        /*
         * we need to tell the server to release the hold (abort is non-zero
         * which means we are trying to hold the job)
         */


        /*
         * send release job request, the job will still be running,
         * so it shouldn't have any holds set so we will send "uos"
         * to clear all holds
         */
        pbs_rlsjob_err(conn, pjob->ji_qs.ji_jobid, (char *)"uos", NULL, &local_errno);

        } /* END if (abort != 0) */


      pbs_disconnect(conn);
      }

    } /* END if (rc != 0) */
  else
    {
    char timestr[80];
    /* checkpoint script returned a zero value.  We assume the checkpoint
        suceeded */

    conn = establish_server_connection(pjob);

    if (conn >= 0)
      {
      sprintf(timestr,"%ld",
          (long)pjob->ji_wattr[JOB_ATR_checkpoint_time].at_val.at_long);
      epoch = (time_t)pjob->ji_wattr[JOB_ATR_checkpoint_time].at_val.at_long;

      sprintf(err_buf,"Job %s was checkpointed and %s to %s/%s at %s",
        pjob->ji_qs.ji_jobid,
        (request_type == PBS_BATCH_HoldJob) ? "terminated" : "continued",
        namebuf,
        pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str,
        ctime(&epoch));

      set_attr(&attrib, ATTR_comment, err_buf);
      set_attr(&attrib, ATTR_checkpoint_name,
          pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
      set_attr(&attrib, ATTR_checkpoint_time, timestr);

      err = pbs_alterjob_err(conn, 
          pjob->ji_qs.ji_jobid, attrib,
          (request_type == PBS_BATCH_HoldJob) ? (char *)CHECKPOINTHOLD : (char *)CHECKPOINTCONT,
          &local_errno);

      if (err != 0)
        {
        sprintf(buf, "pbs_alterjob requested on job %s failed (%d-%s)\n",
            pjob->ji_qs.ji_jobid, err, pbs_strerror(err));
        log_err(-1, __func__, buf);
        if (err == PBSE_UNKJOBID)
          {
          /* TODO: GB - can the job exit while waiting for the checkpoint
              script to exit?? call log_err */
          pbs_disconnect(conn);

          /*
           * If we get an unknown jobid after succesfully doing a non-hold
           * checkpoint, then it is most likely the result of a periodic
           * checkpoint for a job that had a qdel -p done, so we get rid of
           * any local checkpoint / restart files
           */

          if (request_type == 0)
            {
            delete_blcr_checkpoint_files(pjob);
            }
          exit(err);
          }
        }

      pbs_disconnect(conn);

      /* Normally, this is an empty routine and does nothing. */
      rc = site_mom_postchk(pjob,abort);
      }
    }

  exit (rc);
  }  /* END blcr_checkpoint_job() */





/*
 * Checkpoint the job.
 *
 * If abort is TRUE, kill it too.  Return a PBS error code.
 */

int mom_checkpoint_job(

  job *pjob,  /* I */
  int  abort) /* I */

  {
  int             hasold = 0;
  int             sesid = -1;
  int             ckerr;
  unsigned short  momport = 0;

  struct stat     statbuf;
  char            path[MAXPATHLEN + 1];
  char            oldp[MAXPATHLEN + 1];
  char            file[MAXPATHLEN + 1];
  task           *ptask;

  assert(pjob != NULL);

  get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, path, sizeof(path));

  if (stat(path, &statbuf) == 0)
    {
    strcpy(oldp, path);  /* file already exists, rename it */

    strcat(oldp, ".old");

    if (rename(path, oldp) < 0)
      {
      return(errno);
      }

    hasold = 1;
    }

  if (mkdir(path, 0755) == -1)
    {
    if (errno != EEXIST)
      {
      snprintf(log_buffer, sizeof(log_buffer), "Couldn't create directory '%s'", path);
      log_err(errno, __func__, log_buffer);

      return(errno);
      }
    }

  strcpy(file, path);

#ifdef _CRAY

  /*
   * if job is suspended and if <abort> is set, resume job first,
   * this is so job will be "Q"ueued and then back into "R"unning
   * when restarted.
   */

  if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_Suspend) && abort)
    {
    for (ptask = (task *)GET_NEXT(pjob->ji_tasks);
         ptask != NULL;
         ptask = (task *)GET_NEXT(ptask->ti_jobtask))
      {
      sesid = ptask->ti_qs.ti_sid;

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

      /* XXX: What to do if some resume work and others don't? */

      if ((ckerr = resume(C_JOB, sesid)) == 0)
        {
        post_resume(pjob, ckerr);
        }
      else
        {
        sprintf(log_buffer, "checkpoint failed: errno=%d sid=%d",
                errno,
                sesid);

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

        return(errno);
        }
      }
    }

#endif /* _CRAY */

  for (ptask = (task *)GET_NEXT(pjob->ji_tasks);
       ptask != NULL;
       ptask = (task *)GET_NEXT(ptask->ti_jobtask))
    {
    sesid = ptask->ti_qs.ti_sid;

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

    if (mach_checkpoint(ptask, file, abort) == -1)
      goto fail;
    }

  /* Checkpoint successful */

  pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_FILE;

  if (multi_mom)
    {
    momport = pbs_rm_port;
    }
  job_save(pjob, SAVEJOB_FULL, momport); /* to save resources_used so far */

  sprintf(log_buffer, "checkpointed to %s",
          path);

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

  if (hasold)
    remtree(oldp);

  return(PBSE_NONE);

fail:

  /* A checkpoint has failed.  Log and return error. */

  ckerr = errno;

  sprintf(log_buffer, "checkpoint failed:errno=%d sid=%d",
          errno,
          sesid);

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

  /*
  ** See if any checkpoints worked and abort is set.
  ** If so, we need to restart these tasks so the whole job is
  ** still running.  This has to wait until we reap the
  ** aborted task(s).
  */

  if (abort)
    {
    return(PBSE_CKPSHORT);
    }

  /* Clean up files */
  remtree(path);

  if (hasold)
    {
    if (rename(oldp, path) == -1)
      pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHECKPOINT_FILE;
    }

  if (ckerr == EAGAIN)
    {
    return(PBSE_CKPBSY);
    }

  return(ckerr);
  }  /* END mom_checkpoint_job() */






/*
 * post_checkpoint - post processor for start_checkpoint()
 *
 * @see scan_for_terminated() - parent
 *
 * Called from scan_for_terminated() when found in ji_mompost;
 *
 * This sets the "has checkpoint image" bit in the job.
 *
 * job is referenced by parent after calling this routine - do not 'purge'
 * job from inside this routine
 * @param exit_code - exit code. 
 */

void post_checkpoint(

  job *pjob,  /* I - may be purged */
  int  exit_code)    /* I */

  {
  char           path[MAXPATHLEN + 1];
  DIR           *dir;

  struct dirent *pdir;
  tm_task_id     tid;
  task          *ptask;

  int            abort = pjob->ji_flags & MOM_CHECKPOINT_ACTIVE;

  exiting_tasks = 1; /* make sure we call scan_for_exiting() */

  pjob->ji_flags &= ~MOM_CHECKPOINT_ACTIVE;

  /* exit value == 0 means checkpointing worked correctly */
  if (exit_code == 0)
    {
    pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_FILE;

    return;
    }

  /* since checkpointing failed, clear out checkpoint name and time */
  
  pjob->ji_wattr[JOB_ATR_checkpoint_name].at_flags = 0;
  pjob->ji_wattr[JOB_ATR_checkpoint_time].at_flags = 0;

  /*
  ** If we get here, an error happened.  Only try to recover
  ** if we had abort set.
  */

  if (abort == 0)
    {
    return;
    }

  /*
  ** Set a flag for scan_for_exiting() to be able to
  ** deal with a failed checkpoint rather than doing
  ** the usual processing.
  */

  pjob->ji_flags |= MOM_CHECKPOINT_POST;

  /*
  ** Set the TI_FLAGS_CHECKPOINT flag for each task that
  ** was checkpointed and aborted.
  */

  get_chkpt_dir_to_use(pjob, path, sizeof(path));

  dir = opendir(path);

  if (dir == NULL)
    {
    return;
    }

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

    tid = atoi(pdir->d_name);

    if (tid == 0)
      continue;

    ptask = task_find(pjob,tid);

    if (ptask == NULL)
      continue;

    ptask->ti_flags |= TI_FLAGS_CHECKPOINT;
    }  /* END while ((pdir = readdir(dir)) != NULL) */

  closedir(dir);

  return;
  }  /* END post_checkpoint() */





/**
 * start_checkpoint - start a checkpoint going
 *
 * checkpoint done from a child because it takes a while
 *
 * @see blcr_checkpoint() - child
 * @see start_checkpoint() - parent
 */

int start_checkpoint(

  job                  *pjob,
  int                   abort, /* I - boolean - 0 or 1 */
  struct batch_request *preq)  /* may be null */

  {

  pid_t     pid;
  int       rc = PBSE_NONE;
  char      name_buffer[MAXPATHLEN + 1];
  time_t    time_now;

  time_now = time((time_t *)0);

  switch (checkpoint_system_type)

    {
    case CST_MACH_DEP:

      /* NO-OP */

      break;

    case CST_BLCR:

      /* Build the name of the checkpoint file before forking to the child because
       * we want this name to persist and this won't work if we are the child.
       * Notice that the ATR_VFLAG_SEND bit is not set. We don't want this to go
       * to the pbs_server until the checkpoint has completed successfully.
       */

      sprintf(name_buffer,"ckpt.%s.%d",
        pjob->ji_qs.ji_jobid,
        (int)time_now);

      decode_str(&pjob->ji_wattr[JOB_ATR_checkpoint_name], NULL, NULL, name_buffer, 0);

      pjob->ji_wattr[JOB_ATR_checkpoint_name].at_flags =
        ATR_VFLAG_SET | ATR_VFLAG_MODIFY;

      /* Set the checkpoint time so can determine if the checkpoint is recent */
      pjob->ji_wattr[JOB_ATR_checkpoint_time].at_val.at_long = (long)time_now;
      pjob->ji_wattr[JOB_ATR_checkpoint_time].at_flags =
        ATR_VFLAG_SET | ATR_VFLAG_MODIFY;

      /* For BLCR, there must be a directory name in the job attributes. */

      if (!(pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_flags & ATR_VFLAG_SET))
        {
        /* No dir specified, use the default job checkpoint directory
           e.g.  /var/spool/torque/checkpoint/42.host.domain.CK */

        get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, name_buffer, sizeof(name_buffer));

        decode_str(&pjob->ji_wattr[JOB_ATR_checkpoint_dir],NULL,NULL,name_buffer,0);
        }

      break;

    case CST_NONE:

    default:

      return(PBSE_NOSUP);  /* no checkpoint, reject request */

      /*NOTREACHED*/

      break;
    }

  /* now set up as child of MOM */
  pid = fork_me((preq == NULL) ? -1 : preq->rq_conn);

  if (pid > 0)
    {
    /* parent */

    /* MOM_CHECKPOINT_ACTIVE prevents scan_for_exiting from triggering obits while job is checkpointing. */

    pjob->ji_flags |= MOM_CHECKPOINT_ACTIVE;
    pjob->ji_momsubt = pid; /* record pid in job for when child terminates */

    /* Set the address of a function to execute in scan_for_terminated */

    pjob->ji_mompost = (int (*)(job *,int))post_checkpoint;

    if (preq)
      free_br(preq); /* child will send reply */

    }
  else if (pid < 0)
    {
    /* error on fork */

    log_err(errno, __func__, (char *)"cannot fork child process for checkpoint");

    return(PBSE_SYSTEM);
    }
  else
    {
    /* child - does the checkpoint */

    switch (checkpoint_system_type)

      {
      case CST_MACH_DEP:

        rc = mom_checkpoint_job(pjob,abort);

        break;

      case CST_BLCR:

        /* we don't return from here, so we can do checkpoint in this process id */
        rc = blcr_checkpoint_job(pjob,abort,preq);

        break;
      }

    if (rc == PBSE_NONE)
      {
      rc = site_mom_postchk(pjob,abort); /* Normally, this is an empty routine and does nothing. */
      }

    if (preq != NULL)
      {
      /* rc may be 0, req_reject is used to pass auxcode */

      req_reject(rc,PBS_CHECKPOINT_MIGRATE,preq,NULL,NULL); /* BAD reject is used to send OK??? */
      }


    exit(rc); /* zero exit tells main checkpoint ok */
    }

  return(PBSE_NONE);  /* parent return */
  }  /* END start_checkpoint() */





/*
** Restart each task which has exited and has TI_FLAGS_CHECKPOINT turned on.
** If all tasks have been restarted, turn off MOM_CHECKPOINT_POST.
*/

void checkpoint_partial(

  job *pjob)

  {
  char  namebuf[MAXPATHLEN];
  task  *ptask;
  int  texit = 0;

  assert(pjob != NULL);

  get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, namebuf, sizeof(namebuf));

  for (ptask = (task *)GET_NEXT(pjob->ji_tasks);
       ptask != NULL;
       ptask = (task *)GET_NEXT(ptask->ti_jobtask))
    {
    /*
    ** See if the task was marked as one of those that did
    ** actually checkpoint.
    */

    if ((ptask->ti_flags & TI_FLAGS_CHECKPOINT) == 0)
      continue;

    texit++;

    /*
    ** Now see if it was reaped.  We don't want to
    ** fool with it until we see it die.
    */

    if (ptask->ti_qs.ti_status != TI_STATE_EXITED)
      continue;

    texit--;

    if (mach_restart(ptask, namebuf) == -1)
      {
      pjob->ji_flags &= ~MOM_CHECKPOINT_POST;
      kill_job(pjob, SIGKILL, __func__, "failed to restart");
      return;
      }

    ptask->ti_qs.ti_status = TI_STATE_RUNNING;

    ptask->ti_flags &= ~TI_FLAGS_CHECKPOINT;

    task_save(ptask);
    }

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

    struct stat statbuf;

    /*
    ** All tasks should now be running.
    ** Turn off MOM_CHECKPOINT_POST flag so job is back to where
    ** it was before the bad checkpoint attempt.
    */

    pjob->ji_flags &= ~MOM_CHECKPOINT_POST;

    /*
    ** Get rid of incomplete checkpoint directory and
    ** move old checkpoint dir back to regular if it exists.
    */

    remtree(namebuf);
    strcpy(oldname, namebuf);
    strcat(oldname, ".old");

    if (stat(oldname, &statbuf) == 0)
      {
      if (rename(oldname, namebuf) == -1)
        pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHECKPOINT_FILE;
      }
    }
  }  /* END checkpoint_partial() */







/* BLCR version of restart */

int blcr_restart_job(

  job *pjob)

  {
  int            pid;
  char           sid[20];
  char          *arg[20];
  extern char    restart_script_name[MAXPATHLEN + 1];
  task          *ptask;
  char           buf[1024];
  char           namebuf[MAXPATHLEN + 1];
  char           restartfile[MAXPATHLEN + 1];
  char           script_buf[MAXPATHLEN + 1];
#ifdef USEJOBCREATE
	uint64_t       job_id;
#endif /* USEJOBCREATE */
  char          *rsv_id = NULL;
  long long      pagg = 0;


  /* if a restart script is defined launch it */

  if (restart_script_name[0] == '\0')
    {
    log_err(PBSE_RMEXIST, __func__, (char *)"No restart script defined");

    return(PBSE_RMEXIST);
    }

  /* BLCR is not for parallel jobs, there can only be one task in the job. */

  ptask = (task *) GET_NEXT(pjob->ji_tasks);

  if (ptask == NULL)
    {


    /* turns out if we are restarting a complete job then ptask will be
       null and we need to create a task We'll just create one task*/
    if ((ptask = pbs_task_create(pjob, TM_NULL_TASK)) == NULL)
      {
      log_err(PBSE_RMNOPARAM, __func__, (char *)"Job has no tasks");
      return(PBSE_RMNOPARAM);
      }

    strcpy(ptask->ti_qs.ti_parentjobid, pjob->ji_qs.ji_jobid);

    ptask->ti_qs.ti_parentnode = 0;
    ptask->ti_qs.ti_parenttask = 0;
    ptask->ti_qs.ti_task = 0;


    }

#ifdef USEJOBCREATE
  /*
   * Get a job id from the system
   */
  job_id  = get_jobid(pjob->ji_qs.ji_jobid);

  pjob->ji_wattr[JOB_ATR_pagg_id].at_val.at_ll = job_id;
  pjob->ji_wattr[JOB_ATR_pagg_id].at_flags = ATR_VFLAG_SET | ATR_VFLAG_MODIFY;
  if (is_login_node)
    pagg = job_id;
#endif /* USEJOBCREATE */

  /* launch the script and return success */

  pid = fork();

  if (pid < 0)
    {
    /* fork failed */

    return(PBSE_RMSYSTEM);
    }
  else if (pid > 0)
    {
    /* parent */

    ptask->ti_qs.ti_sid = pid;  /* Apparently torque doesn't do anything with the session ID that we pass back here... */

    ptask->ti_qs.ti_status = TI_STATE_RUNNING;
    task_save(ptask);

    return(PBSE_NONE);
    }
  else if (pid == 0)
    {
    /* child: execv the script */

    /* if there are missing .OU or .ER files create them, they were probably
       empty and server didn't send them */
    /* TODO: check return value? */
    create_missing_files(pjob);

    /*
    * Get jobs checkpoint directory.
    */
    get_chkpt_dir_to_use(pjob, namebuf, sizeof(namebuf));

    /* Change the owner of the .SC to be the user */
    snprintf(script_buf, sizeof(script_buf), "%s%s%s",
      path_jobs, pjob->ji_qs.ji_fileprefix, JOB_SCRIPT_SUFFIX);

    if (chown(script_buf,
          pjob->ji_qs.ji_un.ji_momt.ji_exuid,
          pjob->ji_qs.ji_un.ji_momt.ji_exgid) == -1)
      {
      sprintf(log_buffer,"cannot change owner for file %s", script_buf);
      log_err(errno, __func__, log_buffer);
      }

    snprintf(restartfile, sizeof(restartfile), "%s/%s",
      namebuf, pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
   
    /* Change the owner of the checkpoint restart file to be the user */
    if (chown(restartfile,
          pjob->ji_qs.ji_un.ji_momt.ji_exuid,
          pjob->ji_qs.ji_un.ji_momt.ji_exgid) == -1)
      {
      log_err(errno, __func__, (char *)"cannot change checkpoint restart file owner");
      }


    sprintf(sid, "%ld",
            pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long);

    arg[0] = restart_script_name;
    arg[1] = sid;
    arg[2] = SET_ARG(pjob->ji_qs.ji_jobid);
    arg[3] = SET_ARG(pjob->ji_wattr[JOB_ATR_euser].at_val.at_str);
    arg[4] = SET_ARG(pjob->ji_wattr[JOB_ATR_egroup].at_val.at_str);
    arg[5] = SET_ARG(namebuf);
    arg[6] = SET_ARG(pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str);
    arg[7] = NULL;

    snprintf(buf, sizeof(buf), "restart args: %s %s %s %s %s %s %s",
      restart_script_name, sid, pjob->ji_qs.ji_jobid,
      SET_ARG(pjob->ji_wattr[JOB_ATR_euser].at_val.at_str),
      SET_ARG(pjob->ji_wattr[JOB_ATR_egroup].at_val.at_str),
      namebuf,
      SET_ARG(pjob->ji_wattr[JOB_ATR_checkpoint_name].at_val.at_str));

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

    log_close(0);

    if (lockfds >= 0)
      {
      close(lockfds);
      lockfds = -1;
      }

    net_close(-1);

    /* set us up with a new session */
    pid = setsid();

    if (is_login_node == TRUE)
      {
      int       use_nppn = TRUE;
      int       mppdepth = 0;
      char     *mppnodes = NULL;
      int       nppcu = APBASIL_DEFAULT_NPPCU_VALUE; /* default */
      resource *pres = find_resc_entry(
                         &pjob->ji_wattr[JOB_ATR_resource],
                         find_resc_def(svr_resc_def, "procs", svr_resc_size));
     
      if ((pres != NULL) &&
          (pres->rs_value.at_val.at_long != 0))
        use_nppn = FALSE;

      pres = find_resc_entry(&pjob->ji_wattr[JOB_ATR_resource],
                             find_resc_def(svr_resc_def, "mppdepth", svr_resc_size));
      if ((pres != NULL) &&
          (pres->rs_value.at_val.at_long != 0))
        mppdepth = pres->rs_value.at_val.at_long;

      /* look up job nppcu value if it exists */
      if ((pjob->ji_wattr[JOB_ATR_nppcu].at_flags & ATR_VFLAG_SET))
              nppcu = pjob->ji_wattr[JOB_ATR_nppcu].at_val.at_long;

      /* get the mppnodes if it exists */
      pres = find_resc_entry(
             &pjob->ji_wattr[JOB_ATR_resource],
             find_resc_def(svr_resc_def, "mppnodes", svr_resc_size));

      if ((pres != NULL) &&
          (pres->rs_value.at_val.at_str != NULL))
        {
        mppnodes = strdup(pres->rs_value.at_val.at_str);
        }


      if (create_alps_reservation(pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str,
            pjob->ji_wattr[JOB_ATR_job_owner].at_val.at_str,
            pjob->ji_qs.ji_jobid,
            apbasil_path,
            apbasil_protocol,
            pagg,
            use_nppn,
            nppcu,
            mppdepth,
            &rsv_id,
            mppnodes) != PBSE_NONE)
        {
        snprintf(log_buffer, sizeof(log_buffer),
          "Couldn't create the reservation for job %s",
          pjob->ji_qs.ji_jobid);
        log_err(-1, __func__, log_buffer);
        }
      
      if (rsv_id != NULL)
        {
        pjob->ji_wattr[JOB_ATR_reservation_id].at_flags = ATR_VFLAG_SET;
        pjob->ji_wattr[JOB_ATR_reservation_id].at_val.at_str = rsv_id;
        }
      if(mppnodes != NULL) free(mppnodes);
      }

    if (pid < 0)
      {
      perror("setsid");
      exit(-1);
      }

#ifdef ENABLE_CSA
      /*
	     * Add a workload management start record
	     */

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

#endif /* ENABLE_CSA */

    execv(arg[0], arg);
    }  /* END if (pid == 0) */

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





/* start each task based on task checkpoint records located job-specific checkpoint directory */

int mom_restart_job(
    
  job  *pjob)

  {
  char           path[MAXPATHLEN];
  int            i;
  char           namebuf[MAXPATHLEN];
  char          *filnam;
  DIR           *dir;

  struct dirent *pdir;
  tm_task_id     taskid;
  task          *ptask;
  int            tcount = 0;
  long  mach_restart(task *, char *path);

  if (pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str != NULL)
    snprintf(path, sizeof(path), "%s", pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str);
  else
    {
    /* we can't do anything if there's no checkpoint directory specified here */
    return(PBSE_RMEXIST);
    }

  get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, namebuf, sizeof(namebuf));

  if ((dir = opendir(path)) == NULL)
    {
    sprintf(log_buffer, "opendir %s",
      path);

    log_err(errno, __func__, log_buffer);

    return(PBSE_RMEXIST);
    }

  strcpy(namebuf, path);

  strcat(namebuf, "/");

  i = strlen(namebuf);

  filnam = &namebuf[i];

  while ((pdir = readdir(dir)) != NULL)
    {
    if (strlen(pdir->d_name) <= 2)
      continue;

    if ((taskid = (tm_task_id)atoi(pdir->d_name)) == 0)
      {
      sprintf(log_buffer, "%s: garbled filename %s",
              pjob->ji_qs.ji_jobid,
              pdir->d_name);

      goto fail;
      }

    if ((ptask = task_find(pjob, taskid)) == NULL)
      {
      sprintf(log_buffer, "%s: task %d not found",
              pjob->ji_qs.ji_jobid,
              (int)taskid);

      goto fail;
      }

    strcpy(filnam, pdir->d_name);

    if (mach_restart(ptask, namebuf) == -1)
      {
      sprintf(log_buffer, "%s: task %d failed from file %s",
              pjob->ji_qs.ji_jobid,
              (int)taskid,
              namebuf);

      goto fail;
      }

    ptask->ti_qs.ti_status = TI_STATE_RUNNING;

    if (LOGLEVEL >= 6)
      {
      log_record(
        PBSEVENT_ERROR,
        PBS_EVENTCLASS_JOB,
        pjob->ji_qs.ji_jobid,
        "task set to running (mom_restart_job)");
      }

    task_save(ptask);

    tcount++;
    }

  closedir(dir);

  sprintf(log_buffer, "Restarted %d tasks",
          tcount);

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

  return(PBSE_NONE);

fail:

  log_err(errno, __func__, log_buffer);

  closedir(dir);

  return(PBSE_RMEXIST);
  }  /* END mom_restart_job() */






/**
 * mom_checkpoint_init_job_periodic_timer
 *
 * The routine is called from TMomFinalizeJob1 in start_exec.c.
 * This code initializes checkpoint related variables in the job struct.
 *
 * @param pjob Pointer to job structure
 * @see TMomFinalizeJob1
 */
void mom_checkpoint_init_job_periodic_timer(

  job *pjob)

  {
  pbs_attribute  *pattr;
  char           *vp;

  /* Should we set up the job for periodic checkpoint? */

  pattr = &pjob->ji_wattr[JOB_ATR_checkpoint];

  if ((pattr->at_flags & ATR_VFLAG_SET) &&
      (csv_find_string(pattr->at_val.at_str, "c") ||
       csv_find_string(pattr->at_val.at_str, "periodic")))
    {
    /* Yes, what is the interval in minutes. */

    if ((vp = csv_find_value(pattr->at_val.at_str, "c")) ||
        (vp = csv_find_value(pattr->at_val.at_str, "interval")))
      {
      /* has checkpoint time (in minutes), convert to seconds */

      pjob->ji_checkpoint_time = atoi(vp) * 60;
      pjob->ji_checkpoint_next = pjob->ji_checkpoint_time;
      }
    else
      {
      /* pick a default number of minutes */

      pjob->ji_checkpoint_time = default_checkpoint_interval * 60;
      pjob->ji_checkpoint_next = pjob->ji_checkpoint_time;
      }
    }
  }



/**
 * mom_checkpoint_job_has_checkpoint
 *
 * The routine is called from TMomFinalizeJob1 in start_exec.c.
 * It checks to see if the job has a checkpoint file to restart from.
 *
 * @param pjob Pointer to job structure
 * @see TMomFinalizeJob1
 */

int mom_checkpoint_job_has_checkpoint(

  job *pjob)

  {
  /* Has the job has been checkpointed? */

  switch (checkpoint_system_type)

    {
    case CST_MACH_DEP:

      if (pjob->ji_qs.ji_svrflags & (JOB_SVFLG_CHECKPOINT_FILE | JOB_SVFLG_CHECKPOINT_MIGRATEABLE))
        {
        char           buf[MAXPATHLEN + 2];

        struct stat    sb;

        /* Does the checkpoint directory exist? */

        if (pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_flags & ATR_VFLAG_SET)
          {
          /* The job has a checkpoint directory specified, use it. */
          snprintf(buf, sizeof(buf), "%s", pjob->ji_wattr[JOB_ATR_checkpoint_dir].at_val.at_str);
          }
        else
          {
          /* Otherwise, use the default job checkpoint directory /var/spool/torque/checkpoint/42.host.domain.CK */

          get_jobs_default_checkpoint_dir(pjob->ji_qs.ji_fileprefix, buf, sizeof(buf));
          }

        if (stat(buf, &sb) != 0) /* stat(buf) tests if the checkpoint directory exists */
          {
          /* We thought there was a checkpoint but the directory was not there. */
          pjob->ji_qs.ji_svrflags &= ~(JOB_SVFLG_CHECKPOINT_FILE | JOB_SVFLG_CHECKPOINT_MIGRATEABLE);
          break;
          }
        }

      if (LOGLEVEL >= 7)
	      {
        log_ext(-1, "mom_checkpoint_job_has_checkpoint", "TRUE", LOG_DEBUG);
	      }

      return(TRUE); /* Yes, there is a checkpoint. */
      break;

    case CST_BLCR:

      if (pjob->ji_wattr[JOB_ATR_checkpoint_name].at_flags & ATR_VFLAG_SET)
        {
        if (LOGLEVEL >= 7)
	        {
          log_ext(-1, "mom_checkpoint_job_has_checkpoint", "TRUE", LOG_DEBUG);
	        }
        return(TRUE);
        }

      break;
    }

  if (LOGLEVEL >= 7)
    {
    log_ext(-1, "mom_checkpoint_job_has_checkpoint", "FALSE", LOG_DEBUG);
    }

  return(FALSE); /* No checkpoint pbs_attribute on job. */
  }





/**
 * mom_checkpoint_start_restart
 *
 * The routine is called from TMomFinalizeJob1 in start_exec.c.
 * This code initializes checkpoint related variables in the job struct.
 * If there is a checkpoint file, the job is restarted from this image.
 *
 * @param pjob Pointer to job structure
 * @see TMomFinalizeJob1
 */

int mom_checkpoint_start_restart(

  job *pjob)

  {
  int            rc = PBSE_NONE;

  /* At this point we believe that there is a checkpoint image, try to restart it. */

  switch (checkpoint_system_type)

    {
    case CST_MACH_DEP:

      /* perform any site required setup before restart, normally empty and does nothing */

      if ((rc = site_mom_prerst(pjob)) != PBSE_NONE)
        {
        return(rc);
        }

      rc = mom_restart_job(pjob); /* Iterate over files in checkpoint dir, restarting all files found. */

      break;

    case CST_BLCR:

      /* NOTE:  partition creation handled in blcr_restart_job() */

      /* make sure we recreate the nodes file, if needed */

	    if (pjob->ji_flags & MOM_HAS_NODEFILE)
        {
	      if (write_nodes_to_file(pjob) == -1)
          {
          return(FAILURE);
          }

	      if (write_attr_to_file(pjob, JOB_ATR_exec_gpus, "gpu") == -1)
          {
          return(FAILURE);
          }
	      
        if (write_attr_to_file(pjob, JOB_ATR_exec_mics, "mic") == -1)
          {
          return(FAILURE);
          }
        }

      /* perform any site required setup before restart, normally empty and does nothing */

      if ((rc = site_mom_prerst(pjob)) != PBSE_NONE)
        {
        return(rc);
        }

      rc = blcr_restart_job(pjob);

      break;

    case CST_NONE:

    default:

      return(PBSE_NOSUP);

      /*NOTREACHED*/

      break;
    }

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



/* this file creates missing stderr/stdout files before restarting
   the checkpointed job. This was designed for BLCR checkpointing.
   empty .OU or .ER files are not retained by the server, so if we
   are restarting a checkpointed job then they will not get sent back
   out to use.  the blcr restart command will expect these files to exist,
   even if empty.  If any expected files are missing we create them here */

/* TODO: this needs to be modified to work with user .pbs_spool directories */
int create_missing_files(
    
  job *pjob)

  {
  int            should_have_stderr;
  int            should_have_stdout;
  pbs_attribute *pattr;
  char          *pstr;
  char          *namebuf;
  int            bufsize;
  int            files_created = 0;
  int            fd;

  should_have_stderr = TRUE;
  should_have_stdout = TRUE;
  pattr = &pjob->ji_wattr[JOB_ATR_join];

  if (pattr->at_flags & ATR_VFLAG_SET)
    {
    pstr = pattr->at_val.at_str;

    if ((pstr != NULL) && (*pstr != '\0') && (*pstr != 'n'))
      {
      /* if not the first letter, and in list - is joined */

      if ((*pstr != 'e') && (strchr(pstr + 1, (int)'e')))
        {
        should_have_stderr = FALSE; /* being joined */
        }
      else if ((*pstr != 'o') && (strchr(pstr + 1, (int)'o')))
        {
        should_have_stdout = FALSE;
        }
      }
    }



  if (should_have_stdout)
    {
    bufsize = strlen(pjob->ji_qs.ji_fileprefix) + strlen(path_spool) + strlen(JOB_STDOUT_SUFFIX) + 1;
    namebuf = (char *)calloc(bufsize, sizeof(char));

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

    strcpy(namebuf, path_spool);

    strcat(namebuf, pjob->ji_qs.ji_fileprefix);
    strcat(namebuf, JOB_STDOUT_SUFFIX);

    if (access(namebuf, F_OK) != 0)
      {
      if ((fd = creat(namebuf, S_IRUSR | S_IWUSR)) >= 0)
        {
        if (fchown(fd,  pjob->ji_qs.ji_un.ji_momt.ji_exuid, pjob->ji_qs.ji_un.ji_momt.ji_exgid) == -1)
          {
          log_err(errno, "create_missing_files", (char *)"cannot change file owner");
          }
        close(fd);
        ++files_created;
        }
      else
        {
        /* couldn't create the file, why could this happen, TODO: what should we do? */
        }

      }

    free(namebuf);
    }

  if (should_have_stderr)
    {
    bufsize = strlen(pjob->ji_qs.ji_fileprefix) + strlen(path_spool) + strlen(JOB_STDOUT_SUFFIX) + 1;
    namebuf = (char *)calloc(bufsize, sizeof(char));

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

    strcpy(namebuf, path_spool);

    strcat(namebuf, pjob->ji_qs.ji_fileprefix);
    strcat(namebuf, JOB_STDERR_SUFFIX);

    if (access(namebuf, F_OK) != 0)
      {
      if ((fd = creat(namebuf, S_IRUSR | S_IWUSR)) >= 0)
        {
        if (fchown(fd,  pjob->ji_qs.ji_un.ji_momt.ji_exuid, pjob->ji_qs.ji_un.ji_momt.ji_exgid) == -1)
          {
          log_err(errno, "create_missing_files", (char *)"cannot change file ownership");
          }
        close(fd);
        ++files_created;
        }
      else
        {
        /* couldn't create the file, why could this happen, TODO: what should we do? */
        }

      }

    free(namebuf);
    }

  return files_created;
  } /* END create_missing_files() */


