Method GetQueueByAddrScale

phxqueue/consumer/consumer.cpp:782–838

comm::RetCode
Consumer::GetQueueByAddrScale(const vector<consumer::Queue_t> &queues,
                              const consumer::AddrScales &addr_scales,
                              set<size_t> &queue_idxs) {
    queue_idxs.clear();

    map<uint32_t, uint64_t> hash2encoded_addr;

    comm::proto::Addr local_addr;
    local_addr.set_ip(impl_->opt.ip);
    local_addr.set_port(impl_->opt.port);

    uint64_t local_encoded_addr = comm::utils::EncodeAddr(local_addr);

    for (int i{0}; i < addr_scales.size(); ++i) {
        auto &addr = addr_scales[i].addr();
        auto scale = addr_scales[i].scale();

        uint64_t encoded_addr = comm::utils::EncodeAddr(addr);
        for (int j{0}; j < scale; ++j) {
            size_t h = 0;
            h = comm::utils::MurmurHash64(&encoded_addr, sizeof(uint64_t), h);
            h = comm::utils::MurmurHash64(&j, sizeof(int), h);

            
            hash2encoded_addr.emplace(h, encoded_addr);
            //QLInfo("insert h %u encoded_addr %" PRIu64 " j %d", h, encoded_addr, j);
        }
    }

    if (hash2encoded_addr.empty()) {
        QLWarn("hash2encoded_addr empty");
        return comm::RetCode::RET_OK;
    }

    for (int i{0}; i < queues.size(); ++i) {
        auto &&queue = queues[i];
        size_t h = 0;
        h = comm::utils::MurmurHash64(&queue.sub_id, sizeof(uint32_t), h);
        h = comm::utils::MurmurHash64(&queue.store_id, sizeof(uint32_t), h);
        h = comm::utils::MurmurHash64(&queue.queue_id, sizeof(uint32_t), h);


        auto &&it = hash2encoded_addr.lower_bound(h);
        if (hash2encoded_addr.end() == it) it = hash2encoded_addr.begin();

        //QLInfo("find h %u %" PRIu64, h, it->second);

        if (it->second == local_encoded_addr) {
            queue_idxs.insert(i);
        }
    }
    QLInfo("queues.size %zu addrscales.size %zu queue_idxs.size %zu local_encoded_addr %" PRIu64, queues.size(),
           addr_scales.size(), queue_idxs.size(), local_encoded_addr);

    return phxqueue::comm::RetCode::RET_OK;
}