#include <NTSession.h>

Inheritance diagram for tensorflow::NTSession:

Classes
struct	ExecutorsAndKeys

struct	FunctionInfo

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::vector< std::unique_ptr< FunctionInfo > > functions_	GUARDED_BY (executor_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 82 of file NTSession.h.

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

Definition at line 83 of file NTSession.h.

Constructor & Destructor Documentation

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

Definition at line 191 of file NTSession.cc.

References ztail::d, device_mgr_, device_set_, devices_, dqm::qstatus::ERROR, unpackBuffers-CaloStage2::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 231 of file NTSession.cc.

References cancellation_manager_, Close(), ztail::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_);
     }
     for (auto d : device_mgr_->ListDevices()) {
       d->ClearResourceMgr();
     }
     functions_.clear();
     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 854 of file NTSession.cc.

References executor_lock_, cond::persistency::fetch(), dqmdumpme::first, tensorflow::NTSession::ExecutorsAndKeys::graph, triggerObjects_cff::id, input, cmsLHEtoEOSManager::l, dqmiodumpmetadata::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 266 of file NTSession.h.

References closed_lock_, CreateDebuggerState(), DecorateAndPublishGraphForDebug(), tensorflow::NTSession::PerPartitionExecutorsAndLib::device, tensorflow::NTSession::PerPartitionExecutorsAndLib::graph, pfDeepBoostedJetPreprocessParams_cfi::input_names, cmsLHEtoEOSManager::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 1325 of file NTSession.cc.

References cancellation_manager_, closed_lock_, tensorflow::NTSessionFactory::Deregister(), factory_, and cmsLHEtoEOSManager::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 281 of file NTSession.cc.

References ExtendLocked(), graph_def_lock_, init_error_, and cmsLHEtoEOSManager::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 321 of file NTSession.cc.

References pfDeepBoostedJetPreprocessParams_cfi::input_names.

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 1158 of file NTSession.cc.

References KineDebug3::count(), ztail::d, device_mgr_, device_set_, devices_, edge_name_counter_, flib_def_, tensorflow::NTSession::RunStateArgs::graph, graph_def_lock_, tensorflow::NTSession::RunStateArgs::is_partial_run, cmsLHEtoEOSManager::l, eostools::move(), Skims_PA_cff::name, options_, PatBasicFWLiteJetAnalyzer_Selector_cfg::outputs, ZMuMuAnalysisNtupler_cff::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 334 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 122 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 293 of file NTSession.cc.

References CheckNotClosed(), ExtendLocked(), graph_def_lock_, and cmsLHEtoEOSManager::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 299 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 918 of file NTSession.cc.

References CreateGraphs(), tensorflow::NTSession::RunStateArgs::debug_options, DecorateAndPublishGraphForDebug(), device_mgr_, device_set_, executor_lock_, dqmdumpme::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, B2GTnPMonitor_cfi::item, crabWrapper::key, cmsLHEtoEOSManager::l, mps_check::lib, eostools::move(), dqmiodumpmetadata::n, names, node_outputs_callback_, AlcaSiPixelAliHarvester0T_cff::options, options_, convertSQLitetoXML_cfg::output, CalibrationSummaryClient_cfi::params, 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::unique_ptr<FunctionInfo> func_info(new FunctionInfo);
     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, &func_info->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();
     }
     func_info->proc_flr.reset(new ProcessFunctionLibraryRuntime(
         device_mgr_.get(), options_.env, graph_def_version, func_info->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 = func_info->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_);
     functions_.push_back(std::move(func_info));
 
     // 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::vector<std::unique_ptr<FunctionInfo> > functions_ tensorflow::NTSession::GUARDED_BY ( executor_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 1310 of file NTSession.cc.

References ztail::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 117 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 253 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 659 of file NTSession.cc.

References cancellation_manager_, CheckFetch(), CheckNotClosed(), fileCollector::done, executor_lock_, input, crabWrapper::key, cmsLHEtoEOSManager::l, LOG, tensorflow::NTSession::RunState::mu_, Skims_PA_cff::name, operation_timeout_in_ms_, convertSQLitetoXML_cfg::output, contentValuesFiles::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 hlt_jetmet_dqm_QT_fromfile_cfg::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 592 of file NTSession.cc.

References writedatasetfile::args, HltBtagPostValidation_cff::c, cancellation_manager_, CheckNotClosed(), device_mgr_, devices_, executor_lock_, tensorflow::NTSession::RunState::executors_done, GetOrCreateExecutors(), graph_def_lock_, patZpeak::handle, tensorflow::NTSession::RunStateArgs::handle, tensorflow::NTSession::RunStateArgs::is_partial_run, B2GTnPMonitor_cfi::item, tensorflow::NTSession::ExecutorsAndKeys::items, cmsLHEtoEOSManager::l, eostools::move(), options_, tensorflow::NTSession::RunState::rendez, runTheMatrix::ret, personalPlayback::RunState, 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 813 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 1320 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 750 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 313 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 189 of file NTSession.cc.

References HltBtagPostValidation_cff::c.

Referenced by DecorateAndPublishGraphForDebug(), and PRunSetup().

189 { c(); }

HltBtagPostValidation_cff.c

c

Definition: HltBtagPostValidation_cff.py:31

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

private

Definition at line 775 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 1398 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 1383 of file NTSession.cc.

References tensorflow::NTSession::RunState::executors_done, cmsLHEtoEOSManager::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();
     }
   }