1/**************************************************************************/
2/* worker_thread_pool.h */
3/**************************************************************************/
4/* This file is part of: */
5/* GODOT ENGINE */
6/* https://godotengine.org */
7/**************************************************************************/
8/* Copyright (c) 2014-present Godot Engine contributors (see AUTHORS.md). */
9/* Copyright (c) 2007-2014 Juan Linietsky, Ariel Manzur. */
10/* */
11/* Permission is hereby granted, free of charge, to any person obtaining */
12/* a copy of this software and associated documentation files (the */
13/* "Software"), to deal in the Software without restriction, including */
14/* without limitation the rights to use, copy, modify, merge, publish, */
15/* distribute, sublicense, and/or sell copies of the Software, and to */
16/* permit persons to whom the Software is furnished to do so, subject to */
17/* the following conditions: */
18/* */
19/* The above copyright notice and this permission notice shall be */
20/* included in all copies or substantial portions of the Software. */
21/* */
22/* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, */
23/* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF */
24/* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. */
25/* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY */
26/* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, */
27/* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE */
28/* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */
29/**************************************************************************/
30
31#ifndef WORKER_THREAD_POOL_H
32#define WORKER_THREAD_POOL_H
33
34#include "core/os/memory.h"
35#include "core/os/os.h"
36#include "core/os/semaphore.h"
37#include "core/os/thread.h"
38#include "core/templates/local_vector.h"
39#include "core/templates/paged_allocator.h"
40#include "core/templates/rid.h"
41#include "core/templates/safe_refcount.h"
42
43class WorkerThreadPool : public Object {
44 GDCLASS(WorkerThreadPool, Object)
45public:
46 enum {
47 INVALID_TASK_ID = -1
48 };
49
50 typedef int64_t TaskID;
51 typedef int64_t GroupID;
52
53private:
54 struct Task;
55
56 struct BaseTemplateUserdata {
57 virtual void callback() {}
58 virtual void callback_indexed(uint32_t p_index) {}
59 virtual ~BaseTemplateUserdata() {}
60 };
61
62 struct Group {
63 GroupID self;
64 SafeNumeric<uint32_t> index;
65 SafeNumeric<uint32_t> completed_index;
66 uint32_t max = 0;
67 Semaphore done_semaphore;
68 SafeFlag completed;
69 SafeNumeric<uint32_t> finished;
70 uint32_t tasks_used = 0;
71 TightLocalVector<Task *> low_priority_native_tasks;
72 };
73
74 struct Task {
75 Callable callable;
76 void (*native_func)(void *) = nullptr;
77 void (*native_group_func)(void *, uint32_t) = nullptr;
78 void *native_func_userdata = nullptr;
79 String description;
80 Semaphore done_semaphore;
81 bool completed = false;
82 Group *group = nullptr;
83 SelfList<Task> task_elem;
84 uint32_t waiting = 0;
85 bool low_priority = false;
86 BaseTemplateUserdata *template_userdata = nullptr;
87 Thread *low_priority_thread = nullptr;
88 int pool_thread_index = -1;
89
90 void free_template_userdata();
91 Task() :
92 task_elem(this) {}
93 };
94
95 PagedAllocator<Task> task_allocator;
96 PagedAllocator<Group> group_allocator;
97 PagedAllocator<Thread> native_thread_allocator;
98
99 SelfList<Task>::List low_priority_task_queue;
100 SelfList<Task>::List task_queue;
101
102 Mutex task_mutex;
103 Semaphore task_available_semaphore;
104
105 struct ThreadData {
106 uint32_t index;
107 Thread thread;
108 Task *current_low_prio_task = nullptr;
109 };
110
111 TightLocalVector<ThreadData> threads;
112 bool exit_threads = false;
113
114 HashMap<Thread::ID, int> thread_ids;
115 HashMap<TaskID, Task *> tasks;
116 HashMap<GroupID, Group *> groups;
117
118 bool use_native_low_priority_threads = false;
119 uint32_t max_low_priority_threads = 0;
120 uint32_t low_priority_threads_used = 0;
121 uint32_t low_priority_tasks_running = 0;
122 uint32_t low_priority_tasks_awaiting_others = 0;
123
124 uint64_t last_task = 1;
125
126 static void _thread_function(void *p_user);
127 static void _native_low_priority_thread_function(void *p_user);
128
129 void _process_task_queue();
130 void _process_task(Task *task);
131
132 void _post_task(Task *p_task, bool p_high_priority);
133
134 bool _try_promote_low_priority_task();
135 void _prevent_low_prio_saturation_deadlock();
136
137 static WorkerThreadPool *singleton;
138
139 TaskID _add_task(const Callable &p_callable, void (*p_func)(void *), void *p_userdata, BaseTemplateUserdata *p_template_userdata, bool p_high_priority, const String &p_description);
140 GroupID _add_group_task(const Callable &p_callable, void (*p_func)(void *, uint32_t), void *p_userdata, BaseTemplateUserdata *p_template_userdata, int p_elements, int p_tasks, bool p_high_priority, const String &p_description);
141
142 template <class C, class M, class U>
143 struct TaskUserData : public BaseTemplateUserdata {
144 C *instance;
145 M method;
146 U userdata;
147 virtual void callback() override {
148 (instance->*method)(userdata);
149 }
150 };
151
152 template <class C, class M, class U>
153 struct GroupUserData : public BaseTemplateUserdata {
154 C *instance;
155 M method;
156 U userdata;
157 virtual void callback_indexed(uint32_t p_index) override {
158 (instance->*method)(p_index, userdata);
159 }
160 };
161
162protected:
163 static void _bind_methods();
164
165public:
166 template <class C, class M, class U>
167 TaskID add_template_task(C *p_instance, M p_method, U p_userdata, bool p_high_priority = false, const String &p_description = String()) {
168 typedef TaskUserData<C, M, U> TUD;
169 TUD *ud = memnew(TUD);
170 ud->instance = p_instance;
171 ud->method = p_method;
172 ud->userdata = p_userdata;
173 return _add_task(Callable(), nullptr, nullptr, ud, p_high_priority, p_description);
174 }
175 TaskID add_native_task(void (*p_func)(void *), void *p_userdata, bool p_high_priority = false, const String &p_description = String());
176 TaskID add_task(const Callable &p_action, bool p_high_priority = false, const String &p_description = String());
177
178 bool is_task_completed(TaskID p_task_id) const;
179 Error wait_for_task_completion(TaskID p_task_id);
180
181 template <class C, class M, class U>
182 GroupID add_template_group_task(C *p_instance, M p_method, U p_userdata, int p_elements, int p_tasks = -1, bool p_high_priority = false, const String &p_description = String()) {
183 typedef GroupUserData<C, M, U> GroupUD;
184 GroupUD *ud = memnew(GroupUD);
185 ud->instance = p_instance;
186 ud->method = p_method;
187 ud->userdata = p_userdata;
188 return _add_group_task(Callable(), nullptr, nullptr, ud, p_elements, p_tasks, p_high_priority, p_description);
189 }
190 GroupID add_native_group_task(void (*p_func)(void *, uint32_t), void *p_userdata, int p_elements, int p_tasks = -1, bool p_high_priority = false, const String &p_description = String());
191 GroupID add_group_task(const Callable &p_action, int p_elements, int p_tasks = -1, bool p_high_priority = false, const String &p_description = String());
192 uint32_t get_group_processed_element_count(GroupID p_group) const;
193 bool is_group_task_completed(GroupID p_group) const;
194 void wait_for_group_task_completion(GroupID p_group);
195
196 _FORCE_INLINE_ int get_thread_count() const { return threads.size(); }
197
198 static WorkerThreadPool *get_singleton() { return singleton; }
199 void init(int p_thread_count = -1, bool p_use_native_threads_low_priority = true, float p_low_priority_task_ratio = 0.3);
200 void finish();
201 WorkerThreadPool();
202 ~WorkerThreadPool();
203};
204
205#endif // WORKER_THREAD_POOL_H
206