Method TransmitDataCompleteCb

be/src/runtime/krpc-data-stream-sender.cc:513–555

}

void KrpcDataStreamSender::Channel::TransmitDataCompleteCb() {
  std::unique_lock<SpinLock> l(lock_);
  DCHECK(rpc_in_flight_);
  DCHECK_NE(rpc_start_time_ns_, 0);
  int64_t total_time = MonotonicNanos() - rpc_start_time_ns_;
  const kudu::Status controller_status = rpc_controller_.status();
  if (LIKELY(controller_status.ok())) {
    DCHECK(rpc_in_flight_batch_ != nullptr);
    // 'receiver_latency_ns' is calculated with MonoTime, so it must be non-negative.
    DCHECK_GE(resp_.receiver_latency_ns(), 0);
    DCHECK_GE(total_time, resp_.receiver_latency_ns());
    int64_t row_batch_size = RowBatch::GetSerializedSize(*rpc_in_flight_batch_);
    int64_t network_time = total_time - resp_.receiver_latency_ns();
    COUNTER_ADD(parent_->bytes_sent_counter_, row_batch_size);
    if (LIKELY(network_time > 0)) {
      // 'row_batch_size' is bounded by FLAGS_rpc_max_message_size which shouldn't exceed
      // max 32-bit signed value so multiplication below should not overflow.
      DCHECK_LE(row_batch_size, numeric_limits<int32_t>::max());
      int64_t network_throughput = row_batch_size * NANOS_PER_SEC / network_time;
      parent_->network_throughput_counter_->UpdateCounter(network_throughput);
      parent_->network_time_stats_->UpdateCounter(network_time);
    }
    parent_->recvr_time_stats_->UpdateCounter(resp_.receiver_latency_ns());
    if (IsSlowRpc(total_time)) LogSlowRpc("TransmitData", total_time, resp_);
    Status rpc_status = Status::OK();
    int32_t status_code = resp_.status().status_code();
    if (status_code == TErrorCode::DATASTREAM_RECVR_CLOSED) {
      remote_recvr_closed_ = true;
    } else {
      rpc_status = StatusFromProto(resp_.status());
    }
    MarkDone(rpc_status);
  } else {
    if (IsSlowRpc(total_time)) {
      LogSlowFailedRpc("TransmitData", total_time, controller_status);
    }
    DoRpcFn rpc_fn =
        boost::bind(&KrpcDataStreamSender::Channel::DoTransmitDataRpc, this);
    const string& prepend =
        Substitute("TransmitData() to $0 failed", NetworkAddressPBToString(address_));
    HandleFailedRPC(rpc_fn, controller_status, prepend);
  }
}

Status KrpcDataStreamSender::Channel::DoTransmitDataRpc() {
  DCHECK(rpc_in_flight_batch_ != nullptr);