Pipeline: use notify instead of polling for ExchangeReceiver
#9073
Conversation
ti-chi-bot
bot
added
the
release-note-none
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here.
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
bot
added
the
size/XL
SeaRise
changed the title
ExchangeReceiverExchangeReceiver
|
Is this pr ready to review? |
Need to wait #21312 merged |
ti-chi-bot
bot
added
the
needs-rebase
ti-chi-bot
bot
removed
the
needs-rebase
SeaRise
changed the title
ExchangeReceiverExchangeReceiver
|
/cc @windtalker @gengliqi |
| @@ -374,6 +374,8 @@ class DAGContext | |||
| /* const */ bool is_disaggregated_task = false; // a disagg task handling by the write node | |||
| // `tunnel_set` is always set by `MPPTask` and is used later. | |||
| MPPTunnelSetPtr tunnel_set; | |||
| // `mpp_receiver_set` is always set by `MPPTask` and is used later. | |||
There was a problem hiding this comment.
There was a problem hiding this comment.
It can be solved by adding a public function?
There was a problem hiding this comment.
A public function can be added here, but since mpp-tunnel-set is already in the public section, it might be better to maintain consistency.
There was a problem hiding this comment.
But it will change the destructor order in DAGContext, I'm not sure if there is some potential risk of this.
| @@ -152,6 +150,15 @@ MPMCQueueResult ReceivedMessageQueue::pop(size_t stream_id, ReceivedMessagePtr & | |||
| grpc_recv_queue.tryDequeue(); | |||
| #endif | |||
| } | |||
| ExchangeReceiverMetric::subDataSizeMetric(*data_size_in_queue, recv_msg->getPacket().ByteSizeLong()); | |||
There was a problem hiding this comment.
Should be a bug in the previous implementation, this subDataSizeMetric should only be called if recv_msg->getRemainingConsumers()->fetch_sub(1) == 1
|
|
||
| extern thread_local NotifyFuturePtr current_notify_future; | ||
| extern thread_local NotifyFuture * current_notify_future; |
There was a problem hiding this comment.
Looks like this change may increase the risk of using a object that have been already released, is it by design that the current_notify_future will not be released during its lifecycle?
There was a problem hiding this comment.
yes, current_notify_future is only released when the pipeline task changes from wait-for-notify to running
| : queue(log_, std::forward<Args>(args)...) | ||
| {} | ||
|
|
||
| void registerTask(TaskPtr && task) override { queue.registerPipeReadTask(std::move(task)); } |
There was a problem hiding this comment.
maybe rename it to registerPipelineReadTask?
There was a problem hiding this comment.
Because this is an override of NotifyFuture::registerTask, it cannot be renamed
What problem does this PR solve?
Issue Number: ref #8869
Problem Summary:
What is changed and how it works?
The cpu usage of wait reactor is reduced to 0 in tpch50.
Check List
Tests
Tested tpch50, no performance regression
Side effects
Documentation
Release note