#include <NTSession.h>

Inheritance diagram for tensorflow::NTSession:

Classes
struct	ExecutorsAndKeys

struct	PerPartitionExecutorsAndLib

struct	RunState

struct	RunStateArgs

Public Types
typedef std::function< void(Session *)>	CloseCallback

typedef std::vector< std::pair< string, Tensor > >	NamedTensorList

typedef std::unordered_map< StringPiece, Node *, StringPieceHasher >	NameNodeMap

Public Member Functions
::tensorflow::Status	Close () override

::tensorflow::Status	Create (const GraphDef &graph) override

void	ExportCostModels (CostModelManager::CostModelMap *cost_models)

::tensorflow::Status	Extend (const GraphDef &graph) override

::tensorflow::Status	ListDevices (std::vector< DeviceAttributes > *response) override

::tensorflow::Status	LocalDeviceManager (const DeviceMgr **output) override

	NTSession (const SessionOptions &options, const DeviceMgr device_mgr, NTSessionFactory factory)

::tensorflow::Status	PRun (const string &handle, const NamedTensorList &inputs, const std::vector< string > &output_names, std::vector< Tensor > *outputs) override

::tensorflow::Status	PRunSetup (const std::vector< string > &input_names, const std::vector< string > &output_names, const std::vector< string > &target_nodes, string *handle) override

::tensorflow::Status	Reset (const std::vector< string > &containers)

::tensorflow::Status	Run (const NamedTensorList &inputs, const std::vector< string > &output_names, const std::vector< string > &target_nodes, std::vector< Tensor > *outputs) override

::tensorflow::Status	Run (const ::tensorflow::RunOptions &run_options, const NamedTensorList &inputs, const std::vector< string > &output_names, const std::vector< string > &target_nodes, std::vector< Tensor > outputs, RunMetadata run_metadata) override

	~NTSession () override

Private Member Functions
::tensorflow::Status	CheckFetch (const std::vector< std::pair< string, Tensor >> &feeds, const std::vector< string > &fetches, const ExecutorsAndKeys executors_and_keys, const RunState run_state)

::tensorflow::Status	CheckNotClosed ()

::tensorflow::Status	CreateDebuggerState (const DebugOptions &debug_options, int64 session_run_index, int64 executor_step_index, const std::vector< string > &input_names, const std::vector< string > &output_names, const std::vector< string > &target_names, std::unique_ptr< DebuggerStateInterface > *debugger_state)

::tensorflow::Status	CreateGraphs (const BuildGraphOptions &options, std::unordered_map< string, std::unique_ptr< Graph >> outputs, std::unique_ptr< FunctionLibraryDefinition > flib_def, RunStateArgs run_state_args, DataTypeVector input_types, DataTypeVector *output_types)

::tensorflow::Status	DecorateAndPublishGraphForDebug (const DebugOptions &debug_options, Graph graph, Device device)

::tensorflow::Status	ExtendLocked (const GraphDef &graph) EXCLUSIVE_LOCKS_REQUIRED(graph_def_lock_)

::tensorflow::Status	GetOrCreateExecutors (gtl::ArraySlice< string > inputs, gtl::ArraySlice< string > outputs, gtl::ArraySlice< string > target_nodes, ExecutorsAndKeys *executors_and_keys, RunStateArgs run_state_args)

bool graph_created_	GUARDED_BY (graph_def_lock_)

GraphDef graph_def_	GUARDED_BY (graph_def_lock_)

std::unordered_map< string, std::shared_ptr< ExecutorsAndKeys > > executors_	GUARDED_BY (executor_lock_)

std::unordered_map< string, std::unique_ptr< RunState > > partial_runs_	GUARDED_BY (executor_lock_)

std::unordered_map< string, string > stateful_placements_	GUARDED_BY (graph_def_lock_)

std::unique_ptr< GraphExecutionState > execution_state_	GUARDED_BY (graph_def_lock_)

bool closed_	GUARDED_BY (closed_lock_)

Status	MaybeInitializeExecutionState (const GraphDef &graph, bool *out_already_initialized) EXCLUSIVE_LOCKS_REQUIRED(graph_def_lock_)

::tensorflow::Status	RecvPRunOutputs (const std::vector< string > &output_names, const ExecutorsAndKeys executors_and_keys, RunState run_state, std::vector< Tensor > *outputs)

::tensorflow::Status	ResourceHandleToInputTensor (const Tensor &resource_tensor, Tensor *retrieved_tensor)

void	SchedClosure (std::function< void()> c)

::tensorflow::Status	SendPRunInputs (const std::vector< std::pair< string, Tensor >> &inputs, const ExecutorsAndKeys executors_and_keys, IntraProcessRendezvous rendez)

	TF_DISALLOW_COPY_AND_ASSIGN (NTSession)

::tensorflow::Status	WaitForNotification (Notification *n, int64 timeout_in_ms)

void	WaitForNotification (RunState run_state, CancellationManager cm, int64 timeout_in_ms)

Private Attributes
CancellationManager *	cancellation_manager_

mutex	closed_lock_

CostModelManager	cost_model_manager_

const std::unique_ptr< const DeviceMgr >	device_mgr_

DeviceSet	device_set_

std::vector< Device * >	devices_

std::atomic< int64 >	edge_name_counter_ = {0}

mutex	executor_lock_

NTSessionFactory *const	factory_

std::unique_ptr< FunctionLibraryDefinition >	flib_def_

mutex	graph_def_lock_

std::atomic< int64 >	handle_name_counter_ = {0}

Status	init_error_

Executor::Args::NodeOutputsCallback	node_outputs_callback_ = nullptr

const int64	operation_timeout_in_ms_ = 0

const SessionOptions	options_

string	session_handle_

SessionState	session_state_

bool	sync_on_finish_ = true

Static Private Attributes
static std::atomic_int_fast64_t	step_id_counter_

Friends
class	DebugGateway

Detailed Description

Definition at line 71 of file NTSession.h.

Member Typedef Documentation

typedef std::function<void(Session*)> tensorflow::NTSession::CloseCallback

Definition at line 73 of file NTSession.h.

typedef std::vector<std::pair<string, Tensor> > tensorflow::NTSession::NamedTensorList

Definition at line 83 of file NTSession.h.

typedef std::unordered_map<StringPiece, Node*, StringPieceHasher> tensorflow::NTSession::NameNodeMap

Definition at line 84 of file NTSession.h.

Constructor & Destructor Documentation

tensorflow::NTSession::NTSession	(	const SessionOptions &	options,
		const DeviceMgr *	device_mgr,
		NTSessionFactory *	factory
	)

Definition at line 197 of file NTSession.cc.

References edmIntegrityCheck::d, device_mgr_, device_set_, devices_, dqm::qstatus::ERROR, MessageLogger_cfi::INFO, LOG, session_handle_, btagGenBb_cfi::Status, mps_update::status, and sync_on_finish_.

     : options_(options),
       device_mgr_(device_mgr),
       factory_(factory),
       cancellation_manager_(new CancellationManager()),
       operation_timeout_in_ms_(options_.config.operation_timeout_in_ms()) {
   // The default value of sync_on_finish will be flipped soon and this
   // environment variable will be removed as well.
   const Status status =
       ReadBoolFromEnvVar("TF_SYNC_ON_FINISH", true, &sync_on_finish_);
   if (!status.ok()) {
     LOG(ERROR) << status.error_message();
   }
   // NOTE(mrry): We do not need to use a unique string for the session
   // handle, because NTSession owns its devices. This may change
   // in future versions.
   session_handle_ = "no_threads";
   int devices_added = 0;
   if (options.config.log_device_placement()) {
     const string mapping_str = device_mgr_->DeviceMappingString();
     if (mapping_str.empty()) {
       printf("Device mapping: no known devices.\n");
     } else {
       printf("Device mapping:\n%s", mapping_str.c_str());
     }
     LOG(INFO) << "Device mapping:\n" << mapping_str;
   }
   for (auto d : device_mgr_->ListDevices()) {
     devices_.push_back(d);
     device_set_.AddDevice(d);
     d->op_segment()->AddHold(session_handle_);
 
     // The first device added is special: it is the 'client device' (a
     // CPU device) from which we feed and fetch Tensors.
     if (devices_added == 0) {
       device_set_.set_client_device(d);
     }
     ++devices_added;
   }
 }

tensorflow::NTSession::~NTSession ( )

override

Definition at line 240 of file NTSession.cc.

References cancellation_manager_, Close(), edmIntegrityCheck::d, device_mgr_, flib_def_, and session_handle_.

                       {
   if (!closed_) Close().IgnoreError();
   for (auto& it : partial_runs_) {
     it.second.reset(nullptr);
   }
   for (auto& it : executors_) {
     it.second.reset();
   }
   for (auto d : device_mgr_->ListDevices()) {
     d->op_segment()->RemoveHold(session_handle_);
   }
   delete cancellation_manager_;
 
   execution_state_.reset(nullptr);
   flib_def_.reset(nullptr);
 }

Member Function Documentation

Status tensorflow::NTSession::CheckFetch	(	const std::vector< std::pair< string, Tensor >> &	feeds,
		const std::vector< string > &	fetches,
		const ExecutorsAndKeys *	executors_and_keys,
		const RunState *	run_state
	)

private

Definition at line 914 of file NTSession.cc.

References executor_lock_, cond::persistency::fetch(), plotBeamSpotDB::first, tensorflow::NTSession::ExecutorsAndKeys::graph, triggerObjects_cff::id, input, checklumidiff::l, gen::n, tensorflow::NTSession::ExecutorsAndKeys::name_to_node, edm::errors::NotFound, tensorflow::NTSession::RunState::pending_inputs, svgfig::stack, and class-composition::visited.

Referenced by PRun().

                                                             {
   const Graph* graph = executors_and_keys->graph.get();
   const NameNodeMap* name_to_node = &executors_and_keys->name_to_node;
 
   // Build the set of pending feeds that we haven't seen.
   std::unordered_set<TensorId, TensorId::Hasher> pending_feeds;
   {
     mutex_lock l(executor_lock_);
     for (const auto& input : run_state->pending_inputs) {
       // Skip if the feed has already been fed.
       if (input.second) continue;
       TensorId id(ParseTensorName(input.first));
       auto it = name_to_node->find(id.first);
       if (it == name_to_node->end()) {
         return errors::NotFound("Feed ", input.first, ": not found");
       }
       pending_feeds.insert(id);
     }
   }
   for (const auto& it : feeds) {
     TensorId id(ParseTensorName(it.first));
     pending_feeds.erase(id);
   }
 
   // Initialize the stack with the fetch nodes.
   std::vector<const Node*> stack;
   for (const string& fetch : fetches) {
     TensorId id(ParseTensorName(fetch));
     auto it = name_to_node->find(id.first);
     if (it == name_to_node->end()) {
       return errors::NotFound("Fetch ", fetch, ": not found");
     }
     stack.push_back(it->second);
   }
 
   // Any tensor needed for fetches can't be in pending_feeds.
   std::vector<bool> visited(graph->num_node_ids(), false);
   while (!stack.empty()) {
     const Node* n = stack.back();
     stack.pop_back();
 
     for (const Edge* in_edge : n->in_edges()) {
       const Node* in_node = in_edge->src();
       if (pending_feeds.count({in_node->name(), in_edge->src_output()}) > 0) {
         return errors::InvalidArgument("Fetch ", in_node->name(), ":",
                                        in_edge->src_output(),
                                        " can't be computed from the feeds"
                                        " that have been fed so far.");
       }
       if (!visited[in_node->id()]) {
         visited[in_node->id()] = true;
         stack.push_back(in_node);
       }
     }
   }
   return Status::OK();
 }

::tensorflow::Status tensorflow::NTSession::CheckNotClosed ( )

inlineprivate

Definition at line 263 of file NTSession.h.

References closed_lock_, CreateDebuggerState(), DecorateAndPublishGraphForDebug(), tensorflow::NTSession::PerPartitionExecutorsAndLib::device, tensorflow::NTSession::PerPartitionExecutorsAndLib::graph, checklumidiff::l, and btagGenBb_cfi::Status.

Referenced by DecorateAndPublishGraphForDebug(), Extend(), PRun(), and PRunSetup().

                                     {
     mutex_lock l(closed_lock_);
     if (closed_) return errors::Cancelled("Session has been closed.");
     return ::tensorflow::Status::OK();
   }

tensorflow::Status tensorflow::NTSession::Close ( )

override

Definition at line 1386 of file NTSession.cc.

References cancellation_manager_, closed_lock_, tensorflow::NTSessionFactory::Deregister(), factory_, and checklumidiff::l.

Referenced by ~NTSession().

                                   {
   cancellation_manager_->StartCancel();
   {
     mutex_lock l(closed_lock_);
     if (closed_) return ::tensorflow::Status::OK();
     closed_ = true;
   }
   if (factory_ != nullptr) factory_->Deregister(this);
   return ::tensorflow::Status::OK();
 }

Status tensorflow::NTSession::Create ( const GraphDef & graph )

override

Definition at line 288 of file NTSession.cc.

References ExtendLocked(), graph_def_lock_, init_error_, and checklumidiff::l.

                                               {
   TF_RETURN_IF_ERROR(init_error_);
   if (graph.node_size() > 0) {
     mutex_lock l(graph_def_lock_);
     if (graph_created_) {
       return errors::AlreadyExists(
           "A Graph has already been created for this session.");
     }
     return ExtendLocked(graph);
   }
   return Status::OK();
 }

Status tensorflow::NTSession::CreateDebuggerState	(	const DebugOptions &	debug_options,
		int64	session_run_index,
		int64	executor_step_index,
		const std::vector< string > &	input_names,
		const std::vector< string > &	output_names,
		const std::vector< string > &	target_names,
		std::unique_ptr< DebuggerStateInterface > *	debugger_state
	)

private

Definition at line 331 of file NTSession.cc.

Referenced by CheckNotClosed(), and DecorateAndPublishGraphForDebug().

                                                            {
   TF_RETURN_IF_ERROR(
       DebuggerStateRegistry::CreateState(debug_options, debugger_state));
   TF_RETURN_IF_ERROR(debugger_state->get()->PublishDebugMetadata(
       debug_options.global_step(), session_run_index, executor_step_index,
       input_names, output_names, target_names));
   return Status::OK();
 }

Status tensorflow::NTSession::CreateGraphs	(	const BuildGraphOptions &	options,
		std::unordered_map< string, std::unique_ptr< Graph >> *	outputs,
		std::unique_ptr< FunctionLibraryDefinition > *	flib_def,
		RunStateArgs *	run_state_args,
		DataTypeVector *	input_types,
		DataTypeVector *	output_types
	)

private

Definition at line 1214 of file NTSession.cc.

References KineDebug3::count(), edmIntegrityCheck::d, device_mgr_, device_set_, devices_, edge_name_counter_, flib_def_, tensorflow::NTSession::RunStateArgs::graph, graph_def_lock_, tensorflow::NTSession::RunStateArgs::is_partial_run, checklumidiff::l, eostools::move(), dataset::name, options_, PatBasicFWLiteJetAnalyzer_Selector_cfg::outputs, tablePrinter::prefix, alignCSCRings::s, btagGenBb_cfi::Status, and std::swap().

Referenced by GetOrCreateExecutors().

                                   {
   mutex_lock l(graph_def_lock_);
   std::unique_ptr<ClientGraph> client_graph;
 
   std::unique_ptr<GraphExecutionState> temp_exec_state_holder;
   GraphExecutionState* execution_state = nullptr;
   if (options_.config.graph_options().place_pruned_graph()) {
     // Because we are placing pruned graphs, we need to create a
     // new GraphExecutionState for every new unseen graph,
     // and then place it.
     GraphExecutionStateOptions prune_options;
     prune_options.device_set = &device_set_;
     prune_options.session_options = &options_;
     prune_options.stateful_placements = stateful_placements_;
     TF_RETURN_IF_ERROR(GraphExecutionState::MakeForPrunedGraph(
         execution_state_->original_graph_def().library(), prune_options,
         execution_state_->original_graph_def(), subgraph_options,
         &temp_exec_state_holder, &client_graph));
     execution_state = temp_exec_state_holder.get();
   } else {
     execution_state = execution_state_.get();
     TF_RETURN_IF_ERROR(
         execution_state->BuildGraph(subgraph_options, &client_graph));
   }
 
   if (subgraph_options.feed_endpoints.size() !=
       client_graph->feed_types.size()) {
     return errors::Internal(
         "Graph pruning failed: requested number of feed endpoints = ",
         subgraph_options.feed_endpoints.size(),
         " versus number of pruned feed endpoints = ",
         client_graph->feed_types.size());
   }
   if (subgraph_options.fetch_endpoints.size() !=
       client_graph->fetch_types.size()) {
     return errors::Internal(
         "Graph pruning failed: requested number of fetch endpoints = ",
         subgraph_options.fetch_endpoints.size(),
         " versus number of pruned fetch endpoints = ",
         client_graph->fetch_types.size());
   }
 
   auto current_stateful_placements = execution_state->GetStatefulPlacements();
   // Update our current state based on the execution_state's
   // placements.  If there are any mismatches for a node,
   // we should fail, as this should never happen.
   for (auto placement_pair : current_stateful_placements) {
     const string& node_name = placement_pair.first;
     const string& placement = placement_pair.second;
     auto iter = stateful_placements_.find(node_name);
     if (iter == stateful_placements_.end()) {
       stateful_placements_.insert(std::make_pair(node_name, placement));
     } else if (iter->second != placement) {
       return errors::Internal(
           "Stateful placement mismatch. "
           "Current assignment of ",
           node_name, " to ", iter->second, " does not match ", placement);
     }
   }
 
   stateful_placements_ = execution_state->GetStatefulPlacements();
 
   // Remember the graph in run state if this is a partial run.
   if (run_state_args->is_partial_run) {
     run_state_args->graph.reset(new Graph(flib_def_.get()));
     CopyGraph(*execution_state->full_graph(), run_state_args->graph.get());
   }
 
   // Partition the graph across devices.
   PartitionOptions popts;
   popts.node_to_loc = [](const Node* node) {
     assert(node != nullptr);
     return node->assigned_device_name();
   };
   popts.new_name = [this](const string& prefix) {
     return strings::StrCat(prefix, "/_", edge_name_counter_.fetch_add(1));
   };
   popts.get_incarnation = [](const string& name) {
     // The direct session does not have changing incarnation numbers.
     // Just return '1'.
     return 1;
   };
   popts.flib_def = &client_graph->graph.flib_def();
   popts.control_flow_added = false;
 
   std::unordered_map<string, GraphDef> partitions;
   TF_RETURN_IF_ERROR(Partition(popts, &client_graph->graph, &partitions));
 
   std::vector<string> device_names;
   for (auto device : devices_) {
     // Extract the LocalName from the device.
     device_names.push_back(DeviceNameUtils::LocalName(device->name()));
   }
 
   // Check for valid partitions.
   for (const auto& partition : partitions) {
     const string local_partition_name =
         DeviceNameUtils::LocalName(partition.first);
     if (std::count(device_names.begin(), device_names.end(),
                    local_partition_name) == 0) {
       return errors::InvalidArgument(
           "Creating a partition for ", local_partition_name,
           " which doesn't exist in the list of available devices. Available "
           "devices: ",
           str_util::Join(device_names, ","));
     }
   }
 
   for (const auto& partition : partitions) {
     std::unique_ptr<Graph> device_graph(
         new Graph(client_graph->flib_def.get()));
     GraphConstructorOptions device_opts;
     // There are internal operations (e.g., send/recv) that we now allow.
     device_opts.allow_internal_ops = true;
     device_opts.expect_device_spec = true;
     TF_RETURN_IF_ERROR(ConvertGraphDefToGraph(device_opts, partition.second,
                                               device_graph.get()));
     outputs->emplace(partition.first, std::move(device_graph));
   }
 
   GraphOptimizationPassOptions optimization_options;
   optimization_options.session_options = &options_;
   optimization_options.flib_def = client_graph->flib_def.get();
   optimization_options.partition_graphs = outputs;
   TF_RETURN_IF_ERROR(OptimizationPassRegistry::Global()->RunGrouping(
       OptimizationPassRegistry::POST_PARTITIONING, optimization_options));
 
   Status s;
   for (auto& partition : *outputs) {
     const string& partition_name = partition.first;
     std::unique_ptr<Graph>* graph = &partition.second;
 
     VLOG(2) << "Created " << DebugString(graph->get()) << " for "
             << partition_name;
 
     // Give the device an opportunity to rewrite its subgraph.
     Device* d;
     s = device_mgr_->LookupDevice(partition_name, &d);
     if (!s.ok()) break;
     s = d->MaybeRewriteGraph(graph);
     if (!s.ok()) {
       break;
     }
   }
   *flib_def = std::move(client_graph->flib_def);
   std::swap(*input_types, client_graph->feed_types);
   std::swap(*output_types, client_graph->fetch_types);
   return s;
 }

Status tensorflow::NTSession::DecorateAndPublishGraphForDebug	(	const DebugOptions &	debug_options,
		Graph *	graph,
		Device *	device
	)

private

Definition at line 345 of file NTSession.cc.

Referenced by CheckNotClosed(), and GetOrCreateExecutors().

                                                                      {
   std::unique_ptr<DebugGraphDecoratorInterface> decorator;
   TF_RETURN_IF_ERROR(
       DebugGraphDecoratorRegistry::CreateDecorator(debug_options, &decorator));
 
   TF_RETURN_IF_ERROR(decorator->DecorateGraph(graph, device));
   TF_RETURN_IF_ERROR(decorator->PublishGraph(*graph, device->name()));
   return Status::OK();
 }

void tensorflow::NTSession::ExportCostModels ( CostModelManager::CostModelMap * cost_models )

inline

Definition at line 123 of file NTSession.h.

References cost_model_manager_.

                                                                    {
     cost_model_manager_.ExportCostModels(cost_models);
   }

Status tensorflow::NTSession::Extend ( const GraphDef & graph )

override

Definition at line 301 of file NTSession.cc.

References CheckNotClosed(), ExtendLocked(), graph_def_lock_, and checklumidiff::l.

                                               {
   TF_RETURN_IF_ERROR(CheckNotClosed());
   mutex_lock l(graph_def_lock_);
   return ExtendLocked(graph);
 }

Status tensorflow::NTSession::ExtendLocked ( const GraphDef & graph )

private

Definition at line 307 of file NTSession.cc.

References flib_def_, and MaybeInitializeExecutionState().

Referenced by Create(), and Extend().

                                                     {
   bool already_initialized;
   // If this is the first call, we can initialize the execution state
   // with `graph` and do not need to call `Extend()`.
   TF_RETURN_IF_ERROR(
       MaybeInitializeExecutionState(graph, &already_initialized));
   if (already_initialized) {
     TF_RETURN_IF_ERROR(flib_def_->AddLibrary(graph.library()));
     std::unique_ptr<GraphExecutionState> state;
     TF_RETURN_IF_ERROR(execution_state_->Extend(graph, &state));
     execution_state_.swap(state);
   }
   return Status::OK();
 }

Status tensorflow::NTSession::GetOrCreateExecutors	(	gtl::ArraySlice< string >	inputs,
		gtl::ArraySlice< string >	outputs,
		gtl::ArraySlice< string >	target_nodes,
		ExecutorsAndKeys **	executors_and_keys,
		RunStateArgs *	run_state_args
	)

private

Definition at line 975 of file NTSession.cc.

References CreateGraphs(), tensorflow::NTSession::RunStateArgs::debug_options, DecorateAndPublishGraphForDebug(), device_mgr_, device_set_, executor_lock_, plotBeamSpotDB::first, tensorflow::NTSession::RunStateArgs::graph, graph_def_lock_, cuy::graphs, tensorflow::NTSession::RunStateArgs::handle, handle_name_counter_, mps_fire::i, triggerObjects_cff::id, input, tensorflow::NTSession::RunStateArgs::is_partial_run, crabWrapper::key, checklumidiff::l, mps_check::lib, eostools::move(), gen::n, cscdqm::h::names, node_outputs_callback_, AlcaSiPixelAliHarvester0T_cff::options, options_, convertSQLitetoXML_cfg::output, and session_handle_.

Referenced by DecorateAndPublishGraphForDebug(), and PRunSetup().

                                   {
   int64 handle_name_counter_value = -1;
   if (LogMemory::IsEnabled() || run_state_args->is_partial_run) {
     handle_name_counter_value = handle_name_counter_.fetch_add(1);
   }
 
   string debug_tensor_watches_summary;
   if (!run_state_args->debug_options.debug_tensor_watch_opts().empty()) {
     debug_tensor_watches_summary = SummarizeDebugTensorWatches(
         run_state_args->debug_options.debug_tensor_watch_opts());
   }
 
   // Fast lookup path, no sorting.
   const string key = strings::StrCat(
       str_util::Join(inputs, ","), "->", str_util::Join(outputs, ","), "/",
       str_util::Join(target_nodes, ","), "/", run_state_args->is_partial_run,
       "/", debug_tensor_watches_summary);
   // Set the handle, if it's needed to log memory or for partial run.
   if (handle_name_counter_value >= 0) {
     run_state_args->handle =
         strings::StrCat(key, ";", handle_name_counter_value);
   }
 
   // See if we already have the executors for this run.
   {
     mutex_lock l(executor_lock_);  // could use reader lock
     auto it = executors_.find(key);
     if (it != executors_.end()) {
       *executors_and_keys = it->second.get();
       return Status::OK();
     }
   }
 
   // Slow lookup path, the unsorted key missed the cache.
   // Sort the inputs and outputs, and look up with the sorted key in case an
   // earlier call used a different order of inputs and outputs.
   //
   // We could consider some other signature instead of sorting that
   // preserves the same property to avoid the sort in the future.
   std::vector<string> inputs_sorted(inputs.begin(), inputs.end());
   std::sort(inputs_sorted.begin(), inputs_sorted.end());
   std::vector<string> outputs_sorted(outputs.begin(), outputs.end());
   std::sort(outputs_sorted.begin(), outputs_sorted.end());
   std::vector<string> tn_sorted(target_nodes.begin(), target_nodes.end());
   std::sort(tn_sorted.begin(), tn_sorted.end());
 
   const string sorted_key = strings::StrCat(
       str_util::Join(inputs_sorted, ","), "->",
       str_util::Join(outputs_sorted, ","), "/", str_util::Join(tn_sorted, ","),
       "/", run_state_args->is_partial_run, "/", debug_tensor_watches_summary);
   // Set the handle, if its needed to log memory or for partial run.
   if (handle_name_counter_value >= 0) {
     run_state_args->handle =
         strings::StrCat(sorted_key, ";", handle_name_counter_value);
   }
 
   // See if we already have the executors for this run.
   {
     mutex_lock l(executor_lock_);
     auto it = executors_.find(sorted_key);
     if (it != executors_.end()) {
       *executors_and_keys = it->second.get();
       // Insert this under the original key.
       executors_.emplace(key, it->second);
       return Status::OK();
     }
   }
 
   // Nothing found, so create the executors and store in the cache.
   BuildGraphOptions options;
   options.feed_endpoints = inputs_sorted;
   options.fetch_endpoints = outputs_sorted;
   options.target_nodes = tn_sorted;
   options.use_function_convention = !run_state_args->is_partial_run;
   if (!run_state_args->debug_options.debug_tensor_watch_opts().empty()) {
     options.debug_options = run_state_args->debug_options;
   }
 
   std::shared_ptr<ExecutorsAndKeys> ek(new ExecutorsAndKeys);
 
   // The executor_lock_ is intentionally released while executor is
   // being created.
   std::unordered_map<string, std::unique_ptr<Graph>> graphs;
   TF_RETURN_IF_ERROR(CreateGraphs(options, &graphs, &ek->flib_def,
                                   run_state_args, &ek->input_types,
                                   &ek->output_types));
 
   if (run_state_args->is_partial_run) {
     ek->graph = std::move(run_state_args->graph);
     std::unordered_set<StringPiece, StringPieceHasher> names;
     for (const string& input : inputs) {
       TensorId id(ParseTensorName(input));
       names.emplace(id.first);
     }
     for (const string& output : outputs) {
       TensorId id(ParseTensorName(output));
       names.emplace(id.first);
     }
     for (Node* n : ek->graph->nodes()) {
       if (names.count(n->name()) > 0) {
         ek->name_to_node.insert({n->name(), n});
       }
     }
   }
   ek->items.reserve(graphs.size());
   const auto& optimizer_opts =
       options_.config.graph_options().optimizer_options();
 
   int graph_def_version;
   {
     mutex_lock l(graph_def_lock_);
     graph_def_version =
         execution_state_->original_graph_def().versions().producer();
   }
   ek->proc_flr.reset(new ProcessFunctionLibraryRuntime(
       device_mgr_.get(), options_.env, graph_def_version, ek->flib_def.get(),
       optimizer_opts));
 
   GraphOptimizer optimizer(optimizer_opts);
   for (auto iter = graphs.begin(); iter != graphs.end(); ++iter) {
     const string& partition_name = iter->first;
     std::unique_ptr<Graph>& partition_graph = iter->second;
 
     Device* device;
     TF_RETURN_IF_ERROR(device_mgr_->LookupDevice(partition_name, &device));
 
     ek->items.resize(ek->items.size() + 1);
     auto* item = &(ek->items.back());
     auto lib = ek->proc_flr->GetFLR(partition_name);
     if (lib == nullptr) {
       return errors::Internal("Could not find device: ", partition_name);
     }
     item->flib = lib;
 
     LocalExecutorParams params;
     params.device = device;
     params.function_library = lib;
     auto opseg = device->op_segment();
     params.create_kernel = [this, lib, opseg](const NodeDef& ndef,
                                               OpKernel** kernel) {
       // We do not share the kernel via the OpSegment if the node is
       // stateless, or a function.
       // NOTE(mrry): We must not share function kernels (implemented
       // using `CallOp`) between subgraphs, because `CallOp::handle_`
       // is tied to a particular subgraph. Even if the function itself
       // is stateful, the `CallOp` that invokes it is not.
       if (!lib->IsStateful(ndef.op()) ||
           lib->GetFunctionLibraryDefinition()->Find(ndef.op()) != nullptr) {
         return lib->CreateKernel(ndef, kernel);
       }
       auto create_fn = [lib, &ndef](OpKernel** kernel) {
         return lib->CreateKernel(ndef, kernel);
       };
       // Kernels created for subgraph nodes need to be cached.  On
       // cache miss, create_fn() is invoked to create a kernel based
       // on the function library here + global op registry.
       return opseg->FindOrCreate(session_handle_, ndef.name(), kernel,
                                  create_fn);
     };
     params.delete_kernel = [lib](OpKernel* kernel) {
       // If the node is stateful, opseg owns it. Otherwise, delete it.
       if (kernel && !lib->IsStateful(kernel->type_string())) {
         delete kernel;
       }
     };
     params.node_outputs_cb = node_outputs_callback_;
 
     optimizer.Optimize(lib, options_.env, device, &iter->second,
                        /*shape_map=*/nullptr);
 
     // EXPERIMENTAL: tfdbg inserts debug nodes in the graph.
     if (!options.debug_options.debug_tensor_watch_opts().empty()) {
       TF_RETURN_IF_ERROR(DecorateAndPublishGraphForDebug(
           options.debug_options, partition_graph.get(), params.device));
     }
 
     TF_RETURN_IF_ERROR(EnsureMemoryTypes(DeviceType(device->device_type()),
                                          device->name(),
                                          partition_graph.get()));
     // NewLocalExecutor takes ownership of partition_graph.
     item->graph = partition_graph.get();
     item->executor = nullptr;
     item->device = device;
     Executor* executor;
     TF_RETURN_IF_ERROR(
         NewLocalExecutor(params, partition_graph.release(), &executor));
     item->executor.reset(executor);
   }
 
   // Cache the mapping from input/output names to graph elements to
   // avoid recomputing it every time.
   if (!run_state_args->is_partial_run) {
     // For regular `Run()`, we use the function calling convention, and so
     // maintain a mapping from input/output names to
     // argument/return-value ordinal index.
     for (size_t i = 0; i < inputs_sorted.size(); ++i) {
       const string& input = inputs_sorted[i];
       ek->input_name_to_index[input] = i;
     }
     for (size_t i = 0; i < outputs_sorted.size(); ++i) {
       const string& output = outputs_sorted[i];
       ek->output_name_to_index[output] = i;
     }
   } else {
     // For `PRun()`, we use the rendezvous calling convention, and so
     // maintain a mapping from input/output names to rendezvous keys.
     //
     // We always use the first device as the device name portion of the
     // key, even if we're feeding another graph.
     for (size_t i = 0; i < inputs_sorted.size(); ++i) {
       const string& input = inputs_sorted[i];
       ek->input_name_to_rendezvous_key[input] = GetRendezvousKey(
           input, device_set_.client_device()->attributes(), FrameAndIter(0, 0));
     }
     for (size_t i = 0; i < outputs_sorted.size(); ++i) {
       const string& output = outputs_sorted[i];
       ek->output_name_to_rendezvous_key[output] =
           GetRendezvousKey(output, device_set_.client_device()->attributes(),
                            FrameAndIter(0, 0));
     }
   }
 
   // Reacquire the lock, try to insert into the map.
   mutex_lock l(executor_lock_);
 
   // Another thread may have created the entry before us, in which case we will
   // reuse the already created one.
   auto insert_result = executors_.emplace(sorted_key, ek);
   // Insert the value under the original key, so the fast path lookup will work
   // if the user uses the same order of inputs, outputs, and targets again.
   executors_.emplace(key, insert_result.first->second);
   *executors_and_keys = insert_result.first->second.get();
 
   return Status::OK();
 }

bool graph_created_ tensorflow::NTSession::GUARDED_BY ( graph_def_lock_ )

private

GraphDef graph_def_ tensorflow::NTSession::GUARDED_BY ( graph_def_lock_ )

private

std::unordered_map<string, std::shared_ptr<ExecutorsAndKeys> > executors_ tensorflow::NTSession::GUARDED_BY ( executor_lock_ )

private

std::unordered_map<string, std::unique_ptr<RunState> > partial_runs_ tensorflow::NTSession::GUARDED_BY ( executor_lock_ )

private

std::unordered_map<string, string> stateful_placements_ tensorflow::NTSession::GUARDED_BY ( graph_def_lock_ )

private

std::unique_ptr<GraphExecutionState> execution_state_ tensorflow::NTSession::GUARDED_BY ( graph_def_lock_ )

private

bool closed_ tensorflow::NTSession::GUARDED_BY ( closed_lock_ )

private

tensorflow::Status tensorflow::NTSession::ListDevices ( std::vector< DeviceAttributes > * response )

override

Definition at line 1369 of file NTSession.cc.

References edmIntegrityCheck::d, and devices_.

                                            {
   response->clear();
   response->reserve(devices_.size());
   for (Device* d : devices_) {
     const DeviceAttributes& attrs = d->attributes();
     response->emplace_back(attrs);
   }
   return ::tensorflow::Status::OK();
 }

::tensorflow::Status tensorflow::NTSession::LocalDeviceManager ( const DeviceMgr ** output )

inlineoverride

Definition at line 118 of file NTSession.h.

References device_mgr_.

                                                                          {
     *output = device_mgr_.get();
     return ::tensorflow::Status::OK();
   }

Status tensorflow::NTSession::MaybeInitializeExecutionState	(	const GraphDef &	graph,
		bool *	out_already_initialized
	)

private

Definition at line 257 of file NTSession.cc.

References device_set_, flib_def_, AlcaSiPixelAliHarvester0T_cff::options, options_, and groupFilesInBlocks::temp.

Referenced by ExtendLocked().

                                                           {
   // If already initialized, do nothing.
   if (flib_def_ && execution_state_) {
     *out_already_initialized = true;
     return Status::OK();
   }
   // Set up the per-session execution state.
   // NOTE(mrry): The function library created here will be used for
   // all subsequent extensions of the graph.
   flib_def_.reset(
       new FunctionLibraryDefinition(OpRegistry::Global(), graph.library()));
   GraphExecutionStateOptions options;
   options.device_set = &device_set_;
   options.session_options = &options_;
   // TODO(mrry,suharshs): We explicitly copy `graph` so that
   // `MakeForBaseGraph()` can take ownership of its
   // contents. Previously this happened implicitly in calls to the
   // `GraphExecutionState`. Other sessions call
   // `MakeForBaseGraph` in such a way that we can destructively read
   // the passed-in `GraphDef`. In principle we could do the same here,
   // with a wider refactoring; we might revise the direct session so
   // that it copies the graph fewer times.
   GraphDef temp(graph);
   TF_RETURN_IF_ERROR(
       GraphExecutionState::MakeForBaseGraph(&temp, options, &execution_state_));
   graph_created_ = true;
   *out_already_initialized = false;
   return Status::OK();
 }

Status tensorflow::NTSession::PRun	(	const string &	handle,
		const NamedTensorList &	inputs,
		const std::vector< string > &	output_names,
		std::vector< Tensor > *	outputs
	)

override

Definition at line 703 of file NTSession.cc.

References cancellation_manager_, CheckFetch(), CheckNotClosed(), executor_lock_, input, crabWrapper::key, checklumidiff::l, LOG, tensorflow::NTSession::RunState::mu_, dataset::name, operation_timeout_in_ms_, convertSQLitetoXML_cfg::output, CfgNavigationSchool_cfi::parts, tensorflow::NTSession::RunState::pending_inputs, tensorflow::NTSession::RunState::pending_outputs, tensorflow::NTSession::RunState::PendingDone(), RecvPRunOutputs(), tensorflow::NTSession::RunState::rendez, alignCSCRings::s, SendPRunInputs(), session_state_, btagGenBb_cfi::Status, tensorflow::NTSession::RunState::tensor_store, WaitForNotification(), and dqm::qstatus::WARNING.

                                                        {
   TF_RETURN_IF_ERROR(CheckNotClosed());
   std::vector<string> parts = str_util::Split(handle, ';');
   const string& key = parts[0];
   // Get the executors for this partial run.
   ExecutorsAndKeys* executors_and_keys;
   RunState* run_state;
   {
     mutex_lock l(executor_lock_);  // could use reader lock
     auto exc_it = executors_.find(key);
     if (exc_it == executors_.end()) {
       return errors::InvalidArgument(
           "Must run 'setup' before performing partial runs!");
     }
     executors_and_keys = exc_it->second.get();
 
     auto prun_it = partial_runs_.find(handle);
     if (prun_it == partial_runs_.end()) {
       return errors::InvalidArgument(
           "Must run 'setup' before performing partial runs!");
     }
     run_state = prun_it->second.get();
 
     // Make sure that this is a new set of feeds that are still pending.
     for (const auto& input : inputs) {
       auto it = run_state->pending_inputs.find(input.first);
       if (it == run_state->pending_inputs.end()) {
         return errors::InvalidArgument(
             "The feed ", input.first,
             " was not specified in partial_run_setup.");
       } else if (it->second) {
         return errors::InvalidArgument("The feed ", input.first,
                                        " has already been fed.");
       }
     }
     // Check that this is a new set of fetches that are still pending.
     for (const auto& output : output_names) {
       auto it = run_state->pending_outputs.find(output);
       if (it == run_state->pending_outputs.end()) {
         return errors::InvalidArgument(
             "The fetch ", output, " was not specified in partial_run_setup.");
       } else if (it->second) {
         return errors::InvalidArgument("The fetch ", output,
                                        " has already been fetched.");
       }
     }
   }
 
   // Check that this new set of fetches can be computed from all the
   // feeds we have supplied.
   TF_RETURN_IF_ERROR(
       CheckFetch(inputs, output_names, executors_and_keys, run_state));
 
   // Send inputs.
   Status s = SendPRunInputs(inputs, executors_and_keys, run_state->rendez);
 
   // Receive outputs.
   if (s.ok()) {
     s = RecvPRunOutputs(output_names, executors_and_keys, run_state, outputs);
   }
 
   // Save the output tensors of this run we choose to keep.
   if (s.ok()) {
     s = run_state->tensor_store.SaveTensors(output_names, &session_state_);
   }
 
   {
     mutex_lock l(executor_lock_);
     // Delete the run state if there is an error or all fetches are done.
     bool done = true;
     if (s.ok()) {
       {
         mutex_lock l(run_state->mu_);
         if (!run_state->status.ok()) {
           LOG(WARNING) << "An error unrelated to this prun has been detected. "
                        << run_state->status;
         }
       }
       for (const auto& input : inputs) {
         auto it = run_state->pending_inputs.find(input.first);
         it->second = true;
       }
       for (const auto& name : output_names) {
         auto it = run_state->pending_outputs.find(name);
         it->second = true;
       }
       done = run_state->PendingDone();
     }
     if (done) {
       WaitForNotification(run_state, cancellation_manager_,
                           operation_timeout_in_ms_);
       partial_runs_.erase(handle);
     }
   }
 
   return s;
 }

Status tensorflow::NTSession::PRunSetup	(	const std::vector< string > &	input_names,
		const std::vector< string > &	output_names,
		const std::vector< string > &	target_nodes,
		string *	handle
	)

override

Definition at line 626 of file NTSession.cc.

References createfilelist::args, EnergyCorrector::c, cancellation_manager_, CheckNotClosed(), device_mgr_, devices_, executor_lock_, tensorflow::NTSession::RunState::executors_done, GetOrCreateExecutors(), graph_def_lock_, tensorflow::NTSession::RunStateArgs::handle, cmsBatch::handle, tensorflow::NTSession::RunStateArgs::is_partial_run, tensorflow::NTSession::ExecutorsAndKeys::items, checklumidiff::l, eostools::move(), options_, tensorflow::NTSession::RunState::rendez, SchedClosure(), session_state_, btagGenBb_cfi::Status, step_id_counter_, and sync_on_finish_.

                                                 {
   TF_RETURN_IF_ERROR(CheckNotClosed());
   {
     mutex_lock l(graph_def_lock_);
     if (!graph_created_) {
       return errors::InvalidArgument(
           "Session was not created with a graph before PRunSetup()!");
     }
   }
 
   // Check if we already have an executor for these arguments.
   ExecutorsAndKeys* executors_and_keys;
   // TODO(cais): TFDBG support for partial runs.
   DebugOptions debug_options;
   RunStateArgs run_state_args(debug_options);
   run_state_args.is_partial_run = true;
   TF_RETURN_IF_ERROR(GetOrCreateExecutors(input_names, output_names,
                                           target_nodes, &executors_and_keys,
                                           &run_state_args));
 
   // Create the run state and save it for future PRun calls.
   Executor::Args args;
   args.step_id = step_id_counter_.fetch_add(1);
   RunState* run_state =
       new RunState(input_names, output_names, args.step_id, &devices_);
   run_state->rendez = new IntraProcessRendezvous(device_mgr_.get());
   {
     mutex_lock l(executor_lock_);
     if (!partial_runs_
              .emplace(run_state_args.handle,
                       std::unique_ptr<RunState>(run_state))
              .second) {
       return errors::Internal("The handle '", run_state_args.handle,
                               "' created for this partial run is not unique.");
     }
   }
 
   // Start parallel Executors.
   const size_t num_executors = executors_and_keys->items.size();
   ExecutorBarrier* barrier = new ExecutorBarrier(
       num_executors, run_state->rendez, [run_state](const Status& ret) {
         if (!ret.ok()) {
           mutex_lock l(run_state->mu_);
           run_state->status.Update(ret);
         }
         run_state->executors_done.Notify();
       });
 
   args.rendezvous = run_state->rendez;
   args.cancellation_manager = cancellation_manager_;
   args.runner = [this](Executor::Args::Closure c) {
     SchedClosure(std::move(c));
   };
   args.session_state = &session_state_;
   args.tensor_store = &run_state->tensor_store;
   args.step_container = &run_state->step_container;
   if (LogMemory::IsEnabled()) {
     LogMemory::RecordStep(args.step_id, run_state_args.handle);
   }
   args.sync_on_finish = sync_on_finish_;
 
   if (options_.config.graph_options().build_cost_model()) {
     run_state->collector.reset(new StepStatsCollector(nullptr));
     args.stats_collector = run_state->collector.get();
   }
 
   for (auto& item : executors_and_keys->items) {
     item.executor->RunAsync(args, barrier->Get());
   }
 
   *handle = run_state_args.handle;
   return Status::OK();
 }

Status tensorflow::NTSession::RecvPRunOutputs	(	const std::vector< string > &	output_names,
		const ExecutorsAndKeys *	executors_and_keys,
		RunState *	run_state,
		std::vector< Tensor > *	outputs
	)

private

Definition at line 868 of file NTSession.cc.

References operation_timeout_in_ms_, tensorflow::NTSession::ExecutorsAndKeys::output_name_to_rendezvous_key, tensorflow::NTSession::RunState::rendez, alignCSCRings::s, and btagGenBb_cfi::Status.

Referenced by PRun().

                                 {
   Status s;
   if (!output_names.empty()) {
     outputs->resize(output_names.size());
   }
 
   Rendezvous::ParsedKey parsed;
   // Get the outputs from the rendezvous
   for (size_t output_offset = 0; output_offset < output_names.size();
        ++output_offset) {
     const string& output_name = output_names[output_offset];
     auto it =
         executors_and_keys->output_name_to_rendezvous_key.find(output_name);
     if (it == executors_and_keys->output_name_to_rendezvous_key.end()) {
       return errors::Internal("'", output_name,
                               "' is not a pre-defined fetch.");
     }
     const string& output_key = it->second;
     Tensor output_tensor;
     bool is_dead;
     IntraProcessRendezvous* rendez = run_state->rendez;
 
     s = Rendezvous::ParseKey(output_key, &parsed);
     if (s.ok()) {
       // Fetch data from the Rendezvous.
       s = rendez->Recv(parsed, Rendezvous::Args(), &output_tensor, &is_dead,
                        operation_timeout_in_ms_);
       if (is_dead && s.ok()) {
         s = errors::InvalidArgument("The tensor returned for ", output_name,
                                     " was not valid.");
       }
     }
     if (!s.ok()) {
       rendez->StartAbort(s);
       outputs->clear();
       return s;
     }
 
     (*outputs)[output_offset] = output_tensor;
   }
   return Status::OK();
 }

tensorflow::Status tensorflow::NTSession::Reset ( const std::vector< string > & containers )

Definition at line 1380 of file NTSession.cc.

References device_mgr_.

                                          {
   device_mgr_->ClearContainers(containers);
   return ::tensorflow::Status::OK();
 }

Status tensorflow::NTSession::ResourceHandleToInputTensor	(	const Tensor &	resource_tensor,
		Tensor *	retrieved_tensor
	)

private

Definition at line 803 of file NTSession.cc.

References session_state_.

Referenced by DecorateAndPublishGraphForDebug(), and SendPRunInputs().

                                                                             {
   if (resource_tensor.dtype() != DT_RESOURCE) {
     return errors::InvalidArgument(strings::StrCat(
         "ResourceHandleToInputTensor() received non-DT_RESOURCE Tensor: ",
         resource_tensor.dtype()));
   }
 
   const ResourceHandle& resource_handle =
       resource_tensor.scalar<ResourceHandle>()();
 
   if (resource_handle.container() ==
       SessionState::kTensorHandleResourceTypeName) {
     return session_state_.GetTensor(resource_handle.name(), retrieved_tensor);
   } else {
     return errors::InvalidArgument(strings::StrCat(
         "Invalid resource type hash code: ", resource_handle.hash_code(),
         "(name: ", resource_handle.name(),
         " type: ", resource_handle.maybe_type_name(),
         "). Perhaps a resource tensor was being provided as a feed? That is "
         "not currently allowed. Please file an issue at "
         "https://github.com/tensorflow/tensorflow/issues/new, ideally with a "
         "short code snippet that leads to this error message."));
   }
 }

Status tensorflow::NTSession::Run	(	const NamedTensorList &	inputs,
		const std::vector< string > &	output_names,
		const std::vector< string > &	target_nodes,
		std::vector< Tensor > *	outputs
	)

override

Definition at line 322 of file NTSession.cc.

Referenced by DecorateAndPublishGraphForDebug().

                                                       {
   RunMetadata run_metadata;
   return Run(RunOptions(), inputs, output_names, target_nodes, outputs,
              &run_metadata);
 }

::tensorflow::Status tensorflow::NTSession::Run	(	const ::tensorflow::RunOptions &	run_options,
		const NamedTensorList &	inputs,
		const std::vector< string > &	output_names,
		const std::vector< string > &	target_nodes,
		std::vector< Tensor > *	outputs,
		RunMetadata *	run_metadata
	)

override

void tensorflow::NTSession::SchedClosure ( std::function< void()> c )

private

Definition at line 193 of file NTSession.cc.

References EnergyCorrector::c.

Referenced by DecorateAndPublishGraphForDebug(), and PRunSetup().

                                                   {
   c();
 }

Status tensorflow::NTSession::SendPRunInputs	(	const std::vector< std::pair< string, Tensor >> &	inputs,
		const ExecutorsAndKeys *	executors_and_keys,
		IntraProcessRendezvous *	rendez
	)

private

Definition at line 829 of file NTSession.cc.

References input, tensorflow::NTSession::ExecutorsAndKeys::input_name_to_rendezvous_key, ResourceHandleToInputTensor(), alignCSCRings::s, and btagGenBb_cfi::Status.

Referenced by PRun().

                                                                      {
   Status s;
   Rendezvous::ParsedKey parsed;
   // Insert the input tensors into the local rendezvous by their
   // rendezvous key.
   for (const auto& input : inputs) {
     auto it =
         executors_and_keys->input_name_to_rendezvous_key.find(input.first);
     if (it == executors_and_keys->input_name_to_rendezvous_key.end()) {
       return errors::Internal("'", input.first, "' is not a pre-defined feed.");
     }
     const string& input_key = it->second;
 
     s = Rendezvous::ParseKey(input_key, &parsed);
     if (!s.ok()) {
       rendez->StartAbort(s);
       return s;
     }
 
     if (input.second.dtype() == DT_RESOURCE) {
       Tensor tensor_from_handle;
       s = ResourceHandleToInputTensor(input.second, &tensor_from_handle);
       if (s.ok()) {
         s = rendez->Send(parsed, Rendezvous::Args(), tensor_from_handle, false);
       }
     } else {
       s = rendez->Send(parsed, Rendezvous::Args(), input.second, false);
     }
 
     if (!s.ok()) {
       rendez->StartAbort(s);
       return s;
     }
   }
   return Status::OK();
 }

tensorflow::NTSession::TF_DISALLOW_COPY_AND_ASSIGN ( NTSession )

private

tensorflow::Status tensorflow::NTSession::WaitForNotification	(	Notification *	n,
		int64	timeout_in_ms
	)

private

Definition at line 1459 of file NTSession.cc.

References btagGenBb_cfi::Status.

Referenced by DecorateAndPublishGraphForDebug(), PRun(), and WaitForNotification().

                                                      {
   if (timeout_in_ms > 0) {
     const int64 timeout_in_us = timeout_in_ms * 1000;
     const bool notified =
         WaitForNotificationWithTimeout(notification, timeout_in_us);
     if (!notified) {
       return Status(error::DEADLINE_EXCEEDED,
                     "Timed out waiting for notification");
     }
   } else {
     notification->WaitForNotification();
   }
   return Status::OK();
 }

void tensorflow::NTSession::WaitForNotification	(	RunState *	run_state,
		CancellationManager *	cm,
		int64	timeout_in_ms
	)

private

Definition at line 1441 of file NTSession.cc.

References tensorflow::NTSession::RunState::executors_done, checklumidiff::l, tensorflow::NTSession::RunState::mu_, btagGenBb_cfi::Status, mps_update::status, and WaitForNotification().

                                                              {
   const Status status =
       WaitForNotification(&run_state->executors_done, timeout_in_ms);
   if (!status.ok()) {
     {
       mutex_lock l(run_state->mu_);
       run_state->status.Update(status);
     }
     cm->StartCancel();
     // We must wait for the executors to complete, because they have borrowed
     // references to `cm` and other per-step state. After this notification, it
     // is safe to clean up the step.
     run_state->executors_done.WaitForNotification();
   }
 }