Method Lock

phxqueue/consumer/hblock.cpp:772–807

}

comm::RetCode HeartBeatLock::Lock(const int vpid, int &sub_id, int &store_id, int &queue_id) {
    auto &&opt = impl_->consumer->GetConsumerOption();

    auto topic_id = impl_->consumer->GetTopicID();

    comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnLock(topic_id, sub_id, store_id, queue_id);

    if (vpid >= impl_->nproc) {
        QLErr("ERR: vpid err. vpid %u impl_->nproc %u", vpid, impl_->nproc);
        comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnProcLack(topic_id);
        exit(-1);
    }

    Queue_t queue;
    {
        comm::utils::RWLock l(impl_->rwlock, comm::utils::RWLock::LockMode::READ);
        memcpy(&queue, &impl_->buf->queues[vpid], sizeof(Queue_t));
    }
    QLVerb("vpid %u queue sub_id %u store_id %u queue_id %u", vpid, queue.sub_id, queue.store_id, queue.queue_id);


    comm::RetCode ret;
    if (comm::RetCode::RET_OK != (ret = DoLock(vpid, &queue))) {
        if (comm::as_integer(ret) < 0) {
            QLErr("ERR: DoLock ret %d vpid %u", comm::as_integer(ret), vpid);
            comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnLockFail(topic_id, sub_id, store_id, queue_id);
        }
        return ret;
    }
    sub_id = queue.sub_id;
    store_id = queue.store_id;
    queue_id = queue.queue_id;

    comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnLockSucc(topic_id, sub_id, store_id, queue_id);
    return comm::RetCode::RET_OK;
}

comm::RetCode HeartBeatLock::GetQueuesDistribute(vector<Queue_t> &queues) {
    queues.clear();