/*
 * node_func.c - various functions dealing with nodes, properties and
 *   the following global variables:
 * pbsnlist     - the server's global node list
 * svr_totnodes - total number of pbshost entries
 * svr_clnodes  - number of cluster (space-shared) nodes
 *
 * Included functions are:
 * find_nodebyname() - find a node host with a given name
 */
#include <sstream>
#include <fstream>
#include "license_pbs.h" /* See here for the software license */
#include <pbs_config.h>   /* the master config generated by configure */
#include "node_func.h"

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

#include "pbs_ifl.h"
#include "libpbs.h"
#include "list_link.h"
#include "attribute.h"
#include "credential.h"
#include "batch_request.h"
#include "server_limits.h"
#include "server.h"
#include "pbs_job.h"
#include "pbs_nodes.h"
#include "pbs_error.h"
#include "log.h"
#include "dis.h"
#include "../lib/Liblog/pbs_log.h"
#include "../lib/Liblog/log_event.h"
#include "pbs_proto.h"
#include "net_connect.h"
#include "utils.h"
#include "u_tree.h"
#include "node_manager.h" /* is_compose */
#include "req_manager.h" /* mgr_set_node_attr */
#include "../lib/Libnet/lib_net.h" /* pbs_getaddrinfo */
#include "svrfunc.h" /* get_svr_attr_* */
#include "alps_constants.h"
#include "login_nodes.h"
#include "work_task.h"
#include "net_cache.h"
#include "ji_mutex.h"
#include "execution_slot_tracker.hpp"
#include "alps_functions.h"
#include "id_map.hpp"
#include <arpa/inet.h>
#include "threadpool.h"
#include "timer.hpp"
#include "mom_hierarchy_handler.h"
#include "runjob_help.hpp"
#include "policy_values.h"
#include "authorized_hosts.hpp"

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

#define NULLSTR static_cast <const char *>(0);
/* Global Data */

//extern hello_container  failures;
extern struct addrinfo  hints;
extern int              svr_totnodes;
extern int              svr_clnodes;
extern char            *path_nodes_new;
extern char            *path_nodes;
extern char            *path_nodestate;
extern char            *path_nodepowerstate;
extern char            *path_nodenote;
extern int              LOGLEVEL;
extern attribute_def    node_attr_def[];   /* node attributes defs */

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

/* Functions in this file
 * find_nodebyname()   -     given a node host name, search allnodes
 * find_subnodebyname() -     given a subnode name
 * save_characteristic() - save the the characteristics of the node along with
 *  the address of the node
 * chk_characteristic() -  check for changes to the node's set of
 *  characteristics and set appropriate flag bits in the "need_todo"
 *  location depending on which characteristics changed
 * status_nodeattrib() -    add status of each requested (or all) node-pbs_attribute
 *  to the status reply
 * effective_node_delete() -  effectively deletes a node from the server's node
 *  list by setting the node's "deleted" bit
 * setup_notification() -   sets mechanism for notifying other hosts about a new
 *  host
 * process_host_name_part() - processes hostname part of a batch request into a
 *  prop structure, host's IP addresses into an array, and node
 *  node type (cluster/time-shared) into an int variable
 * update_nodes_file() -    used to update the nodes file when certain changes
 *  occur to the server's internal nodes list
 * recompute_ntype_cnts -   Recomputes the current number of cluster nodes and
 *  current number of time-shared nodes
 * create_pbs_node - create basic node structure for adding a node
 */



struct pbsnode *alps_reporter;


/*
 * return 0 if addr is a MOM node and node is in bad state,
 * return 1 otherwise (it is not a MOM node, or it's state is OK)
 */

int addr_ok(

  pbs_net_t       addr,  /* I */
  struct pbsnode *pnode) /* I */

  {
  int           status = 1;  /* assume destination host is healthy */
  int           release_mutex = FALSE;
  time_t        time_now = time(NULL);

  /* if a node wasn't passed in, then find the node */
  if (pnode == NULL)
    {
    pnode = tfind_addr(addr, PBS_MOM_SERVICE_PORT, NULL);

    if (pnode == NULL)
      return(status);
    else
      release_mutex = TRUE;
    }

  if (pnode->nd_state & INUSE_UNKNOWN)
    {
    /* definitely not ok */
    status = 0;
    }
  else if (pnode->nd_state & INUSE_NOT_READY)
    {
    /* the node is ok if it is still talking to us */
    long chk_len = 300; 
    get_svr_attr_l(SRV_ATR_check_rate, &chk_len);
    
    if (pnode->nd_lastupdate != 0)
      {
      if (pnode->nd_lastupdate <= (time_now - chk_len))
        {
        status = 0;
        }
      }
    }

  if (pnode->nd_power_state != POWER_STATE_RUNNING)
    {
    status = 0;
    }

  if (release_mutex == TRUE)
    pnode->unlock_node(__func__, "release_mutex = TRUE", LOGLEVEL);

  return(status);
  }  /* END addr_ok() */



struct pbsnode *find_node_in_allnodes(

  all_nodes  *an,
  const char *nodename)

  {
  struct pbsnode *pnode = NULL;

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

  an->lock();
  pnode = an->find(nodename);
  if (pnode != NULL)
    pnode->lock_node(__func__, 0, LOGLEVEL);

  an->unlock();

  return(pnode);
  } /* END find_node_in_allnodes() */



bool node_exists(

  const char *node_name)

  {
  bool exists = false;

  allnodes.lock();
  if (allnodes.find(node_name) != NULL)
    exists = true;
  allnodes.unlock();

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



bool node_exists_by_id(

  int node_id)

  {
  const char     *name = node_mapper.get_name(node_id);
  return(node_exists(name));
  } // END node_exists_by_id()



struct pbsnode *find_nodebyid(

  int node_id)

  {
  struct pbsnode *pnode = NULL;
  const char     *name = node_mapper.get_name(node_id);

  pnode = find_nodebyname(name);

  return(pnode);
  } /* END find_nodebyid() */



/*
 * find_nodebyname() - find a node host by its name
 */

struct pbsnode *find_nodebyname(

  const char *nodename) /* I */

  {
  FUNCTION_TIMER
  char           *pslash;
  char           *dash = NULL;
  char           *tmp;

  struct pbsnode *pnode = NULL;
  struct pbsnode *numa  = NULL;

  int             numa_index;

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

  if ((pslash = strchr((char *)nodename, '/')) != NULL)
    *pslash = '\0';

  allnodes.lock();

  pnode = allnodes.find(nodename);

  if (pnode != NULL)
    {
    pnode->lock_node(__func__, NULL, LOGLEVEL);
    }
  else
    {
    if (cray_enabled == true)
      {
      if (alps_reporter != NULL)
        {
        alps_reporter->lock_node(__func__, NULL, LOGLEVEL);
        
        alps_reporter->alps_subnodes->lock();
        if ((pnode = (struct pbsnode *)alps_reporter->alps_subnodes->find(nodename)) != NULL)
          {
          pnode->lock_node(__func__, NULL, LOGLEVEL);
          }
        alps_reporter->alps_subnodes->unlock();

        alps_reporter->unlock_node(__func__, NULL, LOGLEVEL);
        }
      }
    else
      {
      /* check if it was a numa node */
      tmp = (char *)nodename;
      while ((tmp = strchr(tmp, '-')) != NULL)
        {
        dash = tmp;
        tmp++;
        }
      
      if (dash != NULL)
        {
        *dash = '\0';
        numa_index = atoi(dash + 1);
        
        if ((pnode = allnodes.find(nodename)) != NULL)
          {
          pnode->lock_node(__func__, NULL, LOGLEVEL);

          /* get the NUMA node */
          numa = AVL_find(numa_index, pnode->nd_mom_port, pnode->node_boards);
          if (numa != NULL)
            numa->lock_node(__func__, NULL, LOGLEVEL);

          pnode->unlock_node(__func__, NULL, LOGLEVEL);
          pnode = numa;
          }
        
        *dash = '-';
        }
      }
    }

  allnodes.unlock();

  if (pslash != NULL)
    *pslash = '/'; /* restore the slash */

  return(pnode);
  }  /* END find_nodebyname() */



/*
 * save_characteristic() -  save the characteristic values of the node along
 *       with the address of the node
 */

void save_characteristic(

  struct pbsnode  *pnode,
  node_check_info *nci)

  {
  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input pbsnode pointer");
    return;
    }
  if (nci == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input node_check_info pointer");
    return;
    }

  nci->state        = pnode->nd_state;
  nci->power_state  = pnode->nd_power_state;
  nci->ntype        = pnode->nd_ntype;
  nci->nprops       = pnode->nd_nprops;
  nci->nstatus      = pnode->nd_nstatus;
  strcpy((char *)nci->ttl,(char *)pnode->nd_ttl);
  nci->acl_size = (pnode->nd_acl == NULL)?0:pnode->nd_acl->as_usedptr;
  nci->rqid = pnode->nd_requestid;
  
  nci->note = pnode->nd_note;
  }  /* END save_characteristic() */





/*
 * chk_characteristic() -  check the value of the characteristics against
 *   that which was saved earlier.
 *   Returns:
 *   -1  if parent address doesn't match saved parent address
 *    0  if successful check.  *pneed_todo gets appropriate
 *       bit(s) set depending on the results of the check.
 *       The "returned" bits get used by the caller.
 */


int chk_characteristic(

  struct pbsnode  *pnode,      /* I */
  node_check_info *nci,        /* I */
  int             *pneed_todo) /* O */

  {
  char  tmpLine[1024];
  char  log_buf[LOCAL_LOG_BUF_SIZE+1];

  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input pbsnode pointer");
    return(PBSE_BAD_PARAMETER);
    }
  if (nci == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input node_check_info pointer");
    return(PBSE_BAD_PARAMETER);
    }
  if (pneed_todo == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input mask pointer");
    return(PBSE_BAD_PARAMETER);
    }

  tmpLine[0] = '\0';

  if (pnode->nd_state != nci->state)
    {
    if ((pnode->nd_state & INUSE_OFFLINE) && 
        !(nci->state & INUSE_OFFLINE))
      {
      *pneed_todo |= WRITENODE_STATE;  /*marked offline */

      strcat(tmpLine, "offline set");
      }
    else if (!(pnode->nd_state & INUSE_OFFLINE) &&
        (nci->state & INUSE_OFFLINE))
      {
      *pneed_todo |= WRITENODE_STATE;  /*removed offline*/

      strcat(tmpLine, "offline cleared");
      }
  
    if (tmpLine[0] != '\0')
      {
      if (LOGLEVEL >= 3)
        {
        snprintf(log_buf, LOCAL_LOG_BUF_SIZE, "node %s state modified (%s)\n",
          pnode->get_name(),
          tmpLine);
        
        log_event(PBSEVENT_ADMIN,PBS_EVENTCLASS_SERVER,"chk_characteristic",log_buf);
        }
      }
    }
  if(pnode->nd_power_state != nci->power_state)
    *pneed_todo |= WRITENODE_POWER_STATE;

  if (pnode->nd_ntype != nci->ntype)
    *pneed_todo |= WRITE_NEW_NODESFILE;

  if ((nci->nprops != pnode->nd_nprops) || 
      strcmp((char *)nci->ttl,(char *)pnode->nd_ttl) ||
      nci->acl_size != ((pnode->nd_acl == NULL)?0:pnode->nd_acl->as_usedptr) ||
      nci->rqid.compare(pnode->nd_requestid))
    *pneed_todo |= WRITE_NEW_NODESFILE;

  if (pnode->nd_note != nci->note)    /* not the same string */
    {
    *pneed_todo |= WRITENODE_NOTE;        /*node's note changed*/
    }

  nci->note.clear();

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




int login_encode_jobs(

  struct pbsnode *pnode,
  tlist_head     *phead)

  {
  job            *pjob;
  std::string     job_str = "";
  char            str_buf[MAXLINE*2];
  svrattrl       *pal;

  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input pbsnode pointer");
    return(PBSE_BAD_PARAMETER);
    }
  if (phead == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL input tlist_head pointer");
    return(PBSE_BAD_PARAMETER);
    }

  for (unsigned int i = 0; i < pnode->nd_job_usages.size(); i++)
    {
    // must be a copy and not a reference to avoid crashes: get_job_usage_info()
    // potentially releases the node mutex meaning a reference could refer tp bad
    // info.
    job_usage_info  jui = pnode->nd_job_usages[i];
    int             jui_index;
    int             jui_iterator = -1;
    int             login_id = -1;

    pjob = get_job_from_job_usage_info(&jui, pnode);
    
    if (pjob != NULL)
      {
      login_id = pjob->ji_wattr[JOB_ATR_login_node_key].at_val.at_long;
      unlock_ji_mutex(pjob, __func__, "1", LOGLEVEL);
      }

    const char *job_id = NULL;
    
    while ((jui_index = jui.est.get_next_occupied_index(jui_iterator)) != -1)
      {
      if (job_id == NULL)
        job_id = job_mapper.get_name(jui.internal_job_id);

      if (pnode->nd_id != login_id)
        {
        if (job_str.length() != 0)
          snprintf(str_buf, sizeof(str_buf), ",%d/%s", jui_index, job_id);
        else
          snprintf(str_buf, sizeof(str_buf), "%d/%s", jui_index, job_id);

        job_str += str_buf;
        }
      }
    }


  if ((pal = attrlist_create((char *)ATTR_NODE_jobs, (char *)NULL, strlen(job_str.c_str()) + 1)) == NULL)
    {
    log_err(ENOMEM, __func__, "");
    return(ENOMEM);
    }

  strcpy((char *)pal->al_value, job_str.c_str());
  pal->al_flags = ATR_VFLAG_SET;

  append_link(phead, &pal->al_link, pal);

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




/* status_nodeattrib() - add status of each requested (or all) node-pbs_attribute to
 *    the status reply
 *
 *      Returns:     0 is success
 *                != 0 is error, if a node-pbs_attribute is incorrectly specified, *bad is
 *   set to the node-pbs_attribute's ordinal position
 */

int status_nodeattrib(

  svrattrl        *pal,    /*a svrattrl from the request  */
  attribute_def   *padef,  /*the defined node attributes   */
  struct pbsnode  *pnode,  /*no longer an pbs_attribute ptr */
  int              limit,  /*number of array elts in padef */
  int              priv,   /*requester's privilege  */

  tlist_head       *phead, /*heads list of svrattrl structs that hang */
                           /*off the brp_attr member of the status sub*/
                           /*structure in the request's "reply area"  */

  int             *bad)    /*if node-pbs_attribute error, record its*/
                           /*list position here                 */

  {
  int   i;
  int   rc = 0;  /*return code, 0 == success*/
  int   index;
  int   nth;  /*tracks list position (ordinal tacker)   */

  pbs_attribute atemp[ND_ATR_LAST]; /*temporary array of attributes   */

  if ((padef == NULL) ||
      (pnode == NULL) ||
      (bad == NULL) ||
      (phead == NULL))
    {
    rc = PBSE_BAD_PARAMETER;
    return(rc);
    }

#ifdef PENABLE_LINUX_CGROUPS
  bool layout_valid = pnode->nd_layout.is_initialized();
#endif

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

  priv &= ATR_DFLAG_RDACC;    /* user-client privilege          */

  for (i = 0;i < ND_ATR_LAST;i++)
    {
    /*set up attributes using data from node*/
    if (i == ND_ATR_state)
      atemp[i].at_val.at_short = pnode->nd_state;
    else if (i == ND_ATR_power_state)
      atemp[i].at_val.at_short = pnode->nd_power_state;
    else if (i == ND_ATR_properties)
      {
      }
    else if (i == ND_ATR_status)
      atemp[i].at_val.at_str = NULL;
    else if (i == ND_ATR_ntype)
      atemp[i].at_val.at_short = pnode->nd_ntype;
    else if (i == ND_ATR_ttl)
      atemp[i].at_val.at_str = (char *)pnode->nd_ttl;
    else if (i == ND_ATR_acl)
      atemp[i].at_val.at_arst = pnode->nd_acl;
    else if (i == ND_ATR_requestid)
      atemp[i].at_val.at_str = (char *)pnode->nd_requestid.c_str();
    else if (i == ND_ATR_jobs)
      atemp[i].at_val.at_jinfo = pnode;
    else if (i == ND_ATR_np)
      atemp[i].at_val.at_long = pnode->nd_slots.get_total_execution_slots();
    else if (i == ND_ATR_note)
      atemp[i].at_val.at_str  = (char *)pnode->nd_note.c_str();
    else if (i == ND_ATR_mom_port)
      atemp[i].at_val.at_long  = pnode->nd_mom_port;
    else if (i == ND_ATR_mom_rm_port)
      atemp[i].at_val.at_long  = pnode->nd_mom_rm_port;
    /* skip NUMA attributes */
    else if (i == ND_ATR_num_node_boards)
      continue;
    else if (i == ND_ATR_numa_str)
      continue;
    else if (i == ND_ATR_gpus_str)
      continue;
    else if (i == ND_ATR_gpustatus)
      atemp[i].at_val.at_arst = pnode->nd_gpustatus;
#ifdef PENABLE_LINUX_CGROUPS
    else if (i == ND_ATR_total_sockets)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getTotalSockets();
      }
    else if (i == ND_ATR_total_numa_nodes)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getTotalChips();
      }
    else if (i == ND_ATR_total_cores)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getTotalCores();
      }
    else if (i == ND_ATR_total_threads)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getTotalThreads();
      }
    else if (i == ND_ATR_dedicated_sockets)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getDedicatedSockets();
      }
    else if (i == ND_ATR_dedicated_numa_nodes)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getDedicatedChips();
      }
    else if (i == ND_ATR_dedicated_cores)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getDedicatedCores();
      }
    else if (i == ND_ATR_dedicated_threads)
      {
      if (layout_valid == false)
        atemp[i].at_val.at_long = 0;
      else
        atemp[i].at_val.at_long = pnode->nd_layout.getDedicatedThreads();
      }
#endif
    else if (i == ND_ATR_gpus)
      {
      if (pnode->nd_ngpus == 0)
        continue;

      atemp[i].at_val.at_long  = pnode->nd_ngpus;
      }
    else if ((padef + i)->at_name != NULL)
      {
      if (!strcmp((padef + i)->at_name, ATTR_NODE_mics))
        {
        if (pnode->nd_nmics == 0)
          continue;

        atemp[i].at_val.at_long  = pnode->nd_nmics;
        }
      else if (!strcmp((padef + i)->at_name, ATTR_NODE_micstatus))
        atemp[i].at_val.at_arst = pnode->nd_micstatus;
      }
    else
      {
      /*we don't ever expect this*/
      *bad = 0;

      return(PBSE_UNKNODEATR);
      }

    atemp[i].at_flags = ATR_VFLAG_SET; /*artificially set the value's flags*/
    }

  if (pal != NULL)
    {
    /*caller has requested status on specific node-attributes*/
    nth = 0;

    while (pal != NULL)
      {
      ++nth;

      index = find_attr(padef, pal->al_name, limit);

      if (index < 0)
        {
        *bad = nth;  /*name in this position can't be found*/

        rc = PBSE_UNKNODEATR;

        break;
        }

      if ((padef + index)->at_flags & priv)
        {
        if ((index == ND_ATR_jobs) &&
            (pnode->nd_is_alps_login == TRUE))
          rc = login_encode_jobs(pnode, phead);
        else if (index == ND_ATR_properties)
          rc = pnode->encode_properties(phead);
        else
          {
          if (index == ND_ATR_status)
            atemp[index].at_val.at_str = strdup(pnode->nd_status.c_str());

          rc = ((padef + index)->at_encode(
                &atemp[index],
                phead,
                (padef + index)->at_name,
                NULL,
                ATR_ENCODE_CLIENT,
                0));

          free(atemp[index].at_val.at_str);
          }

        if (rc < 0)
          {
          rc = -rc;

          break;
          }
        else
          {
          /* encoding was successful */

          rc = 0;
          }
        }

      pal = (svrattrl *)GET_NEXT(pal->al_link);
      }  /* END while (pal != NULL) */
    }    /* END if (pal != NULL) */
  else
    {
    /* non-specific request, return all readable attributes */
    for (index = 0; index < limit; index++)
      {
      if ((index == ND_ATR_jobs) &&
          (pnode->nd_is_alps_login == TRUE))
        rc = login_encode_jobs(pnode, phead);
      else if (index == ND_ATR_properties)
        rc = pnode->encode_properties(phead);
      else if (((padef + index)->at_flags & priv) &&
               !((padef + index)->at_flags & ATR_DFLAG_NOSTAT))
        {
        if (index == ND_ATR_status)
          atemp[index].at_val.at_str = strdup(pnode->nd_status.c_str());

        rc = (padef + index)->at_encode(
               &atemp[index],
               phead,
               (padef + index)->at_name,
               NULL,
               ATR_ENCODE_CLIENT,
               0);
          
        if (index == ND_ATR_status)
          free(atemp[index].at_val.at_str);

        if (rc < 0)
          {
          rc = -rc;

          break;
          }
        else
          {
          /* encoding was successful */

          rc = 0;
          }
        }
      }    /* END for (index) */

    pnode->add_plugin_resources(phead);
    }      /* END else (pal != NULL) */

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



/*
 * effective_node_delete()
 *
 * Checks to make sure we're okay deleting the node and then calls the destructor
 */

void effective_node_delete(

  pbsnode **ppnode)

  {
  struct pbsnode* pnode = NULL;

  if (ppnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL node pointer to pointer delete call");
    return;
    }

  pnode = *ppnode;
  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER, __func__, "NULL node pointer delete call");
    return;
    }

  // If remove_node() fails, the node has been removed and someone else is 
  // deleting it
  if (remove_node(&allnodes, pnode) != PBSE_NONE)
    return;

  pnode->unlock_node(__func__, NULL, LOGLEVEL);

  //The node has been removed from the allnodes array.
  //Give some time for other threads to be done with the node
  //Before wiping it out.
  unsigned char tmp_unlock_count = 0;
  do
    {
    sleep(2);
    pnode->lock_node(__func__, NULL, LOGLEVEL);
    tmp_unlock_count = pnode->nd_tmp_unlock_count;
    pnode->unlock_node(__func__, NULL, LOGLEVEL);
    } while(tmp_unlock_count != 0);
    
  for (unsigned int i = 0; i < pnode->nd_addrs.size(); i++)
    {
    /* del node's IP addresses from tree  */
    auth_hosts.remove_address(pnode->nd_addrs[i], pnode->nd_mom_port);
    }

  delete pnode;

  *ppnode = NULL;

  return;
  }  /* END effective_node_delete() */





/**
 *  NOTE:  pul can return NULL even on SUCCESS of routine
 *
 */

static int process_host_name_part(

  char    *objname, /* node to be's name */
  u_long **pul,  /* 0 terminated host addrs array */
  char   **pname, /* node name w/o any :ts         */
  int     *ntype) /* node type; time-shared, not   */

  {
  char                log_buf[LOCAL_LOG_BUF_SIZE];

  struct addrinfo    *addr_info;
  struct addrinfo    *addr_iter;
  struct sockaddr_in *sai;

  struct in_addr      addr;
  char               *phostname;  /* caller supplied hostname   */
  int                 ipcount = 0;
  int                 len;
  int                 totalipcount;

  char                hname[MAXLINE];
  char                tmpHName[MAXLINE];
  char               *hptr;

  static int          NodeSuffixIsSet = 0;

  static char        *NodeSuffix;

  int                 hindex;
  int                 size = 0;
  int                 rc = PBSE_NONE;
  ulong              *tmp = NULL;

  len = (objname==NULL)?0:strlen(objname);

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

  if (pul == NULL)
    return(PBSE_BAD_PARAMETER);

  phostname = strdup(objname);

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

  *ntype = NTYPE_CLUSTER;

  *pul = NULL;

  if (overwrite_cache(phostname, &addr_info) == false)
    {
    snprintf(log_buf, sizeof(log_buf), "host %s not found", objname);

    log_err(PBSE_UNKNODE, __func__, log_buf);

    free(phostname);
    phostname = NULL;

    return(PBSE_UNKNODE);
    }

  if (LOGLEVEL >= 6)
    {
    char tmpLine[MAXLINE];

    snprintf(tmpLine, sizeof(tmpLine),
      "successfully loaded host structure for '%s'->'%s'",
      phostname,
      addr_info->ai_canonname);

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

  sai = (struct sockaddr_in *)addr_info->ai_addr;
  addr = sai->sin_addr;

  if (addr_info->ai_canonname == NULL)
    {
    free(phostname);
    
    return(PBSE_SYSTEM);
    }

  snprintf(hname, sizeof(hname), "%s", addr_info->ai_canonname);
  
  totalipcount = 0;
  
  if (NodeSuffixIsSet == 0)
    {
    char *node_suffix = NULL;
    get_svr_attr_str(SRV_ATR_NodeSuffix, &node_suffix);

    if (node_suffix != NULL)
      {
      NodeSuffix = strdup(node_suffix);
      }
    
    NodeSuffixIsSet = 1;
    }
  
  if (NodeSuffix != NULL)
    {
    char *ptr;
    
    /* NOTE:  extract outside of loop because hname will be freed */
    ptr = strchr(hname, '.');
    
    if (ptr != NULL)
      {
      *ptr = '\0';
      
      snprintf(tmpHName, sizeof(tmpHName), "%s%s.%s",
        hname,
        NodeSuffix,
        ptr + 1);
      
      *ptr = '.';
      }
    else
      {
      snprintf(tmpHName, sizeof(tmpHName), "%s%s", hname, NodeSuffix);
      }
    }
  
  for (hindex = 0;hindex < 2;hindex++)
    {
    if (hindex == 0)
      {
      hptr = hname;
      }
    else if (NodeSuffix != NULL)
      {
      hptr = tmpHName;
      }
    else
      {
      continue;
      }
    
    if ((rc = pbs_getaddrinfo(hptr, NULL, &addr_iter)) != 0)
      {
      snprintf(log_buf, sizeof(log_buf), "bad cname %s, h_errno=%d errno=%d (%s)",
        hptr,
        h_errno,
        errno,
        pbs_strerror(errno));
      
      log_err(PBSE_UNKNODE, __func__, log_buf);
      
      if (phostname != NULL)
        {
        free(phostname);
        phostname = NULL;
        }
      
      return(PBSE_UNKNODE);
      }
    
    /* count host ipaddrs */
    for (addr_iter = addr_info; addr_iter != NULL; addr_iter = addr_iter->ai_next)
      ipcount++;
    
    if (*pul == NULL)
      {
      size = sizeof(u_long) * (ipcount + 1);
      
      tmp = (u_long *)calloc(1, size);  /* zero-terminate list */
      }
    else
      {
      size += sizeof(u_long) * ipcount;
      
      tmp = (u_long *)realloc(*pul, size);
      }
    
    if (tmp == NULL)
      {
      if (phostname != NULL)
        {
        free(phostname);
        phostname = NULL;
        }

      return(PBSE_SYSTEM);
      }

    *pul = tmp;
    
    for (addr_iter = addr_info; addr_iter != NULL; addr_iter = addr_iter->ai_next)
      {
      u_long ipaddr;
      
      addr = ((struct sockaddr_in *)addr_iter->ai_addr)->sin_addr;
      
      ipaddr = ntohl(addr.s_addr);
      
      (*pul)[totalipcount++] = ipaddr;
      }
    
    (*pul)[totalipcount] = 0;  /* zero-term array ip addrs */
    }  /* END for (hindex) */
  
  *pname = phostname;   /* return node name     */

  return(PBSE_NONE);    /* function successful      */
  }  /* END process_host_name_part() */



/*
 * update_nodes_file - When called, this function will update
 *       the nodes file.  Specifically, it will
 *       walk the server's array of pbsnodes
 *       constructing for each entry a nodes file
 *       line if that entry is not marked as deleted.
 *       These are written to a temporary file.
 *       Upon successful conclusion that file replaces
 *       the nodes file.
*/

int update_nodes_file(
    
  struct pbsnode *held)

  {
  struct pbsnode     *np;
  all_nodes_iterator *iter = NULL;
  FILE               *nin;
  bool                dont_update_file = false;
    
  get_svr_attr_b(SRV_ATR_DontWriteNodesFile, &dont_update_file);
  if (dont_update_file == true)
    return(PBSE_NONE);

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

  if ((nin = fopen(path_nodes_new, "w")) == NULL)
    {
    log_event(
      PBSEVENT_ADMIN,
      PBS_EVENTCLASS_SERVER,
      "nodes",
      (char *)"Node description file update failed");

    return(-1);
    }

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

    fclose(nin);

    return(-1);
    }

  /* for each node ... */
  /* NOTE: DO NOT change this loop to iterate over numa nodes. Since they
   * aren't real hosts they should NOT appear in the nodes file */

  while ((np = next_host(&allnodes,&iter,held)) != NULL)
    {
    np->write_to_nodes_file(nin);

    fflush(nin);

    if (ferror(nin))
      {
      log_event(
        PBSEVENT_ADMIN,
        PBS_EVENTCLASS_SERVER,
        "nodes",
        (char *)"Node description file update failed");

      fclose(nin);
    
      if (held != np)
        np->unlock_node(__func__, "error", LOGLEVEL);

      delete iter;

      return(-1);
      }
    
    if (held != np)
      np->unlock_node(__func__, "loop", LOGLEVEL);
    } /* for each node */
      
  if (iter != NULL)
    delete iter;

  if ((fclose(nin)) != 0)
    {
    log_event(
      PBSEVENT_ADMIN,
      PBS_EVENTCLASS_SERVER,
      "nodes",
      (char *)"fclose of nodes file failed");

    return(-1);
    }

  if (rename(path_nodes_new, path_nodes) != 0)
    {
    log_event(
      PBSEVENT_ADMIN,
      PBS_EVENTCLASS_SERVER,
      "nodes",
      (char *)"replacing old nodes file failed");

    return(-1);
    }

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





/*
 * recompute_ntype_cnts - Recomputes the current number of cluster
 *          nodes and current number of time-shared nodes
 */
void recompute_ntype_cnts(void)

  {
  int              svr_loc_clnodes = 0;

  struct pbsnode  *pnode = NULL;

  node_iterator iter;

  reinitialize_node_iterator(&iter);

  if (svr_totnodes)
    {
    while ((pnode = next_node(&allnodes, pnode, &iter)) != NULL)
      {
      /* count normally */
      svr_loc_clnodes += pnode->nd_slots.get_total_execution_slots();
      }

    svr_clnodes = svr_loc_clnodes;

    if (iter.node_index != NULL)
      {
      delete iter.node_index;
      iter.node_index = NULL;
      }

    if (iter.alps_index != NULL)
      {
      delete iter.alps_index;
      iter.alps_index = NULL;
      }
    }
  } /* END recompute_ntype_cnts() */





/*
 * init_prop - allocate and initialize a prop struct
 *
 * pname points to the property string
 */

prop *init_prop(

  const char *pname) /* I */

  {
  return new prop(pname);
  }  /* END init_prop() */




/*
 * add_execution_slot - create a subnode entry and link to parent node
 *
 *  NOTE: pname arg must be a copy of prop list as it is linked directly in
 */

int add_execution_slot(
    
  struct pbsnode *pnode)

  {
  if (pnode == NULL)
    return(PBSE_RMBADPARAM);
  
  pnode->nd_slots.add_execution_slot();

  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_bool == true))
    job_exclusive_on_use = true;

  if (((pnode->nd_state & INUSE_JOB) != 0) &&
       (!job_exclusive_on_use))
    pnode->nd_state &= ~INUSE_JOB;

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




int create_a_gpusubnode(
    
  struct pbsnode *pnode)

  {
  int     rc = PBSE_NONE;

  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL pbsnode pointer input");
    return(rc);
    }
  
  gpusubn tmp(pnode->nd_gpusn.size());

  /* increment the number of gpu subnodes and gpus free */
  pnode->nd_gpusn.push_back(tmp);
  pnode->nd_ngpus++;
  pnode->nd_ngpus_free++;

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



/*
 * accepts a string of numbers separated by commas. it places the 
 * number in val and advances the string to the next number past the comma
 */

int read_val_and_advance(

  int         *val,
  const char **str)

  {
  const char *comma;

  if ((*str == NULL) ||
      (val == NULL))
    return(PBSE_BAD_PARAMETER);

  *val = atoi(*str);

  comma = strchr(*str,',');

  if (comma != NULL)
    *str += comma - *str + 1;

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





/* creates the private numa nodes on this node 
 *
 * @param pnode - the node that will house the numa nodes
 *
 * @return 0 on success, -1 on failure
 */
static int setup_node_boards(

  struct pbsnode *pnode,
  u_long         *pul)

  {
  int             i;
  int             j;
  struct pbsnode *pn;
  char            pname[MAX_LINE];
  const char     *np_ptr = NULL;
  const char     *gp_ptr = NULL;
  int             np = 0;
  int             gpus;
  int             rc = PBSE_NONE;

  char            log_buf[LOCAL_LOG_BUF_SIZE];

  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input pbsnode poiner");
    return(rc);
    }

  pnode->parent = NULL;

  /* if this isn't a numa node, return no error */
  if ((pnode->num_node_boards == 0) &&
      (pnode->numa_str.size() == 0))
    {
    return(PBSE_NONE);
    }

  /* determine the number of cores per node */
  if (pnode->numa_str.size() != 0)
    {
    np_ptr = pnode->numa_str.c_str();
    }
  else
    np = pnode->nd_slots.get_total_execution_slots() / pnode->num_node_boards;

  /* determine the number of gpus per node */
  if (pnode->gpu_str.size() == 0)
    {
    gp_ptr = pnode->gpu_str.c_str();
    read_val_and_advance(&gpus,&gp_ptr);
    }
  else
    gpus = pnode->nd_ngpus / pnode->num_node_boards;

  for (i = 0; i < pnode->num_node_boards; i++)
    {
    /* each numa node just has a number for a name */
    snprintf(pname,sizeof(pname),"%s-%d",
      pnode->get_name(),
      i);

    pn = new pbsnode(pname, pul, true);

    if ((rc = pn->get_error()) != PBSE_NONE)
      {
      delete pn;
      return(rc);
      }

    /* make sure the server communicates on the correct ports */
    pn->nd_mom_port = pnode->nd_mom_port;
    pn->nd_mom_rm_port = pnode->nd_mom_rm_port;
    memcpy(&pn->nd_sock_addr, &pnode->nd_sock_addr, sizeof(pn->nd_sock_addr));

    /* update the np string pointer */
    if (np_ptr != NULL)
      read_val_and_advance(&np,&np_ptr);

    /* create the subnodes for this node */
    for (j = 0; j < np; j++)
      add_execution_slot(pn);

    /* create the gpu subnodes for this node */
    for (j = 0; j < gpus; j++)
      {
      if (create_a_gpusubnode(pn) != PBSE_NONE)
        {
        /* ERROR */
        delete pn;
        return(PBSE_SYSTEM);
        }
      }

    /* update the gpu string pointer */
    if (gp_ptr != NULL)
      read_val_and_advance(&gpus,&gp_ptr);

    pnode->copy_properties(pn);

    /* add the node to the private tree */
    pnode->node_boards = AVL_insert(i,
        pn->nd_mom_port,
        pn,
        pnode->node_boards);

    /* set my parent node pointer */
    pn->parent = pnode;
    } /* END for each node_board */

  if (LOGLEVEL >= 3)
    {
    snprintf(log_buf,sizeof(log_buf),
      "Successfully created %d numa nodes for node %s\n",
      pnode->num_node_boards,
      pnode->get_name());

    log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_NODE, __func__, log_buf);
    }

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




/* recheck_for_node :
 * This function is called whenever an entry in the nodes file does
 * not resolve on server initialization. This function is called
 * periodically to see if the node is now resolvable and if so
 * add it to the list of available MOM nodes. */

static void recheck_for_node(
   
  struct work_task *ptask)

  {
  node_info *host_info;
  int        rc;
  int        bad;

  if ((host_info = (node_info *)ptask->wt_parm1) == NULL)
    {
    free(ptask->wt_mutex);
    free(ptask);

    return;
    }

  if ((rc = create_pbs_node( host_info->nodename, host_info->plist, host_info->perms, &bad)))
    {
    /* we created a new host_info in create_pbs_node. We
       need to free this one */
    free_attrlist(&host_info->atrlist);

    if (host_info->nodename)
      {
      free(host_info->nodename);
      }

    free(host_info);
    }
  else
    {
    //The node has been created so load reload the heirarchy.
    hierarchy_handler.reloadHierarchy();
    }

  free(ptask->wt_mutex);
  free(ptask);
  return;
  } /* END recheck_for_node() */



/* This is only called if we have resolved the node name and have a valid ip address for it. */

static int finalize_create_pbs_node(char     *pname, /* node name w/o any :ts       */
                                        u_long    *pul,  /* 0 terminated host adrs array*/
                                        int       ntype, /* node type; time-shared, not */
                                        svrattrl  *plist,
                                        int       perms,
                                        int       *bad)
  {
  int        rc = PBSE_NONE;
  pbsnode    *pnode = NULL;
  u_long     addr;
  char       log_buf[LOCAL_LOG_BUF_SIZE];

  if ((pnode = find_nodebyname(pname)) != NULL)
    {
    pnode->unlock_node(__func__, NULL, LOGLEVEL);

    free(pname);
    free(pul);

    return(PBSE_NODEEXIST);
    }

  pnode = new pbsnode(pname, pul, false);

  if ((rc = pnode->get_error()) != PBSE_NONE)
    {
    delete pnode;

    return(rc);
    }

  try
    {
    /* All nodes have at least one execution slot */
    add_execution_slot(pnode);

    rc = mgr_set_node_attr(
           pnode,
           node_attr_def,
           ND_ATR_LAST,
           plist,
           perms,
           bad,
           (void *)pnode,
           ATR_ACTION_ALTER,
           false);

    if (rc != 0)
      {
      effective_node_delete(&pnode);

      return(rc);
      }
    }
  catch(...)
    {
    free(pul);
    free(pname);
    delete pnode;
    return(-1);
    }

  int i;
  for (i = 0; pul[i]; i++)
    {
    if (LOGLEVEL >= 6)
      {
      snprintf(log_buf, LOCAL_LOG_BUF_SIZE,
          "node '%s' allows trust for ipaddr %ld.%ld.%ld.%ld\n",
        pnode->get_name(),
        (pul[i] & 0xff000000) >> 24,
        (pul[i] & 0x00ff0000) >> 16,
        (pul[i] & 0x0000ff00) >> 8,
        (pul[i] & 0x000000ff));

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

    addr = pul[i];
    auth_hosts.add_authorized_address(addr, pnode->nd_mom_port, pnode->get_name());
    }  /* END for (i) */

  if ((rc = setup_node_boards(pnode, pul)) != PBSE_NONE)
    {
    free(pul);
    return(rc);
    }

  free(pul);

  insert_node(&allnodes,pnode);

  svr_totnodes++;

  recompute_ntype_cnts();

  /* SUCCESS */
  return(PBSE_NONE);
  } /* End finalize_create_pbs_node() */

/*
 * create_pbs_node - create pbs node structure, i.e. add a node
 */

int create_pbs_node(

  char     *objname,
  svrattrl *plist,
  int       perms,
  int      *bad)

  {
  char             log_buf[LOCAL_LOG_BUF_SIZE];

  int              ntype; /* node type; time-shared, not */
  char            *pname; /* node name w/o any :ts       */
  u_long          *pul = NULL;  /* 0 terminated host adrs array*/
  int              rc;
  node_info        *host_info;
  time_t           time_now = time(NULL);

  if ((rc = process_host_name_part(objname, &pul, &pname, &ntype)) != 0)
    {
    svrattrl *pal, *pattrl;

    svr_unresolvednodes++;

    if(LOGLEVEL >= 7)
      {
      log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, "Node does not resolve, putting in delayed create.");
      }


    /* the host name in the nodes file did not resolve.
       We will set up a process to check periodically
       to see if the node will resolve later */
    host_info = (node_info *)calloc(1, sizeof(node_info));

    if (host_info == NULL)
      {
      log_err(-1, __func__, "create_pbs_node calloc failed");
      if (pul != NULL)
        free(pul);
      return(PBSE_MEM_MALLOC);
      }

    CLEAR_HEAD(host_info->atrlist);

    /* allocate and copy the objname plist and perms */
    host_info->perms = perms;
    pal = plist;

    while (pal != NULL)
      {
      pattrl = attrlist_create(pal->al_atopl.name, 0, strlen(pal->al_atopl.value) + 1);
      if (pattrl == NULL)
        {
        log_err(-1, __func__, "cannot create node attribute");
        free(host_info);
        if (pul != NULL)
          free(pul);
        return(PBSE_MEM_MALLOC);
        }

      strcpy((char *)pattrl->al_value, pal->al_atopl.value);
      pattrl->al_flags = SET;

      append_link(&host_info->atrlist, &pattrl->al_link, pattrl);
      pal = (svrattrl *)GET_NEXT(pal->al_link);
      }

    pattrl = (svrattrl *)GET_NEXT(host_info->atrlist);
    host_info->plist = pattrl;

    if (objname != NULL)
      {
      host_info->nodename = (char *)calloc(1, strlen(objname)+1);
      
      if (host_info->nodename == NULL)
        {
        free(host_info);
        if (pul != NULL)
          free(pul);
        log_err(-1, __func__, "create_pbs_node calloc failed");
        return(PBSE_MEM_MALLOC);
        }

      strcpy(host_info->nodename, objname);
      }

    if(LOGLEVEL >= 7)
      {
      log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, __func__, "Delayed create for node set up.");
      }

    set_task(WORK_Timed, time_now + 30, recheck_for_node, host_info, FALSE);

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

    return(rc);
    }

  if (pul == NULL)
    {
    free(pname);

    snprintf(log_buf, LOCAL_LOG_BUF_SIZE, 
      "no valid IP addresses found for '%s' - check name service",
      objname);

    log_err(-1, "process_host_name_part", log_buf);

    return(PBSE_SYSTEM);
    }

  return finalize_create_pbs_node(pname,pul,ntype,plist,perms,bad);
  } /* End create_pbs_node() */



#define COLON_OK  0x01
#define COMMA_OK  0x02
#define PLUS_OK   0x04
#define OPAQUE    0x08

/*
 * parse_node_token - parse tokens in the nodes file
 *
 * Token is returned, if null then there was none.
 * If there is an error, then "err" is set non-zero.
 * On following call, with argument "start" as null pointer, then
 * resume where left off.
 *
 * If "cok" is true, then this is first token ( node name) and ':' is
 * allowed and '=' is not.   For following tokens, allow '=' as separator
 * between "keyword" and "value".  Will get value as next token.
 *
 * @param start - begin looking here
 * @param flags - OR'ed together values of:
                  COLON_OK    The colon ':' is allowed
                  COMMA_OK    The comma ',' is allowed
                  PLUS_OK     The plus  '+' is allowed
                  OPAQUE      The string is accepted up to the next white space.
 * @param err (O) - non-zero if there was an error
 * @param term (O) - character terminating the token
 * @return the parsed token or NULL if none are found
 */

char *parse_node_token(

  char **start,
  int    flags,
  int   *err,
  char  *term)

  {
  char *pt = *start;
  char *token;

  *err = 0;

  pt = *start;

  while (*pt && isspace((int)*pt)) /* skip leading whitespace */
    pt++;

  if (*pt == '\0')
    {
    *start = pt;
    return(NULL);  /* no token */
    }

  token = pt;

  /* test for legal characters in token */

  for (;pt[0] != '\0';pt++)
    {
    if (flags & OPAQUE)
      {
      if ((isspace((int)*pt)) ||
          (*pt == '\0'))
        break;

      continue;
      }

    if (isalnum((int)*pt) || strchr("-._[]", *pt) || (*pt == '\0'))
      continue;

    if (isspace((int)*pt))
      break;

    if ((flags & COLON_OK) &&
        (*pt == ':'))
      continue;

    if ((flags & COMMA_OK) &&
        (*pt == ','))
      continue;

    if ((flags & PLUS_OK) &&
        (*pt == '+'))
      continue;

    if (!(flags&COLON_OK) &&
        (*pt == '='))
      break;

    *err = 1;
    }  /* END for() */

  *term = *pt;

  if (*pt != '\0')
    {
    *pt++ = '\0';
    }

  *start = pt;

  return(token);
  }  /* END parse_node_token() */


/*
 * create_pbs_dynamic_node - create pbs node structure, i.e. add a node
 */

int create_pbs_dynamic_node(

  char     *objname,
  svrattrl *plist,
  int       perms,
  int      *bad)

  {
  int     ntype; /* node type; time-shared, not */
  char   *pname = NULL; /* node name w/o any :ts       */
  u_long *pul = NULL;  /* 0 terminated host adrs array*/
  int     rc;
  int     err = 0;
  char    xchar;
  char   *ptr = objname;

  // Call parse_node_token to ensure that the node name doesn't have any 
  // invalid characters. Thi is the same function used when reading th
  // nodes file on startup.
  char *token = parse_node_token(&ptr, COLON_OK, &err, &xchar);

  if ((token == NULL) ||
      (err != 0 ))
    {
    return(PBSE_UNKNODE);
    }

  if ((rc = process_host_name_part(token, &pul, &pname, &ntype)) != 0)
    {
    if (pul != NULL)
      {
      free(pul);
      }

    if (pname != NULL)
      {
      free(pname);
      }

    return(rc);
    }

  return(finalize_create_pbs_node(pname,pul,ntype,plist,perms,bad));
  } // End create_pbs_dynamic_node()



/*
 * add_to_property_list()
 *
 * adds token to the list of properties
 * @pre-cond: token must be a valid string pointer
 * @post-cond: token will be appended to propstr, which is the list of properties.
 */
void add_to_property_list(

  std::string &propstr,
  const char  *token)

  {
  if (token != NULL)
    {
    if (propstr.size() != 0)
      propstr += ",";
    
    propstr += token;
    }
  } // END add_to_property_list()



/*
 * load_node_notes()
 *
 * Loads the notes for each node
 *
 */

void load_node_notes()

  {
  std::string     line;
  std::string     node_name;
  std::string     note;
  std::ifstream   note_file(path_nodenote);

  struct pbsnode *pnode;

  // Read the file. It is in the format:
  // <node name> <note>
  while (std::getline(note_file, line))
    {
    std::size_t space = line.find(" ");

    if (space != std::string::npos)
      {
      node_name = line.substr(0, space);
      note = line.substr(space + 1);

      pnode = find_nodebyname(node_name.c_str());

      if ((pnode == NULL) &&
          (cray_enabled == true) &&
          (isdigit(node_name.at(0))))
        {
        pnode = create_alps_subnode(alps_reporter, node_name.c_str());
        }

      if (pnode != NULL)
        {
        pnode->nd_note = note;
        pnode->unlock_node(__func__, "init - no note", LOGLEVEL);
        }
      }
    }

  note_file.close();
  } // END load_node_notes()



int add_node_attribute_to_list(
    
  char        *token,
  char       **line_ptr,
  tlist_head  *atrlist_ptr,
  int          linenum)

  {
  svrattrl *pal;
  int       err;
  char     *val;
  char      log_buf[LOCAL_LOG_BUF_SIZE];
  char      xchar;

  /* have new style pbs_attribute, keyword=value */
  if (!strcmp(token,"TTL"))
    {
    val = parse_node_token(line_ptr, COLON_OK|COMMA_OK|PLUS_OK, &err, &xchar);
    }
  else if (!strcmp(token,"acl"))
    {
    val = parse_node_token(line_ptr, OPAQUE, &err, &xchar);
    }
  else
    {
    val = parse_node_token(line_ptr, COMMA_OK, &err, &xchar);
    }

  if ((val == NULL) ||
      (err != 0) ||
      (xchar == '='))
    {
    snprintf(log_buf, sizeof(log_buf),
      "token \"%s\" in error on line %d of file nodes",
      token,
      linenum);
    log_err(-1, __func__, log_buf);

    return(-1);
    }

  pal = attrlist_create(token, 0, strlen(val) + 1);

  if (pal == NULL)
    {
    strcpy(log_buf, "cannot create node attribute");
    log_err(-1, __func__, log_buf);

    return(-1);
    }

  strcpy(pal->al_value, val);

  pal->al_flags = SET;

  append_link(atrlist_ptr, &pal->al_link, pal);

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



void add_node_property(

  std::string &propstr,
  const char  *token,
  bool         &is_alps_starter,
  bool         &is_alps_reporter,
  bool         &is_alps_compute)

  {
  /* old style properity */
  if (!strcmp(token, alps_starter_feature))
    is_alps_starter = true;

  if (!strcmp(token, alps_reporter_feature))
    {
    is_alps_reporter = true;

    add_to_property_list(propstr, "cray_compute");
    }
  else
    {
    if (!strcmp(token, "cray_compute"))
      is_alps_compute = true;

    add_to_property_list(propstr, token);
    }
  } // END add_node_property()



int record_node_property_list(
    
  std::string const &propstr,
  tlist_head        *atrlist_ptr)

  {
  svrattrl *pal;
  char      log_buf[LOCAL_LOG_BUF_SIZE];

  /* if any properties, create property attr and add to list */
  if (propstr.size() != 0)
    {
    pal = (svrattrl *)attrlist_create((char *)ATTR_NODE_properties, 0, propstr.size() + 1);
    
    if (pal == NULL)
      {
      strcpy(log_buf, "cannot create node attribute");
      
      log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);
      
      /* FAILURE */
      return(ENOMEM);
      }
    
    strcpy((char *)pal->al_value, propstr.c_str());
    
    pal->al_flags = SET;
    
    append_link(atrlist_ptr, &pal->al_link, pal);
    }

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


int create_node_range(

  char     *nodename,
  char     *open_bracket,
  svrattrl *pal)

  {
  int   num_digits;
  int   start = strtol(open_bracket+1, NULL, 10);
  int   err;
  int   perm = ATR_DFLAG_MGRD | ATR_DFLAG_MGWR;
  int   bad;

  char *dash = strchr(open_bracket,'-');
  char *close_bracket = strchr(open_bracket,']');
  char  tmp_node_name[MAX_LINE];
  char  log_buf[LOCAL_LOG_BUF_SIZE];

  if ((dash == NULL) ||
      (close_bracket == NULL))
    {
    snprintf(log_buf, sizeof(log_buf),
      "malformed nodename with range: %s, must be of form [x-y]\n",
      nodename);
    log_err(-1, __func__, log_buf);

    return(-1);
    }

  int end = strtol(dash+1, NULL, 10);

  /* nullify the open bracket */
  *open_bracket = '\0';

  num_digits = dash - open_bracket - 1;

  /* move past the closing bracket */
  close_bracket++;

  while (start <= end)
    {
    int num_len = 1;
    int tmp = 10;

    snprintf(tmp_node_name, sizeof(tmp_node_name), "%s", nodename);

    /* determine the length of the number */
    while (start / tmp > 0)
      {
      tmp *= 10;
      num_len++;
      }

    /* print extra zeros if needed */
    while (num_len < num_digits)
      {
      strcat(tmp_node_name,"0");

      num_len++;
      }

    sprintf(tmp_node_name+strlen(tmp_node_name),"%d%s",
      start,
      close_bracket);

    err = create_pbs_node(tmp_node_name,pal,perm,&bad);

    if (err != 0)
      break;

    start++;
    }

  return(err);
  } // END create_node_range()



void handle_cray_specific_node_values(
    
  char     *nodename,
  bool      is_alps_reporter,
  bool      is_alps_starter,
  bool      is_alps_compute,
  svrattrl *pal)

  {
  pbsnode *np;
  int      perm = ATR_DFLAG_MGRD | ATR_DFLAG_MGWR;
  char     log_buf[LOCAL_LOG_BUF_SIZE];

  if (cray_enabled == true)
    {
    if (is_alps_reporter == true)
      {
      if ((np = find_nodebyname(nodename)) != NULL)
        {
        np->nd_is_alps_reporter = TRUE;
        alps_reporter = np;
        np->alps_subnodes = new all_nodes();
        np->unlock_node(__func__, NULL, LOGLEVEL);
        }
      }
    else if (is_alps_starter == true)
      {
      if ((np = find_nodebyname(nodename)) != NULL)
        {
        np->nd_is_alps_login = TRUE;
        add_to_login_holder(np);
        np->unlock_node(__func__, NULL, LOGLEVEL);
        }
      }
    else if (is_alps_compute == true)
      {
      np = create_alps_subnode(alps_reporter, nodename);
      // add features
      int bad;
      if (mgr_set_node_attr(np, node_attr_def, ND_ATR_LAST, pal, 
                            perm, &bad, (void *)np, ATR_ACTION_ALTER, false) != PBSE_NONE)
        {
        snprintf(log_buf, sizeof(log_buf),
          "Node %s may not have all attributes initialized correctly", nodename);
        log_err(-1, __func__, log_buf);
        }

      np->unlock_node(__func__, NULL, LOGLEVEL);
      }
    }
  } // END handle_cray_specific_node_values()



/*
 * parse_node_name()
 *
 * Parses the node name from the line passed in
 *
 * @param ptr - the ptr to the current place in the line (I/O)
 * @param linenum - the index of the line we're on (I)
 * @return a pointer to the node name if an acceptable name is found, NULL otherwise
 */

char *parse_node_name(

  char **ptr,
  int   &err,
  int    linenum)

  {
  char  log_buf[LOCAL_LOG_BUF_SIZE];
  char  xchar;
  /* first token is the node name, may have ":ts" appended */
  char *nodename = parse_node_token(ptr, COLON_OK, &err, &xchar);

  if (nodename == NULL)
    {
    // blank line
    return(NULL);
    }

  if (err != 0)
    {
    snprintf(log_buf, LOCAL_LOG_BUF_SIZE,
      "invalid character in token \"%s\" on line %d", nodename, linenum);
    log_err(-1, __func__, log_buf);

    return(NULL);
    }

  // cray allows numeric node names
  if (cray_enabled == false)
    {
    if (!isalpha((int)*nodename))
      {
      snprintf(log_buf, LOCAL_LOG_BUF_SIZE,
        "token \"%s\" doesn't start with alpha on line %d", nodename, linenum);
      log_err(-1, __func__, log_buf);
      err = -1;
      
      return(NULL);
      }
    }

  return(nodename);
  } // END parse_node_name()



int parse_line_in_nodes_file(
    
  char *line,
  int   line_size,
  int   linenum)

  {
  if (line[0] == '#') /* comment */
    {
    memset(line, 0, line_size);
    return(PBSE_NONE);
    }

  bool         is_alps_reporter = false;
  bool         is_alps_starter = false;
  bool         is_alps_compute = false;
  std::string  propstr;
  char        *ptr = line;
  char        *token;
  char        *nodename;
  char         log_buf[LOCAL_LOG_BUF_SIZE];
  char         xchar;
  svrattrl    *pal;
  int          err;
  tlist_head   atrlist;
  int          perm = ATR_DFLAG_MGRD | ATR_DFLAG_MGWR;
  int          bad;

  CLEAR_HEAD(atrlist);

  nodename = parse_node_name(&ptr, err, linenum);

  if (err != PBSE_NONE)
    return(-1);
  else if (nodename == NULL)
    return(PBSE_NONE);

  /* now process remaining tokens (if any), they may be either */
  /* attributes (keyword=value) or old style properties        */
  while ((token = parse_node_token(&ptr, 0, &err, &xchar)) != NULL)
    {
    if (err != 0)
      {
      snprintf(log_buf, sizeof(log_buf),
        "token \"%s\" in error on line %d of file nodes",
        token,
        linenum);
      log_err(-1, __func__, log_buf);
      
      return(-1);
      }

    if (xchar == '=')
      {
      err = add_node_attribute_to_list(token, &ptr, &atrlist, linenum);

      if (err != PBSE_NONE)
        return(err);
      }
    else
      add_node_property(propstr, token, is_alps_starter, is_alps_reporter, is_alps_compute);
    }    /* END while(1) */

  record_node_property_list(propstr, &atrlist);

  /* now create node and subnodes */
  pal = (svrattrl *)GET_NEXT(atrlist);

  err = PBSE_NONE;

  char *open_bracket = strchr(nodename, '[');

  if (open_bracket != NULL)
    {
    err = create_node_range(nodename, open_bracket, pal);
    }
  else if (is_alps_compute == false)
    {
    err = create_pbs_node(nodename, pal, perm, &bad);
    }

  if (err == PBSE_NODEEXIST)
    {
    snprintf(log_buf, sizeof(log_buf), "duplicate node \"%s\"on line %d",
      nodename,
      linenum);
    log_err(-1, __func__, log_buf);

    return(-1);
    }
  else if (err != PBSE_NONE)
    {
    snprintf(log_buf, LOCAL_LOG_BUF_SIZE,
      "could not create node \"%s\", error = %d",
      nodename,
      err);

    log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

    free_attrlist(&atrlist);
    memset(line, 0, line_size);

    return(PBSE_NONE);
    }

  handle_cray_specific_node_values(nodename, is_alps_reporter, is_alps_starter, is_alps_compute, pal);

  if (LOGLEVEL >= 3)
    {
    snprintf(log_buf, sizeof(log_buf), "node '%s' successfully loaded from nodes file", nodename);

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

  free_attrlist(&atrlist);

  memset(line, 0, line_size);
  
  return(PBSE_NONE);
  } // END parse_line_in_nodes_file()



/*
 * parse_nodes_file()
 *
 * Opens and parses the nodes file, creating all relevant nodes
 *
 * @return - PBSE_NONE if the file is successfully parsed and no fatal error occurs,
 *           or non-zero on a fatal failure
 */

int parse_nodes_file()

  {
  FILE              *nin;
  char               log_buf[LOCAL_LOG_BUF_SIZE];
  extern char        server_name[];
  char               line[MAXLINE << 4];

  extern resource_t  next_resource_tag;
 
  if ((nin = fopen(path_nodes, "r")) == NULL)
    {
    snprintf(log_buf, LOCAL_LOG_BUF_SIZE,
      "cannot open node description file '%s' in setup_nodes()\n",
      path_nodes);

    log_event(PBSEVENT_ADMIN,PBS_EVENTCLASS_SERVER,server_name,log_buf);

    return(0);
    }

  next_resource_tag = time(0); /* initialize next resource handle */

  svr_totnodes = 0;

  /* clear out line so we don't have residual data if there is no LF */
  memset(line, 0, sizeof(line));

  for (int linenum = 1; fgets(line, sizeof(line) - 1, nin); linenum++)
    {
    int err = parse_line_in_nodes_file(line, sizeof(line), linenum);

    if (err != PBSE_NONE)
      {
      fclose(nin);
      return(err);
      }
    }  /* END for (linenum) */

  if (cray_enabled == true)
    {
    if (login_node_count() == 0)
      {
      snprintf(log_buf, sizeof(log_buf),
        "pbs_server is Cray enabled but no login nodes are configured. Jobs cannot run. Exiting");
      log_err(-1, __func__, log_buf);

      fclose(nin);
      return(-1);
      }
    }

  fclose(nin);

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



/*
 * Read the file, "nodes", containing the list of properties for each node.
 * The list of nodes is formed and stored in allnodes.
 * Return -1 on error, 0 otherwise.
 *
 * Read the node state file, "node_state", for any "offline"
 * conditions which should be set in the nodes.
*/

int setup_nodes(void)

  {
  FILE              *nin;
  std::string        propstr;
  char               log_buf[LOCAL_LOG_BUF_SIZE];
  int                num;
  int                err;
  char               line[MAXLINE << 4];

  struct pbsnode    *np;

  snprintf(log_buf, sizeof(log_buf), "%s()", __func__);

  log_record(PBSEVENT_SCHED, PBS_EVENTCLASS_REQUEST, __func__, log_buf);

  if ((err = parse_nodes_file()) != PBSE_NONE)
    return(err);
    
  nin = fopen(path_nodestate, "r");

  if (nin != NULL)
    {
    while (fscanf(nin, "%s %d",
                  line,
                  &num) == 2)
      {
      if ((np = find_nodebyname(line)) == NULL)
        {
        if (isdigit(line[0]))
          {
          // If cray enabled, create the node if it looks like a Cray subnode
          np = create_alps_subnode(alps_reporter, line);
          }
        }

      if (np != NULL)
        {
        // Update the state accordingly
        np->nd_state = num;

        /* exclusive bits are calculated later in set_old_nodes() */
        np->nd_state &= ~INUSE_JOB;
        np->unlock_node(__func__, "no match", LOGLEVEL);
        }
      }

    fclose(nin);
    }

  nin = fopen(path_nodepowerstate, "r");

  if (nin != NULL)
    {
    while (fscanf(nin, "%s %d",
                  line,
                  &num) == 2)
      {
      if ((np = find_nodebyname(line)) == NULL)
        {
        if (isdigit(line[0]))
          {
          // If cray enabled, create the node if it looks like a Cray subnode
          np = create_alps_subnode(alps_reporter, line);
          }
        }

      if (np != NULL)
        {
        np->nd_power_state = num;

        np->unlock_node(__func__, "match", LOGLEVEL);
        }
      }

    fclose(nin);
    }

  load_node_notes();

  /* SUCCESS */

  return(0);

  /* FAILURE */

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



/*
 * delete_a_subnode - mark a (last) single subnode entry as deleted
 */

void delete_a_subnode(

  struct pbsnode *pnode)

  {
  pnode->nd_slots.remove_execution_slot();
  return;
  }  /* END delete_a_subnode() */




/*
 * deletes the last gpu subnode
 * frees the node and decrements the number to adjust
 */
static void delete_a_gpusubnode(

  struct pbsnode *pnode)

  {
  // Don't delete unless we have gpu subnodes
  if (pnode->nd_gpusn.size() > 0)
    {
    gpusubn &tmp = pnode->nd_gpusn[pnode->nd_gpusn.size() - 1];

    if (tmp.inuse == false)
      pnode->nd_ngpus_free--;

    /* decrement the number of gpu subnodes */
    pnode->nd_ngpus--;

    pnode->nd_gpusn.pop_back();
    }
  } /* END delete_a_gpusubnode() */





/*
 * node_np_action - action routine for node's np pbs_attribute
 */

int node_np_action(
    
  pbs_attribute *new_attr,     /* derive props into this pbs_attribute*/
  void          *pobj,    /* pointer to a pbsnode struct     */
  int            actmode) /* action mode; "NEW" or "ALTER"   */
  
  {
  struct pbsnode *pnode = (struct pbsnode *)pobj;
  short  old_np;
  short  new_np;

  if (new_attr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input attributes");
    return(PBSE_BAD_PARAMETER);
    }

  if (pobj == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input node");
    return(PBSE_BAD_PARAMETER);
    }

  switch (actmode)
    {

    case ATR_ACTION_NEW:
      new_attr->at_val.at_long = pnode->nd_slots.get_total_execution_slots();
      break;

    case ATR_ACTION_ALTER:
      old_np = pnode->nd_slots.get_total_execution_slots();
      new_np = (short)new_attr->at_val.at_long;

      if (new_np <= 0)
        return PBSE_BADATVAL;

      while (new_np != old_np)
        {

        if (new_np < old_np)
          {
          delete_a_subnode(pnode);
          old_np--;
          svr_clnodes--;
          }
        else
          {
          add_execution_slot(pnode);
          old_np++;
          svr_clnodes++;
          }
        }

      break;
    default:
      log_err(-1,__func__, "unexpected action mode");
      return(-1);
    }

  return 0;
  } /* END node_np_action */



/*
 * node_mom_port_action - action routine for node's port pbs_attribute
 */

int node_mom_port_action(

  pbs_attribute *new_attr,     /*derive props into this pbs_attribute*/
  void          *pobj,    /*pointer to a pbsnode struct     */
  int            actmode) /*action mode; "NEW" or "ALTER"   */

  {
  struct pbsnode *pnode = (struct pbsnode *)pobj;
  int rc = 0;

  if (new_attr == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input attributes");
    return(rc);
    }

  if (pobj == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input node");
    return(rc);
    }

  switch (actmode)
    {

    case ATR_ACTION_NEW:
      new_attr->at_val.at_long = pnode->nd_mom_port;
      break;

    case ATR_ACTION_ALTER:
      pnode->nd_mom_port = new_attr->at_val.at_long;
      break;

    default:

      rc = PBSE_INTERNAL;
    }

  return rc;
  }

/*
 * node_mom_rm_port_action - action routine for node's port pbs_attribute
 */

int node_mom_rm_port_action(

  pbs_attribute *new_attr,     /* derive props into this pbs_attribute*/
  void          *pobj,    /* pointer to a pbsnode struct     */
  int            actmode) /* action mode; "NEW" or "ALTER"   */

  {
  struct pbsnode *pnode = (struct pbsnode *)pobj;
  int rc = 0;

  if (new_attr == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input attributes");
    return(rc);
    }

  if (pobj == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input node");
    return(rc);
    }

  switch (actmode)
    {

    case ATR_ACTION_NEW:
      new_attr->at_val.at_long = pnode->nd_mom_rm_port;
      break;

    case ATR_ACTION_ALTER:
      pnode->nd_mom_rm_port = new_attr->at_val.at_long;
      break;

    default:

      rc = PBSE_INTERNAL;
    }

  return rc;
  }



int node_gpus_action(

  pbs_attribute *new_attr,
  void          *pnode,
  int            actmode)

  {
  struct pbsnode *np = (struct pbsnode *)pnode;
  int             old_gp;
  int             new_gp;
  int             rc = 0;

  if (new_attr == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input attributes");
    return(rc);
    }

  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input node");
    return(rc);
    }

  switch (actmode)
    {
    case ATR_ACTION_NEW:
      new_attr->at_val.at_long = np->nd_ngpus;
      break;

    case ATR_ACTION_ALTER:
      old_gp = np->nd_ngpus;
      new_gp = new_attr->at_val.at_long;

      if (new_gp <= 0)
        return PBSE_BADATVAL;

      while (new_gp != old_gp)
        {

        if (new_gp < old_gp)
          {
          delete_a_gpusubnode((struct pbsnode *)pnode);
          old_gp--;
          }
        else
          {
          create_a_gpusubnode((struct pbsnode *)pnode);
          old_gp++;
          }
        }

      break;

    default:
      rc = PBSE_INTERNAL;
    }

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




int node_mics_action(

  pbs_attribute *new_attr,
  void          *pnode,
  int            actmode)

  {
  struct pbsnode *np = (struct pbsnode *)pnode;
  int             old_mics;
  int             new_mics;
  int             rc = 0;

  switch (actmode)
    {
    case ATR_ACTION_NEW:

      new_attr->at_val.at_long = np->nd_nmics;

      break;

    case ATR_ACTION_ALTER:

      old_mics = np->nd_nmics;
      new_mics = new_attr->at_val.at_long;

      if (new_mics <= 0)
        return(PBSE_BADATVAL);

      np->nd_nmics = new_mics;

      if (new_mics > old_mics)
        {
        np->nd_nmics_free += new_mics - old_mics;
        np->nd_nmics = new_mics;

        if (new_mics > np->nd_nmics_alloced)
          {
          for (int i = 0; i < new_mics - np->nd_nmics_alloced; i++)
            np->nd_micjobids.push_back(-1);
          
          np->nd_nmics_alloced = new_mics;
          }
        }

      break;

    default:
      rc = PBSE_INTERNAL;
    }

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




int node_numa_action(

  pbs_attribute *new_attr,     /* derive status into this pbs_attribute*/
  void          *pnode,   /* pointer to a pbsnode struct     */
  int            actmode) /* action mode; "NEW" or "ALTER"   */

  {

  struct pbsnode *np = (struct pbsnode *)pnode;
  int rc = 0;

  if (new_attr == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input attributes");
    return(rc);
    }

  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc,__func__, "NULL input node");
    return(rc);
    }

  switch (actmode)
    {
    case ATR_ACTION_NEW:
      new_attr->at_val.at_long = np->num_node_boards;
      break;

    case ATR_ACTION_ALTER:
      np->num_node_boards = new_attr->at_val.at_long;
      break;

    default:
      rc = PBSE_INTERNAL;
    }

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



int numa_str_action(

  pbs_attribute *new_attr,     /* derive status into this pbs_attribute*/
  void          *pnode,   /* pointer to a pbsnode struct     */
  int            actmode) /* action mode; "NEW" or "ALTER"   */

  {
  struct pbsnode *np = (struct pbsnode *)pnode;

  if (new_attr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input attributes");
    return(PBSE_BAD_PARAMETER);
    }

  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input node");
    return(PBSE_BAD_PARAMETER);
    }

  switch (actmode)
    {
    case ATR_ACTION_NEW:

      if (np->numa_str.size() != 0)
        {
        new_attr->at_val.at_str = strdup(np->numa_str.c_str());
        }
      else
        new_attr->at_val.at_str = NULL;

      break;

    case ATR_ACTION_ALTER:

      if (new_attr->at_val.at_str != NULL)
        {
        np->numa_str = new_attr->at_val.at_str;
        }
      else
        np->numa_str.clear();

      break;

    default:
      return(PBSE_INTERNAL);
    }

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




int gpu_str_action(

  pbs_attribute *new_attr,
  void          *pnode,
  int            actmode)

  {
  struct pbsnode *np = (struct pbsnode *)pnode;

  if (new_attr == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input attributes");
    return(PBSE_BAD_PARAMETER);
    }

  if (pnode == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input node");
    return(PBSE_BAD_PARAMETER);
    }

  switch (actmode)
    {
    case ATR_ACTION_NEW:

      if (np->gpu_str.size() != 0)
        {
        new_attr->at_val.at_str = strdup(np->gpu_str.c_str());
        }
      else
        new_attr->at_val.at_str = NULL;

      break;

    case ATR_ACTION_ALTER:

      if (new_attr->at_val.at_str != NULL)
        {
        np->gpu_str = new_attr->at_val.at_str;
        }
      else
        np->gpu_str.clear();

      break;

    default:
      return(PBSE_INTERNAL);
    }

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




/* create_partial_pbs_node - similar to create_pbs_node except there will
   only be a name for the new node and no attributes or properties */

int create_partial_pbs_node(

  char          *nodename,
  unsigned long  addr,
  int            perms)

  {
  int              rc;
  int              bad = 0;
  svrattrl        *plist = NULL;
  struct pbsnode  *pnode = NULL;
  u_long          *pul = NULL;

  if (nodename == NULL)
    {
    log_err(PBSE_BAD_PARAMETER,__func__, "NULL input name");
    return(PBSE_BAD_PARAMETER);
    }

  pul = (u_long *)calloc(2, sizeof(u_long));
  if (!pul)
    {
    return(PBSE_SYSTEM);
    }

  memset(pul, 0, sizeof(u_long) * 2);
  *pul = addr;

  pnode = new pbsnode(nodename, pul, false);
    
  free(pul);

  if ((rc = pnode->get_error()) != PBSE_NONE)
    {
    delete pnode;
    return(rc);
    }

  /* create and initialize the first subnode to go with the parent node */

  add_execution_slot(pnode);

  rc = mgr_set_node_attr(
         pnode,
         node_attr_def,
         ND_ATR_LAST,
         plist,
         perms,
         &bad,
         (void *)pnode,
         ATR_ACTION_ALTER,
         false);

  if (rc != 0)
    {
    pnode->lock_node(__func__, NULL, LOGLEVEL);
    effective_node_delete(&pnode);

    return(rc);
    }

  insert_node(&allnodes,pnode);
  auth_hosts.add_authorized_address(addr, pnode->nd_mom_port, pnode->get_name());
  
  svr_totnodes++;
  recompute_ntype_cnts();

  return(PBSE_NONE);     /*create completely successful*/
  } /* END create_partial_pbs_node */





/* 
 * initializes an allocated node iterator 
 */
void reinitialize_node_iterator(

  node_iterator *iter)

  {
  if (iter != NULL)
    {
    iter->node_index = NULL;
    iter->numa_index = -1;
    iter->alps_index = NULL;
    }
  } /* END reinitialize_node_iterator() */




static struct pbsnode *get_my_next_node_board(

  node_iterator  *iter,
  struct pbsnode *pnode)

  {
  struct pbsnode *numa;
  
  iter->numa_index++;
  numa = AVL_find(iter->numa_index, pnode->nd_mom_port, pnode->node_boards);
  
  pnode->unlock_node(__func__, "pnode", LOGLEVEL);
  if (numa != NULL)
    numa->lock_node(__func__, "numa", LOGLEVEL);

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




static struct pbsnode *get_my_next_alps_node(

  node_iterator  *iter,
  struct pbsnode *pnode)

  {
  struct pbsnode *alps_node = next_host(pnode->alps_subnodes, &(iter->alps_index), NULL);

  pnode->unlock_node(__func__, NULL, LOGLEVEL);

  return(alps_node);
  } /* END get_my_next_alps_node() */




/* 
 * @return the next node, from 0->end, accounting for numa nodes
 */
struct pbsnode *next_node(

  all_nodes      *an,
  struct pbsnode *current,
  node_iterator  *iter)

  {
  struct pbsnode *next;
  struct pbsnode *tmp;

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

  if (current == NULL)
    {
    an->lock();

    if (iter->node_index == NULL)
      iter->node_index = an->get_iterator();

    /* the first call to next_node */
    next = iter->node_index->get_next_item();
    if (next != NULL)
      next->lock_node(__func__, "next != NULL", LOGLEVEL);

    an->unlock();

    if (next != NULL)
      {
      /* if I have node_boards, look at those and not me */
      if (next->num_node_boards > 0)
        {
        next = get_my_next_node_board(iter,next);
        }
      else if (next->nd_is_alps_reporter)
        {
        next = get_my_next_alps_node(iter, next);
        }
      }
    } /* END first iteration */
  else
    {
    /* if current is a numa subnode, go back to the parent */
    if ((iter->numa_index >= 0) ||
        (iter->alps_index !=  NULL))
      {
      tmp = current->parent;
      current->unlock_node(__func__, "current == NULL && numa_index > 0", LOGLEVEL);
      if (tmp == NULL) /* TODO: think about this check and apropriate return*/
        {
        log_err(-1, __func__, "current->parent == NULL");
        return(NULL);
        }
      tmp->lock_node(__func__, "tmp && numa_index > 0", LOGLEVEL);
      current = tmp;
      }

    /* move to the next host or get my next node board? */
    if (cray_enabled == true)
      {
      if (current->nd_is_alps_reporter == TRUE)
        {
        if ((next = get_my_next_alps_node(iter, current)) == NULL)
          {
          iter->alps_index = NULL;
          
          an->lock();
          next = iter->node_index->get_next_item();
          
          if (next != NULL)
            {
            next->lock_node(__func__, NULL, LOGLEVEL);
            an->unlock();
            
            if (next->nd_is_alps_reporter)
              next = get_my_next_alps_node(iter, next);
            }
          else
            {
            an->unlock();
            }
          }
        }
      else
        {
        current->unlock_node(__func__, NULL, LOGLEVEL);
        iter->alps_index = NULL;
        
        an->lock();
        next = iter->node_index->get_next_item();
        
        if (next != NULL)
          {
          next->lock_node(__func__, NULL, LOGLEVEL);
          an->unlock();
          
          if (next->nd_is_alps_reporter)
            next = get_my_next_alps_node(iter, next);
          }
        else
          {
          an->unlock();
          }
        }
      }
    else if (iter->numa_index + 1 >= current->num_node_boards)
      {
      /* reset the numa_index to -1 */
      iter->numa_index = -1;

      /* go to the next node in all nodes */
      current->unlock_node(__func__, "next == NULL && numa_index+1", LOGLEVEL);
      an->lock();

      next = iter->node_index->get_next_item();


      if (next != NULL)
        {
        next->lock_node(__func__, "next != NULL && numa_index+1", LOGLEVEL);
        an->unlock();

        if (next->num_node_boards > 0)
          {
          next = get_my_next_node_board(iter, next);
          }
        }
      else
        {
        an->unlock();
        }
      }
    else
      {
      next = get_my_next_node_board(iter, current);
      }
    } /* END all other iterations */

  return(next);
  } /* END next_node() */




/*
 * insert a node into the array 
 *
 * @param pnode - the node to be inserted
 * @return PBSE_NONE on success 
 */

int insert_node(

  all_nodes      *an,    /* M */
  struct pbsnode *pnode) /* I */

  {
  int          rc = 0;

  if (an == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input all_nodes pointer");
    return(rc);
    }
  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input node pointer");
    return(rc);
    }

  an->lock();

  if(!an->insert(pnode, pnode->get_name()))
    {
    rc = ENOMEM;
    log_err(rc, __func__, "No memory to resize the array...SYSTEM FAILURE");
    }
  else
    {
    rc = PBSE_NONE;
    }

  an->unlock();

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




/* 
 * remove a node from the array
 *
 * @param pnode - the node to remove
 * @return PBSE_NONE if the node is removed 
 */

int remove_node(

  all_nodes      *an,
  struct pbsnode *pnode)

  {
  int rc = 0;

  if (an == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input all_nodes pointer");
    return(rc);
    }
  if (pnode == NULL)
    {
    rc = PBSE_BAD_PARAMETER;
    log_err(rc, __func__, "NULL input node pointer");
    return(rc);
    }

  if (an->trylock())
    {
    pnode->tmp_unlock_node(__func__, NULL, LOGLEVEL);
    an->lock();
    pnode->tmp_lock_node(__func__, NULL, LOGLEVEL);
    }

  if (an->remove(pnode->get_name()) == false)
    rc = -1;
  else
    rc = PBSE_NONE;

  an->unlock();

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




struct pbsnode *next_host(

  all_nodes           *an,    /* I */
  all_nodes_iterator **iter,  /* M */
  struct pbsnode      *held)  /* I */

  {
  struct pbsnode *pnode;
  int             id = -1;

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

  if (an->trylock())
    {
    if (held != NULL)
      {
      id = held->nd_id;
      held->unlock_node(__func__, NULL, LOGLEVEL);
      }
    an->lock();
    }
  if (*iter == NULL)
    {
    *iter = an->get_iterator();
    }

  pnode = (*iter)->get_next_item();
  if ((pnode != NULL) &&
      ((pnode != held) && 
       (id == -1)))
    {
    pnode->lock_node(__func__, NULL, LOGLEVEL);
    }

  an->unlock();

  if ((held != pnode) &&
      (id != -1))
    held = find_nodebyid(id);

  return(pnode);
  } /* END next_host() */



