1 /* SPDX-License-Identifier: BSD-3-Clause 2 * Copyright (C) 2020 Intel Corporation. 3 * All rights reserved. 4 */ 5 6 #include "spdk/stdinc.h" 7 8 #include "spdk_cunit.h" 9 #include "common/lib/test_env.c" 10 #include "event/reactor.c" 11 #include "spdk/thread.h" 12 #include "spdk_internal/thread.h" 13 #include "event/scheduler_static.c" 14 #include "../module/scheduler/dynamic/scheduler_dynamic.c" 15 16 static void 17 test_create_reactor(void) 18 { 19 /* See SPDK issue #3004. Seems like a bug with gcc + asan on Fedora 38, so 20 * we need to explicitly align the variable here. 21 */ 22 struct spdk_reactor reactor __attribute__((aligned(SPDK_CACHE_LINE_SIZE))) = {}; 23 24 g_reactors = &reactor; 25 g_reactor_count = 1; 26 27 reactor_construct(&reactor, 0); 28 29 CU_ASSERT(spdk_reactor_get(0) == &reactor); 30 31 spdk_ring_free(reactor.events); 32 reactor_interrupt_fini(&reactor); 33 g_reactors = NULL; 34 } 35 36 static void 37 test_init_reactors(void) 38 { 39 uint32_t core; 40 41 MOCK_SET(spdk_env_get_current_core, 0); 42 43 allocate_cores(3); 44 45 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 46 47 CU_ASSERT(g_reactor_state == SPDK_REACTOR_STATE_INITIALIZED); 48 for (core = 0; core < 3; core++) { 49 CU_ASSERT(spdk_reactor_get(core) != NULL); 50 } 51 52 spdk_reactors_fini(); 53 54 free_cores(); 55 56 MOCK_CLEAR(spdk_env_get_current_core); 57 } 58 59 static void 60 ut_event_fn(void *arg1, void *arg2) 61 { 62 uint8_t *test1 = arg1; 63 uint8_t *test2 = arg2; 64 65 *test1 = 1; 66 *test2 = 0xFF; 67 } 68 69 static void 70 test_event_call(void) 71 { 72 uint8_t test1 = 0, test2 = 0; 73 struct spdk_event *evt; 74 struct spdk_reactor *reactor; 75 76 MOCK_SET(spdk_env_get_current_core, 0); 77 78 allocate_cores(1); 79 80 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 81 82 evt = spdk_event_allocate(0, ut_event_fn, &test1, &test2); 83 CU_ASSERT(evt != NULL); 84 85 MOCK_SET(spdk_env_get_current_core, 0); 86 87 spdk_event_call(evt); 88 89 reactor = spdk_reactor_get(0); 90 CU_ASSERT(reactor != NULL); 91 92 CU_ASSERT(event_queue_run_batch(reactor) == 1); 93 CU_ASSERT(test1 == 1); 94 CU_ASSERT(test2 == 0xFF); 95 96 MOCK_CLEAR(spdk_env_get_current_core); 97 98 spdk_reactors_fini(); 99 100 free_cores(); 101 102 MOCK_CLEAR(spdk_env_get_current_core); 103 } 104 105 static void 106 test_schedule_thread(void) 107 { 108 struct spdk_cpuset cpuset = {}; 109 struct spdk_thread *thread; 110 struct spdk_reactor *reactor; 111 struct spdk_lw_thread *lw_thread; 112 113 MOCK_SET(spdk_env_get_current_core, 0); 114 115 allocate_cores(5); 116 117 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 118 119 spdk_cpuset_set_cpu(&cpuset, 3, true); 120 g_next_core = 4; 121 122 MOCK_SET(spdk_env_get_current_core, 3); 123 124 /* _reactor_schedule_thread() will be called in spdk_thread_create() 125 * at its end because it is passed to SPDK thread library by 126 * spdk_thread_lib_init(). 127 */ 128 thread = spdk_thread_create(NULL, &cpuset); 129 CU_ASSERT(thread != NULL); 130 131 reactor = spdk_reactor_get(3); 132 CU_ASSERT(reactor != NULL); 133 134 CU_ASSERT(event_queue_run_batch(reactor) == 1); 135 136 MOCK_CLEAR(spdk_env_get_current_core); 137 138 lw_thread = TAILQ_FIRST(&reactor->threads); 139 CU_ASSERT(lw_thread != NULL); 140 CU_ASSERT(spdk_thread_get_from_ctx(lw_thread) == thread); 141 142 TAILQ_REMOVE(&reactor->threads, lw_thread, link); 143 reactor->thread_count--; 144 spdk_set_thread(thread); 145 spdk_thread_exit(thread); 146 while (!spdk_thread_is_exited(thread)) { 147 spdk_thread_poll(thread, 0, 0); 148 } 149 spdk_thread_destroy(thread); 150 spdk_set_thread(NULL); 151 152 spdk_reactors_fini(); 153 154 free_cores(); 155 } 156 157 static void 158 test_reschedule_thread(void) 159 { 160 struct spdk_cpuset cpuset = {}; 161 struct spdk_thread *thread; 162 struct spdk_reactor *reactor; 163 struct spdk_lw_thread *lw_thread; 164 165 MOCK_SET(spdk_env_get_current_core, 0); 166 167 allocate_cores(3); 168 169 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 170 171 spdk_cpuset_set_cpu(&g_reactor_core_mask, 0, true); 172 spdk_cpuset_set_cpu(&g_reactor_core_mask, 1, true); 173 spdk_cpuset_set_cpu(&g_reactor_core_mask, 2, true); 174 g_next_core = 0; 175 176 MOCK_SET(spdk_env_get_current_core, 1); 177 /* Create and schedule the thread to core 1. */ 178 spdk_cpuset_set_cpu(&cpuset, 1, true); 179 180 thread = spdk_thread_create(NULL, &cpuset); 181 CU_ASSERT(thread != NULL); 182 lw_thread = spdk_thread_get_ctx(thread); 183 184 reactor = spdk_reactor_get(1); 185 CU_ASSERT(reactor != NULL); 186 187 CU_ASSERT(event_queue_run_batch(reactor) == 1); 188 CU_ASSERT(TAILQ_FIRST(&reactor->threads) == lw_thread); 189 190 spdk_set_thread(thread); 191 192 /* Call spdk_thread_set_cpumask() twice with different cpumask values. 193 * The cpumask of the 2nd call will be used in reschedule operation. 194 */ 195 196 spdk_cpuset_zero(&cpuset); 197 spdk_cpuset_set_cpu(&cpuset, 0, true); 198 CU_ASSERT(spdk_thread_set_cpumask(&cpuset) == 0); 199 200 spdk_cpuset_zero(&cpuset); 201 spdk_cpuset_set_cpu(&cpuset, 2, true); 202 CU_ASSERT(spdk_thread_set_cpumask(&cpuset) == 0); 203 204 CU_ASSERT(lw_thread->resched == true); 205 206 reactor_run(reactor); 207 208 CU_ASSERT(lw_thread->resched == false); 209 CU_ASSERT(TAILQ_EMPTY(&reactor->threads)); 210 211 reactor = spdk_reactor_get(0); 212 CU_ASSERT(reactor != NULL); 213 MOCK_SET(spdk_env_get_current_core, 0); 214 215 CU_ASSERT(event_queue_run_batch(reactor) == 0); 216 217 reactor = spdk_reactor_get(2); 218 CU_ASSERT(reactor != NULL); 219 MOCK_SET(spdk_env_get_current_core, 2); 220 221 CU_ASSERT(event_queue_run_batch(reactor) == 1); 222 223 CU_ASSERT(TAILQ_FIRST(&reactor->threads) == lw_thread); 224 225 MOCK_CLEAR(spdk_env_get_current_core); 226 227 TAILQ_REMOVE(&reactor->threads, lw_thread, link); 228 reactor->thread_count--; 229 spdk_set_thread(thread); 230 spdk_thread_exit(thread); 231 while (!spdk_thread_is_exited(thread)) { 232 spdk_thread_poll(thread, 0, 0); 233 } 234 spdk_thread_destroy(thread); 235 spdk_set_thread(NULL); 236 237 spdk_reactors_fini(); 238 239 free_cores(); 240 } 241 242 static void 243 for_each_reactor_done(void *arg1, void *arg2) 244 { 245 uint32_t *count = arg1; 246 bool *done = arg2; 247 248 (*count)++; 249 *done = true; 250 } 251 252 static void 253 for_each_reactor_cb(void *arg1, void *arg2) 254 { 255 uint32_t *count = arg1; 256 257 (*count)++; 258 } 259 260 static void 261 test_for_each_reactor(void) 262 { 263 uint32_t count = 0, i; 264 bool done = false; 265 struct spdk_reactor *reactor; 266 267 MOCK_SET(spdk_env_get_current_core, 0); 268 269 allocate_cores(5); 270 271 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 272 273 spdk_for_each_reactor(for_each_reactor_cb, &count, &done, for_each_reactor_done); 274 275 MOCK_CLEAR(spdk_env_get_current_core); 276 277 /* We have not processed any event yet, so count and done should be 0 and false, 278 * respectively. 279 */ 280 CU_ASSERT(count == 0); 281 282 /* Poll each reactor to verify the event is passed to each */ 283 for (i = 0; i < 5; i++) { 284 reactor = spdk_reactor_get(i); 285 CU_ASSERT(reactor != NULL); 286 MOCK_SET(spdk_env_get_current_core, i); 287 288 event_queue_run_batch(reactor); 289 CU_ASSERT(count == (i + 1)); 290 CU_ASSERT(done == false); 291 MOCK_CLEAR(spdk_env_get_current_core); 292 } 293 294 MOCK_SET(spdk_env_get_current_core, 0); 295 /* After each reactor is called, the completion calls it one more time. */ 296 reactor = spdk_reactor_get(0); 297 CU_ASSERT(reactor != NULL); 298 299 event_queue_run_batch(reactor); 300 CU_ASSERT(count == 6); 301 CU_ASSERT(done == true); 302 MOCK_CLEAR(spdk_env_get_current_core); 303 304 spdk_reactors_fini(); 305 306 free_cores(); 307 } 308 309 static int 310 poller_run_idle(void *ctx) 311 { 312 uint64_t delay_us = (uint64_t)ctx; 313 314 spdk_delay_us(delay_us); 315 316 return 0; 317 } 318 319 static int 320 poller_run_busy(void *ctx) 321 { 322 uint64_t delay_us = (uint64_t)ctx; 323 324 spdk_delay_us(delay_us); 325 326 return 1; 327 } 328 329 static void 330 test_reactor_stats(void) 331 { 332 struct spdk_cpuset cpuset = {}; 333 struct spdk_thread *thread1, *thread2; 334 struct spdk_reactor *reactor; 335 struct spdk_poller *busy1, *idle1, *busy2, *idle2; 336 struct spdk_thread_stats stats; 337 int rc __attribute__((unused)); 338 339 /* Test case is the following: 340 * Create a reactor on CPU core0. 341 * Create thread1 and thread2 simultaneously on reactor0 at TSC = 100. 342 * Reactor runs 343 * - thread1 for 100 with busy 344 * - thread2 for 200 with idle 345 * - thread1 for 300 with idle 346 * - thread2 for 400 with busy. 347 * Then, 348 * - both elapsed TSC of thread1 and thread2 should be 1100 (= 100 + 1000). 349 * - busy TSC of reactor should be 500 (= 100 + 400). 350 * - idle TSC of reactor should be 500 (= 200 + 300). 351 * 352 * After that reactor0 runs with no threads for 900 TSC. 353 * Create thread1 on reactor0 at TSC = 2000. 354 * Reactor runs 355 * - thread1 for 100 with busy 356 * Then, 357 * - elapsed TSC of thread1 should be 2100 (= 2000+ 100). 358 * - busy TSC of reactor should be 600 (= 500 + 100). 359 * - idle TSC of reactor should be 500 (= 500 + 900). 360 */ 361 362 MOCK_SET(spdk_env_get_current_core, 0); 363 364 allocate_cores(1); 365 366 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 367 368 spdk_cpuset_set_cpu(&cpuset, 0, true); 369 370 reactor = spdk_reactor_get(0); 371 SPDK_CU_ASSERT_FATAL(reactor != NULL); 372 373 /* First reactor_run() sets the tsc_last. */ 374 MOCK_SET(spdk_get_ticks, 100); 375 reactor->tsc_last = spdk_get_ticks(); 376 377 thread1 = spdk_thread_create(NULL, &cpuset); 378 SPDK_CU_ASSERT_FATAL(thread1 != NULL); 379 380 thread2 = spdk_thread_create(NULL, &cpuset); 381 SPDK_CU_ASSERT_FATAL(thread2 != NULL); 382 383 spdk_set_thread(thread1); 384 busy1 = spdk_poller_register(poller_run_busy, (void *)100, 0); 385 CU_ASSERT(busy1 != NULL); 386 387 spdk_set_thread(thread2); 388 idle2 = spdk_poller_register(poller_run_idle, (void *)300, 0); 389 CU_ASSERT(idle2 != NULL); 390 391 _reactor_run(reactor); 392 393 spdk_set_thread(thread1); 394 CU_ASSERT(spdk_thread_get_last_tsc(thread1) == 200); 395 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 396 CU_ASSERT(stats.busy_tsc == 100); 397 CU_ASSERT(stats.idle_tsc == 0); 398 spdk_set_thread(thread2); 399 CU_ASSERT(spdk_thread_get_last_tsc(thread2) == 500); 400 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 401 CU_ASSERT(stats.busy_tsc == 0); 402 CU_ASSERT(stats.idle_tsc == 300); 403 404 CU_ASSERT(reactor->busy_tsc == 100); 405 CU_ASSERT(reactor->idle_tsc == 300); 406 407 /* 100 + 100 + 300 = 500 ticks elapsed */ 408 CU_ASSERT(reactor->tsc_last == 500); 409 410 spdk_set_thread(thread1); 411 spdk_poller_unregister(&busy1); 412 idle1 = spdk_poller_register(poller_run_idle, (void *)200, 0); 413 CU_ASSERT(idle1 != NULL); 414 415 spdk_set_thread(thread2); 416 spdk_poller_unregister(&idle2); 417 busy2 = spdk_poller_register(poller_run_busy, (void *)400, 0); 418 CU_ASSERT(busy2 != NULL); 419 420 _reactor_run(reactor); 421 422 spdk_set_thread(thread1); 423 CU_ASSERT(spdk_thread_get_last_tsc(thread1) == 700); 424 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 425 CU_ASSERT(stats.busy_tsc == 100); 426 CU_ASSERT(stats.idle_tsc == 200); 427 spdk_set_thread(thread2); 428 CU_ASSERT(spdk_thread_get_last_tsc(thread2) == 1100); 429 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 430 CU_ASSERT(stats.busy_tsc == 400); 431 CU_ASSERT(stats.idle_tsc == 300); 432 433 CU_ASSERT(reactor->busy_tsc == 500); 434 CU_ASSERT(reactor->idle_tsc == 500); 435 436 /* 500 + 200 + 400 = 1100 ticks elapsed */ 437 CU_ASSERT(reactor->tsc_last == 1100); 438 439 spdk_set_thread(thread1); 440 spdk_poller_unregister(&idle1); 441 spdk_thread_exit(thread1); 442 443 spdk_set_thread(thread2); 444 spdk_poller_unregister(&busy2); 445 spdk_thread_exit(thread2); 446 447 _reactor_run(reactor); 448 449 /* After 900 ticks new thread is created. */ 450 /* 1100 + 900 = 2000 ticks elapsed */ 451 MOCK_SET(spdk_get_ticks, 2000); 452 _reactor_run(reactor); 453 CU_ASSERT(reactor->tsc_last == 2000); 454 455 thread1 = spdk_thread_create(NULL, &cpuset); 456 SPDK_CU_ASSERT_FATAL(thread1 != NULL); 457 458 spdk_set_thread(thread1); 459 busy1 = spdk_poller_register(poller_run_busy, (void *)100, 0); 460 CU_ASSERT(busy1 != NULL); 461 462 _reactor_run(reactor); 463 464 spdk_set_thread(thread1); 465 CU_ASSERT(spdk_thread_get_last_tsc(thread1) == 2100); 466 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 467 CU_ASSERT(stats.busy_tsc == 100); 468 CU_ASSERT(stats.idle_tsc == 0); 469 470 CU_ASSERT(reactor->busy_tsc == 600); 471 CU_ASSERT(reactor->idle_tsc == 1400); 472 473 /* 2000 + 100 = 2100 ticks elapsed */ 474 CU_ASSERT(reactor->tsc_last == 2100); 475 476 spdk_set_thread(thread1); 477 spdk_poller_unregister(&busy1); 478 spdk_thread_exit(thread1); 479 480 _reactor_run(reactor); 481 482 CU_ASSERT(TAILQ_EMPTY(&reactor->threads)); 483 484 /* No further than 2100 ticks elapsed */ 485 CU_ASSERT(reactor->tsc_last == 2100); 486 487 spdk_reactors_fini(); 488 489 free_cores(); 490 491 MOCK_CLEAR(spdk_env_get_current_core); 492 } 493 494 static uint32_t 495 _run_events_till_completion(uint32_t reactor_count) 496 { 497 struct spdk_reactor *reactor; 498 struct spdk_thread *app_thread = spdk_thread_get_app_thread(); 499 uint32_t i, events; 500 uint32_t total_events = 0; 501 502 do { 503 events = 0; 504 for (i = 0; i < reactor_count; i++) { 505 reactor = spdk_reactor_get(i); 506 CU_ASSERT(reactor != NULL); 507 MOCK_SET(spdk_env_get_current_core, i); 508 events += event_queue_run_batch(reactor); 509 510 /* Some events still require app_thread to run */ 511 MOCK_SET(spdk_env_get_current_core, g_scheduling_reactor->lcore); 512 spdk_thread_poll(app_thread, 0, 0); 513 514 MOCK_CLEAR(spdk_env_get_current_core); 515 } 516 total_events += events; 517 } while (events > 0); 518 519 return total_events; 520 } 521 522 static void 523 test_scheduler(void) 524 { 525 struct spdk_cpuset cpuset = {}; 526 struct spdk_thread *thread[3]; 527 struct spdk_reactor *reactor; 528 struct spdk_poller *busy, *idle; 529 uint64_t reactor_busy_tsc[3], reactor_idle_tsc[3]; 530 uint64_t thread_busy_tsc[3], thread_idle_tsc[3]; 531 uint64_t current_time, end_time, busy_time, idle_time; 532 struct spdk_thread_stats stats; 533 int i; 534 535 MOCK_SET(spdk_env_get_current_core, 0); 536 537 allocate_cores(3); 538 539 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 540 541 spdk_scheduler_set("dynamic"); 542 543 for (i = 0; i < 3; i++) { 544 spdk_cpuset_set_cpu(&g_reactor_core_mask, i, true); 545 } 546 g_next_core = 0; 547 548 /* Create threads. */ 549 for (i = 0; i < 3; i++) { 550 spdk_cpuset_zero(&cpuset); 551 spdk_cpuset_set_cpu(&cpuset, i, true); 552 thread[i] = spdk_thread_create(NULL, &cpuset); 553 CU_ASSERT(thread[i] != NULL); 554 thread_busy_tsc[i] = 0; 555 thread_idle_tsc[i] = 0; 556 } 557 558 for (i = 0; i < 3; i++) { 559 reactor = spdk_reactor_get(i); 560 CU_ASSERT(reactor != NULL); 561 MOCK_SET(spdk_env_get_current_core, i); 562 event_queue_run_batch(reactor); 563 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 564 reactor_busy_tsc[i] = 0; 565 reactor_idle_tsc[i] = 0; 566 } 567 568 g_reactor_state = SPDK_REACTOR_STATE_RUNNING; 569 570 MOCK_SET(spdk_env_get_current_core, 0); 571 572 /* Init threads stats (low load) */ 573 /* Each reactor starts at 100 tsc, 574 * ends at 100 + 100 = 200 tsc. */ 575 current_time = 100; 576 idle_time = 100; 577 busy_time = 0; 578 end_time = current_time + idle_time + busy_time; 579 for (i = 0; i < 3; i++) { 580 spdk_set_thread(thread[i]); 581 idle = spdk_poller_register(poller_run_idle, (void *)idle_time, 0); 582 reactor = spdk_reactor_get(i); 583 CU_ASSERT(reactor != NULL); 584 MOCK_SET(spdk_get_ticks, current_time); 585 reactor->tsc_last = spdk_get_ticks(); 586 _reactor_run(reactor); 587 CU_ASSERT(reactor->tsc_last == end_time); 588 spdk_poller_unregister(&idle); 589 590 CU_ASSERT(spdk_thread_get_last_tsc(thread[i]) == end_time); 591 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 592 CU_ASSERT(stats.busy_tsc == busy_time); 593 thread_busy_tsc[i] = stats.busy_tsc; 594 CU_ASSERT(stats.idle_tsc == idle_time); 595 thread_idle_tsc[i] = stats.idle_tsc; 596 CU_ASSERT(reactor->busy_tsc == busy_time); 597 reactor_busy_tsc[i] = reactor->busy_tsc; 598 CU_ASSERT(reactor->idle_tsc == idle_time); 599 reactor_idle_tsc[i] = reactor->idle_tsc; 600 } 601 CU_ASSERT(spdk_get_ticks() == end_time); 602 current_time = 200; 603 604 MOCK_SET(spdk_env_get_current_core, 0); 605 _reactors_scheduler_gather_metrics(NULL, NULL); 606 607 _run_events_till_completion(3); 608 MOCK_SET(spdk_env_get_current_core, 0); 609 610 /* Threads were idle, so all of them should be placed on core 0. 611 * All reactors start and end at 200 tsc, since for this iteration 612 * the threads have no pollers (so they consume no idle or busy tsc). 613 */ 614 for (i = 0; i < 3; i++) { 615 reactor = spdk_reactor_get(i); 616 CU_ASSERT(reactor != NULL); 617 MOCK_SET(spdk_get_ticks, current_time); 618 _reactor_run(reactor); 619 CU_ASSERT(reactor->tsc_last == current_time); 620 CU_ASSERT(reactor->busy_tsc == reactor_busy_tsc[i]); 621 CU_ASSERT(reactor->idle_tsc == reactor_idle_tsc[i]); 622 spdk_set_thread(thread[i]); 623 CU_ASSERT(spdk_thread_get_last_tsc(thread[i]) == current_time); 624 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 625 CU_ASSERT(stats.busy_tsc == thread_busy_tsc[i]); 626 CU_ASSERT(stats.idle_tsc == thread_idle_tsc[i]); 627 } 628 CU_ASSERT(spdk_get_ticks() == current_time); 629 630 /* 2 threads should be scheduled to core 0 */ 631 reactor = spdk_reactor_get(0); 632 CU_ASSERT(reactor != NULL); 633 MOCK_SET(spdk_env_get_current_core, 0); 634 event_queue_run_batch(reactor); 635 636 reactor = spdk_reactor_get(0); 637 CU_ASSERT(reactor != NULL); 638 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 639 reactor = spdk_reactor_get(1); 640 CU_ASSERT(reactor != NULL); 641 CU_ASSERT(TAILQ_EMPTY(&reactor->threads)); 642 reactor = spdk_reactor_get(2); 643 CU_ASSERT(reactor != NULL); 644 CU_ASSERT(TAILQ_EMPTY(&reactor->threads)); 645 646 /* Make threads busy */ 647 reactor = spdk_reactor_get(0); 648 CU_ASSERT(reactor != NULL); 649 650 /* All threads run on single reactor, 651 * reactor 0 starts at 200 tsc, 652 * ending at 200 + (100 * 3) = 500 tsc. */ 653 MOCK_SET(spdk_get_ticks, current_time); 654 busy_time = 100; 655 idle_time = 0; 656 for (i = 0; i < 3; i++) { 657 spdk_set_thread(thread[i]); 658 busy = spdk_poller_register(poller_run_busy, (void *)busy_time, 0); 659 _reactor_run(reactor); 660 spdk_poller_unregister(&busy); 661 current_time += busy_time; 662 663 CU_ASSERT(reactor->tsc_last == current_time); 664 CU_ASSERT(spdk_thread_get_last_tsc(thread[i]) == current_time); 665 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 666 CU_ASSERT(stats.busy_tsc == thread_busy_tsc[i] + busy_time); 667 CU_ASSERT(stats.idle_tsc == thread_idle_tsc[i] + idle_time);; 668 } 669 CU_ASSERT(reactor->busy_tsc == reactor_busy_tsc[0] + 3 * busy_time); 670 CU_ASSERT(reactor->idle_tsc == reactor_idle_tsc[0] + 3 * idle_time); 671 CU_ASSERT(spdk_get_ticks() == current_time); 672 673 /* Run scheduler again, this time all threads are busy */ 674 MOCK_SET(spdk_env_get_current_core, 0); 675 _reactors_scheduler_gather_metrics(NULL, NULL); 676 677 _run_events_till_completion(3); 678 MOCK_SET(spdk_env_get_current_core, 0); 679 680 /* Threads were busy, 2 will stay on core 0, 1 will move to core 1 */ 681 for (i = 0; i < 3; i++) { 682 MOCK_SET(spdk_env_get_current_core, i); 683 reactor = spdk_reactor_get(i); 684 CU_ASSERT(reactor != NULL); 685 _reactor_run(reactor); 686 } 687 688 for (i = 0; i < 3; i++) { 689 reactor = spdk_reactor_get(i); 690 CU_ASSERT(reactor != NULL); 691 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 692 } 693 694 g_reactor_state = SPDK_REACTOR_STATE_INITIALIZED; 695 696 /* Destroy threads */ 697 for (i = 0; i < 3; i++) { 698 spdk_set_thread(thread[i]); 699 spdk_thread_exit(thread[i]); 700 } 701 for (i = 0; i < 3; i++) { 702 reactor = spdk_reactor_get(i); 703 CU_ASSERT(reactor != NULL); 704 reactor_run(reactor); 705 } 706 707 spdk_set_thread(NULL); 708 709 MOCK_CLEAR(spdk_env_get_current_core); 710 711 spdk_reactors_fini(); 712 713 free_cores(); 714 } 715 716 static void 717 test_bind_thread(void) 718 { 719 struct spdk_cpuset cpuset = {}; 720 struct spdk_thread *thread[3]; 721 struct spdk_reactor *reactor; 722 struct spdk_poller *idle; 723 uint64_t reactor_busy_tsc[3], reactor_idle_tsc[3]; 724 uint64_t thread_busy_tsc[3], thread_idle_tsc[3]; 725 uint64_t current_time, end_time, busy_time, idle_time; 726 struct spdk_thread_stats stats; 727 int i; 728 729 MOCK_SET(spdk_env_get_current_core, 0); 730 731 allocate_cores(3); 732 733 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 734 735 spdk_scheduler_set("dynamic"); 736 737 for (i = 0; i < 3; i++) { 738 spdk_cpuset_set_cpu(&g_reactor_core_mask, i, true); 739 } 740 g_next_core = 0; 741 742 /* Create threads. */ 743 for (i = 0; i < 3; i++) { 744 spdk_cpuset_zero(&cpuset); 745 spdk_cpuset_set_cpu(&cpuset, i, true); 746 thread[i] = spdk_thread_create(NULL, &cpuset); 747 CU_ASSERT(thread[i] != NULL); 748 thread_busy_tsc[i] = 0; 749 thread_idle_tsc[i] = 0; 750 } 751 752 for (i = 0; i < 3; i++) { 753 reactor = spdk_reactor_get(i); 754 CU_ASSERT(reactor != NULL); 755 MOCK_SET(spdk_env_get_current_core, i); 756 event_queue_run_batch(reactor); 757 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 758 reactor_busy_tsc[i] = 0; 759 reactor_idle_tsc[i] = 0; 760 } 761 762 g_reactor_state = SPDK_REACTOR_STATE_RUNNING; 763 764 MOCK_SET(spdk_env_get_current_core, 0); 765 766 /* Init threads stats (low load) */ 767 /* Each reactor starts at 100 tsc, 768 * ends at 100 + 100 = 200 tsc. */ 769 current_time = 100; 770 idle_time = 100; 771 busy_time = 0; 772 end_time = current_time + idle_time + busy_time; 773 for (i = 0; i < 3; i++) { 774 spdk_set_thread(thread[i]); 775 idle = spdk_poller_register(poller_run_idle, (void *)idle_time, 0); 776 reactor = spdk_reactor_get(i); 777 CU_ASSERT(reactor != NULL); 778 MOCK_SET(spdk_get_ticks, current_time); 779 reactor->tsc_last = spdk_get_ticks(); 780 _reactor_run(reactor); 781 CU_ASSERT(reactor->tsc_last == end_time); 782 spdk_poller_unregister(&idle); 783 784 CU_ASSERT(spdk_thread_get_last_tsc(thread[i]) == end_time); 785 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 786 CU_ASSERT(stats.busy_tsc == busy_time); 787 thread_busy_tsc[i] = stats.busy_tsc; 788 CU_ASSERT(stats.idle_tsc == idle_time); 789 thread_idle_tsc[i] = stats.idle_tsc; 790 CU_ASSERT(reactor->busy_tsc == busy_time); 791 reactor_busy_tsc[i] = reactor->busy_tsc; 792 CU_ASSERT(reactor->idle_tsc == idle_time); 793 reactor_idle_tsc[i] = reactor->idle_tsc; 794 } 795 CU_ASSERT(spdk_get_ticks() == end_time); 796 current_time = 200; 797 /* Bind thread 1 */ 798 spdk_thread_bind(thread[1], true); 799 CU_ASSERT(spdk_thread_is_bound(thread[1]) == true); 800 MOCK_SET(spdk_env_get_current_core, 0); 801 _reactors_scheduler_gather_metrics(NULL, NULL); 802 _run_events_till_completion(3); 803 MOCK_SET(spdk_env_get_current_core, 0); 804 805 /* Threads were idle, so all of them should be placed on core 0 except thread 1 806 * since it has been limited on core 1 807 * All reactors start and end at 200 tsc, since for this iteration 808 * the threads have no pollers (so they consume no idle or busy tsc). 809 */ 810 for (i = 0; i < 3; i++) { 811 reactor = spdk_reactor_get(i); 812 CU_ASSERT(reactor != NULL); 813 MOCK_SET(spdk_get_ticks, current_time); 814 _reactor_run(reactor); 815 CU_ASSERT(reactor->tsc_last == current_time); 816 CU_ASSERT(reactor->busy_tsc == reactor_busy_tsc[i]); 817 CU_ASSERT(reactor->idle_tsc == reactor_idle_tsc[i]); 818 spdk_set_thread(thread[i]); 819 CU_ASSERT(spdk_thread_get_last_tsc(thread[i]) == current_time); 820 CU_ASSERT(spdk_thread_get_stats(&stats) == 0); 821 CU_ASSERT(stats.busy_tsc == thread_busy_tsc[i]); 822 CU_ASSERT(stats.idle_tsc == thread_idle_tsc[i]); 823 } 824 CU_ASSERT(spdk_get_ticks() == current_time); 825 826 /* Thread on core 2 should be scheduled to core 0 */ 827 reactor = spdk_reactor_get(0); 828 CU_ASSERT(reactor != NULL); 829 MOCK_SET(spdk_env_get_current_core, 0); 830 event_queue_run_batch(reactor); 831 832 reactor = spdk_reactor_get(0); 833 CU_ASSERT(reactor != NULL); 834 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 835 /* Thread 1 has been limited and stiil on core 0 */ 836 reactor = spdk_reactor_get(1); 837 CU_ASSERT(reactor != NULL); 838 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 839 840 reactor = spdk_reactor_get(2); 841 CU_ASSERT(reactor != NULL); 842 CU_ASSERT(TAILQ_EMPTY(&reactor->threads)); 843 844 g_reactor_state = SPDK_REACTOR_STATE_INITIALIZED; 845 846 /* Destroy threads */ 847 for (i = 0; i < 3; i++) { 848 spdk_set_thread(thread[i]); 849 spdk_thread_exit(thread[i]); 850 } 851 for (i = 0; i < 3; i++) { 852 reactor = spdk_reactor_get(i); 853 CU_ASSERT(reactor != NULL); 854 reactor_run(reactor); 855 } 856 857 spdk_set_thread(NULL); 858 859 MOCK_CLEAR(spdk_env_get_current_core); 860 861 spdk_reactors_fini(); 862 863 free_cores(); 864 } 865 866 uint8_t g_curr_freq; 867 868 static int 869 core_freq_up(uint32_t lcore) 870 { 871 if (g_curr_freq != UINT8_MAX) { 872 g_curr_freq++; 873 } 874 875 return 0; 876 } 877 878 static int 879 core_freq_down(uint32_t lcore) 880 { 881 if (g_curr_freq != 0) { 882 g_curr_freq--; 883 } 884 885 return 0; 886 } 887 888 static int 889 core_freq_max(uint32_t lcore) 890 { 891 g_curr_freq = UINT8_MAX; 892 893 return 0; 894 } 895 896 DEFINE_STUB(core_freq_min, int, (uint32_t lcore_id), 0); 897 DEFINE_STUB(core_caps, int, 898 (uint32_t lcore_id, struct spdk_governor_capabilities *capabilities), 0); 899 DEFINE_STUB(governor_init, int, (void), 0); 900 DEFINE_STUB_V(governor_deinit, (void)); 901 902 static struct spdk_governor governor = { 903 .name = "dpdk_governor", 904 .get_core_curr_freq = NULL, 905 .core_freq_up = core_freq_up, 906 .core_freq_down = core_freq_down, 907 .set_core_freq_max = core_freq_max, 908 .set_core_freq_min = core_freq_min, 909 .get_core_capabilities = core_caps, 910 .init = governor_init, 911 .deinit = governor_deinit, 912 }; 913 914 static void 915 test_governor(void) 916 { 917 struct spdk_cpuset cpuset = {}; 918 struct spdk_thread *thread[2]; 919 struct spdk_lw_thread *lw_thread; 920 struct spdk_reactor *reactor; 921 struct spdk_poller *busy, *idle; 922 uint8_t last_freq = 100; 923 int i; 924 925 MOCK_SET(spdk_env_get_current_core, 0); 926 927 g_curr_freq = last_freq; 928 spdk_governor_register(&governor); 929 930 allocate_cores(2); 931 932 CU_ASSERT(spdk_reactors_init(SPDK_DEFAULT_MSG_MEMPOOL_SIZE) == 0); 933 934 spdk_scheduler_set("dynamic"); 935 spdk_governor_set("dpdk_governor"); 936 937 for (i = 0; i < 2; i++) { 938 spdk_cpuset_set_cpu(&g_reactor_core_mask, i, true); 939 } 940 941 /* Create threads. */ 942 for (i = 0; i < 2; i++) { 943 spdk_cpuset_zero(&cpuset); 944 spdk_cpuset_set_cpu(&cpuset, i, true); 945 thread[i] = spdk_thread_create(NULL, &cpuset); 946 CU_ASSERT(thread[i] != NULL); 947 } 948 949 for (i = 0; i < 2; i++) { 950 reactor = spdk_reactor_get(i); 951 CU_ASSERT(reactor != NULL); 952 MOCK_SET(spdk_env_get_current_core, i); 953 CU_ASSERT(event_queue_run_batch(reactor) == 1); 954 CU_ASSERT(!TAILQ_EMPTY(&reactor->threads)); 955 } 956 957 reactor = spdk_reactor_get(0); 958 CU_ASSERT(reactor != NULL); 959 MOCK_SET(spdk_env_get_current_core, 0); 960 961 g_reactor_state = SPDK_REACTOR_STATE_RUNNING; 962 963 /* TEST 1 */ 964 /* Init thread stats (low load) */ 965 MOCK_SET(spdk_get_ticks, 100); 966 reactor->tsc_last = 100; 967 968 for (i = 0; i < 2; i++) { 969 spdk_set_thread(thread[i]); 970 idle = spdk_poller_register(poller_run_idle, (void *)200, 0); 971 reactor = spdk_reactor_get(i); 972 CU_ASSERT(reactor != NULL); 973 MOCK_SET(spdk_env_get_current_core, i); 974 _reactor_run(reactor); 975 spdk_poller_unregister(&idle); 976 977 /* Update last stats so that we don't have to call scheduler twice */ 978 lw_thread = spdk_thread_get_ctx(thread[i]); 979 lw_thread->current_stats.idle_tsc = 1; 980 } 981 982 MOCK_SET(spdk_env_get_current_core, 0); 983 _reactors_scheduler_gather_metrics(NULL, NULL); 984 985 CU_ASSERT(_run_events_till_completion(2) == 2); 986 MOCK_SET(spdk_env_get_current_core, 0); 987 988 /* Threads were idle, so all of them should be placed on core 0 */ 989 for (i = 0; i < 2; i++) { 990 reactor = spdk_reactor_get(i); 991 CU_ASSERT(reactor != NULL); 992 _reactor_run(reactor); 993 } 994 995 /* 1 thread should be scheduled to core 0 */ 996 reactor = spdk_reactor_get(0); 997 CU_ASSERT(reactor != NULL); 998 MOCK_SET(spdk_env_get_current_core, 0); 999 CU_ASSERT(event_queue_run_batch(reactor) == 1); 1000 1001 /* Main core should be busy less than 50% time now - frequency should be lowered */ 1002 CU_ASSERT(g_curr_freq == last_freq - 1); 1003 1004 last_freq = g_curr_freq; 1005 1006 /* TEST 2 */ 1007 /* Make first threads busy - both threads will be still on core 0, but frequency will have to be raised */ 1008 spdk_set_thread(thread[0]); 1009 busy = spdk_poller_register(poller_run_busy, (void *)1000, 0); 1010 _reactor_run(reactor); 1011 spdk_poller_unregister(&busy); 1012 1013 spdk_set_thread(thread[1]); 1014 idle = spdk_poller_register(poller_run_idle, (void *)100, 0); 1015 _reactor_run(reactor); 1016 spdk_poller_unregister(&idle); 1017 1018 /* Run scheduler again */ 1019 MOCK_SET(spdk_env_get_current_core, 0); 1020 _reactors_scheduler_gather_metrics(NULL, NULL); 1021 1022 i = _run_events_till_completion(2); 1023 /* Six runs when interrupt mode is supported, two if not. */ 1024 CU_ASSERT(i == 6 || i == 2); 1025 MOCK_SET(spdk_env_get_current_core, 0); 1026 1027 /* Main core should be busy more than 50% time now - frequency should be raised */ 1028 CU_ASSERT(g_curr_freq == last_freq + 1); 1029 1030 /* TEST 3 */ 1031 /* Make second thread very busy so that it will be moved to second core */ 1032 spdk_set_thread(thread[1]); 1033 busy = spdk_poller_register(poller_run_busy, (void *)2000, 0); 1034 _reactor_run(reactor); 1035 spdk_poller_unregister(&busy); 1036 1037 /* Update first thread stats */ 1038 spdk_set_thread(thread[0]); 1039 idle = spdk_poller_register(poller_run_idle, (void *)100, 0); 1040 _reactor_run(reactor); 1041 spdk_poller_unregister(&idle); 1042 1043 /* Run scheduler again */ 1044 MOCK_SET(spdk_env_get_current_core, 0); 1045 _reactors_scheduler_gather_metrics(NULL, NULL); 1046 1047 i = _run_events_till_completion(2); 1048 /* Six runs when interrupt mode is supported, two if not. */ 1049 CU_ASSERT(i == 6 || i == 2); 1050 MOCK_SET(spdk_env_get_current_core, 0); 1051 1052 for (i = 0; i < 2; i++) { 1053 reactor = spdk_reactor_get(i); 1054 CU_ASSERT(reactor != NULL); 1055 _reactor_run(reactor); 1056 } 1057 1058 /* Main core frequency should be set to max when we have busy threads on other cores */ 1059 CU_ASSERT(g_curr_freq == UINT8_MAX); 1060 1061 g_reactor_state = SPDK_REACTOR_STATE_INITIALIZED; 1062 1063 /* Destroy threads */ 1064 for (i = 0; i < 2; i++) { 1065 spdk_set_thread(thread[i]); 1066 spdk_thread_exit(thread[i]); 1067 } 1068 for (i = 0; i < 2; i++) { 1069 reactor = spdk_reactor_get(i); 1070 CU_ASSERT(reactor != NULL); 1071 reactor_run(reactor); 1072 } 1073 1074 spdk_set_thread(NULL); 1075 1076 MOCK_CLEAR(spdk_env_get_current_core); 1077 1078 spdk_reactors_fini(); 1079 1080 free_cores(); 1081 } 1082 1083 int 1084 main(int argc, char **argv) 1085 { 1086 CU_pSuite suite = NULL; 1087 unsigned int num_failures; 1088 1089 CU_set_error_action(CUEA_ABORT); 1090 CU_initialize_registry(); 1091 1092 suite = CU_add_suite("app_suite", NULL, NULL); 1093 1094 CU_ADD_TEST(suite, test_create_reactor); 1095 CU_ADD_TEST(suite, test_init_reactors); 1096 CU_ADD_TEST(suite, test_event_call); 1097 CU_ADD_TEST(suite, test_schedule_thread); 1098 CU_ADD_TEST(suite, test_reschedule_thread); 1099 CU_ADD_TEST(suite, test_bind_thread); 1100 CU_ADD_TEST(suite, test_for_each_reactor); 1101 CU_ADD_TEST(suite, test_reactor_stats); 1102 CU_ADD_TEST(suite, test_scheduler); 1103 CU_ADD_TEST(suite, test_governor); 1104 1105 CU_basic_set_mode(CU_BRM_VERBOSE); 1106 CU_basic_run_tests(); 1107 num_failures = CU_get_number_of_failures(); 1108 CU_cleanup_registry(); 1109 1110 return num_failures; 1111 } 1112