Method ProcessDeferredRpc

be/src/runtime/krpc-data-stream-recvr.cc:538–590

}

void KrpcDataStreamRecvr::SenderQueue::ProcessDeferredRpc() {
  // Owns the first entry of 'pending_deferred_rpcs_',
  std::unique_ptr<TransmitDataCtx> ctx;
  Status status;
  {
    unique_lock<SpinLock> l(lock_);
    DCHECK_GT(num_deserialize_tasks_pending_, 0);
    --num_deserialize_tasks_pending_;

    if (pending_deferred_rpcs_.empty()) {
      // No need to respond to RPCs as it is done in Cancel().
      DCHECK(is_cancelled_);
      return;
    }
    // A sender queue cannot be cancelled if there is any deferred RPC.
    DCHECK(!is_cancelled_);

    // Dequeue and process the first entry from 'pending_deferred_rpcs_'.
    ctx = DequeuePendingDeferredRpc();
    TRACE_TO(ctx->rpc_context->trace(), "Processing deferred RPC");
    kudu::Slice tuple_offsets;
    kudu::Slice tuple_data;
    DCHECK_GE(ctx->deserialized_size, 0);
    int64_t batch_size = ctx->deserialized_size;
    status = UnpackRequest(ctx->request, ctx->rpc_context, &tuple_offsets, &tuple_data);
    // Reply with error status if the entry cannot be unpacked.
    if (UNLIKELY(!status.ok())) {
      MarkErrorStatus(status, l);
      recvr_->deferred_rpc_tracker()->Release(ctx->rpc_context->GetTransferSize());
      l.unlock();
      TRACE_TO(ctx->rpc_context->trace(),
          "Error unpacking deferred RPC: $0", status.GetDetail());
      DataStreamService::RespondRpc(status, ctx->response, ctx->rpc_context);
      return;
    }

    DCHECK_GE(pending_deserialized_size_, batch_size);
    pending_deserialized_size_ -= batch_size;

    DCHECK(CanEnqueue(batch_size));
    const RowBatchHeaderPB& header = ctx->request->row_batch_header();
    status = AddBatchWork(
        batch_size, header, tuple_offsets, tuple_data, &l, ctx->rpc_context);
    DCHECK(!status.ok() || !batch_queue_.empty());

    // Release to MemTracker while still holding the lock to prevent race with Close().
    recvr_->deferred_rpc_tracker()->Release(ctx->rpc_context->GetTransferSize());
  }

  // Responds to the sender to ack the insertion of the row batches.
  // No need to hold lock when enqueuing the response.
  DataStreamService::RespondRpc(status, ctx->response, ctx->rpc_context);
}

Status KrpcDataStreamRecvr::SenderQueue::GetBatchSize(
    const TransmitDataRequestPB* request,