Line data Source code
1 : #pragma once
2 :
3 : #include "audioworkgroup.h"
4 : #include "concurrency/threadutils.h"
5 : #include "runtime.h"
6 : #include "concurrentqueue.h"
7 : #include "global/functional/inplace_function.h"
8 : #include "lightweightsemaphore.h"
9 : #include "thirdparty/kors_logger/src/log_base.h"
10 :
11 : #include <atomic>
12 : #include <exception>
13 : #include <memory>
14 : #include <mutex>
15 : #include <set>
16 : #include <string>
17 : #include <thread>
18 : #include <vector>
19 :
20 : namespace muse::audio {
21 : class RealtimeThreadPool
22 : {
23 : struct InflightSemaphoreRelease {
24 0 : explicit InflightSemaphoreRelease(muse::LightweightSemaphore& semaphore)
25 0 : : m_semaphore(semaphore) {}
26 :
27 : InflightSemaphoreRelease(const InflightSemaphoreRelease&) = delete;
28 : InflightSemaphoreRelease& operator=(const InflightSemaphoreRelease&) = delete;
29 :
30 0 : ~InflightSemaphoreRelease() noexcept
31 : {
32 0 : m_semaphore.signal();
33 0 : }
34 :
35 : muse::LightweightSemaphore& m_semaphore;
36 : };
37 :
38 : public:
39 : static constexpr int maxTaskCount = 10000;
40 : using Task = muse::functional::inplace_function<void (), 64>;
41 24 : RealtimeThreadPool(
42 : std::string name,
43 : int num_of_workers = std::thread::hardware_concurrency())
44 24 : {
45 24 : num_of_workers = std::max(0, num_of_workers);
46 24 : m_workers.reserve(num_of_workers);
47 : try {
48 72 : for (size_t i = 0; i < static_cast<size_t>(num_of_workers); ++i) {
49 48 : auto worker = std::make_unique<Worker>();
50 48 : Worker* workerPtr = worker.get();
51 48 : const size_t workerIndex = i;
52 96 : worker->m_thread = std::make_unique<std::thread>(
53 96 : [this, workerPtr, workerIndex, name] {
54 48 : muse::runtime::setThreadName(name + "_" + std::to_string(workerIndex));
55 48 : AudioWorkgroupToken workgroupToken;
56 : for (;;) {
57 48 : Task task;
58 :
59 48 : m_queue.wait_dequeue(task);
60 : {
61 47 : std::lock_guard lock(workerPtr->m_workgroupMutex);
62 48 : workerPtr->m_workgroup.join(workgroupToken);
63 47 : }
64 :
65 47 : if (this->m_should_stop) {
66 93 : return;
67 : }
68 :
69 0 : InflightSemaphoreRelease release(m_inflightSemaphore);
70 0 : task();
71 46 : }
72 94 : });
73 48 : m_workers.push_back(std::move(worker));
74 48 : muse::setThreadPriority(*m_workers.back()->m_thread, ThreadPriority::High);
75 48 : }
76 0 : } catch (...) {
77 0 : terminate();
78 0 : throw;
79 0 : }
80 24 : }
81 :
82 27 : void setAudioWorkgroup(muse::audio::AudioWorkGroup audioworkgroup)
83 : {
84 81 : for (auto& worker : m_workers) {
85 54 : std::lock_guard lock(worker->m_workgroupMutex);
86 54 : worker->m_workgroup = audioworkgroup;
87 54 : }
88 27 : }
89 :
90 0 : bool enqueue(const Task& func)
91 : {
92 0 : m_inflightSemaphore.wait();
93 0 : if (!m_queue.enqueue(func)) {
94 0 : m_inflightSemaphore.signal();
95 0 : return false;
96 : }
97 0 : return true;
98 : }
99 :
100 0 : void participateAndWait()
101 : {
102 0 : Task task;
103 0 : while (m_queue.try_dequeue(task)) {
104 0 : InflightSemaphoreRelease release(m_inflightSemaphore);
105 0 : task();
106 0 : }
107 0 : waitUntilFinished();
108 0 : m_inflightSemaphore.signal(maxTaskCount);
109 0 : }
110 :
111 0 : std::set<std::thread::id> threadIdSet() const
112 : {
113 0 : std::set<std::thread::id> result;
114 :
115 0 : for (const auto& worker : m_workers) {
116 0 : result.insert(worker->m_thread->get_id());
117 : }
118 :
119 0 : return result;
120 0 : }
121 :
122 27 : int getNumberOfWorkers() const
123 : {
124 27 : return static_cast<int>(m_workers.size());
125 : }
126 :
127 24 : ~RealtimeThreadPool()
128 : {
129 24 : terminate();
130 24 : }
131 :
132 : private:
133 0 : void waitUntilFinished()
134 : {
135 0 : auto actuallyAwaited = m_inflightSemaphore.waitMany(maxTaskCount);
136 0 : for (size_t i = 0;
137 0 : i < static_cast<size_t>(maxTaskCount - actuallyAwaited); ++i) {
138 0 : m_inflightSemaphore.wait();
139 : }
140 0 : }
141 :
142 24 : void terminate()
143 : {
144 24 : m_should_stop = true;
145 72 : for (size_t i = 0; i < m_workers.size(); ++i) {
146 48 : IF_ASSERT_FAILED(m_queue.enqueue([] {})) {
147 : // this is _extremely_ unlikely to fail. But if it does the threads would hang indefinitely.
148 0 : std::terminate();
149 : }
150 : }
151 :
152 72 : for (auto& worker : m_workers) {
153 48 : if (worker->m_thread && worker->m_thread->joinable()) {
154 48 : worker->m_thread->join();
155 : }
156 : }
157 24 : }
158 :
159 : struct Worker {
160 : std::unique_ptr<std::thread> m_thread;
161 : AudioWorkGroup m_workgroup;
162 : std::mutex m_workgroupMutex;
163 : };
164 :
165 : muse::BlockingConcurrentQueue<Task > m_queue;
166 : std::vector<std::unique_ptr<Worker> > m_workers;
167 : muse::LightweightSemaphore m_inflightSemaphore { maxTaskCount };
168 : std::atomic<bool> m_should_stop{ false };
169 : }; // namespace muse::audio
170 : } // namespace muse::audio
|
