taskschedulerinternal.cpp 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406
  1. // Copyright 2009-2021 Intel Corporation
  2. // SPDX-License-Identifier: Apache-2.0
  3. #include "taskschedulerinternal.h"
  4. #include "../math/emath.h"
  5. #include "../sys/sysinfo.h"
  6. #include <algorithm>
  7. namespace embree
  8. {
  9. RTC_NAMESPACE_BEGIN
  10. static MutexSys g_mutex;
  11. size_t TaskScheduler::g_numThreads = 0;
  12. __thread TaskScheduler* TaskScheduler::g_instance = nullptr;
  13. std::vector<Ref<TaskScheduler>> g_instance_vector;
  14. __thread TaskScheduler::Thread* TaskScheduler::thread_local_thread = nullptr;
  15. TaskScheduler::ThreadPool* TaskScheduler::threadPool = nullptr;
  16. template<typename Predicate, typename Body>
  17. __forceinline void TaskScheduler::steal_loop(Thread& thread, const Predicate& pred, const Body& body)
  18. {
  19. while (true)
  20. {
  21. /*! some rounds that yield */
  22. for (size_t i=0; i<32; i++)
  23. {
  24. /*! some spinning rounds */
  25. const size_t threadCount = thread.threadCount();
  26. for (size_t j=0; j<1024; j+=threadCount)
  27. {
  28. if (!pred()) return;
  29. if (thread.scheduler->steal_from_other_threads(thread)) {
  30. i=j=0;
  31. body();
  32. }
  33. }
  34. yield();
  35. }
  36. }
  37. }
  38. /*! run this task */
  39. void TaskScheduler::Task::run_internal (Thread& thread) // FIXME: avoid as many dll_exports as possible
  40. {
  41. /* try to run if not already stolen */
  42. if (try_switch_state(INITIALIZED,DONE))
  43. {
  44. Task* prevTask = thread.task;
  45. thread.task = this;
  46. // -- GODOT start --
  47. // try {
  48. // if (context->cancellingException == nullptr)
  49. closure->execute();
  50. // } catch (...) {
  51. // if (context->cancellingException == nullptr)
  52. // context->cancellingException = std::current_exception();
  53. // }
  54. // -- GODOT end --
  55. thread.task = prevTask;
  56. add_dependencies(-1);
  57. }
  58. /* steal until all dependencies have completed */
  59. steal_loop(thread,
  60. [&] () { return dependencies>0; },
  61. [&] () { while (thread.tasks.execute_local_internal(thread,this)); });
  62. /* now signal our parent task that we are finished */
  63. if (parent)
  64. parent->add_dependencies(-1);
  65. }
  66. /*! run this task */
  67. dll_export void TaskScheduler::Task::run (Thread& thread) {
  68. run_internal(thread);
  69. }
  70. bool TaskScheduler::TaskQueue::execute_local_internal(Thread& thread, Task* parent)
  71. {
  72. /* stop if we run out of local tasks or reach the waiting task */
  73. if (right == 0 || &tasks[right-1] == parent)
  74. return false;
  75. /* execute task */
  76. size_t oldRight = right;
  77. tasks[right-1].run_internal(thread);
  78. if (right != oldRight) {
  79. THROW_RUNTIME_ERROR("you have to wait for spawned subtasks");
  80. }
  81. /* pop task and closure from stack */
  82. right--;
  83. if (tasks[right].stackPtr != size_t(-1))
  84. stackPtr = tasks[right].stackPtr;
  85. /* also move left pointer */
  86. if (left >= right) left.store(right.load());
  87. return right != 0;
  88. }
  89. dll_export bool TaskScheduler::TaskQueue::execute_local(Thread& thread, Task* parent) {
  90. return execute_local_internal(thread,parent);
  91. }
  92. bool TaskScheduler::TaskQueue::steal(Thread& thread)
  93. {
  94. size_t l = left;
  95. size_t r = right;
  96. if (l < r)
  97. {
  98. l = left++;
  99. if (l >= r)
  100. return false;
  101. }
  102. else
  103. return false;
  104. if (!tasks[l].try_steal(thread.tasks.tasks[thread.tasks.right]))
  105. return false;
  106. thread.tasks.right++;
  107. return true;
  108. }
  109. /* we steal from the left */
  110. size_t TaskScheduler::TaskQueue::getTaskSizeAtLeft()
  111. {
  112. if (left >= right) return 0;
  113. return tasks[left].N;
  114. }
  115. void threadPoolFunction(std::pair<TaskScheduler::ThreadPool*,size_t>* pair)
  116. {
  117. TaskScheduler::ThreadPool* pool = pair->first;
  118. size_t threadIndex = pair->second;
  119. delete pair;
  120. pool->thread_loop(threadIndex);
  121. }
  122. TaskScheduler::ThreadPool::ThreadPool(bool set_affinity)
  123. : numThreads(0), numThreadsRunning(0), set_affinity(set_affinity), running(false) {}
  124. dll_export void TaskScheduler::ThreadPool::startThreads()
  125. {
  126. if (running) return;
  127. setNumThreads(numThreads,true);
  128. }
  129. void TaskScheduler::ThreadPool::setNumThreads(size_t newNumThreads, bool startThreads)
  130. {
  131. Lock<MutexSys> lock(g_mutex);
  132. assert(newNumThreads);
  133. if (newNumThreads == std::numeric_limits<size_t>::max())
  134. newNumThreads = (size_t) getNumberOfLogicalThreads();
  135. numThreads = newNumThreads;
  136. if (!startThreads && !running) return;
  137. running = true;
  138. size_t numThreadsActive = numThreadsRunning;
  139. mutex.lock();
  140. numThreadsRunning = newNumThreads;
  141. mutex.unlock();
  142. condition.notify_all();
  143. /* start new threads */
  144. for (size_t t=numThreadsActive; t<numThreads; t++)
  145. {
  146. if (t == 0) continue;
  147. auto pair = new std::pair<TaskScheduler::ThreadPool*,size_t>(this,t);
  148. threads.push_back(createThread((thread_func)threadPoolFunction,pair,4*1024*1024,set_affinity ? t : -1));
  149. }
  150. /* stop some threads if we reduce the number of threads */
  151. for (ssize_t t=numThreadsActive-1; t>=ssize_t(numThreadsRunning); t--) {
  152. if (t == 0) continue;
  153. embree::join(threads.back());
  154. threads.pop_back();
  155. }
  156. }
  157. TaskScheduler::ThreadPool::~ThreadPool()
  158. {
  159. /* leave all taskschedulers */
  160. mutex.lock();
  161. numThreadsRunning = 0;
  162. mutex.unlock();
  163. condition.notify_all();
  164. /* wait for threads to terminate */
  165. for (size_t i=0; i<threads.size(); i++)
  166. embree::join(threads[i]);
  167. }
  168. dll_export void TaskScheduler::ThreadPool::add(const Ref<TaskScheduler>& scheduler)
  169. {
  170. mutex.lock();
  171. schedulers.push_back(scheduler);
  172. mutex.unlock();
  173. condition.notify_all();
  174. }
  175. dll_export void TaskScheduler::ThreadPool::remove(const Ref<TaskScheduler>& scheduler)
  176. {
  177. Lock<MutexSys> lock(mutex);
  178. for (std::list<Ref<TaskScheduler> >::iterator it = schedulers.begin(); it != schedulers.end(); it++) {
  179. if (scheduler == *it) {
  180. schedulers.erase(it);
  181. return;
  182. }
  183. }
  184. }
  185. void TaskScheduler::ThreadPool::thread_loop(size_t globalThreadIndex)
  186. {
  187. while (globalThreadIndex < numThreadsRunning)
  188. {
  189. Ref<TaskScheduler> scheduler = NULL;
  190. ssize_t threadIndex = -1;
  191. {
  192. Lock<MutexSys> lock(mutex);
  193. condition.wait(mutex, [&] () { return globalThreadIndex >= numThreadsRunning || !schedulers.empty(); });
  194. if (globalThreadIndex >= numThreadsRunning) break;
  195. scheduler = schedulers.front();
  196. threadIndex = scheduler->allocThreadIndex();
  197. }
  198. scheduler->thread_loop(threadIndex);
  199. }
  200. }
  201. TaskScheduler::TaskScheduler()
  202. : threadCounter(0), anyTasksRunning(0), hasRootTask(false)
  203. {
  204. assert(threadPool);
  205. threadLocal.resize(2 * TaskScheduler::threadCount()); // FIXME: this has to be 2x as in the compatibility join mode with rtcCommitScene the worker threads also join. When disallowing rtcCommitScene to join a build we can remove the 2x.
  206. for (size_t i=0; i<threadLocal.size(); i++)
  207. threadLocal[i].store(nullptr);
  208. }
  209. TaskScheduler::~TaskScheduler()
  210. {
  211. assert(threadCounter == 0);
  212. }
  213. dll_export size_t TaskScheduler::threadID()
  214. {
  215. Thread* thread = TaskScheduler::thread();
  216. if (thread) return thread->threadIndex;
  217. else return 0;
  218. }
  219. dll_export size_t TaskScheduler::threadIndex()
  220. {
  221. Thread* thread = TaskScheduler::thread();
  222. if (thread) return thread->threadIndex;
  223. else return 0;
  224. }
  225. dll_export size_t TaskScheduler::threadCount() {
  226. return threadPool->size();
  227. }
  228. dll_export TaskScheduler* TaskScheduler::instance()
  229. {
  230. if (g_instance == NULL) {
  231. Lock<MutexSys> lock(g_mutex);
  232. g_instance = new TaskScheduler;
  233. g_instance_vector.push_back(g_instance);
  234. }
  235. return g_instance;
  236. }
  237. void TaskScheduler::create(size_t numThreads, bool set_affinity, bool start_threads)
  238. {
  239. if (!threadPool) threadPool = new TaskScheduler::ThreadPool(set_affinity);
  240. threadPool->setNumThreads(numThreads,start_threads);
  241. }
  242. void TaskScheduler::destroy() {
  243. delete threadPool; threadPool = nullptr;
  244. }
  245. dll_export ssize_t TaskScheduler::allocThreadIndex()
  246. {
  247. size_t threadIndex = threadCounter++;
  248. assert(threadIndex < threadLocal.size());
  249. return threadIndex;
  250. }
  251. void TaskScheduler::join()
  252. {
  253. mutex.lock();
  254. size_t threadIndex = allocThreadIndex();
  255. condition.wait(mutex, [&] () { return hasRootTask.load(); });
  256. mutex.unlock();
  257. thread_loop(threadIndex);
  258. }
  259. void TaskScheduler::reset() {
  260. hasRootTask = false;
  261. }
  262. void TaskScheduler::wait_for_threads(size_t threadCount)
  263. {
  264. while (threadCounter < threadCount-1)
  265. pause_cpu();
  266. }
  267. dll_export TaskScheduler::Thread* TaskScheduler::thread() {
  268. return thread_local_thread;
  269. }
  270. dll_export TaskScheduler::Thread* TaskScheduler::swapThread(Thread* thread)
  271. {
  272. Thread* old = thread_local_thread;
  273. thread_local_thread = thread;
  274. return old;
  275. }
  276. dll_export void TaskScheduler::wait()
  277. {
  278. Thread* thread = TaskScheduler::thread();
  279. if (thread == nullptr)
  280. return;
  281. while (thread->tasks.execute_local_internal(*thread,thread->task)) {};
  282. }
  283. void TaskScheduler::thread_loop(size_t threadIndex)
  284. {
  285. /* allocate thread structure */
  286. std::unique_ptr<Thread> mthread(new Thread(threadIndex,this)); // too large for stack allocation
  287. Thread& thread = *mthread;
  288. threadLocal[threadIndex].store(&thread);
  289. Thread* oldThread = swapThread(&thread);
  290. /* main thread loop */
  291. while (anyTasksRunning)
  292. {
  293. steal_loop(thread,
  294. [&] () { return anyTasksRunning > 0; },
  295. [&] () {
  296. anyTasksRunning++;
  297. while (thread.tasks.execute_local_internal(thread,nullptr));
  298. anyTasksRunning--;
  299. });
  300. }
  301. threadLocal[threadIndex].store(nullptr);
  302. swapThread(oldThread);
  303. /* wait for all threads to terminate */
  304. threadCounter--;
  305. #if defined(__WIN32__)
  306. size_t loopIndex = 1;
  307. #endif
  308. #define LOOP_YIELD_THRESHOLD (4096)
  309. while (threadCounter > 0) {
  310. #if defined(__WIN32__)
  311. if ((loopIndex % LOOP_YIELD_THRESHOLD) == 0)
  312. yield();
  313. else
  314. _mm_pause();
  315. loopIndex++;
  316. #else
  317. yield();
  318. #endif
  319. }
  320. }
  321. bool TaskScheduler::steal_from_other_threads(Thread& thread)
  322. {
  323. const size_t threadIndex = thread.threadIndex;
  324. const size_t threadCount = this->threadCounter;
  325. for (size_t i=1; i<threadCount; i++)
  326. {
  327. pause_cpu(32);
  328. size_t otherThreadIndex = threadIndex+i;
  329. if (otherThreadIndex >= threadCount) otherThreadIndex -= threadCount;
  330. Thread* othread = threadLocal[otherThreadIndex].load();
  331. if (!othread)
  332. continue;
  333. if (othread->tasks.steal(thread))
  334. return true;
  335. }
  336. return false;
  337. }
  338. dll_export void TaskScheduler::startThreads() {
  339. threadPool->startThreads();
  340. }
  341. dll_export void TaskScheduler::addScheduler(const Ref<TaskScheduler>& scheduler) {
  342. threadPool->add(scheduler);
  343. }
  344. dll_export void TaskScheduler::removeScheduler(const Ref<TaskScheduler>& scheduler) {
  345. threadPool->remove(scheduler);
  346. }
  347. RTC_NAMESPACE_END
  348. }