Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Reduce the lock granularity of TaskWorkerPool #3571

Merged
merged 6 commits into from
Feb 24, 2022
Merged

Reduce the lock granularity of TaskWorkerPool #3571

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b2a40a9
Optimize task_worker_pool lock granularity.
Astralidea Feb 21, 2022
362566e
Update be/src/agent/task_worker_pool.cpp
Astralidea Feb 23, 2022
f6b8b34
Opt.
Astralidea Feb 23, 2022
cfbe7b0
format
Astralidea Feb 23, 2022
5466674
Fix.
Astralidea Feb 23, 2022
71370f0
Opt.
Astralidea Feb 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions be/src/agent/task_worker_pool.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ const uint32_t PUBLISH_VERSION_MAX_RETRY = 3;
const uint32_t PUBLISH_VERSION_SUBMIT_MAX_RETRY = 10;

std::atomic_ulong TaskWorkerPool::_s_report_version(time(nullptr) * 10000);
std::mutex TaskWorkerPool::_s_task_signatures_lock;
std::mutex TaskWorkerPool::_s_task_signatures_lock[TTaskType::type::TASK_TYPE_COUNT];
std::map<TTaskType::type, std::set<int64_t>> TaskWorkerPool::_s_task_signatures;
FrontendServiceClientCache TaskWorkerPool::_master_service_client_cache;

Expand Down Expand Up @@ -208,7 +208,7 @@ void TaskWorkerPool::submit_tasks(std::vector<TAgentTaskRequest>* tasks) {
const TTaskType::type task_type = (*tasks)[0].task_type;
EnumToString(TTaskType, task_type, type_str);
{
std::lock_guard task_signatures_lock(_s_task_signatures_lock);
std::lock_guard task_signatures_lock(_s_task_signatures_lock[task_type]);
const auto recv_time = time(nullptr);
for (auto it = tasks->begin(); it != tasks->end();) {
TAgentTaskRequest& task_req = *it;
Expand Down Expand Up @@ -250,13 +250,13 @@ void TaskWorkerPool::submit_tasks(std::vector<TAgentTaskRequest>* tasks) {
}

bool TaskWorkerPool::_register_task_info(const TTaskType::type task_type, int64_t signature) {
std::lock_guard task_signatures_lock(_s_task_signatures_lock);
std::lock_guard task_signatures_lock(_s_task_signatures_lock[task_type]);
Astralidea marked this conversation as resolved.
Show resolved Hide resolved
std::set<int64_t>& signature_set = _s_task_signatures[task_type];
return signature_set.insert(signature).second;
}

void TaskWorkerPool::_remove_task_info(const TTaskType::type task_type, int64_t signature) {
std::lock_guard task_signatures_lock(_s_task_signatures_lock);
std::lock_guard task_signatures_lock(_s_task_signatures_lock[task_type]);
_s_task_signatures[task_type].erase(signature);
}

Expand Down Expand Up @@ -1259,9 +1259,11 @@ void* TaskWorkerPool::_report_task_worker_thread_callback(void* arg_this) {
request.__set_backend(worker_pool_this->_backend);

while ((!worker_pool_this->_stopped)) {
{
std::lock_guard task_signatures_lock(_s_task_signatures_lock);
request.__set_tasks(_s_task_signatures);
for (auto& _s_task_signature : _s_task_signatures) {
Astralidea marked this conversation as resolved.
Show resolved Hide resolved
std::lock_guard task_signatures_lock(_s_task_signatures_lock[_s_task_signature.first]);
std::map<TTaskType::type, std::set<int64_t>> one_type_task;
one_type_task[_s_task_signature.first] = _s_task_signature.second;
request.__set_tasks(one_type_task);
Astralidea marked this conversation as resolved.
Show resolved Hide resolved
}

StarRocksMetrics::instance()->report_task_requests_total.increment(1);
Expand Down
2 changes: 1 addition & 1 deletion be/src/agent/task_worker_pool.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ class TaskWorkerPool {
static FrontendServiceClientCache _master_service_client_cache;
static std::atomic_ulong _s_report_version;

static std::mutex _s_task_signatures_lock;
static std::mutex _s_task_signatures_lock[TTaskType::type::TASK_TYPE_COUNT];
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should manage and define a mutex for current active task types. TASK_TYPE_COUNT include may deprecated type.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This only cause create more unused object, but I am not find an easy way to solve this problem.
I'm more afraid that this will be ignored when a new type is added later.

static std::map<TTaskType::type, std::set<int64_t>> _s_task_signatures;

std::atomic<bool> _stopped{false};
Expand Down
4 changes: 3 additions & 1 deletion gensrc/thrift/Types.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,9 @@ enum TTaskType {
// this type of task will replace both ROLLUP and SCHEMA_CHANGE
ALTER,
INSTALL_PLUGIN,
UNINSTALL_PLUGIN
UNINSTALL_PLUGIN,
// this use for calculate enum count
TASK_TYPE_COUNT
Astralidea marked this conversation as resolved.
Show resolved Hide resolved
}

enum TStmtType {
Expand Down