| … | … | @@ -27,6 +27,7 @@ |
| 27 | 27 | |
| 28 | 28 | #include <atomic> |
| 29 | 29 | #include <chrono> |
| 30 | #include <latch> | |
| 30 | 31 | #include <mutex> |
| 31 | 32 | #include <thread> |
| 32 | 33 | #include <vector> |
| … | … | @@ -143,7 +144,7 @@ class async_buffer |
| 143 | 144 | |
| 144 | 145 | // Per-thread barrier synchronization |
| 145 | 146 | std::atomic<unsigned> _barrier_phase; |
| 146 | std::atomic<unsigned> _barrier_remaining; | |
| 147 | std::latch* _barrier_latch; | |
| 147 | 148 | |
| 148 | 149 | void start_writer_thread(); |
| 149 | 150 | void stop_writer_threads(); |
| … | … | @@ -217,7 +218,7 @@ async_buffer<Writer, Element>::async_buffer(Writer* wr, |
| 217 | 218 | _stall_writers = false; |
| 218 | 219 | _in_drain = false; |
| 219 | 220 | _barrier_phase = 0; |
| 220 | _barrier_remaining = 0; | |
| 221 | _barrier_latch = nullptr; | |
| 221 | 222 | |
| 222 | 223 | _high_watermark = DEFAULT_HIGH_WATER_MARK; |
| 223 | 224 | _low_watermark = DEFAULT_LOW_WATER_MARK; |
| … | … | @@ -358,7 +359,7 @@ void async_buffer<Writer, Element>::stop_writer_threads() |
| 358 | 359 | } |
| 359 | 360 | |
| 360 | 361 | _barrier_phase = 0; |
| 361 | _barrier_remaining = 0; | |
| 362 | _barrier_latch = nullptr; | |
| 362 | 363 | |
| 363 | 364 | // Its now OK to start new threads, if desired ...(!) |
| 364 | 365 | _stopping_writers = false; |
| … | … | @@ -452,21 +453,18 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 452 | 453 | // First, drain existing work |
| 453 | 454 | drain(); |
| 454 | 455 | |
| 455 | // Set up the barrier: each thread must process exactly one element | |
| 456 | _barrier_remaining = _thread_count; | |
| 456 | // Set up the barrier using std::latch for synchronization | |
| 457 | std::latch completion(_thread_count); | |
| 458 | _barrier_latch = &completion; | |
| 457 | 459 | _barrier_phase++; |
| 458 | _barrier_remaining.notify_all(); // Wake workers checking phase condition | |
| 459 | 460 | |
| 460 | 461 | // Enqueue one element for each thread. |
| 461 | 462 | for (unsigned int i = 0; i < _thread_count; i++) |
| 462 | 463 | do_insert(elt); |
| 463 | 464 | |
| 464 | 465 | // Wait for all threads to complete their barrier work |
| 465 | while (_barrier_remaining.load() > 0) | |
| 466 | { | |
| 467 | unsigned rem = _barrier_remaining.load(); | |
| 468 | if (rem > 0) _barrier_remaining.wait(rem); | |
| 469 | } | |
| 466 | completion.wait(); | |
| 467 | _barrier_latch = nullptr; | |
| 470 | 468 | } |
| 471 | 469 | |
| 472 | 470 | /// A single write thread. Reads elements from set, and invokes the |
| … | … | @@ -482,17 +480,6 @@ void async_buffer<Writer, Element>::write_loop() |
| 482 | 480 | { |
| 483 | 481 | while (true) |
| 484 | 482 | { |
| 485 | // If a barrier is active and we've done our part, wait for others. | |
| 486 | // This prevents us from grabbing a second barrier element before | |
| 487 | // all other threads have grabbed their first. | |
| 488 | while (_barrier_remaining.load() > 0 and | |
| 489 | my_last_phase >= _barrier_phase.load()) | |
| 490 | { | |
| 491 | unsigned rem = _barrier_remaining.load(); | |
| 492 | if (rem > 0 and my_last_phase >= _barrier_phase.load()) | |
| 493 | _barrier_remaining.wait(rem); | |
| 494 | } | |
| 495 | ||
| 496 | 483 | // Do nothing, if asked to stall. |
| 497 | 484 | while (_stall_writers and _store_set.size() < _low_watermark) |
| 498 | 485 | { |
| … | … | @@ -505,11 +492,10 @@ void async_buffer<Writer, Element>::write_loop() |
| 505 | 492 | unsigned current_phase = _barrier_phase.load(); |
| 506 | 493 | if (my_last_phase < current_phase) |
| 507 | 494 | { |
| 508 | // This is barrier work - process and count toward completion | |
| 495 | // This is barrier work - process and signal completion | |
| 509 | 496 | my_last_phase = current_phase; |
| 510 | 497 | (_writer->*_do_write)(elt); |
| 511 | _barrier_remaining.fetch_sub(1); | |
| 512 | _barrier_remaining.notify_all(); | |
| 498 | _barrier_latch->count_down(); | |
| 513 | 499 | } |
| 514 | 500 | else |
| 515 | 501 | { |
| … | … | @@ -27,6 +27,7 @@ |
| 27 | 27 | |
| 28 | 28 | #include <atomic> |
| 29 | 29 | #include <chrono> |
| 30 | #include <latch> | |
| 30 | 31 | #include <mutex> |
| 31 | 32 | #include <thread> |
| 32 | 33 | #include <vector> |
| … | … | @@ -116,7 +117,7 @@ class async_caller |
| 116 | 117 | |
| 117 | 118 | // Per-thread barrier synchronization |
| 118 | 119 | std::atomic<unsigned> _barrier_phase; |
| 119 | std::atomic<unsigned> _barrier_remaining; | |
| 120 | std::latch* _barrier_latch; | |
| 120 | 121 | |
| 121 | 122 | void start_writer_thread(); |
| 122 | 123 | void stop_writer_threads(); |
| … | … | @@ -180,7 +181,7 @@ async_caller<Writer, Element>::async_caller(Writer* wr, |
| 180 | 181 | _pending = 0; |
| 181 | 182 | _in_drain = false; |
| 182 | 183 | _barrier_phase = 0; |
| 183 | _barrier_remaining = 0; | |
| 184 | _barrier_latch = nullptr; | |
| 184 | 185 | |
| 185 | 186 | _high_watermark = DEFAULT_HIGH_WATER_MARK; |
| 186 | 187 | _low_watermark = DEFAULT_LOW_WATER_MARK; |
| … | … | @@ -278,7 +279,7 @@ void async_caller<Writer, Element>::stop_writer_threads() |
| 278 | 279 | } |
| 279 | 280 | |
| 280 | 281 | _barrier_phase = 0; |
| 281 | _barrier_remaining = 0; | |
| 282 | _barrier_latch = nullptr; | |
| 282 | 283 | |
| 283 | 284 | // Its now OK to start new threads, if desired ...(!) |
| 284 | 285 | _stopping_writers = false; |
| … | … | @@ -366,10 +367,10 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 366 | 367 | // First, drain existing work |
| 367 | 368 | drain(); |
| 368 | 369 | |
| 369 | // Set up the barrier: each thread must process exactly one element | |
| 370 | _barrier_remaining = _thread_count; | |
| 370 | // Set up the latch for barrier completion | |
| 371 | std::latch completion(_thread_count); | |
| 372 | _barrier_latch = &completion; | |
| 371 | 373 | _barrier_phase++; |
| 372 | _barrier_remaining.notify_all(); // Wake workers checking phase condition | |
| 373 | 374 | |
| 374 | 375 | // Enqueue one element for each thread. |
| 375 | 376 | for (unsigned int i = 0; i < _thread_count; i++) |
| … | … | @@ -380,11 +381,8 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 380 | 381 | } |
| 381 | 382 | |
| 382 | 383 | // Wait for all threads to complete their barrier work |
| 383 | while (_barrier_remaining.load() > 0) | |
| 384 | { | |
| 385 | unsigned rem = _barrier_remaining.load(); | |
| 386 | if (rem > 0) _barrier_remaining.wait(rem); | |
| 387 | } | |
| 384 | completion.wait(); | |
| 385 | _barrier_latch = nullptr; | |
| 388 | 386 | } |
| 389 | 387 | |
| 390 | 388 | /// A single write thread. Reads elements from queue, and invokes the |
| … | … | @@ -400,17 +398,6 @@ void async_caller<Writer, Element>::write_loop() |
| 400 | 398 | { |
| 401 | 399 | while (true) |
| 402 | 400 | { |
| 403 | // If a barrier is active and we've done our part, wait for others. | |
| 404 | // This prevents us from grabbing a second barrier element before | |
| 405 | // all other threads have grabbed their first. | |
| 406 | while (_barrier_remaining.load() > 0 and | |
| 407 | my_last_phase >= _barrier_phase.load()) | |
| 408 | { | |
| 409 | unsigned rem = _barrier_remaining.load(); | |
| 410 | if (rem > 0 and my_last_phase >= _barrier_phase.load()) | |
| 411 | _barrier_remaining.wait(rem); | |
| 412 | } | |
| 413 | ||
| 414 | 401 | Element elt = _store_queue.value_pop(); |
| 415 | 402 | _busy_writers ++; |
| 416 | 403 | |
| … | … | @@ -420,8 +407,7 @@ void async_caller<Writer, Element>::write_loop() |
| 420 | 407 | // This is barrier work - process and count toward completion |
| 421 | 408 | my_last_phase = current_phase; |
| 422 | 409 | (_writer->*_do_write)(elt); |
| 423 | _barrier_remaining.fetch_sub(1); | |
| 424 | _barrier_remaining.notify_all(); | |
| 410 | _barrier_latch->count_down(); | |
| 425 | 411 | } |
| 426 | 412 | else |
| 427 | 413 | { |