You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2020/03/19 03:58:13 UTC
[GitHub] [incubator-doris] vagetablechicken opened a new pull request #3143:
Non blocking OlapTableSink
vagetablechicken opened a new pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143
## ImplementaItion Notes
### NodeChannel
1. _cur_batch -> _pending_batches: when _cur_batch is filled up, move it to _pending_batches.
2. add_row() just produce batches.
3. try_send_and_fetch_status() tries to consume one pending batch. If has in flight packet, skip send in this round.
So we can add one sender thread to be in charge of all node channels try_send.
### IndexChannel
1. init(), open() stay the same.
2. Use for_each_node_channel() to expose the detailed changes of NodeChannel.(It's more easy to read & modify)
### Sender thread
See func OlapTableSink::_send_batch_process()
#### Why use polling？
If we use wait/notify, it will notify when generate a new batch. We can't skip sending this batch, coz it won't notify the same batch again. So wait/notify can't avoid blocking simply.
So I choose polling.
It's wasting to continuously try_send(), but it's difficult to set the suitable polling interval. Thus, I add std::this_thread::yield() to give up the time slice, give priority to other process/threads (if there are other process/threads waiting in the queue).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395963965
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +129,128 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, load_id=" << _parent->_load_id
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ LOG(INFO) << name() << " last rpc has responsed";
+ }
+ } else {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status not ok, load_id="
+ << _parent->_load_id << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_rpc_error, _is_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. rpc_error/cancelled/eos: ");
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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);
Review comment:
_plan node in PlanFragmentExec may not limit the mem, so I add mem limit check in NodeChannel::add_row, it'll block the main thread if mem limit exceeded.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken edited a comment on issue #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken edited a comment on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-611406948
@imay @morningman
Sorry about my change is a bit late.
I've changed yeild() to sleep(). And the default interval time is 10ms.
Why 10ms, it's calculated through the avg time of add batch RPCs.
If the case is large, it get faster obviously, so I focus on little cases this time.
I tested two little cases, and got the logs about add batch time. e.g.
`node add batch time(ms)/wait lock time(ms)/num: {10003:(76703)(0)(4336)} {10002:(87961)(0)(4921)} {10006:(80490)(0)(4570)} {10005:(71098)(0)(3984)} {10004:(87736)(0)(4686)}`
So `avg time` = `add batch time` / `num`. And if you think the network time delay is not negligible, you can add it.
In my tests, the avg times are tens of milliseconds. So for the faster sink, I set it to 10ms.
And the test result(interval=10ms) is
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":28582315,"TaskNumber":1,"FileNumber":3,"FileSize":569042681} | 4min37s | 3min27s |
|{"ScannedRows":100967318,"TaskNumber":1,"FileNumber":9,"FileSize":3096583144} | 12min27s | 8min27s |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395956321
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +129,128 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, load_id=" << _parent->_load_id
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ LOG(INFO) << name() << " last rpc has responsed";
+ }
+ } else {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status not ok, load_id="
+ << _parent->_load_id << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_rpc_error, _is_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. rpc_error/cancelled/eos: ");
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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);
Review comment:
As I commented here https://github.com/apache/incubator-doris/pull/3143/files#diff-4007834d7219c2282c8ae5c454f01dbaR67 , the scan node can do the memory limit. I should add more comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on issue #3143: Non
blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-604855354
I wanna change yeild() to sleep(), to avoid CPU busy. I need to prove "even sleep() after each loop, non-blocking sink is faster than blocking one" through testing.
So in the first step, I want the sleep interval to be the config param of BE, and I'll give a default value.
One more thing, the add_row() may be blocked cause mem limit exceeded, I'll add the blocking time to time profile.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken edited a comment on issue #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken edited a comment on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601034196
> Hi @vagetablechicken
>
> Do you have some benchmark for this PR?
I only have two test samples now, testing environment building is in progress in our team.
I use broker load, and the scan node wasn't do mem limit(my test code doesn't contain a80e9bf22968232dead8886d990993eb3b702424). If scan node waits when mem limit exceeded, it must take longer.
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":92095001,"TaskNumber":1,"FileNumber":8,"FileSize":2798532731} | 10min3s | 6min43s |
| {"ScannedRows":951503983,"TaskNumber":1,"FileNumber":300,"FileSize":68560716133} | 4h25min | 2h6min |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken edited a comment on issue #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken edited a comment on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601034196
> Hi @vagetablechicken
>
> Do you have some benchmark for this PR?
I only have two test samples now, testing environment building is in progress in our team.
I use broker load, and the scan node wasn't do mem limit(my test code doesn't contain a80e9bf22968232dead8886d990993eb3b702424). If scan node waits when mem limit exceeded, it must take longer.
5 be:
- E5-2630v2*2 CPU
- 8*16G Mem
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":92095001,"TaskNumber":1,"FileNumber":8,"FileSize":2798532731} | 10min3s | 6min43s |
| {"ScannedRows":951503983,"TaskNumber":1,"FileNumber":300,"FileSize":68560716133} | 4h25min | 2h6min |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman commented on issue #3143: Non blocking
OlapTableSink
Posted by GitBox <gi...@apache.org>.
morningman commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601154604
Compilation failed:
LOG(DEBUG) << name() << " queue_push_ns: " << _queue_push_lock_ns;
There is no `LOG(DEBUG)` macro
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395413391
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -186,80 +257,89 @@ void NodeChannel::cancel() {
closure->ref();
closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _stub->tablet_writer_cancel(&closure->cntl,
- &request,
- &closure->result,
- closure);
+ _stub->tablet_writer_cancel(&closure->cntl, &request, &closure->result, closure);
request.release_id();
- // reset batch
- _batch.reset();
+ // Beware of the destruct sequence. RowBatches will use mem_trackers(include ancestors).
+ // Delete RowBatches here is a better choice to reduce the potential of dtor errors.
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ std::queue<AddBatchReq> empty;
+ std::swap(_pending_batches, empty);
+ _cur_batch.reset();
+ }
}
-Status NodeChannel::_wait_in_flight_packet() {
- if (!_has_in_flight_packet) {
- return Status::OK();
+int NodeChannel::try_send_and_fetch_status() {
+ auto st = none_of({_rpc_error, _is_cancelled, _send_finished});
+ if (!st.ok()) {
+ return 0;
}
- SCOPED_RAW_TIMER(_parent->mutable_wait_in_flight_packet_ns());
- _add_batch_closure->join();
- _has_in_flight_packet = false;
- if (_add_batch_closure->cntl.Failed()) {
- LOG(WARNING) << "failed to send batch, error="
- << berror(_add_batch_closure->cntl.ErrorCode())
- << ", error_text=" << _add_batch_closure->cntl.ErrorText();
- return Status::InternalError("failed to send batch");
- }
+ if (!_add_batch_closure->is_packet_in_flight() && _pending_batches_num > 0) {
+ SCOPED_RAW_TIMER(&_actual_consume_ns);
+ AddBatchReq send_batch;
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ DCHECK(!_pending_batches.empty());
+ send_batch = std::move(_pending_batches.front());
+ _pending_batches.pop();
+ _pending_batches_num--;
+ }
- if (_add_batch_closure->result.has_execution_time_us()) {
- _parent->update_node_add_batch_counter(_node_id,
- _add_batch_closure->result.execution_time_us(),
- _add_batch_closure->result.wait_lock_time_us());
- }
- return {_add_batch_closure->result.status()};
-}
+ auto row_batch = std::move(send_batch.first);
+ auto request = std::move(send_batch.second); // doesn't need to be saved in heap
-Status NodeChannel::_send_cur_batch(bool eos) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
+ // tablet_ids has already set when add row
+ request.set_packet_seq(_next_packet_seq);
+ if (row_batch->num_rows() > 0) {
+ SCOPED_RAW_TIMER(&_serialize_batch_ns);
+ row_batch->serialize(request.mutable_row_batch());
+ }
- // tablet_ids has already set when add row
- _add_batch_request.set_eos(eos);
- _add_batch_request.set_packet_seq(_next_packet_seq);
- if (_batch->num_rows() > 0) {
- SCOPED_RAW_TIMER(_parent->mutable_serialize_batch_ns());
- _batch->serialize(_add_batch_request.mutable_row_batch());
- }
+ _add_batch_closure->reset();
+ _add_batch_closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _add_batch_closure->ref();
- _add_batch_closure->cntl.Reset();
- _add_batch_closure->cntl.set_timeout_ms(_rpc_timeout_ms);
+ if (request.eos()) {
+ for (auto pid : _parent->_partition_ids) {
+ request.add_partition_ids(pid);
+ }
- if (eos) {
- for (auto pid : _parent->_partition_ids) {
- _add_batch_request.add_partition_ids(pid);
+ // eos request must be the last request
+ _add_batch_closure->end_mark();
+ _send_finished = true;
+ DCHECK(_pending_batches_num == 0);
+ LOG(INFO) << name() << " send finished, should wait the last repsonse";
}
- }
- _stub->tablet_writer_add_batch(&_add_batch_closure->cntl,
- &_add_batch_request,
- &_add_batch_closure->result,
- _add_batch_closure);
- _add_batch_request.clear_tablet_ids();
- _add_batch_request.clear_row_batch();
- _add_batch_request.clear_partition_ids();
+ _add_batch_closure->set_in_flight();
+ _stub->tablet_writer_add_batch(&_add_batch_closure->cntl, &request,
+ &_add_batch_closure->result, _add_batch_closure);
- _has_in_flight_packet = true;
- _next_packet_seq++;
+ _next_packet_seq++;
+ }
- _batch->reset();
- return Status::OK();
+ return _send_finished ? 0 : 1;
}
-IndexChannel::~IndexChannel() {
+Status NodeChannel::none_of(std::initializer_list<bool> vars) {
+ bool none = true;
+ std::string vars_str;
Review comment:
I think most of the scenarios are still normal. better to construct strings when it is abnormal
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r406032771
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -877,5 +920,24 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
return filtered_rows;
}
+void OlapTableSink::_send_batch_process() {
+ SCOPED_RAW_TIMER(&_non_blocking_send_ns);
+ while (true) {
+ int running_channels_num = 0;
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel([&](NodeChannel* ch) {
+ running_channels_num += ch->try_send_and_fetch_status();
+ });
+ }
+
+ if (running_channels_num == 0) {
+ LOG(INFO) << "all node channels are stopped(maybe finished/offending/cancelled), "
+ "consumer thread exit.";
+ return;
+ }
+ std::this_thread::yield();
Review comment:
change it to sleep(interval_time)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395951339
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +129,128 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, load_id=" << _parent->_load_id
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ LOG(INFO) << name() << " last rpc has responsed";
Review comment:
Is this log needed?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on issue #3143: Non
blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-611406948
@imay @morningman
Sorry about my change is a bit late.
I've changed yeild() to sleep(). And the default interval time is 10ms.
Why 10ms, it's calculated through the avg time of add batch RPCs.
If the case is large, it get faster obviously, so I focus on little cases this time.
I tested two little cases, and got the logs about add batch time. The log is generated by this piece of code.
https://github.com/apache/incubator-doris/blob/b4014d89d45e7644fe4d098eb0ac937553006bfd/be/src/exec/tablet_sink.cpp#L689-L698
So `avg time` = `add batch time` / `num`. And if you think the network time delay is not negligible, you can add it.
In my tests, the avg times are tens of milliseconds. So for the faster sink, I set it to 10ms.
And the test result(interval=10ms) is
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":28582315,"TaskNumber":1,"FileNumber":3,"FileSize":569042681} | 4min37s | 3min27s |
|{"ScannedRows":100967318,"TaskNumber":1,"FileNumber":9,"FileSize":3096583144} | 12min27s | 8min27s |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman commented on a change in pull request
#3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
morningman commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r398612691
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -186,80 +267,89 @@ void NodeChannel::cancel() {
closure->ref();
closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _stub->tablet_writer_cancel(&closure->cntl,
- &request,
- &closure->result,
- closure);
+ _stub->tablet_writer_cancel(&closure->cntl, &request, &closure->result, closure);
request.release_id();
- // reset batch
- _batch.reset();
+ // Beware of the destruct sequence. RowBatches will use mem_trackers(include ancestors).
+ // Delete RowBatches here is a better choice to reduce the potential of dtor errors.
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ std::queue<AddBatchReq> empty;
+ std::swap(_pending_batches, empty);
+ _cur_batch.reset();
+ }
}
-Status NodeChannel::_wait_in_flight_packet() {
- if (!_has_in_flight_packet) {
- return Status::OK();
+int NodeChannel::try_send_and_fetch_status() {
Review comment:
Why not using bool as return value?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman commented on issue #3143: Non blocking
OlapTableSink
Posted by GitBox <gi...@apache.org>.
morningman commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-604467763
Hi @vagetablechicken , please resolve the conflict.
And Hi @imay , do you have any other comment?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395959565
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -877,5 +920,24 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
return filtered_rows;
}
+void OlapTableSink::_send_batch_process() {
+ SCOPED_RAW_TIMER(&_non_blocking_send_ns);
+ while (true) {
+ int running_channels_num = 0;
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel([&](NodeChannel* ch) {
+ running_channels_num += ch->try_send_and_fetch_status();
+ });
+ }
+
+ if (running_channels_num == 0) {
+ LOG(INFO) << "all node channels are stopped(maybe finished/offending/cancelled), "
+ "consumer thread exit.";
+ return;
+ }
+ std::this_thread::yield();
Review comment:
Yes, it's the easiest to implement, but not good enough. I've consider about writing an algorithm that can choose the property sleep interval, e.g. if every channel is empty, we can sleep for a while. bla bla bla
But taking the trouble to do this may lead to small gains.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395410355
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -47,18 +48,101 @@ class ExprContext;
class TExpr;
namespace stream_load {
-
+
class OlapTableSink;
+// The counter of add_batch rpc of a single node
+struct AddBatchCounter {
+ // total execution time of a add_batch rpc
+ int64_t add_batch_execution_time_us = 0;
+ // lock waiting time in a add_batch rpc
+ int64_t add_batch_wait_lock_time_us = 0;
+ // number of add_batch call
+ int64_t add_batch_num = 0;
+ AddBatchCounter& operator+=(const AddBatchCounter& rhs) {
+ add_batch_execution_time_us += rhs.add_batch_execution_time_us;
+ add_batch_wait_lock_time_us += rhs.add_batch_wait_lock_time_us;
+ add_batch_num += rhs.add_batch_num;
+ return *this;
+ }
+ friend AddBatchCounter operator+(const AddBatchCounter& lhs, const AddBatchCounter& rhs) {
+ AddBatchCounter sum = lhs;
+ sum += rhs;
+ return sum;
+ }
+};
+
+template <typename T>
+class ReusableClosure : public google::protobuf::Closure {
Review comment:
I think should add ref count for this closure. Because this closure is created by channel, and the RPC call has this pointer. However it is not sure when the RPC will call this->Run(). If this channel is destructed before the RPC's callback, it will make invalid memory access. Otherwise it should wait until the RPC finish, which is unnecessary.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on issue #3143: Non blocking
OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-600985445
Hi @vagetablechicken
Do you have some benchmark for this PR?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r399037164
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -186,80 +267,89 @@ void NodeChannel::cancel() {
closure->ref();
closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _stub->tablet_writer_cancel(&closure->cntl,
- &request,
- &closure->result,
- closure);
+ _stub->tablet_writer_cancel(&closure->cntl, &request, &closure->result, closure);
request.release_id();
- // reset batch
- _batch.reset();
+ // Beware of the destruct sequence. RowBatches will use mem_trackers(include ancestors).
+ // Delete RowBatches here is a better choice to reduce the potential of dtor errors.
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ std::queue<AddBatchReq> empty;
+ std::swap(_pending_batches, empty);
+ _cur_batch.reset();
+ }
}
-Status NodeChannel::_wait_in_flight_packet() {
- if (!_has_in_flight_packet) {
- return Status::OK();
+int NodeChannel::try_send_and_fetch_status() {
Review comment:
I want the consumer to have global observation, sum() is the simplest way to do it.
And maybe we can add more status(>0) to let consumer know more infos about channels.
But I think the preliminary design should be simple.
So current design of status value is "0-channel stopped; 1-still working".
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395951806
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -877,5 +920,24 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
return filtered_rows;
}
+void OlapTableSink::_send_batch_process() {
+ SCOPED_RAW_TIMER(&_non_blocking_send_ns);
+ while (true) {
+ int running_channels_num = 0;
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel([&](NodeChannel* ch) {
+ running_channels_num += ch->try_send_and_fetch_status();
+ });
+ }
+
+ if (running_channels_num == 0) {
+ LOG(INFO) << "all node channels are stopped(maybe finished/offending/cancelled), "
+ "consumer thread exit.";
+ return;
+ }
+ std::this_thread::yield();
Review comment:
If there is no other workload, will this cause a busy-loop?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395492007
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -47,18 +48,101 @@ class ExprContext;
class TExpr;
namespace stream_load {
-
+
class OlapTableSink;
+// The counter of add_batch rpc of a single node
+struct AddBatchCounter {
+ // total execution time of a add_batch rpc
+ int64_t add_batch_execution_time_us = 0;
+ // lock waiting time in a add_batch rpc
+ int64_t add_batch_wait_lock_time_us = 0;
+ // number of add_batch call
+ int64_t add_batch_num = 0;
+ AddBatchCounter& operator+=(const AddBatchCounter& rhs) {
+ add_batch_execution_time_us += rhs.add_batch_execution_time_us;
+ add_batch_wait_lock_time_us += rhs.add_batch_wait_lock_time_us;
+ add_batch_num += rhs.add_batch_num;
+ return *this;
+ }
+ friend AddBatchCounter operator+(const AddBatchCounter& lhs, const AddBatchCounter& rhs) {
+ AddBatchCounter sum = lhs;
+ sum += rhs;
+ return sum;
+ }
+};
+
+template <typename T>
+class ReusableClosure : public google::protobuf::Closure {
Review comment:
Actually, I want the NodeChannel to be the owner of add_batch_closure.
https://github.com/apache/incubator-doris/pull/3143/files#diff-4007834d7219c2282c8ae5c454f01dbaR47-R51
In NodeChannel's dtor `_add_batch_closure->release();`, release() will do brpc::Join(). So RPC won't call Run() after channel is destructed.
If OlapTableSink is closed normally, `_add_batch_closure->release();` has no effect.
But as you mentioned about unnecessary waiting RPC finished , if OlapTableSink is closed with error, `_add_batch_closure->release();` in NodeChannel dtor may take some time to wait add_batch repsonse, although the add_batch response is useless under the circumstance that we cancelled the whole sink.
I've considered about brpc::StartCancel(call_id), but I'm not good enough at understanding brpc, so I wrote a comment [here](https://github.com/apache/incubator-doris/pull/3143/files#diff-4007834d7219c2282c8ae5c454f01dbaR249). And just let the channel wait for the useless RPC response.
Maybe we should do brpc::StartCancel(call_id)?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395492007
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -47,18 +48,101 @@ class ExprContext;
class TExpr;
namespace stream_load {
-
+
class OlapTableSink;
+// The counter of add_batch rpc of a single node
+struct AddBatchCounter {
+ // total execution time of a add_batch rpc
+ int64_t add_batch_execution_time_us = 0;
+ // lock waiting time in a add_batch rpc
+ int64_t add_batch_wait_lock_time_us = 0;
+ // number of add_batch call
+ int64_t add_batch_num = 0;
+ AddBatchCounter& operator+=(const AddBatchCounter& rhs) {
+ add_batch_execution_time_us += rhs.add_batch_execution_time_us;
+ add_batch_wait_lock_time_us += rhs.add_batch_wait_lock_time_us;
+ add_batch_num += rhs.add_batch_num;
+ return *this;
+ }
+ friend AddBatchCounter operator+(const AddBatchCounter& lhs, const AddBatchCounter& rhs) {
+ AddBatchCounter sum = lhs;
+ sum += rhs;
+ return sum;
+ }
+};
+
+template <typename T>
+class ReusableClosure : public google::protobuf::Closure {
Review comment:
Actually, I want the NodeChannel to the owner of add_batch_closure.
https://github.com/apache/incubator-doris/pull/3143/files#diff-4007834d7219c2282c8ae5c454f01dbaR47-R51
In NodeChannel's dtor `_add_batch_closure->release();`, release() will do brpc::Join(). So RPC won't call Run() after channel is destructed.
If OlapTableSink is closed normally, `_add_batch_closure->release();` has no effect.
But as you mentioned about unnecessary waiting RPC finished , if OlapTableSink is closed with error, `_add_batch_closure->release();` in NodeChannel dtor may take some time to wait add_batch repsonse, although the add_batch response is useless under the circumstance that we cancelled the whole sink.
I've considered about brpc::StartCancel(call_id), but I'm not good enough at understanding brpc, so I wrote a comment [here](https://github.com/apache/incubator-doris/pull/3143/files#diff-4007834d7219c2282c8ae5c454f01dbaR249). And just let the channel wait for the useless RPC response.
Maybe we should do brpc::StartCancel(call_id)?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395464396
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +127,127 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([&]() {
Review comment:
I think `[this]` is OK
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken edited a comment on issue #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken edited a comment on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601034196
> Hi @vagetablechicken
>
> Do you have some benchmark for this PR?
I only have two test samples now, testing environment building is in progress in our team.
I use broker load, and the scan node wasn't do mem limit(my test code doesn't contain a80e9bf22968232dead8886d990993eb3b702424). If scan node waits when mem limit exceeded, it must take longer.
5 be:
- 24 Intel(R) Xeon(R) CPU E5-2620 v3 @ 2.40GHz
- 128GB Mem
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":92095001,"TaskNumber":1,"FileNumber":8,"FileSize":2798532731} | 10min3s | 6min43s |
| {"ScannedRows":951503983,"TaskNumber":1,"FileNumber":300,"FileSize":68560716133} | 4h25min | 2h6min |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395456198
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +127,127 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([&]() {
Review comment:
The two handlers both need to capture _rpc_error(the member variable), I'm afraid I can only change [&] to [this].
There's an alternate way to explicit capture, although I don't like this style. How about this?
```
auto& rpc_error_ref=_rpc_error;
_add_batch_closure->addFailedHandler([&rpc_error_ref](){})
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken edited a comment on issue #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken edited a comment on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-604855354
I wanna change yeild() to sleep(), to avoid CPU busy. I need to prove "even sleep() after each loop, non-blocking sink is faster than blocking one" through testing.
So in the first step, I want the sleep interval to be the config param of BE, and I'll give a default value.
One more thing, the add_row() may be blocked cause mem limit exceeded, I'll add the blocking time to time profile.
Your comments are welcome.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on issue #3143: Non
blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601034196
> Hi @vagetablechicken
>
> Do you have some benchmark for this PR?
I only have two test samples now, testing environment building is in progress in our team.
I use broker load, test 5 times in each version, and the scan node wasn't do mem limit(my test code doesn't contain a80e9bf22968232dead8886d990993eb3b702424). If scan node waits when mem limit exceeded, it must take longer.
| test sample | origin ver(avg) | nonblocking ver(avg) |
|-|-|-|
| {"ScannedRows":92095001,"TaskNumber":1,"FileNumber":8,"FileSize":2798532731} | 10min3s | 6min43s |
| {"ScannedRows":951503983,"TaskNumber":1,"FileNumber":300,"FileSize":68560716133} | 4h25min | 2h6min |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395953272
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -186,80 +260,90 @@ void NodeChannel::cancel() {
closure->ref();
closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _stub->tablet_writer_cancel(&closure->cntl,
- &request,
- &closure->result,
- closure);
+ _stub->tablet_writer_cancel(&closure->cntl, &request, &closure->result, closure);
request.release_id();
- // reset batch
- _batch.reset();
+ // Beware of the destruct sequence. RowBatches will use mem_trackers(include ancestors).
+ // Delete RowBatches here is a better choice to reduce the potential of dtor errors.
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ std::queue<AddBatchReq> empty;
+ std::swap(_pending_batches, empty);
+ _cur_batch.reset();
+ }
}
-Status NodeChannel::_wait_in_flight_packet() {
- if (!_has_in_flight_packet) {
- return Status::OK();
+int NodeChannel::try_send_and_fetch_status() {
+ auto st = none_of({_rpc_error, _is_cancelled, _send_finished});
+ if (!st.ok()) {
+ return 0;
}
- SCOPED_RAW_TIMER(_parent->mutable_wait_in_flight_packet_ns());
- _add_batch_closure->join();
- _has_in_flight_packet = false;
- if (_add_batch_closure->cntl.Failed()) {
- LOG(WARNING) << "failed to send batch, error="
- << berror(_add_batch_closure->cntl.ErrorCode())
- << ", error_text=" << _add_batch_closure->cntl.ErrorText();
- return Status::InternalError("failed to send batch");
- }
+ if (!_add_batch_closure->is_packet_in_flight() && _pending_batches_num > 0) {
+ SCOPED_RAW_TIMER(&_actual_consume_ns);
+ AddBatchReq send_batch;
+ {
+ std::lock_guard<std::mutex> lg(_pending_batches_lock);
+ DCHECK(!_pending_batches.empty());
+ send_batch = std::move(_pending_batches.front());
+ _pending_batches.pop();
+ _pending_batches_num--;
+ }
- if (_add_batch_closure->result.has_execution_time_us()) {
- _parent->update_node_add_batch_counter(_node_id,
- _add_batch_closure->result.execution_time_us(),
- _add_batch_closure->result.wait_lock_time_us());
- }
- return {_add_batch_closure->result.status()};
-}
+ auto row_batch = std::move(send_batch.first);
+ auto request = std::move(send_batch.second); // doesn't need to be saved in heap
-Status NodeChannel::_send_cur_batch(bool eos) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
+ // tablet_ids has already set when add row
+ request.set_packet_seq(_next_packet_seq);
+ if (row_batch->num_rows() > 0) {
+ SCOPED_RAW_TIMER(&_serialize_batch_ns);
+ row_batch->serialize(request.mutable_row_batch());
+ }
- // tablet_ids has already set when add row
- _add_batch_request.set_eos(eos);
- _add_batch_request.set_packet_seq(_next_packet_seq);
- if (_batch->num_rows() > 0) {
- SCOPED_RAW_TIMER(_parent->mutable_serialize_batch_ns());
- _batch->serialize(_add_batch_request.mutable_row_batch());
- }
+ _add_batch_closure->reset();
+ _add_batch_closure->cntl.set_timeout_ms(_rpc_timeout_ms);
- _add_batch_closure->ref();
- _add_batch_closure->cntl.Reset();
- _add_batch_closure->cntl.set_timeout_ms(_rpc_timeout_ms);
+ if (request.eos()) {
+ for (auto pid : _parent->_partition_ids) {
+ request.add_partition_ids(pid);
+ }
- if (eos) {
- for (auto pid : _parent->_partition_ids) {
- _add_batch_request.add_partition_ids(pid);
+ // eos request must be the last request
+ _add_batch_closure->end_mark();
+ _send_finished = true;
+ DCHECK(_pending_batches_num == 0);
+ LOG(INFO) << name() << " send finished, should wait the last repsonse";
Review comment:
should remove this trace log
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395407530
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -95,72 +193,79 @@ class NodeChannel {
TupleDescriptor* _tuple_desc = nullptr;
const NodeInfo* _node_info = nullptr;
- bool _already_failed = false;
- bool _has_in_flight_packet = false;
// this should be set in init() using config
int _rpc_timeout_ms = 60000;
int64_t _next_packet_seq = 0;
- std::unique_ptr<RowBatch> _batch;
+ std::atomic<bool> _rpc_error{false};
+ std::atomic<bool> _is_cancelled{false};
+
+ std::atomic<bool> _send_finished{false};
+ std::atomic<bool> _add_batches_finished{false};
+
+ bool _eos_is_produced{false}; // only for restricting producer behaviors
+
+ std::unique_ptr<RowDescriptor> _row_desc;
+ int _batch_size = 0;
+ std::unique_ptr<RowBatch> _cur_batch;
+ PTabletWriterAddBatchRequest _cur_add_batch_request;
+
+ std::mutex _pending_batches_lock;
+ using AddBatchReq = std::pair<std::unique_ptr<RowBatch>, PTabletWriterAddBatchRequest>;
+ std::queue<AddBatchReq> _pending_batches;
+ std::atomic<int> _pending_batches_num{0};
+
palo::PInternalService_Stub* _stub = nullptr;
RefCountClosure<PTabletWriterOpenResult>* _open_closure = nullptr;
- RefCountClosure<PTabletWriterAddBatchResult>* _add_batch_closure = nullptr;
+ ReusableClosure<PTabletWriterAddBatchResult>* _add_batch_closure = nullptr;
std::vector<TTabletWithPartition> _all_tablets;
- PTabletWriterAddBatchRequest _add_batch_request;
+ std::vector<TTabletCommitInfo> _tablet_commit_infos;
+
+ AddBatchCounter _add_batch_counter;
+ int64_t _queue_push_lock_ns = 0;
+ int64_t _serialize_batch_ns = 0;
+ int64_t _actual_consume_ns = 0;
};
class IndexChannel {
public:
IndexChannel(OlapTableSink* parent, int64_t index_id, int32_t schema_hash)
- : _parent(parent), _index_id(index_id),
- _schema_hash(schema_hash) {
- }
+ : _parent(parent), _index_id(index_id), _schema_hash(schema_hash) {}
~IndexChannel();
- Status init(RuntimeState* state,
- const std::vector<TTabletWithPartition>& tablets);
- Status open();
- Status add_row(Tuple* tuple, int64_t tablet_id);
+ Status init(RuntimeState* state, const std::vector<TTabletWithPartition>& tablets);
- Status close(RuntimeState* state);
+ Status add_row(Tuple* tuple, int64_t tablet_id);
- void cancel();
+ void for_each_node_channel(std::function<void(NodeChannel*)> func) {
+ for (auto& it : _node_channels) {
+ func(it.second);
+ }
+ }
-private:
- // return true if this load can't success.
- bool _handle_failed_node(NodeChannel* channel);
+ void mark_as_failed(NodeChannel* ch) { _failed_channels.insert(ch->node_id()); }
+ bool has_intolerable_failure();
private:
OlapTableSink* _parent;
int64_t _index_id;
int32_t _schema_hash;
- int _num_failed_channels = 0;
// BeId -> channel
std::unordered_map<int64_t, NodeChannel*> _node_channels;
// from tablet_id to backend channel
std::unordered_map<int64_t, std::vector<NodeChannel*>> _channels_by_tablet;
-};
-
-// The counter of add_batch rpc of a single node
-struct AddBatchCounter {
- // total execution time of a add_batch rpc
- int64_t add_batch_execution_time_ns = 0;
- // lock waiting time in a add_batch rpc
- int64_t add_batch_wait_lock_time_ns = 0;
- // number of add_batch call
- int64_t add_batch_num = 0;
+ // BeId
+ std::set<int64_t> _failed_channels;
};
// write data to Olap Table.
// this class distributed data according to
class OlapTableSink : public DataSink {
Review comment:
Better to add some comments to explain how the work is done, which will make others understand this code easily.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395412001
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +127,127 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([&]() {
Review comment:
I think it is better to declare all your capture other than a &. And what's more, referencing a local variable will lead to invalid memory access, because when the lambda is called, the reference is not valid.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395524007
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -47,18 +48,101 @@ class ExprContext;
class TExpr;
namespace stream_load {
-
+
class OlapTableSink;
+// The counter of add_batch rpc of a single node
+struct AddBatchCounter {
+ // total execution time of a add_batch rpc
+ int64_t add_batch_execution_time_us = 0;
+ // lock waiting time in a add_batch rpc
+ int64_t add_batch_wait_lock_time_us = 0;
+ // number of add_batch call
+ int64_t add_batch_num = 0;
+ AddBatchCounter& operator+=(const AddBatchCounter& rhs) {
+ add_batch_execution_time_us += rhs.add_batch_execution_time_us;
+ add_batch_wait_lock_time_us += rhs.add_batch_wait_lock_time_us;
+ add_batch_num += rhs.add_batch_num;
+ return *this;
+ }
+ friend AddBatchCounter operator+(const AddBatchCounter& lhs, const AddBatchCounter& rhs) {
+ AddBatchCounter sum = lhs;
+ sum += rhs;
+ return sum;
+ }
+};
+
+template <typename T>
+class ReusableClosure : public google::protobuf::Closure {
Review comment:
Yes, you are right.
It is OK to call release before delete it. But it should be commented carefully.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on issue #3143: Non
blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on issue #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#issuecomment-601218611
> Compilation failed:
> LOG(DEBUG) << name() << " queue_push_ns: " << _queue_push_lock_ns;
>
> There is no `LOG(DEBUG)` macro
Oops, my bad. Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395953751
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -611,57 +633,79 @@ Status OlapTableSink::close(RuntimeState* state, Status close_status) {
// only if status is ok can we call this _profile->total_time_counter().
// if status is not ok, this sink may not be prepared, so that _profile is null
SCOPED_TIMER(_profile->total_time_counter());
+ int64_t serialize_batch_ns = 0, queue_push_lock_ns = 0, actual_consume_ns = 0;
{
SCOPED_TIMER(_close_timer);
- for (auto channel : _channels) {
- status = channel->close(state);
- if (!status.ok()) {
- LOG(WARNING) << "close channel failed, load_id=" << print_id(_load_id)
- << ", txn_id=" << _txn_id;
- }
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel(
+ [&](NodeChannel* ch) { WARN_IF_ERROR(ch->mark_close(), ""); });
+ }
+
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel([&](NodeChannel* ch) {
Review comment:
change to explicit capture
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395548920
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -47,18 +48,101 @@ class ExprContext;
class TExpr;
namespace stream_load {
-
+
class OlapTableSink;
+// The counter of add_batch rpc of a single node
+struct AddBatchCounter {
+ // total execution time of a add_batch rpc
+ int64_t add_batch_execution_time_us = 0;
+ // lock waiting time in a add_batch rpc
+ int64_t add_batch_wait_lock_time_us = 0;
+ // number of add_batch call
+ int64_t add_batch_num = 0;
+ AddBatchCounter& operator+=(const AddBatchCounter& rhs) {
+ add_batch_execution_time_us += rhs.add_batch_execution_time_us;
+ add_batch_wait_lock_time_us += rhs.add_batch_wait_lock_time_us;
+ add_batch_num += rhs.add_batch_num;
+ return *this;
+ }
+ friend AddBatchCounter operator+(const AddBatchCounter& lhs, const AddBatchCounter& rhs) {
+ AddBatchCounter sum = lhs;
+ sum += rhs;
+ return sum;
+ }
+};
+
+template <typename T>
+class ReusableClosure : public google::protobuf::Closure {
Review comment:
I think the func name `release()` is a little hard to understand, let me think about it, and I'll add more comments on the usage of ReusableClosure.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull
request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395959565
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -877,5 +920,24 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
return filtered_rows;
}
+void OlapTableSink::_send_batch_process() {
+ SCOPED_RAW_TIMER(&_non_blocking_send_ns);
+ while (true) {
+ int running_channels_num = 0;
+ for (auto index_channel : _channels) {
+ index_channel->for_each_node_channel([&](NodeChannel* ch) {
+ running_channels_num += ch->try_send_and_fetch_status();
+ });
+ }
+
+ if (running_channels_num == 0) {
+ LOG(INFO) << "all node channels are stopped(maybe finished/offending/cancelled), "
+ "consumer thread exit.";
+ return;
+ }
+ std::this_thread::yield();
Review comment:
Yes, it's the easiest to implement, but not good enough. I've consider about writing an algorithm that can choose the property sleep interval, e.g. if every channel is empty, we can sleep for a while. bla bla bla
But taking the trouble to do this may lead to small gains.
Or we should just sleep_for? Long sleep interval saves CPU, but waste time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3143:
Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143: Non blocking OlapTableSink
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r395954239
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,55 +129,128 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, load_id=" << _parent->_load_id
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ LOG(INFO) << name() << " last rpc has responsed";
+ }
+ } else {
+ _rpc_error = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status not ok, load_id="
+ << _parent->_load_id << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_rpc_error, _is_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. rpc_error/cancelled/eos: ");
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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);
Review comment:
we should limit the length of pending_batches.
When sender is blocked, this queue will consume memory without limit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r417728936
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,54 +131,136 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, " << print_load_info()
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ }
+ } else {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status isn't ok, "
+ << print_load_info() << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. cancelled/eos: ");
+ }
+
+ // We use OlapTableSink mem_tracker which has the same ancestor of _plan node,
+ // so in the ideal case, mem limit is a matter for _plan node.
+ // But there is still some unfinished things, we do mem limit here temporarily.
+ while (_parent->_mem_tracker->any_limit_exceeded()) {
+ SCOPED_RAW_TIMER(&_mem_exceeded_block_ns);
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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++;
+ }
+
+ _cur_batch.reset(new RowBatch(*_row_desc, _batch_size, _parent->_mem_tracker));
+ _cur_add_batch_request.clear_tablet_ids();
+
+ row_no = _cur_batch->add_row();
}
DCHECK_NE(row_no, RowBatch::INVALID_ROW_INDEX);
- auto tuple = input_tuple->deep_copy(*_tuple_desc, _batch->tuple_data_pool());
- _batch->get_row(row_no)->set_tuple(0, tuple);
- _batch->commit_last_row();
- _add_batch_request.add_tablet_ids(tablet_id);
+ auto tuple = input_tuple->deep_copy(*_tuple_desc, _cur_batch->tuple_data_pool());
+ _cur_batch->get_row(row_no)->set_tuple(0, tuple);
+ _cur_batch->commit_last_row();
+ _cur_add_batch_request.add_tablet_ids(tablet_id);
return Status::OK();
}
-Status NodeChannel::close(RuntimeState* state) {
- auto st = _close(state);
- _batch.reset();
- return st;
-}
+Status NodeChannel::mark_close() {
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't mark as closed. cancelled/eos: ");
+ }
-Status NodeChannel::_close(RuntimeState* state) {
- return _send_cur_batch(true);
+ _cur_add_batch_request.set_eos(true);
+ {
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ _pending_batches.emplace(std::move(_cur_batch), _cur_add_batch_request);
+ _pending_batches_num++;
+ DCHECK(_pending_batches.back().second.eos());
+ }
+
+ _eos_is_produced = true;
+
+ _cur_batch.reset();
+ return Status::OK();
}
Status NodeChannel::close_wait(RuntimeState* state) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
- Status status(_add_batch_closure->result.status());
- if (status.ok()) {
- for (auto& tablet : _add_batch_closure->result.tablet_vec()) {
- TTabletCommitInfo commit_info;
- commit_info.tabletId = tablet.tablet_id();
- commit_info.backendId = _node_id;
- state->tablet_commit_infos().emplace_back(std::move(commit_info));
- }
+ auto st = none_of({_cancelled, !_eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, skip waiting for close. cancelled/!eos: ");
}
- // clear batch after sendt
- _batch.reset();
- return status;
+
+ // waiting for finished, it may take a long time, so we could't set a timeout
+ // use log to make it easier
+ LOG(INFO) << name() << "start close_wait";
+ while (!_add_batches_finished && !_cancelled) {
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ }
+ LOG(INFO) << name() << "close_wait done";
Review comment:
We just count the total time of channel's close(). There may be some slow BE hide inside. I think this logs are good for analyzing the state at that time. It's convenient for developers.
How about set it to vlog?
----------------------------------------------------------------
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.
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
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r417763651
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,54 +131,136 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, " << print_load_info()
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ }
+ } else {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status isn't ok, "
+ << print_load_info() << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. cancelled/eos: ");
+ }
+
+ // We use OlapTableSink mem_tracker which has the same ancestor of _plan node,
+ // so in the ideal case, mem limit is a matter for _plan node.
+ // But there is still some unfinished things, we do mem limit here temporarily.
+ while (_parent->_mem_tracker->any_limit_exceeded()) {
+ SCOPED_RAW_TIMER(&_mem_exceeded_block_ns);
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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++;
+ }
+
+ _cur_batch.reset(new RowBatch(*_row_desc, _batch_size, _parent->_mem_tracker));
+ _cur_add_batch_request.clear_tablet_ids();
+
+ row_no = _cur_batch->add_row();
}
DCHECK_NE(row_no, RowBatch::INVALID_ROW_INDEX);
- auto tuple = input_tuple->deep_copy(*_tuple_desc, _batch->tuple_data_pool());
- _batch->get_row(row_no)->set_tuple(0, tuple);
- _batch->commit_last_row();
- _add_batch_request.add_tablet_ids(tablet_id);
+ auto tuple = input_tuple->deep_copy(*_tuple_desc, _cur_batch->tuple_data_pool());
+ _cur_batch->get_row(row_no)->set_tuple(0, tuple);
+ _cur_batch->commit_last_row();
+ _cur_add_batch_request.add_tablet_ids(tablet_id);
return Status::OK();
}
-Status NodeChannel::close(RuntimeState* state) {
- auto st = _close(state);
- _batch.reset();
- return st;
-}
+Status NodeChannel::mark_close() {
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't mark as closed. cancelled/eos: ");
+ }
-Status NodeChannel::_close(RuntimeState* state) {
- return _send_cur_batch(true);
+ _cur_add_batch_request.set_eos(true);
+ {
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ _pending_batches.emplace(std::move(_cur_batch), _cur_add_batch_request);
+ _pending_batches_num++;
+ DCHECK(_pending_batches.back().second.eos());
+ }
+
+ _eos_is_produced = true;
+
+ _cur_batch.reset();
+ return Status::OK();
}
Status NodeChannel::close_wait(RuntimeState* state) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
- Status status(_add_batch_closure->result.status());
- if (status.ok()) {
- for (auto& tablet : _add_batch_closure->result.tablet_vec()) {
- TTabletCommitInfo commit_info;
- commit_info.tabletId = tablet.tablet_id();
- commit_info.backendId = _node_id;
- state->tablet_commit_infos().emplace_back(std::move(commit_info));
- }
+ auto st = none_of({_cancelled, !_eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, skip waiting for close. cancelled/!eos: ");
}
- // clear batch after sendt
- _batch.reset();
- return status;
+
+ // waiting for finished, it may take a long time, so we could't set a timeout
+ // use log to make it easier
+ LOG(INFO) << name() << "start close_wait";
+ while (!_add_batches_finished && !_cancelled) {
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ }
+ LOG(INFO) << name() << "close_wait done";
Review comment:
Good advice, I'll fix it.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-doris] imay commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r416586004
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -68,99 +154,126 @@ class NodeChannel {
Status add_row(Tuple* tuple, int64_t tablet_id);
- Status close(RuntimeState* state);
+ // two ways to stop channel:
+ // 1. mark_close()->close_wait() PS. close_wait() will block waiting for the last AddBatch rpc response.
+ // 2. just cancel()
+ Status mark_close();
Status close_wait(RuntimeState* state);
void cancel();
- int64_t node_id() const { return _node_id; }
+ // return:
+ // 0: stopped, send finished(eos request has been sent), or any internal error;
+ // 1: running, haven't reach eos.
+ // only allow 1 rpc in flight
+ int try_send_and_fetch_status();
+
+ void time_report(std::unordered_map<int64_t, AddBatchCounter>& add_batch_counter_map,
+ int64_t* serialize_batch_ns, int64_t* mem_exceeded_block_ns,
+ int64_t* queue_push_lock_ns, int64_t* actual_consume_ns) {
+ add_batch_counter_map[_node_id] += _add_batch_counter;
+ *serialize_batch_ns += _serialize_batch_ns;
+ *mem_exceeded_block_ns += _mem_exceeded_block_ns;
+ *queue_push_lock_ns += _queue_push_lock_ns;
+ *actual_consume_ns += _actual_consume_ns;
+ }
- void set_failed() { _already_failed = true; }
- bool already_failed() const { return _already_failed; }
+ int64_t node_id() const { return _node_id; }
const NodeInfo* node_info() const { return _node_info; }
+ std::string print_load_info() { return _load_info; }
+ std::string name() const {
+ return "NodeChannel[" + std::to_string(_index_id) + "-" + std::to_string(_node_id) + "]";
+ }
-private:
- Status _send_cur_batch(bool eos = false);
- // wait inflight packet finish, return error if inflight packet return failed
- Status _wait_in_flight_packet();
-
- Status _close(RuntimeState* state);
+ Status none_of(std::initializer_list<bool> vars);
private:
OlapTableSink* _parent = nullptr;
int64_t _index_id = -1;
int64_t _node_id = -1;
int32_t _schema_hash = 0;
+ std::string _load_info;
TupleDescriptor* _tuple_desc = nullptr;
const NodeInfo* _node_info = nullptr;
- bool _already_failed = false;
- bool _has_in_flight_packet = false;
// this should be set in init() using config
int _rpc_timeout_ms = 60000;
int64_t _next_packet_seq = 0;
- std::unique_ptr<RowBatch> _batch;
+ // user cancel or get some errors
+ std::atomic<bool> _cancelled{false};
+
+ std::atomic<bool> _send_finished{false};
+ std::atomic<bool> _add_batches_finished{false};
+
+ bool _eos_is_produced{false}; // only for restricting producer behaviors
+
+ std::unique_ptr<RowDescriptor> _row_desc;
+ int _batch_size = 0;
+ std::unique_ptr<RowBatch> _cur_batch;
+ PTabletWriterAddBatchRequest _cur_add_batch_request;
+
+ std::mutex _pending_batches_lock;
+ using AddBatchReq = std::pair<std::unique_ptr<RowBatch>, PTabletWriterAddBatchRequest>;
+ std::queue<AddBatchReq> _pending_batches;
+ std::atomic<int> _pending_batches_num{0};
+
palo::PInternalService_Stub* _stub = nullptr;
RefCountClosure<PTabletWriterOpenResult>* _open_closure = nullptr;
- RefCountClosure<PTabletWriterAddBatchResult>* _add_batch_closure = nullptr;
+ ReusableClosure<PTabletWriterAddBatchResult>* _add_batch_closure = nullptr;
std::vector<TTabletWithPartition> _all_tablets;
- PTabletWriterAddBatchRequest _add_batch_request;
+ std::vector<TTabletCommitInfo> _tablet_commit_infos;
+
+ AddBatchCounter _add_batch_counter;
+ int64_t _serialize_batch_ns = 0;
+
+ int64_t _mem_exceeded_block_ns = 0;
+ int64_t _queue_push_lock_ns = 0;
+ int64_t _actual_consume_ns = 0;
};
class IndexChannel {
public:
IndexChannel(OlapTableSink* parent, int64_t index_id, int32_t schema_hash)
- : _parent(parent), _index_id(index_id),
- _schema_hash(schema_hash) {
- }
+ : _parent(parent), _index_id(index_id), _schema_hash(schema_hash) {}
~IndexChannel();
- Status init(RuntimeState* state,
- const std::vector<TTabletWithPartition>& tablets);
- Status open();
- Status add_row(Tuple* tuple, int64_t tablet_id);
+ Status init(RuntimeState* state, const std::vector<TTabletWithPartition>& tablets);
- Status close(RuntimeState* state);
+ Status add_row(Tuple* tuple, int64_t tablet_id);
- void cancel();
+ void for_each_node_channel(std::function<void(NodeChannel*)> func) {
Review comment:
```suggestion
void for_each_node_channel(const std::function<void(NodeChannel*)>& func) {
```
##########
File path: be/src/common/config.h
##########
@@ -296,6 +296,8 @@ namespace config {
// you may need to increase this timeout if using larger 'streaming_load_max_mb',
// or encounter 'tablet writer write failed' error when loading.
// CONF_Int32(tablet_writer_rpc_timeout_sec, "600");
+ // OlapTableSink sender's send interval, should be less than the real response time of a tablet writer rpc.
+ CONF_Int32(olap_table_sink_send_interval_ms, "10");
Review comment:
```suggestion
CONF_mInt32(olap_table_sink_send_interval_ms, "10");
```
----------------------------------------------------------------
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.
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
[GitHub] [incubator-doris] vagetablechicken commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
vagetablechicken commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r417745660
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -68,99 +154,126 @@ class NodeChannel {
Status add_row(Tuple* tuple, int64_t tablet_id);
- Status close(RuntimeState* state);
+ // two ways to stop channel:
+ // 1. mark_close()->close_wait() PS. close_wait() will block waiting for the last AddBatch rpc response.
+ // 2. just cancel()
+ Status mark_close();
Status close_wait(RuntimeState* state);
void cancel();
- int64_t node_id() const { return _node_id; }
+ // return:
+ // 0: stopped, send finished(eos request has been sent), or any internal error;
+ // 1: running, haven't reach eos.
+ // only allow 1 rpc in flight
+ int try_send_and_fetch_status();
+
+ void time_report(std::unordered_map<int64_t, AddBatchCounter>& add_batch_counter_map,
+ int64_t* serialize_batch_ns, int64_t* mem_exceeded_block_ns,
+ int64_t* queue_push_lock_ns, int64_t* actual_consume_ns) {
+ add_batch_counter_map[_node_id] += _add_batch_counter;
+ *serialize_batch_ns += _serialize_batch_ns;
+ *mem_exceeded_block_ns += _mem_exceeded_block_ns;
+ *queue_push_lock_ns += _queue_push_lock_ns;
+ *actual_consume_ns += _actual_consume_ns;
+ }
- void set_failed() { _already_failed = true; }
- bool already_failed() const { return _already_failed; }
+ int64_t node_id() const { return _node_id; }
const NodeInfo* node_info() const { return _node_info; }
+ std::string print_load_info() const { return _load_info; }
+ std::string name() const {
+ return "NodeChannel[" + std::to_string(_index_id) + "-" + std::to_string(_node_id) + "]";
Review comment:
including gutils/strings/Substitute.h will occur redefinition error
```
In file included from /root/incubator-doris/be/src/gutil/strings/stringpiece.h:128:0,
from /root/incubator-doris/be/src/gutil/strings/substitute.h:9,
from /root/incubator-doris/be/src/exec/tablet_sink.cpp:28:
/root/incubator-doris/be/src/gutil/hash/hash.h:251:26: error: redefinition of 'struct __gnu_cxx::hash<Type*>'
template<class T> struct hash<T*> {
^~~~~~~~
In file included from /root/incubator-doris/thirdparty/installed/include/butil/containers/flat_map.h:101:0,
from /root/incubator-doris/be/src/service/brpc.h:48,
from /root/incubator-doris/be/src/util/ref_count_closure.h:24,
from /root/incubator-doris/be/src/exec/tablet_sink.h:36,
from /root/incubator-doris/be/src/exec/tablet_sink.cpp:18:
/root/incubator-doris/thirdparty/installed/include/butil/containers/hash_tables.h:262:8: note: previous definition of 'struct __gnu_cxx::hash<Type*>'
struct hash<Type*> {
^~~~~~~~~~~
In file included from /root/incubator-doris/be/src/gutil/strings/stringpiece.h:128:0,
from /root/incubator-doris/be/src/gutil/strings/substitute.h:9,
from /root/incubator-doris/be/src/exec/tablet_sink.cpp:28:
/root/incubator-doris/be/src/gutil/hash/hash.h:309:8: error: redefinition of 'struct __gnu_cxx::hash<std::pair<_T1, _T2> >'
struct hash<pair<First, Second> > {
^~~~~~~~~~~~~~~~~~~~~~~~~~
In file included from /root/incubator-doris/thirdparty/installed/include/butil/containers/flat_map.h:101:0,
from /root/incubator-doris/be/src/service/brpc.h:48,
from /root/incubator-doris/be/src/util/ref_count_closure.h:24,
from /root/incubator-doris/be/src/exec/tablet_sink.h:36,
from /root/incubator-doris/be/src/exec/tablet_sink.cpp:18:
/root/incubator-doris/thirdparty/installed/include/butil/containers/hash_tables.h:256:8: note: previous definition of 'struct __gnu_cxx::hash<std::pair<_T1, _T2> >'
struct hash<std::pair<Type1, Type2> > {
```
I will initialize the name_string in init(), to avoid string building in name()
----------------------------------------------------------------
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.
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
[GitHub] [incubator-doris] imay commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r417387499
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -68,99 +154,126 @@ class NodeChannel {
Status add_row(Tuple* tuple, int64_t tablet_id);
- Status close(RuntimeState* state);
+ // two ways to stop channel:
+ // 1. mark_close()->close_wait() PS. close_wait() will block waiting for the last AddBatch rpc response.
+ // 2. just cancel()
+ Status mark_close();
Status close_wait(RuntimeState* state);
void cancel();
- int64_t node_id() const { return _node_id; }
+ // return:
+ // 0: stopped, send finished(eos request has been sent), or any internal error;
+ // 1: running, haven't reach eos.
+ // only allow 1 rpc in flight
+ int try_send_and_fetch_status();
+
+ void time_report(std::unordered_map<int64_t, AddBatchCounter>& add_batch_counter_map,
+ int64_t* serialize_batch_ns, int64_t* mem_exceeded_block_ns,
+ int64_t* queue_push_lock_ns, int64_t* actual_consume_ns) {
+ add_batch_counter_map[_node_id] += _add_batch_counter;
+ *serialize_batch_ns += _serialize_batch_ns;
+ *mem_exceeded_block_ns += _mem_exceeded_block_ns;
+ *queue_push_lock_ns += _queue_push_lock_ns;
+ *actual_consume_ns += _actual_consume_ns;
+ }
- void set_failed() { _already_failed = true; }
- bool already_failed() const { return _already_failed; }
+ int64_t node_id() const { return _node_id; }
const NodeInfo* node_info() const { return _node_info; }
+ std::string print_load_info() const { return _load_info; }
+ std::string name() const {
+ return "NodeChannel[" + std::to_string(_index_id) + "-" + std::to_string(_node_id) + "]";
Review comment:
Prefer strings::Substitute in gutils/strings/Substitute.h.
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -68,99 +154,126 @@ class NodeChannel {
Status add_row(Tuple* tuple, int64_t tablet_id);
- Status close(RuntimeState* state);
+ // two ways to stop channel:
+ // 1. mark_close()->close_wait() PS. close_wait() will block waiting for the last AddBatch rpc response.
+ // 2. just cancel()
+ Status mark_close();
Status close_wait(RuntimeState* state);
void cancel();
- int64_t node_id() const { return _node_id; }
+ // return:
+ // 0: stopped, send finished(eos request has been sent), or any internal error;
+ // 1: running, haven't reach eos.
+ // only allow 1 rpc in flight
+ int try_send_and_fetch_status();
+
+ void time_report(std::unordered_map<int64_t, AddBatchCounter>& add_batch_counter_map,
Review comment:
If param will be modified, prefer pointer rather than reference.
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,54 +131,136 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, " << print_load_info()
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ }
+ } else {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status isn't ok, "
+ << print_load_info() << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. cancelled/eos: ");
+ }
+
+ // We use OlapTableSink mem_tracker which has the same ancestor of _plan node,
+ // so in the ideal case, mem limit is a matter for _plan node.
+ // But there is still some unfinished things, we do mem limit here temporarily.
+ while (_parent->_mem_tracker->any_limit_exceeded()) {
+ SCOPED_RAW_TIMER(&_mem_exceeded_block_ns);
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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++;
+ }
+
+ _cur_batch.reset(new RowBatch(*_row_desc, _batch_size, _parent->_mem_tracker));
+ _cur_add_batch_request.clear_tablet_ids();
+
+ row_no = _cur_batch->add_row();
}
DCHECK_NE(row_no, RowBatch::INVALID_ROW_INDEX);
- auto tuple = input_tuple->deep_copy(*_tuple_desc, _batch->tuple_data_pool());
- _batch->get_row(row_no)->set_tuple(0, tuple);
- _batch->commit_last_row();
- _add_batch_request.add_tablet_ids(tablet_id);
+ auto tuple = input_tuple->deep_copy(*_tuple_desc, _cur_batch->tuple_data_pool());
+ _cur_batch->get_row(row_no)->set_tuple(0, tuple);
+ _cur_batch->commit_last_row();
+ _cur_add_batch_request.add_tablet_ids(tablet_id);
return Status::OK();
}
-Status NodeChannel::close(RuntimeState* state) {
- auto st = _close(state);
- _batch.reset();
- return st;
-}
+Status NodeChannel::mark_close() {
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't mark as closed. cancelled/eos: ");
+ }
-Status NodeChannel::_close(RuntimeState* state) {
- return _send_cur_batch(true);
+ _cur_add_batch_request.set_eos(true);
+ {
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ _pending_batches.emplace(std::move(_cur_batch), _cur_add_batch_request);
+ _pending_batches_num++;
+ DCHECK(_pending_batches.back().second.eos());
+ }
+
+ _eos_is_produced = true;
+
+ _cur_batch.reset();
+ return Status::OK();
}
Status NodeChannel::close_wait(RuntimeState* state) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
- Status status(_add_batch_closure->result.status());
- if (status.ok()) {
- for (auto& tablet : _add_batch_closure->result.tablet_vec()) {
- TTabletCommitInfo commit_info;
- commit_info.tabletId = tablet.tablet_id();
- commit_info.backendId = _node_id;
- state->tablet_commit_infos().emplace_back(std::move(commit_info));
- }
+ auto st = none_of({_cancelled, !_eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, skip waiting for close. cancelled/!eos: ");
}
- // clear batch after sendt
- _batch.reset();
- return status;
+
+ // waiting for finished, it may take a long time, so we could't set a timeout
+ // use log to make it easier
+ LOG(INFO) << name() << "start close_wait";
+ while (!_add_batches_finished && !_cancelled) {
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ }
+ LOG(INFO) << name() << "close_wait done";
Review comment:
Do we need these log?
##########
File path: be/src/exec/tablet_sink.h
##########
@@ -68,99 +154,126 @@ class NodeChannel {
Status add_row(Tuple* tuple, int64_t tablet_id);
- Status close(RuntimeState* state);
+ // two ways to stop channel:
+ // 1. mark_close()->close_wait() PS. close_wait() will block waiting for the last AddBatch rpc response.
+ // 2. just cancel()
+ Status mark_close();
Status close_wait(RuntimeState* state);
void cancel();
- int64_t node_id() const { return _node_id; }
+ // return:
+ // 0: stopped, send finished(eos request has been sent), or any internal error;
+ // 1: running, haven't reach eos.
+ // only allow 1 rpc in flight
+ int try_send_and_fetch_status();
+
+ void time_report(std::unordered_map<int64_t, AddBatchCounter>& add_batch_counter_map,
+ int64_t* serialize_batch_ns, int64_t* mem_exceeded_block_ns,
+ int64_t* queue_push_lock_ns, int64_t* actual_consume_ns) {
+ add_batch_counter_map[_node_id] += _add_batch_counter;
+ *serialize_batch_ns += _serialize_batch_ns;
+ *mem_exceeded_block_ns += _mem_exceeded_block_ns;
+ *queue_push_lock_ns += _queue_push_lock_ns;
+ *actual_consume_ns += _actual_consume_ns;
+ }
- void set_failed() { _already_failed = true; }
- bool already_failed() const { return _already_failed; }
+ int64_t node_id() const { return _node_id; }
const NodeInfo* node_info() const { return _node_info; }
+ std::string print_load_info() const { return _load_info; }
+ std::string name() const {
+ return "NodeChannel[" + std::to_string(_index_id) + "-" + std::to_string(_node_id) + "]";
+ }
-private:
- Status _send_cur_batch(bool eos = false);
- // wait inflight packet finish, return error if inflight packet return failed
- Status _wait_in_flight_packet();
-
- Status _close(RuntimeState* state);
+ Status none_of(std::initializer_list<bool> vars);
private:
OlapTableSink* _parent = nullptr;
int64_t _index_id = -1;
int64_t _node_id = -1;
int32_t _schema_hash = 0;
+ std::string _load_info;
TupleDescriptor* _tuple_desc = nullptr;
const NodeInfo* _node_info = nullptr;
- bool _already_failed = false;
- bool _has_in_flight_packet = false;
// this should be set in init() using config
int _rpc_timeout_ms = 60000;
int64_t _next_packet_seq = 0;
- std::unique_ptr<RowBatch> _batch;
+ // user cancel or get some errors
+ std::atomic<bool> _cancelled{false};
+
+ std::atomic<bool> _send_finished{false};
+ std::atomic<bool> _add_batches_finished{false};
Review comment:
what's the difference between these two flags?
Should comment to let us know
----------------------------------------------------------------
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.
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
[GitHub] [incubator-doris] imay commented on a change in pull request #3143: Non blocking OlapTableSink
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3143:
URL: https://github.com/apache/incubator-doris/pull/3143#discussion_r417744700
##########
File path: be/src/exec/tablet_sink.cpp
##########
@@ -128,54 +131,136 @@ Status NodeChannel::open_wait() {
_open_closure = nullptr;
// add batch closure
- _add_batch_closure = new RefCountClosure<PTabletWriterAddBatchResult>();
- _add_batch_closure->ref();
+ _add_batch_closure = ReusableClosure<PTabletWriterAddBatchResult>::create();
+ _add_batch_closure->addFailedHandler([this]() {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req rpc failed, " << print_load_info()
+ << ", node=" << node_info()->host << ":" << node_info()->brpc_port;
+ });
+
+ _add_batch_closure->addSuccessHandler(
+ [this](const PTabletWriterAddBatchResult& result, bool is_last_rpc) {
+ Status status(result.status());
+ if (status.ok()) {
+ if (is_last_rpc) {
+ for (auto& tablet : result.tablet_vec()) {
+ TTabletCommitInfo commit_info;
+ commit_info.tabletId = tablet.tablet_id();
+ commit_info.backendId = _node_id;
+ _tablet_commit_infos.emplace_back(std::move(commit_info));
+ }
+ _add_batches_finished = true;
+ }
+ } else {
+ _cancelled = true;
+ LOG(WARNING) << "NodeChannel add batch req success but status isn't ok, "
+ << print_load_info() << ", node=" << node_info()->host << ":"
+ << node_info()->brpc_port << ", errmsg=" << status.get_error_msg();
+ }
+
+ if (result.has_execution_time_us()) {
+ _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
+ _add_batch_counter.add_batch_wait_lock_time_us += result.wait_lock_time_us();
+ _add_batch_counter.add_batch_num++;
+ }
+ });
return status;
}
Status NodeChannel::add_row(Tuple* input_tuple, int64_t tablet_id) {
- auto row_no = _batch->add_row();
+ // If add_row() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't add_row. cancelled/eos: ");
+ }
+
+ // We use OlapTableSink mem_tracker which has the same ancestor of _plan node,
+ // so in the ideal case, mem limit is a matter for _plan node.
+ // But there is still some unfinished things, we do mem limit here temporarily.
+ while (_parent->_mem_tracker->any_limit_exceeded()) {
+ SCOPED_RAW_TIMER(&_mem_exceeded_block_ns);
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+
+ auto row_no = _cur_batch->add_row();
if (row_no == RowBatch::INVALID_ROW_INDEX) {
- RETURN_IF_ERROR(_send_cur_batch());
- row_no = _batch->add_row();
+ {
+ SCOPED_RAW_TIMER(&_queue_push_lock_ns);
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ //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++;
+ }
+
+ _cur_batch.reset(new RowBatch(*_row_desc, _batch_size, _parent->_mem_tracker));
+ _cur_add_batch_request.clear_tablet_ids();
+
+ row_no = _cur_batch->add_row();
}
DCHECK_NE(row_no, RowBatch::INVALID_ROW_INDEX);
- auto tuple = input_tuple->deep_copy(*_tuple_desc, _batch->tuple_data_pool());
- _batch->get_row(row_no)->set_tuple(0, tuple);
- _batch->commit_last_row();
- _add_batch_request.add_tablet_ids(tablet_id);
+ auto tuple = input_tuple->deep_copy(*_tuple_desc, _cur_batch->tuple_data_pool());
+ _cur_batch->get_row(row_no)->set_tuple(0, tuple);
+ _cur_batch->commit_last_row();
+ _cur_add_batch_request.add_tablet_ids(tablet_id);
return Status::OK();
}
-Status NodeChannel::close(RuntimeState* state) {
- auto st = _close(state);
- _batch.reset();
- return st;
-}
+Status NodeChannel::mark_close() {
+ auto st = none_of({_cancelled, _eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, can't mark as closed. cancelled/eos: ");
+ }
-Status NodeChannel::_close(RuntimeState* state) {
- return _send_cur_batch(true);
+ _cur_add_batch_request.set_eos(true);
+ {
+ std::lock_guard<std::mutex> l(_pending_batches_lock);
+ _pending_batches.emplace(std::move(_cur_batch), _cur_add_batch_request);
+ _pending_batches_num++;
+ DCHECK(_pending_batches.back().second.eos());
+ }
+
+ _eos_is_produced = true;
+
+ _cur_batch.reset();
+ return Status::OK();
}
Status NodeChannel::close_wait(RuntimeState* state) {
- RETURN_IF_ERROR(_wait_in_flight_packet());
- Status status(_add_batch_closure->result.status());
- if (status.ok()) {
- for (auto& tablet : _add_batch_closure->result.tablet_vec()) {
- TTabletCommitInfo commit_info;
- commit_info.tabletId = tablet.tablet_id();
- commit_info.backendId = _node_id;
- state->tablet_commit_infos().emplace_back(std::move(commit_info));
- }
+ auto st = none_of({_cancelled, !_eos_is_produced});
+ if (!st.ok()) {
+ return st.clone_and_prepend("already stopped, skip waiting for close. cancelled/!eos: ");
}
- // clear batch after sendt
- _batch.reset();
- return status;
+
+ // waiting for finished, it may take a long time, so we could't set a timeout
+ // use log to make it easier
+ LOG(INFO) << name() << "start close_wait";
+ while (!_add_batches_finished && !_cancelled) {
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ }
+ LOG(INFO) << name() << "close_wait done";
Review comment:
If You want to, you can count the time with StopWatch and log the time in one log.
I suggest to use vlog.
----------------------------------------------------------------
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.
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