tensorflow/core/graph/collective_order.cc
// Find all CollectiveReduce nodes and the existing data dependencies between
// them.
Find all CollectiveReduce nodes and the existing data dependencies between
them.
Find all CollectiveReduce nodes and the existing data dependencies between them .
Status DiscoverDataDependencies(
    const Graph* graph, std::vector<Node*>* collective_nodes,
    std::vector<int32>* instance_keys,
    absl::flat_hash_map<Node*, absl::flat_hash_set<int32>>* data_dependencies) {
  Status s;
  auto node_leave = [collective_nodes, instance_keys, data_dependencies,
                     &s](Node* node) {
    int32_t instance_key;
    bool enter_node =
        node->IsCollective() && node->type_string() == "CollectiveReduce";
    if (enter_node) {
      Status get_attr_status =
          GetNodeAttr(node->attrs(), "instance_key", &instance_key);
      s.Update(get_attr_status);
      collective_nodes->push_back(node);
      instance_keys->push_back(instance_key);
      VLOG(2) << "collective node " << node->DebugString();
    }
    data_dependencies->reserve(data_dependencies->size() + 1 +
                               node->out_edges().size());
    const auto& node_deps = (*data_dependencies)[node];
    for (const Edge* out_edge : node->out_edges()) {
      auto& child_deps = (*data_dependencies)[out_edge->dst()];
      child_deps.insert(node_deps.begin(), node_deps.end());
      if (enter_node && s.ok()) {
        child_deps.insert(instance_key);
      }
    }
  };
  ReverseDFS(*graph, nullptr, node_leave);
  return s;
}
Status DiscoverDataDependencies ( const Graph * graph , std :: vector < Node * > * collective_nodes , std :: vector < int32 > * instance_keys , absl :: flat_hash_map < Node * , absl :: flat_hash_set < int32 > > * data_dependencies ) { Status s ; auto node_leave = [ collective_nodes , instance_keys , data_dependencies , & s ] ( Node * node ) { int32_t instance_key ; bool enter_node = node -> IsCollective ( ) && node -> type_string ( ) == " " ; if ( enter_node ) { Status get_attr_status = GetNodeAttr ( node -> attrs ( ) , " " , & instance_key ) ; s . Update ( get_attr_status ) ; collective_nodes -> push_back ( node ) ; instance_keys -> push_back ( instance_key ) ; VLOG ( 2 ) << " " << node -> DebugString ( ) ; } data_dependencies -> reserve ( data_dependencies -> size ( ) + 1 + node -> out_edges ( ) . size ( ) ) ; const auto & node_deps = ( * data_dependencies ) [ node ] ; for ( const Edge * out_edge : node -> out_edges ( ) ) { auto & child_deps = ( * data_dependencies ) [ out_edge -> dst ( ) ] ; child_deps . insert ( node_deps . begin ( ) , node_deps . end ( ) ) ; if ( enter_node && s . ok ( ) ) { child_deps . insert ( instance_key ) ; } } } ; ReverseDFS ( * graph , nullptr , node_leave ) ; return s ; }
Find all CollectiveReduce nodes and the existing data dependencies between
them.
Find all CollectiveReduce nodes and the existing data dependencies between them .
// Algorithm: do Reverse DFS starting at sink.  `node_leave` is called when // all parents of `node` have been visited.  At that point, // `data_dependencies[node]` is a list containing `instance_key` of every // `CollectiveReduce` on which `node` has a data dependency. // For this node's children, add all these instance keys.  Also, if this node // is collective, add as a dependency for the children. // Avoid reference invalidation of `node_deps`.