<commit_before>package org.apache.hadoop.mapred.workflow.scheduling;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.ResourceStatus;
import org.apache.hadoop.mapred.workflow.MachineType;
import org.apache.hadoop.mapred.workflow.TimePriceTable.TableEntry;
import org.apache.hadoop.mapred.workflow.TimePriceTable.TableKey;
import org.apache.hadoop.mapred.workflow.WorkflowConf;

/**
 * Class representing a workflow directed acyclic graph.
 */
public class WorkflowDAG implements Writable {
  
  private static final Log LOG = LogFactory.getLog(WorkflowDAG.class);

  private Set<WorkflowNode> nodes;
  private Set<WorkflowNode> entryNodes = null;
  private Set<WorkflowNode> exitNodes = null;
  private Map<WorkflowNode, Set<WorkflowNode>> successors;
  private Map<WorkflowNode, Set<WorkflowNode>> predecessors;

  public WorkflowDAG() {
    nodes = new HashSet<WorkflowNode>();
    successors = new HashMap<WorkflowNode, Set<WorkflowNode>>();
    predecessors = new HashMap<WorkflowNode, Set<WorkflowNode>>();
  }

  private void addNode(WorkflowNode node) {
    // Initialize the successors and predecessors of a node before it is added.
    successors.put(node, new HashSet<WorkflowNode>());
    predecessors.put(node, new HashSet<WorkflowNode>());

    // Add the node to the workflow.
    nodes.add(node);
  }

  private void removeNode(WorkflowNode node) {
    // Remove pointers to the node, and pointers from the node.
    for (WorkflowNode linkedNode : nodes) {
      successors.get(linkedNode).remove(node);
      predecessors.get(linkedNode).remove(node);
    }
    successors.remove(node);
    predecessors.remove(node);

    // Remove the node itself.
    nodes.remove(node);
  }

  private void addSuccessor(WorkflowNode node, WorkflowNode successor) {
    successors.get(node).add(successor);
  }

  private void addPredecessor(WorkflowNode node, WorkflowNode predecessor) {
    predecessors.get(node).add(predecessor);
  }

  /**
   * Return the predecessors of the specified {@link WorkflowNode}.
   */
  public Set<WorkflowNode> getPredecessors(WorkflowNode node) {
    return predecessors.get(node);
  }

  /**
   * Return the successors of the specified {@link WorkflowNode}.
   */
  public Set<WorkflowNode> getSuccessors(WorkflowNode node) {
    return successors.get(node);
  }

  /**
   * Return the set of nodes for the workflow.
   */
  public Set<WorkflowNode> getNodes() {
    return nodes;
  }

  /**
   * Get the set of entry nodes for the workflow.
   */
  public Set<WorkflowNode> getEntryNodes() {
    if (entryNodes == null) {
      entryNodes = new HashSet<WorkflowNode>();
      for (WorkflowNode node : nodes) {
        if (predecessors.get(node).size() == 0) {
          entryNodes.add(node);
        }
      }
    }
    return entryNodes;
  }

  /**
   * Get the set of exit nodes for the workflow.
   */
  public Set<WorkflowNode> getExitNodes() {
    if (exitNodes == null) {
      exitNodes = new HashSet<WorkflowNode>();
      for (WorkflowNode node : nodes) {
        if (successors.get(node).size() == 0) {
          exitNodes.add(node);
        }
      }
    }
    return exitNodes;
  }

  /**
   * Using the current node-machineType pairings, return a critical path.
   *
   * @param table A time-price table to use for computation of times/prices.
   *
   * @return A list of {@link WorkflowNode}s on the critical path.
   */
  public List<WorkflowNode> getCriticalPath(Map<TableKey, TableEntry> table) {

    LOG.info("Computing Critical path.");
    Map<WorkflowNode, Float> distances = getWorkflowNodeWeights(table);
    List<WorkflowNode> criticalPath = new ArrayList<WorkflowNode>();

    // Get the exit nodes before adding the fake node so it is not included.
    Set<WorkflowNode> exitNodes = getExitNodes();

    // Add a fake exit node connecting to all real exit nodes.
    // This allows computation of only one path, rather than multiple paths.
    WorkflowNode fakeExitNode = new WorkflowNode("fakeNode", 0, 0);
    this.addNode(fakeExitNode);
    LOG.info("Added fake exit node.");

    for (WorkflowNode exit : exitNodes) {
      this.addPredecessor(fakeExitNode, exit);
      LOG.info("Added " + exit.getJobName()
          + " as predecessor of fake exit node.");
    }

    WorkflowNode criticalNode = fakeExitNode;
    do {
      LOG.info("Creating critical path, added " + criticalNode.getJobName());
      criticalNode = getNextCriticalNode(distances, criticalNode);
      criticalPath.add(0, criticalNode);
    } while (criticalNode != null);

    // Null check is after adding, so remove the null element.
    criticalPath.remove(0);
    removeNode(fakeExitNode);
    LOG.info("Removed fake exit node.");

    return criticalPath;
  }

  private WorkflowNode getNextCriticalNode(
      Map<WorkflowNode, Float> distances, WorkflowNode current) {

    float maxDistance = 0;
    WorkflowNode criticalNode = null;

    for (WorkflowNode predecessor : getPredecessors(current)) {
      LOG.info("Checking distance of predecessor " + predecessor.getJobName()
          + " of node " + current.getJobName());
      float distance = distances.get(predecessor);
      if (distance > maxDistance) {
        maxDistance = distance;
        criticalNode = predecessor;
      }
    }
    return criticalNode;
  }

  /**
   * Compute the weights of WorkflowNodes. Each distance is the value of the
   * longest path from a source node to the node, as measured by maximum
   * execution time.
   *
   * @param table A time-price table to use for time/price computations.
   *
   * @return A map of {@link WorkflowNode} to their weightings.
   */
  private Map<WorkflowNode, Float> getWorkflowNodeWeights(
      Map<TableKey, TableEntry> table) {

    LOG.info("Computing workflow node weights.");
    Map<WorkflowNode, Float> distances = new HashMap<WorkflowNode, Float>();
    List<WorkflowNode> ordering = getTopologicalOrdering();

    // Initialize (time) distances.
    for (WorkflowNode node : getNodes()) {
      distances.put(node, Float.MIN_VALUE);
    }
    for (WorkflowNode entry : getEntryNodes()) {
      float maxTime = getNodeMaxTime(table, entry);
      distances.put(entry, maxTime);
      LOG.info("Updated entry node '" + entry.getJobName() + "' weight to "
          + maxTime);
    }
    LOG.info("Initialized node weights.");

    // Relax the nodes to find their proper weight.
    for (WorkflowNode node : ordering) {
      for (WorkflowNode next : getSuccessors(node)) {
        // Add the weight of executing the next node.
        float otherPath = distances.get(node) + getNodeMaxTime(table, next);

        if (distances.get(next) < otherPath) {
          distances.put(next, otherPath);
          LOG.info("Updated " + next.getJobName() + " weight to " + otherPath);
        }
      }
    }

    return distances;
  }

  // Get the execution time of a node.
  // This is the sum of the slowest contained map + reduce tasks.
  private float getNodeMaxTime(Map<TableKey, TableEntry> table,
      WorkflowNode node) {

    float maxMapWeight = 0f;
    float maxRedWeight = 0f;

    for (WorkflowTask task : node.getTasks()) {

      String type = task.getMachineType();
      TableKey key = new TableKey(node.getJobName(), type, task.isMapTask());
      float weight = table.get(key).execTime;

      if (task.isMapTask() && weight > maxMapWeight) {
        maxMapWeight = weight;
      } else if (!task.isMapTask() && weight > maxRedWeight) {
        maxRedWeight = weight;
      }
    }

    return (maxMapWeight + maxRedWeight);
  }

  /**
   * Compute and return the total execution time (makespan) of a WorkflowDAG.
   */
  public float getTime(Map<TableKey, TableEntry> table) {
    // Get the critical path.
    List<WorkflowNode> criticalPath = getCriticalPath(table);

    // Compute the execution time along the path.
    float time = 0f;
    for (WorkflowNode node : criticalPath) {
      time += getNodeMaxTime(table, node);
    }

    return time;
  }

  /**
   * Compute and return the total cost of a WorkflowDAG.
   */
  public float getCost(Map<TableKey, TableEntry> table) throws IOException {
    // Add up the cost of all the nodes/tasks in the dag.
    float cost = 0f;
    for (WorkflowNode node : getNodes()) {
      for (WorkflowTask task : node.getTasks()) {
        String type = task.getMachineType();
        TableKey key = new TableKey(node.getJobName(), type, task.isMapTask());
        TableEntry entry = table.get(key);
        if (entry != null) {
          cost += entry.cost;
        } else {
          // IOException because info wasn't read from the configuration file.<commit_msg>Completed null checks in WorkflowDAG wrt/ the TimePrice table.<commit_after>package org.apache.hadoop.mapred.workflow.scheduling;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.ResourceStatus;
import org.apache.hadoop.mapred.workflow.MachineType;
import org.apache.hadoop.mapred.workflow.TimePriceTable.TableEntry;
import org.apache.hadoop.mapred.workflow.TimePriceTable.TableKey;
import org.apache.hadoop.mapred.workflow.WorkflowConf;

/**
 * Class representing a workflow directed acyclic graph.
 */
public class WorkflowDAG implements Writable {
  
  private static final Log LOG = LogFactory.getLog(WorkflowDAG.class);

  private Set<WorkflowNode> nodes;
  private Set<WorkflowNode> entryNodes = null;
  private Set<WorkflowNode> exitNodes = null;
  private Map<WorkflowNode, Set<WorkflowNode>> successors;
  private Map<WorkflowNode, Set<WorkflowNode>> predecessors;

  public WorkflowDAG() {
    nodes = new HashSet<WorkflowNode>();
    successors = new HashMap<WorkflowNode, Set<WorkflowNode>>();
    predecessors = new HashMap<WorkflowNode, Set<WorkflowNode>>();
  }

  private void addNode(WorkflowNode node) {
    // Initialize the successors and predecessors of a node before it is added.
    successors.put(node, new HashSet<WorkflowNode>());
    predecessors.put(node, new HashSet<WorkflowNode>());

    // Add the node to the workflow.
    nodes.add(node);
  }

  private void removeNode(WorkflowNode node) {
    // Remove pointers to the node, and pointers from the node.
    for (WorkflowNode linkedNode : nodes) {
      successors.get(linkedNode).remove(node);
      predecessors.get(linkedNode).remove(node);
    }
    successors.remove(node);
    predecessors.remove(node);

    // Remove the node itself.
    nodes.remove(node);
  }

  private void addSuccessor(WorkflowNode node, WorkflowNode successor) {
    successors.get(node).add(successor);
  }

  private void addPredecessor(WorkflowNode node, WorkflowNode predecessor) {
    predecessors.get(node).add(predecessor);
  }

  /**
   * Return the predecessors of the specified {@link WorkflowNode}.
   */
  public Set<WorkflowNode> getPredecessors(WorkflowNode node) {
    return predecessors.get(node);
  }

  /**
   * Return the successors of the specified {@link WorkflowNode}.
   */
  public Set<WorkflowNode> getSuccessors(WorkflowNode node) {
    return successors.get(node);
  }

  /**
   * Return the set of nodes for the workflow.
   */
  public Set<WorkflowNode> getNodes() {
    return nodes;
  }

  /**
   * Get the set of entry nodes for the workflow.
   */
  public Set<WorkflowNode> getEntryNodes() {
    if (entryNodes == null) {
      entryNodes = new HashSet<WorkflowNode>();
      for (WorkflowNode node : nodes) {
        if (predecessors.get(node).size() == 0) {
          entryNodes.add(node);
        }
      }
    }
    return entryNodes;
  }

  /**
   * Get the set of exit nodes for the workflow.
   */
  public Set<WorkflowNode> getExitNodes() {
    if (exitNodes == null) {
      exitNodes = new HashSet<WorkflowNode>();
      for (WorkflowNode node : nodes) {
        if (successors.get(node).size() == 0) {
          exitNodes.add(node);
        }
      }
    }
    return exitNodes;
  }

  /**
   * Using the current node-machineType pairings, return a critical path.
   *
   * @param table A time-price table to use for computation of times/prices.
   *
   * @return A list of {@link WorkflowNode}s on the critical path.
   */
  public List<WorkflowNode> getCriticalPath(Map<TableKey, TableEntry> table)
      throws IOException {

    LOG.info("Computing Critical path.");
    Map<WorkflowNode, Float> distances = getWorkflowNodeWeights(table);
    List<WorkflowNode> criticalPath = new ArrayList<WorkflowNode>();

    // Get the exit nodes before adding the fake node so it is not included.
    Set<WorkflowNode> exitNodes = getExitNodes();

    // Add a fake exit node connecting to all real exit nodes.
    // This allows computation of only one path, rather than multiple paths.
    WorkflowNode fakeExitNode = new WorkflowNode("fakeNode", 0, 0);
    this.addNode(fakeExitNode);
    LOG.info("Added fake exit node.");

    for (WorkflowNode exit : exitNodes) {
      this.addPredecessor(fakeExitNode, exit);
      LOG.info("Added " + exit.getJobName()
          + " as predecessor of fake exit node.");
    }

    WorkflowNode criticalNode = fakeExitNode;
    do {
      LOG.info("Creating critical path, added " + criticalNode.getJobName());
      criticalNode = getNextCriticalNode(distances, criticalNode);
      criticalPath.add(0, criticalNode);
    } while (criticalNode != null);

    // Null check is after adding, so remove the null element.
    criticalPath.remove(0);
    removeNode(fakeExitNode);
    LOG.info("Removed fake exit node.");

    return criticalPath;
  }

  private WorkflowNode getNextCriticalNode(
      Map<WorkflowNode, Float> distances, WorkflowNode current) {

    float maxDistance = 0;
    WorkflowNode criticalNode = null;

    for (WorkflowNode predecessor : getPredecessors(current)) {
      LOG.info("Checking distance of predecessor " + predecessor.getJobName()
          + " of node " + current.getJobName());
      float distance = distances.get(predecessor);
      if (distance > maxDistance) {
        maxDistance = distance;
        criticalNode = predecessor;
      }
    }
    return criticalNode;
  }

  /**
   * Compute the weights of WorkflowNodes. Each distance is the value of the
   * longest path from a source node to the node, as measured by maximum
   * execution time.
   *
   * @param table A time-price table to use for time/price computations.
   *
   * @return A map of {@link WorkflowNode} to their weightings.
   */
  private Map<WorkflowNode, Float> getWorkflowNodeWeights(
      Map<TableKey, TableEntry> table) throws IOException {

    LOG.info("Computing workflow node weights.");
    Map<WorkflowNode, Float> distances = new HashMap<WorkflowNode, Float>();
    List<WorkflowNode> ordering = getTopologicalOrdering();

    // Initialize (time) distances.
    for (WorkflowNode node : getNodes()) {
      distances.put(node, Float.MIN_VALUE);
    }
    for (WorkflowNode entry : getEntryNodes()) {
      float maxTime = getNodeMaxTime(table, entry);
      distances.put(entry, maxTime);
      LOG.info("Updated entry node '" + entry.getJobName() + "' weight to "
          + maxTime);
    }
    LOG.info("Initialized node weights.");

    // Relax the nodes to find their proper weight.
    for (WorkflowNode node : ordering) {
      for (WorkflowNode next : getSuccessors(node)) {
        // Add the weight of executing the next node.
        float otherPath = distances.get(node) + getNodeMaxTime(table, next);

        if (distances.get(next) < otherPath) {
          distances.put(next, otherPath);
          LOG.info("Updated " + next.getJobName() + " weight to " + otherPath);
        }
      }
    }

    return distances;
  }

  // Get the execution time of a node.
  // This is the sum of the slowest contained map + reduce tasks.
  private float getNodeMaxTime(Map<TableKey, TableEntry> table,
      WorkflowNode node) throws IOException {

    float maxMapWeight = 0f;
    float maxRedWeight = 0f;

    for (WorkflowTask task : node.getTasks()) {

      String type = task.getMachineType();
      TableKey key = new TableKey(node.getJobName(), type, task.isMapTask());
      TableEntry entry = table.get(key);
      if (entry != null) {
        float weight = entry.execTime;

        if (task.isMapTask() && weight > maxMapWeight) {
          maxMapWeight = weight;
        } else if (!task.isMapTask() && weight > maxRedWeight) {
          maxRedWeight = weight;
        }
      } else {
        // IOException because info wasn't read from the configuration file.
        throw new IOException("Entry for " + node.getJobName() + " ("
            + (task.isMapTask() ? "map" : "reduce") + ") / " + type
            + " does not exist in the time price table.");
      }
    }

    return (maxMapWeight + maxRedWeight);
  }

  /**
   * Compute and return the total execution time (makespan) of a WorkflowDAG.
   */
  public float getTime(Map<TableKey, TableEntry> table) throws IOException {
    // Get the critical path.
    List<WorkflowNode> criticalPath = getCriticalPath(table);

    // Compute the execution time along the path.
    float time = 0f;
    for (WorkflowNode node : criticalPath) {
      time += getNodeMaxTime(table, node);
    }

    return time;
  }

  /**
   * Compute and return the total cost of a WorkflowDAG.
   */
  public float getCost(Map<TableKey, TableEntry> table) throws IOException {
    // Add up the cost of all the nodes/tasks in the dag.
    float cost = 0f;
    for (WorkflowNode node : getNodes()) {
      for (WorkflowTask task : node.getTasks()) {
        String type = task.getMachineType();
        TableKey key = new TableKey(node.getJobName(), type, task.isMapTask());
        TableEntry entry = table.get(key);
        if (entry != null) {
          cost += entry.cost;
        } else {
          // IOException because info wasn't read from the configuration file.
{'fraction_non_alphanumeric': 0.10782561603032764, 'fraction_numerical': 0.0011914432710533442, 'mean_word_length': 3.1026438569206842, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1220859', 'n_tokens_mistral': 5516, 'n_tokens_neox': 5357, 'n_words': 1536}