/*
 * 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.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.NodeConfiguration;
import org.apache.uima.ducc.common.NodeIdentity;
import org.apache.uima.ducc.common.Pair;
import org.apache.uima.ducc.common.admin.event.RmAdminQLoadReply;
import org.apache.uima.ducc.common.admin.event.RmAdminQOccupancyReply;
import org.apache.uima.ducc.common.admin.event.RmAdminReply;
import org.apache.uima.ducc.common.admin.event.RmAdminVaryReply;
import org.apache.uima.ducc.common.admin.event.RmQueriedClass;
import org.apache.uima.ducc.common.admin.event.RmQueriedMachine;
import org.apache.uima.ducc.common.admin.event.RmQueriedNodepool;
import org.apache.uima.ducc.common.component.AbstractDuccComponent;
import org.apache.uima.ducc.common.persistence.rm.IRmPersistence;
import org.apache.uima.ducc.common.persistence.rm.RmPersistenceFactory;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.DuccProperties;
import org.apache.uima.ducc.common.utils.DuccPropertiesResolver;
import org.apache.uima.ducc.common.utils.SystemPropertyResolver;
import org.apache.uima.ducc.common.utils.Version;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.common.utils.id.DuccIdFactory;


/**
 * This process orchestrates scheduling.
 * - Receives requests from clients ( job manager, service manager, etc ) for resources
 * - Forwards requests and current state to pluggable scheduling implementation
 * - Receives a schedule, updates state, sends responses to requestors
 * - Maintains state as needed (work item life cycle etc)
 */
public class Scheduler
//    extends Thread
    implements ISchedulerMain,
    	SchedConstants
{
    IJobManager jobManager;
    static DuccLogger     logger = DuccLogger.getLogger(Scheduler.class, COMPONENT_NAME);

    boolean done = false;
    // Boolean force_epoch = false;
    String ducc_home;
    // Integer epoch = 5;                                                 // scheduling epoch, seconds

    NodeConfiguration configuration = null;                               // UIMA-4142 make it global
    
    String defaultDomain = null;                                          // UIMA-4142
    boolean needRecovery = false;                                         // UIMA-4142 tell outer layer that recovery is required
    AbstractDuccComponent baseComponent;                                  // UIMA-4142, pass in the base for reconfig - reread ducc.properties
    NodePool[] nodepools;                                                 // top-level nodepools
    int max_order = 0;

    //
    // Fair-share and fixed-share use shares only, not machines
    //
    Map<DuccId, Share> busyShares        = new HashMap<DuccId, Share>(); // Running "fair" share jobs

    // incoming reports of machines that are now free
    Map<DuccId, Pair<IRmJob, Share>> vacatedShares= new HashMap<DuccId, Pair<IRmJob, Share>>();
    // boolean growthOccurred = false;                                           // don't care which grew, just that something grew

    List<IRmJob>        incomingJobs    = new ArrayList<IRmJob>();       // coming in from external world but not added our queues yet
    List<IRmJob>        recoveredJobs   = new ArrayList<IRmJob>();       // coming in from external world but we don't now about them, (hopefully
                                                                         //    because we crashed and not for more nefarious reasons)
    List<IRmJob>        completedJobs   = new ArrayList<IRmJob>();       // signaled complete from outside but not yet dealt with
    List<IRmJob>        initializedJobs = new ArrayList<IRmJob>();       // Init is complete so we can begin full (un)fair share allocation

    //HashMap<Node, Node> incomingNodes  = new HashMap<Node, Node>();         // node updates
    Map<Node, Node> deadNodes      = new HashMap<Node, Node>();           // missed too many heartbeats
    Map<Node, Integer> illNodes    = new HashMap<Node, Integer>();        // starting to miss, keep track of how many for the db
    // HashMap<Node, Node> allNodes       = new HashMap<Node, Node>();           // the guys we know
    Map<String, NodePool>    nodepoolsByNode = new HashMap<String, NodePool>(); // all nodes, and their associated pool
    Map<String, String>      shortToLongNode = new HashMap<String, String>();   // 

    Map<String, User>    users     = new HashMap<String, User>();         // Active users - has a job in the system
    //HashMap<DuccId, IRmJob>    runningJobs = new HashMap<DuccId, IRmJob>();

    Map<DuccId, IRmJob>  allJobs = new HashMap<DuccId, IRmJob>();

    Map<ResourceClass, ResourceClass> resourceClasses = new HashMap<ResourceClass, ResourceClass>();
    Map<String, ResourceClass> resourceClassesByName = new HashMap<String, ResourceClass>();

    String defaultFairShareName = null;
    String defaultReserveName = null;

    int defaultNThreads = 1;
    int defaultNTasks = 10;
    int defaultMemory = 15;

    // these two are initialized in constructor
    String schedImplName;
    IScheduler[] schedulers;

    long share_free_dram  = 0;               // 0  GB in KB  - minim memory after shares are allocated
    long dramOverride     = 0;               // if > 0, use this instead of amount reported by agents (modeling and testing)

    int pending_evictions = 0;                    // for queries
    int pending_expansions = 0;                  // for queries

    EvictionPolicy evictionPolicy = EvictionPolicy.SHRINK_BY_MACHINE;

//     int nodeMetricsUpdateRate = 30000;
//     int startupCountdown = 0;       // update each epoch.  only schedule when it's > nodeStability
    int nodeStability = 3;
    boolean stability = false;

    private static DuccIdFactory idFactory;
    IRmPersistence persistence = null;

    // static boolean expandByDoubling = true;
    // static int initializationCap = 2;      // Max allocation until we know initialization works in
                                           // units of *processes*, not shares (i.e.N-shares).

    //
    // Version
    //    0 - major version
    //    6 - minor version
    //    3 - ptf - forced eviction under fragmentation.
    //    4 - defrag code complete
    //  beta - not yet "real"!
    //
    // Bring up to speed with rest of ducc version. 2013-03-06 jrc
    //
    // 1.0.1 - RM can purge non-preemptables except for Unmanaged Reservations.  UIMA-3614
    // 1.0.2 - vary-on, vary-off
    // 1.0.3 - fix bad check in recursion in NodepoolScheduler.doEvictions
    // 1.1.0 - Syncnronize with release
    final static int rmversion_major = 2;
    final static int rmversion_minor = 0;
    final static int rmversion_ptf   = 0;  
    final static String rmversion_string = null;

    boolean initialized = false;           // we refuse nodeupdates until this is true
    public Scheduler(AbstractDuccComponent baseComponent)
    {
        this.baseComponent = baseComponent;                // UIMA-4142, pass in the base for reconfig
    }

    public synchronized void init()
        throws Exception
    {
        String methodName = "init";
        //setName("Scheduler");

        DuccLogger.setUnthreaded();

        String ep         = SystemPropertyResolver.getStringProperty("ducc.rm.eviction.policy", "SHRINK_BY_MACHINE");
        evictionPolicy    = EvictionPolicy.valueOf(ep);        

        // nodepool          = new NodePool(null, evictionPolicy, 0);   // global nodepool
        share_free_dram   = SystemPropertyResolver.getLongProperty("ducc.rm.reserved.dram", share_free_dram) * 1024 * 1024;   // GB -> KB
        ducc_home         = SystemPropertyResolver.getStringProperty("DUCC_HOME");

        // some defaults, for jobs that don't specify them
        defaultNTasks     = SystemPropertyResolver.getIntProperty("ducc.rm.default.tasks", 10); 
        defaultNThreads   = SystemPropertyResolver.getIntProperty("ducc.rm.default.threads", 1);
        defaultMemory     = SystemPropertyResolver.getIntProperty("ducc.rm.default.memory", 15);      // in GB
        // expandByDoubling  = RmUtil.getBooleanProperty("ducc.rm.expand.by.doubling", true);

        nodeStability     = SystemPropertyResolver.getIntProperty("ducc.rm.node.stability", 3);        // number of node metrics updates to wait for before scheduling
                                                                                  // 0 means, just jump right in and don't wait

        dramOverride = SystemPropertyResolver.getLongProperty("ducc.rm.override.dram", 0);
        if ( dramOverride > 0 ) {
            dramOverride = dramOverride * (1024 * 1024);         // convert to KB
        }

        if ( idFactory == null ) {                               // UIMA-4142 only remake it on first boot
            idFactory = new DuccIdFactory(1);
        }

//        try {
//            schedImplName = SystemPropertyResolver.getStringProperty("ducc.rm.scheduler", "org.apache.uima.ducc.rm.ClassBasedScheduler");
//            @SuppressWarnings("unchecked")
//			Class<IScheduler> cl = (Class<IScheduler>) Class.forName(schedImplName);
//            scheduler = (IScheduler) cl.newInstance();
//        } catch (ClassNotFoundException e) {
//            throw new SchedulingException(null, "Cannot find class " + schedImplName);
//        } catch (InstantiationException e) {
//            throw new SchedulingException(null, "Cannot instantiate class " + schedImplName);            
//        } catch (IllegalAccessException e) {
//            throw new SchedulingException(null, "Cannot instantiate class " + schedImplName + ": can't access constructor.");            
//        }

        String class_definitions = SystemPropertyResolver
            .getStringProperty(DuccPropertiesResolver
                               .ducc_rm_class_definitions, "scheduler.classes");
        class_definitions = System.getProperty("DUCC_HOME") + "/resources/" + class_definitions;

        try {
            initClasses();
        } catch ( Exception e ) {
            logger.error(methodName, null, e);
            throw e;
        }

        // we share most of the state with the actual scheduling code - no need to keep passing this around
        // TODO: Make sure these are all Sialized correctly
//         scheduler.setEvictionPolicy(evictionPolicy);
//         scheduler.setClasses(resourceClasses);
//         scheduler.setNodePool(nodepools[0]);

        logger.info(methodName, null, "                       reserved DRAM           : ", (share_free_dram / (1024*1024)), " GB");
        logger.info(methodName, null, "                       DRAM override           : ", (dramOverride / (1024*1024)), " GB");
        logger.info(methodName, null, "                       scheduler               : ", schedImplName);
        logger.info(methodName, null, "                       default threads         : ", defaultNThreads);
        logger.info(methodName, null, "                       default tasks           : ", defaultNTasks);
        logger.info(methodName, null, "                       default memory          : ", defaultMemory);
        logger.info(methodName, null, "                       default fairshare class : ", defaultFairShareName);
        logger.info(methodName, null, "                       default reserve         : ", defaultReserveName);
        logger.info(methodName, null, "                       class definition file   : ", class_definitions);
        logger.info(methodName, null, "                       default domain          : ", defaultDomain);      // UIMA-4142
        logger.info(methodName, null, "                       eviction policy         : ", evictionPolicy);
        logger.info(methodName, null, "                       database enabled        : ", !System.getProperty("ducc.database.host").equals("--disabled--"));
        logger.info(methodName, null, "                       database implementation : ", System.getProperty("ducc.rm.persistence.impl"));
        logger.info(methodName, null, "                       use prediction          : ", SystemPropertyResolver.getBooleanProperty("ducc.rm.prediction", true));
        logger.info(methodName, null, "                       prediction fudge factor : ", SystemPropertyResolver.getIntProperty("ducc.rm.prediction.fudge", 10000));
        logger.info(methodName, null, "                       node stability          : ", nodeStability);
        logger.info(methodName, null, "                       init stability          : ", SystemPropertyResolver.getIntProperty("ducc.rm.init.stability"));
        logger.info(methodName, null, "                       fast recovery           : ", SystemPropertyResolver.getBooleanProperty("ducc.rm.fast.recovery", true));
        logger.info(methodName, null, "                       metrics update rate     : ", SystemPropertyResolver.getIntProperty("ducc.agent.node.metrics.publish.rate", 
                                                                                                                                 DEFAULT_NODE_METRICS_RATE));
        logger.info(methodName, null, "                       initialization cap      : ", SystemPropertyResolver.getIntProperty("ducc.rm.initialization.cap"));
        logger.info(methodName, null, "                       expand by doubling      : ", SystemPropertyResolver.getBooleanProperty("ducc.rm.expand.by.doubling", true));
        logger.info(methodName, null, "                       fragmentation threshold : ", SystemPropertyResolver.getIntProperty("ducc.rm.fragmentation.threshold", 2));
        logger.info(methodName, null, "                       do defragmentation      : ", SystemPropertyResolver.getBooleanProperty("ducc.rm.defragmentation", true));
        logger.info(methodName, null, "                       DUCC home               : ", System.getProperty("DUCC_HOME"));
        logger.info(methodName, null, "                       ActiveMQ URL            : ", SystemPropertyResolver.getStringProperty("ducc.broker.url"));
        logger.info(methodName, null, "                       JVM                     : ", System.getProperty("java.vendor") +
                                                                                      " "+ System.getProperty("java.version"));
        logger.info(methodName, null, "                       JAVA_HOME               : ", System.getProperty("java.home"));
        logger.info(methodName, null, "                       JVM Path                : ", System.getProperty("ducc.jvm"));
        logger.info(methodName, null, "                       JMX URL                 : ", System.getProperty("ducc.jmx.url"));
        logger.info(methodName, null, "                       OS Architecture         : ", System.getProperty("os.arch"));
        logger.info(methodName, null, "                       OS Name                 : ", System.getProperty("os.name"));
        logger.info(methodName, null, "                       DUCC Version            : ", Version.version());
        logger.info(methodName, null, "                       RM Version              : ", ""+ rmversion_major   + "." 
                                                                                             + rmversion_minor   + "." 
                                                                                             + rmversion_ptf);
        persistence = RmPersistenceFactory.getInstance(this.getClass().getName(), "RM");
        persistence.clear();
        initialized = true;
    }

    public RmAdminReply reconfigure()          // UIMA-4142
    {
        String methodName = "reconfigure";

        RmAdminReply ret = new RmAdminReply();
        logger.info(methodName, null, "Reconfiguration starts.");

        setInitialized(false);           // stop receipt of OR and Agent publications

        // First we run the logic that reads the configuration, and if it fails, we abort the reconfig without crashing
        // We'll throw it away because we call init() in a minute, which will do the actual configuration, as if booting
		try {
			readConfiguration();                                    // UIMA-4142
		} catch (Throwable e) {
            setInitialized(true);
            logger.warn(methodName, null, "Reconfiguration aborted:", e.toString());
            ret.setRc(false);
            ret.setMessage("Reconfiguration failed: " + e.toString());
            return ret;
		}

        HashMap<Node, Machine> offlineMachines                   = new HashMap<Node, Machine>();
        for (NodePool np : nodepools) {
            offlineMachines.putAll(np.getOfflineMachines());            
        }

        // (be careful, don't use the value, that must be discarded as it points to the OLD np)
        List<String> offlineHostnames = new ArrayList<String>();
        for ( Machine m : offlineMachines.values()) {
        	logger.info(methodName, null, "Saving offline status of", m.getId());
            offlineHostnames.add(m.getId());
        }
        offlineMachines = null;

        this.configuration = null;
        this.defaultDomain = null;
        this.nodepools = null;
        this.max_order = 0;
        this.busyShares.clear();
        this.vacatedShares.clear();
        this.incomingJobs.clear();
        this.recoveredJobs.clear();
        this.initializedJobs.clear();
        this.deadNodes.clear();
        this.nodepoolsByNode.clear();
        this.shortToLongNode.clear();
        this.users.clear();
        this.allJobs.clear();
        this.resourceClasses.clear();
        this.resourceClassesByName.clear();

        try {
            baseComponent.reloadProperties("ducc.deploy.configuration");
            init();

            if ( offlineHostnames.size() > 0 ) {
                String[] offline = offlineHostnames.toArray(new String[offlineHostnames.size()]);
                varyoff(offline);
            }
        } catch ( Throwable t ) {
        	// TODO do something?  What?  If this fails its pretty awful.
        }
        
        setRecovery(true);               // signal to outer layer that full recovery is needed
        setInitialized(true);            // resume receipt of publications
        logger.info(methodName, null, "Reconfiguration complete.");

        ret.setMessage("Reconfiguration complete.");
        return ret;
    }

    public synchronized void setRecovery(boolean v)
    {
        this.needRecovery = v;
    }

    public synchronized boolean mustRecover()
    {
        return this.needRecovery;
    }

    public synchronized boolean isInitialized()
    {
        return initialized;
    }

    public synchronized void setInitialized(boolean v)
    {
        this.initialized = v;
    }

    public Machine getMachine(Node n)
    {
        return getMachine(n.getNodeIdentity());
    }

    public Machine getMachine(NodeIdentity ni)
    {
        NodePool nodepool = getNodepoolByName(ni);
    	return nodepool.getMachine(ni);        
    }

    public void setJobManager(IJobManager jobmanager)
    {
        this.jobManager = jobmanager;
    }

    public String getDefaultFairShareName()
    {
    	return defaultFairShareName;
    }

    public String getDefaultReserveName()
    {
    	return defaultReserveName;
    }

    public int getDefaultNThreads()
    {
    	return defaultNThreads;
    }

    public int getDefaultNTasks()
    {
    	return defaultNTasks;
    }

    public int getDefaultMemory()
    {
    	return defaultMemory;
    }

    public ResourceClass getResourceClass(String name)
    {
        return resourceClassesByName.get(name);
    }

    public IRmJob getJob(DuccId id)
    {
        return allJobs.get(id);
    }

    public Share getShare(DuccId id)
    {
        return busyShares.get(id);
    }

//    public static int getInitializationCap()
//    {
//        return initializationCap;
//    }
//
//    public static boolean isExpandByDoubling()
//    {
//        return expandByDoubling;
//    }

    /**
     * Calculate share order, given some memory size in GB (as in from a job spec)
     */
    public int calcShareOrder(IRmJob j)
    {
        // Calculate its share order
        long mem = j.getMemory() << 20 ;                             // to KB from GB
        int share_quantum = j.getShareQuantum();
        
        int share_order = (int) (mem / share_quantum);               // liberal calc, round UP
        if ( (mem % share_quantum) > 0 ) {
            share_order++;
        }
        return share_order;
    }


    /**
     * Collect all the classes served by the indicated nodepool (property set).  This fills
     * in the 'ret' map from the parameter 'dp' and recursive calls to the children in dp.

     * @param dp This is the properties object from the configurator for a top-level
     *            nodepool.
     * @param ret This is the map to be filled in by this routine.
     */
    void getClassesForNodepool(DuccProperties dp, Map<ResourceClass, ResourceClass> ret)
    {
        @SuppressWarnings("unchecked")
		List<DuccProperties> class_set = (List<DuccProperties>) dp.get("classes");
        if ( class_set != null ) {
            for ( DuccProperties cl : class_set ) {
                ResourceClass rc = resourceClassesByName.get(cl.getStringProperty("name"));
                ret.put(rc, rc);
            }
        }

        @SuppressWarnings("unchecked")
		List<DuccProperties> children = (List<DuccProperties>) dp.get("children");
        if ( children != null ) {
            for (DuccProperties child : children ) {
                getClassesForNodepool(child, ret);
            }
        }        
    }

    /**
     * Map each node by name into the nodepool it belongs to
     */
    void mapNodesToNodepool(Map<String, String> nodes, NodePool pool)
    {
        if ( nodes == null ) return;

        for ( String s : nodes.keySet() ) {
            updateNodepoolsByNode(s, pool);        // maps from both the fully-qualified name and th shortnmae
        }
    }

    /**
     * (Recursively) build up the heirarchy under the parent nodepool.
     */
    void createSubpools(NodePool parent, List<DuccProperties> children)
    {
        if ( children == null ) return;

        for ( DuccProperties dp : children ) {
            String id = dp.getStringProperty("name");
            @SuppressWarnings("unchecked")
			Map<String, String> nodes = (Map<String, String>) dp.get("nodes");
            int search_order = dp.getIntProperty("search-order", 100);
            NodePool child = parent.createSubpool(id, nodes, search_order);
            mapNodesToNodepool(nodes, child);

            @SuppressWarnings("unchecked")
			List<DuccProperties> grandkids = (List<DuccProperties>) dp.get("children");
            createSubpools(child, grandkids);            
        }
    }

    // UIMA-4142 better modularize this code
    NodeConfiguration readConfiguration()
        throws Exception
    {
        String class_definitions = SystemPropertyResolver
            .getStringProperty(DuccPropertiesResolver
                               .ducc_rm_class_definitions, "scheduler.classes");
        String user_registry = SystemPropertyResolver
            .getStringProperty(DuccPropertiesResolver
                               .ducc_rm_user_registry, "ducc.users");
        class_definitions = System.getProperty("DUCC_HOME") + "/resources/" + class_definitions;
        String me = Scheduler.class.getName() + ".Config";
        DuccLogger initLogger = new DuccLogger(me, COMPONENT_NAME);
        NodeConfiguration nc = new NodeConfiguration(class_definitions, null, user_registry, initLogger);        // UIMA-4142 make the config global
        nc.readConfiguration();
        return nc;                                             // UIMA-4142
    }

    // UIMA-4142, don't pass in class def file, instead use common readConfiguration
    void initClasses()
    {
    	String methodName = "initClasses";
        
		try {
            configuration = readConfiguration();
		} catch (Throwable e) {
            // RM boot.  We must abort being has we have no prior working configuration to fall back to.
            logger.error(methodName, null, e);
            logger.error(methodName, null, "Scheduler exits: unable to read configuration.");
            System.out.println("Scheduler exits: unable to read configuration.");
            e.printStackTrace();
            System.exit(1);
		}

        defaultDomain = configuration.getDefaultDomain();                   // UIMA-4142
        configuration.printConfiguration();

        DuccProperties[] nps = configuration.getToplevelNodepools();
        Map<String, DuccProperties> cls = configuration.getClasses();

        nodepools = new NodePool[nps.length];                   // top-level nodepools
        schedulers = new IScheduler[nps.length];                // a schedler for each top-level nodepool

        // Here build up the ResourceClass definitions
        logger.info(methodName, null, "Classes:");
        logger.info(methodName, null, ResourceClass.getHeader());
        logger.info(methodName, null, ResourceClass.getDashes());
        for ( DuccProperties props : cls.values() ) {
            ResourceClass rc = new ResourceClass(props);
            resourceClasses.put(rc, rc);
            resourceClassesByName.put(rc.getName(), rc);
            logger.info(methodName, null, rc.toString());
        }

        DuccProperties dc = configuration.getDefaultFairShareClass();
        if ( dc != null ) {
            defaultFairShareName = dc.getProperty("name");
        }

        dc = configuration.getDefaultReserveClass();
        if ( dc != null ) {
            defaultReserveName = dc.getProperty("name");
        }

        // Instatntiate one scheduler per top-level nodepool
        try {
            schedImplName = SystemPropertyResolver.getStringProperty("ducc.rm.scheduler", "org.apache.uima.ducc.rm.ClassBasedScheduler");
            @SuppressWarnings("unchecked")
			Class<IScheduler> cl = (Class<IScheduler>) Class.forName(schedImplName);
            for ( int i = 0; i < nps.length; i++ ) {
                logger.info(methodName, null, "Rebuilding", schedImplName, "for top level nodepool", nps[i].get("name"));
                schedulers[i] = (IScheduler) cl.newInstance();
                schedulers[i].setEvictionPolicy(evictionPolicy);
            }

        } catch (ClassNotFoundException e) {
            throw new SchedulingException(null, "Cannot find class " + schedImplName);
        } catch (InstantiationException e) {
            throw new SchedulingException(null, "Cannot instantiate class " + schedImplName);            
        } catch (IllegalAccessException e) {
            throw new SchedulingException(null, "Cannot instantiate class " + schedImplName + ": can't access constructor.");            
        }

        // Here create the nodepool configuration
        for ( int i = 0; i < nps.length; i++ ) {
            DuccProperties np = nps[i];
            String id = np.getStringProperty("name");
            @SuppressWarnings("unchecked")
			Map<String, String> nodes = (Map<String, String>) np.get("nodes");
            int search_order = np.getIntProperty("search-order", 100);
            int q = np.getIntProperty("share-quantum", 15) << 20 ;      // to kB which is how the nodes report in
            nodepools[i] = new NodePool(null, id, nodes, evictionPolicy, 0, search_order, q);
            schedulers[i].setNodePool(nodepools[i]);                    // set its top-level nodepool

            mapNodesToNodepool(nodes, nodepools[i]);
            logger.info(methodName, null, "Created top-level nodepool", id);

            @SuppressWarnings("unchecked")
			List<DuccProperties> children = (List<DuccProperties>) np.get("children");
            createSubpools(nodepools[i], children);

            Map<ResourceClass, ResourceClass> classesForNp = new HashMap<ResourceClass, ResourceClass>();
            getClassesForNodepool(np, classesForNp);           // all classes served by this heirarchy - fills in classesForNp
            for ( ResourceClass rc: classesForNp.values() ) {               // UIMA-4065 tell each cl which np serves it
                String rcid = rc.getNodepoolName();
                if ( rcid != null ) {
                    // set the two-way pointers between rc and np
                    NodePool subpool = nodepools[i].getSubpool(rcid);
                    rc.setNodepool(subpool);                  // rc -> nodepool
                    logger.info(methodName, null, "Assign rc", rc.getName(), "to np", subpool.getId());
                    subpool.addResourceClass(rc);             // nodepool -> rc
                }
            }

            schedulers[i].setClasses(classesForNp);

        }

        // Here create or update Users with constraints from the registry
        Map<String, DuccProperties> usrs = configuration.getUsers();                // UIMA-4275
        for ( Object o : usrs.keySet() ) {                     // iterate over users
            String n = (String) o;
            DuccProperties dp = usrs.get(n);
            for ( Object l : dp.keySet() ) {                  // iterate over limits for the user
                if ( !((String)l).startsWith("max-allotment")) continue; // only this supported at this time
                String val = ((String) dp.get(l)).trim();

                int lim = Integer.parseInt( val ); // verified parsable int during parsing


                User user = users.get(n);
                if (user == null) {
                    user = new User(n);
                    users.put(n, user);
                }

                if ( val.contains(".") ) {
                    String[] tmp = ((String)l).split("\\.");                // max_allotment.classname
                    ResourceClass rc = resourceClassesByName.get(tmp[1]);
                    user.overrideLimit(rc, lim);   // constrain allotment for this class to value in l
                } else {
                    user.overrideGlobalLimit(lim);
                }
            }
        }

    }

    /**
     * Called only from schedule, under the 'this' monitor.
     *
     * We then take the SchedulingUpdate from the IScheduler and dispatches orders to
     * the world to make it happen.
     *
     * For jobs that lose resources, job manager is asked to stop execution in specific shares.
     * For jobs that gain resources, job manager is asked to start execution in specific shares.
     * Jobs that don't change are leftovers.  If they're not running at all, they're in the pending
     *  list; they might also be in the running list but had no allocation changes in the current epoch.
     */
    private JobManagerUpdate dispatch(SchedulingUpdate upd, JobManagerUpdate jmu)
    {
        String methodName = "dispatch";
        HashMap<IRmJob, IRmJob> jobs;

        pending_evictions = 0;                    // for queries
        pending_expansions = 0;                  // for queries

        // Go through shrunken jobs - if they are shrunken to 0, move to dormant
        jobs = upd.getShrunkenJobs();
        for (IRmJob j : jobs.values()) {
            
            logger.trace(methodName, j.getId(), ">>>>>>>>>> SHRINK");

            HashMap<Share, Share> sharesE = j.getAssignedShares();
            HashMap<Share, Share> sharesR = j.getPendingRemoves();
            logger.trace(methodName, j.getId(), "removing", sharesR.size(), "of existing", sharesE.size(), "shares.");
            pending_evictions += (sharesR.size() * j.getShareOrder());

            for ( Share s : sharesE.values() ) {
                logger.trace(methodName, j.getId(), "    current", s.toString());
            }

            for ( Share s : sharesR.values() ) {
                logger.trace(methodName, j.getId(), "    remove ", s.toString());
            }
            logger.trace(methodName, j.getId(), ">>>>>>>>>>");

            jmu.removeShares(j, sharesR);
            // jobManager.stopJob(j, shares);                 // stops job on everything on the pendingRemoves list
            // j.clearPendingRemoves();
        }

        // Go through expanded jobs - if they are dormant, remove from dormant
        //                            then add to running.
        // Tell the server it needs to start some machines for the job
        jobs = upd.getExpandedJobs();
        for (IRmJob j : jobs.values() ) {
            HashMap<Share, Share> sharesE = j.getAssignedShares();
            HashMap<Share, Share> sharesN = j.getPendingShares();        	

            logger.trace(methodName, j.getId(), "<<<<<<<<<<  EXPAND");
            logger.trace(methodName, j.getId(), "adding", sharesN.size(), "new shares to existing", sharesE.size(), "shares.");
            pending_expansions += (sharesN.size() * j.getShareOrder());

            for ( Share s : sharesE.values()) {
                logger.trace(methodName, j.getId(), "    existing ", s.toString());
            }

            for ( Share s : sharesN.values()) {
                logger.trace(methodName, j.getId(), "    expanding", s.toString());
            }
            logger.trace(methodName, j.getId(), "<<<<<<<<<<");

            sharesN = j.promoteShares();
            if ( sharesN.size() == 0 ) {
                // internal error - should not be marked expanded if no machines
                throw new SchedulingException(j.getId(), "Trying to execute expanded job but no pending machines.");
            }

            for ( Share s : sharesN.values()) {                           // update machine books                
                // Sanity checks on the bookkeeping
                busyShares.put(s.getId(), s);                
            }

//            DuccId id = j.getId();                                  // pull from dormant, maybe
//            if ( dormantJobs .containsKey(id) ) {
//                dormantJobs .remove(id);
//            }

            //runningJobs.put(id, j);
            jmu.addShares(j, sharesN);
            // jobManager.executeJob(j, shares);                      // will update job's pending lists

        }

        jobs = upd.getStableJobs();                             // squirrel these away to try next time
        for (IRmJob j: jobs.values()) {
            if ( j.countNShares() < 0 ) {
                throw new SchedulingException(j.getId(), "Share count went negative " + j.countNShares());
            }
            logger.trace(methodName, j.getId(), ".......... STABLE with ", j.countNShares(), " shares.");
        }

        jobs = upd.getDormantJobs();                             // squirrel these away to try next time
        for (IRmJob j: jobs.values()) {
            logger.trace(methodName, j.getId(), ".......... DORMANT");
//            dormantJobs .put(j.getId(), j);
        }

        jobs = upd.getReservedJobs();
        for (IRmJob j: jobs.values()) {
            logger.trace(methodName, j.getId(), "<<<<<<<<<<  RESERVE");

            HashMap<Share, Share> sharesE = j.getAssignedShares();
            HashMap<Share, Share> sharesN = j.getPendingShares();        	

            if ( sharesE.size() == j.getMaxShares() ) {
                logger.trace(methodName, j.getId(), "reserve_stable", sharesE.size(), "machines");
            } else  if ( sharesN.size() == j.getMaxShares() ) {           // reservation is complete but not yet confirmed?
                logger.trace(methodName, j.getId(), "reserve_adding", sharesN.size(), "machines");
                for ( Share s : sharesN.values()) {
                    logger.trace(methodName, j.getId(), "    reserve_expanding ", s.toString());
                }
                jmu.addShares(j, sharesN);                
                j.promoteShares();
            } else {
                logger.trace(methodName, j.getId(), "reserve_pending", j.getMaxShares(), "machines");
            }
            logger.trace(methodName, j.getId(), "<<<<<<<<<<");
        }

        jmu.setAllJobs((HashMap<DuccId, IRmJob>)allJobs);

        jobs = upd.getRefusedJobs();
        Iterator<IRmJob> iter = jobs.values().iterator();
        while ( iter.hasNext() ) {
            IRmJob j = iter.next();
            logger.trace(methodName, j.getId(), ".......... REFUSED");
        }

        return jmu;
    }

    /**
     * We don't accept new work or even Orchestrator state updates until "ready". We do
     * want machines, but be sure the internal structures are protected.
     */
    public synchronized boolean ready()
    {
    	return stability;
    }

    public synchronized void start()
    {
        stability = true;
    }

    public void stop()
    {
        persistence.close();
    }

    protected void handleIllNodes()
    {
    	String methodName = "handleIllNodes";
    	
        if ( ! isInitialized() ) {
            logger.info(methodName, null, "Waiting for (re)initialization.");
            return;
        }

        HashMap<Node, Integer> nodeUpdates = new HashMap<Node, Integer>();
        synchronized(illNodes) {
            nodeUpdates.putAll(illNodes);
            illNodes.clear();
        }
        
        synchronized(this) {
            for ( Node n : nodeUpdates.keySet() ) {
                Machine m = getMachine(n);

                if ( m == null ) {
                    logger.warn(methodName, null, "Cannot find any record of machine", n.getNodeIdentity().getName());
                    continue;
                }

                int count = nodeUpdates.get(n);
                if ( count == 0 ) {
                    m.heartbeatArrives();
                } else {
                    m.heartbeatMissed(count);
                }
            }
        }
    }

    protected void handleDeadNodes()
    {
    	String methodName = "handleDeadNodes";
    	
        if ( ! isInitialized() ) {
            logger.info(methodName, null, "Waiting for (re)initialization.");
            return;
        }

        HashMap<Node, Node> nodeUpdates = new HashMap<Node, Node>();
        synchronized(deadNodes) {
            nodeUpdates.putAll(deadNodes);
            deadNodes.clear();
        }

        synchronized(this) {

            for ( Node n : nodeUpdates.values() ) {
                Machine m = getMachine(n);

                if ( m == null ) {
                    // must have been removed because of earlier missed hb
                    continue;
                }

                logger.warn(methodName, null, "***Purging machine***", m.getId(), "due to missed heartbeats. THreshold:",  nodeStability);
                NodePool np = m.getNodepool();
                np.nodeLeaves(m);
            }
        }        
    }

    /**
     * We first accept any changes and requests from the outside world and place them where they
     * can be acted on in this epoch.
     *
     * We then pass all relevant requests and resources to the IScheduler.  This returns a
     * SchedulingUpdate which is passed to the dispatcher to be acted upon.
     */
    public JobManagerUpdate schedule()
    {
    	String methodName = "schedule";


//         if ( startupCountdown++ < nodeStability ) {
//             logger.info(methodName, null, "Startup countdown:", startupCountdown, "of", nodeStability);
//             return null;
//         }

        if ( ! ready() ) {
            return null;
        }

        if ( ! isInitialized() ) {
            logger.info(methodName, null, "Waiting for (re)initialization.");
            return null;
        }

        // tracking the OR hang problem - are topics being delivered?
        logger.info("nodeArrives", null, "Total arrivals:", total_arrivals);

        synchronized(this) {
            handleIllNodes();
            handleDeadNodes();
            resetNodepools();
        }

        // TODO: Can we combine these two into one?
        SchedulingUpdate upd = new SchedulingUpdate();              // state from internal scheduler
        JobManagerUpdate jmu = new JobManagerUpdate();              // state we forward to job manager

        // int nchanges = 0;
    	

        ArrayList<IRmJob> jobsToRecover = new ArrayList<IRmJob>();
        synchronized(recoveredJobs) {
            jobsToRecover.addAll(recoveredJobs);
            recoveredJobs.clear();
            // nchanges += jobsToRecover.size();
        }

        ArrayList<IRmJob> newJobs = new ArrayList<IRmJob>();
        // 
        // If there are new jobs we need to init some things and start a scheduling cycle.
        //
        synchronized(incomingJobs) {            
            newJobs.addAll(incomingJobs);
            incomingJobs.clear();
            // nchanges += newJobs.size();
        }

        //
        // If some jobs pased initializion we need to signal a scheduling cycle to get
        // them their fair share
        //
//        synchronized(initializedJobs) {            
//            if ( initializedJobs.size() > 0 ) {
//                nchanges++;
//            }
//            initializedJobs.clear();
//        }

        //
        // If some jobs completed we need to process clearning them out and signal a
        // scheduling cycle to try to reuse their resources.
        //
        ArrayList<IRmJob> doneJobs = new ArrayList<IRmJob>();
        synchronized(completedJobs) {
            doneJobs.addAll(completedJobs);
            completedJobs.clear();
            //nchanges += doneJobs.size();
        }

        //
        // If some shares were vacated we need to clear them out and run a scheduling cycle.
        //
        ArrayList<Pair<IRmJob, Share>> doneShares= new ArrayList<Pair<IRmJob, Share>>();
        synchronized(vacatedShares) {
            doneShares.addAll(vacatedShares.values());
            vacatedShares.clear();
            //nchanges += doneShares.size();

            // we use the vacatedShares object to control share growth as well
            //if ( growthOccurred ) nchanges++;
            //growthOccurred = false;
        }

//         boolean must_run = false;
//         synchronized(force_epoch) {
//             must_run = force_epoch;
//             force_epoch = false;
//         }

//         if ( (nchanges == 0) && !must_run ) { 
//             jmu.setAllJobs(allJobs);
//             return jmu;
//         }
// TODO if we remove this code above be sure to clear out all the force_epoch nonsense
// TODO does this even use growthOccurred?

        synchronized(this) {

            // before looking at jobs, insure we're updated after a crash
            for ( IRmJob j : jobsToRecover ) {
                processRecovery(j);
            }

            // process these next to free up resources for the scheduling cycle
            for (Pair<IRmJob, Share> p : doneShares) {
                processCompletion(p.first(), p.second());
            }

            for (IRmJob j : doneJobs) {
                processCompletion(j);
            }

            // update user records, "check in" new jobs
            if ( newJobs.size() > 0 ) {
                logger.info(methodName, null, "Jobs arrive:");
                logger.info(methodName, null, "submit", RmJob.getHeader());
            }

            Iterator<IRmJob> iter = newJobs.iterator();
            while ( iter.hasNext() ) {
                IRmJob j = iter.next();


                if ( j.isRefused() ) {          // the JobManagerConverter has already refused it
                    logger.info(methodName, j.getId(), "Bypassing previously refused job.");
                    upd.refuse(j, j.getRefusalReason());
                }

                String user = j.getUserName();
                User u = users.get(user);
                if ( u == null ) {
                    u = new User(user);
                    users.put(user, u);
                }
                j.setUser(u);

                // Calculate its share order
                int share_order = calcShareOrder(j);
                j.setShareOrder(share_order);

                // Assign it to its priority class
                String clid = j.getClassName();
                ResourceClass prclass = resourceClassesByName.get(clid);

                u.addJob(j);
                allJobs.put(j.getId(), j);
                if ( prclass == null ) {                    
                    upd.refuse(j, "Cannot find priority class " + clid + " for job");
                    continue;
                }

                // UIMA-4275 never refuse impossible work, just let it hang out
//                 if ( share_order > max_order ) {
//                     upd.refuse(j, "Memory requested " + j.getMemory() + "GB exceeds the capacity of any machine in the cluster.");
//                     continue;
//                 }

                /**
                 * We want to allow this - a normal job, submitted to a reservation class.
                   if ( (prclass.getPolicy() == Policy.RESERVE ) && ( ! j.isReservation() ) ) {
                   upd.refuse(j, "Reservaction class " + 
                   prclass.getId() + " specified but work is not a reservation.");
                   continue;
                   }
                */

                if ( ((prclass.getPolicy() != Policy.RESERVE ) && (prclass.getPolicy() != Policy.FIXED_SHARE)) && ( j.isReservation() ) ) {
                    upd.refuse(j, "Class " + prclass.getName() + " is policy " + 
                               prclass.getPolicy() + " but the work is submitted as a reservation.");
                    continue;
                }

                prclass.addJob(j);
                j.setResourceClass(prclass);
                try {
					persistence.addJob(j);
				} catch (Exception e) {
					logger.warn(methodName, j.getId(), "Cannot persist new job in database:", e);					
				}
                logger.info(methodName, j.getId(), "submit", j.toString());
            }

            logger.info(methodName, null, "Scheduling " + newJobs.size(), " new jobs.  Existing jobs: " + allJobs.size());
            for ( int i = 0; i < schedulers.length; i++ ) {
                logger.info(methodName, null, "Run scheduler", i, "with top-level nodepool", nodepools[i].getId());
                schedulers[i].schedule(upd);
            }

            for ( IRmJob j : allJobs.values() ) {       // UIMA-4577 persist 'demand'
                try {
					persistence.updateDemand(j);
				} catch (Exception e) {
					logger.warn(methodName, j.getId(), "Cannot update demand in database:", e);
				}
            }

            logger.info(methodName, null, "--------------- Scheduler returns ---------------");
            logger.info(methodName, null, "\n", upd.toString());
            logger.info(methodName, null, "------------------------------------------------");                
            dispatch(upd, jmu);                 // my own job lists get updated by this

            return jmu;
        }
    }

    synchronized public void shutdown()
    {
        done = true;
    }

//     public void run()
//     {
//     	String methodName = "run";
//         while ( ! done ) {
//             try { sleep(epoch); } catch (InterruptedException e) { }

//             logger.info(methodName, null, "========================== Epoch starts ===========================");
//             try {
//                 schedule();
//             } catch ( SchedulingException e ) {
//                 logger.info(methodName, e.jobid, e);
//             }

//             logger.info(methodName, null, "========================== Epoch ends   ===========================");
//         }
//     }


    /**
     * maps from both the fully-qualified name and th shortnmae
     */
    void updateNodepoolsByNode(String longname, NodePool np)
    {
    	String methodName = "updateNodepoolsByNode";
        String shortname = longname;
        int ndx = longname.indexOf(".");

        logger.info(methodName, null, "Map", longname, "to", np.getId());
        nodepoolsByNode.put(longname, np);

        if ( ndx >= 0 ) {
            shortname = longname.substring(0, ndx);
            nodepoolsByNode.put(shortname, np);
            shortToLongNode.put(shortname, longname);
            logger.info(methodName, null, "Map", shortname, "to", np.getId());
        }
    }

    //
    // Return a nodepool by Node.  If the node can't be associated with a nodepool, return the
    // default nodepool, which is always the first one defined in the config file.
    //
    NodePool getNodepoolByName(NodeIdentity ni)
    {
        NodePool np = nodepoolsByNode.get( ni.getName() );
        if ( np == null ) {
            np = nodepoolsByNode.get( ni.getIp() );
        }
        if ( np == null ) {
            np = nodepools[0];
            updateNodepoolsByNode(ni.getName(), np);     // assign this guy to the default np
            // nodepoolsByNode.put( ni.getName(), np);          // assign this guy to the default np
        }
        return np;
    }

    private int total_arrivals = 0;
    public synchronized void nodeArrives(Node node)
    {        
        String methodName = "nodeArrives";
        if ( ! isInitialized() ) {
            logger.info(methodName, null, "Waiting for (re)initialization; node = " + node.getNodeIdentity().getName());
            return;
        }

        synchronized(illNodes) {        // stop flagging it as a problem
            illNodes.remove(node);
        }

    	// String methodName = "nodeArrives";
        // The first block insures the node is in the scheduler's records as soon as possible

        total_arrivals++;       // report these in the main schedule loop
        
        NodePool np = getNodepoolByName(node.getNodeIdentity());  // finds np assigned in ducc.nodes; if none, returns the default np
        Machine m = np.getMachine(node);
        int share_order = 0;

        // let's always recalculate this in case it changes for whatever bizarre reason (reboot, or pinned process gone, or whatever)
        long allocatable_mem =  node.getNodeMetrics().getNodeMemory().getMemFree() - share_free_dram;
        if ( dramOverride > 0 ) {
            allocatable_mem = dramOverride;
        }
        share_order = (int) (allocatable_mem / np.getShareQuantum());           // conservative - rounds down (this will always cast ok)                
        // NOTE: we cannot set the order into the machine yet, in case it has changed, because NodePool needs to adjust based
        //       on current and new

        max_order = Math.max(share_order, max_order);
        m = np.nodeArrives(node, share_order);                         // announce to the nodepools
        m.heartbeatArrives();
    }

    public void nodeHb(Node n, int count)
    {
        synchronized(illNodes) {
            illNodes.put(n, count);
        }
    }

    public void nodeDeath(Map<Node, Node> nodes)
    {
        synchronized(deadNodes) {
            deadNodes.putAll(nodes);
        }
    }

    /**
     * User passed us a node by name.  Maybe did and maybe didn't qualify it.  
     * Maybe the node checked in qualified maybe it didn't.  Here we try to find
     * something that kind of matches.
     * UIMA-4142.  Technically a bug on vary-on and vary-off but found and fixed as part of
     *             the indicated Jira.
     */
    synchronized String resolve(String node)
    {
        NodePool np = nodepoolsByNode.get(node);
        if ( np == null ) return null;                  // indexed by long and short so if not found we're stuck

        if ( np.hasNode(node) ) return node;            // he knows it by this name we're done

        int ndx = node.indexOf(".");
        if ( ndx > 0 ) {
            // np MUST know it by either long or short or it wouldn't be in nodepoolsByNode
            // so it must be short
            return node.substring(0, ndx);
        } else {
            // and vice-versa, it must be the long
            return shortToLongNode.get(node);
        }
    }

    public synchronized RmAdminReply varyon(String[] nodes)
    {
        String methodName = "varyon";
        RmAdminVaryReply ret = new RmAdminVaryReply();
        StringBuffer sb = new StringBuffer();
        for (String n : nodes ) {

            String rn = resolve(n);
            if ( rn == null ) {
                ret.setRc(false);
                ret.addFailedHost(n);
                sb.append("VaryOn: " + n + " cannot be found in the RM.\n");
            } else {                
                NodePool np = nodepoolsByNode.get(rn);  // if null, resolve will fail
                if ( np == null ) {
                    ret.setRc(false);
                    ret.addFailedHost(rn);
                    sb.append("VaryOn: " + n + " cannot find associated nodepool.\n");
                } else {
                    String repl = np.varyon(rn);
                    logger.info(methodName, null, repl);
                    sb.append(repl);
                    sb.append("\n");
                }
            }
        }
        ret.setMessage(sb.toString());
        return ret;
    }

    public synchronized RmAdminReply varyoff(String[] nodes)
    {
        String methodName = "varyoff";
        RmAdminVaryReply ret = new RmAdminVaryReply();
        StringBuffer sb = new StringBuffer();
        for (String n : nodes ) {

            String rn = resolve(n);
            if ( rn == null ) {
                ret.setRc(false);
                ret.addFailedHost(n);
                sb.append("VaryOff: " + n + " cannot be found in the RM.\n");
            } else {
                NodePool np = nodepoolsByNode.get(rn);  // if null, resolve will fail
                if ( np == null ) {
                    ret.setRc(false);
                    ret.addFailedHost(rn);
                } else {
                    String repl = np.varyoff(rn);
                    logger.info(methodName, null, repl);
                    sb.append(repl);
                    sb.append("\n");
                }
            }
        }
        ret.setMessage(sb.toString());
    	return ret;
    }

    RmQueriedNodepool  getNpStats(NodePool np)
    {

        RmQueriedNodepool ret = new RmQueriedNodepool();
        
        ret.setName(np.getId());
        ret.setOnline(np.countLocalMachines());
        ret.setDead(np.countLocalUnresponsiveMachines());
        ret.setOffline(np.countLocalOfflineMachines());
        
        ret.setSharesAvailable(np.countLocalShares());
        ret.setSharesFree(np.countLocalQShares());

        ret.setAllMachines(np.countAllLocalMachines());

        int[] onlineMachines = np.makeArray();
        int[] freeMachines = np.makeArray();
        for ( int i = 1; i < freeMachines.length; i++ ) {
            freeMachines[i] += np.countFreeMachines(i);         // (these are local, as we want)
        }

        //np.getLocalOnlineByOrder(onlineMachines);
        ret.setOnlineMachines(onlineMachines);
        ret.setFreeMachines(freeMachines);

        ret.setVirtualMachines(np.countLocalVMachinesByOrder());

        // logger.info(methodName, null, np.getId() + ": online", online, "dead", dead, "offline", offline, "shares_available", shares_available, "shares_free", shares_free);
        // logger.info(methodName, null, np.getId() + ": allMachines    ", Arrays.toString(allMachines));
        // logger.info(methodName, null, np.getId() + ": onlineByOrder  ", Arrays.toString(onlineMachines));

        // logger.info(methodName, null, np.getId() + "------- freeMachines should match free -------");
        // logger.info(methodName, null, np.getId() + ": freeMachines   ", Arrays.toString(freeMachines));
        // logger.info(methodName, null, np.getId() + ": free           ", Arrays.toString(free));
        // logger.info(methodName, null, np.getId() + "----------------------------------------------");
        // logger.info(methodName, null, np.getId() + ": virtualMachines", Arrays.toString(virtualMachines));

        return ret;
    }

    void calculateLoad(RmAdminQLoadReply reply)
    {

        for ( ResourceClass cl : resourceClasses.values() ) {
            RmQueriedClass qcl = new RmQueriedClass();

            switch ( cl.getPolicy() ) {
                case FAIR_SHARE:
                    qcl.setPolicy("FAIR_SHARE");
                    break;
                case FIXED_SHARE:
                    qcl.setPolicy("FIXED_SHARE");
                    break;
                case RESERVE:
                    qcl.setPolicy("RESERVE");
                    break;
            }

            // TODO MUST FIX THIS

            // int[] demanded = NodePool.makeArray();
            // int[] awarded  = NodePool.makeArray();

            // HashMap<IRmJob, IRmJob> jobs = cl.getAllJobs();
            // for ( IRmJob j : jobs.values() ) {
            //     int o = j.getShareOrder();
            //     demanded[o] += j.queryDemand();
            //     awarded[o]  += j.countNShares();
            // }
            
            // qcl.setName(cl.getName());
            // qcl.setDemanded(demanded);
            // qcl.setAwarded(awarded);
            // reply.addClass(qcl);
        }
    }

    void listAllNodepools(NodePool parent, ArrayList<NodePool> list)
    {
        list.add(parent);
        for (NodePool np : parent.getChildren().values() ) {
            listAllNodepools(np, list);
        }
    }

    public synchronized RmAdminQLoadReply queryLoad()
    {        

        RmAdminQLoadReply ret = new RmAdminQLoadReply();
        if ( ! ready() ) {
            ret.notReady();
            return ret;
        }

        calculateLoad(ret);

        ArrayList<NodePool> allpools = new ArrayList<NodePool>();
        for ( NodePool np : nodepools ) {
            listAllNodepools(np, allpools);
        }

        for ( NodePool np : allpools ) {
            ret.addNodepool(getNpStats(np));
        }
        
        return ret;
    }


    public synchronized RmAdminQOccupancyReply queryOccupancy()
    {
        RmAdminQOccupancyReply ret = new RmAdminQOccupancyReply();
        if ( ! ready() ) {
            ret.notReady();
            return ret;
        }

        //
        // iterate top-level nodepools to get all their subpools
        //   iterate the subpools to get all their machines
        //      iterage the machines and request a query object
        //         add query object to ret
        // return ret

        // We want to be dependent on common project, not the other way around, so
        // we keep the query objects in common and put knowledge of how to construc
        // them into rm's Machine class.
        //
        // The alternative, passing RM's Machine to the query object creates a circular
        // dependency with RM depending on common and common depending on RM.
        //
        
        //
        // Not a cheap query, by the way.
        //
        // NOTE: No longer used by the rm_qoccupancy script which now goes directly to the database
        //
        
        
        for ( NodePool np : nodepools ) {

            // NOTE:  The offline & dead nodes are also in the AllMachines list so must be removed
            Map<Node, Machine> allMachs     = np.getAllMachines();
            Map<Node, Machine> offline      = np.getOfflineMachines();          // UIMA-4234
            Map<Node, Machine> unresponsive = np.getUnresponsiveMachines();     // UIMA-4234

            for ( Node n : offline.keySet() ) {
                Machine m = offline.get(n);
                RmQueriedMachine qm = m.queryMachine();
                qm.setOffline();
                if ( unresponsive.containsKey(n) ) {
                    unresponsive.remove(n);
                    qm.setUnresponsive();
                }
                ret.addMachine(qm);
                allMachs.remove(n);
            }

            for ( Node n : unresponsive.keySet() ) {
                Machine m = unresponsive.get(n);
                RmQueriedMachine qm = m.queryMachine();
                qm.setUnresponsive();
                ret.addMachine(qm);
                allMachs.remove(n);
            }
            
            for ( Node n : allMachs.keySet() ) { 
              Machine m = allMachs.get(n);
              ret.addMachine(m.queryMachine());
          }
        }

        return ret;
    }

    public synchronized void signalState(DuccId jobid, String state)
    {
        IRmJob j = allJobs.get(jobid);
        if ( j != null ) {                // might not be here yet, we'll get it later
            j.setState(state);
        }
    }

    /**
     * Callback from job manager, need shares for a new fair-share job.
     */
    public void signalNewWork(IRmJob job)
    {
        // We'll synchronize only on the incoming job list 
        synchronized(incomingJobs) {            
            incomingJobs.add(job);
        }
    }

//     public void signalForceEpoch()
//     {
//         synchronized( force_epoch ) {
//             force_epoch = true;
//         }
//     }

    public void signalInitialized(IRmJob job)
    {
        // We'll synchronize only on the incoming job list 
        synchronized(initializedJobs) {            
            initializedJobs.add(job);
        }
    }

    public void signalRecovery(IRmJob job)
    {
        synchronized(recoveredJobs) {
            recoveredJobs.add(job);
        }
    }

    public void jobCancelled(DuccId id)
    {
        // TODO Fill this in.
    }

    /**
     * Callback from job manager when a job completes. We just believe him, no sanity checks or other such stuff.
     */
    public void signalCompletion(DuccId id)
    {
        String methodName = "signalCompletion";
        synchronized(completedJobs) {
            try {
                IRmJob job = allJobs.get(id);
                if ( job == null ) {
                    logger.warn(methodName, id, "Job completion signal: early termination; nothing to complete.");
                    return;  // canceled or terminated very soon.
                }

                logger.info(methodName, id, "Job completion signal.");
                completedJobs.add(job);
            } catch (Throwable t) {
                logger.warn(methodName, id, t);
            }
        }
    }

    /**
     * Callback from job manager when a specific share exits but the job is still alive.
     */
    public void signalCompletion(IRmJob job, Share share)
    {
        String methodName = "signalCompletion";
        synchronized(vacatedShares) {
            logger.info(methodName, job.getId(), "Job vacate signal share: ", share.toString());
            vacatedShares.put(share.getId(), new Pair<IRmJob, Share>(job, share));
        }            
    }

    /**
     * Callback from job manager when a specific share gets a process associated.
     */
//    public void signalGrowth(DuccId jobid, Share share)
//    {
//        String methodName = "signalGrowth";
//        synchronized(vacatedShares) {
//            logger.info(methodName, jobid, "Job growth signal share: ", share.toString());
//            growthOccurred = true;
//        }            
//    }

    /**
     * Called in scheduling cycle, to actually complete the job - avoids deadlock
     */
    private synchronized void processCompletion(IRmJob job)
    {
        String methodName = "processCompletion";
        logger.info(methodName, job.getId(), "Job completes.");

        try {
			persistence.deleteJob(job);     // UIMA-4577
		} catch (Exception e) {
			logger.warn(methodName, job.getId(), "Cannot delete job from database:", e);
		}

        // -- clean up the running jobs list
        IRmJob j = allJobs.remove(job.getId());
        if ( j == null ) {
            logger.info(methodName, job.getId(), "Job is not in run list!");  // can happen if job is refused very early
            return;
        }

        j.markComplete();

        // -- clean up user list
        User user = users.get(j.getUserName());
        user.remove(job);         // UIMA4275 don't clean up users list because it may have registry things in it

        ResourceClass rc = job.getResourceClass();
        if ( rc != null ) {
            rc.removeJob(j);            // also clears it if it's a reservation
        } else if ( !j.isRefused() ) {
            throw new SchedInternalError(j.getId(), "Job exits from class " + job.getClassName() + " but we cannot find the priority class definition.");
        }


        // -- clean up machine lists
        HashMap<Share, Share> shares= job.getAssignedShares();        
        for (Share s: shares.values()) {
            purgeShare(s, job);
        }
        job.removeAllShares();
    }

    /**
     * Called from scheduling cycle - a specific share has run out of work for the give job (but the
     * job is not done yet).
     */
    private synchronized void processCompletion(IRmJob job, Share share)
    {
        String methodName = "processCompletion";
        
        logger.debug(methodName, job.getId(), "Job vacates share ", share.toString());
        //share.removeJob();
        job.removeShare(share);
        purgeShare(share, job);
    }

    /**
     * Log following / reconstruction, needed to init before recovery.
     */
    public void resetNodepools()
    {
        for ( NodePool np : nodepools ) {
            np.reset(np.getMaxOrder());
        }
    }

    /**
     * Determine if the given share is in a nodepool that this job is allowed to be scheduled over.
     * You can get a mismatch if the classes or nodepools are reconfigured and RM is restarted
     * with jobs still in the system.
     *
     * UIMA-4142
     *
     * @param     s  The share to validate.
     * @param     j  The job to validate against.
     * @return true  if s and j are compatible, false otherwise.
     */
    boolean compatibleNodepool(Share s, IRmJob j)
    {
        // cut to the chase and ask the NP directly if this dude is allowed

        NodePool np = s.getNodepool();
        ResourceClass rc = j.getResourceClass();
        Policy p = rc.getPolicy();

        return np.compatibleNodepool(p, rc);
    }

    /**
     * Make this public for log following.
     */
    public synchronized void processRecovery(IRmJob j)
    {
    	String methodName = "processRecovery";

        ResourceClass rc = resourceClassesByName.get(j.getClassName());
        j.setResourceClass(rc);
        
        int share_order = calcShareOrder(j);
        j.setShareOrder(share_order);

        HashMap<Share, Share> shares = j.getRecoveredShares();
        List<Share> sharesToShrink = new ArrayList<Share>();       // UIMA-4142
        StringBuffer sharenames = new StringBuffer();
        for ( Share s : shares.values() ) {
            sharenames.append(s.toString());
            sharenames.append(" ");

            switch ( rc.getPolicy() ) {
                case FAIR_SHARE:
                    s.setShareOrder(share_order);
                    if ( !compatibleNodepool(s, j) ) {            // UIMA-4142
                        sharesToShrink.add(s);
                        break;
                    } 
                    break;
                case FIXED_SHARE:
                    logger.info(methodName, j.getId(), "Set fixed bit for FIXED job");
                    s.setShareOrder(share_order);
                    s.setFixed();
                    if ( !compatibleNodepool(s, j) ) {            // UIMA-4142
                        if ( j.isService() ) {
                            sharesToShrink.add(s);   // nodepool reconfig snafu, SM will reallocate the process
                        } else {
                            logger.warn(methodName, j.getId(), "Share is in incompatible nodepool but cannot be evicted:", s);
                        }
                    }
                    break;
                case RESERVE:
                    logger.info(methodName, j.getId(), "Set fixed bit for RESERVE job");
                    s.setFixed();
                    if ( j.isService() && !compatibleNodepool(s, j) ) {       // UIMA-4142
                        sharesToShrink.add(s);   // nodepool reconfig snafu, SM will reallocate the process
                    }
                    break;
            }

            // if ( rc.getPolicy() != Policy.RESERVE ) {          // if it's RESERVE, the share order is already set from
            //                                                    // the machine when the job arrives. 
            //     s.setShareOrder(share_order);
            // }

            Machine m = s.getMachine();
            NodePool np = m.getNodepool();
            np.connectShare(s, m, j, s.getShareOrder());

            busyShares.put(s.getId(), s);
        }
        String username = j.getUserName();
        User user = users.get(username);
        if ( user == null ) {
            user = new User(username);
            users.put(username, user);
            logger.info(methodName, j.getId(), "&&&&&&&&&&&&&&&& new user", user.toString(), "-------------------");
        }
        j.setUser(user);
        user.addJob(j);

    	j.promoteShares();                       // NOT expanded, just recovered, promote them right away
        j.clearRecoveredShares();

        String clid = j.getClassName();
        ResourceClass prclass = resourceClassesByName.get(clid);
        
        allJobs.put(j.getId(), j);
        prclass.addJob(j);
        j.setResourceClass(prclass);
        logger.info(methodName, j.getId(), "Recovered job:", j.toString());
        logger.info(methodName, j.getId(), "Recovered shares:", sharenames.toString());

        try {
			persistence.addJob(j);
		} catch (Exception e) {
			logger.warn(methodName, j.getId(), "Cannot persist recovered job in database:", j);
		}
        // After a reconfig/restart the share may be in the wrong place, in which case it
        // needs to be removed.  We have to wait until it is fully hooked into the structures
        // before scheduling for removal because it could take a while to go away and
        // we have to be careful not to overcommit.
        // UIMA-4142
        for ( Share s : sharesToShrink ) {
            logger.info(methodName, j.getId(), "Recovery - Removing share from wrong nodepool after reconfiguration:", s);
            j.shrinkByOne(s);
        }

    }

    /**
     * The share is gone, purge from our structures.
     */
    private void purgeShare(Share s, IRmJob j)
    {
        busyShares.remove(s.getId());         // so long, and thanks for all the fish
        Machine m = s.getMachine();
        m.removeShare(s);
    }

    public synchronized static DuccId newId()
    {
        return idFactory.next();
    }

    public synchronized static DuccId newId(long id)
    {
        return idFactory.next(id);
    }

    public void queryMachines()
    {
        for ( NodePool np : nodepools ) {
            np.queryMachines();
        }
    }

    class MachineByOrderSorter
    	implements Comparator<Machine>
    {	
    	public int compare(Machine m1, Machine m2)
        {
            if ( m1.equals(m2) ) return 0;

            if (m1.getShareOrder() == m2.getShareOrder()) {
                return (m1.getId().compareTo(m2.getId()));
            }
            return (int) (m1.getShareOrder() - m2.getShareOrder());
        }
    }


}