/*
 *
 *   Copyright 2007-2008 University Of Southern California
 *
 *   Licensed 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 edu.isi.pegasus.planner.provisioner;

import java.util.*;

/**
 * An instance of this class represents an independent task of a workflow. A task has PET(predicted
 * execution time), list of input data sources(parent tasks), and list of output data(child tasks).
 * This class also used for BTS algorithm such as EST, LFT, schdeuled start/finish time and core
 * methods for BTS algorithm used by OccupationDiagram class.
 *
 * @author Eunkyu Byun
 */
public class Node {
    public static final long DEFAULT_WEIGHT = 1;
    private LinkedList inEdges;
    private LinkedList outEdges;
    private Edge critEdge;

    private String id;
    private String taskName;
    private long weight;

    private long upLen;
    private long downLen;

    private HashSet ancestors;
    private HashSet descendants;

    public void init() {
        this.st = 0;
        this.ancET = -1;
        this.desET = -1;
    }

    /**
     * constructor
     *
     * @param id the unique ID of this task
     * @param name task name(executable name)
     * @param w predicted execution time
     */
    public Node(String id, String name, long w) {
        this(id, name, -1, w);
    }

    public Node(String id, String name, int n, long w) {
        this.upLen = -1;
        this.downLen = -1;
        this.id = id;
        this.taskName = name;
        this.weight = w;
        inEdges = new LinkedList();
        outEdges = new LinkedList();

        ancestors = new HashSet();
        descendants = new HashSet();
        init();
    }

    public Node(String id) {
        this(id, "NullTask", DEFAULT_WEIGHT);
    }

    /**
     * set execution time of this task
     *
     * @param w excution time
     */
    public void setWeight(long w) {
        this.weight = w;
    }

    /**
     * return execution time of this task
     *
     * @return excution time
     */
    public long evalWeight() {
        return weight;
    }

    /**
     * add a data dependency edge to this task
     *
     * @param e the edge to be added
     */
    public void addIn(Edge e) {
        if (!inEdges.contains(e)) {
            inEdges.add(e);
        }
    }

    /**
     * remove a specified data dependency edge from this task
     *
     * @param e the edge to be removed
     */
    public void removeIn(Edge e) {
        //	System.out.println("in "+id+" edge "+e.getID() + " is removedIn");
        inEdges.remove(e);
    }

    /**
     * add a data dependency edge from this task
     *
     * @param e the edge to be added
     */
    public void addOut(Edge e) {
        if (!outEdges.contains(e)) {
            outEdges.add(e);
        }
    }

    /**
     * remove a specified data dependency edge from this task
     *
     * @param e the edge to be removed
     */
    public void removeOut(Edge e) {
        //	System.out.println("in "+id+" edge "+e.getID() + " is removedOut");
        outEdges.remove(e);
    }

    /**
     * check whether all parent tasks are finished.
     *
     * @return true if all parent tasks are finished.
     */
    public boolean checkIn() {
        boolean ret = true;
        Iterator iter = inEdges.iterator();
        while (iter.hasNext()) {
            Edge next = (Edge) iter.next();
            ret &= next.complete;
            st = Math.max(st, next.compTime + next.getCost());
        }
        return ret;
    }

    /**
     * return the list of incoming edges.
     *
     * @return LinkedList of edges.
     */
    public LinkedList getIn() {
        return inEdges;
    }

    /**
     * return the list of outgoing edges.
     *
     * @return LinkedList of edges.
     */
    public LinkedList getOut() {
        return outEdges;
    }

    /**
     * notify child nodes that this task will finish at time et
     *
     * @param et finish time
     */
    public void initOut(boolean s, long et) {
        Iterator iter = outEdges.iterator();
        while (iter.hasNext()) {
            Edge next = (Edge) iter.next();
            next.complete = s;
            if (s) {
                next.compTime = Math.max(next.compTime, et);
            } else {
                next.compTime = et;
            }
        }
    }

    public long st = 0; // scheduled time, used for HEFT algorithm

    // Calculate UpLength
    private void updateUpLen() {
        if (isTop()) {
            this.upLen = 0;
            return;
        }
        long maxUpLen = -1;
        Iterator parents = inEdges.iterator();
        while (parents.hasNext()) {
            Edge next = (Edge) parents.next();
            Node cur = next.getFrom();
            long curUpLen = cur.getUpLen() + cur.evalWeight() + next.getCost();
            if (maxUpLen < curUpLen) {
                maxUpLen = curUpLen;
                critEdge = next;
            }
        }
        this.upLen = maxUpLen;
    }

    /**
     * return the UpLength of this task. UpLength is the longest path from the entry task
     *
     * @return UpLength
     */
    public long getUpLen() {
        if (this.upLen < 0) {
            updateUpLen();
        }
        return this.upLen;
    }

    private void updateDownLen() {
        if (isBottom()) {
            downLen = 0;
            return;
        }

        Iterator iter = outEdges.iterator();
        while (iter.hasNext()) {
            Edge ce = (Edge) iter.next();
            Node cn = ce.getTo();
            downLen = Math.max(downLen, cn.getDownLen() + cn.evalWeight() + ce.getCost());
        }
    }

    /**
     * return the DnLength of this task. DnLength is the longest path to the exit task
     *
     * @return DnLength
     */
    public long getDownLen() {
        if (downLen == -1) updateDownLen();
        return this.downLen;
    }

    /**
     * Check whether this task is the entry task or not.
     *
     * @return true or false
     */
    public boolean isTop() {
        return (inEdges.size() == 0);
    }

    /**
     * Check whether this task is the exit task or not.
     *
     * @return true or false
     */
    public boolean isBottom() {
        return (outEdges.size() == 0);
    }

    public String getID() {
        return id;
    }

    /**
     * update EST(earliest start time) of this task.
     *
     * @param nlb new EST
     * @param set set of ancestor tasks
     */
    public void updateLeftBound(long nlb, /*LinkedList*/ TreeSet set) {
        //	System.out.print("updateLeftBound in "+id+" from "+lb);
        if (nlb > lb) {
            //		System.out.print(" to "+nlb+ "\r\n");
            boolean contain = false;
            if (set != null) contain = set.remove(this);
            this.lb = nlb;
            if (contain) {
                //			System.out.println(id+ " is removed in updateLeftBound");
                boolean test = set.add(this);
                //			System.out.println(test);
            }

            for (int i = 0; i < outEdges.size(); i++) {
                Edge ce = (Edge) outEdges.get(i);
                Node n = ce.getTo();
                n.updateLeftBound(nlb + weight + ce.getCost(), set);
            }
        } // else System.out.print("\r\n");
    }

    /**
     * update LFT(latest finish time) of this task.
     *
     * @param nrb new LFT
     * @param set set of descendant tasks
     */
    public void updateRightBound(long nrb, /*LinkedList*/ TreeSet set) {
        //	System.out.print("updateRightBound in "+id+" from "+rb);
        if (nrb < rb) {
            //		System.out.print(" to "+nrb+"\r\n");
            boolean contain = false;
            if (set != null) contain = set.remove(this);
            this.rb = nrb;
            if (contain) {
                //			System.out.println(id+ " is removed in updateRightBound");
                boolean test = set.add(this);
                //			System.out.println(test);
            }

            for (int i = 0; i < inEdges.size(); i++) {
                Edge ce = (Edge) inEdges.get(i);
                Node n = ce.getFrom();
                n.updateRightBound(nrb - weight - ce.getCost(), set);
            }
        } // else System.out.print("\r\n");
    }

    public boolean stacked = false; // bit for stacked in Occupation diagram
    public long lb; // left bound (EST)
    public long rb; // right bound (LFT)
    public long olb; // original left bound
    public long orb; // original right bound

    public long tempST; // scheduled start time
    public long tempFT; // scheduled finish time

    /**
     * Non-propagate redistribution of this task
     *
     * @param timeMap List of scheduled tasks at each time slot. i.e., Occupation diagram itself
     * @param limit maximum height of Occupation diagram. The result of redistribution should cause
     *     taller height of this limit.
     * @param force Not used.
     * @return true if this task is redistributed.
     */
    public boolean NPbalance(LinkedList[] timeMap, int limit, boolean force) {
        if (weight == 0) return false;
        if (!force && (tempST - lb) < weight && (rb - tempFT) < weight) return false;

        int min = limit;
        int cursor = (int) lb;
        boolean rightBias = (getDesSize() < getAncSize());

        for (int i = (int) lb; i < rb - weight; i++) {
            int localMax = 0;
            for (int j = i; j < (i + weight); j++) {
                int overlap = (j >= tempST && j < tempFT) ? 1 : 0;
                localMax = Math.max(localMax, timeMap[j].size() - overlap);
            }
            if (min > localMax) {
                min = localMax;
                cursor = i;
            }
            if (min == localMax && rightBias) {
                cursor = i;
            }
        }

        if ((min + 1) >= limit && !force) return false;

        //	System.out.println("NPbalance ID:"+id+" limit:"+limit+" from: "+ tempST+" to: " +
        // cursor);

        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].remove(this);
        tempST = cursor;
        tempFT = cursor + weight;
        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].add(this);

        for (int i = 0; i < outEdges.size(); i++) {
            Edge ce = (Edge) outEdges.get(i);
            Node n = ce.getTo();
            n.updateLeftBound(tempFT + ce.getCost(), null);
        }
        for (int i = 0; i < inEdges.size(); i++) {
            Edge ce = (Edge) inEdges.get(i);
            Node n = ce.getFrom();
            n.updateRightBound(tempST - ce.getCost(), null);
        }

        return (min < limit);
    }

    /**
     * Redistribute this task to the earlier time. This causes ancestor tasks to be redistributed
     * too.
     *
     * @param timeMap List of scheduled tasks at each time slot. i.e., Occupation diagram itself
     * @param limit maximum height of Occupation diagram. The result of redistribution should cause
     *     taller height of this limit.
     * @param bound Earliest time this tasks can be scheduled.
     * @return true if this task is redistributed.
     */
    public boolean moveLeft(LinkedList[] timeMap, int limit, long bound) {
        boolean found = false;
        int cursor = (int) (bound - weight);
        for (int i = (int) (bound - weight); i >= olb; i--) {
            int localMax = 0;
            for (int j = i; j < (i + weight); j++) {
                int overlap = (j >= tempST && j < tempFT) ? 1 : 0;
                localMax = Math.max(localMax, timeMap[j].size() - overlap);
            }
            if (limit > (localMax + 1 - pushedParents(i))) {
                cursor = i;
                found = true;
                break;
            }
        }

        //	System.out.println("moveLeft ID:"+id+" limit:"+limit + " bound:"+ bound +" found:" +
        // found+ " " + cursor);

        if (!found) return false; // not possible to move

        // push parent nodes
        for (int i = 0; i < inEdges.size() && found; i++) {
            Edge ce = (Edge) inEdges.get(i);
            Node n = ce.getFrom();
            if (cursor >= lb) {
                n.updateRightBound(cursor - ce.getCost(), null);
            } else {
                found &= n.moveLeft(timeMap, limit, cursor - ce.getCost());
            }
        }
        if (!found) return false; // child block!!

        // re-schedule
        lb = Math.min(lb, cursor);
        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].remove(this);
        tempST = cursor;
        tempFT = cursor + weight;
        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].add(this);

        // release leftbound of child nodes
        for (int i = 0; i < outEdges.size(); i++) {
            Edge ce = (Edge) outEdges.get(i);
            Node n = ce.getTo();
            n.updateLeftBound(tempFT + ce.getCost(), null);
        }
        return true;
    }

    /**
     * Redistribute this task to the later time. This causes descensant tasks to be redistributed
     * too.
     *
     * @param timeMap List of scheduled tasks at each time slot. i.e., Occupation diagram itself
     * @param limit maximum height of Occupation diagram. The result of redistribution should cause
     *     taller height of this limit.
     * @param bound Lastest time this tasks can be scheduled.
     * @return true if this task is redistributed.
     */
    public boolean moveRight(LinkedList[] timeMap, int limit, long bound) {
        boolean found = false;
        int cursor = (int) bound;
        for (int i = (int) bound; i <= (orb - weight); i++) {
            int localMax = 0;
            for (int j = i; j < (i + weight); j++) {
                int overlap = (j >= tempST && j < tempFT) ? 1 : 0;
                localMax = Math.max(localMax, timeMap[j].size() - overlap);
            }
            if (limit > (localMax + 1 - pushedChildren(i + weight))) {
                cursor = i;
                found = true;
                break;
            }
        }

        //	System.out.println("moveRight ID:"+id+" limit:"+limit + " bound:"+ bound +" found:" +
        // found+ " " + cursor);
        if (!found) return false; // not possible to move

        // push child nodes
        for (int i = 0; i < outEdges.size() && found; i++) {
            Edge ce = (Edge) outEdges.get(i);
            Node n = ce.getTo();
            if (cursor + weight <= rb) {
                n.updateLeftBound(cursor + weight + ce.getCost(), null);
            } else {
                found &= n.moveRight(timeMap, limit, cursor + weight + ce.getCost());
            }
        }
        if (!found) return false; // child block!!

        // re-schedule
        rb = Math.max(rb, cursor);
        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].remove(this);
        tempST = cursor;
        tempFT = cursor + weight;
        for (int i = (int) tempST; i < tempFT; i++) timeMap[i].add(this);

        // release leftbound of child nodes
        for (int i = 0; i < inEdges.size(); i++) {
            Edge ce = (Edge) inEdges.get(i);
            Node n = ce.getFrom();
            n.updateRightBound(tempST - ce.getCost(), null);
        }
        return true;
    }

    private int pushedParents(long timeLimit) {
        int result = 0;
        for (int i = 0; i < inEdges.size(); i++) {
            Edge ce = (Edge) inEdges.get(i);
            Node cn = ce.getFrom();
            if (timeLimit < cn.tempFT) {
                result++;
            }
        }
        return result;
    }

    private int pushedChildren(long timeLimit) {
        int result = 0;
        for (int i = 0; i < outEdges.size(); i++) {
            Edge ce = (Edge) outEdges.get(i);
            Node cn = ce.getTo();
            if (timeLimit > cn.tempST) {
                result++;
            }
        }
        return result;
    }

    /** check all child task's start time is scheduled after the finishtime of this task */
    public void checkIntegrity() {
        for (int i = 0; i < outEdges.size(); i++) {
            Edge ce = (Edge) outEdges.get(i);
            Node cn = ce.getTo();
            if ((tempFT + ce.getCost()) > cn.tempST) {
                System.out.println(
                        "Violation! from:"
                                + id
                                + "("
                                + tempFT
                                + ") to "
                                + cn.getID()
                                + "("
                                + cn.tempST
                                + ")");
            }
            cn.checkIntegrity();
        }
    }

    /**
     * build the list of descendant tasks
     *
     * @return the list of descensant tasks
     */
    public Set buildDescendants() {
        if (!isBottom() && descendants.size() == 0) {
            Iterator iter = outEdges.iterator();
            while (iter.hasNext()) {
                Edge ce = (Edge) iter.next();
                Node cn = ce.getTo();
                if (!cn.isBottom()) {
                    descendants.add(cn);
                    descendants.addAll(cn.buildDescendants());
                }
            }
        }
        return descendants;
    }

    /**
     * build the list of ancestor tasks
     *
     * @return the list of ancestor tasks
     */
    public Set buildAncestors() {
        if (!isTop() && ancestors.size() == 0) {
            Iterator iter = inEdges.iterator();
            while (iter.hasNext()) {
                Edge ce = (Edge) iter.next();
                Node cn = ce.getFrom();
                if (!cn.isTop()) {
                    ancestors.add(cn);
                    ancestors.addAll(cn.buildAncestors());
                }
            }
        }
        return ancestors;
    }

    public int getDepSize() {
        return descendants.size() + ancestors.size();
    }

    public int getAncSize() {
        return ancestors.size();
    }

    public int getDesSize() {
        return descendants.size();
    }

    private long ancET;
    private long desET;

    /** @return the sum of execution time of all ancestor tasks */
    public long getAncET() {
        if (ancET < 0) {
            ancET = 0;
            Iterator iter = ancestors.iterator();
            while (iter.hasNext()) {
                Node cn = (Node) iter.next();
                ancET += cn.evalWeight();
            }
        }
        return ancET;
    }

    /** @return the sum of execution time of all descendant tasks */
    public long getDesET() {
        if (desET < 0) {
            desET = 0;
            Iterator iter = descendants.iterator();
            while (iter.hasNext()) {
                Node cn = (Node) iter.next();
                desET += cn.evalWeight();
            }
        }
        return desET;
    }

    /** @return the sum of execution time of all dependent tasks */
    public long getDepET() {
        return getAncET() + getDesET();
    }

    public boolean isAnc(Node n) {
        return ancestors.contains(n);
    }

    public boolean isDes(Node n) {
        return descendants.contains(n);
    }

    public void print() {
        String inList = "";
        Iterator iter = inEdges.iterator();
        while (iter.hasNext()) {
            inList += ((Edge) iter.next()).getID() + ",";
        }

        String outList = "";
        iter = outEdges.iterator();
        while (iter.hasNext()) {
            outList += ((Edge) iter.next()).getID() + ",";
        }

        //	System.out.println("Node_"+id+"("+ taskName +","+ weight +"sec) IN:"+inList+",
        // OUT:"+outList);
        System.out.println(
                "Node_"
                        + id
                        + "("
                        + taskName
                        + ","
                        + weight
                        + ") upLen:"
                        + upLen
                        + ", downLen:"
                        + downLen);
    }

    public int cluster = -1;
    public long tlevel = 0;
    public boolean examined = false;

    /** used for DSC algorithm */
    public boolean isFree() {
        if (examined) return false;
        boolean result = true;
        for (int i = 0; i < inEdges.size(); i++) {
            Edge ce = (Edge) inEdges.get(i);
            Node cn = ce.getFrom();
            if (!cn.examined) {
                result = false;
                break;
            }
        }
        return result;
    }

    public boolean equals(Object e) {
        //	System.out.println("comparison is called this");
        if (e instanceof Node) {
            Node cn = (Node) e;
            boolean same = cn.getID().equals(id);
            //	System.out.println(id+ " " + cn.getID() + " " + same);
            return same;
        } else return false;
    }
}
