networktopology.java

分布式全文搜索工具包 可以支持集群 主要使用java開發 比較方便使用

            + node.toString() + " at an illegal network location");
      }
      if (clusterMap.add(node)) {
        if (rack == null) {
          numOfRacks++;
        }
      }
      LOG.debug("NetworkTopology became:\n" + this.toString());
    } finally {
      netlock.writeLock().unlock();
    }
  }

  /** Remove a node
   * Update node counter & rack counter if neccessary
   * @param node
   *          node to be removed
   */
  public void remove(Node node) {
    if (node == null)
      return;
    netlock.writeLock().lock();
    try {
      if (node instanceof InnerNode) {
        throw new IllegalArgumentException(
            "Not allow to remove an inner node: " + NodeBase.getPath(node));
      }
      LOG.info("Removing a node: " + NodeBase.getPath(node));
      if (clusterMap.remove(node)) {
        InnerNode rack = (InnerNode) getNode(node.getNetworkLocation());
        if (rack == null) {
          numOfRacks--;
        }
      }
      LOG.debug("NetworkTopology became:\n" + this.toString());
    } finally {
      netlock.writeLock().unlock();
    }
  }

  /** Check if the tree contains node <i>node</i>
   * 
   * @param node
   *          a node
   * @return true if <i>node</i> is already in the tree; false otherwise
   */
  public boolean contains(Node node) {
    if (node == null)
      return false;
    netlock.readLock().lock();
    try {
      Node parent = node.getParent();
      for (int level = node.getLevel(); parent != null && level > 0; parent = parent
          .getParent(), level--) {
        if (parent == clusterMap)
          return true;
      }
    } finally {
      netlock.readLock().unlock();
    }
    return false;
  }

  /** Given a string representation of a node, return its reference
   * 
   * @param loc
   *          a path-like string representation of a node
   * @return a reference to the node; null if the node is not in the tree
   */
  protected Node getNode(String loc) {
    loc = NodeBase.normalize(loc);
    if (!NodeBase.ROOT.equals(loc))
      loc = loc.substring(1);
    return clusterMap.getLoc(loc);
  }

  /** Return the total number of racks */
  public int getNumOfRacks() {
    netlock.readLock().lock();
    try {
      return numOfRacks;
    } finally {
      netlock.readLock().unlock();
    }
  }

  /** Return the total number of nodes */
  public int getNumOfLeaves() {
    netlock.readLock().lock();
    try {
      return clusterMap.getNumOfLeaves();
    } finally {
      netlock.readLock().unlock();
    }
  }

  /** Return the distance between two nodes
   * It is assumed that the distance from one node to its parent is 1
   * The distance between two nodes is calculated by summing up their distances
   * to their closest common  ancestor.
   * @param node1 one node
   * @param node2 another node
   * @return the distance between node1 and node2
   * node1 or node2 do not belong to the cluster
   */
  public int getDistance(Node node1, Node node2) {
    if (node1 == node2) {
      return 0;
    }
    Node n1 = node1, n2 = node2;
    int dis = 0;
    netlock.readLock().lock();
    try {
      int level1 = node1.getLevel(), level2 = node2.getLevel();
      while (n1 != null && level1 > level2) {
        n1 = n1.getParent();
        level1--;
        dis++;
      }
      while (n2 != null && level2 > level1) {
        n2 = n2.getParent();
        level2--;
        dis++;
      }
      while (n1 != null && n2 != null && n1.getParent() != n2.getParent()) {
        n1 = n1.getParent();
        n2 = n2.getParent();
        dis += 2;
      }
    } finally {
      netlock.readLock().unlock();
    }
    if (n1 == null) {
      LOG.warn("The cluster does not contain node: " + NodeBase.getPath(node1));
      return Integer.MAX_VALUE;
    }
    if (n2 == null) {
      LOG.warn("The cluster does not contain node: " + NodeBase.getPath(node2));
      return Integer.MAX_VALUE;
    }
    return dis + 2;
  }

  /** Check if two nodes are on the same rack
   * @param node1 one node
   * @param node2 another node
   * @return true if node1 and node2 are pm the same rack; false otherwise
   * @exception IllegalArgumentException when either node1 or node2 is null, or
   * node1 or node2 do not belong to the cluster
   */
  public boolean isOnSameRack(Node node1, Node node2) {
    if (node1 == null || node2 == null) {
      return false;
    }

    netlock.readLock().lock();
    try {
      return node1.getParent() == node2.getParent();
    } finally {
      netlock.readLock().unlock();
    }
  }

  final protected static Random r = new Random();

  /** randomly choose one node from <i>scope</i>
   * if scope starts with ~, choose one from the all nodes except for the
   * ones in <i>scope</i>; otherwise, choose one from <i>scope</i>
   * @param scope range of nodes from which a node will be choosen
   * @return the choosen node
   */
  public Node chooseRandom(String scope) {
    netlock.readLock().lock();
    try {
      if (scope.startsWith("~")) {
        return chooseRandom(NodeBase.ROOT, scope.substring(1));
      } else {
        return chooseRandom(scope, null);
      }
    } finally {
      netlock.readLock().unlock();
    }
  }

  private Node chooseRandom(String scope, String excludedScope) {
    if (excludedScope != null) {
      if (scope.startsWith(excludedScope)) {
        return null;
      }
      if (!excludedScope.startsWith(scope)) {
        excludedScope = null;
      }
    }
    Node node = getNode(scope);
    if (!(node instanceof InnerNode)) {
      return node;
    }
    InnerNode innerNode = (InnerNode) node;
    int numOfDatanodes = innerNode.getNumOfLeaves();
    if (excludedScope == null) {
      node = null;
    } else {
      node = getNode(excludedScope);
      if (!(node instanceof InnerNode)) {
        numOfDatanodes -= 1;
      } else {
        numOfDatanodes -= ((InnerNode) node).getNumOfLeaves();
      }
    }
    int leaveIndex = r.nextInt(numOfDatanodes);
    return innerNode.getLeaf(leaveIndex, node);
  }

  /** return the number of leaves in <i>scope</i> but not in <i>excludedNodes</i>
   * if scope starts with ~, return the number of nodes that are not
   * in <i>scope</i> and <i>excludedNodes</i>; 
   * @param scope a path string that may start with ~
   * @param excludedNodes a list of nodes
   * @return number of available nodes
   */
  public int countNumOfAvailableNodes(String scope, List<Node> excludedNodes) {
    boolean isExcluded = false;
    if (scope.startsWith("~")) {
      isExcluded = true;
      scope = scope.substring(1);
    }
    scope = NodeBase.normalize(scope);
    int count = 0; // the number of nodes in both scope & excludedNodes
    netlock.readLock().lock();
    try {
      for (Node node : excludedNodes) {
        if ((NodeBase.getPath(node) + NodeBase.PATH_SEPARATOR_STR)
            .startsWith(scope + NodeBase.PATH_SEPARATOR_STR)) {
          count++;
        }
      }
      Node n = getNode(scope);
      int scopeNodeCount = 1;
      if (n instanceof InnerNode) {
        scopeNodeCount = ((InnerNode) n).getNumOfLeaves();
      }
      if (isExcluded) {
        return clusterMap.getNumOfLeaves() - scopeNodeCount
            - excludedNodes.size() + count;
      } else {
        return scopeNodeCount - count;
      }
    } finally {
      netlock.readLock().unlock();
    }
  }

  /** convert a network tree to a string */
  public String toString() {
    // print the number of racks
    StringBuffer tree = new StringBuffer();
    tree.append("Number of racks: ");
    tree.append(numOfRacks);
    tree.append("\n");
    // print the number of leaves
    int numOfLeaves = getNumOfLeaves();
    tree.append("Expected number of leaves:");
    tree.append(numOfLeaves);
    tree.append("\n");
    // print nodes
    for (int i = 0; i < numOfLeaves; i++) {
      tree.append(NodeBase.getPath(clusterMap.getLeaf(i, null)));
      tree.append("\n");
    }
    return tree.toString();
  }

  /* swap two array items */
  static private void swap(Node[] nodes, int i, int j) {
    Node tempNode;
    tempNode = nodes[j];
    nodes[j] = nodes[i];
    nodes[i] = tempNode;

  }

  /** Sort nodes array by their distances to <i>reader</i>
   * It linearly scans the array, if a local node is found, swap it with
   * the first element of the array.
   * If a local rack node is found, swap it with the first element following
   * the local node.
   * If neither local node or local rack node is found, put a random replica
   * location at postion 0.
   * It leaves the rest nodes untouched.
   */
  public synchronized void pseudoSortByDistance(Node reader, Node[] nodes) {
    int tempIndex = 0;
    if (reader != null) {
      int localRackNode = -1;
      //scan the array to find the local node & local rack node
      for (int i = 0; i < nodes.length; i++) {
        if (tempIndex == 0 && reader == nodes[i]) { //local node
          //swap the local node and the node at position 0
          if (i != 0) {
            swap(nodes, tempIndex, i);
          }
          tempIndex = 1;
          if (localRackNode != -1) {
            if (localRackNode == 0) {
              localRackNode = i;
            }
            break;
          }
        } else if (localRackNode == -1 && isOnSameRack(reader, nodes[i])) {
          //local rack
          localRackNode = i;
          if (tempIndex != 0)
            break;
        }
      }

      // swap the local rack node and the node at position tempIndex
      if (localRackNode != -1 && localRackNode != tempIndex) {
        swap(nodes, tempIndex, localRackNode);
        tempIndex++;
      }
    }

    // put a random node at position 0 if it is not a local/local-rack node
    if (tempIndex == 0 && nodes.length != 0) {
      swap(nodes, 0, r.nextInt(nodes.length));
    }
  }
}