1 /* 2 Copyright (c) 2020-2021 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 #include "common/test.h" 18 19 #include "tbb/task_group.h" 20 #include "tbb/task_arena.h" 21 #include "tbb/global_control.h" 22 23 #include "common/spin_barrier.h" 24 #include "common/utils.h" 25 26 #include <cstddef> 27 #include <algorithm> 28 #include <numeric> 29 30 //! \file test_arena_priorities.cpp 31 //! \brief Test for [scheduler.task_arena] specification 32 33 //--------------------------------------------------// 34 35 std::vector<tbb::task_arena::priority> g_task_info; 36 37 std::atomic<unsigned> g_task_num; 38 39 std::atomic<bool> g_work_submitted; 40 41 namespace HighPriorityArenasTakeExecutionPrecedence { 42 43 using concurrency_type = int; 44 using arena_info = 45 std::tuple<std::unique_ptr<tbb::task_arena>, 46 concurrency_type, 47 tbb::task_arena::priority, 48 std::unique_ptr<tbb::task_group>>; 49 50 enum arena_info_keys { 51 arena_pointer, arena_concurrency, arena_priority, associated_task_group 52 }; 53 54 void prepare_logging_data(std::vector<tbb::task_arena::priority>& task_log, unsigned overall_tasks_num) { 55 task_log.clear(); 56 task_log.resize(overall_tasks_num); 57 for( auto& record : task_log ) 58 record = tbb::task_arena::priority::normal; 59 } 60 61 template<typename... ArenaArgs> 62 tbb::task_arena* do_allocate_and_construct( const ArenaArgs&... arena_args ) 63 { 64 const int dummy_max_concurrency = 4; 65 const int dummy_reserved_for_masters = 4; 66 67 enum initialization_methods { 68 lazy, 69 explicit_initialize, 70 explicit_initialize_with_different_constructor_parameters, 71 initialization_methods_num 72 }; 73 static initialization_methods initialization_method = lazy; 74 75 tbb::task_arena* result_arena = nullptr; 76 77 switch( initialization_method ) { 78 79 case lazy: 80 result_arena = new tbb::task_arena( arena_args... ); 81 break; 82 83 case explicit_initialize: 84 result_arena = new tbb::task_arena; 85 result_arena->initialize( arena_args... ); 86 break; 87 88 case explicit_initialize_with_different_constructor_parameters: 89 { 90 tbb::task_arena tmp(dummy_max_concurrency, dummy_reserved_for_masters); 91 result_arena = new tbb::task_arena(tmp); 92 result_arena->initialize(arena_args...); 93 break; 94 } 95 96 default: 97 REQUIRE_MESSAGE( false, "Not implemented method of initialization." ); 98 break; 99 } 100 101 int next_value = (initialization_method + 1) % initialization_methods_num; 102 initialization_method = (initialization_methods)next_value; 103 104 return result_arena; 105 } 106 107 template<typename FirstArenaArg> 108 tbb::task_arena* decide_on_arguments( 109 const FirstArenaArg& first_arg, const int reserved_for_masters, 110 tbb::task_arena::priority a_priority ) 111 { 112 const tbb::task_arena::priority default_priority = tbb::task_arena::priority::normal; 113 static bool pass_default_priority_implicitly = false; 114 if( default_priority == a_priority ) { 115 pass_default_priority_implicitly = !pass_default_priority_implicitly; 116 if( pass_default_priority_implicitly ) 117 return do_allocate_and_construct( first_arg, reserved_for_masters ); 118 } 119 return do_allocate_and_construct( first_arg, reserved_for_masters, a_priority ); 120 } 121 122 123 tbb::task_arena* allocate_and_construct_arena( 124 int arena_max_concurrency, tbb::task_arena::priority a_priority ) 125 { 126 const int reserved_for_masters = 0; 127 128 static bool use_constraints = false; 129 use_constraints = !use_constraints; 130 131 if( use_constraints ) { 132 tbb::task_arena::constraints properties{tbb::task_arena::automatic, arena_max_concurrency}; 133 return decide_on_arguments( properties, reserved_for_masters, a_priority ); 134 } 135 136 return decide_on_arguments( arena_max_concurrency, reserved_for_masters, a_priority ); 137 } 138 139 void submit_work( std::vector<arena_info>& arenas, unsigned repeats, utils::SpinBarrier& barrier ) { 140 for( auto& item : arenas ) { 141 tbb::task_arena& arena = *std::get<arena_pointer>(item).get(); 142 concurrency_type concurrency = std::get<arena_concurrency>(item); 143 tbb::task_arena::priority priority_value = std::get<arena_priority>(item); 144 auto& tg = std::get<associated_task_group>(item); 145 146 arena.execute( 147 [repeats, &barrier, &tg, priority_value, concurrency]() { 148 for( unsigned i = 0; i < repeats * concurrency; ++i ) { 149 tg->run( 150 [&barrier, priority_value](){ 151 while( !g_work_submitted.load(std::memory_order_acquire) ) 152 utils::yield(); 153 g_task_info[g_task_num++] = priority_value; 154 barrier.wait(); 155 } 156 ); 157 } 158 } // arena work submission functor 159 ); 160 } 161 } 162 163 void wait_work_completion( 164 std::vector<arena_info>& arenas, std::size_t max_num_threads, unsigned overall_tasks_num ) 165 { 166 if( max_num_threads > 1 ) 167 while( g_task_num < overall_tasks_num ) 168 utils::yield(); 169 170 for( auto& item : arenas ) { 171 tbb::task_arena& arena = *std::get<arena_pointer>(item).get(); 172 auto& tg = std::get<associated_task_group>(item); 173 arena.execute( [&tg]() { tg->wait(); } ); 174 } 175 CHECK_MESSAGE(g_task_num == overall_tasks_num, "Not all tasks were executed."); 176 } 177 178 void test() { 179 180 const std::size_t max_num_threads = tbb::global_control::active_value( 181 tbb::global_control::max_allowed_parallelism 182 ); 183 concurrency_type signed_max_num_threads = static_cast<int>(max_num_threads); 184 if( 1 == max_num_threads ) 185 // Skipping workerless case 186 return; 187 188 INFO( "max_num_threads = " << max_num_threads ); 189 // TODO: iterate over threads to see that the work is going on in low priority arena. 190 191 const int min_arena_concurrency = 2; // implementation detail 192 193 tbb::task_arena::priority high = tbb::task_arena::priority::high; 194 tbb::task_arena::priority normal = tbb::task_arena::priority::normal; 195 tbb::task_arena::priority low = tbb::task_arena::priority::low; 196 197 // TODO: use vector or std::array of priorities instead of the c-style array. 198 199 // TODO: consider extending priorities to have more than three arenas. 200 201 tbb::task_arena::priority priorities[] = {high, normal, low}; // keep it sorted 202 const unsigned priorities_num = sizeof(priorities) / sizeof(priorities[0]); 203 const unsigned overall_arenas_num = priorities_num; 204 205 std::vector<arena_info> arenas; 206 207 std::vector<unsigned> progressing_arenas( overall_arenas_num, 0 ); 208 std::iota( progressing_arenas.begin(), progressing_arenas.end(), 1 ); 209 210 for( const auto& progressing_arenas_num : progressing_arenas ) { 211 212 INFO( "progressing_arenas_num = " << progressing_arenas_num ); 213 214 // TODO: consider populating vector with arenas in separate function. 215 unsigned adjusted_progressing_arenas = progressing_arenas_num; 216 217 arenas.clear(); 218 g_task_num = 0; 219 220 concurrency_type projected_concurrency = 221 (signed_max_num_threads + progressing_arenas_num - 1) / progressing_arenas_num; 222 projected_concurrency = std::max(min_arena_concurrency, projected_concurrency); // implementation detail 223 adjusted_progressing_arenas = signed_max_num_threads / projected_concurrency; 224 225 int threads_left = signed_max_num_threads; 226 227 // Instantiate arenas with necessary concurrency so that progressing arenas consume all 228 // available threads. 229 for( unsigned arena_idx = 0; arena_idx < overall_arenas_num; ++arena_idx ) { 230 tbb::task_arena::priority a_priority = priorities[arena_idx]; 231 232 concurrency_type concurrency = projected_concurrency; 233 concurrency_type actual_concurrency = projected_concurrency; 234 if( threads_left < actual_concurrency || 235 arena_idx == adjusted_progressing_arenas - 1 ) // give all remaining threads to last progressing arena 236 { 237 concurrency = actual_concurrency = threads_left; 238 } 239 240 threads_left -= actual_concurrency; 241 242 if( !actual_concurrency ) { 243 concurrency = tbb::task_arena::automatic; 244 actual_concurrency = signed_max_num_threads; 245 } 246 actual_concurrency = std::max( min_arena_concurrency, actual_concurrency ); // implementation detail 247 248 tbb::task_arena* arena = allocate_and_construct_arena(concurrency, a_priority); 249 arenas.push_back( 250 std::make_tuple( 251 std::unique_ptr<tbb::task_arena>(arena), 252 actual_concurrency, 253 a_priority, 254 std::unique_ptr<tbb::task_group>(new tbb::task_group) 255 ) 256 ); 257 } 258 259 std::rotate( arenas.begin(), arenas.begin() + progressing_arenas_num - 1, arenas.end() ); 260 261 const unsigned repeats = 10; 262 263 unsigned overall_tasks_num = 0; 264 for( auto& item : arenas ) 265 overall_tasks_num += std::get<arena_concurrency>(item) * repeats; 266 267 prepare_logging_data( g_task_info, overall_tasks_num ); 268 269 g_work_submitted = false; 270 271 utils::SpinBarrier barrier{ max_num_threads }; 272 submit_work( arenas, repeats, barrier ); 273 274 g_work_submitted = true; 275 276 wait_work_completion( arenas, max_num_threads, overall_tasks_num ); 277 278 std::map<tbb::task_arena::priority, unsigned> wasted_tasks; 279 280 tbb::task_arena::priority* end_ptr = priorities + adjusted_progressing_arenas; 281 282 { 283 // First epoch - check progressing arenas only 284 unsigned overall_progressing_arenas_tasks_num = 0; 285 std::map<tbb::task_arena::priority, unsigned> per_priority_tasks_num; 286 287 // Due to indeterministic submission of tasks in the beginning, count tasks priorities up 288 // to additional epoch. Assume threads are rebalanced once the work is submitted. 289 unsigned last_task_idx = std::min((repeats + 1) * unsigned(max_num_threads), overall_tasks_num); 290 for( unsigned i = 0; i < last_task_idx; ++i ) { 291 tbb::task_arena::priority p = g_task_info[i]; 292 ++per_priority_tasks_num[p]; 293 294 overall_progressing_arenas_tasks_num += (int)( 295 end_ptr != std::find(priorities, end_ptr, p) 296 ); 297 298 if( i < max_num_threads || i >= repeats * max_num_threads ) 299 ++wasted_tasks[p]; 300 } 301 302 unsigned expected_overall_progressing_arenas_tasks_num = 0; 303 for( unsigned i = 0; i < adjusted_progressing_arenas; ++i ) { 304 tbb::task_arena::priority p = priorities[i]; 305 concurrency_type concurrency = 0; 306 for( auto& item : arenas ) { 307 if( std::get<arena_priority>(item) == p ) { 308 concurrency = std::get<arena_concurrency>(item); 309 break; 310 } 311 } 312 unsigned expected_tasks_num = repeats * concurrency; 313 314 CHECK_MESSAGE( expected_tasks_num == per_priority_tasks_num[p], 315 "Unexpected number of executed tasks in arena with index " << i << " and concurrency = " << concurrency ) ; 316 317 expected_overall_progressing_arenas_tasks_num += expected_tasks_num; 318 } 319 CHECK_MESSAGE( 320 expected_overall_progressing_arenas_tasks_num == overall_progressing_arenas_tasks_num, 321 "Number of tasks for progressing arenas mismatched." 322 ); 323 } 324 { 325 // Other epochs - check remaining arenas 326 std::map<tbb::task_arena::priority, unsigned> per_priority_tasks_num; 327 328 std::size_t lower_priority_start = (repeats + 1) * max_num_threads; 329 for( std::size_t i = lower_priority_start; i < overall_tasks_num; ++i ) 330 ++per_priority_tasks_num[ g_task_info[i] ]; 331 332 for( auto& e : per_priority_tasks_num ) { 333 auto priority = e.first; 334 auto tasks_num = e.second; 335 336 auto priorities_it = std::find( end_ptr, priorities + priorities_num, priority ); 337 CHECK_MESSAGE( priorities_it != priorities + priorities_num, 338 "Tasks from prioritized arena got deferred." ); 339 340 auto it = std::find_if( 341 arenas.begin(), arenas.end(), 342 [priority](arena_info& info) { 343 return std::get<arena_priority>(info) == priority; 344 } 345 ); 346 auto per_arena_tasks_num = repeats * std::get<arena_concurrency>(*it); 347 CHECK_MESSAGE( 348 tasks_num == per_arena_tasks_num - wasted_tasks[priority], 349 "Incorrect number of tasks from deferred (non-progressing) arenas were executed." 350 ); 351 } 352 } // Other epochs 353 } // loop over simultaneously progressing arenas 354 355 INFO( "Done\n" ); 356 } 357 358 } // namespace HighPriorityArenasTakeExecutionPrecedence 359 360 361 // TODO: nested arena case 362 363 //! Test for setting a priority to arena 364 //! \brief \ref requirement 365 TEST_CASE("Arena priorities") { 366 HighPriorityArenasTakeExecutionPrecedence::test(); 367 } 368