| … | … | @@ -142,8 +142,8 @@ class async_buffer |
| 142 | 142 | |
| 143 | 143 | bool _stall_writers; |
| 144 | 144 | |
| 145 | | // Per-thread barrier synchronization |
| 146 | | std::atomic<unsigned> _barrier_phase; |
| 145 | // Barrier synchronization |
| 146 | const Element* _current_barrier; |
| 147 | 147 | std::latch* _barrier_latch; |
| 148 | 148 | |
| 149 | 149 | void start_writer_thread(); |
| … | … | @@ -218,7 +218,7 @@ async_buffer<Writer, Element>::async_buffer(Writer* wr, |
| 218 | 218 | _pending = 0; |
| 219 | 219 | _stall_writers = false; |
| 220 | 220 | _in_drain = false; |
| 221 | | _barrier_phase = 0; |
| 221 | _current_barrier = nullptr; |
| 222 | 222 | _barrier_latch = nullptr; |
| 223 | 223 | |
| 224 | 224 | _high_watermark = DEFAULT_HIGH_WATER_MARK; |
| … | … | @@ -359,7 +359,7 @@ void async_buffer<Writer, Element>::stop_writer_threads() |
| 359 | 359 | (_writer->*_do_write)(elt); |
| 360 | 360 | } |
| 361 | 361 | |
| 362 | | _barrier_phase = 0; |
| 362 | _current_barrier = nullptr; |
| 363 | 363 | _barrier_latch = nullptr; |
| 364 | 364 | |
| 365 | 365 | // Its now OK to start new threads, if desired ...(!) |
| … | … | @@ -446,25 +446,24 @@ void async_buffer<Writer, Element>::barrier() |
| 446 | 446 | /// workers that do caching or lazy initialization or implement |
| 447 | 447 | /// other actions that require synchronization that the default |
| 448 | 448 | /// `barrier(void)` would miss. |
| 449 | /// |
| 449 | 450 | template<typename Writer, typename Element> |
| 450 | 451 | void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 451 | 452 | { |
| 452 | 453 | std::unique_lock<std::mutex> lock(_enqueue_mutex); |
| 453 | 454 | |
| 454 | | // First, drain existing work |
| 455 | 455 | drain(); |
| 456 | 456 | |
| 457 | | // Set up the barrier using std::latch for synchronization |
| 458 | 457 | std::latch completion(_thread_count); |
| 459 | 458 | _barrier_latch = &completion; |
| 460 | | _barrier_phase++; |
| 459 | _current_barrier = &elt; |
| 461 | 460 | |
| 462 | | // Enqueue one element for each thread. |
| 463 | | for (unsigned int i = 0; i < _thread_count; i++) |
| 464 | | do_insert(elt); |
| 461 | _store_set.cancel(); |
| 465 | 462 | |
| 466 | | // Wait for all threads to complete their barrier work |
| 467 | 463 | completion.wait(); |
| 464 | |
| 465 | _store_set.cancel_reset(); |
| 466 | _current_barrier = nullptr; |
| 468 | 467 | _barrier_latch = nullptr; |
| 469 | 468 | } |
| 470 | 469 | |
| … | … | @@ -473,13 +472,9 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 473 | 472 | template<typename Writer, typename Element> |
| 474 | 473 | void async_buffer<Writer, Element>::write_loop() |
| 475 | 474 | { |
| 476 | | // Track which barrier phase this thread has completed. |
| 477 | | // Used to ensure each thread processes exactly one barrier element. |
| 478 | | static thread_local unsigned my_last_phase = 0; |
| 479 | | |
| 480 | | try |
| 475 | while (true) |
| 481 | 476 | { |
| 482 | | while (true) |
| 477 | try |
| 483 | 478 | { |
| 484 | 479 | // Do nothing, if asked to stall. |
| 485 | 480 | while (_stall_writers and _store_set.size() < _low_watermark) |
| … | … | @@ -489,31 +484,24 @@ void async_buffer<Writer, Element>::write_loop() |
| 489 | 484 | |
| 490 | 485 | Element elt = _store_set.value_get(); |
| 491 | 486 | _busy_writers ++; |
| 492 | | |
| 493 | | unsigned current_phase = _barrier_phase.load(); |
| 494 | | if (my_last_phase < current_phase) |
| 487 | (_writer->*_do_write)(elt); |
| 488 | _busy_writers --; |
| 489 | _pending --; |
| 490 | _pending.notify_all(); |
| 491 | } |
| 492 | catch (typename concurrent_set<Element>::Canceled& e) |
| 493 | { |
| 494 | if (_current_barrier != nullptr) |
| 495 | 495 | { |
| 496 | | // This is barrier work - process and signal completion |
| 497 | | my_last_phase = current_phase; |
| 498 | | (_writer->*_do_write)(elt); |
| 496 | (_writer->*_do_write)(*_current_barrier); |
| 499 | 497 | _barrier_latch->count_down(); |
| 500 | | } |
| 501 | | else |
| 502 | | { |
| 503 | | // Regular work |
| 504 | | (_writer->*_do_write)(elt); |
| 498 | _barrier_latch->wait(); |
| 505 | 499 | } |
| 506 | 500 | |
| 507 | | _busy_writers --; |
| 508 | | _pending --; |
| 509 | | _pending.notify_all(); |
| 501 | // We are so out of here. Nothing to do, just exit this thread. |
| 502 | if (_stopping_writers) return; |
| 510 | 503 | } |
| 511 | 504 | } |
| 512 | | catch (typename concurrent_set<Element>::Canceled& e) |
| 513 | | { |
| 514 | | // We are so out of here. Nothing to do, just exit this thread. |
| 515 | | return; |
| 516 | | } |
| 517 | 505 | } |
| 518 | 506 | |
| 519 | 507 | |