| … | … | @@ -27,7 +27,6 @@ |
| 27 | 27 | |
| 28 | 28 | #include <atomic> |
| 29 | 29 | #include <chrono> |
| 30 | | #include <latch> |
| 31 | 30 | #include <mutex> |
| 32 | 31 | #include <thread> |
| 33 | 32 | #include <vector> |
| … | … | @@ -144,7 +143,7 @@ class async_buffer |
| 144 | 143 | |
| 145 | 144 | // Barrier synchronization |
| 146 | 145 | const Element* _current_barrier; |
| 147 | | std::latch* _barrier_latch; |
| 146 | std::atomic<unsigned> _barrier_count; |
| 148 | 147 | |
| 149 | 148 | void start_writer_thread(); |
| 150 | 149 | void stop_writer_threads(); |
| … | … | @@ -219,7 +218,7 @@ async_buffer<Writer, Element>::async_buffer(Writer* wr, |
| 219 | 218 | _stall_writers = false; |
| 220 | 219 | _in_drain = false; |
| 221 | 220 | _current_barrier = nullptr; |
| 222 | | _barrier_latch = nullptr; |
| 221 | _barrier_count = 0; |
| 223 | 222 | |
| 224 | 223 | _high_watermark = DEFAULT_HIGH_WATER_MARK; |
| 225 | 224 | _low_watermark = DEFAULT_LOW_WATER_MARK; |
| … | … | @@ -360,7 +359,7 @@ void async_buffer<Writer, Element>::stop_writer_threads() |
| 360 | 359 | } |
| 361 | 360 | |
| 362 | 361 | _current_barrier = nullptr; |
| 363 | | _barrier_latch = nullptr; |
| 362 | _barrier_count = 0; |
| 364 | 363 | |
| 365 | 364 | // Its now OK to start new threads, if desired ...(!) |
| 366 | 365 | _stopping_writers = false; |
| … | … | @@ -454,17 +453,23 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 454 | 453 | |
| 455 | 454 | drain(); |
| 456 | 455 | |
| 457 | | std::latch completion(_thread_count); |
| 458 | | _barrier_latch = &completion; |
| 456 | _barrier_count = _thread_count + 1; |
| 459 | 457 | _current_barrier = &elt; |
| 460 | 458 | |
| 461 | 459 | _store_set.cancel(); |
| 462 | 460 | |
| 463 | | completion.wait(); |
| 461 | // Wait for all workers to finish processing |
| 462 | unsigned cnt = _barrier_count.load(); |
| 463 | while (1 < cnt) |
| 464 | { |
| 465 | _barrier_count.wait(cnt); |
| 466 | cnt = _barrier_count.load(); |
| 467 | } |
| 464 | 468 | |
| 465 | | _store_set.cancel_reset(); |
| 466 | 469 | _current_barrier = nullptr; |
| 467 | | _barrier_latch = nullptr; |
| 470 | _store_set.cancel_reset(); |
| 471 | _barrier_count--; |
| 472 | _barrier_count.notify_all(); |
| 468 | 473 | } |
| 469 | 474 | |
| 470 | 475 | /// A single write thread. Reads elements from set, and invokes the |
| … | … | @@ -494,8 +499,16 @@ void async_buffer<Writer, Element>::write_loop() |
| 494 | 499 | if (_current_barrier != nullptr) |
| 495 | 500 | { |
| 496 | 501 | (_writer->*_do_write)(*_current_barrier); |
| 497 | | _barrier_latch->count_down(); |
| 498 | | _barrier_latch->wait(); |
| 502 | _barrier_count--; |
| 503 | _barrier_count.notify_all(); |
| 504 | |
| 505 | // Wait for barrier to complete |
| 506 | unsigned cnt = _barrier_count.load(); |
| 507 | while (0 < cnt) |
| 508 | { |
| 509 | _barrier_count.wait(cnt); |
| 510 | cnt = _barrier_count.load(); |
| 511 | } |
| 499 | 512 | } |
| 500 | 513 | |
| 501 | 514 | // We are so out of here. Nothing to do, just exit this thread. |