| … | … | @@ -334,9 +334,14 @@ void async_buffer<Writer, Element>::stop_writer_threads() |
| 334 | 334 | |
| 335 | 335 | _stopping_writers = true; |
| 336 | 336 | |
| 337 | | // Wait until the writer threads are (mostly) done. |
| 338 | | while (0 < _pending) |
| 339 | | std::this_thread::sleep_for(std::chrono::milliseconds(1)); |
| 337 | // Wait until the writer threads are (mostly) done. There |
| 338 | // might still be some lingering enqueues after this exits. |
| 339 | unsigned long pend = _pending.load(); |
| 340 | while (pend != 0) |
| 341 | { |
| 342 | _pending.wait(pend); |
| 343 | pend = _pending.load(); |
| 344 | } |
| 340 | 345 | |
| 341 | 346 | // Now tell all the threads that they are done. |
| 342 | 347 | // I.e. cancel all the threads. |
| … | … | @@ -491,8 +496,10 @@ void async_buffer<Writer, Element>::write_loop() |
| 491 | 496 | _busy_writers ++; |
| 492 | 497 | (_writer->*_do_write)(elt); |
| 493 | 498 | _busy_writers --; |
| 494 | | _pending --; |
| 495 | | _pending.notify_all(); |
| 499 | |
| 500 | unsigned long old_pend = _pending.fetch_sub(1); |
| 501 | if (1 == old_pend) |
| 502 | _pending.notify_all(); |
| 496 | 503 | } |
| 497 | 504 | catch (typename concurrent_set<Element>::Canceled& e) |
| 498 | 505 | { |
| … | … | @@ -506,7 +513,7 @@ void async_buffer<Writer, Element>::write_loop() |
| 506 | 513 | |
| 507 | 514 | // barrier() will drop the count to zero after |
| 508 | 515 | // all workers have run. |
| 509 | | unsigned cnt = _barrier_count.load(); |
| 516 | unsigned int cnt = _barrier_count.load(); |
| 510 | 517 | while (0 < cnt) |
| 511 | 518 | { |
| 512 | 519 | _barrier_count.wait(cnt); |
| … | … | @@ -533,7 +540,10 @@ void async_buffer<Writer, Element>::do_insert(Element&& elt) |
| 533 | 540 | if (not inserted) |
| 534 | 541 | { |
| 535 | 542 | _duplicate_count++; |
| 536 | | _pending --; |
| 543 | |
| 544 | unsigned long old_pend = _pending.fetch_sub(1); |
| 545 | if (1 == old_pend) |
| 546 | _pending.notify_all(); |
| 537 | 547 | } |
| 538 | 548 | } |
| 539 | 549 | |