TwoTierProrater

/*
 * Copyright 2010 Fraunhofer Gesellschaft, Munich, Germany,
 * for its Fraunhofer Institute for Computer Architecture and Software
 * Technology (FIRST), Berlin, Germany. All rights reserved.
 * http://www.first.fraunhofer.de/
 */

package net.kwfgrid.gwes.prorater;

import net.kwfgrid.gworkflowdl.structure.Transition;
import net.kwfgrid.gworkflowdl.structure.OperationClass;
import net.kwfgrid.gworkflowdl.structure.OperationCandidate;
import net.kwfgrid.gworkflowdl.structure.Operation;
import net.kwfgrid.gwes.exception.OperationMapperException;
import net.kwfgrid.gwes.GenericWorkflowHandler;
import net.kwfgrid.gwes.Constants;
import net.kwfgrid.gwes.operationmapper.OperationMapper;

import java.util.*;
import java.io.IOException;

import org.apache.log4j.Logger;

/**
 * @author Andreas Hoheisel
 *         (<a href="http://www.andreas-hoheisel.de">www.andreas-hoheisel.de</a>)
 * @author Dietmar Sommerfeld, GWDG
 * @version $Id: RecurrentProrater.java 1419 2010-11-01 14:12:17Z hoheisel $
 */
public class TwoTierProrater extends OperationMapper implements Runnable {

    /**
     * key is hardware identifier, value is Node object.
     */
    private static HashMap<String, Node> nodes;

    /**
     * key is hardware identifier, value is time of last assignment of this hardware
     */
    private static HashMap<String, Long> lastAssignments;

    /**
     * log4j logger.
     */
    final static Logger logger = Logger.getLogger(TwoTierProrater.class);

    /**
     * Nodes with a quality below MIN_QUALITY are skipped.
     */
    private static float MIN_QUALITY;

    /**
     * Wait period in Milliseconds after last usage of hardware.
     */
    private static int WAIT_AFTER_LAST_ASSIGNMENT;

    /**
     * Polling interval in Milliseconds.
     */
    private static long POLL_INTERVAL;

    /**
     * Holds transitions to be processed.
     */
    private static Vector<Transition> transitions;

    /**
     * Holds transitions that were recently processed.
     */
    private static Vector<Transition> transitionsDone;

    /**
     * key is transition, value is parent workflow ID.
     */
    private static HashMap<Transition, String> parents;

    /**
     * All processing steps are done by one single thread for all workflows.
     */
    private static TwoTierProrater prorater;
    private static Thread thread;

    private static final Object lock = new Object();

    static boolean abort;

    /**
     * Constructor.
     */
    public TwoTierProrater() {
        if (transitions == null) {
            transitions = new Vector<Transition>();
            logger.debug("Initializing TwoTier Prorater...");

            MIN_QUALITY = Float.parseFloat(System.getProperty(Constants.PROP_SYSTEM_RESOURCE_PRORATER_MIN_QUALITY, "0.3"));
            WAIT_AFTER_LAST_ASSIGNMENT = Integer.parseInt(System.getProperty(Constants.PROP_SYSTEM_RESOURCE_PRORATER_WAIT, "12000"));
            POLL_INTERVAL = Long.parseLong(System.getProperty(Constants.PROP_SYSTEM_RESOURCE_PRORATER_INTERVAL,"1000"));

            transitionsDone = new Vector<Transition>();
            parents = new HashMap<Transition, String>();
            prorater = new TwoTierProrater();
            abort = false;
            // start singleton as thread
            thread = new Thread(prorater);
            thread.start();
        }
    }

    public boolean processTransition(GenericWorkflowHandler handler, Transition transition) throws OperationMapperException {
       	// if a transition was scheduled in the previous round do not schedule it again right away
        if (!transitions.contains(transition) && !transitionsDone.contains(transition)) {
        	synchronized (lock) {
            	logger.debug("Adding transition to prorater queue: " + transition.getID() + "\tfrom workflow " + handler.getID());
            	transitions.add(transition);
                parents.put(transition, handler.getID());	// keep a reference to originating workflow for prioritization
            }
        }
        return false;
    }

    public boolean processTransitions(GenericWorkflowHandler handler, ArrayList<Transition> transitions) throws OperationMapperException {
        for (Transition t : transitions) {
            processTransition(handler,t);
        }
        return false;
    }

    /**
     * @see Thread#run()
     */
    public void run() {
        while(!abort) {
            synchronized (lock) {
                transitionsDone.clear();
                if (transitions.size() > 0) {
                    try {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Scheduling the following transitions:");
                            for (Transition t:transitions) logger.debug(t.getID());
                        }
                        initialize();
                        decide();
                        if (logger.isDebugEnabled()) {
                            if (transitions.size() > 0) {
                                logger.debug("Unscheduled transitions:");
                                for (Transition t:transitions) logger.debug(t.getID());
                            }
                        }
                    } catch (Exception e) {
                        logger.error("Exception during processTransitions(): " + e, e);
                    }
                }
            }

            try {
                Thread.sleep(POLL_INTERVAL);
            } catch (InterruptedException e) {
                logger.error("exception:\n" + e, e);
                thread.interrupt();
            }
        }
    }

    private void initialize() throws IOException {
        logger.debug("Initialize:");
        long now = System.currentTimeMillis();
        if (nodes == null) nodes = new HashMap<String, Node>();
        if (lastAssignments == null) lastAssignments = new HashMap<String, Long>();
        nodes.clear();
        for (Transition transition : transitions) {
            synchronized (transition.getOperation()) {
                //logger.debug("entered synchronized block");
                Operation operation = transition.getOperation();
                if (operation == null) {
                    logger.warn("No <operation> available for transition "+transition.getID());
                    continue;
                }
                Object oo = operation.get();

                if (oo instanceof OperationClass) {

                    OperationCandidate[] operationCandidates = ((OperationClass) oo).getOperationCandidates();
                    if (operationCandidates == null) {
                        logger.warn("No <operationCandidate> available for transition "+transition.getID());
                        continue;
                    }

                    int priority = 0;
                    // <property name="priority">1</property>
                    String propStr = transition.getProperties().get(Constants.PROP_TRANSITION_PRIORITY);
                    if (propStr != null) {
                        try {
                            priority = Integer.parseInt(propStr);
                        } catch (NumberFormatException e) {
                            // do nothing.
                        }
                    }

                    for (OperationCandidate operationCandidate : operationCandidates) {
                        String resourceName = operationCandidate.getResourceName();
                        if (resourceName == null) {
                            logger.warn("No <resourceName> available for operationCandidate of transition "+transition.getID());
                            continue;
                        }
                        // skip hardware that has been recently assigned (wait for load update)
                        Long last = lastAssignments.get(resourceName);
                        if (last != null && now - last < WAIT_AFTER_LAST_ASSIGNMENT) {
                            //logger.debug("Skipping resource  " + resourceName);
                        	continue;
                        }

                        Node node = nodes.get(resourceName);
                        if (node == null) {
                            node = new Node(resourceName);
                            node.updateQuality(operationCandidate.getQuality());
                            nodes.put(resourceName, node);
                        }

                        // only operationCandidates with a high node quality get into the game
                        if (node.quality > MIN_QUALITY) {
                            node.add(new Alternative(transition, operationCandidate, priority));
                        }
                        // for the others set the quality
                        else {
                            operationCandidate.setQuality(node.quality);
                        }

                    }
                }
            }
        }

        if (logger.isDebugEnabled()) {
            logger.debug("--------- last resource assignments ---------");
            for (String hw : lastAssignments.keySet()) {
                logger.debug(hw + ":\t" + (now - lastAssignments.get(hw)) + "ms ago");
            }
        }
    }

    private void decide() {
        if (nodes.size() == 0) return;
        logger.debug("Decide:");
        ArrayList<Node> nc = generateClusteredRandomDistribution(nodes);
        for (Node node : nc) {
            int nr = node.alternatives.size();	// number of alternative operationCandidates involving this host
            if (nr == 0) continue;
            //logger.debug("Scheduling node: " + node.hostname);
            for (int i = 0; i < nr; i++) {
                Alternative a = node.alternatives.get(i);
                if (!transitionsDone.contains(a.transition)) {
                	// look for transitions from same workflow with higher priority
                	for (int j = i + 1; j < nr; j++) {
                        Alternative b = node.alternatives.get(j);
                        if (!transitionsDone.contains(b.transition)) {
                        	if (b.priority > a.priority && parents.get(a.transition).equals(parents.get(b.transition))) {
                                logger.debug("Priority: " + b.transition.getID() + " > " + a.transition.getID());
                                a = b;
                            }
                		}
                	}
                    transitionsDone.add(a.transition);
                    transitions.remove(a.transition);
                    parents.remove(a.transition);
                    lastAssignments.put(node.hostname, System.currentTimeMillis());
                    a.operationCandidate.setSelected(true);
                    logger.debug("Decision: " + a.transition.getID() + ", priority " + a.priority + ": " + a.operationCandidate.getResourceName());
                    break;
                }
            }
        }
    }

    private ArrayList<Node> generateClusteredRandomDistribution(HashMap<String, Node> nodes) {

        // at least 3 members per quality cluster, maximum 5 quality clusters
        int nClusters = (int) Math.ceil((double) nodes.size() / 3d);
        if (nClusters < 1) nClusters = 1;
        else if (nClusters > 5) nClusters = 5;
        int nMembers = (int) Math.ceil((double) nodes.size() / (double )nClusters);
        if (logger.isDebugEnabled()) {
            logger.debug(" generateClusteredRandomDistribution");
            logger.debug(" nodes.size="+nodes.size()+" nClusters="+nClusters+" nMembers="+nMembers);
        }

        ArrayList<Node> nc = new ArrayList<Node>();

        // not enough members for clustering
        if (nClusters == 1) {
            nc.addAll(nodes.values());
            Collections.shuffle(nc);
        }
        // more than one cluster
        else {
            // sort nodes by quality
            ArrayList<Node> sortedNodes = new ArrayList<Node>();
            sortedNodes.addAll(nodes.values());
            Collections.sort(sortedNodes);
            Collections.reverse(sortedNodes);

            // loop clusters
            ArrayList<Node> clusterNodes = new ArrayList<Node>();
            Iterator<Node> nodeIter = sortedNodes.iterator();
            for (int iCluster = 0; iCluster < nClusters; iCluster++) {
                clusterNodes.clear();
                // fill cluster
                for (int iMember = 0; iMember < nMembers; iMember++) {
                    if (nodeIter.hasNext()) clusterNodes.add(nodeIter.next());
                }
                // randomize cluster
                Collections.shuffle(clusterNodes);
                // put cluster to result
                nc.addAll(clusterNodes);
            }
            // remaining nodes
            while (nodeIter.hasNext()) {
                nc.add(nodeIter.next());
            }
        }
        if (logger.isDebugEnabled()) {
            for (Node node : nc) {
                logger.debug(" node.quality="+node.quality+" host="+node.hostname);
            }
        }
        return nc;
    }

}