[refine](pipelineX)Make the 'set ready' logic of SenderQueue in pipelineX the same as that in the pipeline.

be/src/vec/runtime/vdata_stream_recvr.cpp

@@ -100,7 +100,8 @@ Status VDataStreamRecvr::SenderQueue::_inner_get_batch_without_lock(Block* block
  auto [next_block, block_byte_size] = std::move(_block_queue.front());
  _recvr->update_blocks_memory_usage(-block_byte_size);
  _block_queue.pop_front();
- if (_block_queue.size() == 0 && _dependency) {
+ _record_debug_info();
+ if (_block_queue.empty() && _dependency) {
  if (!_is_cancelled && _num_remaining_senders > 0) {
  _dependency->block();
  }
@@ -122,6 +123,16 @@ Status VDataStreamRecvr::SenderQueue::_inner_get_batch_without_lock(Block* block
  return Status::OK();
 }
 
+void VDataStreamRecvr::SenderQueue::try_set_dep_ready_without_lock() {
+ if (!_dependency) {
+ return;
+ }
+ const bool should_wait = !_is_cancelled && _block_queue.empty() && _num_remaining_senders > 0;
+ if (!should_wait) {
+ _dependency->set_ready();
+ }
+}
+
 Status VDataStreamRecvr::SenderQueue::add_block(const PBlock& pblock, int be_number,
  int64_t packet_seq,
  ::google::protobuf::Closure** done) {
@@ -178,9 +189,8 @@ Status VDataStreamRecvr::SenderQueue::add_block(const PBlock& pblock, int be_num
 
  if (!empty) {
  _block_queue.emplace_back(std::move(block), block_byte_size);
- if (_dependency) {
- _dependency->set_ready();
- }
+ _record_debug_info();
+ try_set_dep_ready_without_lock();
  }
  // if done is nullptr, this function can't delay this response
  if (done != nullptr && _recvr->exceeds_limit(block_byte_size)) {
@@ -234,9 +244,8 @@ void VDataStreamRecvr::SenderQueue::add_block(Block* block, bool use_move) {
 
  if (!empty) {
  _block_queue.emplace_back(std::move(nblock), block_mem_size);
- if (_dependency) {
- _dependency->set_ready();
- }
+ _record_debug_info();
+ try_set_dep_ready_without_lock();
  _data_arrival_cv.notify_one();
  }
 
@@ -269,13 +278,12 @@ void VDataStreamRecvr::SenderQueue::decrement_senders(int be_number) {
  _sender_eos_set.insert(be_number);
  DCHECK_GT(_num_remaining_senders, 0);
  _num_remaining_senders--;
+ _record_debug_info();
  VLOG_FILE << "decremented senders: fragment_instance_id="
  << print_id(_recvr->fragment_instance_id()) << " node_id=" << _recvr->dest_node_id()
  << " #senders=" << _num_remaining_senders;
  if (_num_remaining_senders == 0) {
- if (_dependency) {
- _dependency->set_ready();
- }
+ try_set_dep_ready_without_lock();
  _data_arrival_cv.notify_one();
  }
 }
@@ -288,9 +296,7 @@ void VDataStreamRecvr::SenderQueue::cancel(Status cancel_status) {
  }
  _is_cancelled = true;
  _cancel_status = cancel_status;
- if (_dependency) {
- _dependency->set_ready();
- }
+ try_set_dep_ready_without_lock();
  VLOG_QUERY << "cancelled stream: _fragment_instance_id="
  << print_id(_recvr->fragment_instance_id())
  << " node_id=" << _recvr->dest_node_id();
@@ -318,9 +324,7 @@ void VDataStreamRecvr::SenderQueue::close() {
  // is clear will be memory leak
  std::lock_guard<std::mutex> l(_lock);
  _is_cancelled = true;
- if (_dependency) {
- _dependency->set_ready();
- }
+ try_set_dep_ready_without_lock();
 
  for (auto closure_pair : _pending_closures) {
  closure_pair.first->Run();
@@ -561,9 +565,8 @@ void VDataStreamRecvr::PipSenderQueue::add_block(Block* block, bool use_move) {
  return;
  }
  _block_queue.emplace_back(std::move(nblock), block_mem_size);
- if (_dependency) {
- _dependency->set_ready();
- }
+ _record_debug_info();
+ try_set_dep_ready_without_lock();
  COUNTER_UPDATE(_recvr->_local_bytes_received_counter, block_mem_size);
  _recvr->update_blocks_memory_usage(block_mem_size);
  _data_arrival_cv.notify_one();

be/src/vec/runtime/vdata_stream_recvr.h

@@ -30,6 +30,7 @@
 #include <memory>
 #include <mutex>
 #include <ostream>
+#include <sstream>
 #include <thread>
 #include <unordered_map>
 #include <unordered_set>
@@ -240,6 +241,52 @@ class VDataStreamRecvr::SenderQueue {
  friend struct pipeline::ExchangeDataDependency;
  Status _inner_get_batch_without_lock(Block* block, bool* eos);
 
+ void try_set_dep_ready_without_lock();
+
+ // To record information about several variables in the event of a DCHECK failure.
+ // DCHECK(_is_cancelled || !_block_queue.empty() || _num_remaining_senders == 0)
+#ifndef NDEBUG
+ constexpr static auto max_record_number = 128;
+ std::list<size_t> _record_block_queue;
+ std::list<int> _record_num_remaining_senders;
+#else
+#endif
+
+ // only in debug
+ ALWAYS_INLINE inline void _record_debug_info() {
+#ifndef NDEBUG
+ if (_record_block_queue.size() > max_record_number) {
+ _record_block_queue.pop_front();
+ }
+ if (_record_num_remaining_senders.size() > max_record_number) {
+ _record_num_remaining_senders.pop_front();
+ }
+ _record_block_queue.push_back(_block_queue.size());
+ _record_num_remaining_senders.push_back(_num_remaining_senders);
+#else
+#endif
+ }
+
+ ALWAYS_INLINE inline std::string _debug_string_info() {
+#ifndef NDEBUG
+ std::stringstream out;
+ DCHECK_EQ(_record_block_queue.size(), _record_num_remaining_senders.size());
+ out << "record_debug_info [ \n";
+
+ auto it1 = _record_block_queue.begin();
+ auto it2 = _record_num_remaining_senders.begin();
+ for (; it1 != _record_block_queue.end(); it1++, it2++) {
+ out << "( "
+ << "_block_queue size : " << *it1 << " , _num_remaining_senders : " << *it2
+ << " ) \n";
+ }
+ out << " ]\n";
+ return out.str();
+#else
+#endif
+ return "";
+ }
+
  // Not managed by this class
  VDataStreamRecvr* _recvr = nullptr;
  std::mutex _lock;
@@ -272,7 +319,8 @@ class VDataStreamRecvr::PipSenderQueue : public SenderQueue {
  DCHECK(_is_cancelled || !_block_queue.empty() || _num_remaining_senders == 0)
  << " _is_cancelled: " << _is_cancelled
  << ", _block_queue_empty: " << _block_queue.empty()
- << ", _num_remaining_senders: " << _num_remaining_senders;
+ << ", _num_remaining_senders: " << _num_remaining_senders << "\n"
+ << _debug_string_info();
  return _inner_get_batch_without_lock(block, eos);
  }