Method Cancel

be/src/runtime/krpc-data-stream-recvr.cc:660–692

}

void KrpcDataStreamRecvr::SenderQueue::Cancel() {
  vector<std::unique_ptr<TransmitDataCtx>> rpcs_to_close;
  {
    unique_lock<SpinLock> l(lock_);
    if (is_cancelled_) return;
    is_cancelled_ = true;

    // Collect and dequeue deferred RPCs. Respond later without holding lock_.
    rpcs_to_close.reserve(deferred_rpcs_.size() + pending_deferred_rpcs_.size());
    while (!pending_deferred_rpcs_.empty()) {
      unique_ptr<TransmitDataCtx> ctx = DequeuePendingDeferredRpc();
      recvr_->deferred_rpc_tracker()->Release(ctx->rpc_context->GetTransferSize());
      rpcs_to_close.push_back(move(ctx));
    }
    while (!deferred_rpcs_.empty()) {
      unique_ptr<TransmitDataCtx> ctx = DequeueDeferredRpc();
      recvr_->deferred_rpc_tracker()->Release(ctx->rpc_context->GetTransferSize());
      rpcs_to_close.push_back(move(ctx));
    }
  }
  for (auto& ctx: rpcs_to_close) {
    RespondClosed(ctx.get());
  }
  rpcs_to_close.clear();
  VLOG(2) << "cancelled stream: fragment_instance_id="
          << PrintId(recvr_->fragment_instance_id())
          << " node_id=" << recvr_->dest_node_id();
  // Wake up all threads waiting to produce/consume batches. They will all
  // notice that the stream is cancelled and handle it.
  data_arrival_cv_.notify_all();
  PeriodicCounterUpdater::StopTimeSeriesCounter(
      recvr_->bytes_received_time_series_counter_);
}

void KrpcDataStreamRecvr::SenderQueue::Close() {
  unique_lock<SpinLock> l(lock_);