| 648 | } |
| 649 | |
| 650 | void KrpcDataStreamSender::Channel::EndDataStreamCompleteCb() { |
| 651 | std::unique_lock<SpinLock> l(lock_); |
| 652 | DCHECK(rpc_in_flight_); |
| 653 | DCHECK_NE(rpc_start_time_ns_, 0); |
| 654 | int64_t total_time_ns = MonotonicNanos() - rpc_start_time_ns_; |
| 655 | const kudu::Status controller_status = rpc_controller_.status(); |
| 656 | if (LIKELY(controller_status.ok())) { |
| 657 | // 'receiver_latency_ns' is calculated with MonoTime, so it must be non-negative. |
| 658 | DCHECK_GE(eos_resp_.receiver_latency_ns(), 0); |
| 659 | DCHECK_GE(total_time_ns, eos_resp_.receiver_latency_ns()); |
| 660 | int64_t network_time_ns = total_time_ns - eos_resp_.receiver_latency_ns(); |
| 661 | parent_->network_time_stats_->UpdateCounter(network_time_ns); |
| 662 | parent_->recvr_time_stats_->UpdateCounter(eos_resp_.receiver_latency_ns()); |
| 663 | if (IsSlowRpc(total_time_ns)) LogSlowRpc("EndDataStream", total_time_ns, eos_resp_); |
| 664 | MarkDone(StatusFromProto(eos_resp_.status())); |
| 665 | } else { |
| 666 | if (IsSlowRpc(total_time_ns)) { |
| 667 | LogSlowFailedRpc("EndDataStream", total_time_ns, controller_status); |
| 668 | } |
| 669 | DoRpcFn rpc_fn = |
| 670 | boost::bind(&KrpcDataStreamSender::Channel::DoEndDataStreamRpc, this); |
| 671 | const string& prepend = |
| 672 | Substitute("EndDataStream() to $0 failed", NetworkAddressPBToString(address_)); |
| 673 | HandleFailedRPC(rpc_fn, controller_status, prepend); |
| 674 | } |
| 675 | } |
| 676 | |
| 677 | Status KrpcDataStreamSender::Channel::DoEndDataStreamRpc() { |
| 678 | DCHECK(rpc_in_flight_); |
nothing calls this directly
no test coverage detected