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

[chore](enhancement) remove atomic load shared ptr usage (#49781)

Browse Source
This commit is contained in:
yiguolei
2025-04-03 17:57:12 +08:00
committed by GitHub
parent 2b08429577
commit 672829096e
2 changed files with 8 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
be/src/pipeline/task_queue.cpp
View File

@ -154,22 +154,20 @@ void MultiCoreTaskQueue::close() {
PipelineTask* MultiCoreTaskQueue::take(int core_id) {
PipelineTask* task = nullptr;
auto prio_task_queue_list =
std::atomic_load_explicit(&_prio_task_queue_list, std::memory_order_relaxed);
while (!_closed) {
DCHECK(prio_task_queue_list->size() > core_id)
<< " list size: " << prio_task_queue_list->size() << " core_id: " << core_id
DCHECK(_prio_task_queue_list->size() > core_id)
<< " list size: " << _prio_task_queue_list->size() << " core_id: " << core_id
<< " _core_size: " << _core_size << " _next_core: " << _next_core.load();
task = (*prio_task_queue_list)[core_id]->try_take(false);
task = (*_prio_task_queue_list)[core_id]->try_take(false);
if (task) {
task->set_core_id(core_id);
break;
}
task = _steal_take(core_id, *prio_task_queue_list);
task = _steal_take(core_id, *_prio_task_queue_list);
if (task) {
break;
}
task = (*prio_task_queue_list)[core_id]->take(WAIT_CORE_TASK_TIMEOUT_MS /* timeout_ms */);
task = (*_prio_task_queue_list)[core_id]->take(WAIT_CORE_TASK_TIMEOUT_MS /* timeout_ms */);
if (task) {
task->set_core_id(core_id);
break;
@ -211,9 +209,7 @@ Status MultiCoreTaskQueue::push_back(PipelineTask* task) {
Status MultiCoreTaskQueue::push_back(PipelineTask* task, int core_id) {
DCHECK(core_id < _core_size);
task->put_in_runnable_queue();
auto prio_task_queue_list =
std::atomic_load_explicit(&_prio_task_queue_list, std::memory_order_relaxed);
return (*prio_task_queue_list)[core_id]->push(task);
return (*_prio_task_queue_list)[core_id]->push(task);
}
} // namespace pipeline

6
be/src/pipeline/task_queue.h
View File

@ -144,10 +144,8 @@ public:
void update_statistics(PipelineTask* task, int64_t time_spent) override {
task->inc_runtime_ns(time_spent);
auto prio_task_queue_list =
std::atomic_load_explicit(&_prio_task_queue_list, std::memory_order_relaxed);
(*prio_task_queue_list)[task->get_core_id()]->inc_sub_queue_runtime(task->get_queue_level(),
time_spent);
(*_prio_task_queue_list)[task->get_core_id()]->inc_sub_queue_runtime(
task->get_queue_level(), time_spent);
}
private: