Method Sync

phxqueue/consumer/hblock.cpp:552–626

comm::RetCode HeartBeatLock::Sync() {
    QLInfo("Sync begin");

    comm::RetCode ret;

    const int topic_id = impl_->consumer->GetTopicID();

    UpdateProcUsed();

    comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnSync(topic_id);
    comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnProcUsed(topic_id, impl_->nproc, impl_->proc_used);
    if (impl_->proc_used >= impl_->nproc) {
        comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnProcUsedExceed(topic_id, impl_->nproc, impl_->proc_used);
        QLErr("ERR: impl_->proc_used(%d) >= nproc(%d)", impl_->proc_used, impl_->nproc);
    }

    QLInfo("nproc %d proc_used %d", impl_->nproc, impl_->proc_used);

    SubID2AddrScales sub_id2addr_scales;
    if (comm::RetCode::RET_OK != (ret = GetAddrScale(sub_id2addr_scales))) {
        QLErr("ERR: GetAddrScale ret %d", comm::as_integer(ret));
        return ret;
    }

    std::set<int> valid_sub_ids;
    for (auto &&kv : sub_id2addr_scales) {
        valid_sub_ids.insert(kv.first);
    }
    ClearInvalidSubIDs(valid_sub_ids);

    uint64_t conf_last_mod_time = config::GlobalConfig::GetThreadInstance()->GetLastModTime(topic_id);

    map<int, vector<Queue_t> > sub_id2pending_queues;
    for (auto &&it : sub_id2addr_scales) {
        auto &&sub_id = it.first;
        auto &&addr_scale = it.second;

        QLVerb("sub_id %d addr_scale.size %zu", sub_id, addr_scale.size());

        uint64_t scale_hash = CalHash(addr_scale);

        comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnScaleHash(topic_id, sub_id, scale_hash);

        if (LOCK_BUF_MAGIC == impl_->buf->magic &&
            conf_last_mod_time == impl_->conf_last_mod_time &&
            scale_hash == impl_->scale_hashs[sub_id]) {
            QLInfo("no need to adjust scale. sub_id %d scale_hash %" PRIu64, sub_id, scale_hash);
            continue;
        }

        comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnAdjustScale(topic_id, sub_id);

        std::vector<Queue_t> all_queues;
        if (comm::RetCode::RET_OK != (ret = GetAllQueues(sub_id, all_queues))) {
            QLErr("ERR: GetAllQueues ret %d", comm::as_integer(ret));
            return ret;
        }