You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/09/11 06:48:39 UTC
(doris) branch spill_and_reserve updated (c1d8332c53b -> 8f27ff3e848)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch spill_and_reserve
in repository https://gitbox.apache.org/repos/asf/doris.git
from c1d8332c53b [fix](exchange) fix coredump of update counter is block is null
new 12c06129483 [opt](scanner profile) More counter for scanner (#40144)
new 8f27ff3e848 [fix](scanner) Fix deadlock when scanner submit failed (#40495)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/pipeline/exec/scan_operator.cpp | 8 ++
be/src/pipeline/exec/scan_operator.h | 2 +
be/src/vec/exec/scan/scanner_context.cpp | 93 ++++++++++++++++------
be/src/vec/exec/scan/scanner_context.h | 17 ++--
be/src/vec/exec/scan/scanner_scheduler.cpp | 55 ++++++++-----
be/src/vec/exec/scan/scanner_scheduler.h | 2 +-
be/src/vec/exec/scan/vscanner.cpp | 1 +
...e_by_frontend.groovy => scanner_profile.groovy} | 84 +++++++++++--------
8 files changed, 175 insertions(+), 87 deletions(-)
copy regression-test/suites/query_profile/{test_execute_by_frontend.groovy => scanner_profile.groovy} (55%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/02: [fix](scanner) Fix deadlock when scanner submit failed (#40495)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch spill_and_reserve
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8f27ff3e8481f9b23d5584e8d5ec6a7db4c4f705
Author: zhiqiang <se...@163.com>
AuthorDate: Tue Sep 10 11:00:56 2024 +0800
[fix](scanner) Fix deadlock when scanner submit failed (#40495)
We have dead lock when submit scanner to scheduler failed.
pstack looks like
```txt
Thread 2012 (Thread 0x7f87363fb700 (LWP 4179707) "Pipe_normal [wo"):
#0 0x00007f8b8f3dc82d in __lll_lock_wait () from /lib64/libpthread.so.0
#1 0x00007f8b8f3d5ad9 in pthread_mutex_lock () from /lib64/libpthread.so.0
#2 0x000055b20f333e7a in __gthread_mutex_lock (__mutex=0x7f8733d960a8) at /mnt/disk1/hezhiqiang/toolchains/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/x86_64-linux-gnu/c++/11/bits/gthr-default
.h:749
#3 std::mutex::lock (this=0x7f8733d960a8) at /mnt/disk1/hezhiqiang/toolchains/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_mutex.h:100
#4 std::lock_guard<std::mutex>::lock_guard (__m=..., this=<optimized out>) at /mnt/disk1/hezhiqiang/toolchains/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_mutex.h:229
#5 doris::vectorized::ScannerContext::append_block_to_queue (this=<optimized out>, scan_task=...) at /mnt/disk1/hezhiqiang/doris/be/src/vec/exec/scan/scanner_context.cpp:234
#6 0x000055b20f32c0f9 in doris::vectorized::ScannerScheduler::submit (this=<optimized out>, ctx=..., scan_task=...) at /mnt/disk1/hezhiqiang/doris/be/src/vec/exec/scan/scanner_scheduler.cpp:209
#7 0x000055b20f3338fc in doris::vectorized::ScannerContext::submit_scan_task (this=this@entry=0x7f8733d96010, scan_task=...) at /mnt/disk1/hezhiqiang/doris/be/src/vec/exec/scan/scanner_context.cpp:217
#8 0x000055b20f3346cd in doris::vectorized::ScannerContext::get_block_from_queue (this=0x7f8733d96010, state=<optimized out>, block=0x7f871f728de0, eos=0x7f871abce470, id=<optimized out>) at /mnt/disk1/hezhiqiang/doris/be/src/vec/exec/scan/scanner_context.cpp:290
#9 0x000055b214cb4f13 in doris::pipeline::ScanOperatorX<doris::pipeline::OlapScanLocalState>::get_block (this=<optimized out>, state=0x7f872f0eb400, block=0x7f8b8f3dc82d <__lll_lock_wait+29>, eos=0x7f871abce470) at /mnt/disk1/hezhiqiang/doris/be/src/pipeline/exec/scan_operator.cpp:1292
#10 0x000055b2142b5772 in doris::pipeline::ScanOperatorX<doris::pipeline::OlapScanLocalState>::get_block_after_projects (this=0x80, state=0x0, block=0x7f8b8f3dc82d <__lll_lock_wait+29>, eos=0x7f8733d960a8) at /mnt/disk1/hezhiqiang/doris/be/src/pipeline/exec/scan_operator.h:363
#11 0x000055b2142e7880 in doris::pipeline::StatefulOperatorX<doris::pipeline::StreamingAggLocalState>::get_block (this=0x7f871f9bee00, state=0x7f872f0eb400, block=0x7f8716d49060, eos=0x7f87363f4937) at /mnt/disk1/hezhiqiang/doris/be/src/pipeline/exec/operator.cpp:587
```
Deallock happens with following
```cpp
Status ScannerContext::get_block_from_queue {
std::unique_lock l(_transfer_lock);
...
if (scan_task->is_eos()) {
...
} else {
// resubmit current running scanner to read the next block
submit_scan_task(scan_task);
}
}
ScannerContext::submit_scan_task(std::shared_ptr<ScanTask> scan_task) {
_scanner_scheduler->submit(shared_from_this(), scan_task);
}
void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
std::shared_ptr<ScanTask> scan_task) {
...
if (auto ret = sumbit_task(); !ret) {
scan_task->set_status(Status::InternalError(
"Failed to submit scanner to scanner pool reason:" + std::string(ret.msg()) +
"|type:" + std::to_string(type)));
ctx->append_block_to_queue(scan_task);
return;
}
}
void ScannerContext::append_block_to_queue(std::shared_ptr<ScanTask> scan_task) {
...
std::lock_guard<std::mutex> l(_transfer_lock);
...
}
```
Since mutex in cpp is not re-enterable, so the scanner thread will
deadlock with itself.
This pr fix the problem by making `ScannerScheduler::submit` return a
Status instead of doing append failed task to the ScannerContext. The
caller itself will decide where resubmit the scanner or just abort the
execution of the query.
---
be/src/vec/exec/scan/scanner_context.cpp | 44 ++++++++++++++++++++++--------
be/src/vec/exec/scan/scanner_context.h | 4 +--
be/src/vec/exec/scan/scanner_scheduler.cpp | 31 +++++++++++----------
be/src/vec/exec/scan/scanner_scheduler.h | 2 +-
4 files changed, 53 insertions(+), 28 deletions(-)
diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp
index 5cc20c214c1..cbb3d0f5723 100644
--- a/be/src/vec/exec/scan/scanner_context.cpp
+++ b/be/src/vec/exec/scan/scanner_context.cpp
@@ -152,7 +152,7 @@ Status ScannerContext::init() {
for (int i = 0; i < _max_thread_num; ++i) {
std::weak_ptr<ScannerDelegate> next_scanner;
if (_scanners.try_dequeue(next_scanner)) {
- submit_scan_task(std::make_shared<ScanTask>(next_scanner));
+ RETURN_IF_ERROR(submit_scan_task(std::make_shared<ScanTask>(next_scanner)));
_num_running_scanners++;
}
}
@@ -196,10 +196,10 @@ bool ScannerContext::empty_in_queue(int id) {
return _blocks_queue.empty();
}
-void ScannerContext::submit_scan_task(std::shared_ptr<ScanTask> scan_task) {
+Status ScannerContext::submit_scan_task(std::shared_ptr<ScanTask> scan_task) {
_scanner_sched_counter->update(1);
_num_scheduled_scanners++;
- _scanner_scheduler->submit(shared_from_this(), scan_task);
+ return _scanner_scheduler->submit(shared_from_this(), scan_task);
}
void ScannerContext::append_block_to_queue(std::shared_ptr<ScanTask> scan_task) {
@@ -247,10 +247,15 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
auto scan_task = _blocks_queue.front();
DCHECK(scan_task);
+ // The abnormal status of scanner may come from the execution of the scanner itself,
+ // or come from the scanner scheduler, such as TooManyTasks.
if (!scan_task->status_ok()) {
+ // TODO: If the scanner status is TooManyTasks, maybe we can retry the scanner after a while.
+ _process_status = scan_task->get_status();
_set_scanner_done();
- return scan_task->get_status();
+ return _process_status;
}
+
if (!scan_task->cached_blocks.empty()) {
auto [current_block, block_size] = std::move(scan_task->cached_blocks.front());
scan_task->cached_blocks.pop_front();
@@ -263,13 +268,20 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
block->swap(*current_block);
return_free_block(std::move(current_block));
} else {
+ // This scan task do not have any cached blocks.
_blocks_queue.pop_front();
- if (scan_task->is_eos()) { // current scanner is finished, and no more data to read
+ // current scanner is finished, and no more data to read
+ if (scan_task->is_eos()) {
_num_finished_scanners++;
std::weak_ptr<ScannerDelegate> next_scanner;
// submit one of the remaining scanners
if (_scanners.try_dequeue(next_scanner)) {
- submit_scan_task(std::make_shared<ScanTask>(next_scanner));
+ auto submit_status = submit_scan_task(std::make_shared<ScanTask>(next_scanner));
+ if (!submit_status.ok()) {
+ _process_status = submit_status;
+ _set_scanner_done();
+ return _process_status;
+ }
} else {
// no more scanner to be scheduled
// `_free_blocks` serve all running scanners, maybe it's too large for the remaining scanners
@@ -284,11 +296,16 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
}
} else {
// resubmit current running scanner to read the next block
- submit_scan_task(scan_task);
+ Status submit_status = submit_scan_task(scan_task);
+ if (!submit_status.ok()) {
+ _process_status = submit_status;
+ _set_scanner_done();
+ return _process_status;
+ }
}
}
// scale up
- _try_to_scale_up();
+ RETURN_IF_ERROR(_try_to_scale_up());
}
if (_num_finished_scanners == _all_scanners.size() && _blocks_queue.empty()) {
@@ -303,7 +320,7 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
return Status::OK();
}
-void ScannerContext::_try_to_scale_up() {
+Status ScannerContext::_try_to_scale_up() {
// Four criteria to determine whether to increase the parallelism of the scanners
// 1. It ran for at least `SCALE_UP_DURATION` ms after last scale up
// 2. Half(`WAIT_BLOCK_DURATION_RATIO`) of the duration is waiting to get blocks
@@ -320,7 +337,7 @@ void ScannerContext::_try_to_scale_up() {
// when _last_wait_duration_ratio > 0, it has scaled up before.
// we need to determine if the scale-up is effective:
// the wait duration ratio after last scaling up should less than 80% of `_last_wait_duration_ratio`
- return;
+ return Status::OK();
}
bool is_scale_up = false;
@@ -335,7 +352,10 @@ void ScannerContext::_try_to_scale_up() {
// get enough memory to launch one more scanner.
std::weak_ptr<ScannerDelegate> scale_up_scanner;
if (_scanners.try_dequeue(scale_up_scanner)) {
- submit_scan_task(std::make_shared<ScanTask>(scale_up_scanner));
+ // Just return error to caller.
+ // Because _try_to_scale_up is called under _transfer_lock locked, if we add the scanner
+ // to the block queue, we will get a deadlock.
+ RETURN_IF_ERROR(submit_scan_task(std::make_shared<ScanTask>(scale_up_scanner)));
_num_running_scanners++;
_scale_up_scanners_counter->update(1);
is_scale_up = true;
@@ -350,6 +370,8 @@ void ScannerContext::_try_to_scale_up() {
_total_wait_block_time = 0;
}
}
+
+ return Status::OK();
}
Status ScannerContext::validate_block_schema(Block* block) {
diff --git a/be/src/vec/exec/scan/scanner_context.h b/be/src/vec/exec/scan/scanner_context.h
index f93d01eef88..03c4e5a4f1b 100644
--- a/be/src/vec/exec/scan/scanner_context.h
+++ b/be/src/vec/exec/scan/scanner_context.h
@@ -139,7 +139,7 @@ public:
// set the next scanned block to `ScanTask::current_block`
// set the error state to `ScanTask::status`
// set the `eos` to `ScanTask::eos` if there is no more data in current scanner
- void submit_scan_task(std::shared_ptr<ScanTask> scan_task);
+ Status submit_scan_task(std::shared_ptr<ScanTask> scan_task);
// append the running scanner and its cached block to `_blocks_queue`
void append_block_to_queue(std::shared_ptr<ScanTask> scan_task);
@@ -186,7 +186,7 @@ protected:
/// 3. `_free_blocks_memory_usage` < `_max_bytes_in_queue`, remains enough memory to scale up
/// 4. At most scale up `MAX_SCALE_UP_RATIO` times to `_max_thread_num`
void _set_scanner_done();
- void _try_to_scale_up();
+ Status _try_to_scale_up();
RuntimeState* _state = nullptr;
pipeline::ScanLocalStateBase* _local_state = nullptr;
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index e30983932ee..444ff4dbb0c 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -120,23 +120,23 @@ Status ScannerScheduler::init(ExecEnv* env) {
return Status::OK();
}
-void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
- std::shared_ptr<ScanTask> scan_task) {
+Status ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
+ std::shared_ptr<ScanTask> scan_task) {
scan_task->last_submit_time = GetCurrentTimeNanos();
if (ctx->done()) {
- return;
+ return Status::OK();
}
auto task_lock = ctx->task_exec_ctx();
if (task_lock == nullptr) {
LOG(INFO) << "could not lock task execution context, query " << ctx->debug_string()
<< " maybe finished";
- return;
+ return Status::OK();
}
if (ctx->thread_token != nullptr) {
std::shared_ptr<ScannerDelegate> scanner_delegate = scan_task->scanner.lock();
if (scanner_delegate == nullptr) {
- return;
+ return Status::OK();
}
scanner_delegate->_scanner->start_wait_worker_timer();
@@ -153,13 +153,12 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
});
if (!s.ok()) {
scan_task->set_status(s);
- ctx->append_block_to_queue(scan_task);
- return;
+ return s;
}
} else {
std::shared_ptr<ScannerDelegate> scanner_delegate = scan_task->scanner.lock();
if (scanner_delegate == nullptr) {
- return;
+ return Status::OK();
}
scanner_delegate->_scanner->start_wait_worker_timer();
@@ -187,14 +186,18 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
return scan_sched->submit_scan_task(simple_scan_task);
};
- if (auto ret = sumbit_task(); !ret) {
- scan_task->set_status(Status::InternalError(
- "Failed to submit scanner to scanner pool reason:" + std::string(ret.msg()) +
- "|type:" + std::to_string(type)));
- ctx->append_block_to_queue(scan_task);
- return;
+ Status submit_status = sumbit_task();
+ if (!submit_status.ok()) {
+ // User will see TooManyTasks error. It looks like a more reasonable error.
+ Status scan_task_status = Status::TooManyTasks(
+ "Failed to submit scanner to scanner pool reason:" +
+ std::string(submit_status.msg()) + "|type:" + std::to_string(type));
+ scan_task->set_status(scan_task_status);
+ return scan_task_status;
}
}
+
+ return Status::OK();
}
std::unique_ptr<ThreadPoolToken> ScannerScheduler::new_limited_scan_pool_token(
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h
index ddc61396e23..439291f2107 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -57,7 +57,7 @@ public:
[[nodiscard]] Status init(ExecEnv* env);
- void submit(std::shared_ptr<ScannerContext> ctx, std::shared_ptr<ScanTask> scan_task);
+ Status submit(std::shared_ptr<ScannerContext> ctx, std::shared_ptr<ScanTask> scan_task);
void stop();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/02: [opt](scanner profile) More counter for scanner (#40144)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch spill_and_reserve
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 12c061294833427b9f8c8e6e25d7128b3c4f7ad7
Author: zhiqiang <se...@163.com>
AuthorDate: Fri Sep 6 17:49:24 2024 +0800
[opt](scanner profile) More counter for scanner (#40144)
New profile metrics to monitor the schedule process of scanner.
```
VScanner:
...
- PeakMemoryUsage: 16.31 MB
- PeakRunningScanner: 1
...
```
In general, the value of `PeakMemoryUsage` is increased when any of the
scan tasks gets a block, decreased when ScanOperator consumes block from
block_queue.
---
be/src/pipeline/exec/scan_operator.cpp | 8 ++
be/src/pipeline/exec/scan_operator.h | 2 +
be/src/vec/exec/scan/scanner_context.cpp | 49 +++++++---
be/src/vec/exec/scan/scanner_context.h | 13 +--
be/src/vec/exec/scan/scanner_scheduler.cpp | 24 ++++-
be/src/vec/exec/scan/vscanner.cpp | 1 +
.../suites/query_profile/scanner_profile.groovy | 104 +++++++++++++++++++++
7 files changed, 177 insertions(+), 24 deletions(-)
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
index d7af0053944..cd65d61f641 100644
--- a/be/src/pipeline/exec/scan_operator.cpp
+++ b/be/src/pipeline/exec/scan_operator.cpp
@@ -995,6 +995,9 @@ Status ScanLocalState<Derived>::_start_scanners(
_scanner_ctx = vectorized::ScannerContext::create_shared(
state(), this, p._output_tuple_desc, p.output_row_descriptor(), scanners, p.limit(),
state()->scan_queue_mem_limit(), _scan_dependency,
+ // NOTE: This will logic makes _max_thread_num of ScannerContext to be C(num of cores) * 2
+ // For a query with C/2 instance and M scan node, scan task of this query will be C/2 * M * C*2
+ // and will be C*C*N at most.
// 1. If data distribution is ignored , we use 1 instance to scan.
// 2. Else if this operator is not file scan operator, we use config::doris_scanner_thread_pool_thread_num scanners to scan.
// 3. Else, file scanner will consume much memory so we use config::doris_scanner_thread_pool_thread_num / query_parallel_instance_num scanners to scan.
@@ -1055,6 +1058,9 @@ Status ScanLocalState<Derived>::_init_profile() {
_memory_usage_counter = ADD_LABEL_COUNTER_WITH_LEVEL(_scanner_profile, "MemoryUsage", 1);
_free_blocks_memory_usage =
_scanner_profile->AddHighWaterMarkCounter("FreeBlocks", TUnit::BYTES, "MemoryUsage", 1);
+ _scanner_peak_memory_usage =
+ _scanner_profile->AddHighWaterMarkCounter("PeakMemoryUsage", TUnit::BYTES);
+
_newly_create_free_blocks_num =
ADD_COUNTER(_scanner_profile, "NewlyCreateFreeBlocksNum", TUnit::UNIT);
_scale_up_scanners_counter = ADD_COUNTER(_scanner_profile, "NumScaleUpScanners", TUnit::UNIT);
@@ -1073,6 +1079,8 @@ Status ScanLocalState<Derived>::_init_profile() {
_max_scanner_thread_num = ADD_COUNTER(_runtime_profile, "MaxScannerThreadNum", TUnit::UNIT);
+ _peak_running_scanner =
+ _scanner_profile->AddHighWaterMarkCounter("PeakRunningScanner", TUnit::UNIT);
return Status::OK();
}
diff --git a/be/src/pipeline/exec/scan_operator.h b/be/src/pipeline/exec/scan_operator.h
index 823440f52c0..f84153e596d 100644
--- a/be/src/pipeline/exec/scan_operator.h
+++ b/be/src/pipeline/exec/scan_operator.h
@@ -109,6 +109,8 @@ protected:
RuntimeProfile::Counter* _newly_create_free_blocks_num = nullptr;
// Max num of scanner thread
RuntimeProfile::Counter* _max_scanner_thread_num = nullptr;
+ RuntimeProfile::HighWaterMarkCounter* _peak_running_scanner = nullptr;
+ RuntimeProfile::HighWaterMarkCounter* _scanner_peak_memory_usage = nullptr;
// time of get block from scanner
RuntimeProfile::Counter* _scan_timer = nullptr;
RuntimeProfile::Counter* _scan_cpu_timer = nullptr;
diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp
index bab11616c77..5cc20c214c1 100644
--- a/be/src/vec/exec/scan/scanner_context.cpp
+++ b/be/src/vec/exec/scan/scanner_context.cpp
@@ -73,11 +73,23 @@ ScannerContext::ScannerContext(
limit = -1;
}
MAX_SCALE_UP_RATIO = _state->scanner_scale_up_ratio();
+ // _max_thread_num controls how many scanners of this ScanOperator can be submitted to scheduler at a time.
+ // The overall target of our system is to make full utilization of the resources.
+ // At the same time, we dont want too many tasks are queued by scheduler, that makes the query
+ // waiting too long, and existing task can not be scheduled in time.
+ // First of all, we try to make sure _max_thread_num of a ScanNode of a query on a single backend is less than
+ // config::doris_scanner_thread_pool_thread_num.
+ // For example, on a 64-core machine, the default value of config::doris_scanner_thread_pool_thread_num will be 64*2 =128.
+ // and the num_parallel_instances of this scan operator will be 64/2=32.
+ // For a query who has two scan nodes, the _max_thread_num of each scan node instance will be 128 / 32 = 4.
+ // We have 32 instances of this scan operator, so for the ScanNode, we have 4 * 32 = 128 scanner tasks can be submitted at a time.
+ // Remember that we have to ScanNode in this query, so the total number of scanner tasks can be submitted at a time is 128 * 2 = 256.
_max_thread_num =
_state->num_scanner_threads() > 0
? _state->num_scanner_threads()
: config::doris_scanner_thread_pool_thread_num / num_parallel_instances;
_max_thread_num = _max_thread_num == 0 ? 1 : _max_thread_num;
+ // In some situation, there are not too many big tablets involed, so we can reduce the thread number.
_max_thread_num = std::min(_max_thread_num, (int32_t)scanners.size());
// 1. Calculate max concurrency
// For select * from table limit 10; should just use one thread.
@@ -116,7 +128,6 @@ Status ScannerContext::init() {
_scanner_sched_counter = _local_state->_scanner_sched_counter;
_newly_create_free_blocks_num = _local_state->_newly_create_free_blocks_num;
_scanner_wait_batch_timer = _local_state->_scanner_wait_batch_timer;
- _free_blocks_memory_usage_mark = _local_state->_free_blocks_memory_usage;
_scanner_ctx_sched_time = _local_state->_scanner_ctx_sched_time;
_scale_up_scanners_counter = _local_state->_scale_up_scanners_counter;
@@ -157,9 +168,11 @@ vectorized::BlockUPtr ScannerContext::get_free_block(bool force) {
vectorized::BlockUPtr block = nullptr;
if (_free_blocks.try_dequeue(block)) {
DCHECK(block->mem_reuse());
- _free_blocks_memory_usage -= block->allocated_bytes();
- _free_blocks_memory_usage_mark->set(_free_blocks_memory_usage);
- } else if (_free_blocks_memory_usage < _max_bytes_in_queue || force) {
+ _block_memory_usage -= block->allocated_bytes();
+ // A free block is reused, so the memory usage should be decreased
+ // The caller of get_free_block will increase the memory usage
+ update_peak_memory_usage(-block->allocated_bytes());
+ } else if (_block_memory_usage < _max_bytes_in_queue || force) {
_newly_create_free_blocks_num->update(1);
block = vectorized::Block::create_unique(_output_tuple_desc->slots(), 0,
true /*ignore invalid slots*/);
@@ -168,11 +181,13 @@ vectorized::BlockUPtr ScannerContext::get_free_block(bool force) {
}
void ScannerContext::return_free_block(vectorized::BlockUPtr block) {
- if (block->mem_reuse() && _free_blocks_memory_usage < _max_bytes_in_queue) {
- _free_blocks_memory_usage += block->allocated_bytes();
- _free_blocks_memory_usage_mark->set(_free_blocks_memory_usage);
+ if (block->mem_reuse() && _block_memory_usage < _max_bytes_in_queue) {
+ size_t block_size_to_reuse = block->allocated_bytes();
+ _block_memory_usage += block_size_to_reuse;
block->clear_column_data();
- _free_blocks.enqueue(std::move(block));
+ if (_free_blocks.enqueue(std::move(block))) {
+ update_peak_memory_usage(block_size_to_reuse);
+ }
}
}
@@ -242,8 +257,8 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
if (_estimated_block_size > block_size) {
_estimated_block_size = block_size;
}
- _free_blocks_memory_usage -= block_size;
- _free_blocks_memory_usage_mark->set(_free_blocks_memory_usage);
+ _block_memory_usage -= block_size;
+ update_peak_memory_usage(-current_block->allocated_bytes());
// consume current block
block->swap(*current_block);
return_free_block(std::move(current_block));
@@ -263,8 +278,7 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
for (int i = 0; i < free_blocks_for_each; ++i) {
vectorized::BlockUPtr removed_block;
if (_free_blocks.try_dequeue(removed_block)) {
- _free_blocks_memory_usage -= block->allocated_bytes();
- _free_blocks_memory_usage_mark->set(_free_blocks_memory_usage);
+ _block_memory_usage -= block->allocated_bytes();
}
}
}
@@ -314,8 +328,7 @@ void ScannerContext::_try_to_scale_up() {
int num_add = int(std::min(_num_running_scanners * SCALE_UP_RATIO,
_max_thread_num * MAX_SCALE_UP_RATIO - _num_running_scanners));
if (_estimated_block_size > 0) {
- int most_add =
- (_max_bytes_in_queue - _free_blocks_memory_usage) / _estimated_block_size;
+ int most_add = (_max_bytes_in_queue - _block_memory_usage) / _estimated_block_size;
num_add = std::min(num_add, most_add);
}
for (int i = 0; i < num_add; ++i) {
@@ -445,4 +458,12 @@ void ScannerContext::_set_scanner_done() {
_dependency->set_always_ready();
}
+void ScannerContext::update_peak_running_scanner(int num) {
+ _local_state->_peak_running_scanner->add(num);
+}
+
+void ScannerContext::update_peak_memory_usage(int64_t usage) {
+ _local_state->_scanner_peak_memory_usage->add(usage);
+}
+
} // namespace doris::vectorized
diff --git a/be/src/vec/exec/scan/scanner_context.h b/be/src/vec/exec/scan/scanner_context.h
index d97fc731fe5..f93d01eef88 100644
--- a/be/src/vec/exec/scan/scanner_context.h
+++ b/be/src/vec/exec/scan/scanner_context.h
@@ -122,10 +122,12 @@ public:
vectorized::BlockUPtr get_free_block(bool force);
void return_free_block(vectorized::BlockUPtr block);
- inline void inc_free_block_usage(size_t usage) {
- _free_blocks_memory_usage += usage;
- _free_blocks_memory_usage_mark->set(_free_blocks_memory_usage);
- }
+ inline void inc_block_usage(size_t usage) { _block_memory_usage += usage; }
+
+ // Caller should make sure the pipeline task is still running when calling this function
+ void update_peak_running_scanner(int num);
+ // Caller should make sure the pipeline task is still running when calling this function
+ void update_peak_memory_usage(int64_t usage);
// Get next block from blocks queue. Called by ScanNode/ScanOperator
// Set eos to true if there is no more data to read.
@@ -223,7 +225,6 @@ protected:
RuntimeProfile::Counter* _scanner_sched_counter = nullptr;
RuntimeProfile::Counter* _newly_create_free_blocks_num = nullptr;
RuntimeProfile::Counter* _scanner_wait_batch_timer = nullptr;
- RuntimeProfile::HighWaterMarkCounter* _free_blocks_memory_usage_mark = nullptr;
RuntimeProfile::Counter* _scanner_ctx_sched_time = nullptr;
RuntimeProfile::Counter* _scale_up_scanners_counter = nullptr;
QueryThreadContext _query_thread_context;
@@ -231,7 +232,7 @@ protected:
// for scaling up the running scanners
size_t _estimated_block_size = 0;
- std::atomic_long _free_blocks_memory_usage = 0;
+ std::atomic_long _block_memory_usage = 0;
int64_t _last_scale_up_time = 0;
int64_t _last_fetch_time = 0;
int64_t _total_wait_block_time = 0;
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index 351f5d4e275..e30983932ee 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -34,6 +34,7 @@
#include "common/status.h"
#include "olap/tablet.h"
#include "runtime/exec_env.h"
+#include "runtime/memory/mem_tracker.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "util/async_io.h" // IWYU pragma: keep
@@ -210,6 +211,9 @@ void ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
return;
}
+ ctx->update_peak_running_scanner(1);
+ Defer defer([&] { ctx->update_peak_running_scanner(-1); });
+
std::shared_ptr<ScannerDelegate> scanner_delegate = scan_task->scanner.lock();
if (scanner_delegate == nullptr) {
return;
@@ -267,13 +271,18 @@ void ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
if (free_block == nullptr) {
break;
}
+ // We got a new created block or a reused block.
+ ctx->update_peak_memory_usage(free_block->allocated_bytes());
+ ctx->update_peak_memory_usage(-free_block->allocated_bytes());
status = scanner->get_block_after_projects(state, free_block.get(), &eos);
+ // Projection will truncate useless columns, makes block size change.
+ auto free_block_bytes = free_block->allocated_bytes();
+ ctx->update_peak_memory_usage(free_block_bytes);
first_read = false;
if (!status.ok()) {
LOG(WARNING) << "Scan thread read VScanner failed: " << status.to_string();
break;
}
- auto free_block_bytes = free_block->allocated_bytes();
raw_bytes_read += free_block_bytes;
if (!scan_task->cached_blocks.empty() &&
scan_task->cached_blocks.back().first->rows() + free_block->rows() <=
@@ -281,18 +290,25 @@ void ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
size_t block_size = scan_task->cached_blocks.back().first->allocated_bytes();
vectorized::MutableBlock mutable_block(
scan_task->cached_blocks.back().first.get());
+ ctx->update_peak_memory_usage(-mutable_block.allocated_bytes());
status = mutable_block.merge(*free_block);
+ ctx->update_peak_memory_usage(mutable_block.allocated_bytes());
if (!status.ok()) {
LOG(WARNING) << "Block merge failed: " << status.to_string();
break;
}
+ scan_task->cached_blocks.back().second = mutable_block.allocated_bytes();
scan_task->cached_blocks.back().first.get()->set_columns(
std::move(mutable_block.mutable_columns()));
+
+ // Return block succeed or not, this free_block is not used by this scan task any more.
+ ctx->update_peak_memory_usage(-free_block_bytes);
+ // If block can be reused, its memory usage will be added back.
ctx->return_free_block(std::move(free_block));
- ctx->inc_free_block_usage(
- scan_task->cached_blocks.back().first->allocated_bytes() - block_size);
+ ctx->inc_block_usage(scan_task->cached_blocks.back().first->allocated_bytes() -
+ block_size);
} else {
- ctx->inc_free_block_usage(free_block->allocated_bytes());
+ ctx->inc_block_usage(free_block->allocated_bytes());
scan_task->cached_blocks.emplace_back(std::move(free_block), free_block_bytes);
}
} // end for while
diff --git a/be/src/vec/exec/scan/vscanner.cpp b/be/src/vec/exec/scan/vscanner.cpp
index 43d791caffa..a78f8956025 100644
--- a/be/src/vec/exec/scan/vscanner.cpp
+++ b/be/src/vec/exec/scan/vscanner.cpp
@@ -143,6 +143,7 @@ Status VScanner::get_block(RuntimeState* state, Block* block, bool* eof) {
}
if (state->is_cancelled()) {
+ // TODO: Should return the specific ErrorStatus instead of just Cancelled.
return Status::Cancelled("cancelled");
}
*eof = *eof || _should_stop;
diff --git a/regression-test/suites/query_profile/scanner_profile.groovy b/regression-test/suites/query_profile/scanner_profile.groovy
new file mode 100644
index 00000000000..38216d211e6
--- /dev/null
+++ b/regression-test/suites/query_profile/scanner_profile.groovy
@@ -0,0 +1,104 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import groovy.json.StringEscapeUtils
+
+
+def getProfileList = {
+ def dst = 'http://' + context.config.feHttpAddress
+ def conn = new URL(dst + "/rest/v1/query_profile").openConnection()
+ conn.setRequestMethod("GET")
+ def encoding = Base64.getEncoder().encodeToString((context.config.feHttpUser + ":" +
+ (context.config.feHttpPassword == null ? "" : context.config.feHttpPassword)).getBytes("UTF-8"))
+ conn.setRequestProperty("Authorization", "Basic ${encoding}")
+ return conn.getInputStream().getText()
+}
+
+
+def getProfile = { id ->
+ def dst = 'http://' + context.config.feHttpAddress
+ def conn = new URL(dst + "/api/profile/text/?query_id=$id").openConnection()
+ conn.setRequestMethod("GET")
+ def encoding = Base64.getEncoder().encodeToString((context.config.feHttpUser + ":" +
+ (context.config.feHttpPassword == null ? "" : context.config.feHttpPassword)).getBytes("UTF-8"))
+ conn.setRequestProperty("Authorization", "Basic ${encoding}")
+ return conn.getInputStream().getText()
+}
+
+suite('scanner_profile') {
+ sql """
+ DROP TABLE IF EXISTS scanner_profile;
+ """
+ sql """
+ CREATE TABLE if not exists `scanner_profile` (
+ `id` INT,
+ `name` varchar(32)
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`id`) BUCKETS 10
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ // Insert data to table
+ sql """
+ insert into scanner_profile values
+ (1, "A"),(2, "B"),(3, "C"),(4, "D"),(5,"E"),(6,"F"),(7,"G"),(8,"H"),(9,"K");
+ """
+ sql """
+ insert into scanner_profile values
+ (10, "A"),(20, "B"),(30, "C"),(40, "D"),(50,"E"),(60,"F"),(70,"G"),(80,"H"),(90,"K");
+ """
+ sql """
+ insert into scanner_profile values
+ (101, "A"),(201, "B"),(301, "C"),(401, "D"),(501,"E"),(601,"F"),(701,"G"),(801,"H"),(901,"K");
+ """
+ sql """
+ insert into scanner_profile values
+ (1010, "A"),(2010, "B"),(3010, "C"),(4010, "D"),(5010,"E"),(6010,"F"),(7010,"G"),(8010,"H"),(9010,"K");
+ """
+
+ def uuidString = UUID.randomUUID().toString()
+ sql "set enable_profile=true"
+ // With Limit, MaxScannerThreadNum = 1
+ sql """
+ select "with_limit_1_${uuidString}", * from scanner_profile limit 10;
+ """
+
+ def wholeString = getProfileList()
+ List profileData = new JsonSlurper().parseText(wholeString).data.rows
+ String queryIdWithLimit1 = "";
+
+
+ logger.info("{}", uuidString)
+
+ for (def profileItem in profileData) {
+ if (profileItem["Sql Statement"].toString().contains("with_limit_1_${uuidString}")) {
+ queryIdWithLimit1 = profileItem["Profile ID"].toString()
+ logger.info("profileItem: {}", profileItem)
+ }
+ }
+
+ logger.info("queryIdWithLimit1_${uuidString}: {}", queryIdWithLimit1)
+
+ assertTrue(queryIdWithLimit1 != "")
+ def String profileWithLimit1 = getProfile(queryIdWithLimit1).toString()
+ logger.info("query profile {}", profileWithLimit1)
+ assertTrue(profileWithLimit1.contains("- PeakRunningScanner: 1"))
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org