DagTask.java

package mvd.jester.model;

import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import org.jgrapht.Graphs;
import org.jgrapht.experimental.dag.DirectedAcyclicGraph;
import org.jgrapht.graph.DefaultEdge;
import org.jgrapht.traverse.BreadthFirstIterator;

public class DagTask implements Task {

    private DirectedAcyclicGraph<Job, DefaultEdge> jobDag;
    private final Set<Segment> workloadDistribution;
    private final long workload;
    private final long criticalPath;
    private final long period;
    private final long deadline;
    private final long numberOfThreads;

    public DagTask(DirectedAcyclicGraph<Job, DefaultEdge> jobDag, long period,
            long numberOfThreads) {
        this.jobDag = jobDag;
        this.period = period;
        this.deadline = period;
        this.numberOfThreads = numberOfThreads;
        this.workload = DagUtils.calculateWorkload(this.jobDag);
        this.criticalPath = DagUtils.calculateCriticalPath(this.jobDag);
        this.workloadDistribution =
                DagUtils.calculateWorkloadDistribution(this.jobDag, this.criticalPath);
    }

    public double getUtilization() {
        return (double) workload / period;
    }

    /**
     * @return the deadline
     */
    public long getDeadline() {
        return deadline;
    }

    /**
     * @return the jobDag
     */
    public DirectedAcyclicGraph<Job, DefaultEdge> getJobDag() {
        return jobDag;
    }

    /**
     * @param jobDag the jobDag to set
     */
    public void setJobDag(DirectedAcyclicGraph<Job, DefaultEdge> jobDag) {
        this.jobDag = jobDag;
    }

    /**
     * @return the period
     */
    public long getPeriod() {
        return period;
    }

    /**
     * @return the workload
     */
    public long getWorkload() {
        return workload;
    }

    /**
     * @return the criticalPath
     */
    public long getCriticalPath() {
        return criticalPath;
    }


    /**
     * @return the workloadDistribution
     */
    public Set<Segment> getWorkloadDistribution() {
        return workloadDistribution;
    }

    @Override
    public long getMaximumParallelism() {
        long max = 0;
        for (Segment s : workloadDistribution) {
            if (max < s.getNumberOfJobs()) {
                max = s.getNumberOfJobs();
            }
        }
        return max;
    }


    @Override
    public long getNumberOfThreads() {
        return numberOfThreads;
    }

    public static class DagUtils {
        public static long calculateWorkload(DirectedAcyclicGraph<Job, DefaultEdge> jobDag) {
            long workload = 0;

            for (Job job : jobDag) {
                workload += job.getWcet();
            }
            return workload;
        }

        public static long calculateCriticalPath(DirectedAcyclicGraph<Job, DefaultEdge> jobDag) {
            long criticalPath = 0;
            for (Job job : jobDag) {
                Set<DefaultEdge> edges = jobDag.incomingEdgesOf(job);
                long longestRelativeCompletionTime = 0;
                for (DefaultEdge e : edges) {
                    Job source = jobDag.getEdgeSource(e);
                    longestRelativeCompletionTime =
                            longestRelativeCompletionTime >= source.getRelativeCompletionTime()
                                    ? longestRelativeCompletionTime
                                    : source.getRelativeCompletionTime();
                }

                job.setRelativeCompletionTime(longestRelativeCompletionTime + job.getWcet());
                criticalPath = job.getRelativeCompletionTime();
            }

            return criticalPath;
        }

        public static LinkedHashSet<Segment> calculateWorkloadDistribution(
                DirectedAcyclicGraph<Job, DefaultEdge> jobDag, long criticalPath) {
            LinkedHashSet<Segment> segments = new LinkedHashSet<>();
            long segmentDuration = 0;
            long segmentHeight = 1;
            for (long t = 0; t < criticalPath; ++t) {
                long currentHeight = 0;
                for (Job j : jobDag) {
                    if (t >= j.getRelativeCompletionTime() - j.getWcet()
                            && t < j.getRelativeCompletionTime()) {
                        currentHeight++;
                    }
                }
                if (currentHeight == segmentHeight) {
                    segmentDuration++;
                } else {
                    segments.add(new Segment(segmentDuration, segmentHeight));
                    segmentDuration = 1;
                    segmentHeight = currentHeight;
                }
            }
            segments.add(new Segment(segmentDuration, segmentHeight));

            return segments;
        }

        public static DirectedAcyclicGraph<Job, DefaultEdge> createNFJGraph(
                DirectedAcyclicGraph<Job, DefaultEdge> jobDag) {
            DirectedAcyclicGraph<Job, DefaultEdge> modifiedJobDag =
                    new DirectedAcyclicGraph<>(DefaultEdge.class);
            Graphs.addGraph(modifiedJobDag, jobDag);
            LinkedList<Job> joinNodes = new LinkedList<>();
            List<Job> forkNodes = new LinkedList<>();

            BreadthFirstIterator<Job, DefaultEdge> breadthFirstIterator =
                    new BreadthFirstIterator<>(modifiedJobDag);
            while (breadthFirstIterator.hasNext()) {
                Job j = breadthFirstIterator.next();
                if (modifiedJobDag.inDegreeOf(j) > 1) {
                    joinNodes.add(j);
                }
                if (modifiedJobDag.outDegreeOf(j) > 1) {
                    forkNodes.add(j);
                }
            }

            Job sink = joinNodes.getLast();

            for (Job j : joinNodes) {
                Set<DefaultEdge> edgeSet = new HashSet<>(modifiedJobDag.incomingEdgesOf(j));
                for (DefaultEdge e : edgeSet) {
                    Job predecessor = modifiedJobDag.getEdgeSource(e);
                    boolean satisfiesProposition =
                            DagUtils.checkForFork(modifiedJobDag, j, forkNodes, predecessor);
                    if (!satisfiesProposition) {
                        modifiedJobDag.removeEdge(e);
                        if (modifiedJobDag.outgoingEdgesOf(predecessor).isEmpty()) {
                            try {
                                modifiedJobDag.addDagEdge(predecessor, sink);
                            } catch (Exception ex) {
                            }
                        }
                    }

                    if (modifiedJobDag.inDegreeOf(j) == 1) {
                        break;
                    }
                }
                // Find fork node f following the path along this edge e
                // if f has successor that is not ancestor of j -> e is conflicting edge
                // get sorcetarget of e
                // remove e
                // if sourcetarget has no successor -> connect sourcetraget to sink
                // if indegree = 1 -> break;
            }

            // if (!DagUtils.checkProperty1(modifiedJobDag)) {
            // throw new RuntimeException("abs");
            // }
            return modifiedJobDag;
        }

        private static boolean checkProperty1(DirectedAcyclicGraph<Job, DefaultEdge> jobDag) {
            LinkedList<Job> joinNodes = new LinkedList<>();
            List<Job> forkNodes = new LinkedList<>();

            BreadthFirstIterator<Job, DefaultEdge> breadthFirstIterator =
                    new BreadthFirstIterator<>(jobDag);
            while (breadthFirstIterator.hasNext()) {
                Job j = breadthFirstIterator.next();
                if (jobDag.inDegreeOf(j) > 1) {
                    joinNodes.add(j);
                }
                if (jobDag.outDegreeOf(j) > 1) {
                    forkNodes.add(j);
                }
            }

            for (Job j : joinNodes) {
                nextFork: for (Job f : forkNodes) {
                    Set<DefaultEdge> edgeSet = jobDag.getAllEdges(f, j);

                    for (DefaultEdge e : edgeSet) {
                        Job a = jobDag.getEdgeSource(e);
                        if (a != f) {
                            Set<Job> succAndPred = new HashSet<>();
                            succAndPred.addAll(Graphs.predecessorListOf(jobDag, a));
                            succAndPred.addAll(Graphs.successorListOf(jobDag, a));

                            for (Job b : succAndPred) {
                                if (!((jobDag.getAncestors(jobDag, j).contains(b) || b == j)
                                        && (jobDag.getDescendants(jobDag, f).contains(b)
                                                || b == f))) {
                                    continue nextFork;
                                }
                            }
                        }
                    }
                    return true;
                }
            }
            return false;
        }

        private static boolean checkForFork(DirectedAcyclicGraph<Job, DefaultEdge> jobDag,
                Job joinNode, List<Job> forkNodes, Job job) {
            List<Job> pred = Graphs.predecessorListOf(jobDag, job);

            for (Job p : pred) {
                if (forkNodes.contains(p)) {
                    for (DefaultEdge successorEdge : jobDag.outgoingEdgesOf(p)) {
                        Job successor = jobDag.getEdgeSource(successorEdge);
                        if (jobDag.getAncestors(jobDag, joinNode).contains(successor)) {
                            return false;
                        }
                    }
                } else {
                    return checkForFork(jobDag, joinNode, forkNodes, p);
                }
            }

            return true;
        }

    }
}