You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "Mryange (via GitHub)" <gi...@apache.org> on 2023/11/02 03:40:45 UTC
[PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Mryange opened a new pull request, #26281:
URL: https://github.com/apache/doris/pull/26281
## Proposed changes
Issue Number: close #xxx
<!--Describe your changes.-->
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791728200
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 46.48 seconds
stream load tsv: 557 seconds loaded 74807831229 Bytes, about 128 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 29.1 seconds inserted 10000000 Rows, about 343K ops/s
storage size: 17162061619 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791685508
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1790025598
TeamCity be ut coverage result:
Function Coverage: 37.37% (8481/22694)
Line Coverage: 29.72% (68592/230810)
Region Coverage: 28.37% (35490/125083)
Branch Coverage: 25.11% (18114/72152)
Coverage Report: http://coverage.selectdb-in.cc/coverage/b0e8df48b0adb7e5d834a55a1ae97883bbe042c7_b0e8df48b0adb7e5d834a55a1ae97883bbe042c7/report/index.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1790046834
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 46.68 seconds
stream load tsv: 573 seconds loaded 74807831229 Bytes, about 124 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 64 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 29.2 seconds inserted 10000000 Rows, about 342K ops/s
storage size: 17162641041 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791794864
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 45.96 seconds
stream load tsv: 555 seconds loaded 74807831229 Bytes, about 128 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 29.0 seconds inserted 10000000 Rows, about 344K ops/s
storage size: 17162092491 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26281:
URL: https://github.com/apache/doris/pull/26281#discussion_r1379559779
##########
be/src/pipeline/pipeline_x/dependency.cpp:
##########
@@ -294,6 +317,18 @@
return results;
}
+void SetSharedState::set_probe_finished_children(int child_id) {
Review Comment:
warning: method 'set_probe_finished_children' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_x/dependency.h:784:
```diff
- void set_probe_finished_children(int child_id);
+ static void set_probe_finished_children(int child_id);
```
##########
be/src/pipeline/pipeline_x/dependency.cpp:
##########
@@ -21,10 +21,33 @@
#include <mutex>
#include "common/logging.h"
+#include "pipeline/pipeline_task.h"
+#include "pipeline/pipeline_x/pipeline_x_task.h"
#include "runtime/memory/mem_tracker.h"
namespace doris::pipeline {
+void Dependency::add_block_task(PipelineXTask* task) {
+ std::unique_lock<std::mutex> lc(_task_lock);
+ DCHECK(task->get_state() != PipelineTaskState::RUNNABLE);
+ DCHECK(avoid_using_blocked_queue(task->get_state()));
+ _block_task.push_back(task);
+}
+
+void Dependency::try_to_wake_up_task() {
Review Comment:
warning: method 'try_to_wake_up_task' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_x/dependency.h:106:
```diff
- void try_to_wake_up_task();
+ static void try_to_wake_up_task();
```
##########
be/src/pipeline/pipeline_x/pipeline_x_task.h:
##########
@@ -70,54 +70,76 @@ class PipelineXTask : public PipelineTask {
// must be call after all pipeline task is finish to release resource
Status close(Status exec_status) override;
- bool source_can_read() override {
- if (_dry_run) {
- return true;
- }
+ Dependency* read_blocked_dependency() {
Review Comment:
warning: method 'read_blocked_dependency' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Dependency* read_blocked_dependency() {
```
##########
be/src/pipeline/pipeline_x/dependency.cpp:
##########
@@ -21,10 +21,33 @@
#include <mutex>
#include "common/logging.h"
+#include "pipeline/pipeline_task.h"
+#include "pipeline/pipeline_x/pipeline_x_task.h"
#include "runtime/memory/mem_tracker.h"
namespace doris::pipeline {
+void Dependency::add_block_task(PipelineXTask* task) {
Review Comment:
warning: method 'add_block_task' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_x/dependency.h:104:
```diff
- void add_block_task(PipelineXTask* task);
+ static void add_block_task(PipelineXTask* task);
```
##########
be/src/pipeline/pipeline_x/pipeline_x_task.h:
##########
@@ -70,54 +70,76 @@
// must be call after all pipeline task is finish to release resource
Status close(Status exec_status) override;
- bool source_can_read() override {
- if (_dry_run) {
- return true;
- }
+ Dependency* read_blocked_dependency() {
for (auto* op_dep : _read_dependencies) {
auto* dep = op_dep->read_blocked_by();
if (dep != nullptr) {
dep->start_read_watcher();
- push_blocked_task_to_dependency(dep);
- return false;
+ set_state(PipelineTaskState::BLOCKED_FOR_SOURCE);
+ return dep;
}
}
- return true;
+ return nullptr;
}
- bool runtime_filters_are_ready_or_timeout() override {
+ bool source_can_read() override {
+ if (_dry_run) {
+ return true;
+ }
+ auto* dep = read_blocked_dependency();
+ return dep == nullptr;
+ }
+
+ RuntimeFilterDependency* filter_blocked_dependency() {
auto* dep = _filter_dependency->filter_blocked_by();
if (dep != nullptr) {
- push_blocked_task_to_dependency(dep);
- return false;
+ set_state(PipelineTaskState::BLOCKED_FOR_RF);
+ dep->add_block_task(this);
+ return dep;
}
- return true;
+ return nullptr;
}
- bool sink_can_write() override {
+ bool runtime_filters_are_ready_or_timeout() override {
+ auto* dep = filter_blocked_dependency();
+ return dep == nullptr;
+ }
+
+ WriteDependency* write_blocked_dependency() {
auto* dep = _write_dependencies->write_blocked_by();
if (dep != nullptr) {
dep->start_write_watcher();
- push_blocked_task_to_dependency(dep);
- return false;
+ set_state(PipelineTaskState::BLOCKED_FOR_SINK);
+ dep->add_block_task(this);
+ return dep;
}
- return true;
+ return nullptr;
+ }
+
+ bool sink_can_write() override {
+ auto* dep = write_blocked_dependency();
+ return dep == nullptr;
}
Status finalize() override;
std::string debug_string() override;
- bool is_pending_finish() override {
+ FinishDependency* finish_blocked_dependency() {
Review Comment:
warning: method 'finish_blocked_dependency' can be made static [readability-convert-member-functions-to-static]
```suggestion
static FinishDependency* finish_blocked_dependency() {
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange closed pull request #26281: [feature](pipelineX) avoid using blocked queue in sink/filter
URL: https://github.com/apache/doris/pull/26281
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1790014438
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791711355
TeamCity be ut coverage result:
Function Coverage: 37.28% (8452/22674)
Line Coverage: 29.66% (68440/230740)
Region Coverage: 28.32% (35409/125021)
Branch Coverage: 25.06% (18073/72112)
Coverage Report: http://coverage.selectdb-in.cc/coverage/b0e8df48b0adb7e5d834a55a1ae97883bbe042c7_b0e8df48b0adb7e5d834a55a1ae97883bbe042c7/report/index.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "Mryange (via GitHub)" <gi...@apache.org>.
Mryange commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791768449
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature](pipelineX) avoid using blocked queue in sink/filter [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26281:
URL: https://github.com/apache/doris/pull/26281#issuecomment-1791786921
TeamCity be ut coverage result:
Function Coverage: 37.27% (8451/22674)
Line Coverage: 29.65% (68421/230740)
Region Coverage: 28.32% (35405/125021)
Branch Coverage: 25.05% (18065/72112)
Coverage Report: http://coverage.selectdb-in.cc/coverage/325c5cd3ef6563b1eefed470fd1064377ba756b0_325c5cd3ef6563b1eefed470fd1064377ba756b0/report/index.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org