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

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

#include <string>
#include <sstream>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <ctype.h>
#include <errno.h>
#include <time.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <stdarg.h>
#include <assert.h>
#if defined(NTOHL_NEEDS_ARPA_INET_H) && defined(HAVE_ARPA_INET_H)
#include <arpa/inet.h>
#endif
#include <vector>

#include "portability.h"
#include "libpbs.h"
#include "server_limits.h"
#include "list_link.h"
#include "attribute.h"
#include "resource.h"
#include "server.h"
#include "net_connect.h"
#include "batch_request.h"
#include "work_task.h"
#include "svrfunc.h"
#include "pbs_job.h"
#include "log.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "pbs_nodes.h"
#include "dis.h"
#include "dis_init.h"
#include "resmon.h"
#include "mcom.h"
#include "utils.h"
#include "u_tree.h"
#include "threadpool.h"
#include "node_func.h" /* find_nodebyname */
#include "../lib/Libutils/u_lock_ctl.h" /* lock_node, unlock_node */
#include "../lib/Libnet/lib_net.h" /* socket_read_flush */
#include "svr_func.h" /* get_svr_attr_* */
#include "alps_functions.h"
#include "login_nodes.h"
#include "svr_connect.h" /* svr_disconnect_sock */
#include "net_cache.h"
#include "ji_mutex.h"
#include "alps_constants.h"
#include "mutex_mgr.hpp"
#include "timer.hpp"

#define IS_VALID_STR(STR)  (((STR) != NULL) && ((STR)[0] != '\0'))

extern int              LOGLEVEL;

#if !defined(H_ERRNO_DECLARED) && !defined(_AIX)
/*extern int              h_errno;*/
#endif
  

int                     svr_totnodes = 0; /* total number nodes defined       */
int                     svr_clnodes  = 0; /* number of cluster nodes     */
int                     svr_chngNodesfile = 0; /* 1 signals want nodes file update */
int                     gpu_mode_rqstd = -1;  /* default gpu mode requested */
int                     gpu_err_reset = FALSE;    /* was a gpu errcount reset requested */
/* on server shutdown, (qmgr mods)  */

all_nodes               allnodes;

static int              num_addrnote_tasks = 0; /* number of outstanding send_cluster_addrs tasks */
pthread_mutex_t        *addrnote_mutex = NULL;

extern int              server_init_type;
extern int              has_nodes;

extern int create_a_gpusubnode(struct pbsnode *);
int        is_gpustat_get(struct pbsnode *np, char **str_ptr);

extern int              ctnodes(char *);

extern char            *path_home;
extern char            *path_nodes;
extern char            *path_nodes_new;
extern char            *path_nodestate;
extern char            *path_nodenote;
extern char            *path_nodenote_new;
extern char             server_name[];

extern struct server    server;
extern tlist_head       svr_newnodes;
extern attribute_def    node_attr_def[];   /* node attributes defs */
extern int              SvrNodeCt;
extern struct all_jobs  alljobs;

extern int              multi_mom;

#define SKIP_NONE       0
#define SKIP_EXCLUSIVE  1
#define SKIP_ANYINUSE   2
#define SKIP_NONE_REUSE 3

#ifndef MAX_BM
#define MAX_BM          64
#endif

int handle_complete_first_time(job *pjob);
int is_compute_node(char *node_id);
int hasprop(struct pbsnode *, struct prop *);
int add_job_to_node(struct pbsnode *,struct pbssubn *,short,job *);
int node_satisfies_request(struct pbsnode *,char *);
int reserve_node(struct pbsnode *,short,job *,char *,struct howl **);
int build_host_list(struct howl **,struct pbssubn *,struct pbsnode *);
int procs_available(int proc_ct);
void check_nodes(struct work_task *);
int gpu_entry_by_id(struct pbsnode *,char *, int);
job *get_job_from_jobinfo(struct jobinfo *,struct pbsnode *);

/* marks a stream as finished being serviced */
pthread_mutex_t        *node_state_mutex = NULL;





/**
**      Modified by Tom Proett <proett@nas.nasa.gov> for PBS.
*/

AvlTree                 ipaddrs = NULL;


/**
 * specialized version of tfind for looking in the ipadders tree
 * @param key - the node we are searching for
 * @return a pointer to the pbsnode
*/

struct pbsnode *tfind_addr(

  const u_long  key,
  uint16_t      port,
  char         *job_momname)

  {
  struct pbsnode *pn = AVL_find(key,port,ipaddrs);

  if (pn == NULL)
    return(NULL);

  lock_node(pn, __func__, "pn", LOGLEVEL);

  if (pn->num_node_boards == 0)
    return(pn);
  else
    {
    char *dash = NULL;
    char *plus = NULL;
    char *tmp = job_momname;

    struct pbsnode *numa = NULL;

    int index;

    plus = strchr(tmp,'+');

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

    while ((tmp = strchr(tmp,'-')) != NULL)
      {
      dash = tmp;
      tmp++;
      }

    if (dash == NULL)
      {
      /* node has numa nodes but no dashes in exec host?? */
      log_err(-1, __func__, "Numa node but there's no dash in exec_host?");

      return(pn);
      }

    index = atoi(dash+1);

    numa = AVL_find(index, pn->nd_mom_port, pn->node_boards);

    unlock_node(pn, __func__, "pn->numa", LOGLEVEL);
    lock_node(numa, __func__, "numa", LOGLEVEL);

    if (plus != NULL)
      *plus = '+';

    return(numa);
    }
  } /* END tfind_addr() */





/* update_node_state - central location for updating node state */
/* NOTE:  called each time a node is marked down, each time a MOM reports node  */
/*        status, and when pbs_server sends hello/cluster_addrs */

void update_node_state(

  struct pbsnode *np,         /* I (modified) */
  int             newstate)   /* I (one of INUSE_*) */

  {
  char            log_buf[LOCAL_LOG_BUF_SIZE];

  /* No need to do anything if newstate == oldstate */
  if (np->nd_state == newstate)
    return;

  /*
   * LOGLEVEL >= 4 logs all state changes
   *          >= 2 logs down->(busy|free) changes
   *          (busy|free)->down changes are always logged
   */

  if (LOGLEVEL >= 4)
    {
    sprintf(log_buf, "adjusting state for node %s - state=%d, newstate=%d",
      (np->nd_name != NULL) ? np->nd_name : "NULL",
      np->nd_state,
      newstate);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  log_buf[0] = '\0';

  if (newstate & INUSE_DOWN)
    {
    if (!(np->nd_state & INUSE_DOWN))
      {
      sprintf(log_buf, "node %s marked down",
        (np->nd_name != NULL) ? np->nd_name : "NULL");

      np->nd_state |= INUSE_DOWN;
      np->nd_state &= ~INUSE_UNKNOWN;
      }

    /* ignoring the obvious possibility of a "down,busy" node */
    }  /* END if (newstate & INUSE_DOWN) */
  else if (newstate & INUSE_BUSY)
    {
    if ((!(np->nd_state & INUSE_BUSY) && (LOGLEVEL >= 4)) ||
        ((np->nd_state & INUSE_DOWN) && (LOGLEVEL >= 2)))
      {
      sprintf(log_buf, "node %s marked busy",
        (np->nd_name != NULL) ? np->nd_name : "NULL");
      }

    np->nd_state |= INUSE_BUSY;

    np->nd_state &= ~INUSE_UNKNOWN;

    if (np->nd_state & INUSE_DOWN)
      {
      np->nd_state &= ~INUSE_DOWN;
      }
    }  /* END else if (newstate & INUSE_BUSY) */
  else if (newstate == INUSE_FREE)
    {
    if (((np->nd_state & INUSE_DOWN) && (LOGLEVEL >= 2)) ||
        ((np->nd_state & INUSE_BUSY) && (LOGLEVEL >= 4)))
      {
      sprintf(log_buf, "node %s marked free",
        (np->nd_name != NULL) ? np->nd_name : "NULL");
      }

    np->nd_state &= ~INUSE_BUSY;

    np->nd_state &= ~INUSE_UNKNOWN;

    if (np->nd_state & INUSE_DOWN)
      {
      np->nd_state &= ~INUSE_DOWN;
      }
    }    /* END else if (newstate == INUSE_FREE) */

  if (newstate & INUSE_UNKNOWN)
    {
    np->nd_state |= INUSE_UNKNOWN;
    }

  if ((LOGLEVEL >= 2) && (log_buf[0] != '\0'))
    {
    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  return;
  }  /* END update_node_state() */




int check_node_for_job(

  struct pbsnode *pnode,
  char           *jobid)

  {
  for (int i = 0; i < (int)pnode->nd_job_usages.size(); i++)
    {
    job_usage_info *jui = pnode->nd_job_usages[i];

    if (!strcmp(jobid, jui->jobid))
      return(TRUE);
    }

  /* not found */
  return(FALSE);
  } /* END check_node_for_job() */



/*
 * record_reported_time()
 *
 * @pre-cond: vp must be a character pointer to a job id
 * @post-cond: the job's last reported time is updated. This is used to safeguard against
 * deleting jobs pre-maturely
 *
 * @param vp - the jobid of the job
 *
 */
void *record_reported_time(

  void *vp)

  {
  char *job_and_node = (char *)vp;

  if (job_and_node != NULL)
    {
    char *jobid = job_and_node;
    char *node_id;
    char *colon = strchr(job_and_node, ':');

    if (colon != NULL)
      {
      *colon = '\0';
      node_id = colon + 1;

      job *pjob = svr_find_job(jobid, TRUE);

      if (pjob != NULL)
        {
        mutex_mgr job_mutex(pjob->ji_mutex, true);

        if (pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str != NULL)
          {
          if (!strncmp(node_id, pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str, strlen(node_id)))
            pjob->ji_last_reported_time = time(NULL);
          }
        }
      }

    free(job_and_node);
    }

  return(NULL);
  } /* END record_reported_time() */




/*
 * is_job_on_node - return TRUE if this jobid is present on pnode
 */

int is_job_on_node(

  struct pbsnode *pnode, /* I */
  char           *jobid) /* I */

  {
  struct pbsnode *numa;

  int             present = FALSE;
  char           *at;
  int             i;

  if ((at = strchr(jobid, (int)'@')) != NULL)
    *at = '\0'; /* strip off @server_name */

  if (pnode->num_node_boards > 0)
    {
    /* check each subnode on each numa node for the job */
    for (i = 0; i < pnode->num_node_boards; i++)
      {
      numa = AVL_find(i,pnode->nd_mom_port,pnode->node_boards);

      lock_node(numa, __func__, "before check_node_for_job numa", LOGLEVEL);
      present = check_node_for_job(pnode,jobid);
      unlock_node(numa, __func__, "after check_node_for_job numa", LOGLEVEL);

      /* leave loop if we found the job */
      if (present != FALSE)
        break;
      } /* END for each numa node */
    }
  else
    {
    present = check_node_for_job(pnode, jobid);
    if (present == TRUE)
      {
      char *job_and_node = (char *)calloc(1, strlen(jobid) + 2 + strlen(pnode->nd_name));
      sprintf(job_and_node, "%s:%s", jobid, pnode->nd_name);
      enqueue_threadpool_request(record_reported_time, job_and_node);
      }
    }

  if (at != NULL)
    *at = '@';  /* restore @server_name */

  return(present);
  }  /* END is_job_on_node() */




/* If nodes have similiar names this will make sure the name is an exact match.
 * Not just found inside another name.
 * i.e. Machines by the name of gpu, gpuati, gpunvidia. If searching for gpu...
 * List format is similiar to: gpuati+gpu/1+gpunvidia/4+gpu/5
 */
int node_in_exechostlist(
    
  char *node_name,
  char *node_ehl)

  {
  int   rc = FALSE;
  char *cur_pos = node_ehl;
  char *new_pos = cur_pos;
  int   name_len = strlen(node_name);

  while (1)
    {
    if ((new_pos = strstr(cur_pos, node_name)) == NULL)
      break;
    else if (new_pos == node_ehl)
      {
      if ((new_pos+name_len == NULL) ||
          (*(new_pos+name_len) == '+') ||
          (*(new_pos+name_len) == '/'))
        {
        rc = TRUE;
        break;
        }
      }
    else if (*(new_pos-1) == '+')
      {
      if ((new_pos+name_len == NULL) ||
          (*(new_pos+name_len) == '+') ||
          (*(new_pos+name_len) == '/'))
        {
        rc = TRUE;
        break;
        }
      }

    cur_pos = new_pos+1;
    }

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




int kill_job_on_mom(

  char           *jobid,
  struct pbsnode *pnode)

  {
  batch_request *preq;
  int            rc = -1;
  int            conn;
  int            local_errno = 0;
  char           log_buf[LOCAL_LOG_BUF_SIZE];

  /* job is reported by mom but server has no record of job */
  sprintf(log_buf, "stray job %s found on %s", jobid, pnode->nd_name);
  log_err(-1, __func__, log_buf);
  
  conn = svr_connect(pnode->nd_addrs[0], pnode->nd_mom_port, &local_errno, pnode, NULL, ToServerDIS);

  if (conn >= 0)
    {
    if ((preq = alloc_br(PBS_BATCH_SignalJob)) == NULL)
      {
      log_err(-1, __func__, "unable to allocate SignalJob request-trouble!");
      svr_disconnect(conn);
      }
    else
      {
      snprintf(preq->rq_ind.rq_signal.rq_jid, sizeof(preq->rq_ind.rq_signal.rq_jid), "%s", jobid);
      snprintf(preq->rq_ind.rq_signal.rq_signame, sizeof(preq->rq_ind.rq_signal.rq_signame), "SIGKILL");
      preq->rq_extra = strdup(SYNC_KILL);
      
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      rc = issue_Drequest(conn, preq, true);
      free_br(preq);
      lock_node(pnode, __func__, NULL, LOGLEVEL);
      }
    }

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



pthread_mutex_t jobsKilledMutex = PTHREAD_MUTEX_INITIALIZER;
std::vector<std::string> jobsKilled;
#define JOB_SYNC_TIMEOUT 60 //Once a kill job has been sent to a MOM, don't send another for five minutes.



/*
 * Delayed task to remove a killed job from the list in
 * case it needs to be removed again.
 */

void remove_job_from_already_killed_list(struct work_task *pwt)
  {
  std::string *pJobID = (std::string *)pwt->wt_parm1;

  free(pwt->wt_mutex);
  free(pwt);

  if(pJobID == NULL) return;

  pthread_mutex_lock(&jobsKilledMutex);

  for(std::vector<std::string>::iterator i = jobsKilled.begin();i != jobsKilled.end();)
    {
    if((*i).compare(*pJobID) == 0)
      {
      i = jobsKilled.erase(i);
      if(i == jobsKilled.end())
        {
        break;
        }
      continue;
      }
    i++;
    }
  pthread_mutex_unlock(&jobsKilledMutex);

  delete pJobID;

  }

/*
 * If a job is not supposed to be on a node and we have
 * not sent a kill to that job in the last 5 minutes
 * then the job should be killed.
 */

bool job_should_be_killed(
    
  char           *jobid,
  struct pbsnode *pnode)

  {
  bool  should_be_on_node = true;
  bool  should_kill_job = false;
  job *pjob;
  
  if (strstr(jobid, server_name) != NULL)
    {
    if ((is_job_on_node(pnode, jobid)) == FALSE)
      {
      /* must lock the job before the node */
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      pjob = svr_find_job(jobid, TRUE);
      lock_node(pnode, __func__, NULL, LOGLEVEL);
      
      if (pjob != NULL)
        {
        /* job exists, but doesn't currently have resources assigned
         * to this node double check the job struct because we
         * could be in the middle of moving the job around because
         * of data staging, suspend, or rerun */            
        mutex_mgr job_mgr(pjob->ji_mutex,true);
        if (pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str == NULL)
          {
          should_be_on_node = false;
          }
        else if (node_in_exechostlist(pnode->nd_name, pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str) == FALSE)
          {
          should_be_on_node = false;
          }
        }
      else
        should_be_on_node = false;
      }
    }

  if(!should_be_on_node)
    {
    bool jobAlreadyKilled = false;
    //Job should not be on the node, see if we have already sent a kill for this job.
    pthread_mutex_lock(&jobsKilledMutex);

    for(std::vector<std::string>::iterator i = jobsKilled.begin();(i != jobsKilled.end())&&(jobAlreadyKilled == false);i++)
      {
      if((*i).compare(jobid) == 0)
        {
        jobAlreadyKilled = true;
        }
      }
    if(!jobAlreadyKilled)
      {
      should_kill_job = true;
      }
    pthread_mutex_unlock(&jobsKilledMutex);
    }

  return(should_kill_job);
  } /* END job_should_be_on_node() */


void *finish_job(

  void *vp)

  {
  char *jobid = (char *)vp;
  job  *pjob;

  if (jobid == NULL)
    return(NULL);
  else if ((pjob = svr_find_job(jobid, TRUE)) == NULL)
    {
    free(jobid);

    return(NULL);
    }
  mutex_mgr job_mgr(pjob->ji_mutex,true);
  job_mgr.set_unlock_on_exit(false);

  free(jobid);

  free_nodes(pjob);

  svr_setjobstate(pjob, JOB_STATE_COMPLETE, JOB_SUBSTATE_COMPLETE, FALSE);

  handle_complete_first_time(pjob);
  /* pjob->ji_mutex is always returned unlocked from handle_complete_first_time */

  return(NULL);
  } /* END finish_job() */




int remove_jobs_that_have_disappeared(

  struct pbsnode  *pnode,
  resizable_array *reported_ms_jobs,
  time_t           timestamp)

  {
  int   iter = -1;
  char  log_buf[LOCAL_LOG_BUF_SIZE];
  char *jobid;
  
  while ((jobid = (char *)pop_thing(pnode->nd_ms_jobs)) != NULL)
    {
    job *pjob;

    /* locking priority is job before node */
    unlock_node(pnode, __func__, NULL, LOGLEVEL);
    pjob = svr_find_job(jobid, TRUE);
    lock_node(pnode, __func__, NULL, LOGLEVEL);

    if (pjob == NULL)
      {
      free(jobid);

      continue;
      }
    mutex_mgr job_mgr(pjob->ji_mutex,true);

    /* 45 seconds is typically the time between intervals for each update 
     * from the mom. Add this in case a stale update is processed and the job
     * hadn't started at the time the update was sent */
    if ((pjob->ji_qs.ji_state >= JOB_STATE_EXITING) ||
        (pjob->ji_qs.ji_substate < JOB_SUBSTATE_RUNNING) ||
        (pjob->ji_wattr[JOB_ATR_start_time].at_val.at_long > timestamp - 45))
      {
      free(jobid);

      job_mgr.unlock();
      continue;
      }

    job_mgr.unlock();
    /* mom didn't report this job - it has exited unexpectedly */
    snprintf(log_buf, sizeof(log_buf),
      "Server thinks job %s is on node %s, but the mom doesn't. Terminating job",
      jobid, pnode->nd_name);
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, jobid, log_buf);  

    enqueue_threadpool_request(finish_job, jobid);
    }

  /* now replace the old resizable array with the ones currently reported */
  while ((jobid = (char *)next_thing(reported_ms_jobs, &iter)) != NULL)
    insert_thing(pnode->nd_ms_jobs, jobid);

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



/*
 * is_jobid_in_mom()
 * returns: true if jobid was found; false otherwise.
 */
bool is_jobid_in_mom(

  const char *jobs,
  const char *jobid)

  {
  char *joblist = strdup(jobs);
  char *jobptr = joblist;
  char *jobidstr = NULL;

  jobidstr = threadsafe_tokenizer(&jobptr, " ");
  while (jobidstr != NULL)
    {
    if (strcmp(jobid, jobidstr) == 0)
      {
      free(joblist);
      return(true);
      }

    jobidstr = threadsafe_tokenizer(&jobptr, " ");
    }

  free(joblist);

  return(false);
  }



/*
 * sync_node_jobs_with_moms() - remove any jobs in the pbsnode (np) that was not
 * reported by the mom that it's currently running in its status update.
 */
void sync_node_jobs_with_moms(

  struct pbsnode *np,        /* I */
  const char *jobs_in_mom)   /* I */

  {
  std::vector<std::string> jobsRemoveFromNode;
  bool removealljobs = (strlen(jobs_in_mom) == 0);

  for (int i = 0; i < (int)np->nd_job_usages.size(); i++)
    {
    bool removejob = false;
    job_usage_info *jui = np->nd_job_usages[i];
    char *jobid = jui->jobid;

    if (!removealljobs)
      {
      char *p = strstr((char *)jobs_in_mom, jobid);
      /* job is in the node but not in mom */
      if (!p)
        removejob = true;
      else if (is_jobid_in_mom(jobs_in_mom, jobid) == false)
        removejob = true;
      }
    if (removejob || removealljobs)
      {
      unlock_node(np, __func__, NULL, LOGLEVEL);
      job *pjob = svr_find_job(jobid, TRUE);
      lock_node(np, __func__, NULL, LOGLEVEL);
      if (pjob)
        unlock_ji_mutex(pjob, __func__, NULL, LOGLEVEL);
      else
        jobsRemoveFromNode.push_back(std::string (jobid));
      }
    }

  char log_buf[LOCAL_LOG_BUF_SIZE + 1];
  for (std::vector<std::string>::iterator it = jobsRemoveFromNode.begin();
             it != jobsRemoveFromNode.end(); it++)
    {
    snprintf(log_buf, sizeof(log_buf),
      "Job %s was not reported in %s update status. Freeing job from node.", (*it).c_str(), np->nd_name);
    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);
    remove_job_from_node(np, (*it).c_str());
    }
  } /* end of sync_node_jobs_with_moms */





/*
 * sync_node_jobs() - determine if a MOM has a stale job and possibly delete it
 *
 * This function is called every time we get a node stat from the pbs_mom.
 *
 * NOTE: changed to be processed in a thread so that processing here doesn't hinder
 * the server's ability to reply to the status
 *
 * @see is_stat_get()
 */

void *sync_node_jobs(

  void *vp)

  {
  struct pbsnode       *np;
  sync_job_info        *sji = (sync_job_info *)vp;
  char                 *raw_input;
  char                 *node_id;
  char                 *jobstring_in;
  char                 *joblist;
  char                 *jobidstr;
  long                  job_sync_timeout = JOB_SYNC_TIMEOUT;
  char                 *jobs_in_mom = NULL;

  if (vp == NULL)
    return(NULL);

  raw_input = sji->input;

  /* raw_input's format is:
   *   node name:<JOBID>[ <JOBID>]... */
  if ((jobstring_in = strchr(raw_input, ':')) != NULL)
    {
    node_id = raw_input;
    *jobstring_in = '\0';
    jobstring_in++;
    }
  else
    {
    /* bad input */
    free(raw_input);
    free(sji);

    return(NULL);
    }
    
  if ((np = find_nodebyname(node_id)) == NULL)
    {
    free(raw_input);
    free(sji);

    return(NULL);
    }

  /* FORMAT <JOBID>[ <JOBID>]... */
  jobs_in_mom = strdup(jobstring_in);
  joblist = jobstring_in;
  jobidstr = threadsafe_tokenizer(&joblist, (char *)" ");

  get_svr_attr_l(SRV_ATR_job_sync_timeout, &job_sync_timeout);

  while ((jobidstr != NULL) && 
         (isdigit(*jobidstr)) != FALSE)
    {
    if (job_should_be_killed(jobidstr, np))
      {
      if (kill_job_on_mom(jobidstr, np) == PBSE_NONE)
        {
        pthread_mutex_lock(&jobsKilledMutex);
        std::string str(jobidstr);
        jobsKilled.push_back(str);
        pthread_mutex_unlock(&jobsKilledMutex);
        set_task(WORK_Timed, 
                 time(NULL) + job_sync_timeout,
                 remove_job_from_already_killed_list,
                 (void *)new std::string(jobidstr),
                 FALSE);
        }
      }
    
    jobidstr = threadsafe_tokenizer(&joblist, " ");
    } /* END while ((jobidstr != NULL) && ...) */

  /* SUCCESS */
  free(raw_input);

  free(sji);

  if (jobs_in_mom)
    {
    sync_node_jobs_with_moms(np, jobs_in_mom);
    free(jobs_in_mom);
    }

  unlock_node(np, __func__, NULL, LOGLEVEL);

  return(NULL);
  }  /* END sync_node_jobs() */



/*
 *      setup_notification -  Sets up the  mechanism for notifying
 *                            other members of the server's node
 *                            pool that a new node was added manually
 *                            via qmgr.  Actual notification occurs some
 *                            time later through the send_cluster_addrs mechanism
 */

void setup_notification(
    
  char *pname) /* I */

  {
  struct pbsnode *pnode;
  new_node       *nnew;

  if (pname != NULL)
    {
    if ((pnode = find_nodebyname(pname)) != NULL)
      {
      /* call it offline until after all nodes get the new ipaddr */
      pnode->nd_state |= INUSE_OFFLINE;
      
      nnew = (new_node *)calloc(1, sizeof(new_node));
      
      if (nnew == NULL)
        {
        unlock_node(pnode, __func__, "nnew == NULL", LOGLEVEL);
        return;
        }
      
      CLEAR_LINK(nnew->nn_link);
      
      nnew->nn_name = strdup(pname);
      
      append_link(&svr_newnodes, &nnew->nn_link, nnew);
      
      unlock_node(pnode, __func__, "nnew != NULL", LOGLEVEL);
      }
    }

  if (addrnote_mutex == NULL)
    {
    addrnote_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));
    pthread_mutex_init(addrnote_mutex,NULL);
    }

  pthread_mutex_lock(addrnote_mutex);
  num_addrnote_tasks++;
  pthread_mutex_unlock(addrnote_mutex);

  return;
  } /* END setup_notification() */







/*
 **     reset gpu data in case mom reconnects with changed gpus.
 **     If we have real gpus, not virtual ones, then clear out gpu_status,
 **     gpus count and remove gpu subnodes.
 */

void clear_nvidia_gpus(

  struct pbsnode *np)  /* I */

  {
  pbs_attribute   temp;

  if ((np->nd_gpus_real) && (np->nd_ngpus > 0))
    {
    /* delete gpusubnodes by freeing it */
    free(np->nd_gpusn);
    np->nd_gpusn = NULL;

    /* reset # of gpus, etc */
    np->nd_ngpus = 0;
    np->nd_ngpus_free = 0;

    /* unset "gpu_status" node attribute */

    memset(&temp, 0, sizeof(temp));

    if (decode_arst(&temp, NULL, NULL, NULL, 0))
      {
      log_err(-1, __func__, "clear_nvidia_gpus:  cannot initialize attribute\n");

      return;
      }

    node_gpustatus_list(&temp, np, ATR_ACTION_ALTER);
    }

  return;
  }  /* END clear_nvidia_gpus() */





/* EOF on a stream received (either stream or addr must be specified) */
/* mark node down */
/* NOTE: pass in stream = -1 if you wish the stream to be optional */

void stream_eof(

  int       stream,  /* I (optional) */
  u_long    addr,  /* I (optional) */
  uint16_t  port,  /* I (optional) */
  int       ret)     /* I (ignored) */

  {
  char            log_buf[LOCAL_LOG_BUF_SIZE];
  enum conn_type  cntype = ToServerDIS;
  int             conn;
  int             my_err = 0;

  struct pbsnode *np = NULL;

  if (LOGLEVEL >= 6)
    {
    sprintf(log_buf, "stream: %d, addr: %ld, port %d", stream, addr, port);
    LOG_EVENT(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);
    }

  if (addr != 0)
    {
    np = AVL_find(addr, port, ipaddrs);
    }

  if (np == NULL)
    {
    /* cannot locate node */

    return;
    }

  /* Before we mark this node down see if we can connect */
  lock_node(np, __func__, "parent", LOGLEVEL);
  conn = svr_connect(addr, port, &my_err, np, NULL, cntype);
  if(conn >= 0)
    {
    unlock_node(np, __func__, "parent", LOGLEVEL);
    svr_disconnect(conn);
    return;
    }


  sprintf(log_buf,
    "connection to %s is no longer valid, connection may have been closed remotely, remote service may be down, or message may be corrupt (%s).  setting node state to down",
    np->nd_name,
    dis_emsg[ret]);

  log_err(-1, __func__, log_buf);

  /* mark node and all subnodes as down */

  if (np->num_node_boards > 0)
    {
    int             i;
    struct pbsnode *pnode;

    for (i = 0; i < np->num_node_boards; i++)
      {
      pnode = AVL_find(i,np->nd_mom_port,np->node_boards);

      lock_node(pnode, __func__, "subs", LOGLEVEL);
      update_node_state(pnode,INUSE_DOWN);
      unlock_node(pnode, __func__, "subs", LOGLEVEL);
      }
    }
  else
    update_node_state(np, INUSE_DOWN);

  unlock_node(np, __func__, "parent", LOGLEVEL);

  return;
  }  /* END stream_eof() */


/*
 * wrapper task that check_nodes places in the thread pool's queue
 */
void *check_nodes_work(

  void *vp)

  {
  work_task        *ptask = (struct work_task *)vp;

  struct pbsnode   *np = NULL;
  long              chk_len = 300;
  char              log_buf[LOCAL_LOG_BUF_SIZE];
  time_t            time_now = time(NULL);

  node_iterator     iter;
  
  /* load min refresh interval */
  get_svr_attr_l(SRV_ATR_check_rate, &chk_len);

  if (LOGLEVEL >= 5)
    {
    sprintf(log_buf, "verifying nodes are active (min_refresh = %d seconds)", (int)chk_len);

    log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, log_buf);
    }

  /* evaluate all nodes */
  reinitialize_node_iterator(&iter);

  while ((np = next_node(&allnodes,np,&iter)) != NULL)
    {
    if (!(np->nd_state & INUSE_DOWN))
      {
      if (np->nd_lastupdate < (time_now - chk_len)) 
        {
        if (LOGLEVEL >= 6)
          {
          sprintf(log_buf, "node %s not detected in %ld seconds, contacting mom",
          np->nd_name,
          (long int)(time_now - np->nd_lastupdate));
          
          log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, log_buf);
          }
          
        if (LOGLEVEL >= 0)
          {
          sprintf(log_buf, "node %s not detected in %ld seconds, marking node down",
            np->nd_name,
            (long int)(time_now - np->nd_lastupdate));
          
          log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, log_buf);
          }
        
          update_node_state(np, (INUSE_DOWN));    
          
        /* The node is up. Do not mark the node down, but schedule a check_nodes */
        }
      }
    } /* END for each node */

  if (ptask->wt_parm1 == NULL)
    {
    set_task(WORK_Timed, time_now + chk_len, check_nodes, (char *)NULL,FALSE);
    }

  /* since this is done via threading, we now free the task here */
  free(ptask->wt_mutex);
  free(ptask);

  return(NULL);
  } /* check_nodes_work() */




/*
 * Mark any nodes that haven't checked in as down.
 * If the node isn't down then it checks to see that the
 * last update hasn't been too long ago.
 */

void check_nodes(

  struct work_task *ptask)  /* I (modified) */

  {
  int rc = enqueue_threadpool_request(check_nodes_work,ptask);

  if (rc)
    {
    log_err(rc, __func__, "Unable to enqueue check nodes task into the threadpool");
    }
  }  /* END check_nodes() */







void *write_node_state_work(

  void *vp)

  {
  struct pbsnode *np;
  static char    *fmt = (char *)"%s %d\n";
  static FILE    *nstatef = NULL;
  int             iter = -1;

  int             savemask;

  pthread_mutex_lock(node_state_mutex);

  if (LOGLEVEL >= 5)
    {
    DBPRT(("write_node_state_work: entered\n"))
    }

  /* don't store volatile states like down and unknown */

  savemask = INUSE_OFFLINE | INUSE_RESERVE;

  if (nstatef != NULL)
    {
    fseek(nstatef, 0L, SEEK_SET); /* rewind and clear */

    if (ftruncate(fileno(nstatef), (off_t)0) != 0)
      {
      log_err(errno, __func__, "could not truncate file");

      pthread_mutex_unlock(node_state_mutex);
      
      return(NULL);
      }
    }
  else
    {
    /* need to open for first time, temporary-move to pbsd_init */

    if ((nstatef = fopen(path_nodestate, "w+")) == NULL)
      {
      log_err(errno, __func__, "could not open file");

      pthread_mutex_unlock(node_state_mutex);
      
      return(NULL);
      }
    }

  /*
  ** The only state that carries forward is if the
  ** node has been marked offline.
  */

  while ((np = next_host(&allnodes,&iter,NULL)) != NULL)
    {
    if (np->nd_state & INUSE_OFFLINE)
      {
      fprintf(nstatef, fmt, np->nd_name, np->nd_state & savemask);
      }

    unlock_node(np, __func__, NULL, LOGLEVEL);
    } /* END for each node */

  if (fflush(nstatef) != 0)
    {
    log_err(errno, __func__, "failed saving node state to disk");
    }

  fclose(nstatef);
  nstatef = NULL;

  pthread_mutex_unlock(node_state_mutex);

  return(NULL);
  } /* END write_node_state_work() */





void write_node_state(void)

  {
  int rc = enqueue_threadpool_request(write_node_state_work,NULL);

  if (rc)
    {
    log_err(rc, __func__, "Unable to enqueue write_node_state_work task into the threadpool");
    }
  }  /* END write_node_state() */



/* Create a new node_note file then overwrite the previous one.
 *
 *   The note file could get up to:
 *      (# of nodes) * (2 + MAX_NODE_NAME + MAX_NOTE)  bytes in size
 */
int write_node_note(void)

  {
  struct pbsnode *np;
  int             iter = -1;
  FILE           *nin;

  if (LOGLEVEL >= 2)
    {
    DBPRT(("%s: entered\n", __func__))
    }

  if ((nin = fopen(path_nodenote_new, "w")) == NULL)
    goto err1;

  if (svr_totnodes == 0)
    {
    log_event(
      PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, "Server has empty nodes list");

    fclose(nin);

    return(-1);
    }

  /* for each node ... */
  while ((np = next_host(&allnodes, &iter, NULL)) != NULL)
    {
    /* write node name followed by its note string */
    if ((np->nd_note != NULL) && 
        (np->nd_note[0] != '\0'))
      {
      fprintf(nin, "%s %s\n", np->nd_name, np->nd_note);
      }
    
    unlock_node(np, __func__, NULL, LOGLEVEL);
    }

  fflush(nin);

  if (ferror(nin))
    {
    fclose(nin);
    goto err1;
    }

  fclose(nin);

  if (rename(path_nodenote_new, path_nodenote) != 0)
    {
    log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__,
      (char *)"replacing old node note file failed");

    return(-1);
    }

  return(PBSE_NONE);

err1:
  log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__,
    (char *)"Node note file update failed");

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



/*
 * free_prop - free list of prop structures created by proplist()
 */

static void free_prop(

  struct prop *prop)

  {
  struct prop *pp;

  for (pp = prop;pp != NULL;pp = prop)
    {
    prop = pp->next;

    free(pp->name);
    free(pp);
    }  /* END for (pp) */

  return;
  }    /* END free_prop() */




void *node_unreserve_work(

  void *vp)

  {
  resource_t       handle = *((resource_t *)vp);

  struct  pbsnode *np;
  int              iter = -1;

  /* clear old reserve */
  while ((np = next_host(&allnodes,&iter,NULL)) != NULL)
    {
    if (handle == RESOURCE_T_ALL)
      np->nd_np_to_be_used = 0;

    unlock_node(np, "node_unreserve_work", NULL, LOGLEVEL);
    }

  return(NULL);
  } /* END node_unreserve_work() */





/*
 * unreserve - unreserve nodes
 *
 * If handle is set to a existing resource_t, then release all nodes
 * associated with that handle, otherwise, (this is dangerous)
 * if handle == RESOURCE_T_ALL, release all nodes period.
 */

void node_unreserve(

  resource_t handle)

  {
  int rc = enqueue_threadpool_request(node_unreserve_work,NULL);

  if (rc)
    {
    log_err(rc, __func__, "Unable to enqueue node_unreserve task into the threadpool");
    }
  }  /* END node_unreserve() */




/*
** Look through the property list and make sure that all
** those marked are contained in the node.
*/

int hasprop(

  struct pbsnode *pnode,
  struct prop    *props)

  {
  struct  prop    *need;

  for (need = props;need;need = need->next)
    {

    struct prop *pp;

    if (need->mark == 0) /* not marked, skip */
      continue;

    for (pp = pnode->nd_first;pp != NULL;pp = pp->next)
      {
      if (strcmp(pp->name, need->name) == 0)
        break;  /* found it */
      }

    if (pp == NULL)
      {
      return(0);
      }
    }

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





/*
 * see if node has the number of processors required
 * if free == SKIP_NONE,  check against total number of processors, else
 * if free != SKIP_NONE,  check against number free
 *
 * Return 1 if possible, 0 if not
 */

static int hasppn(

  struct pbsnode *pnode,     /* I */
  int             node_req,  /* I */
  int             free)      /* I */

  {
  if ((free != SKIP_NONE) &&
      (free != SKIP_NONE_REUSE) &&
      (pnode->nd_slots.get_number_free() >= node_req))
    {
    return(1);
    }

  if ((free == SKIP_NONE) && 
      (pnode->nd_slots.get_total_execution_slots() >= node_req))
    {
    return(1);
    }

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




/*
** Count how many gpus are available for use on this node
*/
static int gpu_count(

  struct pbsnode *pnode,  /* I */
  int    freeonly)        /* I */

  {
  int  count = 0;
  char log_buf[LOCAL_LOG_BUF_SIZE];

  if ((pnode->nd_state & INUSE_OFFLINE) ||
      (pnode->nd_state & INUSE_UNKNOWN) ||
      (pnode->nd_state & INUSE_DOWN))
    {
    if (LOGLEVEL >= 7)
      {
      sprintf(log_buf,
        "Counted %d gpus %s on node %s that was skipped",
        count,
        (freeonly? "free":"available"),
        pnode->nd_name);
    
      log_ext(-1, __func__, log_buf, LOG_DEBUG);
      }
    return (count);
    }

  if (pnode->nd_gpus_real)
    {
    int j;

    for (j = 0; j < pnode->nd_ngpus; j++)
      {
      struct gpusubn *gn = pnode->nd_gpusn + j;

      /* always ignore unavailable gpus */
      if (gn->state == gpu_unavailable)
        continue;

      if (!freeonly)
        {
        count++;
        }
      else if ((gn->state == gpu_unallocated) ||
               ((gn->state == gpu_shared) &&
                (gpu_mode_rqstd == gpu_normal)))
        {
        count++;;
        }
      }
    }
  else
    {
    /* virtual gpus */
    if (freeonly)
      {
      count = pnode->nd_ngpus_free;
      }
    else
      {
      count = pnode->nd_ngpus;
      }
    }

  if (LOGLEVEL >= 7)
    {
    sprintf(log_buf,
      "Counted %d gpus %s on node %s",
      count,
      (freeonly? "free":"available"),
      pnode->nd_name);

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

  return(count);
  }  /* END gpu_count() */





/*
** get gpu index for this gpuid
*/
int gpu_entry_by_id(

  struct pbsnode *pnode,  /* I */
  char   *gpuid,
  int    get_empty)

  {
  if (pnode->nd_gpus_real)
    {
    int j;

    for (j = 0; j < pnode->nd_ngpus; j++)
      {
      struct gpusubn *gn = pnode->nd_gpusn + j;

      if ((gn->gpuid != NULL) && (strcmp(gpuid, gn->gpuid) == 0))
        {
        return(j);
        }
      }
    }

  /*
   * we did not find the entry.  if get_empty is set then look for an empty
   * slot.  If none is found then try to add a new entry to nd_gpusn
   */

  if (get_empty)
    {
    int j;

    for (j = 0; j < pnode->nd_ngpus; j++)
      {
      struct gpusubn *gn = pnode->nd_gpusn + j;

      if (gn->gpuid == NULL)
        {
        return(j);
        }
      }

    create_a_gpusubnode(pnode);
    return (pnode->nd_ngpus - 1);    
    }

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




/*
** Parse a number in a spec.
** Return 0 if okay, 1 if no number exists, -1 on error
*/

static int number(

  char **ptr,
  int   *num)

  {
  char  holder[80];
  int   i = 0;
  char *str = *ptr;
  char  log_buf[LOCAL_LOG_BUF_SIZE];

  while (isdigit(*str) && (unsigned int)(i + 1) < sizeof holder)
    holder[i++] = *str++;

  if (i == 0)
    {
    return(1);
    }

  holder[i] = '\0';

  if ((i = atoi(holder)) <= 0)
    {
    sprintf(log_buf, "zero illegal");

    return(-1);
    }

  *ptr = str;

  *num = i;

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




/*
** Check string to see if it is a legal property name.
** If not, return 1.
** *prop set to static char array containing the properity,
** must be copied.
*/

static int property(

  char **ptr,
  char **prop)

  {
  char        *str = *ptr;
  char        *dest = *prop;
  int          i = 0;
  char         log_buf[LOCAL_LOG_BUF_SIZE];
  long         cray_enabled = FALSE;

  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);

  if (!isalpha(*str))
    {
    if ((cray_enabled == FALSE) ||
        (is_compute_node(str) == FALSE))
      {
      sprintf(log_buf, "first character of property (%s) not a letter", str);
      
      return(1);
      }
    }

  while (isalnum(*str) || *str == '-' || *str == '.' || *str == '=' || *str == '_')
    dest[i++] = *str++;

  dest[i] = '\0';

  /* skip over "/vp_number" */

  if (*str == '/')
    {
    do
      {
      str++;
      }
    while (isdigit(*str));
    }

  *ptr = str;

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





/*
** Create a property list from a string.
** Return 0 if all is well, 1 otherwise.
*/

int proplist(

  char        **str,
  struct prop **plist,
  int          *node_req,
  int          *gpu_req,
  int          *mic_req)

  {
  struct prop *pp;
  char         name_storage[80];
  char        *pname;
  char        *pequal;
  int          have_gpus = FALSE;
  char         log_buf[LOCAL_LOG_BUF_SIZE];

  *node_req = 1; /* default to 1 processor per node */

  pname  = name_storage;
  *pname = '\0';

  for (;;)
    {
    if (property(str, &pname))
      {
      return(1);
      }

    if (*pname == '\0')
      break;

    if ((pequal = strchr(pname, (int)'=')) != NULL)
      {
      /* special property */

      /* identify the special property and place its value */
      /* into node_req       */

      *pequal = '\0';

      if (strcmp(pname, "ppn") == 0)
        {
        pequal++;

        if ((number(&pequal, node_req) != 0) || (*pequal != '\0'))
          {
          return(1);
          }
        }
      else if (strcmp(pname, "mics") == 0)
        {
        pequal++;

        if ((number(&pequal, mic_req) != PBSE_NONE) ||
            (*pequal != '\0'))
          {
          return(1);
          }
        }
      else if (strcmp(pname, "gpus") == 0)
        {
        pequal++;

        if ((number(&pequal, gpu_req) != 0) || (*pequal != '\0'))
          {
          return(1);
          }

        have_gpus = TRUE;
        gpu_err_reset = FALSE; /* default to no */

        /* default value if no other gets specified */

        gpu_mode_rqstd = gpu_exclusive_thread;
        }
      else
        {
        return(1); /* not recognized - error */
        }
      }
    else if (have_gpus && (!strcasecmp(pname, "exclusive_thread")))
      {
      gpu_mode_rqstd = gpu_exclusive_thread;
      }
    else if (have_gpus && (!strcasecmp(pname, "exclusive")))
      {
      gpu_mode_rqstd = gpu_exclusive_thread;
      }
    else if (have_gpus && (!strcasecmp(pname, "exclusive_process")))
      {
      gpu_mode_rqstd = gpu_exclusive_process;
      }
    else if (have_gpus && (!strcasecmp(pname, "default")))
      {
      gpu_mode_rqstd = gpu_normal;
      }
    else if (have_gpus && (!strcasecmp(pname, "shared")))
      {
      gpu_mode_rqstd = gpu_normal;
      }
    else if (have_gpus && (!strcasecmp(pname, "reseterr")))
      {
      gpu_err_reset = TRUE;
      }
    else
      {
      pp = (struct prop *)calloc(1, sizeof(struct prop));

      pp->mark = 1;
      pp->name = strdup(pname);
      pp->next = *plist;

      *plist = pp;
      }

    if ((have_gpus) && (LOGLEVEL >= 7))
      {
      sprintf(log_buf,
        "proplist: set needed gpu mode to %d",
        gpu_mode_rqstd);

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

    if (**str != ':')
      break;

    (*str)++;
    }  /* END for(;;) */

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




/*
** Add the "global" spec to every sub-spec in "spec".
**      RETURNS:  allocated string buffer (must be freed externally)
*/

static char *mod_spec(

  char *spec,    /* I */
  char *global)  /* I */

  {
  char  *line;
  char  *cp;
  int    len;
  int    nsubspec;

  nsubspec = 1;

  for (cp = spec;*cp != '\0';cp++)
    {
    if (*cp == '+')
      {
      nsubspec++;
      }
    }

  len = strlen(global);

  line = (char *)calloc(1, nsubspec * (len + 1) + strlen(spec) + 1);

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

    return(NULL);
    }

  cp = line;

  while (*spec)
    {
    if (*spec == '+')
      {
      *cp++ = ':';

      strcpy(cp, global);

      cp += len;
      }

    *cp++ = *spec++;
    }

  *cp++ = ':';

  strcpy(cp, global);

  return(line);
  }  /* END mod_spec() */



/*
 * Test a procs specification.
 *
 * Return >0 - number of procs counted in the spec if it works,
 *         0 - if it cannot be satisfied now,
 *        -1 - if it can never be satisfied.
 *
 */
int procs_available(
    
  int proc_ct)

  {
  int             iter = -1;
  int             procs_avail = 0;
  struct pbsnode *pnode;

  if (proc_ct > svr_clnodes)
    {
    /* user requested more processors than are available on the system*/
    return(-1);
    }

  while ((pnode = next_host(&allnodes,&iter,NULL)) != NULL)
    {
    procs_avail += pnode->nd_slots.get_number_free();

    unlock_node(pnode, "procs_available", NULL, LOGLEVEL);
    }

  if (proc_ct > procs_avail)
    {
    return(0);
    }

  return(procs_avail);
  } /* END procs_available() */




int node_is_spec_acceptable(

  struct pbsnode   *pnode,
  single_spec_data *spec,
  char             *ProcBMStr,
  int              *eligible_nodes)

  {
  struct prop    *prop = spec->prop;

  int             ppn_req = spec->ppn;
  int             gpu_req = spec->gpu;
  int             mic_req = spec->mic;
  int             gpu_free;
  int             np_free;
  int             mic_free;

#ifdef GEOMETRY_REQUESTS
  if (IS_VALID_STR(ProcBMStr))
    {
    if (pnode->nd_state != INUSE_FREE)
      return(FALSE);

    if (node_satisfies_request(pnode, ProcBMStr) == FALSE)
      return(FALSE);
    }
#endif

  /* NYI: check if these are necessary */
  pnode->nd_flag = okay;

  if (pnode->nd_ntype != NTYPE_CLUSTER)
    return(FALSE);

  /* make sure that the node has properties */
  if (hasprop(pnode, prop) == FALSE)
    return(FALSE);

  if ((hasppn(pnode, ppn_req, SKIP_NONE) == FALSE) ||
      (gpu_count(pnode, FALSE) < gpu_req) ||
      (pnode->nd_nmics < mic_req))
    return(FALSE);

  (*eligible_nodes)++;

  if ((pnode->nd_state & (INUSE_OFFLINE | INUSE_DOWN | INUSE_RESERVE | INUSE_JOB)) != 0)
    return(FALSE);

  gpu_free = gpu_count(pnode, TRUE) - pnode->nd_ngpus_to_be_used;
  np_free  = pnode->nd_slots.get_number_free() - pnode->nd_np_to_be_used;
  mic_free = pnode->nd_nmics_free - pnode->nd_nmics_to_be_used;
  
  if ((ppn_req > np_free) ||
      (gpu_req > gpu_free) ||
      (mic_req > mic_free))
    return(FALSE);

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




int parse_req_data(
    
  complete_spec_data *all_reqs)

  {
  int               i;
  int               j = 0;
  long              cray_enabled = FALSE;

  single_spec_data *req;

  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);
  all_reqs->total_nodes = 0;

  for (i = 0; i < all_reqs->num_reqs; i++)
    {
    req = all_reqs->reqs + i;
    req->nodes = 1;
    req->gpu   = 0;
    req->ppn   = 1;
    req->prop  = NULL;

    if ((cray_enabled == FALSE) ||
        (is_compute_node(all_reqs->req_start[i]) == FALSE))
      {
      if ((j = number(&(all_reqs->req_start[i]), &(req->nodes))) == -1)
        return(j);
      }

    if (j == 0)
      {
      /* there was a number */
      if (*(all_reqs->req_start[i]) != '\0')
        {
        if (*(all_reqs->req_start[i]) == ':')
          all_reqs->req_start[i]++;
        
        if (proplist(&(all_reqs->req_start[i]), &(req->prop), &(req->ppn), &(req->gpu), &(req->mic)))
          return(-1);
        }
      }
    else
      {
      if (*(all_reqs->req_start[i]) != '\0')
        {
        if (proplist(&(all_reqs->req_start[i]), &(req->prop), &(req->ppn), &(req->gpu), &(req->mic)))
          return(-1);
        }
      }

    all_reqs->total_nodes += req->nodes;
    }

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




/* 
 * builds the node_job_add_info struct that will be used by set_nodes
 * instead of looping over different nodes.
 */

int save_node_for_adding(
    
  node_job_add_info *naji,
  struct pbsnode    *pnode,
  single_spec_data  *req,
  char              *first_node_name,
  int                is_external_node,
  int                req_rank)

  {
  node_job_add_info *to_add;
  node_job_add_info *old_next;
  node_job_add_info *cur_naji;
  bool               first = false;

  if ((first_node_name[0] != '\0') &&
      (!strcmp(first_node_name, pnode->nd_name)))
    {
    pnode->nd_order = 0;
    first = true;
    req_rank = 0;
    }
  else
    pnode->nd_order = 1;

  if (naji->node_name[0] == '\0')
    {
    /* first */
    snprintf(naji->node_name, sizeof(naji->node_name), "%s", pnode->nd_name);
    naji->ppn_needed = req->ppn;
    naji->gpu_needed = req->gpu;
    naji->mic_needed = req->mic;
    naji->is_external = is_external_node;
    naji->req_rank = req_rank;
    }
  else
    {
    /* second */
    if ((to_add = (node_job_add_info *)calloc(1, sizeof(node_job_add_info))) == NULL)
      {
      log_err(ENOMEM, __func__, "Cannot allocate memory!");

      return(ENOMEM);
      }
    
    if (first == true)
      {
      /* move the first element here and place me first */
      memcpy(to_add, naji, sizeof(node_job_add_info));

      snprintf(naji->node_name, sizeof(naji->node_name), "%s", pnode->nd_name);
      naji->ppn_needed = req->ppn;
      naji->gpu_needed = req->gpu;
      naji->mic_needed = req->mic;
      naji->is_external = is_external_node;
      naji->req_rank = req_rank;
      }
    else
      {
      /* initialize to_add */
      snprintf(to_add->node_name, sizeof(to_add->node_name), "%s", pnode->nd_name);
      to_add->ppn_needed = req->ppn;
      to_add->gpu_needed = req->gpu;
      to_add->mic_needed = req->mic;
      to_add->is_external = is_external_node;
      to_add->req_rank = req_rank;
      }

    /* fix pointers, NOTE: works even if old_next == NULL */
    cur_naji = naji;
    old_next = cur_naji->next;
    while(old_next != NULL)
      {
      if(to_add->req_rank <= old_next->req_rank)
        {
        cur_naji->next = to_add;
        to_add->next = old_next;
        to_add = NULL;
        break;
        }
      cur_naji = old_next;
      old_next = cur_naji->next;
      }
    if(to_add != NULL)
      {
      cur_naji->next = to_add;
      to_add->next = NULL;
      }
    }

  /* count off the number we have reserved */
  pnode->nd_np_to_be_used    += req->ppn;
  pnode->nd_ngpus_to_be_used += req->gpu;

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




/*
 * if there is a node being requested, the spec should look like
 * node_name[:ppn=X][+]...
 * otherwise it should look like:
 * <NUM_NODES>[:ppn=X][+]...
 *
 * If a specific node is being requested first, copy just the
 * name into first_node_name  
 */

void set_first_node_name(
    
  char *spec_param,      /* I */
  char *first_node_name) /* O */

  {
  int   i;
  int   len;

  if (isdigit(spec_param[0]) == TRUE)
    {
    first_node_name[0] = '\0';
    }
  else
    {
    len = strlen(spec_param);
    
    for (i = 0; i < len; i++)
      {
      /* a ':' means you've moved on to ppn and a + means its the next req */
      if ((spec_param[i] == ':') ||
          (spec_param[i] == '+') ||
          (spec_param[i] == '|'))
        break;
      else
        first_node_name[i] = spec_param[i];
      }
    
    /* make sure you NULL terminate */
    first_node_name[i] = '\0';
    }

  } /* END set_first_node_name() */




int is_reserved_property(

  char *prop)

  {
  if ((strncmp(prop, "ppn", strlen("ppn")) == 0) ||
      (strncmp(prop, "gpus", strlen("gpus") == 0)) ||
      (strncasecmp(prop, "exclusive_thread", strlen("exclusive_thread")) == 0) ||
      (strncasecmp(prop, "exclusive", strlen("exclusive")) == 0) ||
      (strncasecmp(prop, "exclusive_process", strlen("exclusive_process")) == 0) ||
      (strncasecmp(prop, "default", strlen("default")) == 0) ||
      (strncasecmp(prop, "shared", strlen("shared")) == 0) ||
      (strncasecmp(prop, "reseterr", strlen("reseterr")) == 0))
    return(TRUE);
  else
    return(FALSE);
  } /* END is_reserved_property() */




int is_compute_node(

  char *node_id)

  {
  struct pbsnode *pnode;
  int             rc = FALSE;
  char           *colon;
  char           *plus;

  if ((colon = strchr(node_id, ':')) != NULL)
    {
    if ((!strcmp(colon + 1, "external")) ||
        (!strcmp(colon + 1, alps_reporter_feature)) || 
        (!strcmp(colon + 1, alps_starter_feature)))
      {
      return(rc);
      }
    else
      *colon = '\0';
    }
  
  if ((plus = strchr(node_id, '+')) != NULL)
    *plus = '\0';

  if ((pnode = find_nodebyname(node_id)) != NULL)
    {
    rc = TRUE;
    unlock_node(pnode, __func__, NULL, LOGLEVEL);
    }

  if (colon != NULL)
    *colon = ':';

  if (plus != NULL)
    *plus = '+';

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




void release_node_allocation(
    
  node_job_add_info *naji)

  {
  node_job_add_info *current = NULL;
  struct pbsnode    *pnode = NULL;

  current = naji;
  while (current != NULL) 
    {
    if ((pnode = find_nodebyname(current->node_name)) != NULL)
      {
      pnode->nd_np_to_be_used    -= current->ppn_needed;
      pnode->nd_ngpus_to_be_used -= current->gpu_needed;
      pnode->nd_nmics_to_be_used -= current->mic_needed;
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      }
    current = current->next;
    }
  } /* END release_node_allocation() */




int check_for_node_type(

  complete_spec_data *all_reqs,
  enum node_types     nt)

  {
  single_spec_data *req;
  int               i;
  int               found_type = FALSE;
  struct pbsnode   *pnode;
  struct pbsnode   *reporter = alps_reporter;
  struct prop      *p;

  if (reporter == NULL)
    {
    /* this shouldn't be possible */
    log_err(-1, __func__, "Checking for node types with a non-cray enabled pbs_server??");
    return(-1);
    }


  for (i = 0; i < all_reqs->num_reqs; i++)
    {
    req = all_reqs->reqs + i;

    for (p = req->prop; p != NULL; p = p->next)
      {
      if ((!strcmp(p->name, "cray_compute")) ||
          (!strcmp(p->name, alps_starter_feature)))
        continue;

      lock_node(reporter, __func__, NULL, LOGLEVEL);
      pnode = find_node_in_allnodes(&(reporter->alps_subnodes), p->name);
      unlock_node(reporter, __func__, NULL, LOGLEVEL);

      if (pnode != NULL)
        {
        unlock_node(pnode, __func__, NULL, LOGLEVEL);

        if (nt == ND_TYPE_CRAY)
          {
          found_type = TRUE;
  
          break;
          }
        }
      else if (nt != ND_TYPE_CRAY)
        {
        int login = FALSE;

        pnode = find_nodebyname(p->name);

        if (pnode != NULL)
          {
          if (pnode->nd_is_alps_login == TRUE)
            login = TRUE;

          unlock_node(pnode, __func__, NULL, LOGLEVEL);

          if (nt == ND_TYPE_EXTERNAL)
            {
            if (login == FALSE)
              found_type = TRUE;
            }
          else if (nt == ND_TYPE_LOGIN)
            if (login == TRUE)
              found_type = TRUE;

          break;
          }
        }
      }

    if (found_type == TRUE)
      break;
    }

  return(found_type);
  } /* END check_for_node_type() */




enum job_types find_job_type(

  complete_spec_data *all_reqs)

  {
  enum job_types jt = JOB_TYPE_cray;
  
  if (check_for_node_type(all_reqs, ND_TYPE_CRAY) == TRUE)
    {
    if (check_for_node_type(all_reqs, ND_TYPE_EXTERNAL) == TRUE)
      jt = JOB_TYPE_heterogeneous;
    else
      jt = JOB_TYPE_cray;
    }
  else if (check_for_node_type(all_reqs, ND_TYPE_EXTERNAL) == TRUE)
    {
    jt = JOB_TYPE_normal;
    }
  else if (check_for_node_type(all_reqs, ND_TYPE_LOGIN) == TRUE)
    jt = JOB_TYPE_login;

  return(jt);
  } /* END find_job_type() */




int add_login_node_if_needed(

  char              **first_node_name_ptr,
  char               *login_prop,
  node_job_add_info  *naji)

  {
  char             *first_node_name = *first_node_name_ptr;
  struct pbsnode   *login = find_nodebyname(first_node_name);
  int               need_to_add_login = FALSE;
  int               rc = PBSE_NONE;
  int               dummy1;
  int               dummy2;
  int               dummy3;
  struct prop      *prop = NULL;
  single_spec_data  req;

  if (login == NULL)
    need_to_add_login = TRUE;
  else
    {
    if (login->nd_is_alps_login == FALSE)
      need_to_add_login = TRUE;

    unlock_node(login, __func__, NULL, LOGLEVEL);
    }

  if (need_to_add_login == TRUE)
    {
    if (login_prop != NULL)
      {
      if (proplist(&login_prop, &prop, &dummy1, &dummy2, &dummy3) != PBSE_NONE)
        {
        if (LOGLEVEL >= 3)
          {
          char log_buf[LOCAL_LOG_BUF_SIZE];
          snprintf(log_buf, sizeof(log_buf), "Malformed property list '%s', continuing.", login_prop);
          log_err(-1, __func__, log_buf);
          }
        }
      }

    if ((login = get_next_login_node(prop)) == NULL)
      rc = -1;
    else
      {
      if (naji != NULL)
        {
        /* add to naji */
        req.nodes = 1;
        req.ppn = 1;
        req.gpu = 0;
        req.mic = 0;
        req.prop = NULL;
        save_node_for_adding(naji, login, &req, login->nd_name, FALSE, -1);
        strcpy(*first_node_name_ptr, login->nd_name);
        }
      
      rc = PBSE_NONE;

      unlock_node(login, __func__, NULL, LOGLEVEL);
      }

    if (prop != NULL)
      free_prop(prop);
    }

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




int node_is_external(

  struct pbsnode *pnode)

  {
  int is_external = FALSE;

  /* all logins have nd_is_alps_login set to true.
   * all cray computes have their parent pointer set to alps_reporter.
   * if neither of these are found, it must be an external node */
  if ((pnode->nd_is_alps_login == FALSE) &&
      (pnode->parent == NULL))
    is_external = TRUE;
  
  return(is_external);
  } /* END node_is_external() */




/*
 * Test a node specification.
 *
 * Return >0 - number of nodes counted in the spec if it works,
 *         0 - if it cannot be satisfied,
 *        -1 - if it can never be satisfied.
 * Okay to bail early if "early" is true.
 * VPs selected are marked "thinking"
 */

int node_spec(

  char               *spec_param, /* I */
  int                 early,      /* I (boolean) */
  int                 exactmatch, /* I (boolean) - NOT USED */
  char               *ProcBMStr,  /* I */
  char               *FailNode,   /* O (optional,minsize=1024) */
  node_job_add_info  *naji,       /* O (optional) */
  char               *EMsg,       /* O (optional,minsize=1024) */
  char               *login_prop, /* I (optional) */
  alps_req_data     **ard_array,  /* O (optional) */
  int                *num_reqs,   /* O (optional) */
  enum job_types     &job_type)

  {
  FUNCTION_TIMER

  struct pbsnode      *pnode;
  char                 first_node_name[PBS_MAXHOSTNAME + 1];
  char                *first_name_ptr;
  node_iterator        iter;
  char                 log_buf[LOCAL_LOG_BUF_SIZE];

  char                *globs;
  char                *cp;
  char                *hold;
  int                  i;
  int                  num;
  int                  rc;
  int                  eligible_nodes = 0;
  complete_spec_data   all_reqs;
  char                *spec;
  char                *plus;
  long                 cray_enabled = FALSE;
  int                  num_alps_reqs = 0;

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

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

  if (LOGLEVEL >= 6)
    {
    sprintf(log_buf, "entered spec=%.4000s", spec_param);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

    DBPRT(("%s\n", log_buf));
    }
  
  job_type = JOB_TYPE_normal;

  set_first_node_name(spec_param, first_node_name);
  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);

  spec = strdup(spec_param);

  if (spec == NULL)
    {
    /* FAILURE */
    sprintf(log_buf, "cannot alloc memory");

    if (LOGLEVEL >= 1)
      {
      log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
      }

    if (EMsg != NULL)
      {
      snprintf(EMsg, 1024, "%s", log_buf);
      }

    return(-1);
    }

  if ((globs = strchr(spec, '#')) != NULL)
    {
    *globs++ = '\0';

    globs = strdup(globs);

    while ((cp = strrchr(globs, '#')) != NULL)
      {
      *cp++ = '\0';

      hold = mod_spec(spec, cp);
      
      free(spec);
      
      spec = hold;
      }

    hold = mod_spec(spec, globs);
    
    free(spec);
    
    spec = hold;

    free(globs);
    }  /* END if ((globs = strchr(spec,'#')) != NULL) */

  all_reqs.num_reqs = 1;
  plus = spec;

  /* count number of reqs */
  while (*plus != '\0')
    {
    if ((*plus == '+') ||
        (*plus == '|'))
      all_reqs.num_reqs++;

    plus++;
    }

  /* allocate space in all_reqs */
  all_reqs.reqs      = (single_spec_data *)calloc(all_reqs.num_reqs, sizeof(single_spec_data));
  all_reqs.req_start = (char **)calloc(all_reqs.num_reqs, sizeof(char *));

  if ((all_reqs.reqs == NULL) ||
      (all_reqs.req_start == NULL))
    {
    if (all_reqs.reqs != NULL)
      free(all_reqs.reqs);
    else if (all_reqs.req_start != NULL)
      free(all_reqs.req_start);

    log_err(ENOMEM, __func__, "Cannot allocate memory!");
    free(spec);
    return(-1);
    }

  /* set up pointers for reqs */
  plus = spec;
  i = 0;
  all_reqs.req_start[i] = spec;
  i++;

  while (*plus != '\0')
    {
    /* make the '+' NULL and advance past it */
    if (*plus == '|')
      num_alps_reqs++;

    if ((*plus == '|') ||
        (*plus == '+'))
      {
      all_reqs.reqs[i].req_id = num_alps_reqs;
      
      *plus = '\0';
      plus++;
      
      /* advance past "nodes=" */
      if (!strncmp(plus, "nodes=", strlen("nodes=")))
        plus += strlen("nodes=");
      
      all_reqs.req_start[i] = plus;
      i++;
      }
    else
      plus++;
    }

  /* now parse each spec into the data */
  if ((rc = parse_req_data(&all_reqs)) != PBSE_NONE)
    {
    /* FAILURE */
    for (i = 0; i < all_reqs.num_reqs; i++)
      free_prop(all_reqs.reqs[i].prop);
    
    free(all_reqs.reqs);
    free(all_reqs.req_start);

    free(spec);

    return(rc);
    }

  num = all_reqs.total_nodes;

#ifndef CRAY_MOAB_PASSTHRU
  /* If we restart pbs_server while the cray is down, pbs_server won't know about
   * the computes. Don't perform this check for this case. */
  if ((cray_enabled != TRUE) || 
      (alps_reporter == NULL) ||
      (alps_reporter->alps_subnodes.ra->num != 0))
    {
    if (num > svr_clnodes)
      {
      /* FAILURE */

      free(spec);

      sprintf(log_buf, "job allocation request exceeds available cluster nodes, %d requested, %d available",
        num,
        svr_clnodes);

      if (LOGLEVEL >= 6)
        {
        log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
        }

      if (EMsg != NULL)
        {
        snprintf(EMsg, 1024, "%s", log_buf);
        }

      return(-1);
      }
    }
#endif

  if (LOGLEVEL >= 6)
    {
    sprintf(log_buf, "job allocation debug: %d requested, %d svr_clnodes, %d svr_totnodes",
      num,
      svr_clnodes,
      svr_totnodes);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

    DBPRT(("%s\n", log_buf));
    }

  if (cray_enabled == TRUE)
    {
    job_type = find_job_type(&all_reqs);

    first_name_ptr = first_node_name;

    if ((job_type == JOB_TYPE_cray) ||
        (job_type == JOB_TYPE_heterogeneous))
      {
      /* naji == NULL indicates that this is a qsub not a run command,
       * we only need to assign the login when the job is run */
      if (naji != NULL)
        {
        if (add_login_node_if_needed(&first_name_ptr, login_prop, naji) != PBSE_NONE)
          {
          snprintf(log_buf, sizeof(log_buf), 
            "Couldn't find an acceptable login node for spec '%s' with feature request '%s'",
            spec_param,
            (login_prop != NULL) ? login_prop : "null");
          
          log_err(-1, __func__, log_buf);
          
          free(spec);
          
          return(PBSE_LOGIN_BUSY);
          }
        }
      }

    if ((num_alps_reqs > 0) &&
        (ard_array != NULL) &&
        (job_type == JOB_TYPE_cray))
      {
      *ard_array = (alps_req_data *)calloc(num_alps_reqs + 1, sizeof(alps_req_data));
      
      for (i = 0; i <= num_alps_reqs; i++)
        (*ard_array)[i].node_list = get_dynamic_string(-1, NULL);

      *num_reqs = num_alps_reqs + 1;
      }
    }

  reinitialize_node_iterator(&iter);
  pnode = NULL;

  /* iterate over all nodes */
  while ((pnode = next_node(&allnodes,pnode,&iter)) != NULL)
    {
    /* check each req against this node to see if it satisfies it */
    for (i = 0; i < all_reqs.num_reqs; i++)
      {
      single_spec_data *req = all_reqs.reqs + i;

      if (req->nodes > 0)
        {
        if (node_is_spec_acceptable(pnode, req, ProcBMStr, &eligible_nodes) == TRUE)
          {
          if (naji != NULL)
            {
            /* for heterogeneous jobs on the cray, record the external 
             * nodes in a separate attribute */
            if ((job_type == JOB_TYPE_heterogeneous) &&
                (node_is_external(pnode) == TRUE))
              save_node_for_adding(naji, pnode, req, first_node_name, TRUE, i+1);
            else
              save_node_for_adding(naji, pnode, req, first_node_name, FALSE, i+1);

            if ((num_alps_reqs > 0) &&
                (ard_array != NULL) &&
                (*ard_array != NULL))
              {
              if ((*ard_array)[req->req_id].node_list->used != 0)
                append_char_to_dynamic_string((*ard_array)[req->req_id].node_list, ',');

              append_dynamic_string((*ard_array)[req->req_id].node_list, pnode->nd_name);

              if (req->ppn > (*ard_array)[req->req_id].ppn)
                (*ard_array)[req->req_id].ppn = req->ppn;
              }
            }

          /* decrement needed nodes */
          all_reqs.total_nodes--;
          req->nodes--;
    
          /* are all reqs satisfied? */
          if (all_reqs.total_nodes == 0)
            break;
          }
        }
      }

    /* are all reqs satisfied? */
    if (all_reqs.total_nodes == 0)
      {
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      break;
      }
    } /* END for each node */

  for (i = 0; i < all_reqs.num_reqs; i++)
    if (all_reqs.reqs[i].prop != NULL)
      free_prop(all_reqs.reqs[i].prop);
  
  free(all_reqs.reqs);
  free(all_reqs.req_start);

  free(spec);

  /* If we restart pbs_server while the cray is down, pbs_server won't know about
   * the computes. Don't perform this check for this case. */
  if ((cray_enabled != TRUE) || 
      (alps_reporter == NULL) ||
      (alps_reporter->alps_subnodes.ra->num != 0))
    {
#ifndef CRAY_MOAB_PASSTHRU
    if (eligible_nodes < num)
      {
      if ((SvrNodeCt == 0) || (SvrNodeCt < num))
        {
        /* sufficient eligible nodes do not exist */
        /* FAILURE */
        sprintf(log_buf,
          "job requesting nodes that will never be available - spec = %s",
          spec_param);

        log_err(-1, __func__, log_buf);
        if (naji != NULL)
          release_node_allocation(naji);

        return(-1);
        }
      }
#endif
    }

  if (all_reqs.total_nodes > 0)
    {
    /* nodes not currently available */
    /* FAILURE */
    sprintf(log_buf,
      "job allocation request exceeds currently available cluster nodes, %d requested, %d available",
      num,
      num - all_reqs.total_nodes);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

    if (EMsg != NULL)
      {
      snprintf(EMsg, MAXLINE, "%s", log_buf);
      }

    if (naji != NULL)
      release_node_allocation(naji);

    return(0);
    } /* END if (all_reqs.total_nodes > 0) */

  /* SUCCESS - spec is ok */
  if (LOGLEVEL >= 6)
    {
    sprintf(log_buf, "job allocation debug(3): returning %d requested", num);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

    DBPRT(("%s\n", log_buf));
    }

  return(num);
  }  /* END node_spec() */




#ifdef GEOMETRY_REQUESTS
/**
 * get_bitmap
 *
 * @param pjob (I) - the job whose bitmap is be retrieved
 * @param ProcBMPtr (O) - the ptr to the string where the bitmap will be stored
 * @param ProcBMSize (I) - the size of the string ProcBMPtr points to
 * @return FAILURE if there is no specified bitmap or either pjob or ProcBMStrPtr are NULL
 * @return SUCCESS otherwise
 */
int get_bitmap(

  job  *pjob,        /* I */
  int   ProcBMSize,  /* I */
  char *ProcBMPtr)   /* O */

  {
  resource     *presc;
  resource_def *prd;

  char          LocalBM[MAX_BM];

  if ((pjob == NULL) ||
      (ProcBMPtr == NULL))
    {
    return(FAILURE);
    }

  LocalBM[0] = '\0';

  /* read the bitmap from the resource list */
  prd = find_resc_def(svr_resc_def,"procs_bitmap",svr_resc_size);
  presc = find_resc_entry(&pjob->ji_wattr[JOB_ATR_resource],prd);
  
  if ((presc != NULL) && 
      (presc->rs_value.at_flags & ATR_VFLAG_SET))
    {
    snprintf(LocalBM,sizeof(LocalBM),"%s",presc->rs_value.at_val.at_str);
    }
  else
    {
    /* fail if there was no bitmap given */

    return(FAILURE);
    }

  if (LocalBM[0] == '\0')
    {
    /* fail if there was no bitmap given */

    return(FAILURE);
    }
  else
    {
    snprintf(ProcBMPtr,sizeof(LocalBM),"%s",LocalBM);
    return(SUCCESS);
    }
  } /* end get_bitmap() */




/**
 * node_satisfies_request
 *
 * @param pnode (I) - the node to check for validity
 * @param ProcBMStr (I) - the bitmap of procs requested
 * @return TRUE - if the node satisfies the bitmap, FALSE otherwise
 * @return BM_ERROR if the bitmap isn't valid
 *
 * NOTE: must always be called by a thread already locking the pnode's mutex
 */
int node_satisfies_request(

  struct pbsnode *pnode,     /* I */
  char           *ProcBMStr) /* I */

  {
  int BMLen;
  int BMIndex;

  if (IS_VALID_STR(ProcBMStr) == FALSE)
    return(BM_ERROR);

  /* nodes are exclusive when we're using bitmaps */
  if (pnode->nd_state != INUSE_FREE)
    return(FALSE);

  BMLen = strlen(ProcBMStr);

  /* process in reverse because ProcBMStr[0] referes to core index 0 */
  BMIndex = BMLen-1;

  /* check if the requested processors are available on this node */
  for (int i = 0; i < pnode->nd_slots.get_total_execution_slots() && BMIndex >= 0; i++)
    {
    /* don't check cores that aren't requested */
    if (ProcBMStr[BMIndex--] != '1')
      continue;

    if (pnode->nd_slots.is_occupied(i) == true)
      return(FALSE);
    }

  if (BMIndex >= 0)
    {
    /* this means we didn't finish checking the string -
     * the node doesn't have enough processors */

    return(FALSE);
    }

  /* passed all checks, we're good */
  return(TRUE);
  } /* END node_satisfies_request() */




/**
 * reserve_node
 *
 * @param pnode - node to reserve
 * @param pjob - the job to be added to the node
 * @param hlistptr - a pointer to the host list 
 */

job_reservation_info *reserve_node(

  struct pbsnode                      *pnode,     /* I/O */
  job                                 *pjob,      /* I */
  char                                *ProcBMStr) /* I */

  {
  if ((pnode == NULL) ||
      (pjob == NULL)  ||
      (ProcBMStr == NULL))
    {
    return(NULL);
    }

  int                   BMIndex = strlen(ProcBMStr) - 1;
  job_reservation_info *node_info = (job_reservation_info *)calloc(1, sizeof(job_reservation_info));

  /* now reserve each node */
  for (int i = 0; i < pnode->nd_slots.get_total_execution_slots() && BMIndex >= 0; i++)
    {
    /* ignore unrequested cores */
    if (ProcBMStr[BMIndex--] != '1')
      continue;

    pnode->nd_slots.reserve_execution_slot(i, node_info->est);
    }

  if (BMIndex >= 0)
    {
    /* failure */
    free(node_info);
    return(NULL);
    }
    
  job_usage_info *jui = new job_usage_info(pjob->ji_qs.ji_jobid);
    
  jui->est = node_info->est;
  snprintf(node_info->node_name, sizeof(node_info->node_name), "%s", pnode->nd_name);
  node_info->port = pnode->nd_mom_rm_port;
  pnode->nd_job_usages.push_back(jui);
    
  /* mark the node as exclusive */
  pnode->nd_state = INUSE_JOB;

  return(node_info);
  }
#endif /* GEOMETRY_REQUESTS */




/**
 * adds this job to the node's list of jobs
 * checks to be sure not to add duplicates
 *
 * conditionally updates the subnode's state
 * decrements the amount of needed nodes
 *
 * @param pnode - the node that the job is running on
 * @param nd_psn - the subnode (processor) that the job is running on
 * @param newstate - the state nodes are transitioning to when used
 * @param pjob - the job that is going to be run
 */
int add_job_to_node(

  struct pbsnode *pnode,     /* I/O */
  struct pbssubn *snp,       /* I/O */
  short           newstate,  /* I */
  job            *pjob)      /* I */

  {
  struct jobinfo *jp;
  char            log_buf[LOCAL_LOG_BUF_SIZE];

  /* NOTE:  search existing job array.  add job only if job not already in place */
  if (LOGLEVEL >= 5)
    {
    sprintf(log_buf, "allocated node %s/%d to job %s (nsnfree=%d)",
      pnode->nd_name,
      snp->index,
      pjob->ji_qs.ji_jobid,
      pnode->nd_slots.get_number_free());

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    DBPRT(("%s\n", log_buf));
    }

  for (jp = snp->jobs;jp != NULL;jp = jp->next)
    {
    if (!(strcmp(jp->jobid, pjob->ji_qs.ji_jobid)))
      break;
    }

  if (jp == NULL)
    {
    /* add job to front of subnode job array */
    jp = (struct jobinfo *)calloc(1, sizeof(struct jobinfo));

    jp->next = snp->jobs;
    snp->jobs = jp;
    strcpy(jp->jobid, pjob->ji_qs.ji_jobid);

    /* if no free VPs, set node state */
    if ((pnode->nd_slots.get_number_free() <= 0) ||
        (pjob->ji_wattr[JOB_ATR_node_exclusive].at_val.at_long == TRUE))
      pnode->nd_state = newstate;

    if (snp->inuse == INUSE_FREE)
      {
      snp->inuse = newstate;
      }
    }

  /* decrement the amount of nodes needed */
  --pnode->nd_np_to_be_used;

  return(SUCCESS);
  } /* END add_job_to_node() */


    

int add_job_to_gpu_subnode(
    
  struct pbsnode *pnode,
  struct gpusubn *gn,
  job            *pjob)

  {
  if (!pnode->nd_gpus_real)
    {
    /* update the gpu subnode */
    strcpy(gn->jobid, pjob->ji_qs.ji_jobid);
    gn->inuse = TRUE;

    /* update the main node */
    pnode->nd_ngpus_free--;
    }

  gn->job_count++;
  pnode->nd_ngpus_to_be_used--;

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




int add_job_to_mic(

  struct pbsnode *pnode,
  int             index,
  job            *pjob)

  {
  int rc = -1;

  if (pnode->nd_micjobs[index].jobid[0] == '\0')
    {
    strcpy(pnode->nd_micjobs[index].jobid, pjob->ji_qs.ji_jobid);
    pnode->nd_nmics_free--;
    pnode->nd_nmics_to_be_used--;
    rc = PBSE_NONE;
    }

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




int remove_job_from_nodes_mics(

  struct pbsnode *pnode,
  job            *pjob)

  {
  short i;

  for (i = 0; i < pnode->nd_nmics; i++)
    {
    if (!strcmp(pnode->nd_micjobs[i].jobid, pjob->ji_qs.ji_jobid))
      pnode->nd_micjobs[i].jobid[0] = '\0';
    }

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




/**
 * builds the host list (hlist)
 *
 * @param pnode - the node being added to the host list
 * @param hlist - the host list being built
 */ 
int build_host_list(

  struct howl    **hlistptr,  /* O */
  struct pbssubn  *snp,       /* I */
  struct pbsnode  *pnode)     /* I */
  
  {
  struct howl *curr;
  struct howl *prev;
  struct howl *hp;

  /* initialize the pointers */
  curr = (struct howl *)calloc(1, sizeof(struct howl));
  curr->order = pnode->nd_order;
  curr->name  = pnode->nd_name;
  curr->index = snp->index;
  curr->port = pnode->nd_mom_rm_port;

  /* find the proper place in the list */
  for (prev = NULL, hp = *hlistptr;hp;prev = hp, hp = hp->next)
    {
    if (curr->order <= hp->order)
      break;
    }  /* END for (prev) */

  /* set the correct pointers in the list */
  curr->next = hp;

  if (prev == NULL)
    *hlistptr = curr;
  else
    prev->next = curr;

  return(SUCCESS);
  }



int add_gpu_to_hostlist(
    
  struct howl    **hlistptr,
  struct gpusubn  *gn,
  struct pbsnode  *pnode)

  {
  struct howl *curr;
  struct howl *prev;
  struct howl *hp;
  char        *gpu_name;
  static char *gpu = (char *)"gpu";

  /* create gpu_name */
  gpu_name = (char *)calloc(1, strlen(pnode->nd_name) + strlen(gpu) + 2);
  sprintf(gpu_name, "%s-%s", pnode->nd_name, gpu);


  /* initialize the pointers */
  curr = (struct howl *)calloc(1, sizeof(struct howl));
  curr->order = pnode->nd_order;
  curr->name  = gpu_name;
  curr->index = gn->index;
  curr->port = pnode->nd_mom_rm_port;

  /* find the proper place in the list */
  for (prev = NULL, hp = *hlistptr;hp;prev = hp, hp = hp->next)
    {
    if (curr->order <= hp->order)
      break;
    }  /* END for (prev) */

  /* set the correct pointers in the list */
  curr->next = hp;

  if (prev == NULL)
    *hlistptr = curr;
  else
    prev->next = curr;

  return(SUCCESS);
  } /* END add_gpu_to_hostlist() */



/*
 * checks the gpus of pnode and places them in gpu_list as necessary
 */

int place_gpus_in_hostlist(

  struct pbsnode     *pnode,
  job                *pjob,
  node_job_add_info  *naji,
  struct howl       **gpu_list)

  {
  int             j;
  struct gpusubn *gn;

  char            log_buf[LOCAL_LOG_BUF_SIZE];

  /* place the gpus in the hostlist as well */
  for (j = 0; j < pnode->nd_ngpus && naji->gpu_needed > 0; j++)
    {
    sprintf(log_buf,
      "node: %s j %d ngpus %d need %d",
      pnode->nd_name,
      j,
      pnode->nd_ngpus,
      pnode->nd_ngpus_needed);
    
    if (LOGLEVEL >= 7)
      {
      log_ext(-1, __func__, log_buf, LOG_DEBUG);
      }
    DBPRT(("%s\n", log_buf));
    
    gn = pnode->nd_gpusn + j;

    if (pnode->nd_gpus_real)
      {
      if ((gn->state == gpu_unavailable) ||
          (gn->state == gpu_exclusive) ||
          ((((int)gn->mode == gpu_normal)) &&
           (gpu_mode_rqstd != gpu_normal) &&
           (gn->state != gpu_unallocated)))
        continue;
      }
    else
      {
      if ((gn->state == gpu_unavailable) ||
          (gn->inuse == TRUE))
        continue;
      }

    if ((gn->state == gpu_unavailable) ||
        ((gn->state == gpu_exclusive) && pnode->nd_gpus_real) ||
        ((pnode->nd_gpus_real) &&
         ((int)gn->mode == gpu_normal) &&
         ((gpu_mode_rqstd != gpu_normal) && (gn->state != gpu_unallocated))) ||
        ((!pnode->nd_gpus_real) && 
         (gn->inuse == TRUE)))
      continue;
    
    add_job_to_gpu_subnode(pnode,gn,pjob);
    naji->gpu_needed--;
    
    sprintf(log_buf,
      "ADDING gpu %s/%d to exec_gpus still need %d",
      pnode->nd_name,
      j,
      pnode->nd_ngpus_needed);

    if (LOGLEVEL >= 7)
      {
      log_ext(-1, __func__, log_buf, LOG_DEBUG);
      }
    DBPRT(("%s\n", log_buf));
    
    add_gpu_to_hostlist(gpu_list,gn,pnode);
    
    /*
     * If this a real gpu in exclusive/single job mode, or a gpu in default
     * mode and the job requested an exclusive mode then we change state
     * to exclusive so we cannot assign another job to it
     */
    
    if ((pnode->nd_gpus_real) && 
        ((gn->mode == gpu_exclusive_thread) ||
         (gn->mode == gpu_exclusive_process) ||
         ((gn->mode == gpu_normal) && 
          ((gpu_mode_rqstd == gpu_exclusive_thread) ||
           (gpu_mode_rqstd == gpu_exclusive_process)))))
      {
      gn->state = gpu_exclusive;
      
      sprintf(log_buf,
        "Setting gpu %s/%d to state EXCLUSIVE for job %s",
        pnode->nd_name,
        j,
        pjob->ji_qs.ji_jobid);
      
      if (LOGLEVEL >= 7)
        {
        log_ext(-1, __func__, log_buf, LOG_DEBUG);
        }
      }
    
    /*
     * If this a real gpu in shared/default job mode and the state is
     * unallocated then we change state to shared so only other shared jobs
     * can use it
     */
    
    if ((pnode->nd_gpus_real) && (gn->mode == gpu_normal) && 
        (gpu_mode_rqstd == gpu_normal) && (gn->state == gpu_unallocated))
      {
      gn->state = gpu_shared;
      
      sprintf(log_buf,
        "Setting gpu %s/%d to state SHARED for job %s",
        pnode->nd_name,
        j,
        pjob->ji_qs.ji_jobid);
      
      if (LOGLEVEL >= 7)
        {
        log_ext(-1, __func__, log_buf, LOG_DEBUG);
        }
      }
    }

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




int add_mic_to_list(

  struct howl    **mic_list,
  struct pbsnode  *pnode,
  int              index)

  {
  struct howl *curr;
  struct howl *prev;
  struct howl *hp;
  char        *name;
  static char *mic = (char *)"mic";

  /* create gpu_name */
  name = (char *)calloc(1, strlen(pnode->nd_name) + strlen(mic) + 2);
  sprintf(name, "%s-%s", pnode->nd_name, mic);


  /* initialize the pointers */
  curr = (struct howl *)calloc(1, sizeof(struct howl));
  curr->order = pnode->nd_order;
  curr->name  = name;
  curr->index = index;
  curr->port = pnode->nd_mom_rm_port;

  /* find the proper place in the list */
  for (prev = NULL, hp = *mic_list; hp != NULL; prev = hp, hp = hp->next)
    {
    if (curr->order <= hp->order)
      break;
    }  /* END for (prev) */

  /* set the correct pointers in the list */
  curr->next = hp;

  if (prev == NULL)
    *mic_list = curr;
  else
    prev->next = curr;

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




int place_mics_in_hostlist(

  struct pbsnode    *pnode,
  job               *pjob,
  node_job_add_info *naji,
  struct howl      **mic_list)

  {
  int i;

  for (i = 0; i < pnode->nd_nmics && naji->mic_needed > 0; i++)
    {
    if (add_job_to_mic(pnode, i, pjob) == PBSE_NONE)
      {
      naji->mic_needed--;
      add_mic_to_list(mic_list, pnode, i);
      }
    }

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




/*
 * checks the subnodes of pnode and places them in the host list
 * as necessary
 */

job_reservation_info *place_subnodes_in_hostlist(

  job                *pjob,
  struct pbsnode     *pnode,
  node_job_add_info  *naji,
  char               *ProcBMStr)

  {
#ifdef GEOMETRY_REQUESTS
  if (IS_VALID_STR(ProcBMStr))
    {
    job_reservation_info *node_info = reserve_node(pnode, pjob, ProcBMStr);

    if (node_info != NULL)
      {
      // nodes are used exclusively for GEOMETRY_REQUESTS
      pnode->nd_np_to_be_used = 0;
      naji->ppn_needed = 0;
      }

    return(node_info);
    }

#endif
  job_reservation_info   *node_info = (job_reservation_info *)calloc(1, sizeof(job_reservation_info));

  if (pnode->nd_slots.reserve_execution_slots(naji->ppn_needed, node_info->est) == PBSE_NONE)
    {
    /* SUCCESS */
    pnode->nd_np_to_be_used -= naji->ppn_needed;
    naji->ppn_needed = 0;

    node_info->port = pnode->nd_mom_rm_port;
    
    job_usage_info *jui = new job_usage_info(pjob->ji_qs.ji_jobid);
    jui->est = node_info->est;
    
    snprintf(node_info->node_name, sizeof(node_info->node_name), "%s", pnode->nd_name);
    pnode->nd_job_usages.push_back(jui);

    bool job_exclusive_on_use = false;
    if ((server.sv_attr[SRV_ATR_JobExclusiveOnUse].at_flags & ATR_VFLAG_SET) &&
        (server.sv_attr[SRV_ATR_JobExclusiveOnUse].at_val.at_long != 0))
      job_exclusive_on_use = true;
    
    if ((pnode->nd_slots.get_number_free() <= 0) ||
        (pjob->ji_wattr[JOB_ATR_node_exclusive].at_val.at_long == TRUE) ||
        (job_exclusive_on_use))
      pnode->nd_state |= INUSE_JOB;
    }
  else
    {
    free(node_info);
    node_info = NULL;
    }

  return(node_info);
  } /* END place_subnodes_in_hostlist() */



/*
 * takes a struct howl and translates it to a string that will
 * become a job pbs_attribute (exec_hosts, exec_gpus, exec_ports)
 * NOTE: frees list (the struct howl)
 */

int translate_howl_to_string(

  struct howl  *list,
  char         *EMsg,
  int          *NCount,
  char        **str_ptr,
  char        **portstr_ptr,
  int           port)

  {
  struct howl *hp;
  struct howl *next;
  size_t       len = 1;
  int          count = 1;
  char        *str;
  char        *end;
  char        *portlist = NULL;
  char        *endport;

  for (hp = list;hp != NULL;hp = hp->next)
    {
    len += (strlen(hp->name) + 8);
    count++;
    }

  if ((str = (char *)calloc(1, len + 1)) == NULL)
    {
    log_err(ENOMEM, __func__, "Cannot allocate memory!");

    if (EMsg != NULL)
      sprintf(EMsg,"no nodes can be allocated to job");
    
    return(PBSE_RESCUNAV);
    }

  *str = '\0';

  if (port == TRUE)
    {
    /* port list will have a string of sister port addresses */
    if ((portlist = (char *)calloc(1, (count * PBS_MAXPORTNUM) + count)) == NULL)
      {
      log_err(ENOMEM, __func__, "Cannot allocate memory!");
      
      if (EMsg != NULL)
        sprintf(EMsg,"no nodes can be allocated to job");

      free(str);
      
      return(PBSE_RESCUNAV);
      }
  
    *portlist = '\0';
    }

  /* now copy in name+name+... */
  *NCount = 0;

  for (hp = list,end = str,endport = portlist; hp != NULL; hp = next)
    {
    (*NCount)++;

    sprintf(end, "%s/%d+",
      hp->name,
      hp->index);

    end += strlen(end);

    if (port == TRUE)
      {
      sprintf(endport, "%d+", hp->port);
      endport += strlen(endport);
      }

    next = hp->next;

    free(hp);
    }

  /* strip trailing '+' and assign pointers */
  str[strlen(str) - 1] = '\0';
  *str_ptr = str;

  if (port == TRUE)
    {
    portlist[strlen(portlist) - 1] = '\0';
    *portstr_ptr = portlist;
    }

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



int translate_job_reservation_info_to_string(
    
  std::vector<job_reservation_info *>  &host_info, 
  int                                  *NCount, 
  std::stringstream                    &exec_host_output,
  std::stringstream                    *exec_port_output)

  {
  bool              first = true;

  for (int hi_index = 0; hi_index < (int)host_info.size(); hi_index++)
    {
    job_reservation_info *jri = host_info[hi_index];
    int                   jri_index;
    int                   jri_iterator = -1;
    
    (*NCount)++;

    while ((jri_index = jri->est.get_next_occupied_index(jri_iterator)) != -1)
      {
      if (first == false)
        {
        exec_host_output << "+";

        if (exec_port_output != NULL)
          *exec_port_output << "+";
        }

      first = false;

      exec_host_output << jri->node_name << "/" << jri_index;

      if (exec_port_output != NULL)
        *exec_port_output << jri->port;
      }
    }

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



/*
 * free the struct that holds the information for where the job
 * will be placed  
 **/
void free_naji(
    
  node_job_add_info *naji)

  {
  node_job_add_info *current = NULL;
  node_job_add_info *tmp = NULL;
  node_job_add_info *first = naji;

  current = naji;
  while (current != NULL) 
    {
    tmp = current;
    current = current->next;
    free(tmp);
    if (current == first)
      break;
    }
  } /* END free_naji() */



/*
 * external nodes refers only to nodes outside of the cray
 * for jobs that also have cray compute nodes
 */

int record_external_node(

  job            *pjob,
  struct pbsnode *pnode)

  {
  char         *external_nodes;
  unsigned int  len;

  if (pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str == NULL)
    {
    pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str = strdup(pnode->nd_name);
    pjob->ji_wattr[JOB_ATR_external_nodes].at_flags |= ATR_VFLAG_SET;
    }
  else
    {
    len = strlen(pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str) + strlen(pnode->nd_name) + 2;
    external_nodes = (char *)calloc(1, len);

    snprintf(external_nodes, len, "%s+%s",
      pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str, pnode->nd_name);

    free(pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str);

    pjob->ji_wattr[JOB_ATR_external_nodes].at_val.at_str = external_nodes;
    }

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



/*
 * builds the hostlist based on the nodes=... part of the request
 */

int build_hostlist_nodes_req(

  job                                  *pjob,      /* M */
  char                                 *EMsg,      /* O */
  char                                 *spec,      /* I */
  short                                 newstate,  /* I */
  std::vector<job_reservation_info *>  &host_info, /* O */
  struct howl                         **gpu_list,  /* O */
  struct howl                         **mic_list,  /* O */ 
  node_job_add_info                    *naji,      /* I - freed */
  char                                 *ProcBMStr) /* I */

  {
  struct pbsnode    *pnode = NULL;

  node_job_add_info *current;
  char               log_buf[LOCAL_LOG_BUF_SIZE];
  bool               failure = false;

  current = naji;

  while (current != NULL)
    {
    if ((pnode = find_nodebyname(current->node_name)) != NULL)
      {
      if (failure == true)
        {
        /* just remove the marked request from the node */
        pnode->nd_np_to_be_used    -= current->ppn_needed;
        pnode->nd_ngpus_to_be_used -= current->gpu_needed;
        pnode->nd_nmics_to_be_used -= current->mic_needed;
        }
      else
        {
        job_reservation_info *host_single = place_subnodes_in_hostlist(pjob, pnode, current, ProcBMStr);

        if (host_single != NULL)
          {
          host_info.push_back(host_single);
          place_gpus_in_hostlist(pnode, pjob, current, gpu_list);
          place_mics_in_hostlist(pnode, pjob, current, mic_list);
        
          if (current->is_external == TRUE)
            {
            record_external_node(pjob, pnode);
            }
          }

        /* NOTE: continue through the loop if failure is true just to clean up amounts needed */

        if ((naji->gpu_needed > 0) || 
            (naji->ppn_needed > 0) ||
            (naji->mic_needed > 0))
          {
          failure = true;
       
          /* remove any remaining things marked on the node */
          pnode->nd_np_to_be_used    -= current->ppn_needed;
          pnode->nd_ngpus_to_be_used -= current->gpu_needed;
          pnode->nd_nmics_to_be_used -= current->mic_needed;
          }
        }

      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      }

    current = current->next;
    } /* END processing reserved nodes */
   
  free_naji(naji);

  if (failure == true)
    {
    /* did not satisfy the request */
    if (EMsg != NULL)
      {
      sprintf(log_buf,
        "could not locate requested gpu resources '%.4000s' (node_spec failed) %s",
        spec,
        EMsg);
      
      log_record(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);
      }

    return(PBSE_RESCUNAV);
    }

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




int build_hostlist_procs_req(

  job                                 *pjob,     /* M */
  int                                  procs,    /* I */
  short                                newstate, /* I */
  std::vector<job_reservation_info *> &host_info) /* O */

  {
  int             procs_needed;
  node_iterator   iter;
  struct pbsnode *pnode = NULL;

  /* did we have a request for procs? Do those now */
  if (procs > 0)
    {
    /* check to see if a -l nodes request was made */
    if (pjob->ji_have_nodes_request)
      {
      procs_needed = procs;
      }
    else
      {
      /* the qsub request used -l procs only. No -l nodes=x
         was given in the qsub request.
         TORQUE allocates 1 node by default if a -l nodes specification
         is not given.
      */
      if (procs > 1)
        {
        procs_needed = procs - 1;
        }
      else
        procs_needed = 1;
      }
  
    reinitialize_node_iterator(&iter);

    while ((pnode = next_node(&allnodes,pnode,&iter)) != NULL)
      {
      int execution_slots_free = pnode->nd_slots.get_number_free();

      if (execution_slots_free > 0)
        {
        job_reservation_info   *node_info = (job_reservation_info *)calloc(1, sizeof(job_reservation_info));
        if (pnode->nd_slots.reserve_execution_slots(execution_slots_free, node_info->est) == PBSE_NONE)
          {
          procs_needed -= execution_slots_free;

          host_info.push_back(node_info);
          node_info->port = pnode->nd_mom_rm_port;
          }
        else
          free(node_info);
        }
      } /* END for each node */
    } /* if (procs > 0) */

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




int add_to_ms_list(
    
  char *node_name,
  job  *pjob)

  {
  struct pbsnode *pnode = find_nodebyname(node_name);

  if (pnode != NULL)
    {
    insert_thing(pnode->nd_ms_jobs, strdup(pjob->ji_qs.ji_jobid));

    unlock_node(pnode, __func__, NULL, LOGLEVEL);
    }

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




void free_alps_req_data_array(
    
  alps_req_data *ard_array,
  int            num_reqs)

  {
  int i;

  for (i = 0; i < num_reqs; i++)
    free_dynamic_string(ard_array[i].node_list);

  free(ard_array);
  } /* END free_alps_req_data_array() */




int add_multi_reqs_to_job(
    
  job           *pjob,
  int            num_reqs,
  alps_req_data *ard_array)

  {
  int             i;
  dynamic_string *attr_str;
  char            buf[MAXLINE];

  if (ard_array == NULL)
    return(PBSE_NONE);

  attr_str = ard_array[0].node_list;

  for (i = 0; i < num_reqs; i++)
    {
    if (i != 0)
      {
      append_char_to_dynamic_string(attr_str, '|');
      append_dynamic_string(attr_str, ard_array[i].node_list->str);
      }

    snprintf(buf, sizeof(buf), "*%d", ard_array[i].ppn);
    append_dynamic_string(attr_str, buf);
    }

  if (pjob->ji_wattr[JOB_ATR_multi_req_alps].at_val.at_str != NULL)
    free(pjob->ji_wattr[JOB_ATR_multi_req_alps].at_val.at_str);

  pjob->ji_wattr[JOB_ATR_multi_req_alps].at_val.at_str = strdup(attr_str->str);
  pjob->ji_wattr[JOB_ATR_multi_req_alps].at_flags |= ATR_VFLAG_SET;

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



int free_hostinfo(

  std::vector<job_reservation_info *>  &host_info) /* O */

  {
  for (unsigned int i = 0; i < host_info.size(); i++)
    {
    job_reservation_info *host_single = host_info[i];

    if (host_single != NULL)
      free(host_single);
    }
  return(PBSE_NONE);
  }


/*
 * set_nodes() - Call node_spec() to allocate nodes then set them inuse.
 * Build list of allocated nodes to pass back in rtnlist.
 *      Return: PBS error code
 */

int set_nodes(

  job   *pjob,        /* I */
  char  *spec,        /* I */
  int    procs,       /* I */
  char **rtnlist,     /* O */
  char **rtnportlist, /* O */
  char  *FailHost,    /* O (optional,minsize=1024) */
  char  *EMsg)        /* O (optional,minsize=1024) */

  {
  FUNCTION_TIMER
  std::vector<job_reservation_info *> host_info;
  std::stringstream                   exec_hosts;
  std::stringstream                   exec_ports;
/*  job_reservation_info *gpu_info  = NULL;
  job_reservation_info *mic_info  = NULL; */
  struct howl       *gpu_list = NULL;
  struct howl       *mic_list = NULL;

  int                i;
  int                rc;
  int                NCount = 0;
  short              newstate;

  char              *login_prop = NULL;
  char              *gpu_str = NULL;
  char              *mic_str = NULL;
  char               ProcBMStr[MAX_BM];
  char               log_buf[LOCAL_LOG_BUF_SIZE];
  node_job_add_info *naji = NULL;
  alps_req_data     *ard_array = NULL;
  int                num_reqs = 0;
  long               cray_enabled = FALSE; 
  enum job_types     job_type;

  int gpu_flags = 0;

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

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

  if (LOGLEVEL >= 3)
    {
    sprintf(log_buf, "allocating nodes for job %s with node expression '%.4000s'",
      pjob->ji_qs.ji_jobid,
      spec);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  ProcBMStr[0] = '\0';
#ifdef GEOMETRY_REQUESTS
  get_bitmap(pjob,sizeof(ProcBMStr),ProcBMStr);
#endif /* GEOMETRY_REQUESTS */

  naji = (node_job_add_info *)calloc(1, sizeof(node_job_add_info));

  if (pjob->ji_wattr[JOB_ATR_login_prop].at_flags & ATR_VFLAG_SET)
    login_prop = pjob->ji_wattr[JOB_ATR_login_prop].at_val.at_str;

  /* allocate nodes */
  if ((i = node_spec(spec,
                     1,
                     1,
                     ProcBMStr,
                     FailHost,
                     naji,
                     EMsg,
                     login_prop,
                     &ard_array,
                     &num_reqs,
                     job_type)) == 0)
    {
    /* no resources located, request failed */
    if (EMsg != NULL)
      {
      sprintf(log_buf,
        "could not locate requested resources '%.4000s' (node_spec failed) %s",
        spec,
        EMsg);

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

    free_naji(naji);
    free_alps_req_data_array(ard_array, num_reqs);

    return(PBSE_RESCUNAV);
    }
  else if (i == PBSE_LOGIN_BUSY)
    {
    free_naji(naji);
    free_alps_req_data_array(ard_array, num_reqs);
    return(i);
    }
  else if (i < 0)
    {
    /* request failed, corrupt request */
    log_err(PBSE_UNKNODE, __func__, "request failed, corrupt request");
    free_naji(naji);
    free_alps_req_data_array(ard_array, num_reqs);
    return(PBSE_UNKNODE);
    }
 
  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);
  if (cray_enabled == TRUE)
    {
    // JOB_TYPE_normal means no component from the Cray will be used
    if ((job_type != JOB_TYPE_normal) && 
        (naji->next != NULL))
      {
      pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str = strdup(naji->node_name);
      pjob->ji_wattr[JOB_ATR_login_node_id].at_flags = ATR_VFLAG_SET;
      }
    }

  newstate = INUSE_JOB;

  if ((rc = build_hostlist_nodes_req(pjob,
                                     EMsg,
                                     spec,
                                     newstate,
                                     host_info,
                                     &gpu_list,
                                     &mic_list,
                                     naji,
                                     ProcBMStr)) != PBSE_NONE)
    {
    free_nodes(pjob);
    free_alps_req_data_array(ard_array, num_reqs);
    free_hostinfo(host_info);
    return(rc);
    }

  if ((rc = build_hostlist_procs_req(pjob, procs, newstate, host_info)) != PBSE_NONE)
    {
    free_nodes(pjob);
    free_alps_req_data_array(ard_array, num_reqs);
    free_hostinfo(host_info);
    return(rc);
    }

  if (host_info.empty() == true)
    {
    if (LOGLEVEL >= 1)
      {
      sprintf(log_buf, "no nodes can be allocated to job %s",
        pjob->ji_qs.ji_jobid);

      log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
      }

    if (EMsg != NULL)
      sprintf(EMsg, "no nodes can be allocated to job");
    
    free_alps_req_data_array(ard_array, num_reqs);

    free_hostinfo(host_info);
    return(PBSE_RESCUNAV);
    }  /* END if (host_info.size() == 0) */

  pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HasNodes;  /* indicate has nodes */

  /* build list of allocated nodes, gpus, and ports */
  rc = translate_job_reservation_info_to_string(host_info, &NCount, exec_hosts, &exec_ports);
  if (rc != PBSE_NONE)
    {
    free_alps_req_data_array(ard_array, num_reqs);
    free_hostinfo(host_info);
    return(rc);
    }

  *rtnlist = strdup(exec_hosts.str().c_str());
  *rtnportlist = strdup(exec_ports.str().c_str());

  // JOB_TYPE_normal means no component from the Cray will be used
  if ((cray_enabled == TRUE) &&
      (job_type != JOB_TYPE_normal))
    {
    char *plus = strchr(*rtnlist, '+');

    /* only do this if there's more than one host in the host list */
    if (plus != NULL)
      {
      char *to_free = *rtnlist;

      *plus = '\0';
      *rtnlist = strdup(plus + 1);
      free(to_free);
      }
    }

  if (mic_list != NULL)
    {
    if ((rc = translate_howl_to_string(mic_list, EMsg, &NCount, &mic_str, NULL, FALSE)) != PBSE_NONE)
      {
      free_hostinfo(host_info);
      return(rc);
      }

    job_attr_def[JOB_ATR_exec_mics].at_free(
      &pjob->ji_wattr[JOB_ATR_exec_mics]);

    job_attr_def[JOB_ATR_exec_mics].at_decode(
      &pjob->ji_wattr[JOB_ATR_exec_mics],
      NULL,
      NULL,
      mic_str,
      0);

    free(mic_str);
    }

  if (gpu_list != NULL)
    {
    if ((rc = translate_howl_to_string(gpu_list, EMsg, &NCount, &gpu_str, NULL, FALSE)) != PBSE_NONE)
      {
      free_alps_req_data_array(ard_array, num_reqs);
      free_hostinfo(host_info);
      return(rc);
      }

    job_attr_def[JOB_ATR_exec_gpus].at_free(
      &pjob->ji_wattr[JOB_ATR_exec_gpus]);
    
    job_attr_def[JOB_ATR_exec_gpus].at_decode(
      &pjob->ji_wattr[JOB_ATR_exec_gpus],
      NULL,
      NULL,
      gpu_str,
      0);  /* O */
    
    free(gpu_str);

    if (gpu_mode_rqstd != -1)
      gpu_flags = gpu_mode_rqstd;
    if (gpu_err_reset)
      gpu_flags += 1000;

    if (gpu_flags >= 0)
      {
      pjob->ji_wattr[JOB_ATR_gpu_flags].at_val.at_long = gpu_flags;
      pjob->ji_wattr[JOB_ATR_gpu_flags].at_flags = ATR_VFLAG_SET | ATR_VFLAG_MODIFY;
      
      if (LOGLEVEL >= 7)
        {
        sprintf(log_buf, "setting gpu_flags for job %s to %d %ld",
          pjob->ji_qs.ji_jobid,
          gpu_flags,
          pjob->ji_wattr[JOB_ATR_gpu_flags].at_val.at_long);

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

  if (LOGLEVEL >= 3)
    {
    snprintf(log_buf, sizeof(log_buf), "job %s allocated %d nodes (nodelist=%.4000s)",
      pjob->ji_qs.ji_jobid,
      NCount,
      *rtnlist);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  add_multi_reqs_to_job(pjob, num_reqs, ard_array);
  free_alps_req_data_array(ard_array, num_reqs);
  free_hostinfo(host_info);

  /* SUCCESS */

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




/* count the number of requested processors in a node spec
 * return processors requested on success
 * return -1 on error 
 */ 
int procs_requested(
    
  char *spec)

  {
  char        *str;
  char        *globs;
  char        *cp;
  char        *hold;
  int          num_nodes = 0;
  int          num_procs = 1;
  int          total_procs = 0;
  int          num_gpus = 0;
  int          num_mics = 0;
  int          i;
  struct prop *prop = NULL;
  char        *tmp_spec;
  char         log_buf[LOCAL_LOG_BUF_SIZE];

  tmp_spec = strdup(spec);  
  
  if (tmp_spec == NULL)
    {
    /* FAILURE */

    sprintf(log_buf,"cannot alloc memory");

    if (LOGLEVEL >= 1)
      {
      log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
      }

    return(-2);
    }

  /* Check to see if we have a global modifier */
  if ((globs = strchr(tmp_spec, '#')) != NULL)
    {
    *globs++ = '\0';

    globs = strdup(globs);

    while ((cp = strrchr(globs, '#')) != NULL)
      {
      *cp++ = '\0';

      hold = mod_spec(spec, cp);
      
      free(tmp_spec);
      
      tmp_spec = hold;
      }

    hold = mod_spec(tmp_spec, globs);
    
    free(tmp_spec);
    
    tmp_spec = hold;

    free(globs);
    }  /* END if ((globs = strchr(spec,'#')) != NULL) */

  str = tmp_spec;

  do
    {
    if ((i = number(&str, &num_nodes)) == -1)
      {
      free(tmp_spec);
      /* Bad string syntax. Fail */
      return(-1);
      }

    if (i == 0)
      {
      /* number exists */
      if (*str == ':')
        {
        /* there are properties */

        str++;

        if (proplist(&str, &prop, &num_procs, &num_gpus, &num_mics))
          {
          free(tmp_spec);
          if (prop != NULL)
            free_prop(prop);
          return(-1);
          }
        }
      }
    else
      {
      /* no number */
      num_nodes = 1;
      if (proplist(&str, &prop, &num_procs, &num_gpus, &num_mics))
        {
        /* must be a prop list with no number in front */
        free(tmp_spec);
        if (prop != NULL)
          free_prop(prop);

        return(-1);
        }
      }
    total_procs += num_procs * num_nodes;
    } while(*str++ == '+');
  
  free(tmp_spec);
  
  return(total_procs);
  } /* END procs_requested() */





/*
 * node_avail_complex -
 * *navail is set to number available
 * *nalloc is set to number allocated
 * *nresvd is set to number reserved
 * *ndown  is set to number down/offline
 *      return -1 on failure
 */

int node_avail_complex(

  char *spec,   /* I - node spec */
  int  *navail, /* O - number available */
  int  *nalloc, /* O - number allocated */
  int  *nresvd, /* O - number reserved  */
  int  *ndown)  /* O - number down      */

  {
  int            ret;
  enum job_types job_type;

  ret = node_spec(spec, 1, 0, NULL, NULL, NULL, NULL, NULL, NULL, NULL, job_type);

  *navail = ret;
  *nalloc = 0;
  *nresvd = 0;
  *ndown  = 0;

  return(ret);
  }  /* END node_avail_complex() */





/*
 * node_avail - report if nodes requested are available
 *
 * Return 0 when no error in request and
 *  *navail is set to number available
 *  *nalloc is set to number allocated
 *  *nresvd is set to number reserved
 *  *ndown  is set to number down/offline
 *      !=0 error number when error in request
 */

int node_avail(

  char *spec,  /* I  - node spec */
  int  *navail, /* O - number available */
  int *nalloc, /* O - number allocated */
  int *nresvd, /* O - number reserved  */
  int *ndown)  /* O - number down      */

  {
  int             j;
  int             holdnum;

  struct pbsnode *pn;
  char           *pc;

  struct prop    *prop = NULL;
  register int    xavail;
  register int    xalloc;
  register int    xresvd;
  register int    xdown;
  int             node_req = 1;
  int             gpu_req = 0;
  int             mic_req = 0;

  node_iterator   iter;

  if (spec == NULL)
    {
    log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, "no spec");

    return(RM_ERR_NOPARAM);
    }

  pc = spec;

  if ((strchr(spec, (int)'+') == NULL) && (number(&pc, &holdnum) == 1))
    {
    /* A simple node spec - reply with numbers of avaiable, */
    /* allocated, reserved, and down nodes that match the */
    /* the spec, null or simple number means all  */

    xavail = 0;
    xalloc = 0;
    xresvd = 0;
    xdown  = 0;

    /* find number of a specific type of node */

    if (*pc)
      {
      if (proplist(&pc, &prop, &node_req, &gpu_req, &mic_req))
        {
        return(RM_ERR_BADPARAM);
        }
      }

    reinitialize_node_iterator(&iter);
    pn = NULL;

    while ((pn = next_node(&allnodes, pn, &iter)) != NULL)
      {
      if ((pn->nd_ntype == NTYPE_CLUSTER) && hasprop(pn, prop))
        {
        if (pn->nd_state & (INUSE_OFFLINE | INUSE_DOWN))
          ++xdown;
        else if (hasppn(pn, node_req, SKIP_ANYINUSE))
          ++xavail;
        else if (hasppn(pn, node_req, SKIP_NONE))
          {
          /* node has enough processors, are they busy or reserved? */
          j = pn->nd_slots.get_total_execution_slots() - pn->nd_np_to_be_used;
          
          if (j >= node_req)
            ++xresvd;
          else
            ++xalloc;
          }
        }
      } /* END for each node */

    free_prop(prop);

    *navail = xavail;

    *nalloc = xalloc;

    *nresvd = xresvd;

    *ndown  = xdown;

    return(0);
    }
  else if (number(&pc, &holdnum) == -1)
    {
    /* invalid spec */

    return(RM_ERR_BADPARAM);
    }

  /* not a simple spec - determine if supplied complex */
  /* node spec can be satisified from avail nodes */
  /* navail set to >0 if can be satified now  */
  /*    0 if not now but possible  */
  /*   -l if never possible   */

  node_avail_complex(spec, navail, nalloc, nresvd, ndown);

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




/*
 * node_reserve - Reserve nodes
 *
 * Returns: >0 - reservation succeeded, number of nodes reserved
 *    0 - None or partial reservation
 *   -1 - requested reservation impossible
 */

int node_reserve(

  char       *nspec, /* In     - a node specification */
  resource_t  tag)   /* In/Out - tag for resource if reserved */

  {

  struct pbsnode    *pnode;
  int                ret_val;

  node_iterator      iter;
  char               log_buf[LOCAL_LOG_BUF_SIZE];
  node_job_add_info  *naji = NULL;
  enum job_types      job_type;

  DBPRT(("%s: entered\n", __func__))

  if ((nspec == NULL) || (*nspec == '\0'))
    {
    log_event(PBSEVENT_ADMIN, PBS_EVENTCLASS_SERVER, __func__, "no spec");

    return(-1);
    }

  naji = (node_job_add_info *)calloc(1, sizeof(node_job_add_info));

  if ((ret_val = node_spec(nspec, 0, 0, NULL, NULL, naji, NULL, NULL, NULL, NULL, job_type)) >= 0)
    {
    /*
    ** Zero or more of the needed Nodes are available to be
    ** reserved.
    */
    reinitialize_node_iterator(&iter);
    pnode = NULL;

    while ((pnode = next_node(&allnodes,pnode,&iter)) != NULL)
      {
      if (pnode->nd_flag != thinking)
        {
        continue;   /* skip this one */
        }


      if (pnode->nd_np_to_be_used == pnode->nd_slots.get_total_execution_slots())
        pnode->nd_state |= INUSE_RESERVE;
      } /* END for each node */
    }
  else
    {
    /* could never satisfy the reservation */

    snprintf(log_buf, sizeof(log_buf), "can never reserve %s", nspec);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  free_naji(naji);

  return(ret_val);
  }  /* END node_reserve() */




char *get_next_exec_host(

  char **current)

  {
  char *name_ptr = *current;
  char *plus;
  char *slash;
  
  if (name_ptr != NULL)
    {
    if ((plus = strchr(name_ptr, '+')) != NULL)
      {
      *current = plus + 1;
      *plus = '\0';
      }
    else
      *current = NULL;

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

  return(name_ptr);
  } /* END get_next_exec_host() */



int remove_job_from_nodes_gpus(

  struct pbsnode *pnode,
  job            *pjob)

  {
  struct gpusubn *gn;
  char           *gpu_str = NULL;
  int             i;
  char            log_buf[LOCAL_LOG_BUF_SIZE];
  std::string     tmp_str;
  char            num_str[6];
 
  if (pjob->ji_wattr[JOB_ATR_exec_gpus].at_flags & ATR_VFLAG_SET)
    gpu_str = pjob->ji_wattr[JOB_ATR_exec_gpus].at_val.at_str;

  if (gpu_str != NULL)
    {
    /* reset gpu nodes */
    for (i = 0; i < pnode->nd_ngpus; i++)
      {
      gn = pnode->nd_gpusn + i;
      
      if (pnode->nd_gpus_real)
        {
        /* reset real gpu nodes */
        tmp_str = pnode->nd_name;
        tmp_str += "-gpu/";
        sprintf (num_str, "%d", i);
        tmp_str += num_str;
        
        /* look thru the string and see if it has this host and gpuid.
         * exec_gpus string should be in format of 
         * <hostname>-gpu/<index>[+<hostname>-gpu/<index>...]
         *
         * if we are using the gpu node exclusively or if shared mode and
         * this is last job assigned to this gpu then set it's state
         * unallocated so its available for a new job. Takes time to get the
         * gpu status report from the moms.
         */
        
        if (strstr(gpu_str, tmp_str.c_str()) != NULL)
          {
          gn->job_count--;
          
          if ((gn->mode == gpu_exclusive_thread) ||
              (gn->mode == gpu_exclusive_process) ||
              ((gn->mode == gpu_normal) && 
               (gn->job_count == 0)))
            {
            gn->state = gpu_unallocated;
            
            if (LOGLEVEL >= 7)
              {
              sprintf(log_buf, "freeing node %s gpu %d for job %s",
                pnode->nd_name,
                i,
                pjob->ji_qs.ji_jobid);
              
              log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
              }
            
            }
          }
        }
      else
        {
        if (!strcmp(gn->jobid, pjob->ji_qs.ji_jobid))
          {
          gn->inuse = FALSE;
          memset(gn->jobid, 0, sizeof(gn->jobid));
          
          pnode->nd_ngpus_free++;
          }
        }
      }
    }

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




int remove_job_from_node(

  struct pbsnode *pnode,
  const char     *jobid)

  {
  char log_buf[LOCAL_LOG_BUF_SIZE];

  for (int i = 0; i < (int)pnode->nd_job_usages.size(); i++)
    {
    job_usage_info *jui = pnode->nd_job_usages[i];

    if (!strcmp(jui->jobid, jobid))
      {
      pnode->nd_slots.unreserve_execution_slots(jui->est);
      pnode->nd_job_usages.erase(pnode->nd_job_usages.begin() + i);

      if (LOGLEVEL >= 6)
        {
        sprintf(log_buf, "increased execution slot free count to %d of %d\n",
          pnode->nd_slots.get_number_free(),
          pnode->nd_slots.get_total_execution_slots());
        
        log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
        }

      pnode->nd_state &= ~INUSE_JOB;
      delete jui;
      i--; /* the array has shrunk by 1 so we need to reduce i by one */

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




/*
 * free_nodes - free nodes allocated to a job
 */

void free_nodes(

  job *pjob)  /* I (modified) */

  {
  struct pbsnode *pnode;

  char            log_buf[LOCAL_LOG_BUF_SIZE];
  char           *exec_hosts = NULL;
  char           *host_ptr = NULL;
  char           *hostname;
  char           *previous_hostname = NULL;

  if (LOGLEVEL >= 3)
    {
    sprintf(log_buf, "freeing nodes for job %s", pjob->ji_qs.ji_jobid);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
    }

  if (pjob->ji_wattr[JOB_ATR_exec_host].at_flags & ATR_VFLAG_SET)
    {
    if (pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str != NULL)
      {
      exec_hosts = strdup(pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str);
      host_ptr = exec_hosts;
      }
    }

  while ((hostname = get_next_exec_host(&host_ptr)) != NULL)
    {
    if ((previous_hostname) != NULL)
      {
      if (!strcmp(hostname, previous_hostname))
        continue;
      }

    previous_hostname = hostname;

    if ((pnode = find_nodebyname(hostname)) != NULL)
      {
      remove_job_from_node(pnode, pjob->ji_qs.ji_jobid);
      remove_job_from_nodes_gpus(pnode, pjob);
      remove_job_from_nodes_mics(pnode, pjob);
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      }
    }

  free(exec_hosts);

  if (pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str != NULL)
    {
    if ((pnode = find_nodebyname(pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str)) != NULL)
      {
      remove_job_from_node(pnode, pjob->ji_qs.ji_jobid);
      unlock_node(pnode, __func__, NULL, LOGLEVEL);
      }
    }

  pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_HasNodes;

  return;
  }  /* END free_nodes() */




struct pbsnode *get_compute_node(

  char *node_name)

  {
  struct pbsnode *ar = alps_reporter;
  struct pbsnode *compute_node = NULL;
  unsigned int    i;
  unsigned int    len = strlen(node_name);

  for (i = 0; i < len; i++)
    {
    if (isdigit(node_name[i]) == FALSE)
      {
      /* found a non-numeric character - not a compute node */
      return(NULL);
      }
    }

  lock_node(ar, __func__, NULL, LOGLEVEL);
  compute_node = create_alps_subnode(ar, node_name);
  unlock_node(ar, __func__, NULL, LOGLEVEL);

  return(compute_node);
  } /* END get_compute_node() */




/*
 * set_one_old - set a named node as allocated to a job
 */

void set_one_old(

  char *name,
  job  *pjob)

  {
  int             index;

  struct pbsnode *pnode;
  char           *pc;
  long            cray_enabled = FALSE;

  if ((pc = strchr(name, (int)'/')))
    {
    index = atoi(pc + 1);

    *pc = '\0';
    }
  else
    {
    index = 0;
    }

  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);

  pnode = find_nodebyname(name);

  if (cray_enabled == TRUE)
    {
    if (pnode == NULL)
      pnode = get_compute_node(name);

    if (pnode != NULL)
      {
      if (pnode->parent == alps_reporter)
        {
        while (index >= pnode->nd_slots.get_total_execution_slots())
          {
          add_execution_slot(pnode);
          }
        }
      }
    }

  if (pnode != NULL)
    {
    bool found = false;
    /* Mark node as being IN USE ...  */

    for (int i = 0; i < (int)pnode->nd_job_usages.size(); i++)
      {
      job_usage_info *jui = pnode->nd_job_usages[i];

      if (!strcmp(jui->jobid, pjob->ji_qs.ji_jobid))
        {
        found = true;

        while (index >= jui->est.get_total_execution_slots())
          jui->est.add_execution_slot();

        jui->est.mark_as_used(index);
        pnode->nd_slots.mark_as_used(index);
        }
      }

    if (found == false)
      {
      job_usage_info *jui = new job_usage_info(pjob->ji_qs.ji_jobid);
        
      while (index >= jui->est.get_total_execution_slots())
        jui->est.add_execution_slot();

      jui->est.mark_as_used(index);
      pnode->nd_slots.mark_as_used(index);
      pnode->nd_job_usages.push_back(jui);
      }

    if (pnode->nd_slots.get_number_free() <= 0)
      pnode->nd_state |= INUSE_JOB;

    unlock_node(pnode, __func__, NULL, LOGLEVEL);
    }

  return;
  }  /* END set_one_old() */





/*
 * set_old_nodes - set "old" nodes as in use - called from pbsd_init()
 * when recovering a job in the running state.
 */

void set_old_nodes(

  job *pjob)  /* I (modified) */

  {
  char     *old;
  char     *po;
  long      cray_enabled = FALSE;

  if (pjob->ji_wattr[JOB_ATR_exec_host].at_flags & ATR_VFLAG_SET)
    {
    old = strdup(pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str);

    if (old == NULL)
      {
      /* FAILURE - cannot alloc memory */

      return;
      }

    while ((po = strrchr(old, (int)'+')) != NULL)
      {
      *po++ = '\0';

      set_one_old(po, pjob);
      }

    set_one_old(old, pjob);

    free(old);
    } /* END if pjobs exec host is set */

  /* record the job on the alps_login if cray_enabled */
  get_svr_attr_l(SRV_ATR_CrayEnabled, &cray_enabled);
  if ((cray_enabled == TRUE) &&
      (pjob->ji_wattr[JOB_ATR_login_node_id].at_flags & ATR_VFLAG_SET))
    {
    set_one_old(pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str, pjob);
    }

  return;
  }  /* END set_old_nodes() */

    

job *get_job_from_job_usage_info(
    
  job_usage_info *jui,
  struct pbsnode *pnode)

  {
  job *pjob;

  unlock_node(pnode, __func__, NULL, LOGLEVEL);
  pjob = svr_find_job(jui->jobid, TRUE);
  lock_node(pnode, __func__, NULL, LOGLEVEL);

  return(pjob);
  }


  
job *get_job_from_jobinfo(
    
  struct jobinfo *jp,
  struct pbsnode *pnode)
  
  {
  job *pjob;

  unlock_node(pnode, __func__, NULL, LOGLEVEL);
  pjob = svr_find_job(jp->jobid, TRUE);
  lock_node(pnode, __func__, NULL, LOGLEVEL);

  return(pjob);
  } /* END get_job_from_jobinfo() */


/* END node_manager.c */

