| 526 | } |
| 527 | |
| 528 | bool TaskControl::steal_task(bthread_t* tid, size_t* seed, size_t offset) { |
| 529 | auto tag = tls_task_group->tag(); |
| 530 | |
| 531 | if (_priority_queues[tag].steal(tid)) { |
| 532 | return true; |
| 533 | } |
| 534 | |
| 535 | // 1: Acquiring fence is paired with releasing fence in _add_group to |
| 536 | // avoid accessing uninitialized slot of _groups. |
| 537 | const size_t ngroup = tag_ngroup(tag).load(butil::memory_order_acquire/*1*/); |
| 538 | if (0 == ngroup) { |
| 539 | return false; |
| 540 | } |
| 541 | |
| 542 | // NOTE: Don't return inside `for' iteration since we need to update |seed| |
| 543 | bool stolen = false; |
| 544 | size_t s = *seed; |
| 545 | auto& groups = tag_group(tag); |
| 546 | for (size_t i = 0; i < ngroup; ++i, s += offset) { |
| 547 | TaskGroup* g = groups[s % ngroup]; |
| 548 | // g is possibly NULL because of concurrent _destroy_group |
| 549 | if (g) { |
| 550 | if (g->_rq.steal(tid)) { |
| 551 | stolen = true; |
| 552 | break; |
| 553 | } |
| 554 | if (g->_remote_rq.pop(tid)) { |
| 555 | stolen = true; |
| 556 | break; |
| 557 | } |
| 558 | } |
| 559 | } |
| 560 | *seed = s; |
| 561 | return stolen; |
| 562 | } |
| 563 | |
| 564 | void TaskControl::signal_task(int num_task, bthread_tag_t tag) { |
| 565 | if (num_task <= 0) { |