Method RunSync

phxqueue/consumer/hblock.cpp:275–305

}

void HeartBeatLock::RunSync() {
    comm::RetCode ret;
    pid_t pid = fork();
    if (pid == 0) {

        prctl(PR_SET_PDEATHSIG, SIGHUP);

        impl_->consumer->OnRunSync();

        //oss_call_stat_attach(vpid);
        //mmlb_stat_attach(vpid);

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

        while (true) {
            if (0 > comm::as_integer(ret = Sync())) {
                QLErr("ERR: Sync fail. ret %d", comm::as_integer(ret));
                comm::ConsumerHeartBeatLockBP::GetThreadInstance()->OnSyncFail(topic_id);
            }

            shared_ptr<const config::TopicConfig> topic_config;
            if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(topic_id, topic_config))) {
                QLErr("ERR: GetTopicConfigByTopicID ret %d topic_id %u", ret, topic_id);
                continue;
            }

            sleep(topic_config->GetProto().topic().scheduler_get_scale_interval_s());
        }
        exit(0);
    }
}

static size_t CalHash(const vector<comm::proto::AddrScale> &addr_scales) {
    size_t h = crc32(0, Z_NULL, 0);