tensorflow/core/tpu/graph_rewrite/combine_tpu_embedding_load_retrieve_pass.cc
// Computes an array of ports containing the source of each table/parameter
// combination. Fills in any unused ports with {nullptr, 0}.
Computes an array of ports containing the source of each table/parameter
combination. Fills in any unused ports with {nullptr, 0}.
Computes an array of ports containing the source of each table / parameter combination . Fills in any unused ports with { nullptr 0 } .
Status CombinePerTableParametersForLoad(
    const absl::flat_hash_set<Node*>& load_nodes,
    std::vector<Node*>* nodes_per_table,
    std::vector<bool>* is_debug_load_retrieve_node,
    const tensorflow::tpu::TPUEmbeddingConfiguration& tpu_embedding_config,
    const TableNameToIntegerMap& table_name_to_id_map,
    LoadCombinedParametersType* combined_inputs, int* num_shards) {
  TF_RETURN_IF_ERROR(GetLoadOrRetrieveNodesByTable(
      load_nodes, tpu_embedding_config, table_name_to_id_map, "Load",
      nodes_per_table, is_debug_load_retrieve_node, num_shards));
  const int num_tables = tpu_embedding_config.table_descriptor_size();
  for (auto& v : *combined_inputs) {
    v.clear();
    v.resize(num_tables, Port{nullptr, 0});
  }
  for (int table_id = 0; table_id < num_tables; ++table_id) {
    Node* neighbor = (*nodes_per_table)[table_id];
    CHECK_NE(neighbor, nullptr);  
    const auto& opt_params = tpu_embedding_config.table_descriptor(table_id)
                                 .optimization_parameters();
    std::vector<tpu::StateVariableSpecification> state_variable_specs;
    Status status = tpu::GetOptimizationAlgorithmStateVariables(
        opt_params, &state_variable_specs);
    if (!status.ok()) {
      return status;
    }
    CHECK_LE(state_variable_specs.size(),  
             tpu::kMaxAuxiliaryParameterCount + 1);
    for (int parameter_num = 0;
         parameter_num < tpu::kMaxAuxiliaryParameterCount + 1;
         ++parameter_num) {
      (*combined_inputs)[parameter_num][table_id] = Port{nullptr, 0};
    }
    for (const auto& e : neighbor->in_edges()) {
      if (!e->IsControlEdge()) {
        TF_RET_CHECK(e->dst_input() >= 0);
        TF_RET_CHECK(e->dst_input() < state_variable_specs.size());
        TF_RET_CHECK(state_variable_specs[e->dst_input()].has_user_defined() ||
                     (*is_debug_load_retrieve_node)[table_id]);
        TF_RET_CHECK((*combined_inputs)[e->dst_input()][table_id].node ==
                     nullptr);
        (*combined_inputs)[e->dst_input()][table_id] =
            Port{e->src(), e->src_output()};
      }
    }
    for (int parameter_num = 0;
         parameter_num < tpu::kMaxAuxiliaryParameterCount + 1;
         ++parameter_num) {
      const auto node = (*combined_inputs)[parameter_num][table_id].node;
      if (parameter_num < state_variable_specs.size() &&
          (state_variable_specs[parameter_num].has_user_defined() ||
           (*is_debug_load_retrieve_node)[table_id])) {
        if (node == nullptr) {
          return errors::InvalidArgument(absl::StrFormat(
              "Found missing parameter in slot %d of table %s.", parameter_num,
              tpu_embedding_config.table_descriptor(table_id).name()));
        }
      } else {
        if (node != nullptr) {
          return errors::InvalidArgument(absl::StrFormat(
              "Found extra parameter in slot %d of table %s.", parameter_num,
              tpu_embedding_config.table_descriptor(table_id).name()));
        }
      }
    }
  }
  return Status::OK();
}
Status CombinePerTableParametersForLoad ( const absl :: flat_hash_set < Node * > & load_nodes , std :: vector < Node * > * nodes_per_table , std :: vector < bool > * is_debug_load_retrieve_node , const tensorflow :: tpu :: TPUEmbeddingConfiguration & tpu_embedding_config , const TableNameToIntegerMap & table_name_to_id_map , LoadCombinedParametersType * combined_inputs , int * num_shards ) { TF_RETURN_IF_ERROR ( GetLoadOrRetrieveNodesByTable ( load_nodes , tpu_embedding_config , table_name_to_id_map , " " , nodes_per_table , is_debug_load_retrieve_node , num_shards ) ) ; const int num_tables = tpu_embedding_config . table_descriptor_size ( ) ; for ( auto & v : * combined_inputs ) { v . clear ( ) ; v . resize ( num_tables , Port { nullptr , 0 } ) ; } for ( int table_id = 0 ; table_id < num_tables ; ++ table_id ) { Node * neighbor = ( * nodes_per_table ) [ table_id ] ; CHECK_NE ( neighbor , nullptr ) ; const auto & opt_params = tpu_embedding_config . table_descriptor ( table_id ) . optimization_parameters ( ) ; std :: vector < tpu :: StateVariableSpecification > state_variable_specs ; Status status = tpu :: GetOptimizationAlgorithmStateVariables ( opt_params , & state_variable_specs ) ; if ( ! status . ok ( ) ) { return status ; } CHECK_LE ( state_variable_specs . size ( ) , tpu :: kMaxAuxiliaryParameterCount + 1 ) ; for ( int parameter_num = 0 ; parameter_num < tpu :: kMaxAuxiliaryParameterCount + 1 ; ++ parameter_num ) { ( * combined_inputs ) [ parameter_num ] [ table_id ] = Port { nullptr , 0 } ; } for ( const auto & e : neighbor -> in_edges ( ) ) { if ( ! e -> IsControlEdge ( ) ) { TF_RET_CHECK ( e -> dst_input ( ) >= 0 ) ; TF_RET_CHECK ( e -> dst_input ( ) < state_variable_specs . size ( ) ) ; TF_RET_CHECK ( state_variable_specs [ e -> dst_input ( ) ] . has_user_defined ( ) || ( * is_debug_load_retrieve_node ) [ table_id ] ) ; TF_RET_CHECK ( ( * combined_inputs ) [ e -> dst_input ( ) ] [ table_id ] . node == nullptr ) ; ( * combined_inputs ) [ e -> dst_input ( ) ] [ table_id ] = Port { e -> src ( ) , e -> src_output ( ) } ; } } for ( int parameter_num = 0 ; parameter_num < tpu :: kMaxAuxiliaryParameterCount + 1 ; ++ parameter_num ) { const auto node = ( * combined_inputs ) [ parameter_num ] [ table_id ] . node ; if ( parameter_num < state_variable_specs . size ( ) && ( state_variable_specs [ parameter_num ] . has_user_defined ( ) || ( * is_debug_load_retrieve_node ) [ table_id ] ) ) { if ( node == nullptr ) { return errors :: InvalidArgument ( absl :: StrFormat ( " " , parameter_num , tpu_embedding_config . table_descriptor ( table_id ) . name ( ) ) ) ; } } else { if ( node != nullptr ) { return errors :: InvalidArgument ( absl :: StrFormat ( " " , parameter_num , tpu_embedding_config . table_descriptor ( table_id ) . name ( ) ) ) ; } } } } return Status :: OK ( ) ; }
Computes an array of ports containing the source of each table/parameter
combination.
Computes an array of ports containing the source of each table / parameter combination .
// Set dummy values for unused tensor inputs/outputs. // Crash OK // Crash OK