Skip to content

Commit b7ba892

Browse files
author
TSUNG-WEI HUANG
committed
added pause to optimize scan loop
1 parent fb289ca commit b7ba892

5 files changed

Lines changed: 144 additions & 58 deletions

File tree

‎taskflow/algorithm/scan.hpp‎

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,6 @@ struct ScanData {
4444
// Function: scan_loop
4545
template <typename S, typename Iterator, typename B>
4646
void scan_loop(
47-
tf::Runtime& rt,
4847
S& sdata,
4948
B bop,
5049
Iterator d_beg,
@@ -64,11 +63,13 @@ void scan_loop(
6463
if(w==0) {
6564
return;
6665
}
67-
68-
// need to do public corun because multiple workers can call this
69-
rt.executor().corun_until([&sdata](){
70-
return sdata.counter.load(std::memory_order_acquire) == 0;
71-
});
66+
67+
// simply do a loop until the counter becomes zero; we don't do corun
68+
// as the block scan is typically very fast, and stealing a task can cause
69+
// the worker to evict cached data from the block scan
70+
spin_until([&](){
71+
return sdata.counter.load(std::memory_order_acquire) == 0; }
72+
);
7273

7374
// block addup
7475
for(size_t i=0; i<block_size; i++) {
@@ -136,7 +137,7 @@ auto make_inclusive_scan_task(B first, E last, D d_first, BOP bop) {
136137
}
137138

138139
// block scan
139-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
140+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
140141
};
141142

142143
std::advance(s_beg, block_size);
@@ -207,7 +208,7 @@ auto make_inclusive_scan_task(B first, E last, D d_first, BOP bop, T init) {
207208
}
208209

209210
// block scan
210-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
211+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
211212
};
212213

213214
std::advance(s_beg, block_size);
@@ -280,7 +281,7 @@ auto make_transform_inclusive_scan_task(
280281
}
281282

282283
// block scan
283-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
284+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
284285
};
285286

286287
std::advance(s_beg, block_size);
@@ -352,7 +353,7 @@ auto make_transform_inclusive_scan_task(
352353
}
353354

354355
// block scan
355-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
356+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
356357
};
357358

358359
std::advance(s_beg, block_size);
@@ -436,7 +437,7 @@ auto make_exclusive_scan_task(
436437
*d_beg++ = local;
437438

438439
// block scan
439-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
440+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
440441
};
441442

442443
std::advance(s_beg, block_size);
@@ -523,7 +524,7 @@ auto make_transform_exclusive_scan_task(
523524
*d_beg++ = local;
524525

525526
// block scan
526-
detail::scan_loop(rt, *scan_data, bop, result, W, w, block_size);
527+
detail::scan_loop(*scan_data, bop, result, W, w, block_size);
527528
};
528529

529530
std::advance(s_beg, block_size);

‎taskflow/core/executor.hpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1717,9 +1717,9 @@ inline void Executor::_process_exception(Worker&, Node* node) {
17171717
}
17181718
}
17191719

1720-
// TODO: for now, we simply store the exception in this node; this can happen
1721-
// for execution that doesn't have any external control to capture the exception
1722-
// (e.g., silent async)
1720+
// for now, we simply store the exception in this node; this can happen in an
1721+
// execution that does not have any external control to capture the exception,
1722+
// such as silent async task
17231723
node->_exception_ptr = std::current_exception();
17241724
}
17251725

‎taskflow/core/tsq.hpp‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,22 +9,22 @@
99
*/
1010

1111
#ifndef TF_DEFAULT_BOUNDED_TASK_QUEUE_LOG_SIZE
12-
/**
13-
@def TF_DEFAULT_BOUNDED_TASK_QUEUE_LOG_SIZE
14-
15-
This macro defines the default size of the bounded task queue in Log2.
16-
Bounded task queue is used by each worker.
17-
*/
12+
/**
13+
@def TF_DEFAULT_BOUNDED_TASK_QUEUE_LOG_SIZE
14+
15+
This macro defines the default size of the bounded task queue in Log2.
16+
Bounded task queue is used by each worker.
17+
*/
1818
#define TF_DEFAULT_BOUNDED_TASK_QUEUE_LOG_SIZE 8
1919
#endif
2020

2121
#ifndef TF_DEFAULT_UNBOUNDED_TASK_QUEUE_LOG_SIZE
22-
/**
23-
@def TF_DEFAULT_UNBOUNDED_TASK_QUEUE_LOG_SIZE
24-
25-
This macro defines the default size of the unbounded task queue in Log2.
26-
Unbounded task queue is used by the executor.
27-
*/
22+
/**
23+
@def TF_DEFAULT_UNBOUNDED_TASK_QUEUE_LOG_SIZE
24+
25+
This macro defines the default size of the unbounded task queue in Log2.
26+
Unbounded task queue is used by the executor.
27+
*/
2828
#define TF_DEFAULT_UNBOUNDED_TASK_QUEUE_LOG_SIZE 10
2929
#endif
3030

‎taskflow/utility/os.hpp‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,78 @@ inline bool has_env(const std::string& str) {
182182
//#endif
183183
//}
184184

185+
/**
186+
@fn pause
187+
188+
The `pause()` instruction is used in spin-wait loops to hint the CPU that the current
189+
thread is in a busy-wait state.
190+
It helps reduce power consumption and improves performance on hyper-threaded processors
191+
by preventing the CPU from consuming unnecessary cycles while waiting.
192+
It is particularly useful in low-contention scenarios, where the thread
193+
is likely to quickly acquire the lock or condition it's waiting for,
194+
avoiding an expensive context switch.
195+
On modern x86 processors, this instruction can be invoked using @c __builtin_ia32_pause()
196+
in GCC/Clang or @c _mm_pause() in MSVC.
197+
In non-x86 architectures, alternative mechanisms such as yielding the CPU may be used instead.
198+
199+
*/
200+
inline void pause() {
201+
#if defined(__x86_64__) || defined(_M_X64) || defined(__i386__) || defined(_M_IX86)
202+
// x86 and x86_64: Use the PAUSE instruction
203+
#if defined(_MSC_VER)
204+
// Microsoft Visual C++
205+
_mm_pause();
206+
#elif defined(__GNUC__) || defined(__clang__)
207+
// GCC and Clang
208+
__builtin_ia32_pause();
209+
#else
210+
asm volatile("pause" ::: "memory");
211+
#endif
212+
213+
#elif defined(__aarch64__) || defined(__arm__)
214+
// ARM and AArch64: Use the YIELD instruction
215+
#if defined(__GNUC__) || defined(__clang__)
216+
asm volatile("yield" ::: "memory");
217+
#endif
185218

219+
#else
220+
// Fallback: Portable yield for unknown architectures
221+
std::this_thread::yield();
222+
#endif
223+
}
224+
225+
/**
226+
@brief spins until the given predicate becomes true
227+
228+
@tparam P the type of the predicate function or callable.
229+
@param predicate the callable that returns a boolean value, which is checked in the loop.
230+
231+
This function repeatedly checks the provided predicate in a spin-wait loop
232+
and uses a backoff strategy to minimize CPU waste during the wait. Initially,
233+
it uses the `pause()` instruction for the first 100 iterations to hint to the
234+
CPU that the thread is waiting, thus reducing power consumption and avoiding
235+
unnecessary cycles. After 100 iterations, it switches to yielding the CPU using
236+
`std::this_thread::yield()` to allow other threads to run and improve system
237+
responsiveness.
238+
239+
The function operates as follows:
240+
1. For the first 100 iterations, it invokes `pause()` to reduce power consumption
241+
during the spin-wait.
242+
2. After 100 iterations, it uses `std::this_thread::yield()` to relinquish the
243+
CPU, allowing other threads to execute.
244+
245+
@note This function is useful when you need to wait for a condition to be true, but
246+
want to optimize CPU usage during the wait by using a busy-wait approach with
247+
backoff.
248+
249+
*/
250+
template <typename P>
251+
void spin_until(P&& predicate) {
252+
size_t num_pauses = 0;
253+
while(!predicate()) {
254+
(num_pauses++ < 100) ? pause() : std::this_thread::yield();
255+
}
256+
}
186257

187258
} // end of namespace tf -----------------------------------------------------
188259

‎unittests/test_exceptions.cpp‎

Lines changed: 45 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -965,7 +965,7 @@ TEST_CASE("Exception.ModuleTask.4threads" * doctest::timeout(300)) {
965965
}
966966

967967
// ----------------------------------------------------------------------------
968-
// Exception.AsyncTask
968+
// Exception.Async
969969
// ----------------------------------------------------------------------------
970970

971971
void async_task(unsigned W) {
@@ -983,35 +983,35 @@ void async_task(unsigned W) {
983983
});
984984
REQUIRE_THROWS_WITH_AS(fu2.get(), "x", std::runtime_error);
985985

986-
//// exception is caught without any action
987-
//executor.silent_async([](){
988-
// throw std::runtime_error("y");
989-
//});
986+
// exception is caught without any action
987+
executor.silent_async([](){
988+
throw std::runtime_error("y");
989+
});
990990

991-
//executor.wait_for_all();
991+
executor.wait_for_all();
992992
}
993993

994-
TEST_CASE("Exception.AsyncTask.1thread" * doctest::timeout(300)) {
994+
TEST_CASE("Exception.Async.1thread" * doctest::timeout(300)) {
995995
async_task(1);
996996
}
997997

998-
TEST_CASE("Exception.AsyncTask.2threads" * doctest::timeout(300)) {
998+
TEST_CASE("Exception.Async.2threads" * doctest::timeout(300)) {
999999
async_task(2);
10001000
}
10011001

1002-
TEST_CASE("Exception.AsyncTask.3threads" * doctest::timeout(300)) {
1002+
TEST_CASE("Exception.Async.3threads" * doctest::timeout(300)) {
10031003
async_task(3);
10041004
}
10051005

1006-
TEST_CASE("Exception.AsyncTask.4threads" * doctest::timeout(300)) {
1006+
TEST_CASE("Exception.Async.4threads" * doctest::timeout(300)) {
10071007
async_task(4);
10081008
}
10091009

10101010
// ----------------------------------------------------------------------------
10111011
// Async Task with Runtime
10121012
// ----------------------------------------------------------------------------
10131013

1014-
void async_task_with_runtime(unsigned W) {
1014+
void async_with_runtime(unsigned W) {
10151015

10161016
tf::Executor executor(W);
10171017
std::vector<std::future<void>> futures;
@@ -1025,29 +1025,36 @@ void async_task_with_runtime(unsigned W) {
10251025
for(auto& fu : futures) {
10261026
REQUIRE_THROWS_WITH_AS(fu.get(), "x", std::runtime_error);
10271027
}
1028+
1029+
// silently caught by the task
1030+
executor.silent_async([](tf::Runtime&){
1031+
throw std::runtime_error("x");
1032+
});
1033+
1034+
executor.wait_for_all();
10281035
}
10291036

1030-
TEST_CASE("Exception.AsyncTaskWithRuntime.1thread" * doctest::timeout(300)) {
1031-
async_task_with_runtime(1);
1037+
TEST_CASE("Exception.Async.Runtime.1thread" * doctest::timeout(300)) {
1038+
async_with_runtime(1);
10321039
}
10331040

1034-
TEST_CASE("Exception.AsyncTaskWithRuntime.2threads" * doctest::timeout(300)) {
1035-
async_task_with_runtime(2);
1041+
TEST_CASE("Exception.Async.Runtime.2threads" * doctest::timeout(300)) {
1042+
async_with_runtime(2);
10361043
}
10371044

1038-
TEST_CASE("Exception.AsyncTaskWithRuntime.3threads" * doctest::timeout(300)) {
1039-
async_task_with_runtime(3);
1045+
TEST_CASE("Exception.Async.Runtime.3threads" * doctest::timeout(300)) {
1046+
async_with_runtime(3);
10401047
}
10411048

1042-
TEST_CASE("Exception.AsyncTaskWithRuntime.4threads" * doctest::timeout(300)) {
1043-
async_task_with_runtime(4);
1049+
TEST_CASE("Exception.Async.Runtime.4threads" * doctest::timeout(300)) {
1050+
async_with_runtime(4);
10441051
}
10451052

10461053
// ----------------------------------------------------------------------------
10471054
// Dependent Async Task with Runtime
10481055
// ----------------------------------------------------------------------------
10491056

1050-
void dependent_async_task_with_runtime(unsigned W) {
1057+
void dependent_async_with_runtime(unsigned W) {
10511058

10521059
tf::Executor executor(W);
10531060
std::vector<std::future<void>> futures;
@@ -1061,22 +1068,29 @@ void dependent_async_task_with_runtime(unsigned W) {
10611068
for(auto& fu : futures) {
10621069
REQUIRE_THROWS_WITH_AS(fu.get(), "x", std::runtime_error);
10631070
}
1071+
1072+
// silently caught by the task
1073+
executor.silent_dependent_async([](tf::Runtime&){
1074+
throw std::runtime_error("x");
1075+
});
1076+
1077+
executor.wait_for_all();
10641078
}
10651079

1066-
TEST_CASE("Exception.DependentAsyncTaskWithRuntime.1thread" * doctest::timeout(300)) {
1067-
dependent_async_task_with_runtime(1);
1080+
TEST_CASE("Exception.DependentAsync.Runtime.1thread" * doctest::timeout(300)) {
1081+
dependent_async_with_runtime(1);
10681082
}
10691083

1070-
TEST_CASE("Exception.DependentAsyncTaskWithRuntime.2threads" * doctest::timeout(300)) {
1071-
dependent_async_task_with_runtime(2);
1084+
TEST_CASE("Exception.DependentAsync.Runtime.2threads" * doctest::timeout(300)) {
1085+
dependent_async_with_runtime(2);
10721086
}
10731087

1074-
TEST_CASE("Exception.DependentAsyncTaskWithRuntime.3threads" * doctest::timeout(300)) {
1075-
dependent_async_task_with_runtime(3);
1088+
TEST_CASE("Exception.DependentAsync.Runtime.3threads" * doctest::timeout(300)) {
1089+
dependent_async_with_runtime(3);
10761090
}
10771091

1078-
TEST_CASE("Exception.DependentAsyncTaskWithRuntime.4threads" * doctest::timeout(300)) {
1079-
dependent_async_task_with_runtime(4);
1092+
TEST_CASE("Exception.DependentAsync.Runtime.4threads" * doctest::timeout(300)) {
1093+
dependent_async_with_runtime(4);
10801094
}
10811095

10821096
/*
@@ -1137,15 +1151,15 @@ void runtime_async_task(unsigned W) {
11371151
REQUIRE(flag == 0);
11381152
}
11391153
1140-
TEST_CASE("Exception.RuntimeAsyncTask.2threads" * doctest::timeout(300)) {
1154+
TEST_CASE("Exception.RuntimeAsync.2threads" * doctest::timeout(300)) {
11411155
runtime_async_task(2);
11421156
}
11431157
1144-
TEST_CASE("Exception.RuntimeAsyncTask.3threads" * doctest::timeout(300)) {
1158+
TEST_CASE("Exception.RuntimeAsync.3threads" * doctest::timeout(300)) {
11451159
runtime_async_task(3);
11461160
}
11471161
1148-
TEST_CASE("Exception.RuntimeAsyncTask.4threads" * doctest::timeout(300)) {
1162+
TEST_CASE("Exception.RuntimeAsync.4threads" * doctest::timeout(300)) {
11491163
runtime_async_task(4);
11501164
}
11511165
*/

0 commit comments

Comments
 (0)