#include "license_pbs.h" /* See here for the software license */
#include <pbs_config.h>   /* the master config generated by configure */
/* this file contains functions for manipulating job arrays

  included functions:

  is_array() determine if jobnum is actually an array identifyer
  get_array() return array struct for given "parent id"
  array_save() save array struct to disk
  array_get_parent_id() return id of parent job if job belongs to a job array
  array_recov() recover the array struct for a job array at restart
  array_delete() free memory used by struct and delete sved struct on disk
 */

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <limits.h> /* INT_MAX */

/* this macro is for systems like BSD4 that do not have O_SYNC in fcntl.h,
 * but do have O_FSYNC! */

#ifndef O_SYNC
#define O_SYNC O_FSYNC
#endif /* !O_SYNC */

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <pthread.h>

#include <libxml/parser.h>
#include <libxml/tree.h>

#include "pbs_ifl.h"
#include "log.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "lib_ifl.h"
#include "list_link.h"
#include "attribute.h"
#include "server_limits.h"
#include "server.h"
#include "pbs_job.h"
#include "queue.h"
#include "pbs_error.h"
#include "svrfunc.h"
#include "work_task.h"
#include "utils.h"
#include "array.h"
#include "svr_func.h"
#include "job_func.h" /* svr_job_purge */
#include "ji_mutex.h"
#include "mutex_mgr.hpp"
#include "batch_request.h"
#include "alps_constants.h"

#include "../lib/Libutils/u_lock_ctl.h" /* lock_ss, unlock_ss */


extern int array_upgrade(job_array *, int, int, int *);
extern char *get_correct_jobname(const char *jobid);
extern int count_user_queued_jobs(pbs_queue *,char *);
extern void post_modify_arrayreq(batch_request *preq);

/* global data items used */

/* list of job arrays */
extern struct server   server;

all_jobs          array_summary;
all_arrays allarrays;

extern char *path_arrays;
extern char *path_jobs;
extern int    LOGLEVEL;
extern char *pbs_o_host;


extern int array_259_upgrade;

int         is_num(const char *);
int         array_request_token_count(const char *);
int         array_request_parse_token(char *, int *, int *);
job_array  *next_array_check(int *, job_array *);
void        force_purge_work(job *pjob);

#define     BUFSIZE 256

/* search job array list to determine if id is a job array */

int is_array(
    
  char *id)

  {
  int        rc = FALSE;

  char      *bracket_ptr;
  char      *end_bracket_ptr;
  char      *tmpjobid;
  char       jobid[PBS_MAXSVRJOBID];
  char       temp_jobid[PBS_MAXSVRJOBID];

  tmpjobid = get_correct_jobname(id);
  if (tmpjobid == NULL)
    {
    /* Maybe we should just return ENOMEM? */
    snprintf(jobid, sizeof(jobid), "%s", id);
    }
  else
    {
    snprintf(jobid, sizeof(jobid), "%s", tmpjobid);
    free(tmpjobid);
    }

  /* Check to see if we have an array dependency */
  /* If there is an array dependency count then we will */
  /* have an id of something like arrayid[][1]. We need to take */
  /* off the [1] so we can compare the array id with and existing */
  /* array entry. */
  if ((bracket_ptr = strchr(jobid,'[')) != NULL)
    {
    /* Make sure the next character is ']' */
    if (*(++bracket_ptr) != ']')
      {
      /* If we do not have a ']' then we have bad syntax. */
      return(FALSE);
      }

    if (*(++bracket_ptr) == '[')
      {
      /* we made it to here. That means we have a count inside
         brackets. Just truncate them for the name comparison */
      end_bracket_ptr = strchr(bracket_ptr, ']');
      if (end_bracket_ptr == NULL)
        {
        /*  we do not have a ']' then we have bad syntax. */
        return(FALSE);
        }
      /* advance end_bracket_ptr one. We should be either NULL or '.' */
      end_bracket_ptr++;

      /* truncate the string */
      *bracket_ptr = 0; /* this makes jobid just the arrayid name */
      /* append the rest of the job id */
      snprintf(temp_jobid, sizeof(jobid), "%s%s", jobid, end_bracket_ptr);
      snprintf(jobid, sizeof(jobid), "%s", temp_jobid);
      }
    }
  else
    {
    /* No '[' then we do not have an array */
    return (FALSE);
    }

  allarrays.lock();

  if (allarrays.find(jobid) != NULL)
    rc = TRUE;

  allarrays.unlock();

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



/* return a server's array info struct corresponding to an array id */
job_array *get_array(
    
  const char *id)

  {
  job_array *pa;
  char      *tmpjobid;

  tmpjobid = get_correct_jobname(id);
  if (tmpjobid == NULL)
    return(NULL);

  allarrays.lock();

  pa = allarrays.find(tmpjobid);

  if (pa != NULL)
    lock_ai_mutex(pa, __func__, NULL, LOGLEVEL);

  allarrays.unlock();

  free(tmpjobid);

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



/*
 * get_and_remove_array()
 *
 * Gets the array with parent_id == id if present, and removes it from allarrays,
 * then it is returned.
 *
 * @param id - the id of the array to remove and return
 * @return - a pointer to the removed array or NULL if not found
 */

job_array *get_and_remove_array(

  const char *id)

  {
  job_array *pa = NULL;
  char      *tmpjobid;

  tmpjobid = get_correct_jobname(id);
  if (tmpjobid == NULL)
    return(NULL);

  allarrays.lock();

  pa = allarrays.find(tmpjobid);

  if (pa != NULL)
    {
    lock_ai_mutex(pa, __func__, NULL, LOGLEVEL);
    allarrays.remove(id);
    }

  allarrays.unlock();

  free(tmpjobid);

  return(pa);
  } // END get_and_remove_array()



void add_string_field_node(

    xmlNodePtr *rnode,
    const char *content, 
    const char *tag,
    int        *node_count)

  {
  xmlNodePtr root_node = *rnode;
  if (xmlNewChild(root_node, NULL, (xmlChar *)tag, (xmlChar *)content))
    *node_count += 1;
  }

void add_integer_field_node(

  xmlNodePtr *rnode,
  int field,
  const char *tag,
  int *node_count)

  {
  char content[BUFSIZE];
  xmlNodePtr root_node = *rnode;
  
  snprintf(content, sizeof(content), "%d", field);
  if (xmlNewChild(root_node, NULL, (xmlChar *)tag, (xmlChar *)content))
    *node_count += 1;
  }


int add_token_xml(

  xmlNodePtr *tnode, 
  const array_request_node *rn, 
  int *node_count)

  {
  int rc = -1;
  
  xmlNodePtr tokenNode = *tnode;
  xmlNodePtr node;
  char buf[BUFSIZE];

  snprintf(buf, sizeof(buf), "%s%d", TOKEN_TAG,*node_count);
  if ((node = xmlNewNode(NULL, (xmlChar *)buf)))
    {
    if (xmlAddChild(tokenNode, node))
      {
      snprintf(buf, sizeof(buf), "%u", (unsigned int)rn->start);
      if (xmlSetProp(node, (const xmlChar *)START_TAG, (const xmlChar *)buf))
        {
        snprintf(buf, sizeof(buf), "%u", (unsigned int)rn->end);
        if (xmlSetProp(node, (const xmlChar *)END_TAG, (const xmlChar *)buf))
          {
          *node_count += 1;
          rc = PBSE_NONE;
          }
        }
      }
    }
  return rc;
  }



int array_info_xml(

  xmlNodePtr       *rnode,   /* M */ /* dom's root node */
  const array_info *ai_qs)   /* I */ /* info to written to xml */

  {
  int node_count = 0;

  add_integer_field_node(rnode, ai_qs->struct_version, ARRAY_STRUCT_VERSION_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->array_size, ARRAY_SIZE_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_jobs, NUM_JOBS_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->slot_limit, SLOT_LIMIT_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->jobs_running, JOBS_RUNNING_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->jobs_done, JOBS_DONE_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_cloned, NUM_CLONED_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_started, NUM_STARTED_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_failed, NUM_FAILED_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_successful, NUM_SUCCESSFUL_TAG, &node_count);
  add_string_field_node(rnode, ai_qs->owner, OWNER_TAG, &node_count);
  add_string_field_node(rnode, ai_qs->parent_id, PARENT_TAG, &node_count);
  add_string_field_node(rnode, ai_qs->fileprefix, ARRAY_FILEPREFIX_TAG, &node_count);
  add_string_field_node(rnode, ai_qs->submit_host, SUBMIT_HOST_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_purged, NUM_PURGED_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->num_idle, NUM_IDLE_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->idle_slot_limit, IDLE_SLOT_LIMIT_TAG, &node_count);
  add_integer_field_node(rnode, ai_qs->highest_id_created, HIGHEST_ID_CREATED_TAG, &node_count);

  // NOTE: This must be updated 
  if (node_count == 18)
    return(PBSE_NONE);

  return(-1);
  } // END array_info_xml()



int array_save_xml(

  const job_array *pa,       /* I */  /* array info to be written to xml */
  const char      *filename, /* I */  /* xml filename */
  char            *log_buf,  /* O */  /* error buffer */
  size_t           buflen)   /* I */  /* length of error buffer */

  {
  int rc = -1;
  xmlDocPtr doc = NULL; 
  xmlNodePtr root_node;

  if ((doc = xmlNewDoc((const xmlChar*) "1.0")))
    {
    if ((root_node = xmlNewNode(NULL, (const xmlChar*) ARRAY_TAG)))
      {
      xmlDocSetRootElement(doc, root_node);
      if ((rc = array_info_xml(&root_node, (const array_info*) &(pa->ai_qs))) == PBSE_NONE)
        {
        if (xmlNewChild(root_node, NULL, (xmlChar *)RANGE_TAG, (xmlChar *)pa->ai_qs.range_str.c_str()))
          {
          lock_ss();

          int lenwritten = xmlSaveFormatFileEnc(filename, doc, NULL, 1);

          unlock_ss();

          if (!(lenwritten))
            {
            rc = -1;
            snprintf(log_buf, buflen, "unable to write document to disk (file %s) for job array %s", 
              filename, pa->ai_qs.parent_id);
            }
          }
        }
      }
    else
      snprintf(log_buf, buflen, "Can't create root node on the document for job array file %s", filename);
    xmlFreeDoc(doc);
    }
  else
    snprintf(log_buf, buflen, "unable to create document for the xml file job %s", filename);

  /* error message will be printed out by the caller */
  return rc;
  } // END array_save_xml()



/* save a job array struct to disk returns zero if no errors*/
int array_save(
    
  job_array *pa)

  {
  char namebuf[MAXPATHLEN];
  char log_buf[LOCAL_LOG_BUF_SIZE];
  // get adjusted path_arrays path
  std::string adjusted_path_arrays = get_path_jobdata(pa->ai_qs.parent_id, path_arrays);

  snprintf(namebuf, sizeof(namebuf), "%s%s%s",
    adjusted_path_arrays.c_str(), pa->ai_qs.fileprefix, ARRAY_FILE_SUFFIX);

  /* error buf is filled in array_save_xml or its subroutines */
  if (array_save_xml((const job_array *)pa, namebuf, log_buf, sizeof(log_buf)) != PBSE_NONE)
    {
    log_event(PBSEVENT_SYSTEM,PBS_EVENTCLASS_JOB,pa->ai_qs.parent_id,log_buf);
    unlink(namebuf);
    return -1;
    }

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




/* if a job belongs to an array, this will return the id of the parent job
 * returns job id if not array parent id
 */
void array_get_parent_id(
    
  char *job_id,
  char *parent_id)

  {
  char *c;
  char *pid;
  int bracket = 0;

  c = job_id;
  *parent_id = '\0';
  pid = parent_id;

  /* copy up to the '[' */

  while (!bracket && *c != '\0')
    {
    if (*c == '[')
      {
      bracket = 1;
      }
    *pid = *c;
    c++;
    pid++;
    }

  /* skip the until the closing bracket */
  while (*c != ']' && *c != '\0')
    {
    c++;
    }

  /* copy the rest of the id */
  *pid = '\0';
  strcat(pid, c);

  } /* END array_get_parent_id() */




int read_and_convert_259_array(
    
  int         fd,
  job_array  *pa,
  const char *path)

  {
  char           log_buf[LOCAL_LOG_BUF_SIZE];
  int            len;
  job_array_259 *pa_259; /* This is for a backward compatibility problem put 
                            into 2.5.9 and 3.0.3 */
  /* allocate the storage for the struct */
  pa_259 = (job_array_259*)calloc(1, sizeof(job_array_259));

  if (pa_259 == NULL)
    {
    return PBSE_SYSTEM;
    }

  len = read_ac_socket(fd, &(pa_259->ai_qs), sizeof(pa_259->ai_qs));
  if (len < 0) 
    {
    sprintf(log_buf, "error reading %s", path);
    log_err(errno, "read_and_convert_259_array", log_buf);
    free(pa_259);
    close(fd);
    return PBSE_BAD_ARRAY_DATA;
    }

  if (pa_259->ai_qs.struct_version == ARRAY_QS_STRUCT_VERSION)
    {
    sprintf(log_buf, "Already at array structure version 4. Restart pbs_server without -u option");
    log_err(errno, "read_and_convert_259_array", log_buf);
    free(pa_259);
    close(fd);
    return PBSE_BAD_ARRAY_DATA;
    }
    
  if (pa_259->ai_qs.struct_version != 3)
    {
    sprintf(log_buf, "Cannot upgrade array version %d to %d", 
             pa_259->ai_qs.struct_version, ARRAY_QS_STRUCT_VERSION);
    log_err(errno, "read_and_convert_259_array", log_buf);
    free(pa_259);
    close(fd);
    return PBSE_BAD_ARRAY_DATA;
    }
  
  pa->ai_qs.struct_version = ARRAY_QS_STRUCT_VERSION;
  pa->ai_qs.array_size = pa_259->ai_qs.array_size;
  pa->ai_qs.num_jobs = pa_259->ai_qs.num_jobs;
  pa->ai_qs.slot_limit = pa_259->ai_qs.slot_limit;
  pa->ai_qs.jobs_running = pa_259->ai_qs.jobs_running;
  pa->ai_qs.jobs_done = pa_259->ai_qs.jobs_done;
  pa->ai_qs.num_cloned = pa_259->ai_qs.num_cloned;
  pa->ai_qs.num_started = pa_259->ai_qs.num_started;
  pa->ai_qs.num_failed = pa_259->ai_qs.num_failed;
  pa->ai_qs.num_successful = pa_259->ai_qs.num_successful;
  pa->ai_qs.num_purged = pa_259->ai_qs.num_purged;

  for (struct array_depend *pdep = (struct array_depend *)GET_NEXT(pa_259->ai_qs.deps); 
       pdep != NULL;
       pdep = (struct array_depend *)GET_NEXT(pa_259->ai_qs.deps))
    pa->ai_qs.deps.push_back(pdep);

  snprintf(pa->ai_qs.owner, sizeof(pa->ai_qs.owner), "%s", pa_259->ai_qs.owner);
  snprintf(pa->ai_qs.parent_id, sizeof(pa->ai_qs.parent_id), "%s", pa_259->ai_qs.parent_id);
  snprintf(pa->ai_qs.fileprefix, sizeof(pa->ai_qs.fileprefix), "%s", pa_259->ai_qs.fileprefix);
  snprintf(pa->ai_qs.submit_host, sizeof(pa->ai_qs.submit_host), "%s", pa_259->ai_qs.submit_host);

  free(pa_259);

  array_save(pa);

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



/*
 * assign_array_info_fields()
 *
 * Sets a value in the array according the specified xml
 * @param pa_new - a pointer to a pointer to the new array
 * @param xml_node - a pointer to the xml node
 * @param log_buf - a pointer to the error buffer
 * @param buflen - the length of the error buffer
 * @param num_tokens - the number of tokens in the range, if present
 * @return PBSE_NONE on success, -1 if the xml tag is unreadable
 */

int assign_array_info_fields(

  job_array  **pa_new,       /* O */ /* Array Job to recover information from file */
  xmlNodePtr   xml_node,       /* I */ /*Root element of the dom */
  char        *log_buf,        /* O */ /* Error buffer */
  size_t       buflen,        /* I */ /* Error buffer length */
  int         *num_tokens)    /* O */ /* Number of tokens, value to be kept in another variable */

  {
  xmlChar *content = xmlNodeGetContent(xml_node);
  int      rc = PBSE_NONE;

  if (!content)
    {
    snprintf(log_buf, buflen, "missing content on array xml, tag was %s", (const char*) xml_node->name);
    return -1;
    }
  
  job_array *pa = *pa_new;
  size_t nameLen = strlen((const char *) xml_node->name);

  /* The length comparison first so to avoid buffer overrun with strcmp */
  if ((nameLen == strlen(ARRAY_STRUCT_VERSION_TAG)) && 
    (!(strcmp((char *)xml_node->name, ARRAY_STRUCT_VERSION_TAG)))) 
    pa->ai_qs.struct_version = atoi((const char *)content);
  else if ((nameLen == strlen(ARRAY_SIZE_TAG)) && 
    (!(strcmp((char *)xml_node->name, ARRAY_SIZE_TAG)))) 
    pa->ai_qs.array_size = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_JOBS_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_JOBS_TAG)))) 
    pa->ai_qs.num_jobs = atoi((const char *)content);
  else if ((nameLen == strlen(SLOT_LIMIT_TAG)) &&
    (!(strcmp((char *)xml_node->name, SLOT_LIMIT_TAG)))) 
    pa->ai_qs.slot_limit = atoi((const char *)content);
  else if ((nameLen == strlen(JOBS_RUNNING_TAG)) &&
    (!(strcmp((char *)xml_node->name, JOBS_RUNNING_TAG)))) 
    pa->ai_qs.jobs_running = atoi((const char *)content);
  else if ((nameLen == strlen(JOBS_DONE_TAG)) && 
    (!(strcmp((char *)xml_node->name, JOBS_DONE_TAG)))) 
    pa->ai_qs.jobs_done = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_CLONED_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_CLONED_TAG)))) 
    pa->ai_qs.num_cloned = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_STARTED_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_STARTED_TAG)))) 
    pa->ai_qs.num_started = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_FAILED_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_FAILED_TAG)))) 
    pa->ai_qs.num_failed = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_PURGED_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_PURGED_TAG)))) 
    pa->ai_qs.num_purged = atoi((const char *)content);
  else if ((nameLen == strlen(OWNER_TAG)) &&
    (!(strcmp((char *)xml_node->name, OWNER_TAG)))) 
    snprintf(pa->ai_qs.owner, PBS_MAXUSER + PBS_MAXSERVERNAME + 2, "%s", (const char *)content);
  else if ((nameLen == strlen(PARENT_TAG)) &&
    (!(strcmp((char *)xml_node->name, PARENT_TAG)))) 
    snprintf(pa->ai_qs.parent_id, PBS_MAXSVRJOBID + 1, "%s", (const char *)content);
  else if ((nameLen == strlen(ARRAY_FILEPREFIX_TAG)) &&
    (!(strcmp((char *)xml_node->name, ARRAY_FILEPREFIX_TAG)))) 
    snprintf(pa->ai_qs.fileprefix, PBS_JOBBASE + 1, "%s", (const char *)content);
  else if ((nameLen == strlen(SUBMIT_HOST_TAG)) &&
    (!(strcmp((char *)xml_node->name, SUBMIT_HOST_TAG)))) 
    snprintf(pa->ai_qs.submit_host, PBS_MAXSERVERNAME + 1, "%s", (const char *)content);
  else if (((nameLen - strlen(NUM_SUCCESSFUL_TAG) < 2) &&
    (!(strncmp((char *)xml_node->name, NUM_SUCCESSFUL_TAG, strlen(NUM_SUCCESSFUL_TAG)))))) 
    pa->ai_qs.num_successful = atoi((const char *)content);
  else if ((nameLen == strlen(NUM_TOKENS_TAG)) &&
    (!(strcmp((char *)xml_node->name, NUM_TOKENS_TAG)))) 
    *num_tokens = atoi((const char *)content);
  else if (!strcmp((const char *)xml_node->name, RANGE_TAG))
    {
    pa->ai_qs.range_str = (const char *)content;
    }
  else if (!strcmp((const char *)xml_node->name, NUM_IDLE_TAG))
    {
    pa->ai_qs.num_idle = strtol((const char *)content, NULL, 10);
    }
  else if (!strcmp((const char *)xml_node->name, IDLE_SLOT_LIMIT_TAG))
    {
    pa->ai_qs.idle_slot_limit = strtol((const char *)content, NULL, 10);
    }
  else if (!strcmp((const char *)xml_node->name, HIGHEST_ID_CREATED_TAG))
    {
    pa->ai_qs.highest_id_created = strtol((const char *)content, NULL, 10);
    }
  else
    {
    snprintf(log_buf, buflen, "unknown tag \"%s\" on array xml", (const char*) xml_node->name);
    rc = -1;
    }

  xmlFree(content);
  return rc;
  } // END assign_array_info_fields()



int parse_num_tokens(

  job_array **new_pa, 
  xmlNodePtr  tokensNode, 
  char       *log_buf, 
  size_t       buflen)
  
  {
  xmlNodePtr  cur_node = NULL;
  job_array  *pa = *new_pa;
  int         rc = PBSE_NONE;
  bool        element_found = false;
  char        range_buf[MAXLINE];

  for (cur_node = tokensNode->children; cur_node != NULL; cur_node = cur_node->next)
    {
    /* skip text children, only process elements */
    if (!strcmp((const char *)cur_node->name, text_name))
      continue;

    element_found = true;

    xmlChar *start_attr, *end_attr;
    start_attr = end_attr = NULL;
    mbool_t gotBothAttr = false;
    if ((start_attr = xmlGetProp(cur_node, (xmlChar *)START_TAG)))
      {
      if ((end_attr = xmlGetProp(cur_node, (xmlChar *)END_TAG)))
        {
        int start = atoi((char *)start_attr);
        int end = atoi((char *)end_attr);

        if (pa->ai_qs.range_str.size() == 0)
          {
          if (start != end)
            snprintf(range_buf, sizeof(range_buf), "%d-%d", start, end);
          else
            snprintf(range_buf, sizeof(range_buf), "%d", start);
          }
        else
          {
          if (start != end)
            snprintf(range_buf, sizeof(range_buf), ",%d-%d", start, end);
          else
            snprintf(range_buf, sizeof(range_buf), ",%d", start);
          }

        gotBothAttr = true;
        }
      }

    if (start_attr)
      xmlFree(start_attr);
    if (end_attr)
      xmlFree(end_attr);

    if (!gotBothAttr)
      {
      snprintf(log_buf, buflen, "%s", "missing start/end attributes for array tokens in the array xml");
      rc = -1;
      break;
      }
    }
   
  if (element_found == false)
    {
    snprintf(log_buf, buflen, "%s", "no \"tokens\" xml element was found");
    rc = -1;
    }

  return rc;
  }



/*
 * parse_array_dom()
 *
 * @pre-cond: pa hasn't been allocated or recovered
 * @pre-cond: root_element is a valid xmlNodePtr pointing to the root of this
 * array's xml document
 * @post-cond: pa and its attributes have been adequately recovered.
 */

int parse_array_dom(

  job_array  **pa,         /* O */ /* Array Job to recover information from file */
  xmlNodePtr root_element, /* I */ /*Root element of the dom */
  char       *log_buf,      /* O */ /* Error buffer */
  size_t      buflen)      /* I */ /* Error buffer length */

  {
  xmlNodePtr cur_node = NULL;
  xmlNodePtr tokensNode = NULL;
  int        rc = -1;
  int        num_tokens = 0;
  bool       element_found = false;

  for (cur_node = root_element->children; cur_node != NULL; cur_node = cur_node->next)
    {
    /* skip text children, only process elements */
    if (!strcmp((const char *)cur_node->name, text_name))
      continue;

    element_found = true;
      
    if (!(strcmp((const char*)cur_node->name, TOKENS_TAG)))
      tokensNode = cur_node;
    else if ((rc = assign_array_info_fields(pa, cur_node, log_buf, buflen, &num_tokens)))
      break;
    }

  if ((!rc))
    {
    job_array  *new_pa = *pa;
    if (new_pa->ai_qs.array_size > 0)
      if ((new_pa->job_ids = (char **)calloc(new_pa->ai_qs.array_size, sizeof(char *))) == NULL)
        {
        snprintf(log_buf, buflen, "%s", "unable to allocate memory for array job_ids strings");
        rc = -1;
        }

    if ((!(rc)) &&
        (num_tokens > 0) && tokensNode)
      rc = parse_num_tokens(pa, tokensNode, log_buf, buflen);

    new_pa->initialize_uncreated_ids();
    }

  if (element_found == false)
    snprintf(log_buf, buflen, "%s", "xml file is empty");

  return(rc);
  }


int array_recov_xml(

  const char *filename,    /* I */ /* File containing array information to read */
  job_array **pa,          /* O */ /* Array Job to recover information from file */
  char       *log_buf,     /* O */ /* Error buffer */
  size_t      buflen)      /* I */ /* Error buffer length */

  {
  xmlDoc *doc = NULL;
  xmlNode *root_element = NULL;
  int rc = PBSE_INVALID_SYNTAX;

  /*parse the file and get the DOM */
  doc = xmlReadFile(filename, NULL, 0);

  if (doc == NULL)
    return(rc);

  /*Get the root element node */
  if (!(root_element = xmlDocGetRootElement(doc)))
    {
    xmlFreeDoc(doc);
    return(rc);
    }
  else
    {
    if (strcmp((const char *) root_element->name, ARRAY_TAG))
      {
      snprintf(log_buf, buflen, "missing root tag %s", ARRAY_TAG);
      
      /* set return code of -1 as we do have an AR xml but it did not have the right root elem. */
      xmlFreeDoc(doc);
      return(-1); 
      }

    if (parse_array_dom(pa, root_element, log_buf, buflen))
      {
      xmlFreeDoc(doc);
      return(-1);
      }
    }
      
  xmlFreeDoc(doc);

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


int array_recov_binary(

  const char *path,    /* I */ /* File containing array information to read */
  job_array **new_pa,  /* O */ /* Array Job to recover information from file */
  char       *log_buf, /* O */ /* Error buffer */
  size_t      buflen)  /* I */ /* Error buffer length */

  {
  job_array *pa = *new_pa;
  int   old_version = ARRAY_QS_STRUCT_VERSION;
  int   fd;
  int   num_tokens;
  int   i;
  int   len;
  int   rc = -1;
  struct stat s_buf;
  char                request_buf[MAXLINE];

  fd = open(path, O_RDONLY, 0);
  if (fd < 0)
    {
    snprintf(log_buf, buflen, "failed to open %s", path);
    log_err(errno, __func__, log_buf);
    return(PBSE_SYSTEM);
    }

  if (fstat(fd, &s_buf) < 0)
    {
    snprintf(log_buf, buflen, "failed to fstat %s", path);
    log_err(errno, __func__, log_buf);

    close(fd);
    return(PBSE_SYSTEM);
    }

  /* if we have a zero length file, do not proceed */
  if (s_buf.st_size == 0)
    {
    snprintf(log_buf, buflen, "%s is empty", path);
    log_err(-1, __func__, log_buf);

    close(fd);
    return(PBSE_SYSTEM);
    }

  if (array_259_upgrade)
    {
    rc = read_and_convert_259_array(fd, pa, path);
    if (rc != PBSE_NONE)
      {
      close(fd);
      return(rc);
      }
    }
  else
    {

    /* read the file into the struct previously allocated.
     */

    len = read_ac_socket(fd, &(pa->ai_qs), sizeof(array_info));
    if ((len < 0) || ((len < (int)sizeof(array_info)) && (pa->ai_qs.struct_version == ARRAY_QS_STRUCT_VERSION)))
      {
      pa->ai_qs.deps.clear();
      snprintf(log_buf, buflen, "error reading %s", path);
      close(fd);
      return(PBSE_SYSTEM);
      }

    if (pa->ai_qs.struct_version != ARRAY_QS_STRUCT_VERSION)
      {
      rc = array_upgrade(pa, fd, pa->ai_qs.struct_version, &old_version);
      if (rc)
        {
        pa->ai_qs.deps.clear();
        snprintf(log_buf, buflen, 
          "Cannot upgrade array version %d to %d", pa->ai_qs.struct_version, ARRAY_QS_STRUCT_VERSION);
        close(fd);
        return(rc);
        }
      }
    }

  // make sure array_size is valid
  if (pa->ai_qs.array_size <= 0)
    {
    snprintf(log_buf, buflen, 
      "array size must be positive");
    close(fd);
    return PBSE_SYSTEM;
    }

  pa->job_ids = (char **)calloc(pa->ai_qs.array_size, sizeof(char *));
  if(pa->job_ids == NULL)
    {
    close(fd);
    return PBSE_SYSTEM;
    }

  /* check to see if there is any additional info saved in the array file */
  /* check if there are any array request tokens that haven't been fully
     processed */

  if (old_version > 1)
    {
    if (read_ac_socket(fd, &num_tokens, sizeof(int)) != sizeof(int))
      {
      snprintf(log_buf, buflen, "error reading token count from %s", path);
      close(fd);
      return(PBSE_SYSTEM);
      }

    for (i = 0; i < num_tokens; i++)
      {
      array_request_node rn;

      if (read_ac_socket(fd, &rn, sizeof(array_request_node)) != sizeof(array_request_node))
        {
        snprintf(log_buf, buflen, "error reading array_request_node from %s", path);

        close(fd);
        return(PBSE_SYSTEM);
        }

      if (pa->ai_qs.range_str.size() == 0)
        {
        if (rn.start != rn.end)
          snprintf(request_buf, sizeof(request_buf), "%d-%d", rn.start, rn.end);
        else
          snprintf(request_buf, sizeof(request_buf), "%d", rn.start);
        }
      else
        {
        if (rn.start != rn.end)
          snprintf(request_buf, sizeof(request_buf), ",%d-%d", rn.start, rn.end);
        else
          snprintf(request_buf, sizeof(request_buf), ",%d", rn.start);
        }

      pa->ai_qs.range_str += request_buf;
      }
    }

  close(fd);

  if (old_version != ARRAY_QS_STRUCT_VERSION)
    {
    /* resave the array struct if the version on disk is older than the current */
    array_save(pa);
    }

  return PBSE_NONE;
  } /* END array_recov_binary */



/* array_recov reads in  an array struct saved to disk and inserts it into
   the servers list of arrays */
int array_recov(

  const char  *path, 
  job_array  **new_pa)

  {
  job_array *pa;
  char  log_buf[LOCAL_LOG_BUF_SIZE];
  int   rc;

  *new_pa = NULL;
  bool   binary_conversion = false;

  /* allocate the storage for the struct */
  pa = new job_array();

  if (pa == NULL)
    {
    return(PBSE_SYSTEM);
    }

  if ((rc = array_recov_xml(path, &pa, log_buf, sizeof(log_buf))) && rc == PBSE_INVALID_SYNTAX)
    {
    rc = array_recov_binary(path, &pa, log_buf, sizeof(log_buf));
    binary_conversion = true;
    }

  if (rc != PBSE_NONE) 
    {
    delete pa;
    log_err(-1, __func__, log_buf);
    return rc;
    }

  if (binary_conversion)
    if (array_save_xml((const job_array *)pa, path, log_buf, sizeof(log_buf)) != PBSE_NONE)
      log_event(PBSEVENT_SYSTEM,PBS_EVENTCLASS_JOB,pa->ai_qs.parent_id,log_buf);

  lock_ai_mutex(pa, __func__, NULL, LOGLEVEL);

  /* link the struct into the servers list of job arrays */
  insert_array(pa);

  *new_pa = pa;

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



/* 
 * array_delete()
 * Delete a job array from memory and disk. 
 *
 * @pre-cond: There should be no sub jobs left for this array.
 *  of jobs that belong to the array becomes zero.
 *  returns zero if there are no errors, non-zero otherwise
 */
int array_delete(
    
  const char *array_id)

  {
  char                     path[MAXPATHLEN + 1];
  char                     log_buf[LOCAL_LOG_BUF_SIZE];
  std::string              adjusted_path_arrays;

  /* first thing to do is take this out of the servers list of all arrays */
  job_array *pa = get_and_remove_array(array_id);

  // The array has vanished - success
  if (pa == NULL)
    return(PBSE_NONE);

  // get adjusted path_arrays path
  adjusted_path_arrays = get_path_jobdata(pa->ai_qs.fileprefix, path_arrays);

  /* delete the on disk copy of the struct */
  snprintf(path, sizeof(path), "%s%s%s",
    adjusted_path_arrays.c_str(), pa->ai_qs.fileprefix, ARRAY_FILE_SUFFIX);

  if (unlink(path))
    {
    sprintf(log_buf, "unable to delete %s", path);
    log_err(errno, __func__, log_buf);
    }

  /* purge the "template" job, 
     this also deletes the shared script file for the array*/
  if (pa->ai_qs.parent_id[0] != '\0')
    {
    job *pjob;
    if ((pjob = svr_find_job(pa->ai_qs.parent_id, FALSE)) != NULL)
      svr_job_purge(pjob);
    }

  /* free the memory allocated for the struct */
  delete pa;

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



/* 
 * set_slot_limit()
 * sets how many jobs can be run from this array at once
 *
 * @param request - the string array request
 * @param pa - the array to receive a slot limit
 *
 * @return 0 on SUCCESS
 */

int set_slot_limit(

  char      *request, /* I */
  job_array *pa)      /* O */

  {
  char *pcnt;
  long  max_limit;

  /* check for a max slot limit */
  if (get_svr_attr_l(SRV_ATR_MaxSlotLimit, &max_limit) != PBSE_NONE)
    max_limit = NO_SLOT_LIMIT;

  if ((pcnt = strchr(request,'%')) != NULL)
    {
    /* remove '%' from the request, or else it can't be parsed */
    while (*pcnt == '%')
      {
      *pcnt = '\0';
      pcnt++;
      }

    /* read the number if one is given */
    if (strlen(pcnt) > 0)
      {
      pa->ai_qs.slot_limit = atoi(pcnt);
      if ((max_limit != NO_SLOT_LIMIT) &&
          (max_limit < pa->ai_qs.slot_limit))
        {
        return(INVALID_SLOT_LIMIT);
        }
      }
    else
      {
      pa->ai_qs.slot_limit = max_limit;
      }
    }
  else
    {
    pa->ai_qs.slot_limit = max_limit;
    }

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



int setup_array_struct(
    
  job *pjob)

  {
  job_array          *pa;
  int                 rc;

  if (pjob == NULL)
    return RM_ERR_BADPARAM;

  pa = new job_array();

  pa->set_array_id(pjob->ji_qs.ji_jobid);
  pa->set_arrays_fileprefix(pjob->ji_qs.ji_fileprefix);
  pa->set_owner(pjob->ji_wattr[JOB_ATR_job_owner].at_val.at_str);
  pa->set_submit_host(get_variable(pjob, pbs_o_host));

  mutex_mgr pa_mutex = mutex_mgr(pa->ai_mutex);

  if (job_save(pjob, SAVEJOB_FULL, 0) != 0)
    {
    /* the array is deleted in svr_job_purge */
    pa_mutex.unlock();
    /* Does job array need to be removed? */

    if (LOGLEVEL >= 6)
      log_record(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, "cannot save job");

    svr_job_purge(pjob);
    delete pa;

    return(1);
    }

  if ((rc = pa->set_slot_limit(pjob->ji_wattr[JOB_ATR_job_array_request].at_val.at_str)))
    {
    pa_mutex.unlock();
    array_delete(pjob->ji_qs.ji_jobid);

    return(rc);
    }

  long requested_limit = -1;

  if (pjob->ji_wattr[JOB_ATR_idle_slot_limit].at_flags & ATR_VFLAG_SET)
    requested_limit = pjob->ji_wattr[JOB_ATR_idle_slot_limit].at_val.at_long;

  if ((rc = pa->set_idle_slot_limit(requested_limit)))
    {
    pa_mutex.unlock();
    array_delete(pjob->ji_qs.ji_jobid);

    return(rc);
    }

  if ((rc = pa->parse_array_request(pjob->ji_wattr[JOB_ATR_job_array_request].at_val.at_str)))
    {
    pa_mutex.unlock();
    array_delete(pjob->ji_qs.ji_jobid);
    return(rc);
    }

  array_save(pa);

  strcpy(pjob->ji_arraystructid, pa->ai_qs.parent_id);

  insert_array(pa);

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



int is_num(
    
  const char *str)

  {
  int i;
  int len;

  len = strlen(str);

  if (len == 0)
    {
    return(FALSE);
    }

  for (i = 0; i < len; i++)
    {
    if (str[i] < '0' || str[i] > '9')
      {
      return(FALSE);
      }
    }

  return(TRUE);
  } /* END is_num() */



int array_request_token_count(
    
  const char *str)

  {
  int token_count = 1;
  int len = strlen(str);
  int i;

  if (len == 0)
    {
    return 0;
    }
  for (i = 0; i < len; i++)
    {
    if (str[i] == ',')
      {
      token_count++;
      }
    }

  return(token_count);
  } /* END array_request_token_count() */



int array_request_parse_token(
    
  char *str,   /* I */
  int  *start, /* O */
  int  *end)   /* O */

  {
  int   num_ids;
  long  start_l;
  long  end_l;
  char *idx;
  char *ridx;

  if ((NULL == str) ||
      (NULL == start) ||
      (NULL == end))
    {
    return 0;
    }
  idx = index(str, '-');
  ridx = rindex(str, '-');

  /* token is not a range, parse it as a single task id */
  if (idx == NULL)
    {
    /* make sure it is a number...*/
    if (!is_num(str))
      {
      start_l = -1;
      end_l = -1;
      }
    else
      {
      /* token is a number, set start_l and end_l to the value */ 
      start_l = strtol(str, NULL, 10);
      end_l = start_l;
      }
    }
  /* index and rindex found the same '-' character, this is a range */
  else if (idx == ridx)
    {
    *idx = '\0';
    idx++;

    /* check for an invalid range */
    if ((!is_num(str)) || 
        (!is_num(idx)))
      {
      start_l = -1;
      end_l = -1;
      }
    else
      {
      /* both sides of the range were numbers so we set start_l and end_l
         we will check later to make sure that the range is "good" */
      start_l = strtol(str, NULL, 10);
      end_l = strtol(idx, NULL, 10);
      }
    }
  /* index/rindex found different '-' characters, this can't be a good range */
  else
    {
    start_l = -1;
    end_l = -1;
    }

  /* restore the string so this function is non-destructive to the token */
  if ((idx != NULL) &&
      (idx == ridx + 1))
    {
    idx--;
    *idx = '-';
    }

  /* make sure the start or end of the range is not out of the range for 
     job array task IDs, and make sure that end_l is not less than start_l 
     (it is OK for end_l to == start_l)*/
  if ((start_l < 0) ||
      (start_l >= INT_MAX) ||
      (end_l < 0) ||
      (end_l >= INT_MAX) ||
      (start_l > PBS_MAXJOBARRAY) ||
      (end_l > PBS_MAXJOBARRAY) ||
      (end_l < start_l))
    {
    *start = -1;
    *end = -1;
    num_ids = 0;
    }
  else
    {
    /* calculate the number of task IDs in the range, and cast the start_l and
       end_l to ints when setting start and end (we already confirmed that 
       start_l and end_l are > 0 and <= INT_MAX, so we will not under/overflow)
      */
    num_ids = end_l - start_l + 1;
    *start = (int)start_l;
    *end   = (int)end_l;
    }

  return(num_ids);
  } /* END array_request_parse_token() */



/*
 * delete_array_range()
 *
 * deletes a range from a specific array
 *
 * @param pa - the array whose jobs are deleted
 * @param range_str - the user-given range to delete 
 * @return - the number of jobs skipped, -1 if range error 
 */
int delete_array_range(

  job_array *pa,
  char      *range_str,
  bool       purge)

  {
  job                *pjob;
  char               *range;
  std::vector<int>    range_vec;

  int                 num_skipped = 0;
  int                 num_deleted = 0;
  bool                deleted;
  int                 running;
  long                cancel_exit_code = 0;

  /* get just the numeric range specified, '=' should
   * always be there since we put it there in qdel */
  if ((range = strchr(range_str,'=')) == NULL)
    return(-1);
  range++; /* move past the '=' */

  if (translate_range_string_to_vector(range, range_vec) != PBSE_NONE)
    {
    /* don't delete jobs if range error */

    return(-1);
    }

  get_svr_attr_l(SRV_ATR_ExitCodeCanceledJob, &cancel_exit_code);

  for (size_t i = 0; i < range_vec.size(); i++)
    {
    int index = range_vec[i];

    /* don't stomp on other memory */
    if (index >= pa->ai_qs.array_size)
      continue;
    
    if (pa->job_ids[index] == NULL)
      continue;

    if ((pjob = svr_find_job(pa->job_ids[index], FALSE)) == NULL)
      {
      free(pa->job_ids[index]);
      pa->job_ids[index] = NULL;
      }
    else
      {
      mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
      if (pjob->ji_qs.ji_state >= JOB_STATE_EXITING)
        {
        /* invalid state for request,  skip */
        continue;
        }

      int old_state = pjob->ji_qs.ji_state;

      running = (pjob->ji_qs.ji_state == JOB_STATE_RUNNING);

      pthread_mutex_unlock(pa->ai_mutex);
      if (purge == true)
        {
        deleted = true;

        try
          {
          force_purge_work(pjob);
          }
        catch (int err)
          {
          if (err != PBSE_JOB_RECYCLED)
            {
            char log_buf[LOCAL_LOG_BUF_SIZE];
            sprintf(log_buf, "Error when purging %s", pa->job_ids[i]);
            log_err(err, __func__, log_buf);
            deleted = false;
            }
          }
        }
      else
        deleted = attempt_delete(pjob);
      
      // Both attempt_delete and force_purge_work unlock pjob
      pjob_mutex.set_unlock_on_exit(false);
      pthread_mutex_lock(pa->ai_mutex);

      if (deleted == false)
        {
        /* if the job was deleted, this mutex would be taked care of elsewhere. When it fails,
         * release it here */
        num_skipped++;
        }
      else 
        {
        /* running jobs will increase the deleted count when their obit is reported */
        if (running == FALSE)
          num_deleted++;

        pa->update_array_values(old_state, aeTerminate, pa->job_ids[i], cancel_exit_code);
        }
      }
    }

  pa->ai_qs.num_failed += num_deleted;

  return(num_skipped);
  } /* END delete_array_range() */




/* 
 * first_job_index()
 *
 * @param pa - the array
 * @return the index of the first job in the array
 */
int first_job_index(

  job_array *pa)

  {
  int i;

  for (i = 0; i < pa->ai_qs.array_size; i++)
    {
    if (pa->job_ids[i] != NULL)
      return(i);
    }

  return(-1);
  } /* END first_job_index() */



/* 
 * delete_whole_array()
 *
 * iterates over the array and deletes the whole thing
 *
 * @param pa - the array to be deleted
 * @param purge - true if the array should be purged, false if it's a normal delete
 * @return - the number of jobs skipped
 */
int delete_whole_array(

  job_array *pa,
  bool       purge)

  {
  int  i;
  int  num_skipped = 0;
  int  num_jobs = 0;
  int  num_deleted = 0;
  bool deleted;
  int  running;
  long cancel_exit_code = 0;

  std::string array_id = pa->ai_qs.parent_id;

  get_svr_attr_l(SRV_ATR_ExitCodeCanceledJob, &cancel_exit_code);

  job *pjob;

  for (i = 0; i < pa->ai_qs.array_size; i++)
    {
    if (pa->job_ids[i] == NULL)
      continue;

    if ((pjob = svr_find_job(pa->job_ids[i], FALSE)) == NULL)
      {
      free(pa->job_ids[i]);
      pa->job_ids[i] = NULL;
      }
    else
      {
      mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
      num_jobs++;

      if ((pjob->ji_qs.ji_state >= JOB_STATE_EXITING) &&
          (purge == false))
        {
        /* invalid state for request,  skip */
        continue;
        }

      int old_state = pjob->ji_qs.ji_state;
        
      running = (pjob->ji_qs.ji_state == JOB_STATE_RUNNING);

      pthread_mutex_unlock(pa->ai_mutex);

      if (purge)
        {
        deleted = true;

        try
          {
          force_purge_work(pjob);
          }
        catch (int err)
          {
          if (err != PBSE_JOB_RECYCLED)
            {
            char log_buf[LOCAL_LOG_BUF_SIZE];
            sprintf(log_buf, "Error when purging %s", pa->job_ids[i]);
            log_err(err, __func__, log_buf);
            deleted = false;
            }
          }
        }
      else
        deleted = attempt_delete(pjob);

      /* we come out of attempt_delete unlocked */
      pjob_mutex.set_unlock_on_exit(false);

      if (deleted == false)
        {
        /* if the job was deleted, this mutex would be taked care of elsewhere.
         * When it fails, release it here */
        num_skipped++;
        }
      else if (running == FALSE)
        {
        /* running jobs will increase the deleted count when their obit is reported */
        num_deleted++;
        }

      if ((pa = get_array(array_id.c_str())) == NULL)
        break;
      
      if ((deleted == true) &&
          (purge == false))
        pa->update_array_values(old_state, aeTerminate, pa->job_ids[i], cancel_exit_code);
      }
    }

  if (pa != NULL)
    pa->ai_qs.num_failed += num_deleted;

  if (num_jobs == 0)
    return(NO_JOBS_IN_ARRAY);

  return(num_skipped);
  } /* END delete_whole_array() */



/*
 * hold_array_range()
 * 
 * holds just a specified range from an array
 * @param pa - the array to be acted on
 * @param range_str - string specifying the range 
 */
int hold_array_range(

  job_array     *pa,         /* O */
  char          *range_str,  /* I */
  pbs_attribute *temphold)   /* I */

  {
  job                *pjob;
  std::vector<int>    range_vec;
  
  char *range = strchr(range_str,'=');
  if (range == NULL)
    return(PBSE_IVALREQ);

  range++; /* move past the '=' */
 
  if (translate_range_string_to_vector(range, range_vec))
    {
    /* don't hold the jobs if range error */
    
    return(PBSE_IVALREQ);
    }
  else 
    {
    for (size_t i = 0; i < range_vec.size(); i++)
      {
      int index = range_vec[i];
    
      /* don't stomp on other memory */
      if (index >= pa->ai_qs.array_size)
        continue;
      
      if (pa->job_ids[index] == NULL)
        continue;

      if ((pjob = svr_find_job(pa->job_ids[index], FALSE)) == NULL)
        {
        free(pa->job_ids[index]);
        pa->job_ids[index] = NULL;
        }
      else
        {
        hold_job(temphold,pjob);
        unlock_ji_mutex(pjob, __func__, NULL, LOGLEVEL);
        }
      }
    }

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




int release_array_range(

  job_array            *pa,
  struct batch_request *preq,
  char                 *range_str)

  {
  int                 rc;
  job                *pjob;
  std::vector<int>    range_vec;
  
  char *range = strchr(range_str,'=');
  if (range == NULL)
    return(PBSE_IVALREQ);

  range++; /* move past the '=' */
  
  if (translate_range_string_to_vector(range, range_vec))
    {
    /* don't hold the jobs if range error */
    
    return(PBSE_IVALREQ);
    }
  
  for (size_t i = 0; i < range_vec.size(); i++)
    {
    int index = range_vec[i];
    
    /* don't stomp on other memory */
    if (index >= pa->ai_qs.array_size)
      continue;

    if (pa->job_ids[index] == NULL)
      continue;

    if ((pjob = svr_find_job(pa->job_ids[index], FALSE)) == NULL)
      {
      free(pa->job_ids[index]);
      pa->job_ids[index] = NULL;
      }
    else
      {
      mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
      if ((rc = release_job(preq, pjob, pa)))
        {
        return(rc);
        }
      }
    }

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




int modify_array_range(

  job_array     *pa,              /* I/O */
  char          *range,           /* I */
  svrattrl      *plist,           /* I */
  batch_request *preq,            /* I */
  int            checkpoint_req)  /* I */

  {
  int                 rc = PBSE_NONE;
  job                *pjob;
  std::vector<int>    range_vec;

  if (translate_range_string_to_vector(range, range_vec))
    {
    /* don't hold the jobs if range error */
    
    return(FAILURE);
    }
  else 
    {
    bool array_gone = false;

    for (size_t i = 0; i < range_vec.size() && array_gone == false; i++)
      {
      int index = range_vec[i];

      if ((index >= pa->ai_qs.array_size) ||
          (pa->job_ids[index] == NULL))
        continue;

      if ((pjob = svr_find_job(pa->job_ids[index], FALSE)) == NULL)
        {
        free(pa->job_ids[index]);
        pa->job_ids[index] = NULL;
        }
      else
        {
        struct batch_request *array_req = duplicate_request(preq, index);
        mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
        pthread_mutex_unlock(pa->ai_mutex);
        array_req->rq_noreply = TRUE;
        modify_job((void **)&pjob, plist, array_req, checkpoint_req, NO_MOM_RELAY);
        pa = get_jobs_array(&pjob);

        if (pa == NULL)
          {
          array_gone = true;

          if (pjob == NULL)
            pjob_mutex.set_unlock_on_exit(false);
          break;
          }
        
        if (pjob == NULL)
          {
          pjob_mutex.set_unlock_on_exit(false);
          pa->job_ids[i] = NULL;
          }
        }
      }
    }

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


/*
 * set_slot_hold()
 *
 * Sets a hold on pjob if it isn't NULL and isn't held, incrementing difference, which should be negative
 * when passed into this function.
 */

void set_slot_hold(
    
  job *pjob,
  int &difference)

  {
  if (pjob != NULL)
    {
    if ((pjob->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l) == 0)
      {
      pjob->ji_wattr[JOB_ATR_hold].at_val.at_long |= HOLD_l;
      pjob->ji_wattr[JOB_ATR_hold].at_flags |= ATR_VFLAG_SET;

      difference++;
      }
    }
  } // END set_slot_hold()



/*
 * Releases the slot hold on pjob, if it has one
 *
 * @param job - the job whose hold should be released
 * @param difference - a count to be decremented for each hold released
 */

void release_slot_hold(

  job *pjob,
  int &difference)

  {
  if (pjob != NULL)
    {
    if (pjob->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l)
      {
      pjob->ji_wattr[JOB_ATR_hold].at_val.at_long &= ~HOLD_l;
      if (pjob->ji_wattr[JOB_ATR_hold].at_val.at_long == 0)
        pjob->ji_wattr[JOB_ATR_hold].at_flags &= ~ATR_VFLAG_SET;

      difference--;
      }
    }
  } // END release_slot_hold()



/*
 * update_slot_values()
 *
 * Updates the job array to have the correct jobs on hold and the correct running count
 *
 * @param pa - the job array
 * @param actually_running - the number of jobs that are actually running
 * @param number_queued - the actual number of queued jobs in the job array
 * @param held - the job pointer whose mutex we're holding
 * @param candidates - a vector of job ids that could have holds released
 * @return PBSE_NONE
 */

int update_slot_values(

  job_array                *pa,
  int                       actually_running,
  int                       number_queued,
  job                      *held,
  std::vector<std::string> &candidates)

  {
  unsigned int i = 0;
  int difference = pa->ai_qs.slot_limit - actually_running - number_queued;
  
  pa->ai_qs.jobs_running = actually_running;

  if (difference > 0)
    {
    release_slot_hold(held, difference);
  
    while ((difference > 0) &&
           (i < candidates.size()))
      {
      for (; i < candidates.size(); i++)
        {
        if (held != NULL)
          {
          if (held->ji_qs.ji_jobid == candidates[i])
            continue;
          }
        
        job *pj = svr_find_job(pa->job_ids[i], TRUE);
        
        if (pj != NULL)
          {
          release_slot_hold(pj, difference);
          unlock_ji_mutex(pj, __func__, NULL, LOGLEVEL);
          break;
          }
        }
      }
    }

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



/*
 * check_array_slot_limits()
 * Iterates over each job in this job's array to get a correct count of running and queued
 * subjobs.
 *
 * @param pjob - the job whose array we're checking
 * @return PBSE_NONE on success of PBSE_JOB_RECYCLED if the job is lost while locking it's array
 */

int check_array_slot_limits(

  job       *pjob,
  job_array *pa_held)

  {
  std::vector<std::string>  candidates;
  job_array                *pa;

  if (pa_held == NULL)
    {
    pa = get_jobs_array(&pjob);

    if (pa == NULL)
      {
      if (pjob == NULL)
        return(PBSE_JOB_RECYCLED);
      else
        return(PBSE_NONE);
      }
    }
  else
    pa = pa_held;

  mutex_mgr array_mgr(pa->ai_mutex, true);

  // Don't unlock the array if we held it coming in
  if (pa_held != NULL)
    array_mgr.set_unlock_on_exit(false);

  if (pa->ai_qs.slot_limit != NO_SLOT_LIMIT)
    {
    int  jobs_currently_running = 0;
    int  number_queued = 0;

    if (pjob->ji_qs.ji_state == JOB_STATE_RUNNING)
      jobs_currently_running++;
    else if (pjob->ji_qs.ji_state == JOB_STATE_QUEUED)
      number_queued++;
    
    job *pj;
    int  i = 0;

    // Only loop until we verify that we have the correct running count
    while ((jobs_currently_running < pa->ai_qs.jobs_running) &&
           (i < pa->ai_qs.num_jobs))
      {
      for (; i < pa->ai_qs.num_jobs; i++)
        {
        if (pa->job_ids[i] == NULL)
          continue;

        if (!strcmp(pjob->ji_qs.ji_jobid, pa->job_ids[i]))
          continue;

        if ((pj = svr_find_job(pa->job_ids[i], TRUE)) == NULL)
          {
          free(pa->job_ids[i]);
          pa->job_ids[i] = NULL;
          }
        else
          {
          mutex_mgr pj_mutex = mutex_mgr(pj->ji_mutex, true);

          if (pj->ji_qs.ji_state == JOB_STATE_RUNNING)
            {
            jobs_currently_running++;
            break;
            }
          else if (pj->ji_qs.ji_state == JOB_STATE_QUEUED)
            {
            number_queued++;
            }
          else if (pj->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l)
            {
            if ((int)candidates.size() < pa->ai_qs.jobs_running)
              candidates.push_back(pj->ji_qs.ji_jobid);
            }
          }
        }
      }

    if (jobs_currently_running != pa->ai_qs.jobs_running)
      update_slot_values(pa, jobs_currently_running, number_queued, pjob, candidates);
    }

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



/*
 * update_slot_held_jobs()
 *
 * Set or release up to num_to_release jobs that are currently held for slot limits in pa
 *
 * @param pa - the array whose jobs we are releasing
 * @param num_to_release - the maximum number of jobs whose slot limits should be released, or if negative,
 *                         the maximum number of jobs whose slot limits should be set.
 */

void update_slot_held_jobs(
    
  job_array *pa,
  int        num_to_release)

  {
  for (int i = 0; i < pa->ai_qs.num_jobs && num_to_release > 0; i++)
    {
    job *pjob = svr_find_job(pa->job_ids[i], TRUE);

    if (pjob != NULL)
      {
      release_slot_hold(pjob, num_to_release);
      unlock_ji_mutex(pjob, __func__, NULL, LOGLEVEL);
      }
    }
  
  for (int i = 0; i < pa->ai_qs.num_jobs && num_to_release < 0; i++)
    {
    job *pjob = svr_find_job(pa->job_ids[i], TRUE);

    if (pjob != NULL)
      {
      set_slot_hold(pjob, num_to_release);
      unlock_ji_mutex(pjob, __func__, NULL, LOGLEVEL);
      }
    }
  } // END update_slot_held_jobs()



/*
 * sets the state of the array summary job that is used strictly
 * for qstat displays.
 */

void update_array_statuses()

  {
  job_array      *pa;
  job            *pjob;
  all_arrays_iterator *iter = NULL;
  unsigned int    running;
  int             queued;
  unsigned int    complete;
  char            log_buf[LOCAL_LOG_BUF_SIZE];
  char            jobid[PBS_MAXSVRJOBID+1];

  while ((pa = next_array(&iter)) != NULL)
    {
    // If the array has been deleted, force the state to show as complete
    if (pa->is_deleted())
      {
      running = 0;
      complete = 1;
      queued = 0;
      }
    else
      {
      running  = pa->ai_qs.jobs_running;
      complete = pa->ai_qs.num_failed + pa->ai_qs.num_successful;
      queued   = pa->ai_qs.num_jobs - running - complete;
      if (queued < 0)
        queued = 0;
      }
    
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf, "%s: unlocking ai_mutex", __func__);
      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);
      }

    strcpy(jobid, pa->ai_qs.parent_id);
    unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
    
    if ((pjob = svr_find_job(jobid, TRUE)) != NULL)
      {
      mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);
      if (running > 0)
        {
        svr_setjobstate(pjob, JOB_STATE_RUNNING, pjob->ji_qs.ji_substate, FALSE);
        }
      else if ((complete > 0) && 
               (queued == 0))
        {
        svr_setjobstate(pjob, JOB_STATE_COMPLETE, pjob->ji_qs.ji_substate, FALSE);
        }
      else 
        {
        /* default to just calling the array queued */
        svr_setjobstate(pjob, JOB_STATE_QUEUED, pjob->ji_qs.ji_substate, FALSE);
        }
      }
    } /* END for each array */

  if (iter != NULL)
    delete iter;
  } /* END update_array_statuses() */




/* num_array_jobs()
 *
 * determine the number of jobs in the array from the array request 
 *
 * @param req_str - the string of the array request
 * @return - the number of jobs in the array, -1 on error 
 */

int num_array_jobs(

  const char *req_str) /* I */

  {
  int    num_jobs = 0;
  int    start;
  int    end;

  char  *delim = (char *)",";
  char  *ptr;
  char  *dash;
  char  *tmp_ptr;
  char   tmp_str[MAXPATHLEN];

  if (req_str == NULL)
    return(-1);

  snprintf(tmp_str, sizeof(tmp_str), "%s", req_str);
  tmp_ptr = tmp_str;
  ptr = threadsafe_tokenizer(&tmp_ptr, delim);

  while (ptr != NULL)
    {
    if ((dash = strchr(ptr,'-')) != NULL)
      {
      /* this is a range */
      start = atoi(ptr);
      end   = atoi(dash+1);

      /* check for invalid range */
      if (end < start)
        return(-1);

      num_jobs += end - start + 1;
      }
    else
      {
      /* just one job */
      num_jobs++;
      }

    ptr = threadsafe_tokenizer(&tmp_ptr, delim);
    }

  return(num_jobs);
  } /* END num_array_jobs */


/* 
 * insert pa into the global array 
 */
int insert_array(

  job_array *pa)

  {
  int rc = PBSE_NONE;

  allarrays.lock();

  if (!allarrays.insert(pa, pa->ai_qs.parent_id))
    {
    rc = ENOMEM;
    log_err(rc, __func__, "No memory to resize the array...SYSTEM FAILURE\n");
    }

  allarrays.unlock();

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





int remove_array(

  job_array *pa)

  {
  int  rc = PBSE_NONE;
  char arrayid[PBS_MAXSVRJOBID+1];

  if (allarrays.trylock())
    {
    strcpy(arrayid, pa->ai_qs.parent_id);

    unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);
    allarrays.lock();

    pa = allarrays.find(arrayid);

    if (pa != NULL)
      lock_ai_mutex(pa, __func__, "2", LOGLEVEL);
    }

  if (pa == NULL)
    rc = PBSE_NONE;
  else
    if (!allarrays.remove(pa->ai_qs.parent_id))
      rc = THING_NOT_FOUND;

  allarrays.unlock();

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





job_array *next_array(

    all_arrays_iterator **iter)

  {
  job_array *pa = NULL;

  allarrays.lock();

  if(*iter == NULL)
    {
    *iter = allarrays.get_iterator();
    }

  pa = (*iter)->get_next_item();
  
  if (pa != NULL)
    lock_ai_mutex(pa, __func__, NULL, LOGLEVEL);

  allarrays.unlock();

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




job_array *next_array_check(

  all_arrays_iterator **iter,
  job_array *owned)

  {
  job_array *pa = NULL;

  allarrays.lock();
  if(*iter == NULL)
    {
    *iter = allarrays.get_iterator();
    }

  pa = (*iter)->get_next_item();

  if ((pa != NULL) &&
      (pa != owned))
    lock_ai_mutex(pa, __func__, NULL, LOGLEVEL);
  
  allarrays.unlock();

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

/* END array_func.c */

