Method CanEnqueue

be/src/runtime/krpc-data-stream-recvr.cc:348–359

}

inline bool KrpcDataStreamRecvr::SenderQueue::CanEnqueue(int64_t batch_size) const {
  lock_.DCheckLocked();
  // The queue is truly empty iff there is no pending insert. It's important that we
  // enqueue the new batch regardless of buffer limit if the queue is currently empty.
  // In the case of a merging receiver, batches are received from a specific queue
  // based on data order, and the pipeline will stall if the merger is waiting for data
  // from an empty queue that cannot be filled because the limit has been reached.
  bool queue_empty = batch_queue_.empty()
      && num_pending_enqueue_ == 0 && num_deserialize_tasks_pending_ == 0;
  bool result = queue_empty || !ExceedsLimit(batch_size);
  return result;
}

void KrpcDataStreamRecvr::SenderQueue::EnqueueDeferredRpc(
    unique_ptr<TransmitDataCtx> payload) {