Method CheckMem

phxqueue/consumer/consumer.cpp:950–992

}

void Consumer::CheckMem(const int vpid) {
    comm::RetCode ret;

    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 topic_id %d", comm::as_integer(ret), impl_->topic_id);
        return;
    }

    auto mem_size_limit = topic_config->GetProto().topic().consumer_max_mem_size_mb_per_proc();
    if (!mem_size_limit) return;

    static time_t last_check_time = 0;
    auto now = time(NULL);
    if (last_check_time + 60 < now) {
        last_check_time = now;

        comm::ConsumerBP::GetThreadInstance()->OnMemCheck(impl_->topic_id);

        comm::utils::MemStat mem_stat;
        if (!mem_stat.Stat()) return;

        if (mem_stat.resident > mem_stat.share) {
            auto mem_size = (mem_stat.resident - mem_stat.share) / 256;
            QLInfo("mem_size %d mem_size_limit %d", mem_size, mem_size_limit);
            uint32_t fixed_limit = mem_size_limit + mem_size_limit / 100.0 * (vpid % 20);
            if (mem_size > fixed_limit) {
                comm::ConsumerBP::GetThreadInstance()->OnMemCheckUnpass(impl_->topic_id);

                QLErr("ERR: memory size %u > %u MB, kill it, res %lu, share %lu, size %lu, mem_size_limit %u",
                      mem_size,
                      fixed_limit,
                      mem_stat.resident,
                      mem_stat.share,
                      mem_stat.size,
                      mem_size_limit);
                exit(-1);
            }
        }

        comm::ConsumerBP::GetThreadInstance()->OnMemCheckPass(impl_->topic_id);
    }
}

void Consumer::CustomGetRequest(const comm::proto::ConsumerContext &cc,
                                const comm::proto::GetRequest &req,