Method TakeOverEarlySender

be/src/runtime/krpc-data-stream-recvr.cc:601–643

}

void KrpcDataStreamRecvr::SenderQueue::TakeOverEarlySender(
    unique_ptr<TransmitDataCtx> ctx) {
  // TakeOverEarlySender() is called by the same thread which calls Close().
  // The receiver cannot be closed while this function is in progress so
  // 'recvr_->mgr_' shouldn't be NULL.
  DCHECK(TestInfo::is_test() || FragmentInstanceState::IsFragmentExecThread());
  DCHECK(!recvr_->closed_ && recvr_->mgr_ != nullptr);
  COUNTER_ADD(recvr_->total_received_batches_counter_, 1);
  int64_t serialized_size;
  Status status = GetBatchSize(ctx->request, ctx->rpc_context, &serialized_size);
  if (UNLIKELY(!status.ok())) {
    {
      unique_lock<SpinLock> l(lock_);
      if (!is_cancelled_) MarkErrorStatus(status, l);
    }
    TRACE_TO(
        ctx->rpc_context->trace(), "Error unpacking request: $0", status.GetDetail());
    DataStreamService::RespondRpc(status, ctx->response, ctx->rpc_context);
    return;
  }
  COUNTER_ADD(recvr_->bytes_received_counter_, serialized_size);
  DCHECK_GE(ctx->deserialized_size, 0);
  int sender_id = ctx->request->sender_id();
  int num_to_dequeue = 0;
  {
    unique_lock<SpinLock> l(lock_);
    // Only enqueue a deferred RPC if the sender queue is not yet cancelled.
    if (UNLIKELY(is_cancelled_)) {
      l.unlock();
      TRACE_TO(ctx->rpc_context->trace(), "Recvr closed");
      RespondClosed(ctx.get());
      return;
    }
    recvr_->deferred_rpc_tracker()->Consume(ctx->rpc_context->GetTransferSize());
    EnqueueDeferredRpc(move(ctx));
    num_to_dequeue = CollectPendingDeferredRpcs();
  }

  if (num_to_dequeue > 0) {
    recvr_->mgr_->EnqueueDeserializeTask(recvr_->fragment_instance_id(),
        recvr_->dest_node_id(), sender_id, num_to_dequeue);
  }
}

bool KrpcDataStreamRecvr::SenderQueue::DecrementSenders() {
  lock_guard<SpinLock> l(lock_);