| … | … | @@ -162,7 +162,8 @@ class async_buffer |
| 162 | 162 | void insert(const Element&); |
| 163 | 163 | void flush(); |
| 164 | 164 | void barrier(void); |
| 165 | void barrier(const Element&); | |
| 165 | template<typename Factory> | |
| 166 | void barrier(Factory&& make_element); | |
| 166 | 167 | |
| 167 | 168 | void set_watermarks(size_t, size_t); |
| 168 | 169 | void stall(bool); |
| … | … | @@ -435,18 +436,23 @@ void async_buffer<Writer, Element>::barrier() |
| 435 | 436 | drain(); |
| 436 | 437 | } |
| 437 | 438 | |
| 438 | /// Barrier that ensures every worker processes the given element. | |
| 439 | /// Barrier that ensures every worker processes a provided element. | |
| 439 | 440 | /// This first drains existing work queues, then gives each worker |
| 440 | /// a copy of `elt` for processing. It then waits for all workers | |
| 441 | /// to complete. This guarantees that every worker has run at least | |
| 442 | /// once. | |
| 441 | /// an element provided by the factory. It then waits for all | |
| 442 | /// workers to complete. This guarantees that every worker has run | |
| 443 | /// at least once. | |
| 444 | /// | |
| 445 | /// The factory callable is invoked once per worker thread to create | |
| 446 | /// a unique element. Since the async_buffer is deduplicating for | |
| 447 | /// work elements, each element must be distinct. | |
| 443 | 448 | /// |
| 444 | 449 | /// The intended use case for this function are systems that have |
| 445 | 450 | /// workers that do caching or lazy initialization or implement |
| 446 | 451 | /// other actions that require synchronization that the default |
| 447 | 452 | /// `barrier(void)` would miss. |
| 448 | 453 | template<typename Writer, typename Element> |
| 449 | void async_buffer<Writer, Element>::barrier(const Element& elt) | |
| 454 | template<typename Factory> | |
| 455 | void async_buffer<Writer, Element>::barrier(Factory&& make_element) | |
| 450 | 456 | { |
| 451 | 457 | std::unique_lock<std::mutex> lock(_enqueue_mutex); |
| 452 | 458 | |
| … | … | @@ -460,7 +466,7 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 460 | 466 | |
| 461 | 467 | // Enqueue one element for each thread. |
| 462 | 468 | for (unsigned int i = 0; i < _thread_count; i++) |
| 463 | do_insert(elt); | |
| 469 | do_insert(make_element()); | |
| 464 | 470 | |
| 465 | 471 | // Wait for all threads to complete their barrier work |
| 466 | 472 | completion.wait(); |
| … | … | @@ -135,7 +135,8 @@ class async_caller |
| 135 | 135 | void enqueue(Element&&); |
| 136 | 136 | void flush_queue(); |
| 137 | 137 | void barrier(void); |
| 138 | void barrier(const Element&); | |
| 138 | template<typename Factory> | |
| 139 | void barrier(Factory&& make_element); | |
| 139 | 140 | |
| 140 | 141 | void set_watermarks(size_t, size_t); |
| 141 | 142 | |
| … | … | @@ -349,18 +350,22 @@ void async_caller<Writer, Element>::barrier() |
| 349 | 350 | drain(); |
| 350 | 351 | } |
| 351 | 352 | |
| 352 | /// Barrier that ensures every worker processes the given element. | |
| 353 | /// Barrier that ensures every worker processes an element. | |
| 353 | 354 | /// This first drains existing work queues, then gives each worker |
| 354 | /// a copy of `elt` for processing. It then waits for all workers | |
| 355 | /// to complete. This guarantees that every worker has run at least | |
| 356 | /// once. | |
| 355 | /// an element provided by the factory. It then waits for all | |
| 356 | /// workers to complete. This guarantees that every worker has run | |
| 357 | /// at least once. | |
| 358 | /// | |
| 359 | /// The factory callable is invoked once per worker thread to create | |
| 360 | /// an element. | |
| 357 | 361 | /// |
| 358 | 362 | /// The intended use case for this function are systems that have |
| 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. |
| 362 | 366 | template<typename Writer, typename Element> |
| 363 | void async_caller<Writer, Element>::barrier(const Element& elt) | |
| 367 | template<typename Factory> | |
| 368 | void async_caller<Writer, Element>::barrier(Factory&& make_element) | |
| 364 | 369 | { |
| 365 | 370 | std::unique_lock<std::mutex> lock(_enqueue_mutex); |
| 366 | 371 | |
| … | … | @@ -376,7 +381,7 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 376 | 381 | for (unsigned int i = 0; i < _thread_count; i++) |
| 377 | 382 | { |
| 378 | 383 | _pending ++; |
| 379 | _store_queue.push(elt); | |
| 384 | _store_queue.push(make_element()); | |
| 380 | 385 | _item_count++; |
| 381 | 386 | } |
| 382 | 387 |