DolphinScheduler 源码分析之 DAG类

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.dolphinscheduler.common.graph;

import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * analysis of DAG
 * 对DAG图的分析
 * Node: node 指代一个节点，其实一般是Int类型的数字，比如 1
 * NodeInfo：node description information 指代节点的描述信息，其实一般是String类型 ，比如 “v(1)”
 * EdgeInfo: edge description information 其实一般是String类型 比如 “edge(1 -> 2)”
 */
public class DAG<Node, NodeInfo, EdgeInfo> {


  private static final Logger logger = LoggerFactory.getLogger(DAG.class);

  private final ReadWriteLock lock = new ReentrantReadWriteLock();

  /**
   * node map, key is node, value is node information
   * 节点映射，键是节点，值是节点信息
   */
  private volatile Map<Node, NodeInfo> nodesMap;

  /**
   * edge map. key is node of origin;value is Map with key for destination node and value for edge
   * 边的映射。key是起始节点，value是一个Map，这个Map又是以目标节点作为Key，边的信息作为value的。
   */
  private volatile Map<Node, Map<Node, EdgeInfo>> edgesMap;

  /**
   * reversed edge set，key is node of destination, value is Map with key for origin node and value for edge
   * 反转的边的映射。key是目标点，value是一个Map。这个Map又是以起始点为Key，边的信息作为value的。
   */
  private volatile Map<Node, Map<Node, EdgeInfo>> reverseEdgesMap;


  public DAG() {
    nodesMap = new HashMap<>();
    edgesMap = new HashMap<>();
    reverseEdgesMap = new HashMap<>();
  }


  /**
   * add node information
   * 添加一个节点
   *
   * @param node          node
   * @param nodeInfo      node information
   */
  public void addNode(Node node, NodeInfo nodeInfo) {
    lock.writeLock().lock();

    try{
      nodesMap.put(node, nodeInfo);
    }finally {
      lock.writeLock().unlock();
    }

  }


  /**
   * add edge
   * 添加边
   * @param fromNode node of origin 起始点
   * @param toNode   node of destination 目标点
   * @return The result of adding an edge. returns false if the DAG result is a ring result
   * 返回值是一个波尔类型的值，如果添加边之后会形成一个环图，那么就会返回false，并且不会添加这条边，如果添加边之后不会形成一个环图，那么就会返回true。
   */
  public boolean addEdge(Node fromNode, Node toNode) {
    return addEdge(fromNode, toNode, false);
  }


  /**
   * add edge
   * 添加边
   * @param fromNode        node of origin 起始点
   * @param toNode          node of destination 目标点
   * @param createNode      whether the node needs to be created if it does not exist
   *                        如果我添加新的边的时候，起始点或者目标点不存在。
   *                        那么我需要看这个createNode参数，如果这里传入true，代表同时可以建立Node，如果是false，代表不可以擅自新建Node。
   * @return The result of adding an edge. returns false if the DAG result is a ring result
   * 返回值是一个波尔类型的值，如果添加边之后会形成一个环图，那么就会返回false，并且不会添加这条边，如果添加边之后不会形成一个环图，那么就会返回true。
   */
  private boolean addEdge(Node fromNode, Node toNode, boolean createNode) {
    return addEdge(fromNode, toNode, null, createNode);
  }


  /**
   * add edge
   * 添加边
   *
   * @param fromNode        node of origin 起始节点
   * @param toNode          node of destination 目标节点
   * @param edge            edge description 边描述
   * @param createNode      whether the node needs to be created if it does not exist
   *                        如果我添加新的边的时候，起始点或者目标点不存在。
   *                        那么我需要看这个createNode参数，如果这里传入true，代表同时可以建立Node，如果是false，代表不可以擅自新建Node。
   * @return The result of adding an edge. returns false if the DAG result is a ring result
   * 返回值是一个波尔类型的值，如果添加边之后会形成一个环图，那么就会返回false，并且不会添加这条边，如果添加边之后不会形成一个环图，那么就会返回true。
   */
  public boolean addEdge(Node fromNode, Node toNode, EdgeInfo edge, boolean createNode) {
    lock.writeLock().lock();

    try{

      // Whether an edge can be successfully added(fromNode -> toNode)
      // 判断边是否可以成功被添加（起始点-->目标点）
      if (!isLegalAddEdge(fromNode, toNode, createNode)) {
        logger.error("serious error: add edge({} -> {}) is invalid, cause cycle！", fromNode, toNode);
        return false;
      }

      addNodeIfAbsent(fromNode, null);
      addNodeIfAbsent(toNode, null);

      addEdge(fromNode, toNode, edge, edgesMap);
      addEdge(toNode, fromNode, edge, reverseEdgesMap);

      return true;
    }finally {
      lock.writeLock().unlock();
    }

  }


  /**
   * whether this node is contained
   * 检查node的map中是否存在这个节点
   *
   * @param node node 节点
   * @return true if contains 返回true，如果存在的话
   */
  public boolean containsNode(Node node) {
    lock.readLock().lock();

    try{
      return nodesMap.containsKey(node);
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * whether this edge is contained
   * 检查是否存在边
   *
   * @param fromNode node of origin 起始点
   * @param toNode   node of destination 目标点
   * @return true if contains 返回true如果存在的话
   */
  public boolean containsEdge(Node fromNode, Node toNode) {
    lock.readLock().lock();
    try{
      Map<Node, EdgeInfo> endEdges = edgesMap.get(fromNode);
      if (endEdges == null) {
        return false;
      }

     return endEdges.containsKey(toNode);
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * get node description
   * 获取node的nodeInfo
   *
   * @param node node 要查询的node
   * @return node description 返回节点的描述信息
   */
  public NodeInfo getNode(Node node) {
    lock.readLock().lock();

    try{
      return nodesMap.get(node);
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * Get the number of nodes
   * 获取node的数量
   *
   * @return the number of nodes 返回node的数量
   */
  public int getNodesCount() {
    lock.readLock().lock();

    try{
      return nodesMap.size();
    }finally {
      lock.readLock().unlock();
    }
  }

  /**
   * Get the number of edges
   * 获取边的数量
   * @return the number of edges 返回边的数量
   */
  public int getEdgesCount() {
    lock.readLock().lock();
    try{
      int count = 0;

      for (Map.Entry<Node, Map<Node, EdgeInfo>> entry : edgesMap.entrySet()) {
        count += entry.getValue().size();
      }

      return count;
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * get the start node of DAG
   * 获取一幅图中只有出度，没有入度的节点的集合
   *
   * @return the start node of DAG 返回一幅图中只有出度，没有入度的节点的集合
   */
  public Collection<Node> getBeginNode() {
    lock.readLock().lock();

    try{
      return CollectionUtils.subtract(nodesMap.keySet(), reverseEdgesMap.keySet());
    }finally {
      lock.readLock().unlock();
    }

  }


  /**
   * get the end node of DAG
   * 获取一幅图中只有入度，没有出度的节点的集合
   *
   * @return the end node of DAG 返回一幅图中只有入度，没有出度的节点的集合
   */
  public Collection<Node> getEndNode() {

    lock.readLock().lock();

    try{
      return CollectionUtils.subtract(nodesMap.keySet(), edgesMap.keySet());
    }finally {
      lock.readLock().unlock();
    }

  }


  /**
   * Gets all previous nodes of the node
   * 传入一个node，返回所有指向这个node的node集合
   *
   * @param node node id to be calculated 传入要查询的node
   * @return all previous nodes of the node 返回所有指向这个node的node集合
   */
  public Set<Node> getPreviousNodes(Node node) {
    lock.readLock().lock();

    try{
      return getNeighborNodes(node, reverseEdgesMap);
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * Get all subsequent nodes of the node
   * 传入一个node，返回所有的该节点指向的node集合
   *
   * @param node node id to be calculated 传入要查询的node的id
   * @return all subsequent nodes of the node 返回的该节点指向的所有node
   */
  public Set<Node> getSubsequentNodes(Node node) {
    lock.readLock().lock();

    try{
      return getNeighborNodes(node, edgesMap);
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * Gets the degree of entry of the node
   * 计算一个节点的入度
   *
   * @param node node id 被计算的节点
   * @return the degree of entry of the node 节点的入度
   */
  public int getIndegree(Node node) {
    lock.readLock().lock();

    try{
      return getPreviousNodes(node).size();
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * whether the graph has a ring
   * 判断这个图是否存在环
   *
   * @return true if has cycle, else return false.
   * 如果存在环，返回true。没有形成环状，返回false
   */
  public boolean hasCycle() {
    lock.readLock().lock();
    try{
        return !topologicalSortImpl().getKey();
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   * Only DAG has a topological sort
   * 返回一个DAG图的拓扑排序
   * @return topologically sorted results, returns false if the DAG result is a ring result
   * 返回一个DAG图的拓扑排序，如果这个DAG其实是环图，那么返回false，也就是非DAG图不存在拓扑排序。
   * @throws Exception errors
   */
  public List<Node> topologicalSort() throws Exception {
    lock.readLock().lock();

    try{
      Map.Entry<Boolean, List<Node>> entry = topologicalSortImpl();

      if (entry.getKey()) {
        return entry.getValue();
      }

      throw new Exception("serious error: graph has cycle ! ");
    }finally {
      lock.readLock().unlock();
    }
  }


  /**
   *  if tho node does not exist,add this node
   * 如果节点不存在，则添加该节点
   * @param node    node 节点id
   * @param nodeInfo node information 节点描述信息
   */
  private void addNodeIfAbsent(Node node, NodeInfo nodeInfo) {
    if (!containsNode(node)) {
      addNode(node, nodeInfo);
    }
  }


  /**
   * add edge
   * 添加边
   *
   * @param fromNode node of origin 起始点
   * @param toNode   node of destination 目标点
   * @param edge  edge description 边的描述信息
   * @param edges edge set 所有边的集合
   */
  private void addEdge(Node fromNode, Node toNode, EdgeInfo edge, Map<Node, Map<Node, EdgeInfo>> edges) {
    edges.putIfAbsent(fromNode, new HashMap<>());
    Map<Node, EdgeInfo> toNodeEdges = edges.get(fromNode);
    toNodeEdges.put(toNode, edge);
  }


  /**
   * Whether an edge can be successfully added(fromNode -> toNode)
   * 判断是否新的 边 可以被成功添加进现有的图中（起始点-->目标点）
   * need to determine whether the DAG has cycle
   * 需要判断是否会形成一个环状图
   * @param fromNode     node of origin 起始点
   * @param toNode       node of destination 目标点
   * @param createNode whether to create a node
   *                   如果我添加新的边的时候，起始点或者目标点不存在。
   *                   那么我需要看这个createNode参数，如果这里传入true，代表同时可以建立Node，如果是false，代表不可以擅自新建Node。
   * @return true if added
   * 返回true，如果节点被成功添加，否则返回false。
   */
  private boolean isLegalAddEdge(Node fromNode, Node toNode, boolean createNode) {
      if (fromNode.equals(toNode)) {
          logger.error("edge fromNode({}) can't equals toNode({})", fromNode, toNode);
          return false;
      }

      if (!createNode) {
          if (!containsNode(fromNode) || !containsNode(toNode)){
              logger.error("edge fromNode({}) or toNode({}) is not in vertices map", fromNode, toNode);
              return false;
          }
      }

      // Whether an edge can be successfully added(fromNode -> toNode),need to determine whether the DAG has cycle!
      // 判断是否新的边能被成功添加进图中（起始点-->目标点），需要判断是否会形成有环图。
      int verticesCount = getNodesCount();

      Queue<Node> queue = new LinkedList<>();

      queue.add(toNode);

      // if DAG doesn't find fromNode, it's not has cycle!
      // 把目标点添加到队列，然后不断循环检查，目标节点的下一个节点会不会是起始点
      while (!queue.isEmpty() && (--verticesCount > 0)) {
          Node key = queue.poll();

          for (Node subsequentNode : getSubsequentNodes(key)) {
              if (subsequentNode.equals(fromNode)) {
                  return false;
              }

              queue.add(subsequentNode);
          }
      }

      return true;
  }


  /**
   * Get all neighbor nodes of the node
   * 获取待查询节点的下一个节点的集合，一个节点可能指向多个节点
   *
   * @param node   Node id to be calculated 需要查询的node id
   * @param edges neighbor edge information 现有的所有边的信息
   * @return all neighbor nodes of the node 返回相邻的节点的集合
   */
  private Set<Node> getNeighborNodes(Node node, final Map<Node, Map<Node, EdgeInfo>> edges) {
    final Map<Node, EdgeInfo> neighborEdges = edges.get(node);

    if (neighborEdges == null) {
      return Collections.EMPTY_MAP.keySet();
    }

    return neighborEdges.keySet();
  }



  /**
   * Determine whether there are ring and topological sorting results
   * 确定是否有环，返回拓扑排序结果
   * Directed acyclic graph (DAG) has topological ordering
   *  有向无环有向无环图(DAG)具有拓扑序
   * Breadth First Search：广度优先搜索：
   *    1、Traversal of all the vertices in the graph, the degree of entry is 0 vertex into the queue
   *    1、遍历图中的所有顶点，进入队列的度为0
   *    2、Poll a vertex in the queue to update its adjacency (minus 1) and queue the adjacency if it is 0 after minus 1
   *    2、轮询队列中的一个顶点，以更新其邻接点(减一) ，如果该顶点在减一之后是0，则对其邻接点进行排队
   *    3、Do step 2 until the queue is empty
   *    3、执行步骤2，直到队列是空的
   * If you cannot traverse all the nodes, it means that the current graph is not a directed acyclic graph.
   * 如果你不能遍历所有的节点，这意味着当前的图不是一个有向无环图。
   * There is no topological sort.
   * 没有拓扑排序。
   *
   *
   * @return key Returns the state
   * if success (acyclic) is true, failure (acyclic) is looped,
   * and value (possibly one of the topological sort results)
   * 如果成功(非循环)为 true，返回true和拓扑排序序列。如果是循环图，则返回false。
   */
  private Map.Entry<Boolean, List<Node>> topologicalSortImpl() {
    // node queue with degree of entry 0
    //入度为0的所有节点
    Queue<Node> zeroIndegreeNodeQueue = new LinkedList<>();
    // save result
    //保存结果
    List<Node> topoResultList = new ArrayList<>();
    // save the node whose degree is not 0
    //保存所有的入度不为0的节点
    Map<Node, Integer> notZeroIndegreeNodeMap = new HashMap<>();

    // Scan all the vertices and push vertexs with an entry degree of 0 to queue
    //扫描所有顶点并将入度为0的节点推入队列
    for (Map.Entry<Node, NodeInfo> vertices : nodesMap.entrySet()) {
      Node node = vertices.getKey();
      int inDegree = getIndegree(node);

      if (inDegree == 0) {
        zeroIndegreeNodeQueue.add(node);
        topoResultList.add(node);
      } else {
        notZeroIndegreeNodeMap.put(node, inDegree);
      }
    }

    /**
     * After scanning, there is no node with 0 degree of entry,
     * 如果这幅图连一个入度为0的节点都没有的话，说明整个图就是一个环了
     * indicating that there is a ring, and return directly
     * 这种情况 就直接返回false就好了
     */
    if(zeroIndegreeNodeQueue.isEmpty()){
      return new AbstractMap.SimpleEntry(false, topoResultList);
    }

    // The topology algorithm is used to delete nodes with 0 degree of entry and its associated edges
    //拓扑算法用于删除具有0入度的节点及其关联边
    while (!zeroIndegreeNodeQueue.isEmpty()) {
      Node v = zeroIndegreeNodeQueue.poll();
      // Get the neighbor node
      //获取相邻节点
      Set<Node> subsequentNodes = getSubsequentNodes(v);

      for (Node subsequentNode : subsequentNodes) {

        Integer degree = notZeroIndegreeNodeMap.get(subsequentNode);

        if(--degree == 0){
          topoResultList.add(subsequentNode);
          zeroIndegreeNodeQueue.add(subsequentNode);
          notZeroIndegreeNodeMap.remove(subsequentNode);
        }else{
          notZeroIndegreeNodeMap.put(subsequentNode, degree);
        }

      }
    }

    // if notZeroIndegreeNodeMap is empty,there is no ring!
    //如果非0入度的Map是空的，说明没有环，返回拓扑排序
    AbstractMap.SimpleEntry resultMap = new AbstractMap.SimpleEntry(notZeroIndegreeNodeMap.size() == 0 , topoResultList);
    return resultMap;

  }

}

爱词霸

 

 

划词翻译

自动发声

自动添加生词本