-
Notifications
You must be signed in to change notification settings - Fork 409
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Pipeline: support io priority queue (#7722)
ref #6518
- Loading branch information
Showing
31 changed files
with
474 additions
and
81 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
130 changes: 130 additions & 0 deletions
130
dbms/src/Flash/Pipeline/Schedule/TaskQueues/IOPriorityQueue.cpp
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| // Copyright 2023 PingCAP, Ltd. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| #include <Flash/Pipeline/Schedule/TaskQueues/IOPriorityQueue.h> | ||
| #include <Flash/Pipeline/Schedule/Tasks/TaskHelper.h> | ||
| #include <common/likely.h> | ||
|
|
||
| namespace DB | ||
| { | ||
| IOPriorityQueue::~IOPriorityQueue() | ||
| { | ||
| RUNTIME_ASSERT(io_in_task_queue.empty(), logger, "all task should be taken before it is destructed"); | ||
| RUNTIME_ASSERT(io_out_task_queue.empty(), logger, "all task should be taken before it is destructed"); | ||
| } | ||
|
|
||
| bool IOPriorityQueue::take(TaskPtr & task) | ||
| { | ||
| std::unique_lock lock(mu); | ||
| while (true) | ||
| { | ||
| bool io_out_first = ratio_of_out_to_in * total_io_in_time_microsecond >= total_io_out_time_microsecond; | ||
| auto & first_queue = io_out_first ? io_out_task_queue : io_in_task_queue; | ||
| auto & next_queue = io_out_first ? io_in_task_queue : io_out_task_queue; | ||
| if (!first_queue.empty()) | ||
| { | ||
| task = std::move(first_queue.front()); | ||
| first_queue.pop_front(); | ||
| return true; | ||
| } | ||
| if (!next_queue.empty()) | ||
| { | ||
| task = std::move(next_queue.front()); | ||
| next_queue.pop_front(); | ||
| return true; | ||
| } | ||
| if (unlikely(is_finished)) | ||
| return false; | ||
| cv.wait(lock); | ||
| } | ||
| } | ||
|
|
||
| void IOPriorityQueue::updateStatistics(const TaskPtr &, ExecTaskStatus exec_task_status, UInt64 inc_ns) | ||
| { | ||
| switch (exec_task_status) | ||
| { | ||
| case ExecTaskStatus::IO_IN: | ||
| total_io_in_time_microsecond += (inc_ns / 1000); | ||
| break; | ||
| case ExecTaskStatus::IO_OUT: | ||
| total_io_out_time_microsecond += (inc_ns / 1000); | ||
| break; | ||
| default:; // ignore not io status. | ||
| } | ||
| } | ||
|
|
||
| bool IOPriorityQueue::empty() const | ||
| { | ||
| std::lock_guard lock(mu); | ||
| return io_out_task_queue.empty() && io_in_task_queue.empty(); | ||
| } | ||
|
|
||
| void IOPriorityQueue::finish() | ||
| { | ||
| { | ||
| std::lock_guard lock(mu); | ||
| is_finished = true; | ||
| } | ||
| cv.notify_all(); | ||
| } | ||
|
|
||
| void IOPriorityQueue::submitTaskWithoutLock(TaskPtr && task) | ||
| { | ||
| auto status = task->getStatus(); | ||
| switch (status) | ||
| { | ||
| case ExecTaskStatus::IO_IN: | ||
| io_in_task_queue.push_back(std::move(task)); | ||
| break; | ||
| case ExecTaskStatus::IO_OUT: | ||
| io_out_task_queue.push_back(std::move(task)); | ||
| break; | ||
| default: | ||
| throw Exception(fmt::format("Unexpected status: {}, IOPriorityQueue only accepts tasks with IO status", magic_enum::enum_name(status))); | ||
| } | ||
| } | ||
|
|
||
| void IOPriorityQueue::submit(TaskPtr && task) | ||
| { | ||
| if unlikely (is_finished) | ||
| { | ||
| FINALIZE_TASK(task); | ||
| return; | ||
| } | ||
|
|
||
| { | ||
| std::lock_guard lock(mu); | ||
| submitTaskWithoutLock(std::move(task)); | ||
| } | ||
| cv.notify_one(); | ||
| } | ||
|
|
||
| void IOPriorityQueue::submit(std::vector<TaskPtr> & tasks) | ||
| { | ||
| if unlikely (is_finished) | ||
| { | ||
| FINALIZE_TASKS(tasks); | ||
| return; | ||
| } | ||
|
|
||
| if (tasks.empty()) | ||
| return; | ||
| std::lock_guard lock(mu); | ||
| for (auto & task : tasks) | ||
| { | ||
| submitTaskWithoutLock(std::move(task)); | ||
| cv.notify_one(); | ||
| } | ||
| } | ||
| } // namespace DB |
63 changes: 63 additions & 0 deletions
63
dbms/src/Flash/Pipeline/Schedule/TaskQueues/IOPriorityQueue.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| // Copyright 2023 PingCAP, Ltd. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| #pragma once | ||
|
|
||
| #include <Flash/Pipeline/Schedule/TaskQueues/TaskQueue.h> | ||
|
|
||
| #include <deque> | ||
| #include <mutex> | ||
|
|
||
| namespace DB | ||
| { | ||
| /// The queue only used by io thread pool. | ||
| /// In IOPriorityQueue, the priority of io_out is higher than io_in, which means the ratio of the total execution time of io_out to io_in is `ratio_of_in_to_out`:1. | ||
| /// Because the IO_OUT task usually writes the data in the memory to the external storage and releases the occupied memory, | ||
| /// while the IO_IN task usually reads the data from the external storage into the memory and occupies the memory. | ||
| /// Prioritizing the execution of IO_OUT tasks can effectively reduce the memory usage. | ||
| class IOPriorityQueue : public TaskQueue | ||
| { | ||
| public: | ||
| // // The ratio of total execution time between io_in and io_out is 3:1. | ||
| static constexpr size_t ratio_of_out_to_in = 3; | ||
|
|
||
| ~IOPriorityQueue() override; | ||
|
|
||
| void submit(TaskPtr && task) override; | ||
|
|
||
| void submit(std::vector<TaskPtr> & tasks) override; | ||
|
|
||
| bool take(TaskPtr & task) override; | ||
|
|
||
| void updateStatistics(const TaskPtr &, ExecTaskStatus exec_task_status, UInt64 inc_ns) override; | ||
|
|
||
| bool empty() const override; | ||
|
|
||
| void finish() override; | ||
|
|
||
| private: | ||
| void submitTaskWithoutLock(TaskPtr && task); | ||
|
|
||
| private: | ||
| mutable std::mutex mu; | ||
| std::condition_variable cv; | ||
| std::atomic_bool is_finished = false; | ||
|
|
||
| std::deque<TaskPtr> io_in_task_queue; | ||
| std::atomic_uint64_t total_io_in_time_microsecond{0}; | ||
|
|
||
| std::deque<TaskPtr> io_out_task_queue; | ||
| std::atomic_uint64_t total_io_out_time_microsecond{0}; | ||
| }; | ||
| } // namespace DB |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.