yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804068926
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,135 +159,48 @@ Status
PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return
Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a
multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine
this logic
+Status Channel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
+ return _send_local_block(exec_status, eos);
}
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
- }
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(std::move(_pblock), eos, exec_status));
return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
- if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState,
Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(), block.bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block.rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(&block, _parent->sender_id(), true);
- if (eos) {
- _local_recvr->remove_sender(_parent->sender_id(), _be_number,
exec_status);
- }
- return Status::OK();
- } else {
- _serializer.reset_block();
- return _receiver_status;
+ if (!block.empty() || eos) {
+ RETURN_IF_ERROR(send_local_block(&block, eos, true));
}
+ _serializer.reset_block();
+ return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Block* block, bool can_be_moved) {
+Status Channel::send_local_block(Block* block, bool eos, bool can_be_moved) {
SCOPED_TIMER(_parent->local_send_timer());
if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState,
Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(),
block->bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(block, _parent->sender_id(), can_be_moved);
- return Status::OK();
- } else {
- return _receiver_status;
- }
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_remote_block(PBlock* block, bool eos, Status
exec_status) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>)
{
+ COUNTER_UPDATE(_parent->local_bytes_send_counter(), block->bytes());
+ COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- SCOPED_TIMER(_parent->brpc_send_timer());
- if (_send_remote_block_callback == nullptr) {
- _send_remote_block_callback =
DummyBrpcCallback<PTransmitDataResult>::create_shared();
- } else {
- RETURN_IF_ERROR(_wait_last_brpc());
- _send_remote_block_callback->cntl_->Reset();
- }
- VLOG_ROW << "Channel<Parent>::send_batch() instance_id=" <<
print_id(_fragment_instance_id)
- << " dest_node=" << _dest_node_id << " to_host=" <<
_brpc_dest_addr.hostname
- << " _packet_seq=" << _packet_seq << " row_desc=" <<
_row_desc.debug_string();
-
- _brpc_request->set_eos(eos);
- if (!exec_status.ok()) {
- exec_status.to_protobuf(_brpc_request->mutable_exec_status());
- }
- if (block != nullptr && !block->column_metas().empty()) {
- _brpc_request->set_allocated_block(block);
- }
- _brpc_request->set_packet_seq(_packet_seq++);
-
- _send_remote_block_callback->cntl_->set_timeout_ms(_brpc_timeout_ms);
- if (config::exchange_sink_ignore_eovercrowded) {
- _send_remote_block_callback->cntl_->ignore_eovercrowded();
- }
-
- {
- auto send_remote_block_closure =
- AutoReleaseClosure<PTransmitDataParams,
DummyBrpcCallback<PTransmitDataResult>>::
- create_unique(_brpc_request,
_send_remote_block_callback);
- if (enable_http_send_block(*_brpc_request)) {
- RETURN_IF_ERROR(transmit_block_httpv2(
- _state->exec_env(), std::move(send_remote_block_closure),
_brpc_dest_addr));
- } else {
- transmit_blockv2(*_brpc_stub,
std::move(send_remote_block_closure));
+ const auto sender_id = _parent->sender_id();
+ if (!block->empty()) [[likely]] {
+ _local_recvr->add_block(block, sender_id, can_be_moved);
}
- }
-
- if (block != nullptr) {
- static_cast<void>(_brpc_request->release_block());
- }
- return Status::OK();
-}
-template <typename Parent>
-Status Channel<Parent>::add_rows(Block* block, const std::vector<uint32_t>&
rows, bool eos) {
- if (_fragment_instance_id.lo == -1) {
+ if (eos) [[unlikely]] {
+ _local_recvr->remove_sender(sender_id, _be_number, Status::OK());
+ }
return Status::OK();
+ } else {
+ return _receiver_status;
}
-
- bool serialized = false;
- RETURN_IF_ERROR(
- _serializer.next_serialized_block(block, _ch_cur_pb_block, 1,
&serialized, eos, &rows));
- if (serialized) {
- RETURN_IF_ERROR(send_current_block(false, Status::OK()));
- }
-
- return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::close_wait(RuntimeState* state) {
+Status Channel::close_wait(RuntimeState* state) {
Review Comment:
这个close wait 函数还有意义吗？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
