Method AddBatch

be/src/runtime/krpc-data-stream-recvr.cc:479–536

}

void KrpcDataStreamRecvr::SenderQueue::AddBatch(const TransmitDataRequestPB* request,
    TransmitDataResponsePB* response, RpcContext* rpc_context) {
  // TODO: Add timers for time spent in this function and queue time in 'batch_queue_'.
  const RowBatchHeaderPB& header = request->row_batch_header();
  kudu::Slice tuple_offsets;
  kudu::Slice tuple_data;
  int64_t batch_size = RowBatch::GetDeserializedSize(header);
  Status status = UnpackRequest(request, rpc_context, &tuple_offsets, &tuple_data);
  if (UNLIKELY(!status.ok())) {
    {
      unique_lock<SpinLock> l(lock_);
      MarkErrorStatus(status, l);
    }
    TRACE_TO(rpc_context->trace(), "Error unpacking request: $0", status.GetDetail());
    DataStreamService::RespondRpc(status, response, rpc_context);
    return;
  }
  COUNTER_ADD(recvr_->total_received_batches_counter_, 1);
  // To be consistent with the senders, only count the sidecars size.
  COUNTER_ADD(recvr_->bytes_received_counter_, tuple_data.size() + tuple_offsets.size());

  {
    unique_lock<SpinLock> l(lock_);
    // There should be one or more senders left when this function is called. The reason
    // is that EndDataStream RPC is not sent until all outstanding TransmitData() RPC has
    // been replied to. There is at least one TransmitData() RPC which hasn't yet been
    // responded to if we reach here.
    DCHECK_GT(num_remaining_senders_, 0);
    if (UNLIKELY(is_cancelled_)) {
      lock_.unlock();
      Status cancel_status = Status::Expected(TErrorCode::DATASTREAM_RECVR_CLOSED,
          PrintId(recvr_->fragment_instance_id()), recvr_->dest_node_id());
      TRACE_TO(rpc_context->trace(), "Receiver was cancelled");
      DataStreamService::RespondRpc(cancel_status, response, rpc_context);
      return;
    }

    // If there's something in the queue or this batch will push us over the buffer
    // limit we need to wait until the queue gets drained. We store the rpc context
    // so that we can signal it at a later time to resend the batch that we couldn't
    // process here. If there are already deferred RPCs waiting in queue, the new
    // batch needs to line up after the deferred RPCs to avoid starvation of senders
    // in the non-merging case.
    if (!deferred_rpcs_.empty() || !CanEnqueue(batch_size)) {
      recvr_->deferred_rpc_tracker()->Consume(rpc_context->GetTransferSize());
      auto payload = make_unique<TransmitDataCtx>(
          request, response, rpc_context);
      EnqueueDeferredRpc(move(payload));
      return;
    }

    // At this point, we are committed to inserting the row batch into 'batch_queue_'.
    status = AddBatchWork(batch_size, header, tuple_offsets, tuple_data, &l, rpc_context);
  }

  // Respond to the sender to ack the insertion of the row batches.
  DataStreamService::RespondRpc(status, response, rpc_context);
}