xref: /oneTBB/src/tbb/waiters.h (revision 229b63ee)
1 /*
2     Copyright (c) 2005-2023 Intel Corporation
3 
4     Licensed under the Apache License, Version 2.0 (the "License");
5     you may not use this file except in compliance with the License.
6     You may obtain a copy of the License at
7 
8         http://www.apache.org/licenses/LICENSE-2.0
9 
10     Unless required by applicable law or agreed to in writing, software
11     distributed under the License is distributed on an "AS IS" BASIS,
12     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13     See the License for the specific language governing permissions and
14     limitations under the License.
15 */
16 
17 #ifndef _TBB_waiters_H
18 #define _TBB_waiters_H
19 
20 #include "oneapi/tbb/detail/_task.h"
21 #include "scheduler_common.h"
22 #include "arena.h"
23 #include "threading_control.h"
24 
25 namespace tbb {
26 namespace detail {
27 namespace r1 {
28 
29 inline d1::task* get_self_recall_task(arena_slot& slot);
30 
31 class waiter_base {
32 public:
33     waiter_base(arena& a, int yields_multiplier = 1) : my_arena(a), my_backoff(int(a.my_num_slots), yields_multiplier) {}
34 
35     bool pause() {
36         if (my_backoff.pause()) {
37             my_arena.out_of_work();
38             return true;
39         }
40 
41         return false;
42     }
43 
44     void reset_wait() {
45         my_backoff.reset_wait();
46     }
47 
48 protected:
49     arena& my_arena;
50     stealing_loop_backoff my_backoff;
51 };
52 
53 class outermost_worker_waiter : public waiter_base {
54 public:
55     using waiter_base::waiter_base;
56 
57     bool continue_execution(arena_slot& slot, d1::task*& t) const {
58         __TBB_ASSERT(t == nullptr, nullptr);
59 
60         if (is_worker_should_leave(slot)) {
61             // Leave dispatch loop
62             return false;
63         }
64 
65         t = get_self_recall_task(slot);
66         return true;
67     }
68 
69     void pause(arena_slot&) {
70         waiter_base::pause();
71     }
72 
73 
74     d1::wait_context* wait_ctx() {
75         return nullptr;
76     }
77 
78     static bool postpone_execution(d1::task&) {
79         return false;
80     }
81 
82 private:
83     using base_type = waiter_base;
84 
85     bool is_worker_should_leave(arena_slot& slot) const {
86         bool is_top_priority_arena = my_arena.is_top_priority();
87         bool is_task_pool_empty = slot.task_pool.load(std::memory_order_relaxed) == EmptyTaskPool;
88 
89         if (is_top_priority_arena) {
90             // Worker in most priority arena do not leave arena, until all work in task_pool is done
91             if (is_task_pool_empty && my_arena.is_recall_requested()) {
92                 return true;
93             }
94         } else {
95             if (my_arena.is_recall_requested()) {
96                 // If worker has work in task pool, we must notify other threads,
97                 // because can appear missed wake up of other threads
98                 if (!is_task_pool_empty) {
99                     my_arena.advertise_new_work<arena::wakeup>();
100                 }
101                 return true;
102             }
103         }
104 
105         return false;
106     }
107 };
108 
109 class sleep_waiter : public waiter_base {
110 protected:
111     using waiter_base::waiter_base;
112 
113     template <typename Pred>
114     void sleep(std::uintptr_t uniq_tag, Pred wakeup_condition) {
115         my_arena.get_waiting_threads_monitor().wait<thread_control_monitor::thread_context>(wakeup_condition,
116             market_context{uniq_tag, &my_arena});
117     }
118 };
119 
120 class external_waiter : public sleep_waiter {
121 public:
122     external_waiter(arena& a, d1::wait_context& wo)
123         : sleep_waiter(a, /*yields_multiplier*/10), my_wait_ctx(wo)
124         {}
125 
126     bool continue_execution(arena_slot& slot, d1::task*& t) const {
127         __TBB_ASSERT(t == nullptr, nullptr);
128         if (!my_wait_ctx.continue_execution())
129             return false;
130         t = get_self_recall_task(slot);
131         return true;
132     }
133 
134     void pause(arena_slot&) {
135         if (!sleep_waiter::pause()) {
136             return;
137         }
138 
139         auto wakeup_condition = [&] { return !my_arena.is_empty() || !my_wait_ctx.continue_execution(); };
140 
141         sleep(std::uintptr_t(&my_wait_ctx), wakeup_condition);
142         my_backoff.reset_wait();
143     }
144 
145     d1::wait_context* wait_ctx() {
146         return &my_wait_ctx;
147     }
148 
149     static bool postpone_execution(d1::task&) {
150         return false;
151     }
152 
153 private:
154     d1::wait_context& my_wait_ctx;
155 };
156 
157 #if __TBB_RESUMABLE_TASKS
158 
159 class coroutine_waiter : public sleep_waiter {
160 public:
161     using sleep_waiter::sleep_waiter;
162 
163     bool continue_execution(arena_slot& slot, d1::task*& t) const {
164         __TBB_ASSERT(t == nullptr, nullptr);
165         t = get_self_recall_task(slot);
166         return true;
167     }
168 
169     void pause(arena_slot& slot) {
170         if (!sleep_waiter::pause()) {
171             return;
172         }
173 
174         suspend_point_type* sp = slot.default_task_dispatcher().m_suspend_point;
175 
176         auto wakeup_condition = [&] { return !my_arena.is_empty() || sp->m_is_owner_recalled.load(std::memory_order_relaxed); };
177 
178         sleep(std::uintptr_t(sp), wakeup_condition);
179         my_backoff.reset_wait();
180     }
181 
182     void reset_wait() {
183         my_backoff.reset_wait();
184     }
185 
186     d1::wait_context* wait_ctx() {
187         return nullptr;
188     }
189 
190     static bool postpone_execution(d1::task& t) {
191         return task_accessor::is_resume_task(t);
192     }
193 };
194 
195 #endif // __TBB_RESUMABLE_TASKS
196 
197 } // namespace r1
198 } // namespace detail
199 } // namespace tbb
200 
201 #endif // _TBB_waiters_H
202