| … | … | @@ -143,7 +143,7 @@ class async_buffer |
| 143 | 143 | |
| 144 | 144 | // Barrier synchronization |
| 145 | 145 | const Element* _current_barrier; |
| 146 | std::atomic<unsigned> _barrier_count; | |
| 146 | std::atomic<int> _barrier_count; | |
| 147 | 147 | |
| 148 | 148 | void start_writer_thread(); |
| 149 | 149 | void stop_writer_threads(); |
| … | … | @@ -463,18 +463,33 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 463 | 463 | |
| 464 | 464 | _store_set.cancel(); |
| 465 | 465 | |
| 466 | // Wait for all workers to finish processing | |
| 467 | unsigned cnt = _barrier_count.load(); | |
| 466 | // Wait for all workers to finish processing the barrier element. | |
| 467 | // Workers decrement from _thread_count+1 down to 1, then wait. | |
| 468 | int cnt = _barrier_count.load(); | |
| 468 | 469 | while (1 < cnt) |
| 469 | 470 | { |
| 470 | 471 | _barrier_count.wait(cnt); |
| 471 | 472 | cnt = _barrier_count.load(); |
| 472 | 473 | } |
| 473 | 474 | |
| 475 | // All workers have checked in. Unblock them. | |
| 474 | 476 | _current_barrier = nullptr; |
| 475 | 477 | _store_set.cancel_reset(); |
| 476 | _barrier_count--; | |
| 478 | _barrier_count--; // Now 0 | |
| 477 | 479 | _barrier_count.notify_all(); |
| 480 | ||
| 481 | // Wait for all workers to acknowledge and resume. | |
| 482 | // Each worker decrements once more after waking, going negative. | |
| 483 | // All workers have started when count reaches minus _thread_count | |
| 484 | // Failure to wait, here, will result in confusion if another | |
| 485 | // barrier arrives. | |
| 486 | int target = -static_cast<int>(_thread_count); | |
| 487 | cnt = _barrier_count.load(); | |
| 488 | while (target < cnt) | |
| 489 | { | |
| 490 | _barrier_count.wait(cnt); | |
| 491 | cnt = _barrier_count.load(); | |
| 492 | } | |
| 478 | 493 | } |
| 479 | 494 | |
| 480 | 495 | /// A single write thread. Reads elements from set, and invokes the |
| … | … | @@ -507,18 +522,25 @@ void async_buffer<Writer, Element>::write_loop() |
| 507 | 522 | { |
| 508 | 523 | (_writer->*_do_write)(*_current_barrier); |
| 509 | 524 | |
| 510 | unsigned int old_count = _barrier_count.fetch_sub(1); | |
| 525 | // Last one out tells the master thread. | |
| 526 | int old_count = _barrier_count.fetch_sub(1); | |
| 511 | 527 | if (2 == old_count) |
| 512 | 528 | _barrier_count.notify_all(); |
| 513 | 529 | |
| 514 | 530 | // barrier() will drop the count to zero after |
| 515 | 531 | // all workers have run. |
| 516 | unsigned int cnt = _barrier_count.load(); | |
| 532 | int cnt = _barrier_count.load(); | |
| 517 | 533 | while (0 < cnt) |
| 518 | 534 | { |
| 519 | 535 | _barrier_count.wait(cnt); |
| 520 | 536 | cnt = _barrier_count.load(); |
| 521 | 537 | } |
| 538 | ||
| 539 | // Ack that we're resuming normal operation. | |
| 540 | // Last one notifies the master thread. | |
| 541 | old_count = _barrier_count.fetch_sub(1); | |
| 542 | if (1 - static_cast<int>(_thread_count) == old_count) | |
| 543 | _barrier_count.notify_all(); | |
| 522 | 544 | } |
| 523 | 545 | |
| 524 | 546 | // We are so out of here. Nothing to do, just exit this thread. |
| … | … | @@ -116,7 +116,7 @@ class async_caller |
| 116 | 116 | |
| 117 | 117 | // Barrier synchronization |
| 118 | 118 | const Element* _current_barrier; |
| 119 | std::atomic<unsigned> _barrier_count; | |
| 119 | std::atomic<int> _barrier_count; | |
| 120 | 120 | |
| 121 | 121 | void start_writer_thread(); |
| 122 | 122 | void stop_writer_threads(); |
| … | … | @@ -376,18 +376,33 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 376 | 376 | |
| 377 | 377 | _store_queue.cancel(); |
| 378 | 378 | |
| 379 | // Wait for all workers to finish processing | |
| 380 | unsigned cnt = _barrier_count.load(); | |
| 379 | // Wait for all workers to finish processing the barrier element. | |
| 380 | // Workers decrement from _thread_count+1 down to 1, then wait. | |
| 381 | int cnt = _barrier_count.load(); | |
| 381 | 382 | while (1 < cnt) |
| 382 | 383 | { |
| 383 | 384 | _barrier_count.wait(cnt); |
| 384 | 385 | cnt = _barrier_count.load(); |
| 385 | 386 | } |
| 386 | 387 | |
| 388 | // All workers have checked in. Unblock them. | |
| 387 | 389 | _current_barrier = nullptr; |
| 388 | 390 | _store_queue.cancel_reset(); |
| 389 | _barrier_count--; | |
| 391 | _barrier_count--; // Now 0 | |
| 390 | 392 | _barrier_count.notify_all(); |
| 393 | ||
| 394 | // Wait for all workers to acknowledge and resume. | |
| 395 | // Each worker decrements once more after waking, going negative. | |
| 396 | // All workers have started when count reaches minus _thread_count | |
| 397 | // Failure to wait, here, will result in confusion if another | |
| 398 | // barrier arrives. | |
| 399 | int target = -static_cast<int>(_thread_count); | |
| 400 | cnt = _barrier_count.load(); | |
| 401 | while (target < cnt) | |
| 402 | { | |
| 403 | _barrier_count.wait(cnt); | |
| 404 | cnt = _barrier_count.load(); | |
| 405 | } | |
| 391 | 406 | } |
| 392 | 407 | |
| 393 | 408 | /// A single write thread. Reads elements from queue, and invokes the |
| … | … | @@ -414,18 +429,25 @@ void async_caller<Writer, Element>::write_loop() |
| 414 | 429 | { |
| 415 | 430 | (_writer->*_do_write)(*_current_barrier); |
| 416 | 431 | |
| 417 | unsigned int old_count = _barrier_count.fetch_sub(1); | |
| 432 | // Last one out tells the master thread. | |
| 433 | int old_count = _barrier_count.fetch_sub(1); | |
| 418 | 434 | if (2 == old_count) |
| 419 | 435 | _barrier_count.notify_all(); |
| 420 | 436 | |
| 421 | 437 | // barrier() will drop the count to zero after |
| 422 | 438 | // all workers have run. |
| 423 | unsigned int cnt = _barrier_count.load(); | |
| 439 | int cnt = _barrier_count.load(); | |
| 424 | 440 | while (0 < cnt) |
| 425 | 441 | { |
| 426 | 442 | _barrier_count.wait(cnt); |
| 427 | 443 | cnt = _barrier_count.load(); |
| 428 | 444 | } |
| 445 | ||
| 446 | // Ack that we're resuming normal operation. | |
| 447 | // Last one notifies the master thread. | |
| 448 | old_count = _barrier_count.fetch_sub(1); | |
| 449 | if (1 - static_cast<int>(_thread_count) == old_count) | |
| 450 | _barrier_count.notify_all(); | |
| 429 | 451 | } |
| 430 | 452 | |
| 431 | 453 | // We are so out of here. Nothing to do, just exit this thread. |