This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new fa0e0a606b [Fix](autoinc) Ensure that _fetch_autoinc_id_executor is
destructed after the destructions of AutoIncIDBuffers (#24832)
fa0e0a606b is described below
commit fa0e0a606b7f93e06c0bdbd7bf1818660cb67399
Author: bobhan1 <[email protected]>
AuthorDate: Sun Sep 24 21:44:02 2023 +0800
[Fix](autoinc) Ensure that _fetch_autoinc_id_executor is destructed after
the destructions of AutoIncIDBuffers (#24832)
---
be/src/vec/sink/autoinc_buffer.cpp | 11 +----------
be/src/vec/sink/autoinc_buffer.h | 22 ++++++++++++----------
2 files changed, 13 insertions(+), 20 deletions(-)
diff --git a/be/src/vec/sink/autoinc_buffer.cpp
b/be/src/vec/sink/autoinc_buffer.cpp
index 58527f38f7..74747c3c33 100644
--- a/be/src/vec/sink/autoinc_buffer.cpp
+++ b/be/src/vec/sink/autoinc_buffer.cpp
@@ -27,20 +27,11 @@
namespace doris::vectorized {
-FetchAutoIncIDExecutor::FetchAutoIncIDExecutor() {
- ThreadPoolBuilder("AsyncFetchAutoIncIDExecutor")
- .set_min_threads(config::auto_inc_fetch_thread_num)
- .set_max_threads(config::auto_inc_fetch_thread_num)
- .set_max_queue_size(std::numeric_limits<int>::max())
- .build(&_pool);
-}
-
AutoIncIDBuffer::AutoIncIDBuffer(int64_t db_id, int64_t table_id, int64_t
column_id)
: _db_id(db_id),
_table_id(table_id),
_column_id(column_id),
- _rpc_token(FetchAutoIncIDExecutor::GetInstance()->_pool->new_token(
- ThreadPool::ExecutionMode::CONCURRENT)) {}
+ _rpc_token(GlobalAutoIncBuffers::GetInstance()->create_token()) {}
void AutoIncIDBuffer::set_batch_size_at_least(size_t batch_size) {
if (batch_size > _batch_size) {
diff --git a/be/src/vec/sink/autoinc_buffer.h b/be/src/vec/sink/autoinc_buffer.h
index 3443386c82..a8753a473d 100644
--- a/be/src/vec/sink/autoinc_buffer.h
+++ b/be/src/vec/sink/autoinc_buffer.h
@@ -27,16 +27,6 @@ namespace doris::vectorized {
class VOlapTableSink;
class OlapTableBlockConvertor;
-struct FetchAutoIncIDExecutor {
- FetchAutoIncIDExecutor();
- static FetchAutoIncIDExecutor* GetInstance() {
- static FetchAutoIncIDExecutor instance;
- return &instance;
- }
-
- std::unique_ptr<ThreadPool> _pool;
-};
-
struct AutoIncIDAllocator {
int64_t next_id() {
DCHECK(!ids.empty());
@@ -104,12 +94,23 @@ public:
return &buffers;
}
+ GlobalAutoIncBuffers() {
+ static_cast<void>(ThreadPoolBuilder("AsyncFetchAutoIncIDExecutor")
+
.set_min_threads(config::auto_inc_fetch_thread_num)
+
.set_max_threads(config::auto_inc_fetch_thread_num)
+
.set_max_queue_size(std::numeric_limits<int>::max())
+ .build(&_fetch_autoinc_id_executor));
+ }
~GlobalAutoIncBuffers() {
for (auto [_, buffer] : _buffers) {
delete buffer;
}
}
+ std::unique_ptr<ThreadPoolToken> create_token() {
+ return
_fetch_autoinc_id_executor->new_token(ThreadPool::ExecutionMode::CONCURRENT);
+ }
+
AutoIncIDBuffer* get_auto_inc_buffer(int64_t db_id, int64_t table_id,
int64_t column_id) {
std::lock_guard<std::mutex> lock(_mutex);
auto key = std::make_tuple(db_id, table_id, column_id);
@@ -121,6 +122,7 @@ public:
}
private:
+ std::unique_ptr<ThreadPool> _fetch_autoinc_id_executor;
std::map<std::tuple<int64_t, int64_t, int64_t>, AutoIncIDBuffer*> _buffers;
std::mutex _mutex;
};
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
