Skip to content

Commit 4066de3

Browse files
yiguoleiyiguolei
authored
[bugfix](scannercore) scanner will core in deconstructor during collect profile (#28727)
1 parent 3d2b4ae commit 4066de3

11 files changed

Lines changed: 213 additions & 336 deletions

‎be/src/exec/exec_node.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,10 @@ class ExecNode {
239239

240240
size_t children_count() const { return _children.size(); }
241241

242+
// when the fragment is normal finished, call this method to do some finish work
243+
// such as send the last buffer to remote.
244+
virtual Status try_close(RuntimeState* state) { return Status::OK(); }
245+
242246
protected:
243247
friend class DataSink;
244248

‎be/src/pipeline/exec/scan_operator.cpp‎

Lines changed: 19 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -64,14 +64,6 @@ bool ScanOperator::can_read() {
6464
}
6565
}
6666

67-
bool ScanOperator::is_pending_finish() const {
68-
return _node->_scanner_ctx && !_node->_scanner_ctx->no_schedule();
69-
}
70-
71-
Status ScanOperator::try_close(RuntimeState* state) {
72-
return _node->try_close(state);
73-
}
74-
7567
bool ScanOperator::runtime_filters_are_ready_or_timeout() {
7668
return _node->runtime_filters_are_ready_or_timeout();
7769
}
@@ -81,9 +73,8 @@ std::string ScanOperator::debug_string() const {
8173
fmt::format_to(debug_string_buffer, "{}, scanner_ctx is null: {} ",
8274
SourceOperator::debug_string(), _node->_scanner_ctx == nullptr);
8375
if (_node->_scanner_ctx) {
84-
fmt::format_to(debug_string_buffer, ", num_running_scanners = {}, num_scheduling_ctx = {} ",
85-
_node->_scanner_ctx->get_num_running_scanners(),
86-
_node->_scanner_ctx->get_num_scheduling_ctx());
76+
fmt::format_to(debug_string_buffer, ", num_running_scanners = {}",
77+
_node->_scanner_ctx->get_num_running_scanners());
8778
}
8879
return fmt::to_string(debug_string_buffer);
8980
}
@@ -101,9 +92,6 @@ std::string ScanOperator::debug_string() const {
10192
template <typename Derived>
10293
ScanLocalState<Derived>::ScanLocalState(RuntimeState* state, OperatorXBase* parent)
10394
: ScanLocalStateBase(state, parent) {
104-
_finish_dependency = std::make_shared<FinishDependency>(
105-
parent->operator_id(), parent->node_id(), parent->get_name() + "_FINISH_DEPENDENCY",
106-
state->get_query_ctx());
10795
_filter_dependency = std::make_shared<RuntimeFilterDependency>(
10896
parent->operator_id(), parent->node_id(), parent->get_name() + "_FILTER_DEPENDENCY",
10997
state->get_query_ctx());
@@ -177,7 +165,6 @@ Status ScanLocalState<Derived>::open(RuntimeState* state) {
177165

178166
auto status = _eos ? Status::OK() : _prepare_scanners();
179167
if (_scanner_ctx) {
180-
_finish_dependency->block();
181168
DCHECK(!_eos && _num_scanners->value() > 0);
182169
RETURN_IF_ERROR(_scanner_ctx->init());
183170
RETURN_IF_ERROR(state->exec_env()->scanner_scheduler()->submit(_scanner_ctx));
@@ -570,15 +557,14 @@ std::string ScanLocalState<Derived>::debug_string(int indentation_level) const {
570557
PipelineXLocalState<>::debug_string(indentation_level), _eos.load());
571558
if (_scanner_ctx) {
572559
fmt::format_to(debug_string_buffer, "");
573-
fmt::format_to(
574-
debug_string_buffer,
575-
", Scanner Context: (_is_finished = {}, _should_stop = {}, "
576-
"_num_running_scanners={}, "
577-
"_num_scheduling_ctx = {}, _num_unfinished_scanners = {}, status = {}, error = {})",
578-
_scanner_ctx->is_finished(), _scanner_ctx->should_stop(),
579-
_scanner_ctx->get_num_running_scanners(), _scanner_ctx->get_num_scheduling_ctx(),
580-
_scanner_ctx->get_num_unfinished_scanners(), _scanner_ctx->status().to_string(),
581-
_scanner_ctx->status_error());
560+
fmt::format_to(debug_string_buffer,
561+
", Scanner Context: (_is_finished = {}, _should_stop = {}, "
562+
"_num_running_scanners={}, "
563+
" _num_unfinished_scanners = {}, status = {}, error = {})",
564+
_scanner_ctx->is_finished(), _scanner_ctx->should_stop(),
565+
_scanner_ctx->get_num_running_scanners(),
566+
_scanner_ctx->get_num_unfinished_scanners(),
567+
_scanner_ctx->status().to_string(), _scanner_ctx->status_error());
582568
}
583569

584570
return fmt::to_string(debug_string_buffer);
@@ -1226,24 +1212,27 @@ template <typename Derived>
12261212
Status ScanLocalState<Derived>::_prepare_scanners() {
12271213
std::list<vectorized::VScannerSPtr> scanners;
12281214
RETURN_IF_ERROR(_init_scanners(&scanners));
1215+
// Init scanner wrapper
1216+
for (auto it = scanners.begin(); it != scanners.end(); ++it) {
1217+
_scanners.emplace_back(std::make_shared<vectorized::ScannerDelegate>(*it));
1218+
}
12291219
if (scanners.empty()) {
12301220
_eos = true;
12311221
_scan_dependency->set_ready();
12321222
} else {
12331223
COUNTER_SET(_num_scanners, static_cast<int64_t>(scanners.size()));
1234-
RETURN_IF_ERROR(_start_scanners(scanners));
1224+
RETURN_IF_ERROR(_start_scanners(_scanners));
12351225
}
12361226
return Status::OK();
12371227
}
12381228

12391229
template <typename Derived>
12401230
Status ScanLocalState<Derived>::_start_scanners(
1241-
const std::list<vectorized::VScannerSPtr>& scanners) {
1231+
const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners) {
12421232
auto& p = _parent->cast<typename Derived::Parent>();
12431233
_scanner_ctx = PipScannerContext::create_shared(state(), this, p._output_tuple_desc, scanners,
12441234
p.limit(), state()->scan_queue_mem_limit(),
1245-
p._col_distribute_ids, 1, _scan_dependency,
1246-
_finish_dependency);
1235+
p._col_distribute_ids, 1, _scan_dependency);
12471236
return Status::OK();
12481237
}
12491238

@@ -1319,9 +1308,6 @@ Status ScanLocalState<Derived>::_init_profile() {
13191308

13201309
_max_scanner_thread_num = ADD_COUNTER(_runtime_profile, "MaxScannerThreadNum", TUnit::UNIT);
13211310

1322-
_wait_for_finish_dependency_timer =
1323-
ADD_TIMER(_runtime_profile, "WaitForPendingFinishDependency");
1324-
13251311
return Status::OK();
13261312
}
13271313

@@ -1429,17 +1415,6 @@ Status ScanOperatorX<LocalStateType>::open(RuntimeState* state) {
14291415
return Status::OK();
14301416
}
14311417

1432-
template <typename LocalStateType>
1433-
Status ScanOperatorX<LocalStateType>::try_close(RuntimeState* state) {
1434-
auto& local_state = get_local_state(state);
1435-
if (local_state._scanner_ctx) {
1436-
// mark this scanner ctx as should_stop to make sure scanners will not be scheduled anymore
1437-
// TODO: there is a lock in `set_should_stop` may cause some slight impact
1438-
local_state._scanner_ctx->set_should_stop();
1439-
}
1440-
return Status::OK();
1441-
}
1442-
14431418
template <typename Derived>
14441419
Status ScanLocalState<Derived>::close(RuntimeState* state) {
14451420
if (_closed) {
@@ -1451,10 +1426,9 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {
14511426

14521427
SCOPED_TIMER(exec_time_counter());
14531428
if (_scanner_ctx) {
1454-
_scanner_ctx->clear_and_join(reinterpret_cast<ScanLocalStateBase*>(this), state);
1429+
_scanner_ctx->stop_scanners(state);
14551430
}
14561431
COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());
1457-
COUNTER_SET(_wait_for_finish_dependency_timer, _finish_dependency->watcher_elapse_time());
14581432
COUNTER_SET(_wait_for_rf_timer, _filter_dependency->watcher_elapse_time());
14591433

14601434
return PipelineXLocalState<>::close(state);
@@ -1511,7 +1485,7 @@ Status ScanOperatorX<LocalStateType>::get_block(RuntimeState* state, vectorized:
15111485
if (eos) {
15121486
source_state = SourceState::FINISHED;
15131487
// reach limit, stop the scanners.
1514-
local_state._scanner_ctx->set_should_stop();
1488+
local_state._scanner_ctx->stop_scanners(state);
15151489
}
15161490

15171491
return Status::OK();

‎be/src/pipeline/exec/scan_operator.h‎

Lines changed: 6 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@
3131
namespace doris {
3232
class ExecNode;
3333
} // namespace doris
34+
namespace doris::vectorized {
35+
class ScannerDelegate;
36+
}
3437

3538
namespace doris::pipeline {
3639
class PipScannerContext;
@@ -48,13 +51,9 @@ class ScanOperator : public SourceOperator<ScanOperatorBuilder> {
4851

4952
bool can_read() override; // for source
5053

51-
bool is_pending_finish() const override;
52-
5354
bool runtime_filters_are_ready_or_timeout() override;
5455

5556
std::string debug_string() const override;
56-
57-
Status try_close(RuntimeState* state) override;
5857
};
5958

6059
class ScanDependency final : public Dependency {
@@ -171,7 +170,6 @@ class ScanLocalStateBase : public PipelineXLocalState<>, public vectorized::Runt
171170
RuntimeProfile::Counter* _wait_for_scanner_done_timer = nullptr;
172171
// time of prefilter input block from scanner
173172
RuntimeProfile::Counter* _wait_for_eos_timer = nullptr;
174-
RuntimeProfile::Counter* _wait_for_finish_dependency_timer = nullptr;
175173
RuntimeProfile::Counter* _wait_for_rf_timer = nullptr;
176174
};
177175

@@ -214,7 +212,6 @@ class ScanLocalState : public ScanLocalStateBase {
214212
Dependency* dependency() override { return _scan_dependency.get(); }
215213

216214
RuntimeFilterDependency* filterdependency() override { return _filter_dependency.get(); };
217-
Dependency* finishdependency() override { return _finish_dependency.get(); }
218215

219216
protected:
220217
template <typename LocalStateType>
@@ -350,7 +347,7 @@ class ScanLocalState : public ScanLocalStateBase {
350347
Status _prepare_scanners();
351348

352349
// Submit the scanner to the thread pool and start execution
353-
Status _start_scanners(const std::list<vectorized::VScannerSPtr>& scanners);
350+
Status _start_scanners(const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners);
354351

355352
// For some conjunct there is chance to elimate cast operator
356353
// Eg. Variant's sub column could eliminate cast in storage layer if
@@ -413,14 +410,13 @@ class ScanLocalState : public ScanLocalStateBase {
413410

414411
std::shared_ptr<RuntimeFilterDependency> _filter_dependency;
415412

416-
std::shared_ptr<Dependency> _finish_dependency;
413+
// ScanLocalState owns the ownership of scanner, scanner context only has its weakptr
414+
std::list<std::shared_ptr<vectorized::ScannerDelegate>> _scanners;
417415
};
418416

419417
template <typename LocalStateType>
420418
class ScanOperatorX : public OperatorX<LocalStateType> {
421419
public:
422-
Status try_close(RuntimeState* state) override;
423-
424420
Status init(const TPlanNode& tnode, RuntimeState* state) override;
425421
Status prepare(RuntimeState* state) override { return OperatorXBase::prepare(state); }
426422
Status open(RuntimeState* state) override;

‎be/src/vec/exec/scan/pip_scanner_context.h‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -31,24 +31,23 @@ class PipScannerContext : public vectorized::ScannerContext {
3131
public:
3232
PipScannerContext(RuntimeState* state, vectorized::VScanNode* parent,
3333
const TupleDescriptor* output_tuple_desc,
34-
const std::list<vectorized::VScannerSPtr>& scanners, int64_t limit_,
35-
int64_t max_bytes_in_blocks_queue, const std::vector<int>& col_distribute_ids,
36-
const int num_parallel_instances)
34+
const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners,
35+
int64_t limit_, int64_t max_bytes_in_blocks_queue,
36+
const std::vector<int>& col_distribute_ids, const int num_parallel_instances)
3737
: vectorized::ScannerContext(state, parent, output_tuple_desc, scanners, limit_,
3838
max_bytes_in_blocks_queue, num_parallel_instances),
3939
_col_distribute_ids(col_distribute_ids),
4040
_need_colocate_distribute(!_col_distribute_ids.empty()) {}
4141

4242
PipScannerContext(RuntimeState* state, ScanLocalStateBase* local_state,
4343
const TupleDescriptor* output_tuple_desc,
44-
const std::list<vectorized::VScannerSPtr>& scanners, int64_t limit_,
45-
int64_t max_bytes_in_blocks_queue, const std::vector<int>& col_distribute_ids,
46-
const int num_parallel_instances,
47-
std::shared_ptr<pipeline::ScanDependency> dependency,
48-
std::shared_ptr<pipeline::Dependency> finish_dependency)
44+
const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners,
45+
int64_t limit_, int64_t max_bytes_in_blocks_queue,
46+
const std::vector<int>& col_distribute_ids, const int num_parallel_instances,
47+
std::shared_ptr<pipeline::ScanDependency> dependency)
4948
: vectorized::ScannerContext(state, output_tuple_desc, scanners, limit_,
5049
max_bytes_in_blocks_queue, num_parallel_instances,
51-
local_state, dependency, finish_dependency),
50+
local_state, dependency),
5251
_need_colocate_distribute(false) {}
5352

5453
Status get_block_from_queue(RuntimeState* state, vectorized::BlockUPtr* block, bool* eos,
@@ -111,9 +110,6 @@ class PipScannerContext : public vectorized::ScannerContext {
111110
return Status::OK();
112111
}
113112

114-
// We should make those method lock free.
115-
bool done() override { return _is_finished || _should_stop; }
116-
117113
void append_blocks_to_queue(std::vector<vectorized::BlockUPtr>& blocks) override {
118114
const int queue_size = _blocks_queues.size();
119115
const int block_size = blocks.size();

0 commit comments

Comments
 (0)