| 806 | } |
| 807 | |
| 808 | Status DebugGrpcIO::SendTensorThroughGrpcStream( |
| 809 | const DebugNodeKey& debug_node_key, const Tensor& tensor, |
| 810 | const uint64 wall_time_us, const string& grpc_stream_url, |
| 811 | const bool gated) { |
| 812 | if (gated && |
| 813 | !IsReadGateOpen(grpc_stream_url, debug_node_key.debug_node_name)) { |
| 814 | return Status::OK(); |
| 815 | } else { |
| 816 | std::vector<Event> events; |
| 817 | TF_RETURN_IF_ERROR(WrapTensorAsEvents(debug_node_key, tensor, wall_time_us, |
| 818 | kGrpcMessageSizeLimitBytes, &events)); |
| 819 | for (const Event& event : events) { |
| 820 | TF_RETURN_IF_ERROR( |
| 821 | SendEventProtoThroughGrpcStream(event, grpc_stream_url)); |
| 822 | } |
| 823 | if (IsWriteGateOpen(grpc_stream_url, debug_node_key.debug_node_name)) { |
| 824 | DebugGrpcChannel* debug_grpc_channel = nullptr; |
| 825 | TF_RETURN_IF_ERROR( |
| 826 | GetOrCreateDebugGrpcChannel(grpc_stream_url, &debug_grpc_channel)); |
| 827 | debug_grpc_channel->ReceiveAndProcessEventReplies(1); |
| 828 | // TODO(cais): Support new tensor value carried in the EventReply for |
| 829 | // overriding the value of the tensor being published. |
| 830 | } |
| 831 | return Status::OK(); |
| 832 | } |
| 833 | } |
| 834 | |
| 835 | Status DebugGrpcIO::ReceiveEventReplyProtoThroughGrpcStream( |
| 836 | EventReply* event_reply, const string& grpc_stream_url) { |
nothing calls this directly
no test coverage detected