/*
    * 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.DatabaseException;
  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
   * @author Helge Rosé
   * @version $Id: RecurrentProrater.java 1540 2011-08-17 13:30:37Z hoheisel $
   */
  public class RecurrentProrater extends OperationMapper implements Runnable {
  
      /**
       * key is hardware identifier, value is Node object.
       */
      static HashMap<String, Node> nodes;
  
      /**
       * key is hardware identifier, value is time of last assignment of this hardware
       */
      static HashMap<String, Long> lastAssignments;
  
      /**
       * log4j logger.
       */
      final static Logger logger = Logger.getLogger(SimpleProrater.class);
  
      /**
       * Nodes with a quality below MIN_QUALITY are skipped.
       */
      static float MIN_QUALITY;
  
      /**
       * Wait period in Milliseconds after last usage of hardware.
       */
      static int WAIT_AFTER_LAST_ASSIGNMENT;
  
      /**
       * Polling interval in Milliseconds.
       */
      static long POLL_INTERVAL;
  
      /**
       * All processing steps are done by one single thread for all workflows.
       */
      static RecurrentProrater p;
      static Thread t;
  
      final Object lock = new Object();
  
      /**
       * Holds transitions to be processed.
       */
      private static Vector<Transition> transitions;
  
      static boolean abort; 
  
      /**
       * Constructor.
       */
      public RecurrentProrater() {
          if (transitions == null) {
              transitions = new Vector<Transition>();
              logger.debug("Initializing Recurrent Prorater...");
  
              MIN_QUALITY = Float.parseFloat(System.getProperty(Constants.PROP_SYSTEM_RESOURCE_PRORATER_MIN_QUALITY, "0.1"));
              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"));
  
              p = new RecurrentProrater();
              abort = false;
              // start singleton as thread
              t = new Thread(p);
              t.start();
          }
      }
  
      public boolean processTransition(GenericWorkflowHandler handler, Transition transition) throws OperationMapperException {
         if (!transitions.contains(transition) && transition.getAbstractionLevel() == Operation.BLUE) {
             synchronized(p.lock) {
                 transitions.add(transition);
             }
         }
         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 (p.lock) {
                 if (transitions.size()>0) {
                     try {
                         if (logger.isDebugEnabled()) {
                             logger.debug("Scheduling the following transitions:");
                             for (Transition t:transitions) logger.debug(t.getID());
                         }
                         initialize(transitions);
                         decide();
                         transitions.clear();
                     } catch (Exception e) {
                         logger.error("Exception during processTransitions(): " + e, e);
                     }
                 }
             }
 
             try {
                 Thread.sleep(POLL_INTERVAL);
             } catch (InterruptedException e) {
                 logger.error("exception:\n" + e, e);
                 t.interrupt();
             }
         }
     }
 
     static void initialize(Vector<Transition> transitions) throws IOException {
         logger.debug("Init:");
         long now = System.currentTimeMillis();
         if (nodes == null) nodes = new HashMap<String, Node>();
         if (lastAssignments == null) lastAssignments = new HashMap<String, Long>();
         nodes.clear();
         int priority = 0;
         for (Transition transition : transitions) {
             synchronized (transition.getOperation()) {
                 Operation operation = transition.getOperation();
                 if (operation == null) {
                     logger.warn("No <operation> available for transition "+transition.getID());
                     continue;
                 }
                 Object oo = transition.getOperation().get();
                 if (oo instanceof OperationClass) {
 
                     OperationCandidate[] operationCandidates = ((OperationClass) oo).getOperationCandidates();
                     if (operationCandidates == null) {
                         logger.warn("No <operationCandidate> available for transition "+transition.getID());
                         continue;
                     }
 
                     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) continue;
                         // <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.
                             }
                         }
                         Alternative alternative = new Alternative(transition, operationCandidate, priority);
                         Node node = nodes.get(resourceName);
 
                         if (node == null) {
                             node = new Node(resourceName);
                             node.updateQuality(operationCandidate.getQuality());
                             nodes.put(resourceName, node);
                         }
 
                         // Only nodes with a high quality get into the game
                         if (node.quality > MIN_QUALITY) {
                             node.add(alternative);
                         }
                         // for the others set the quality
                         else {
                             alternative.operationCandidate.setQuality(node.quality);
                         }
 
                     }
                     ++priority;
                 }
             }
         }
 
         if (logger.isDebugEnabled()) {
             logger.debug("--------- last resource assignments ---------");
             for (String hw : lastAssignments.keySet()) {
                 logger.debug(hw + ": " + lastAssignments.get(hw));
             }
         }
     }
 
     /**
      * @return True if there has been a decision, false if not.
      */
     static boolean decide() {
         boolean ret = false;
         if (nodes.size() == 0) return ret;
         logger.debug("Decide:");
         ArrayList<Transition> tdone = new ArrayList<Transition>();
         List<Node> nc = generateClusteredRandomDistribution(nodes);
         for (Node node : nc) {
             int nr = node.alternatives.size();	// number of alternative operationCandidates involving this host
             if (nr == 0) continue;
             if (logger.isDebugEnabled()) {
                 logger.debug("Node: " + node.hostname);
             }
             Collections.sort(node.alternatives);
             Collections.reverse(node.alternatives);	// transitions with high priority first
             for (int i = 0; i < nr; ++i) {
                 Alternative a = node.alternatives.get(i);
                 if (!tdone.contains(a.transition)) {
                     // find equal priorities
                     int tries, j = i;
                     while (j + 1 < nr && a.priority == node.alternatives.get(j + 1).priority) ++j;
                     // choose random transition in case of equal priorities
                     if (j > i) {
                         //System.out.println("node: "+node.hostname+" equal priorities: "+i+"-"+j);
                         tries = j - i;
                         while (tries-- > 0) {
                             Alternative b = node.alternatives.get(i + (int) ((j - i) * Math.random() + 0.5));
                             if (!tdone.contains(b.transition)) {
                                 a = b;
                                 break;
                             }
                         }
                     }
                     a.operationCandidate.setSelected(true);
                     ret = true;
                     tdone.add(a.transition);
                     lastAssignments.put(node.hostname, System.currentTimeMillis());
                     logger.debug("  " + a.priority + ": " + a.operationCandidate.getResourceName() + ": " + a.transition.getID());
                     break;
                 }
             }
         }
         return ret;
     }
 
     public static List<Node> generateClusteredRandomDistribution(Map<String, Node> nodes) {
 
         // at least 3 members per quality cluster, maximum 5 quality clusters
         int nClusters = nodes.size() / 3;
         if (nClusters < 1) nClusters = 1;
         else if (nClusters > 5) nClusters = 5;
         int nMembers = nodes.size() / nClusters;
         if (logger.isDebugEnabled()) {
             logger.debug("nodes.size="+nodes.size()+" nClusters="+nClusters+" nMembers="+nMembers);
         }
 
         List<Node> nc = new ArrayList<Node>();
 
         // not enough members for clustering
         if (nClusters == 1) {
             nc.addAll(nodes.values());
             Collections.shuffle(nc, new Random(System.currentTimeMillis()));
         }
         // more than one cluster
         else {
             // sort nodes by quality
             List<Node> sortedNodes = new ArrayList<Node>();
             sortedNodes.addAll(nodes.values());
             Collections.sort(sortedNodes);
             Collections.reverse(sortedNodes);
 
             // loop clusters
             List<Node> clusterNodes = new ArrayList<Node>();
             Iterator 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((Node)nodeIter.next());
                 }
                 // randomize cluster
                 Collections.shuffle(clusterNodes,new Random(System.currentTimeMillis()));
                 // put cluster to result
                 nc.addAll(clusterNodes);
             }
             // remaining nodes
             while (nodeIter.hasNext()) {
                 nc.add((Node)nodeIter.next());
             }
         }
         if (logger.isDebugEnabled()) {
             for (Node node : nc) {
                 logger.debug("node quality="+node.quality+" host="+node.hostname);
             }
         }
         return nc;
     }
 
 }