This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 569ab30556 [bug](NodeChannel) fix OOM caused by pending queue in sink
send (#12359) (#12362)
569ab30556 is described below
commit 569ab3055699c975af5b64af01871c7597565c3e
Author: zhengyu <[email protected]>
AuthorDate: Wed Sep 7 20:49:08 2022 +0800
[bug](NodeChannel) fix OOM caused by pending queue in sink send (#12359)
(#12362)
Each NodeChannel has its own queue, with size up to 1/20 exec_mem_limit.
User will crash into OOM if set exec_mem_limit high. This commit uses
fixed number to control the total max memory used by NodeChannels.
Signed-off-by: freemandealer <[email protected]>
---
be/src/common/config.h | 3 +++
be/src/exec/tablet_sink.cpp | 5 ++++-
be/src/exec/tablet_sink.h | 2 +-
be/src/vec/sink/vtablet_sink.cpp | 4 ++++
4 files changed, 12 insertions(+), 2 deletions(-)
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 0d52f5f844..6a21eaf746 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -854,6 +854,9 @@ CONF_Int32(doris_remote_scanner_thread_pool_queue_size,
"10240");
// This config should be removed when the new scan node is ready.
CONF_Bool(enable_new_scan_node, "true");
+// limit the queue of pending batches which will be sent by a single
nodechannel
+CONF_mInt64(nodechannel_pending_queue_max_bytes, "67108864");
+
#ifdef BE_TEST
// test s3
CONF_String(test_s3_resource, "resource");
diff --git a/be/src/exec/tablet_sink.cpp b/be/src/exec/tablet_sink.cpp
index cd4d8c5c8d..32bb9af088 100644
--- a/be/src/exec/tablet_sink.cpp
+++ b/be/src/exec/tablet_sink.cpp
@@ -113,7 +113,6 @@ Status NodeChannel::init(RuntimeState* state) {
_rpc_timeout_ms = state->query_options().query_timeout * 1000;
_timeout_watch.start();
- _max_pending_batches_bytes = _parent->_load_mem_limit / 20; //TODO:
session variable percent
return Status::OK();
}
@@ -317,6 +316,10 @@ Status NodeChannel::add_row(Tuple* input_tuple, int64_t
tablet_id) {
//To simplify the add_row logic, postpone adding batch into req
until the time of sending req
_pending_batches.emplace(std::move(_cur_batch),
_cur_add_batch_request);
_pending_batches_num++;
+ VLOG_DEBUG << "OlapTableSink:" << _parent << " NodeChannel:" <<
this
+ << " pending_batches_bytes:" << _pending_batches_bytes
+ << " jobid:" << std::to_string(_state->load_job_id())
+ << " tabletid:" << tablet_id << " loadinfo:" <<
_load_info;
}
_cur_batch.reset(new RowBatch(*_row_desc, _batch_size));
diff --git a/be/src/exec/tablet_sink.h b/be/src/exec/tablet_sink.h
index 7ba1b385a2..c1660a31f0 100644
--- a/be/src/exec/tablet_sink.h
+++ b/be/src/exec/tablet_sink.h
@@ -286,7 +286,7 @@ protected:
// limit _pending_batches size
std::atomic<size_t> _pending_batches_bytes {0};
- size_t _max_pending_batches_bytes {10 * 1024 * 1024};
+ size_t _max_pending_batches_bytes
{(size_t)config::nodechannel_pending_queue_max_bytes};
std::mutex _pending_batches_lock; // reuse for vectorized
std::atomic<int> _pending_batches_num {0}; // reuse for vectorized
diff --git a/be/src/vec/sink/vtablet_sink.cpp b/be/src/vec/sink/vtablet_sink.cpp
index d84aceac77..025de0a035 100644
--- a/be/src/vec/sink/vtablet_sink.cpp
+++ b/be/src/vec/sink/vtablet_sink.cpp
@@ -201,6 +201,10 @@ Status VNodeChannel::add_block(vectorized::Block* block,
_pending_batches_bytes += _cur_mutable_block->allocated_bytes();
_pending_blocks.emplace(std::move(_cur_mutable_block),
_cur_add_block_request);
_pending_batches_num++;
+ VLOG_DEBUG << "VOlapTableSink:" << _parent << " VNodeChannel:" <<
this
+ << " pending_batches_bytes:" << _pending_batches_bytes
+ << " jobid:" << std::to_string(_state->load_job_id())
+ << " loadinfo:" << _load_info;
}
_cur_mutable_block.reset(new vectorized::MutableBlock({_tuple_desc}));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
