Method UpdateLimitInfo

phxqueue/consumer/freqman.cpp:184–418

comm::RetCode FreqMan::UpdateLimitInfo() {
    QLVerb("start");

    comm::RetCode ret;

    uint64_t now = comm::utils::Time::GetSteadyClockMS();
    uint64_t diff_time_ms = now - impl_->last_update_time;
    impl_->last_update_time = now;

    auto &&opt(impl_->consumer->GetConsumerOption());

    shared_ptr<const config::TopicConfig> topic_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(impl_->topic_id, topic_config))) {
        QLErr("GetTopicConfigByTopicID ret %d", as_integer(ret));
        return ret;
    }
    const int batch_limit = topic_config->GetProto().topic().batch_limit();

    shared_ptr<const config::StoreConfig> store_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetStoreConfig(impl_->topic_id, store_config))) {
        QLErr("GetStoreConfig ret %d", as_integer(ret));
        return ret;
    }

    set<int> store_ids;
    if (comm::RetCode::RET_OK != (ret = store_config->GetAllStoreID(store_ids))) {
        QLErr("GetAllStoreID ret %d", as_integer(ret));
        return ret;
    }

    vector<shared_ptr<config::proto::FreqInfo> > freq_infos;
    if (comm::RetCode::RET_OK != (ret = topic_config->GetAllFreqInfo(freq_infos))) {
        QLErr("GetAllFreqInfo ret %d", as_integer(ret));
        return ret;
    }

    vector<Queue_t> queues;
    if (comm::RetCode::RET_OK != (ret = impl_->consumer->GetQueuesDistribute(queues))) {
        QLErr("GetQueuesDistribute ret %d", as_integer(ret));
        return ret;
    }

    if (queues.size() != opt->nprocs) {
        QLErr("queues.size(%d) != opt->nprocs(%d)", queues.size(), opt->nprocs);
        return comm::RetCode::RET_ERR_UNEXPECTED;
    }

    unique_ptr<bool[]> updated(new bool[opt->nprocs]);
    memset(updated.get(), 0, sizeof(bool) * opt->nprocs);

    map<tuple<int, int, int>, double> pub_sub_queue_info_id2max_handle_pct;
    for (auto &&freq_info : freq_infos) {
        auto pub_id = freq_info->pub_id();
        auto sub_id = freq_info->sub_id();
        auto handle_id = freq_info->handle_id();
        auto limit_per_min = freq_info->limit_per_min();
        auto queue_info_id = freq_info->queue_info_id();