| … | … | @@ -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> |
| … | … | @@ -115,9 +114,9 @@ class async_caller |
| 115 | 114 | unsigned int _thread_count; |
| 116 | 115 | bool _stopping_writers; |
| 117 | 116 | |
| 118 | | // Per-thread barrier synchronization |
| 119 | | std::atomic<unsigned> _barrier_phase; |
| 120 | | std::latch* _barrier_latch; |
| 117 | // Barrier synchronization |
| 118 | const Element* _current_barrier; |
| 119 | std::atomic<unsigned> _barrier_count; |
| 121 | 120 | |
| 122 | 121 | void start_writer_thread(); |
| 123 | 122 | void stop_writer_threads(); |
| … | … | @@ -180,8 +179,8 @@ async_caller<Writer, Element>::async_caller(Writer* wr, |
| 180 | 179 | _busy_writers = 0; |
| 181 | 180 | _pending = 0; |
| 182 | 181 | _in_drain = false; |
| 183 | | _barrier_phase = 0; |
| 184 | | _barrier_latch = nullptr; |
| 182 | _current_barrier = nullptr; |
| 183 | _barrier_count = 0; |
| 185 | 184 | |
| 186 | 185 | _high_watermark = DEFAULT_HIGH_WATER_MARK; |
| 187 | 186 | _low_watermark = DEFAULT_LOW_WATER_MARK; |
| … | … | @@ -254,9 +253,14 @@ void async_caller<Writer, Element>::stop_writer_threads() |
| 254 | 253 | |
| 255 | 254 | _stopping_writers = true; |
| 256 | 255 | |
| 257 | | // Wait until the writer threads are (mostly) done. |
| 258 | | while (0 < _pending) |
| 259 | | std::this_thread::sleep_for(std::chrono::milliseconds(1)); |
| 256 | // Wait until the writer threads are (mostly) done. There |
| 257 | // might still be some lingering enqueues after this exits. |
| 258 | unsigned long pend = _pending.load(); |
| 259 | while (pend != 0) |
| 260 | { |
| 261 | _pending.wait(pend); |
| 262 | pend = _pending.load(); |
| 263 | } |
| 260 | 264 | |
| 261 | 265 | // Now tell all the threads that they are done. |
| 262 | 266 | // I.e. cancel all the threads. |
| … | … | @@ -278,8 +282,8 @@ void async_caller<Writer, Element>::stop_writer_threads() |
| 278 | 282 | (_writer->*_do_write)(elt); |
| 279 | 283 | } |
| 280 | 284 | |
| 281 | | _barrier_phase = 0; |
| 282 | | _barrier_latch = nullptr; |
| 285 | _current_barrier = nullptr; |
| 286 | _barrier_count = 0; |
| 283 | 287 | |
| 284 | 288 | // Its now OK to start new threads, if desired ...(!) |
| 285 | 289 | _stopping_writers = false; |
| … | … | @@ -359,30 +363,31 @@ void async_caller<Writer, Element>::barrier() |
| 359 | 363 | /// workers that do caching or lazy initialization or implement |
| 360 | 364 | /// other actions that require synchronization that the default |
| 361 | 365 | /// `barrier(void)` would miss. |
| 366 | /// |
| 362 | 367 | template<typename Writer, typename Element> |
| 363 | 368 | void async_caller<Writer, Element>::barrier(const Element& elt) |
| 364 | 369 | { |
| 365 | 370 | std::unique_lock<std::mutex> lock(_enqueue_mutex); |
| 366 | 371 | |
| 367 | | // First, drain existing work |
| 368 | 372 | drain(); |
| 369 | 373 | |
| 370 | | // Set up the latch for barrier completion |
| 371 | | std::latch completion(_thread_count); |
| 372 | | _barrier_latch = &completion; |
| 373 | | _barrier_phase++; |
| 374 | _barrier_count = _thread_count + 1; |
| 375 | _current_barrier = &elt; |
| 376 | |
| 377 | _store_queue.cancel(); |
| 374 | 378 | |
| 375 | | // Enqueue one element for each thread. |
| 376 | | for (unsigned int i = 0; i < _thread_count; i++) |
| 379 | // Wait for all workers to finish processing |
| 380 | unsigned cnt = _barrier_count.load(); |
| 381 | while (1 < cnt) |
| 377 | 382 | { |
| 378 | | _pending ++; |
| 379 | | _store_queue.push(elt); |
| 380 | | _item_count++; |
| 383 | _barrier_count.wait(cnt); |
| 384 | cnt = _barrier_count.load(); |
| 381 | 385 | } |
| 382 | 386 | |
| 383 | | // Wait for all threads to complete their barrier work |
| 384 | | completion.wait(); |
| 385 | | _barrier_latch = nullptr; |
| 387 | _current_barrier = nullptr; |
| 388 | _store_queue.cancel_reset(); |
| 389 | _barrier_count--; |
| 390 | _barrier_count.notify_all(); |
| 386 | 391 | } |
| 387 | 392 | |
| 388 | 393 | /// A single write thread. Reads elements from queue, and invokes the |
| … | … | @@ -390,41 +395,43 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 390 | 395 | template<typename Writer, typename Element> |
| 391 | 396 | void async_caller<Writer, Element>::write_loop() |
| 392 | 397 | { |
| 393 | | // Track which barrier phase this thread has completed. |
| 394 | | // Used to ensure each thread processes exactly one barrier element. |
| 395 | | static thread_local unsigned my_last_phase = 0; |
| 396 | | |
| 397 | | try |
| 398 | while (true) |
| 398 | 399 | { |
| 399 | | while (true) |
| 400 | try |
| 400 | 401 | { |
| 401 | 402 | Element elt = _store_queue.value_pop(); |
| 402 | 403 | _busy_writers ++; |
| 404 | (_writer->*_do_write)(elt); |
| 405 | _busy_writers --; |
| 403 | 406 | |
| 404 | | unsigned current_phase = _barrier_phase.load(); |
| 405 | | if (my_last_phase < current_phase) |
| 406 | | { |
| 407 | | // This is barrier work - process and count toward completion |
| 408 | | my_last_phase = current_phase; |
| 409 | | (_writer->*_do_write)(elt); |
| 410 | | _barrier_latch->count_down(); |
| 411 | | } |
| 412 | | else |
| 407 | unsigned long old_pend = _pending.fetch_sub(1); |
| 408 | if (1 == old_pend) |
| 409 | _pending.notify_all(); |
| 410 | } |
| 411 | catch (typename concurrent_queue<Element>::Canceled& e) |
| 412 | { |
| 413 | if (_current_barrier != nullptr) |
| 413 | 414 | { |
| 414 | | // Regular work |
| 415 | | (_writer->*_do_write)(elt); |
| 415 | (_writer->*_do_write)(*_current_barrier); |
| 416 | |
| 417 | unsigned int old_count = _barrier_count.fetch_sub(1); |
| 418 | if (2 == old_count) |
| 419 | _barrier_count.notify_all(); |
| 420 | |
| 421 | // barrier() will drop the count to zero after |
| 422 | // all workers have run. |
| 423 | unsigned int cnt = _barrier_count.load(); |
| 424 | while (0 < cnt) |
| 425 | { |
| 426 | _barrier_count.wait(cnt); |
| 427 | cnt = _barrier_count.load(); |
| 428 | } |
| 416 | 429 | } |
| 417 | 430 | |
| 418 | | _busy_writers --; |
| 419 | | _pending --; |
| 420 | | _pending.notify_all(); |
| 431 | // We are so out of here. Nothing to do, just exit this thread. |
| 432 | if (_stopping_writers) return; |
| 421 | 433 | } |
| 422 | 434 | } |
| 423 | | catch (typename concurrent_queue<Element>::Canceled& e) |
| 424 | | { |
| 425 | | // We are so out of here. Nothing to do, just exit this thread. |
| 426 | | return; |
| 427 | | } |
| 428 | 435 | } |
| 429 | 436 | |
| 430 | 437 | |