Method DistubePendingQueues

phxqueue/consumer/hblock.cpp:195–273

}

void HeartBeatLock::DistubePendingQueues(const map<int, vector<Queue_t>> &sub_id2pending_queues) {
    //if (impl_->ossid) OssAttrInc(impl_->ossid, 20, 1);

    comm::utils::RWLock l(impl_->rwlock, comm::utils::RWLock::LockMode::WRITE);

    for (auto &&kv : sub_id2pending_queues) {
        auto &&sub_id = kv.first;
        auto &&pending_queues = kv.second;

        vector<bool> done;
        done.resize(pending_queues.size(), false);

        size_t idx;

        // if there is a proc handling a queue in pending_queues, keep doing that. prevent from lock switching.
        for (int vpid{0}; vpid < impl_->nproc; ++vpid) {
            Queue_t *queue{&impl_->buf->queues[vpid]};

            if (LOCK_ITEM_MAGIC == queue->magic && sub_id == queue->sub_id) {
                for (idx = 0; idx < pending_queues.size(); ++idx) {
                    auto &&pending_queue = pending_queues[idx];
                    if (queue->sub_id == pending_queue.sub_id &&
                        queue->store_id == pending_queue.store_id &&
                        queue->queue_id == pending_queue.queue_id) {

                        if (!done[idx]) {
                            done[idx] = true;
                            QLInfo("QUEUEINFO: vpid %d keep sub %u store %u queue %u", vpid, queue->sub_id, queue->store_id, queue->queue_id);
                        } else {
                            memset(queue, 0, sizeof(Queue_t));
                        }

                        break;
                    }
                }
                if (pending_queues.size() == idx) {
                    memset(queue, 0, sizeof(Queue_t));
                }
            }
        }

        // the left nproc handle the left queues in pending_queues
        for (int vpid{0}; vpid < impl_->nproc; ++vpid) {
            Queue_t *queue{&impl_->buf->queues[vpid]};

            if (LOCK_ITEM_MAGIC != queue->magic) {
                idx = 0;
                for (; idx < pending_queues.size(); ++idx) {
                    if (!done[idx]) break;
                }
                if (idx == pending_queues.size()) {
                    queue->magic = 0;
                } else {
                    auto &&pending_queue = pending_queues[idx];

                    queue->magic = LOCK_ITEM_MAGIC;
                    queue->sub_id = pending_queue.sub_id;
                    queue->store_id = pending_queue.store_id;