database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[improve](move-memtable) limit task num in load stream flush token (#28748)

Browse Source
This commit is contained in:
Kaijie Chen
2023-12-21 12:19:58 +08:00
committed by GitHub
parent 425bbaa4fd
commit db523dafcb
3 changed files with 15 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
be/src/common/config.cpp
View File

@ -776,6 +776,8 @@ DEFINE_Int64(load_stream_max_buf_size, "20971520"); // 20MB
DEFINE_Int32(load_stream_messages_in_batch, "128");
// brpc streaming StreamWait seconds on EAGAIN
DEFINE_Int32(load_stream_eagain_wait_seconds, "60");
// max tasks per flush token in load stream
DEFINE_Int32(load_stream_flush_token_max_tasks, "2");
// max send batch parallelism for OlapTableSink
// The value set by the user for send_batch_parallelism is not allowed to exceed max_send_batch_parallelism_per_job,

2
be/src/common/config.h
View File

@ -837,6 +837,8 @@ DECLARE_Int64(load_stream_max_buf_size);
DECLARE_Int32(load_stream_messages_in_batch);
// brpc streaming StreamWait seconds on EAGAIN
DECLARE_Int32(load_stream_eagain_wait_seconds);
// max tasks per flush token in load stream
DECLARE_Int32(load_stream_flush_token_max_tasks);
// max send batch parallelism for OlapTableSink
// The value set by the user for send_batch_parallelism is not allowed to exceed max_send_batch_parallelism_per_job,

13
be/src/runtime/load_stream.cpp
View File

@ -18,6 +18,7 @@
#include "runtime/load_stream.h"
#include <brpc/stream.h>
#include <bthread/bthread.h>
#include <bthread/condition_variable.h>
#include <bthread/mutex.h>
#include <olap/rowset/rowset_factory.h>
@ -136,7 +137,11 @@ Status TabletStream::append_data(const PStreamHeader& header, butil::IOBuf* data
LOG(INFO) << "write data failed " << *this;
}
};
return _flush_tokens[new_segid % _flush_tokens.size()]->submit_func(flush_func);
auto& flush_token = _flush_tokens[new_segid % _flush_tokens.size()];
while (flush_token->num_tasks() >= config::load_stream_flush_token_max_tasks) {
bthread_usleep(10 * 1000); // 10ms
}
return flush_token->submit_func(flush_func);
}
Status TabletStream::add_segment(const PStreamHeader& header, butil::IOBuf* data) {
@ -170,7 +175,11 @@ Status TabletStream::add_segment(const PStreamHeader& header, butil::IOBuf* data
LOG(INFO) << "add segment failed " << *this;
}
};
return _flush_tokens[new_segid % _flush_tokens.size()]->submit_func(add_segment_func);
auto& flush_token = _flush_tokens[new_segid % _flush_tokens.size()];
while (flush_token->num_tasks() >= config::load_stream_flush_token_max_tasks) {
bthread_usleep(10 * 1000); // 10ms
}
return flush_token->submit_func(add_segment_func);
}
Status TabletStream::close() {