| 847 | |
| 848 | template <class PropagatorStateType> |
| 849 | void ExecutorState<PropagatorStateType>::BatchProcess(std::vector<TaggedNode> nodes, |
| 850 | int nodes_count, |
| 851 | int64_t scheduled_nsec) { |
| 852 | WithContext wc(context_); |
| 853 | TaggedNodeSeq ready; |
| 854 | TaggedNodeReadyQueue inline_ready; |
| 855 | |
| 856 | // Parameters passed to OpKernel::Compute. |
| 857 | TensorValueVec inputs; |
| 858 | AllocatorAttributeVec input_alloc_attrs; |
| 859 | |
| 860 | OpKernelContext::Params params; |
| 861 | params.step_id = step_id_; |
| 862 | params.round_step_id = round_step_id_; |
| 863 | // Override device's threadpool if user provides an intra_op_threadpool |
| 864 | Device* device = immutable_state_.params().device; |
| 865 | if (user_device_) { |
| 866 | params.device = user_device_.get(); |
| 867 | } else { |
| 868 | params.device = device; |
| 869 | } |
| 870 | params.start_time_usecs = start_time_usecs_; |
| 871 | params.deadline = deadline_; |
| 872 | params.log_memory = log_memory_; |
| 873 | params.rendezvous = rendezvous_; |
| 874 | params.create_rendezvous = create_rendezvous_; |
| 875 | params.global_rendezvous = global_rendezvous_; |
| 876 | params.collective_executor = collective_executor_; |
| 877 | params.session_state = session_state_; |
| 878 | params.session_handle = session_handle_; |
| 879 | params.session_metadata = session_metadata_; |
| 880 | params.tensor_store = tensor_store_; |
| 881 | params.cancellation_manager = cancellation_manager_; |
| 882 | params.call_frame = call_frame_; |
| 883 | params.function_library = immutable_state_.params().function_library; |
| 884 | params.resource_manager = device->resource_manager(); |
| 885 | params.step_container = step_container_; |
| 886 | params.slice_reader_cache = slice_reader_cache_; |
| 887 | params.inputs = &inputs; |
| 888 | params.input_alloc_attrs = &input_alloc_attrs; |
| 889 | params.runner = &runner_; |
| 890 | params.run_all_kernels_inline = |
| 891 | (executor_policy_ == ExecutorPolicy::USE_INLINE_EXECUTOR); |
| 892 | params.stats_collector = stats_collector_; |
| 893 | params.inc_num_deferred_ops_function = [this]() { |
| 894 | mutex_lock lock(num_deferred_ops_mu_); |
| 895 | num_deferred_ops_++; |
| 896 | }; |
| 897 | params.dec_num_deferred_ops_function = [this]() { |
| 898 | bool finish_when_deferred_ops_done = false; |
| 899 | { |
| 900 | mutex_lock lock(num_deferred_ops_mu_); |
| 901 | num_deferred_ops_--; |
| 902 | if (num_deferred_ops_ == 0) { |
| 903 | finish_when_deferred_ops_done = finish_when_deferred_ops_done_; |
| 904 | } |
| 905 | } |
| 906 | // Invoke Finish if the graph processing has completed. Finish is always |
no test coverage detected