| 777 | } |
| 778 | |
| 779 | KrpcDataStreamRecvr::KrpcDataStreamRecvr(KrpcDataStreamMgr* stream_mgr, |
| 780 | MemTracker* parent_tracker, const RowDescriptor* row_desc, |
| 781 | const RuntimeState& runtime_state, const TUniqueId& fragment_instance_id, |
| 782 | PlanNodeId dest_node_id, int num_senders, bool is_merging, int64_t total_buffer_limit, |
| 783 | RuntimeProfile* profile, BufferPool::ClientHandle* client) |
| 784 | : mgr_(stream_mgr), |
| 785 | runtime_state_(runtime_state), |
| 786 | fragment_instance_id_(fragment_instance_id), |
| 787 | dest_node_id_(dest_node_id), |
| 788 | row_desc_(row_desc), |
| 789 | is_merging_(is_merging), |
| 790 | closed_(false), |
| 791 | total_buffer_limit_(total_buffer_limit), |
| 792 | deferred_rpc_tracker_(new MemTracker(-1, "KrpcDeferredRpcs", parent_tracker)), |
| 793 | parent_tracker_(parent_tracker), |
| 794 | buffer_pool_client_(client), |
| 795 | profile_(profile), |
| 796 | dequeue_profile_(RuntimeProfile::Create(&pool_, RuntimeProfile::DEQUEUE, false)), |
| 797 | enqueue_profile_(RuntimeProfile::Create(&pool_, RuntimeProfile::ENQUEUE, false)) { |
| 798 | // Create one queue per sender if is_merging is true. |
| 799 | int num_queues = is_merging ? num_senders : 1; |
| 800 | sender_queues_.reserve(num_queues); |
| 801 | int num_sender_per_queue = is_merging ? 1 : num_senders; |
| 802 | for (int i = 0; i < num_queues; ++i) { |
| 803 | SenderQueue* queue = pool_.Add(new SenderQueue(this, num_sender_per_queue)); |
| 804 | sender_queues_.push_back(queue); |
| 805 | } |
| 806 | num_active_queues_.Store(num_queues); |
| 807 | |
| 808 | // Add the profiles of the dequeuing side (i.e. GetBatch()) and the enqueuing side |
| 809 | // (i.e. AddBatchWork()) as children of the owning exchange node's profile. |
| 810 | profile_->AddChild(dequeue_profile_); |
| 811 | profile_->AddChild(enqueue_profile_); |
| 812 | |
| 813 | // Initialize various counters for measuring dequeuing from queues. |
| 814 | bytes_dequeued_counter_ = |
| 815 | ADD_COUNTER(dequeue_profile_, "TotalBytesDequeued", TUnit::BYTES); |
| 816 | bytes_dequeued_time_series_counter_ = ADD_SYSTEM_TIME_SERIES_COUNTER( |
| 817 | dequeue_profile_, "BytesDequeued", bytes_dequeued_counter_); |
| 818 | queue_get_batch_timer_ = ADD_TIMER(dequeue_profile_, "TotalGetBatchTime"); |
| 819 | data_wait_timer_ = |
| 820 | ADD_CHILD_TIMER(dequeue_profile_, "DataWaitTime", "TotalGetBatchTime"); |
| 821 | inactive_timer_ = profile_->inactive_timer(); |
| 822 | first_batch_wait_total_timer_ = |
| 823 | ADD_TIMER(dequeue_profile_, "FirstBatchWaitTime"); |
| 824 | |
| 825 | // Initialize various counters for measuring enqueuing into queues. |
| 826 | bytes_received_counter_ = |
| 827 | ADD_COUNTER(enqueue_profile_, "TotalBytesReceived", TUnit::BYTES); |
| 828 | bytes_received_time_series_counter_ = ADD_SYSTEM_TIME_SERIES_COUNTER( |
| 829 | enqueue_profile_, "BytesReceived", bytes_received_counter_); |
| 830 | deserialize_row_batch_timer_ = |
| 831 | ADD_TIMER(enqueue_profile_, "DeserializeRowBatchTime"); |
| 832 | total_eos_received_counter_ = |
| 833 | ADD_COUNTER(enqueue_profile_, "TotalEosReceived", TUnit::UNIT); |
| 834 | total_early_senders_counter_ = |
| 835 | ADD_COUNTER(enqueue_profile_, "TotalEarlySenders", TUnit::UNIT); |
| 836 | total_received_batches_counter_ = |
nothing calls this directly
no test coverage detected