FonsecaNelis.java

package mvd.jester.tests;

import java.math.RoundingMode;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.math.LongMath;
import org.jgrapht.experimental.dag.DirectedAcyclicGraph;
import org.jgrapht.graph.DefaultEdge;
import mvd.jester.info.SchedulingInfo;
import mvd.jester.info.TerminationInfo;
import mvd.jester.model.DagTask;
import mvd.jester.model.Job;
import mvd.jester.model.Segment;
import mvd.jester.model.SortedTaskSet;
import mvd.jester.model.Task;
import mvd.jester.model.TreeJob;
import mvd.jester.utils.DagUtils;
import mvd.jester.priority.PriorityManager;
import mvd.jester.priority.RateMonotonic;
import mvd.jester.utils.BinaryDecompositionTree;
import mvd.jester.utils.BinaryDecompositionTree.Node;
import mvd.jester.utils.BinaryDecompositionTree.NodeType;

public class FonsecaNelis extends AbstractTest<DagTask> {

    private final Map<Task, TerminationInfo> responseTimes;
    private final PriorityManager priorityManager;
    private final Map<Task, Set<Segment>> carryOutSegments;

    public FonsecaNelis(final long numberOfProcessors) {
        super(numberOfProcessors);
        this.responseTimes = new HashMap<>();
        this.priorityManager = new RateMonotonic();
        this.carryOutSegments = new HashMap<>();
    }

    @Override
    public PriorityManager getPriorityManager() {
        return priorityManager;
    }

    @Override
    public SchedulingInfo runSchedulabilityCheck(final SortedTaskSet<DagTask> tasks) {
        carryOutSegments.clear();
        responseTimes.clear();
        createNFJandDecompositionTree(tasks);
        for (final DagTask t : tasks) {
            final long responseTime = calculateResponseTime(tasks, t);
            responseTimes.put(t, new TerminationInfo(t.getDeadline(), responseTime));
        }

        return new SchedulingInfo(responseTimes.values());
    }

    private void createNFJandDecompositionTree(final SortedTaskSet<DagTask> tasks) {
        for (final DagTask t : tasks) {
            final DirectedAcyclicGraph<Job, DefaultEdge> jobDag = t.getJobDag();

            final DirectedAcyclicGraph<Job, DefaultEdge> nfjJobDag =
                    DagUtils.createNFJGraph(jobDag);
            final BinaryDecompositionTree<Job> tree = DagUtils.createDecompositionTree(nfjJobDag);

            carryOutSegments.put(t, constructCarryOutDistribution(nfjJobDag, tree));
        }
    }

    private Set<Segment> constructCarryOutDistribution(
            final DirectedAcyclicGraph<Job, DefaultEdge> nfjDag,
            final BinaryDecompositionTree<Job> tree) {
        final Set<Segment> carryOutWorkload = new LinkedHashSet<>();
        final BinaryDecompositionTree<TreeJob> modifiedTree = transformTree(tree);

        boolean isEmpty = false;
        do {
            final Set<TreeJob> parallelJobs = getMaximumParallelism(modifiedTree.getRootNode());
            final Optional<TreeJob> min =
                    parallelJobs.stream().min((p1, p2) -> Long.compare(p1.getWcet(), p2.getWcet()));
            if (min.isPresent()) {
                final long width = min.get().getWcet();
                carryOutWorkload.add(new Segment(width, parallelJobs.size()));
                for (final TreeJob p : parallelJobs) {
                    p.setWcet(p.getWcet() - width);
                }
            } else {
                break;
            }
            isEmpty = parallelJobs.isEmpty();
        } while (!isEmpty);

        return carryOutWorkload;
    }


    private BinaryDecompositionTree<TreeJob> transformTree(
            final BinaryDecompositionTree<Job> tree) {
        final BinaryDecompositionTree<TreeJob> modifiedTree = new BinaryDecompositionTree<>();
        final Node<TreeJob> root = transformNode(null, tree.getRootNode());
        modifiedTree.setRoot(root);
        return modifiedTree;
    }

    private Node<TreeJob> transformNode(final Node<TreeJob> parent, final Node<Job> node) {
        if (node.getNodeType().equals(NodeType.LEAF)) {
            return new Node<TreeJob>(parent, new TreeJob(node.getObject()));
        } else {
            final Node<TreeJob> modifiedNode = new Node<TreeJob>(parent, node.getNodeType());
            modifiedNode.setLeftNode(transformNode(modifiedNode, node.getLeftNode()));
            modifiedNode.setRightNode(transformNode(modifiedNode, node.getRightNode()));
            return modifiedNode;
        }
    }

    private Set<TreeJob> getMaximumParallelism(final Node<TreeJob> node) {
        final NodeType nodeType = node.getNodeType();
        if (nodeType.equals(NodeType.PAR)) {
            final Set<TreeJob> left = getMaximumParallelism(node.getLeftNode());
            final Set<TreeJob> right = getMaximumParallelism(node.getRightNode());
            return Sets.union(left, right);
        } else if (nodeType.equals(NodeType.SEQ)) {
            final Set<TreeJob> left = getMaximumParallelism(node.getLeftNode());
            final Set<TreeJob> right = getMaximumParallelism(node.getRightNode());
            if (left.size() >= right.size()) {
                return left;
            } else {
                return right;
            }
        } else {
            if (node.getObject().getWcet() > 0) {
                return new HashSet<>(Arrays.asList(node.getObject()));
            } else {
                return new HashSet<>();
            }
        }
    }

    private long calculateResponseTime(final SortedTaskSet<DagTask> tasks, final DagTask task) {
        final long criticalPath = task.getCriticalPath();
        long responseTime = criticalPath;
        long previousResponseTime = 0;

        do {
            previousResponseTime = responseTime;
            double taskInterference = 0;
            for (final DagTask t : tasks) {
                if (t.getPeriod() < task.getPeriod()) {
                    taskInterference += getTaskInterference(t, responseTime);
                }
            }

            taskInterference /= numberOfProcessors;

            final double selfInterference = getSelfInterference(task);

            final long totalInterference = (long) Math.floor(taskInterference + selfInterference);
            responseTime = criticalPath + totalInterference;

        } while (responseTime != previousResponseTime);

        return responseTime;
    }

    private double getTaskInterference(final DagTask task, final long interval) {
        final long period = task.getPeriod();
        final long criticalPath = task.getCriticalPath();
        final long numberOfPeriods =
                LongMath.divide(interval - criticalPath, period, RoundingMode.FLOOR);
        final long carryInAndOutInterval = interval - Math.max(0, numberOfPeriods) * period;
        final long numberOfBodyJobs =
                LongMath.divide(interval - carryInAndOutInterval, period, RoundingMode.FLOOR);

        final long bodyWorkload = Math.max(0, numberOfBodyJobs) * task.getWorkload();


        final long carryInAndOutWorkload = getCarryInAndOutWorkload(task,
                task.getWorkloadDistribution(), carryOutSegments.get(task), carryInAndOutInterval);

        return carryInAndOutWorkload + bodyWorkload;
    }

    private long getCarryInAndOutWorkload(final DagTask task,
            final Set<Segment> carryInDistribution, final Set<Segment> carryOutDistribution,
            final long carryInAndOutPeriod) {
        long workload = getCarryOutWorkload(task, carryOutDistribution, carryInAndOutPeriod);
        long carryInPeriod = task.getPeriod() - responseTimes.get(task).getResponseTime();
        long carryOutPeriod = 0;
        final List<Segment> reverseCarryInList = Lists.newArrayList(carryInDistribution);
        Collections.reverse(reverseCarryInList);

        for (final Segment s : reverseCarryInList) {
            carryInPeriod += s.getJobWcet();
            carryOutPeriod = carryInAndOutPeriod - carryInPeriod;
            final long carryInWorkload =
                    getCarryInWorkload(task, carryInDistribution, carryInPeriod);
            final long carryOutWorkload =
                    getCarryOutWorkload(task, carryOutDistribution, carryOutPeriod);
            workload = Math.max(workload, carryInWorkload + carryOutWorkload);
        }

        workload = Math.max(workload,
                getCarryInWorkload(task, carryInDistribution, carryInAndOutPeriod));
        carryOutPeriod = 0;

        for (final Segment s : carryOutDistribution) {
            carryOutPeriod += s.getJobWcet();
            carryInPeriod = carryInAndOutPeriod - carryOutPeriod;
            final long carryInWorkload =
                    getCarryInWorkload(task, carryInDistribution, carryInPeriod);
            final long carryOutWorkload =
                    getCarryOutWorkload(task, carryOutDistribution, carryOutPeriod);
            workload = Math.max(workload, carryInWorkload + carryOutWorkload);
        }

        return workload;
    }

    private long getCarryOutWorkload(final DagTask task, final Set<Segment> carryOutDistribution,
            final long carryOutPeriod) {
        long workload = 0;
        final long taskWorkload = task.getWorkload();
        final long criticalPath = task.getCriticalPath();
        final List<Segment> distributionList = Lists.newArrayList(carryOutDistribution);

        for (int i = 0; i < distributionList.size(); ++i) {
            final Segment s = distributionList.get(i);
            long weightOfPreviousSegments = 0;
            for (int j = 0; j < i; ++j) {
                weightOfPreviousSegments += distributionList.get(j).getJobWcet();
            }
            final long width = carryOutPeriod - weightOfPreviousSegments;

            workload += Math.max(Math.min(width, s.getJobWcet()), 0) * s.getNumberOfJobs();
        }

        final long improvedWorkloadFromTask =
                taskWorkload - Math.max(0, criticalPath - carryOutPeriod);
        final long improvedWorkloadFromProcessors = carryOutPeriod * numberOfProcessors;

        return Math.min(Math.min(improvedWorkloadFromTask, improvedWorkloadFromProcessors),
                workload);
    }

    private long getCarryInWorkload(final DagTask task, final Set<Segment> carryInDistribution,
            final long carryInPeriod) {
        long workload = 0;
        final long period = task.getPeriod();
        final long responseTime = responseTimes.get(task).getResponseTime();
        final List<Segment> distributionList = Lists.newArrayList(carryInDistribution);

        for (int b = 0; b < carryInDistribution.size(); ++b) {
            final Segment s = distributionList.get(b);
            long weightOfRemainingSegments = 0;
            for (int p = b + 1; p < carryInDistribution.size(); ++p) {
                weightOfRemainingSegments += distributionList.get(p).getJobWcet();
            }
            final long width = carryInPeriod - period + responseTime - weightOfRemainingSegments;

            workload += Math.max(Math.min(width, s.getJobWcet()), 0) * s.getNumberOfJobs();
        }

        final long improvedWorkload =
                Math.max(carryInPeriod - (period - responseTime), 0) * numberOfProcessors;

        return Math.min(improvedWorkload, workload);
    }

    private double getSelfInterference(final DagTask task) {
        final long criticalPath = task.getCriticalPath();
        final long workload = task.getWorkload();

        return (double) (workload - criticalPath) / numberOfProcessors;
    }

    @Override
    public String getName() {
        return "FonsecaNelis";
    }


}