
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
*/
package org.apache.uima.ducc.rm.scheduler;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.uima.ducc.common.Node;
import org.apache.uima.ducc.common.NodeIdentity;
import org.apache.uima.ducc.common.persistence.rm.IRmPersistence;
import org.apache.uima.ducc.common.persistence.rm.IRmPersistence.RmNodes;
import org.apache.uima.ducc.common.persistence.rm.RmPersistenceFactory;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType;


class NodePool
	implements SchedConstants
{
	static DuccLogger logger = DuccLogger.getLogger(NodePool.class, COMPONENT_NAME);
    String id;
    NodePool parent = null;

    int depth;
    int updated = 0;
    int search_order = 100;
    int share_quantum = 0;

    EvictionPolicy evictionPolicy = EvictionPolicy.SHRINK_BY_MACHINE;

    HashMap<String, NodePool> children                       = new HashMap<String, NodePool>();                // key is name of resource class
    Map<String, String>   subpoolNames                       = new HashMap<String, String>();                  // if a subpool, this names the membership

    HashMap<Node, Machine> allMachines                       = new HashMap<Node, Machine>();                   // all active machines in the system
    HashMap<Node, Machine> unresponsiveMachines              = new HashMap<Node, Machine>();                   // machines with excessive missed heartbeats
    HashMap<Node, Machine> offlineMachines                   = new HashMap<Node, Machine>();
    HashMap<Integer, HashMap<Node, Machine>> machinesByOrder = new HashMap<Integer, HashMap<Node, Machine>>(); // All schedulable machines, not necessarily free
    HashMap<String, Machine>                 machinesByName  = new HashMap<String, Machine>();                 // by name, for nodepool support
    HashMap<String, Machine>                 deadByName      = new HashMap<String, Machine>();                 // anything we move to offline or unresponsive,
                                                                                                               // but with the same name we used, because
                                                                                                               // sometimes stupid domain gets in the way
    HashMap<String, Machine>                 machinesByIp    = new HashMap<String, Machine>();                 // by IP, for nodepool support

    HashMap<Share, Share>                    allShares       = new HashMap<Share, Share>();

    HashMap<Node, Machine>                   preemptables    = new HashMap<Node, Machine>();                   // candidates for preemption for reservations
    int total_shares = 0;

    Map<ResourceClass, ResourceClass>        allClasses      = new HashMap<ResourceClass, ResourceClass>();    // all the classes directly serviced by me
    //
    // There are "theoretical" shares based on actual capacities of
    // the machines.  They are used for the "how much" part of the
    // calculations. They aren't correlated with any actual share
    // objects.
    //
    // nMachinesByOrder is initialzed on every scheduling cycle to the number of physical
    // machines of each share order in the system.  During scheduling, we start giving them
    // away, and must reduce the count accordingly.  It is an optimization representing the number
    // of completely free machines - machines that have not had any resources scheduled against them
    // in this cycle.
    //
    //                            assume 2x16G, 0x32G, 1x48G, 4x64G machines ------------------------------+
    //                            the arrays look like this for each init of the scheduler:                v
    int nMachinesByOrder[];       // number of full, free machines of each share order                [ 0  2 0 1  4 ] - physical machines
    int vMachinesByOrder[];       // number of partial machines, indexed by free space
    int nSharesByOrder[];         // shares of each size for each share order                         [ 0 21 9 5  4 ] - collective N Shares for each order
    //int nFreeSharesByOrder[];     // for each order, the theoretical number of shares to give away  [ 0  1 0 3 16 ] - free Q shares per order

    int nPendingByOrder[];        // number of N-shares with pending evictinos

    //int neededByOrder[];         // for each order, how many N-shares do I want to add?

    //     int shareExpansion[]; 
    Map<Integer, Integer> onlineMachinesByOrder = new HashMap<Integer, Integer>();  // all online machines

    // Indexed by available free shares, the specific machines with the indicated free space
    HashMap<Integer, Map<Node, Machine>> virtualMachinesByOrder = new HashMap<Integer, Map<Node, Machine>>();  // UIMA-4142
    GlobalOrder maxorder = null;

    IRmPersistence persistence = null;
    boolean canReserve = false;       // if we contain a class with policy Reserve, then stuff in this pool is reservable

//     NodePool(NodePool parent, String id, EvictionPolicy ep, int order)
//     {
//         this.parent = parent;
//         this.id = id;
//         this.evictionPolicy = ep;
//         this.depth = 0;
//         this.order = order;
//     }

    NodePool(NodePool parent, String id, Map<String, String> nodes, EvictionPolicy ep, int depth, int search_order, int share_quantum)
    {
    	String methodName = "NodePool.<init>";
        this.parent = parent;
        this.id = id;
        this.subpoolNames = nodes;
        if ( nodes == null ) {            // unlikely, but not illegal
            this.subpoolNames = new HashMap<String, String>();
            logger.warn(methodName, null, "Nodepool", id, ": no nodes in node list");
        } 
        this.evictionPolicy = ep;
        this.depth = depth;
        this.search_order = search_order;
        this.share_quantum = share_quantum;              // in KB

        if ( parent == null ) {
            maxorder = new GlobalOrder();
        } else {
            maxorder = parent.getGlobalOrder();
        }

        persistence = RmPersistenceFactory.getInstance(this.getClass().getName(), "RM");
    }

    void addResourceClass(ResourceClass cl)
    {  // UIMA-4065
        allClasses.put(cl, cl);
        if ( cl.getPolicy() == Policy.RESERVE) canReserve = true;
    }

    NodePool getParent()
    {
        return this.parent;
    }

    String getId()
    {
        return id;
    }

    int getShareQuantum()
    {
        return share_quantum;
    }

    int getDepth()
    {
        return depth;
    }

    int countShares()
    {
        return allShares.size();
    }

    int countOccupiedShares()
    {
        int count = allShares.size();
        for ( NodePool np : children.values() ) {
            count += np.countOccupiedShares();
        }
        return count;
    }

    //
    // Note, this will only be accurate AFTER reset, but before actuall allocation of
    // shares begins.  After allocation, and before the next reset this will return junk.
    //
    // It is intended to be called from ResourceClass.canUseBonus()
    // UIMA-4065
    int countAssignableShares(int order)
    {
        String methodName = "countAssignableShares";
        // first calculate my contribution
        int ret = nSharesByOrder[order];
        for (ResourceClass rc : allClasses.values() ) {
            int[] gbo = rc.getGivenByOrder();
            if ( gbo != null ) {
                ret -= gbo[order];
            }
        }
        logger.trace(methodName, null, "Shares available for", id, ":", ret);
        // now accumulate the kid's contribution
        for ( NodePool np : children.values() ) {
            ret += np.countAssignableShares(order);
        }
        return ret;
    }

    void removeShare(Share s)
    {
        allShares.remove(s);
    }

    boolean containsPoolNode(Node n)
    {
        // allow the names to be machines or ip addresses
        if ( subpoolNames.containsKey( n.getNodeIdentity().getIp()   )) return true;
        if ( subpoolNames.containsKey( n.getNodeIdentity().getName() )) return true;
        return false;
    }

    /**
     * How many do I have, including recusring down the children?
     */
    int countMachines()
    {
        int count = allMachines.size();
        for ( NodePool np : children.values() ) {
            count += np.countMachines();
        }
        return count;
    }
     /**
     * How many do I have, including recursing down the children?
     */
    int countUnresponsiveMachines()
    {
        int count = unresponsiveMachines.size();
        for ( NodePool np : children.values() ) {
            count += np.countUnresponsiveMachines();
        }
        return count;
    }

     /**
     * How many do I have, just me.
     */
    int countLocalUnresponsiveMachines()
    {
        return unresponsiveMachines.size();
    }

    /**
     * How many do I have, including recusring down the children?
     */
    int countOfflineMachines()
    {
        int count = offlineMachines.size();
        for ( NodePool np : children.values() ) {
            count += np.countOfflineMachines();
        }
        return count;
    }

    /**
     * How many do I have, just me.
     */
    int countLocalOfflineMachines()
    {
        return offlineMachines.size();
    }

    /**
     * Return nodes varied off for me and my kids.
     * UIMA-4142, RM reconfiguration
     */
    Map<Node, Machine> getOfflineMachines()
    {
        @SuppressWarnings("unchecked")
		Map<Node, Machine> ret = (Map<Node, Machine>) offlineMachines.clone();
        for (NodePool np : children.values()) {
            ret.putAll(np.getOfflineMachines());
        }
        return ret;
    }

    /**
     * Return nodes varied off for me and my kids.
     * UIMA-4234, More info in query occupancy
     */
    Map<Node, Machine> getUnresponsiveMachines()
    {
        @SuppressWarnings("unchecked")
		Map<Node, Machine> ret = (Map<Node, Machine>) unresponsiveMachines.clone();
        for (NodePool np : children.values()) {
            ret.putAll(np.unresponsiveMachines);
        }
        return ret;
    }

    /**
     * Non-recursive machine count.
     */
    int countLocalMachines()
    {
        return allMachines.size();
    }

    /**
     * Non recursive share count;
     */
    int countLocalShares()
    {
        return total_shares;
    }
    
    /**
     * Counts just local, for reservations.
     */
    int countFreeMachines(int order)
    {
        int cnt = 0;

        HashMap<Node, Machine> mlist = null;
        mlist = machinesByOrder.get(order);
        if ( mlist == null ) return 0;

        for ( Machine m : mlist.values() ) {
            if ( isSchedulable(m) && m.isFree() ) {
                cnt++;
            }
        }
        return cnt;
    }

    /**
     * Counts all known machines, just me.
     */
    int[] countLocalFreeMachines()
    {
        return nMachinesByOrder.clone();
    }

    int countTotalShares()
    {
        int answer = total_shares;
        for ( NodePool np : children.values() ) {
            answer += np.countTotalShares();
        }
        return answer;
    }

    /**
     * Total Q shares in the nodepool that are not yet given away in the scheduling cycle.
     */ 
    int countQShares()
    {
        int count = nSharesByOrder[1];
        for ( NodePool np : children.values() ) {
            count += np.countQShares();
        }
        return count;
    }

    /**
     * Total Q shares in the nodepool still available, just me.
     */ 
    int countLocalQShares()
    {
        return nSharesByOrder[1];
    }


    int countAllMachinesByOrder(int o)
    {
        int count = 0;
        if ( machinesByOrder.containsKey(o) ) {
            count = machinesByOrder.get(o).size();
        }

        for ( NodePool np : children.values() ) {
            count += np.countAllMachinesByOrder(o);
        }
        return count;
    }

    int[] countAllLocalMachines()
    {
        int[] ret = makeArray();
        for ( int o : machinesByOrder.keySet() ) {
            ret[o] = machinesByOrder.get(o).size();
        }
        return ret;
    }

    /**
     * Returns N-Shares, recursing down
     */
    int countNSharesByOrder(int o)
    {
        int count = nSharesByOrder[o];
        for ( NodePool np : children.values() ) {
            count += np.countNSharesByOrder(o);
        }
        return count;
    }

    /**
     * Returns N-Shares, local
     */
    int countLocalNSharesByOrder(int o)
    {
        return nSharesByOrder[o];
    }

    /**
     * Returns number of N-shares that are still busy but pending eviction.
     */
    int countPendingSharesByOrder(int o)
    {
        int count = nPendingByOrder[o];
        for ( NodePool np : children.values() ) {
            count += np.countPendingSharesByOrder(o);
        }
        return count;
    }

    /**
     * Helper for compatibleNodepool(), recurses down children.
     * UIMA-4142
     */
    private boolean isCompatibleNodepool(Policy p, ResourceClass rc)
    {
        if ( allClasses.containsKey(rc) ) return true;
        for (NodePool np : children.values()) {
            if ( np.isCompatibleNodepool(p, rc) ) return true;
        }      
        return false;
    }

    /**
     * Helper for compatibleNodepool(), find the top of the heirarchy.
     * UIMA-4142
     */
    NodePool findTopOfHeirarchy()
    {
        NodePool ret = this;
        while (ret.getParent() != null) {
            ret = ret.getParent();
        }
        return ret;
    }

    /**
     * Interrogate whether work assigned to the indicated rc could end up here.  
     *
     * If it's a fair-share allocation, we need to interrogate 'me', my children, and 
     * my ancestors.
     *
     * If it's something else, it must reside reight here.
     *
     * This is called during recovery; a change to the class or np config can cause incompatibilities
     * with previously scheduled work after a restart.
     *
     * UIMA-4142
     *
     * @param     p   The scheduling policy; determines whether descent into child pools is allowed.
     * @param    rc  The rc to check
     * @return true If work scheduled to the RC is compatible.
     */
    boolean compatibleNodepool(Policy p, ResourceClass rc)
    {
        switch ( p ) {
            case FAIR_SHARE:
                NodePool top = findTopOfHeirarchy();
                return top.isCompatibleNodepool(p, rc);
            case FIXED_SHARE:
            case RESERVE:
                if ( allClasses.containsKey(rc) ) return true;
        }

        return false;
    }

    int[] cloneNSharesByOrder()
    {
        int[] cln =  nSharesByOrder.clone();
        for ( NodePool np : children.values() ) {
            int[] subcln = np.cloneNSharesByOrder();
            for ( int i = 0; i < cln.length; i++ ) {
                cln[i] += subcln[i];
            }
        }
        return cln;
    }

    int[] cloneVMachinesByOrder()
    {
        int[] cln =  nMachinesByOrder.clone();
        for ( int i = 0; i < cln.length; i++ ) {
            cln[i] += vMachinesByOrder[i];
        }

        for ( NodePool np : children.values() ) {
            int[] subcln = np.cloneVMachinesByOrder();
            for ( int i = 0; i < cln.length; i++ ) {
                cln[i] += subcln[i];
            }
        }
        return cln;
    }

    public GlobalOrder getGlobalOrder()
    {
        return maxorder;
    }

    public void updateMaxOrder(int order)
    {
        maxorder.update(order);
    }

    public int getMaxOrder()
    {
        return maxorder.getOrder();              // must always be the same for parent and all children
    }

    public int getArraySize()
    {
        return getMaxOrder() + 1;                      // a bit bigger, because we're 1-indexed for easier counting
                                                       // same for parent and children
    }

    public int[] makeArray()                           // common static code because getting this right everywhere is painful
    {
        return new int[getArraySize()];
    }


    int getSearchOrder()
    {
        return this.search_order;
    }

    public Machine getMachine(Node n)
    {
        Machine m = allMachines.get(n);

        if ( m == null ) {
            for ( NodePool np : children.values() ) {
                m = np.getMachine(n);
                if ( m != null ) {
                    break;
                }
            }
        }
        return m;
    }

    public Machine getMachine(NodeIdentity ni)
    {    	
        Machine m = machinesByIp.get(ni.getIp());
        if ( m == null ) {
            for ( NodePool np : children.values() ) {
                m = np.getMachine(ni);
                if ( m != null ) break;
            }                
        }
        return m;
    }

    boolean containsMachine(Machine m)
    {
        Map<Node, Machine> allm = getAllMachines();
        return allm.containsKey(m.getNode());
    }

    @SuppressWarnings("unchecked")
	HashMap<Node, Machine> getAllMachinesForPool()
    {
        return (HashMap<Node, Machine>) allMachines.clone();
    }

    HashMap<Node, Machine> getAllMachines()
    {
        @SuppressWarnings("unchecked")
		HashMap<Node, Machine> machs = (HashMap<Node, Machine>) allMachines.clone();
        for ( NodePool np : children.values() ) {
            HashMap<Node, Machine> m = np.getAllMachines();
            if ( m != null ) {
                machs.putAll(m);
            }
        }                
    
        return machs;
    }

    HashMap<String, Machine> getMachinesByName()
    {
        @SuppressWarnings("unchecked")
		HashMap<String, Machine> machs = (HashMap<String, Machine>) machinesByName.clone();
        for ( NodePool np : children.values() ) {
            HashMap<String, Machine> m = np.getMachinesByName();
            if ( m != null ) {
                machs.putAll(m);
            }
        }                

        return machs;
    }

    HashMap<String, Machine> getMachinesByIp()
    {
        @SuppressWarnings("unchecked")
		HashMap<String, Machine> machs = (HashMap<String, Machine>) machinesByIp.clone();
        for ( NodePool np : children.values() ) {
            HashMap<String, Machine> m = np.getMachinesByIp();
            if ( m != null ) {
                machs.putAll(m);
            }
        }                

        return machs;
    }

    @SuppressWarnings("unchecked")
	HashMap<Node, Machine> getMachinesByOrder(int order)
    {

        HashMap<Node, Machine> machs;

        if( machinesByOrder.containsKey(order) ) {
            machs = (HashMap<Node, Machine>) machinesByOrder.get(order).clone();
        } else {
            machs = new HashMap<Node, Machine>();
        }

        for ( NodePool np : children.values() ) {
            HashMap<Node, Machine> m = np.getMachinesByOrder(order);
            machs.putAll(m);
        }                

        return machs;
    }

    @SuppressWarnings("unchecked")
	Map<Node, Machine> getVirtualMachinesByOrder(int order)
    {
        Map<Node, Machine> machs;

        if( virtualMachinesByOrder.containsKey(order) ) {
        	HashMap<Node, Machine> tmp = (HashMap<Node, Machine>) virtualMachinesByOrder.get(order);
            machs = (HashMap<Node, Machine>) tmp.clone();
        } else {
            machs = new HashMap<Node, Machine>();
        }

//         for ( NodePool np : children.values() ) {
//             HashMap<Machine, Machine> m = np.getVirtualMachinesByOrder(order);
//             machs.putAll(m);
//         }                

        return machs;
    }

    /**
     * Work out the N shares for each share class.
     *
     * Note: This is a helper class, not made public, and does not need to account for child nodepools.
     *       If you need the recursion use countOutNSharesByOrder().
     *
     * Internally, only call this if you mess with the counting arrays.  If you call somebody else who
     * messes with the counting arrays, leave it to them to call this.
     */
    protected void calcNSharesByOrder()
    {
        int len = nMachinesByOrder.length;

        // init nSharesByorder to the sum of 'n and 'v MachinesByOrder
        System.arraycopy(nMachinesByOrder, 0, nSharesByOrder, 0, len);
        for ( int i = 0; i < getMaxOrder() + 1; i++ ) {
            nSharesByOrder[i] += vMachinesByOrder[i];
        }

        for ( int o = 1; o < len; o++ ) {                     // counting by share order
            //nFreeSharesByOrder[o] = nMachinesByOrder[o] * o;
            for ( int p = o+1; p < len; p++ ) {
                if ( nSharesByOrder[p] != 0 ) {
                    nSharesByOrder[o] += (p / o) * nSharesByOrder[p];
                }
            }
        }
    }

    protected int[] countMachinesByOrder()
    {
        int[] ans = nMachinesByOrder.clone();
        for ( NodePool np : children.values() ) {
            int[] tmp = np.countMachinesByOrder();
            for ( int i = 0; i < getArraySize(); i++ ) {
                ans[i] += tmp[i];
            }
        }
        return ans;
    }

    
    protected int[] countVMachinesByOrder()
    {
        int[] ans = vMachinesByOrder.clone();
        for ( NodePool np : children.values() ) {
            int[] tmp = np.countVMachinesByOrder();
            for ( int i = 0; i < getArraySize(); i++ ) {
                ans[i] += tmp[i];
            }
        }
        return ans;
    }

    protected int[] countLocalVMachinesByOrder()
    {
        return vMachinesByOrder.clone();
    }

    protected int[] countAllNSharesByOrder()
    {
        int[] ans = nSharesByOrder.clone();
        for ( NodePool np : children.values() ) {
            int[] tmp = np.countAllNSharesByOrder();
            for ( int i = 0; i < getArraySize(); i++ ) {
                ans[i] += tmp[i];
            }
        }
        return ans;
    }


    /**
     * Common code to connect a share into the system, used when assigning a new
     * share (from within NodePool), or when reconnecting a share during job recovery
     * (from JobManagerCoverter).
     */
    public synchronized void connectShare(Share s, Machine m, IRmJob j, int order)
    {
        String methodName = "connectShare";
        logger.info(methodName, j.getId(), "share", s,  "order", order, "machine", m);
        j.assignShare(s);
        m.assignShare(s);
        rearrangeVirtual(m, order, j.getSchedulingPolicy());
        allShares.put(s, s);        
    }

    void rearrangeVirtual(Machine m, int order, Policy policy)
         
    {
    	String methodName = "rearrangeVirtual";
        if ( allMachines.containsKey(m.key()) ) {
            int v_order = m.getVirtualShareOrder();
            int r_order = m.getShareOrder();

            // UIMA-4913 Avoid index-outta-bounds when a machine's size changes
            // If Share appears bigger than remaining free-space pretend the share is smaller,
            // or if share is smaller but is a whole machine reservation, then pretend the
            // reservation matches the machine, so in both cases the free space = 0
            if (order > v_order) {
                logger.warn(methodName, null, m.getId(), "found a share of size", order, "on a machine with only", v_order, "free slots - set free=0");
                order = v_order;
            } else if (order < v_order && policy == Policy.RESERVE){
                logger.warn(methodName, null, m.getId(), "found a RESERVE share of size", order, "on a machine with", v_order, "free slots - set free=0");
                order = v_order;
            }

            logger.trace(methodName, null, m.getId(), "order", order, "v_order", v_order, "r_order", r_order);

            if ( v_order == r_order ) {
                nMachinesByOrder[r_order]--;
            } else {
                vMachinesByOrder[v_order]--;
            }
           
            Map<Node, Machine> vlist = virtualMachinesByOrder.get(v_order);
            if ( vlist == null ) {
                // Delivered under UIMA-4275 as that is when I decided to try to avoid NPE here.
                //
                // This is fatal, the internal records are all wrong.  Usually this is because of some
                // external snafu, such as mixing and matching ducc clusters on the same broker.
                // There's really not much we can do though.  There's a good chance that continuing
                // will cause NPE elsewhere.  Maybe we can just ignore it and let it leak?
                logger.error(methodName, null, "ERROR: bad virtual machine list.", m.getId(), "order", order, "v_order", v_order, "r_order", r_order);
                return;
            }
            vlist.remove(m.key());
            
            v_order -= order;
            m.setVirtualShareOrder(v_order);
            if (v_order != 0 ) {
                vlist = virtualMachinesByOrder.get(v_order);
                if ( vlist == null ) {
                    vlist = new HashMap<Node, Machine>();
                    virtualMachinesByOrder.put(v_order, vlist);
                }
                vlist.put(m.key(), m);
                vMachinesByOrder[v_order]++;
            }
            calcNSharesByOrder();
        } else {
            for ( NodePool np : children.values() ) {
                np.rearrangeVirtual(m, order, policy);
            }
        }
    }


    void accountForShares(HashMap<Share, Share> shares)
    {
        if ( shares == null ) return;

        for ( Share s : shares.values() ) {
            int order = s.getShareOrder();
            Machine m = s.getMachine();
            Policy policy = s.getJob().getSchedulingPolicy();
            rearrangeVirtual(m, order, policy);
        }
    }

    /**
     * Prepare the structures for scheduling.  These get modified in place by the scheduler.
     *
     * @param order is the hightest order of any job that can potentially get scheduled.
     *        We need this to insure the tables have sufficient space and we don't get NPEs.
     */
    void reset(int order)
    {
    	String methodName = "reset";
        //
        // TODO: Not all of these are used in every reset cycle.  Maybe we should break up the
        //       reset code so it matches the cycles better.  otoh, this isn't a performance-intensive
        //       scheduler so do we care?
        //
        updateMaxOrder(order);
        logger.info(methodName, null, "Nodepool:", id, "Maxorder set to", getMaxOrder());

        nSharesByOrder     = makeArray();
        nMachinesByOrder   = makeArray();
        vMachinesByOrder   = makeArray();
        //nFreeSharesByOrder = new int[maxorder + 1];
        //neededByOrder      = new int[maxorder + 1];

        nPendingByOrder = makeArray();

        // UIMA-4142 Must set vMachinesByOrder and virtualMachinesByOrder independently of
        //           machinesByOrder because blacklisting can cause v_order != r_order
        //           during reset.
        // UIMA-4910 Ignore unusable machines 
        virtualMachinesByOrder.clear();
        for ( Machine m : allMachines.values() ) {
            if ( !isSchedulable(m) ) {
              continue;               // Ignore unusable machines
            }
            m.resetVirtualShareOrder();
            int v_order = m.getVirtualShareOrder();
            int r_order = m.getShareOrder();
            
            Map<Node, Machine> ml = null;
            if ( v_order == r_order ) {
                nMachinesByOrder[r_order]++;
            } else {
                vMachinesByOrder[v_order]++;
            }

            ml = virtualMachinesByOrder.get(v_order);
            if ( ml == null ) {
                ml = new HashMap<Node, Machine>();
                virtualMachinesByOrder.put(v_order, ml);
            }

            ml.put(m.key(), m);
        }

        // UIMA 4142 this old calc isn't right any more because blacklisting can cause
        //     v_order != r_order during reset
        // virtualMachinesByOrder = new HashMap<Integer, HashMap<Node, Machine>>();
        // for ( Integer i : machinesByOrder.keySet() ) {

        //     @SuppressWarnings("unchecked")
        //     HashMap<Node, Machine> ml = (HashMap<Node, Machine>) machinesByOrder.get(i).clone();

        //     virtualMachinesByOrder.put(i, ml);
        //     nMachinesByOrder[i] = ml.size();
        // }


        calcNSharesByOrder();

        for ( NodePool np : children.values() ) {
            np.reset(order);
        }

        if ( (parent == null) && ( updated > 0 ) ) {
        	// top-level nodepool will recurse
            logger.info(methodName, null, "Scheduling Tables:\n", toString());
            updated = 0;
        }
    }

    void resetPreemptables()
    {
    	String methodName = "resetPreemptables";
        logger.info(methodName, null, "Resetting preemptables in nodepool", id);

        // UIMA-4064 Need to do this recrsively
        preemptables.clear();
        for ( NodePool np : children.values() ) {
            np.resetPreemptables();
        }

    }


    /**
     * Return the specified subpool, or me!
     */
    NodePool getSubpool(String name)
    {
        if ( name.equals(id) ) {
            return this;
        }

        for ( NodePool np : children.values() ) {
            NodePool ret = np.getSubpool(name);
            if (ret != null) return ret;
        }
        return null;
    }

    /**
     * Do "I" contain the indicated nodepool? More accurately, can "I" access everything in the
     * indicated nodepool?
     */
    boolean containsSubpool(NodePool np)
    {
        if ( np == this ) {
            return true;
        }

        for ( NodePool cnp : children.values() ) {
            if (cnp.containsSubpool(np)) return true;
        }
        return false;
    }

    HashMap<String, NodePool> getChildren()
    {
        return children;
    }
    
    List<NodePool> getChildrenAscending()
    {
        ArrayList<NodePool> sorted = new ArrayList<NodePool>();
        if ( children.size() > 0 ) {
            sorted.addAll(children.values());
            Collections.sort(sorted, new NodepoolAscendingSorter());
        }
        return sorted;
    }

    List<NodePool> getChildrenDescending()
    {
        ArrayList<NodePool> sorted = new ArrayList<NodePool>();
        if ( children.size() > 0 ) {
            sorted.addAll(children.values());
            Collections.sort(sorted, new NodepoolDescendingSorter());
        }
        return sorted;
    }

    /**
     * Subpools are always associated with a classname.
     *
     * We can assume that all node updates are refused until all subpools are created
     * so we don't have to worry about updating the pools until nodeArrives(), below.
     */
    NodePool createSubpool(String className, Map<String, String> names, int order)
    {
        NodePool np = new NodePool(this, className, names, evictionPolicy, depth + 1, order, share_quantum);
        children.put(className, np);
        return np;
    }

//    private synchronized void incrementOnlineByOrder(int order)
//    {
//        if ( ! onlineMachinesByOrder.containsKey(order) ) {
//            onlineMachinesByOrder.put(order, 1);
//        } else {
//            onlineMachinesByOrder.put(order, onlineMachinesByOrder.get(order) + 1);
//        }
//    }

//    private synchronized void decrementOnlineByOrder(int order)
//    {
//        onlineMachinesByOrder.put(order, onlineMachinesByOrder.get(order) - 1);
//    }

//    synchronized void getLocalOnlineByOrder(int[] ret)         // for queries, just me
//    {
//        for ( int o: onlineMachinesByOrder.keySet() ) {
//            ret[o] += onlineMachinesByOrder.get(o);
//        }
//    }

//    synchronized void getOnlineByOrder(int[] ret)         // for queries
//    {
//        for ( int o: onlineMachinesByOrder.keySet() ) {
//            ret[o] += onlineMachinesByOrder.get(o);
//        }
//        for ( NodePool child : children.values() ) {
//            child.getOnlineByOrder(ret);
//        }
//    }


    void signalDb(Machine m, RmNodes key, Object value)
    {
    	String methodName = "signalDb";
        try {
			persistence.setNodeProperty(m.getNode().getNodeIdentity().getName(), key, value);
		} catch (Exception e) {
			logger.warn(methodName, null, "Cannot update DB property", key, "for machine", m);
		}
    }

    Map<RmNodes, Object> initDbProperties(Machine m)
    {
        Node n = m.getNode();
        NodeIdentity nid = n.getNodeIdentity();
        
        Map<RmNodes, Object> props = new HashMap<RmNodes, Object>();
        props.put(RmNodes.Name, nid.getName());
        props.put(RmNodes.Ip, nid.getIp());
        props.put(RmNodes.Nodepool, id);
        props.put(RmNodes.Quantum, share_quantum / ( 1024*1024));
        
        props.put(RmNodes.Memory       , m.getMemory() / (1024*1024));
        props.put(RmNodes.ShareOrder  , m.getShareOrder());
        props.put(RmNodes.Blacklisted  , m.isBlacklisted());

        // init these here, but must be maintained by machine
        props.put(RmNodes.Heartbeats   , 0);
        props.put(RmNodes.SharesLeft   , m.countFreeShares());     // qshares remaining
        props.put(RmNodes.Assignments  , m.countProcesses());      // processes
        props.put(RmNodes.NPAssignments, m.countNpShares());
        
        props.put(RmNodes.Reservable   , canReserve);

        StringBuffer buf = new StringBuffer();
        for ( ResourceClass cl : allClasses.keySet() ) {
            buf.append(cl.getName());
            buf.append(" ");
        }
        props.put(RmNodes.Classes, buf.toString());
        return props;
    }

    void adjustMachinesByOrder(int neworder, Machine m)
    {
        int oldorder = m.getShareOrder();
        if ( oldorder != neworder ) {                  // can change. e.g. if it was taken offline for
            HashMap<Node, Machine> mlist = machinesByOrder.get(oldorder);
            mlist.remove(m.key());
            m.setShareOrder(neworder);                          //    hardware changes.
            signalDb(m, RmNodes.ShareOrder, neworder);          // Jira 4913 Update DB so ducc-mon can show the current size 
            mlist = machinesByOrder.get(neworder);
            if ( mlist == null ) {
                mlist = new HashMap<Node, Machine>();
                machinesByOrder.put(neworder, mlist);
            }
            mlist.put(m.key(), m);                     
        }        
    }

    /**
     * Handle a new node update.
     */
    Machine nodeArrives(Node node, int order)
    {
        String methodName = "nodeArrives";
        // Note: the caller of this method MUST (aka IS REQUIRED) to insure this this is the
        //       right nodepool as we do not recurse.

        updateMaxOrder(order);

        String n = node.getNodeIdentity().getName();

        // if it's offline it can't be restored like this.
        if ( offlineMachines.containsKey(node) ) {
            Machine m = offlineMachines.get(node);
            logger.trace(methodName, null, "Node ", m.getId(), " is offline, not activating.");
            return m;
        }
        // logger.info(methodName, null, "NODEARRIVES", n, "pass offline Machines");

        // if it was dead, then it isn't any more, AND it's mine, so I need to restart it
        if ( unresponsiveMachines.containsKey(node) ) {          // reactive the node

            logger.info(methodName, null, "RECOVER NODE", n);
            Machine m = unresponsiveMachines.remove(node);       // not unresponsive any more

            // Deal with memory on the machine changing
            adjustMachinesByOrder(order, m);

            // Note: The machine must be on all the other lists by definition since it wasn't taken off when it went offline

            signalDb(m, RmNodes.Responsive, true);
            logger.info(methodName, null, "Nodepool:", id, "Host reactivated ", m.getId(), "shares", order, m.toString());
            return m;
        }
        // logger.info(methodName, null, "NODEARRIVES", n, "pass unresponsive Machines");

        // ok, it is my problem?  If so, then it isn't offline or dead, so it's ok, and we're done
        if ( allMachines.containsKey(node) ) {                   // already known, do nothing
            Machine m = allMachines.get(node);

            // Deal with memory on the machine changing
            adjustMachinesByOrder(order, m);

            logger.trace(methodName, null, "Node ", m.getId(), " is already known, not adding.");
            return m;
        }
        // logger.info(methodName, null, "NODEARRIVES", n, "pass allMachines");

        // If we fall through it's a new one.
        Machine machine = new Machine(node);                     // brand new machine, make it active
        Node key = machine.key();
        machine.setShareOrder(order);
        allMachines.put(key, machine);                 // global list
        machinesByName.put(machine.getId(), machine);
        machinesByIp.put(machine.getIp(), machine);
        //incrementOnlineByOrder(order);
        machine.setNodepool(this);

        total_shares += order;     

        // index it by its share order to make it easier to find        
        HashMap<Node, Machine> mlist = machinesByOrder.get(order);
        if ( mlist == null ) {
            mlist = new HashMap<Node, Machine>();
            machinesByOrder.put(order, mlist);
        }
        mlist.put(key, machine);        

        logger.info(methodName, null, "Nodepool:", id, "Host added:", id, ": ", machine.getId(), "Nodefile:", subpoolNames.get(machine.getId()), // UIMA-4142, add file nodefile
                    String.format("shares %2d total %4d:", order, total_shares), machine.toString()); 
        updated++;

        Map<RmNodes, Object> props = initDbProperties(allMachines.get(key));
        props.put(RmNodes.Responsive, true);
        props.put(RmNodes.Online, true);
        try {
            persistence.createMachine(machine.getId(), props);
        } catch (Exception e) {
            logger.warn(methodName, null, "Cannot write machine to DB:", machine.getId(), e);
        }
        
        return machine;
    }

    /**
     * Purge all or some of the work on a machine that has died, or been taken offline
     * 
     * @param m            node being removed
     * @param removeAll    true if all work is to be purged, otherwise just the preemptable work
     * 
     * Ignore unmanaged reservations as they have no ducc-managed work
     * Purge just fair-share/preemptable work if being varyed off, or all work if node has died (UIMA-4752)
     */
    void disable(Machine m, boolean removeAll)
    {
        String methodName = "disable";

        logger.info(methodName, null, "Nodepool:", id, "Host disabled:", m.getId(), "Looking for shares to clear");

        String eventType = removeAll ? "Host dead:" : "Host offline:";

        HashMap<Share, Share> shares = m.getActiveShares();
        for (Share s : shares.values()) {
            IRmJob j = s.getJob();

            if ( j.getDuccType() == DuccType.Reservation ) {
              logger.info(methodName, null, "Nodepool:", id, eventType, m.getId(), "Not purging", j.getDuccType());
              continue;
            }
            if ( removeAll || j.getSchedulingPolicy() == Policy.FAIR_SHARE )  {
                // NOTE: Currently will never get a Pop ... just a Service of type Other !!
                logger.info(methodName, j.getId(), "Nodepool:", id, eventType, j.getDuccType(), "purge:", m.getId());
                if (j.getDuccType() == DuccType.Service || j.getDuccType() == DuccType.Pop) {
                  j.markComplete();      // UIMA-4327 Must avoid reallocation, these guys are toast if they get purged.
                  logger.info(methodName, j.getId(), "Nodepool:", id, eventType, m.getId(), "Mark service/pop completed.");
                }
                j.shrinkByOne(s);   // De-allocate this share
                s.purge();          // This bit tells OR not to wait for confirmation from the agent

                int order = s.getShareOrder();
                nPendingByOrder[order]++;
            } else {
                logger.info(methodName, j.getId(), "Nodepool:", id, eventType, m.getId(), "Not purging NP work - ", j.getDuccType());
            } 
            
        }
    }

    void nodeLeaves(Machine m)
    {
        // note, simpler than varyoff because we really don't care about unusual
        // conditions since there's nobody to tell
        if ( allMachines.containsKey(m.key()) ) {            
            disable(m, true);    // Remove all work
            unresponsiveMachines.put(m.key(), m);
            signalDb(m, RmNodes.Responsive, false);
        } else {
            for ( NodePool np : children.values() ) {
                np.nodeLeaves(m);
            }
        }
    }

    // UIMA-4142
    // helper for CLI things that refer to things by name only.  do we know about anything by this
    // name?  see resolve() in Scheduler.java.
    boolean hasNode(String n)
    {
        return machinesByName.containsKey(n);
    }

    NodePool findNodepoolByNodename(String n)
    {
        if ( hasNode(n) ) {
            return this;
        } else {
            for ( NodePool np : children.values() ) {
                NodePool ret = np.findNodepoolByNodename(n);
                if ( ret != null ) {
                    return ret;
                }
            }            
        }
        return null;
    }

    private String doVaryOff(String node)
    {
    	// caller must insure node is known to "me"
        Machine m = machinesByName.get(node);
        if (offlineMachines.containsKey(m.key()) ) {            
            return "VaryOff: Nodepool " + id + " - Already offline: " + node;
        }

        if ( unresponsiveMachines.containsKey(m.key()) ) {
            // lets be friendly and tell caller it's also unresponsive
            offlineMachines.put(m.key(), m);
            signalDb(m, RmNodes.Online, false);
            return "VaryOff: Nodepool " + id + " - Unresponsive machine, marked offline: " + node;
        }

        offlineMachines.put(m.key(), m);
        disable(m, false);                    // Remove just pre-emptable work
        signalDb(m, RmNodes.Online, false);
        return "VaryOff: " + node + " - OK.";
    }

    String varyoff(String node)
    {
        // note, vaguely trickier than 'nodeLeaves' because we need to catch the
        // potential user confusions and reflect them back.
        NodePool np = findNodepoolByNodename(node);
        if ( np == null ) {
            return "VaryOff: Nodepool " + id + " - Cannot find machine: " + node;
        }

        // note we only call this if we know for sure the node can be found and associated with a NP
        return np.doVaryOff(node);   // must direct to the correct context
    }

    private String doVaryOn(String node)
    {

    	// caller must insure node is known to "me"
        Machine m = machinesByName.get(node);
        Node key = m.key();

        if ( ! offlineMachines.containsKey(key) ) {
            return "VaryOn: Nodepool " + id + " - Already online: " + m.getId();
        }
        
        offlineMachines.remove(key);
        signalDb(m, RmNodes.Online, true);

        return "VaryOn: Nodepool " + id + " - Machine marked online: " + node;
    }

    /**
     * We're going to just take it off the offline list and if it happens to come back, fine, it will get picked up
     * in nodeArrives as a new machine.
     */
    String varyon(String node)
    {        
        NodePool np = findNodepoolByNodename(node);
        if ( np == null ) {
            return "VaryOff: Nodepool " + id + " - Cannot find machine: " + node;
        }

        return np.doVaryOn(node);         // must pass to the right nodepool, can't do it "here" 
    }

    boolean isSchedulable(Machine m)
    {
        if ( m.isBlacklisted() )                         return false;
        if ( unresponsiveMachines.containsKey(m.key()) ) return false;
        if ( offlineMachines.containsKey(m.key()) )      return false;

        return true;
    }

    /**
     * ------------------------------------------------------------------------------------------
     * Routines used during the counting phase
     * ------------------------------------------------------------------------------------------
     */

    /**
     * A quick check to see if there are any machines of the right size. We make a more
     * comprehensive check to see if they're usable in countFreeableMachines later. We do this
     * so we can try to return an accurate reason for deferral.
     */
    int countReservables(IRmJob j)
    {
        int order = j.getShareOrder();
        if ( ! machinesByOrder.containsKey(order) ) return 0;
        return machinesByOrder.get(order).size();
    }

    /**
     * Count total physical machines that could accomodate a 'fixed' request that the job
     * will fit in.
     */
    int countFixable(IRmJob j)
    {
        int order = j.getShareOrder();
        int ret = 0;
        for ( int i = order; i < getMaxOrder(); i++ ) {
            if ( machinesByOrder.containsKey(order) ) {
                ret += machinesByOrder.get(order).size();
            }
        }
        return ret;
    }

    
    /**
     * Adjust counts for something that takes full machines, like a reservation.
     * If "enforce" is set the machine order must match, otherwise we just do best effort to match.
     *
     * This is intended for use by reservations only; as such it does NOT recurse into child nodepools.
     *
     * We save some trouble for later by remembering which machines we counted - we wouldn't be 
     * counting them if we didn't know FOR SURE at this point that we need them.
     * Sort on least eviction cost to get the cheapest set of preemptables.
     *
     * @returns number of machines given
     *          and updates the table of preemptables
     */
    int countFreeableMachines(IRmJob j, int needed)
    {
        String methodName = "countFreeableMachines";

        logger.info(methodName, j.getId(), "Enter nodepool", id, "preemptables.size() =", preemptables.size());
        int order = j.getShareOrder();

        ArrayList<Machine>  machs = new ArrayList<Machine>();
        if ( machinesByOrder.containsKey(order) ) {
            machs.addAll(machinesByOrder.get(order).values());            // candidates
        } else {
            return 0;                                                     // no candidates
        }

        StringBuffer sb = new StringBuffer("Machines to search:");
        for ( Machine m : machs ) {
            sb.append(" ");
            sb.append(m.getId());
        }
        logger.info(methodName, j.getId(), sb.toString());

        Collections.sort(machs, new MachineByAscendingEvictionCostSorter());

        int given = 0;           // total to give, free or freeable
        Iterator<Machine> iter = machs.iterator();
        ArrayList<Machine> pables = new ArrayList<Machine>();
        
        while ( iter.hasNext() && (given < needed) ) {
            Machine m = iter.next();
            logger.info(methodName, j.getId(), "Examining", m.getId());
            if ( !isSchedulable(m) ) {
              logger.info(methodName, j.getId(), "Bypass because machine", m.getId(), "is offline or unresponsive or blacklisted");
              continue;
            }
            if ( preemptables.containsKey(m.key()) ) {         // already counted, don't count twice
                logger.info(methodName, j.getId(), "Bypass because machine", m.getId(), "already counted.");
                continue;
            }

            if ( m.isFree() ) {
                logger.info(methodName, j.getId(), "Giving", m.getId(), "because it is free");
                given++;
                continue;
            }

            if ( m.isFreeable() ) {
                logger.info(methodName, j.getId(), "Giving", m.getId(), "because it is freeable");
                given++;
                pables.add(m);
            } else {
                logger.info(methodName, j.getId(), "Bypass because machine", m.getId(), "is not freeable");
            }
        }

        // Remember how many full machines we need to free up when we get to preemption stage.

        for ( Machine m : pables ) {
            logger.info(methodName, j.getId(), "Setting up", m.getId(), "for eviction");
            preemptables.put(m.key(), m);
            nMachinesByOrder[m.getShareOrder()]--;
        }

        calcNSharesByOrder();
        return given;
    }

    /**
     * @param nrequested is number of N shares to remove
     * @param order is the order that is affected
     */
    int countOutNSharesByOrder(int order, int nrequested)
    {
        int given = 0;                                         // track count given, for recursion

        int rem = 0;
        int low = order;

        while ( (given < nrequested ) && ( low <= getMaxOrder() ) ) {

            int avail = vMachinesByOrder[low] + nMachinesByOrder[low];
            if ( avail > 0 ) {
                if (vMachinesByOrder[low] > 0 ) {
                    vMachinesByOrder[low]--;
                } else {
                    nMachinesByOrder[low]--;
                }
                
                given++;
                rem = low - order;
                if ( rem > 0 ) {
                    vMachinesByOrder[rem]++;
                    low = Math.max(rem, order);
                }

            } else {
                low++;
            }
        }

        // oops, I can't do this myself, make a child do it.
        int k = nrequested - given;            // the number of shares we need
        if ( k > 0 ) {
            Iterator<NodePool> iter = children.values().iterator();
            while ( iter.hasNext() && ( k > 0 ) ) {
                NodePool np  = iter.next();
                given       += np.countOutNSharesByOrder(order, k);
                k            = nrequested - given;
            }
        }

        calcNSharesByOrder();

        return given;
    }

    /********************************************************************************************
     *
     * Routines used in the 'what-of' phase.
     *
     * All the counting is done - we have to reset all the counts before starting to call any of these.
     *
     *******************************************************************************************/

    /**
     * We need to make enough space for 'cnt' full machines.
     *
     * Returns number of machines that are freeable, up to 'needed', or 0, if we can't get enough.
     * If we return 0, we must defer the reservation.
     */
    protected int setupPreemptions(int needed, int order)
    {
        String methodName = "setupPreemptions";
        int given = 0;

        Iterator<Machine> iter = preemptables.values().iterator();

        while ( iter.hasNext() && (given < needed) ) {
            Machine m = iter.next();
            int o = m.getShareOrder();
            if ( order != o ) {
                continue;
            }
            logger.info(methodName, null, "Clearing", m.getId(), "from preemptable list for reservations.");
            HashMap<Share, Share> shares = m.getActiveShares();
            for ( Share s : shares.values() ) {
                if ( s.isPreemptable() ) {
                    IRmJob j = s.getJob();
                    j.shrinkByOne(s);
                    nPendingByOrder[order]++;
                } else {
                    // if the share was evicted or purged we don't care.  otherwise, it SHOULD be evictable so we
                    // log its state to try to figure out why it didn't evict
                    if ( ! (s.isEvicted() || s.isPurged() ) ) {
                        IRmJob j = s.getJob();                    
                        logger.warn(methodName, j.getId(), "Found non-preemptable share", s.getId(), "fixed:", s.isFixed(), 
                                    "j.NShares", j.countNShares(), "j.NSharesGiven", j.countNSharesGiven());
                    }
                }
            }
            given++;
            iter.remove();
        }
       
        return given;
    }

    /**
     * Here we have to dig around and find either fully free machines, or machines that we
     * can preempt to fully free it.
     */
    void  findMachines(IRmJob job, ResourceClass rc)
    {
    	String methodName = "findMachines";        

        int order = job.getShareOrder();

        int counted = job.countNSharesGiven();      // allotment from the counter
        int current = job.countNShares();           // currently allocated, plus pending, less those removed by earlier preemption
        int needed = (counted - current);

        logger.info(methodName, job.getId(), "counted", counted, "current", current, "needed", needed, "order", order);
        if ( needed <= 0 ) return;

        int cnt = countFreeMachines(order);
        if ( cnt < needed ) {
            // Get the preemptions started
            logger.info(methodName, job.getId(), "Setup preemptions.  Have", cnt, "free machines, needed", needed);
            setupPreemptions(needed-cnt, order); 
        }

        // something awful happened if we throw here.
        if ( ! machinesByOrder.containsKey(order) ) {       // hosed if this happens
            throw new SchedInternalError(job.getId(), "Scheduling counts are wrong - machinesByOrder does not match nMachinesByOrder");
        }
        
        // Since all are the same size and only empty ones are considered, no need to sort
        //machs = sortedForReservation(machinesByOrder.get(order));
        
        for ( Machine mm : machinesByOrder.get(order).values() ) {
            if ( isSchedulable(mm) && mm.isFree() ) {
                Share s = new Share(mm, job, mm.getShareOrder());
                s.setFixed();
                connectShare(s, mm, job, mm.getShareOrder());
                if ( --needed == 0 ) break;
            }
        }

    }

    /**
     * All the jobs passed in here are assigned to this nodepool.  In the case of the global nodepool
     * they're assigned implicitly; all others are assigned explicitly.
     *
     * The only tricky bit here is that there may be reservations waiting for us to clear out a full
     * machine, in which case we *might* have to over-preempt.  The doAdditions code is expected to
     * notice this and compensate by adding new allocations on other nodes.  Counting is expected
     * to guarantee that other nodes will exist, but it may take a preemption cycle or to go clear them.
     *
     * For the most part this should be pretty stable though.
     */

    /**
     * Shares come in sorted by largest first.  We iterate looking for a combination of shares that
     * leaves space of size 'order' free on the machine, using the fewest number of shares evacuated.
     */
    ArrayList<Share> evacuateLargest(int order, ArrayList<Share> shares)
    {
        int found_order = 0;

        // special case to avoid running off the end of the list
        if ( shares.size() == 1 ) {                              // terminate recursion at last share
            // Need to recheck if it's preemptable - if the job has had other preemptions then
            // we need to avoid over-preemptiong.
            Share s = shares.get(0);
            if ( s.isPreemptable() && (s.getShareOrder() == order) ) { 
                return shares;                                   // with success
            }
            return null;                                         // or with failure
        }

        ArrayList<Share> slist = new ArrayList<Share>();
        for ( Share s : shares ) {
            found_order = s.getShareOrder();
            if ( s.isPreemptable() && (found_order == order) ) {  // exact match, end recursion
                slist.add(s);
                return slist;
            }

            int new_order = order - found_order;                 // now looking for next order after removing size of what we just fond
            @SuppressWarnings("unchecked")
			ArrayList<Share> new_shares = (ArrayList<Share>) shares.clone();        //  ... and after removing the share we just found without destroying
            new_shares.remove(0);                                //      the incoming list

            ArrayList<Share> found_shares =  evacuateLargest(new_order, new_shares);
            if ( s.isPreemptable() && (found_shares != null) ) {                        // ... else we just advance to the next and try the search again
                slist.add(s);                                    // making progress, end recursion
                slist.addAll(found_shares);
                return slist;
            }
        }
        return null;                                            // found nothing, heck
    }

    private void doEvictions(int[] neededByOrder, HashMap<Integer, HashMap<IRmJob, IRmJob>> candidates, boolean force)
    {

        for ( int nbo = getMaxOrder(); nbo > 0; nbo-- ) {

            if ( neededByOrder[nbo] == 0 ) {                                  // these are N-shares
                continue;
            }
            for ( int oo = getMaxOrder(); oo > 0; oo-- ) {
                HashMap<IRmJob, IRmJob> jobs = candidates.get(oo);
                if ( jobs == null ) {
                    continue;
                }

                Iterator<IRmJob> iter = jobs.values().iterator();             // he has something to give.  is it enough?
                while ( iter.hasNext() && (neededByOrder[nbo] > 0) ) {
                    IRmJob j = iter.next();
                    int loss = 0;

                    switch ( evictionPolicy ) {
                        case SHRINK_BY_MACHINE:
                            // minimize fragmentation
                            loss = j.shrinkByOrderByMachine(neededByOrder[nbo], nbo, force, this); // pass in number of N-shares of given order that we want
                                                                                                       // returns number of quantum shares it had to relinquish 
                            break;
                        case SHRINK_BY_INVESTMENT: 
                            // minimize lost work
                            loss = j.shrinkByInvestment(neededByOrder[nbo], nbo, force, this);    // pass in number of N-shares of given order that we want
                                                                                                       // returns number of quantum shares it had to relinquish 
                            break;
                    }

                    neededByOrder[nbo]   -= loss;
                    neededByOrder[0]     -= loss;
                    nPendingByOrder[oo]  += loss;

                    if ( j.countNShares() == 0 ) {                            // nothing left? don't look here any more
                        iter.remove();
                    }
                }

            }
        }
    }

    /**
     * Here we tell the NP how much we need cleared up.  It will look around and try to do that.
     * @deprecated No longer used, the doEvictions code in NodepoolScheduler handles evictions by itself.
     *             Keeping this for a while for reference.  UIMA-4275
     */
    void doEvictionsByMachine(int [] neededByOrder, boolean force)
    {
    	String methodName = "doEvictions";
        //
        // Collect losers that are also squatters, by order, and try them first
        //
        String type;
        type = force ? "forced" : "natural";

        logger.debug(methodName, null, getId(),  "NeededByOrder", type, "on entrance eviction", Arrays.toString(neededByOrder));

        for ( NodePool np : getChildrenDescending() ) {
            logger.info(methodName, null, "Recurse to", np.getId(), "from", getId(), "force:", force);
            np.doEvictionsByMachine(neededByOrder, force);
            logger.info(methodName, null, "Recurse from", np.getId(), "proceed with logic for", getId(), "force", force);
        }

        // 
        // Adjust neededByOrder to reflect the number of shares that need to be preempted by subtracting the
        // number of shares that already are free
        //
        for ( int nbo = getMaxOrder(); nbo > 0; nbo-- ) {
            // UIMA-4065 - I think that subtracting countPendingSharesByOrder() amounts to double counting because it
            //             will reflect any evictions from the depth-first recursion.  Instead, we would subtract only
            //             our own shares.
            //
            // int needed = Math.max(0, neededByOrder[nbo] - countNSharesByOrder(nbo) - countPendingSharesByOrder(nbo)); 
            int needed = Math.max(0, neededByOrder[nbo] - countNSharesByOrder(nbo) - nPendingByOrder[nbo]);
            neededByOrder[nbo] = needed;
            neededByOrder[0] += needed;
        }

        logger.debug(methodName, null, getId(),  "NeededByOrder", type, "after adjustments for pending eviction:", Arrays.toString(neededByOrder));

        HashMap<Integer, HashMap<IRmJob, IRmJob>> squatters = new HashMap<Integer, HashMap<IRmJob, IRmJob>>();
        HashMap<Integer, HashMap<IRmJob, IRmJob>> residents = new HashMap<Integer, HashMap<IRmJob, IRmJob>>();

        for ( Share s : allShares.values() ) {
            HashMap<Integer, HashMap<IRmJob, IRmJob>> map = null;
            boolean is_candidate = force ? s.isForceable() : s.isPreemptable();
            if ( is_candidate ) {
                IRmJob j = s.getJob();
                ResourceClass rc = j.getResourceClass();
                if ( rc.getNodepoolName().equals(id) ) {
                    map = residents;
                } else {
                    map = squatters;
                }

                int order = j.getShareOrder();
                HashMap<IRmJob, IRmJob> jmap = null;
                if ( map.containsKey(order) ) {
                    jmap = map.get(order);
                } else {
                    jmap = new HashMap<IRmJob, IRmJob>();
                    map.put(order, jmap);
                }
                jmap.put(j, j);
            }
        }

        doEvictions(neededByOrder, squatters, force);
        logger.debug(methodName, null, getId(), "NeededByOrder", type, "after eviction of squatters:", Arrays.toString(neededByOrder));
        if ( neededByOrder[0] <= 0 )  {
            return;
        }

        doEvictions(neededByOrder, residents, force);
        logger.debug(methodName, null, getId(), "NeededByOrder", type, "after eviction of residents:", Arrays.toString(neededByOrder));
    }


    // For FIXED: find shares ith caps, disallow vertical stacking.  UIMA-4712
    int findSharesHorizontal( IRmJob j )
    {
        return findShares(j, true, false);
    }

    // For FIXED: find shares ith caps, do allow vertical stacking.  UIMA-4712
    int findSharesVertical( IRmJob j )
    {
        return findShares(j, true, true);
    }

    // For FAIR_SHRE: find shares, caller controls caps, allow vertical stacking.  UIMA-4712
    int findShares( IRmJob j, boolean honorCaps ) 
    {
        return findShares(j, honorCaps, true);
    }

    int findShares( IRmJob j, boolean honorCaps, boolean allowVertical )  // UIMA-4712, allowVertical
    {
        String methodName = "findShares";

        int counted = j.countNSharesGiven();      // allotment from the counter
        int current = j.countNShares();           // currently allocated, plus pending, less those removed by earlier preemption
        int needed = (counted - current);
        int order = j.getShareOrder();
        int given = 0;        
        boolean expansionStopped = false;         // UIMA-4275

        logger.debug(methodName, j.getId(), "counted", counted, "current", current, "needed", needed, "order", order, "given", given);

        if ( needed > 0 ) {
            whatof: {
                for ( int i = order; i < getArraySize(); i++ ) {
                    if ( nSharesByOrder[i] == 0 ) {
                        continue;                                            // nothing here to give
                    }
                    
                    Map<Node, Machine> machs = getVirtualMachinesByOrder(i);
                    ArrayList<Machine> ml = new ArrayList<Machine>();
                    ml.addAll(machs.values());
                    
                    for ( Machine m : ml ) {                                // look for space
                        if ( !isSchedulable(m) ) continue;                  // nope
                        if ( (!allowVertical) && (m.hasVerticalConflict(j)) ) continue;  // UIMA-4712
                        int g = Math.min(needed, m.countFreeShares(order)); // adjust by the order supported on the machine
                        for ( int ndx= 0;  ndx < g; ndx++ ) {
                            if ( honorCaps && j.exceedsFairShareCap() ) {                // UIMA-4275
                                // can't take any more shares, probably because of caps
                                expansionStopped = true;
                                break whatof;
                            } else {
                                Share s = new Share(m, j, order);
                                connectShare(s, m, j, order);
                                logger.info(methodName, j.getId(), "Connecting new share", s.toString());
                                //j.assignShare(s);
                                //m.assignShare(s);
                                //rearrangeVirtual(m, order);
                                //allShares.put(s, s);
                            }
                        }
                        
                        given += g;
                        needed -= g;
                        if ( needed == 0 ) {
                            break whatof;
                        }
                    }
                }
            }

            //calcNSharesByOrder();
        }

        if ( (needed > 0) && ( !expansionStopped ) && ( j.getSchedulingPolicy() == Policy.FAIR_SHARE) ) {            // UIMA-4275
            for ( NodePool np : getChildrenAscending() ) {

                StringBuffer sb = new StringBuffer();
                for ( NodePool sp : getChildrenAscending() ) {
                    sb.append(sp.getId());
                    sb.append(" ");
                }
                logger.info(methodName, null, np.getId(), "Doing expansions in this order:", sb.toString());

                int g = np.findShares(j, honorCaps, allowVertical);
                given += g;
                needed -= g;

                if ( needed == 0 ) {
                    break;
                }
            }
        }

        return given;
    }

    /**
     * Bop through the jobs, and if their current counts exceed their current assignment, find
     * something to give them.
     *
     * It's possible that a job had evictions as a result of clearing space for a reservation - 
     * so we need to check the count of allocated shares, the count of pending removals, and the
     * current share count assignment.
     */
    HashMap<IRmJob, IRmJob>  doExpansion(List<IRmJob> jobs)
    {
        String methodName = "doExpansion";
        HashMap<IRmJob, IRmJob> expansions = new HashMap<IRmJob, IRmJob>();

        StringBuffer sb = new StringBuffer();
        sb.append("NP: ");
        sb.append(getId());
        sb.append(" Expansions in this order: ");
        for ( IRmJob j : jobs ) {
            if ( j.isCompleted() ) continue;  // deal with races while job is completing

            j.undefer();
            sb.append(j.getId());
            sb.append(":");
            if ( findShares(j, false) > 0 ) {   // always fair-share, so don't do caps yet. UIMA-4712
                sb.append("found ");
                expansions.put(j, j);
            } else {
                sb.append("notfound ");
            }
            if ( j.countNShares() == 0 ) j.setReason("Waiting for preemptions.");
        }
        logger.info(methodName, null, sb.toString());
        return expansions;
    }

    /**
     * This prints garbage unless you call reset() first.
     */
    public String toString()
    {
        StringBuffer sb = new StringBuffer();
        sb.append("--------------------------------------------------------------------------------\n");
        sb.append("Nodepool ");
        sb.append(id);
        sb.append(" depth ");
        sb.append(depth);
        sb.append(": ");
        sb.append("\n");

        //
        // Print the key tables.  First the header ...
        //
        int len = nMachinesByOrder.length;
        StringBuffer sbsb = new StringBuffer("%18s ");
        for ( int i = 0; i < len; i++ ) {
            sbsb.append("%4s ");
        }
        sbsb.append("\n");
        String fmt = sbsb.toString();
        Object[] vals = new Object[len + 2];
        vals[0] = "Order";
        for ( int i = 0; i < len; i++ ) {
            vals[i+1] = Integer.toString(i);
        }

        sb.append(String.format(fmt, vals));

        // Now nMachinesByorder

        sbsb = new StringBuffer("%18s ");
        for ( int i = 0; i < len; i++ ) {
            sbsb.append("%4d ");
        }
        sbsb.append("\n");
        fmt = sbsb.toString();

        vals[0] = "nMachinesByOrder";
        int[] counts = countMachinesByOrder();
        for ( int i = 0; i < len; i++ ) {
            vals[i+1] = counts[i];
        }
        sb.append(String.format(fmt, vals));

        vals[0] = "vMachinesByOrder";
        counts = countVMachinesByOrder();
        for ( int i = 0; i < len; i++ ) {
            vals[i+1] = counts[i];
        }
        sb.append(String.format(fmt, vals));

        // Now nSharesByorder

        vals[0] = "nSharesByOrder";
        counts = countAllNSharesByOrder();
        for ( int i = 0; i < len; i++ ) {
            vals[i+1] = counts[i];
        }
        sb.append(String.format(fmt, vals));

        // Now nFreeByorder

//        vals[0] = "nFreeSharesByOrder";
//        counts = countFreeSharesByOrder();
//        for ( int i = 0; i < len; i++ ) {
//            vals[i+1] = counts[i];
//        }
//        sb.append(String.format(fmt, vals));
       
        sb.append("--------------------------------------------------------------------------------\n");
        
        for ( NodePool np: children.values () ) {
            sb.append(np.toString());
        }
        
        return sb.toString();
    }

    public void queryMachines()
    {
    	String methodName = "queryMachines";
        ArrayList<Machine> machines = new ArrayList<Machine>();
        machines.addAll(getAllMachines().values());
        logger.info(methodName, null, "================================== Query Machines Nodepool:", id, "=========================");
        StringBuffer buf = new StringBuffer();
        buf.append(Machine.getHeader());
        buf.append("\n");
        buf.append(Machine.getDashes());            
        buf.append("\n"); 
        Collections.sort(machines, new MachineByOrderSorter());
        
        for ( Machine m : machines) {
            buf.append(m.toString());
            int remaining = m.countFreeShares();
            if ( remaining > 0 ) {
                buf.append("[" + m.countFreeShares() + "]");
            }
            buf.append("\n");
        }
        logger.info(methodName, null, "\n", buf.toString());
        logger.info(methodName, null, "================================== End Query Machines Nodepool:", id, "======================");
    }

    //
    // Order shares by INCREASING preemption cost (all free followed by those with least eviction cost)
    // Don't need to check for unschedulable or un-freeable as they will be ignored later.
    //
    class MachineByAscendingEvictionCostSorter implements Comparator<Machine> {
        public int compare(Machine m1, Machine m2) {
            if (m1.equals(m2))
                return 0;

            if (m1.isFree()) {
                if (m2.isFree())
                    return 0;
                else
                    return -1; // m2 not free, m1 to the front of the list
            } else if (m2.isFree())
                return 1;      // m1 not free, m2 to the front of the list

            // Sort the lowest eviction cost first
            // Since totals are the same, most free shares ==> smallest eviction cost
            switch (evictionPolicy) {
                case SHRINK_BY_MACHINE :
                    return m2.countFreeShares() - m1.countFreeShares();
                case SHRINK_BY_INVESTMENT :
                    return m1.getInvestment() - m2.getInvestment();
                default:
                    return 0;
            }
        }
    }

    static private class NodepoolAscendingSorter
        implements Comparator<NodePool>
    {
        public int compare(NodePool n1, NodePool n2)
        {
            return (n1.getSearchOrder() - n2.getSearchOrder());
        }
    }

    static private class NodepoolDescendingSorter
        implements Comparator<NodePool>
    {
        public int compare(NodePool n1, NodePool n2)
        {
            return (n2.getSearchOrder() - n1.getSearchOrder());
        }
    }

    class InvestmentSorter
    	implements Comparator<Share>
    {	
    	public int compare(Share s1, Share s2)
        {
            return (int) (s1.getInvestment() - s2.getInvestment());           // lowest investment
                                                                      // if we're not tracking investment we
                                                                      // don't care anyway so this works fine
        }
    }

    class DescendingShareOrderSorter
    	implements Comparator<Share>
    {	
    	public int compare(Share s1, Share s2)
        {
            return (int) (s2.getShareOrder() - s1.getShareOrder());
        }
    }


    class MachineByOrderSorter
    	implements Comparator<Machine>
    {	
    	public int compare(Machine m1, Machine m2)
        {
            return m2.getShareOrder() - m1.getShareOrder();
        }
    }

    class MachineByAscendingOrderSorter
    	implements Comparator<Machine>
    {	
    	public int compare(Machine m1, Machine m2)
        {
            return m1.getShareOrder() - m2.getShareOrder();
        }
    }

    class GlobalOrder
    {
        int maxorder = 0;

        GlobalOrder()
        {
            this.maxorder = 0;
        }

        synchronized void reset()
        {
            this.maxorder = 0;
        }

        synchronized void update(int order)
        {
            this.maxorder = Math.max(maxorder, order);
        }

        synchronized int getOrder()
        {
            return maxorder;
        }
    }
}