Method KeepMaster

phxqueue/scheduler/keepmasterthread.cpp:120–211

}

comm::RetCode KeepMasterThread::KeepMaster() {
    const int topic_id{impl_->scheduler->GetTopicID()};

    comm::SchedulerKeepMasterBP::GetThreadInstance()->OnKeepMaster(topic_id);

    if (impl_->scheduler->NeedDropMaster()) {
        QLInfo("MASTERSTAT: NeedDropMaster return true");

        return comm::RetCode::RET_ERR_SVR_BLOCK;
    }

    comm::RetCode ret;

    string client_id{impl_->scheduler->GetSchedulerOption()->ip};

    int lock_id{-1};
    if (comm::RetCode::RET_OK != (ret = GetLockID(lock_id))) {
        QLErr("GetLockID ret %d", as_integer(ret));

        return ret;
    }

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

        return ret;
    }

    const string lock_key(topic_config->GetProto().topic().scheduler_lock_key());

    static thread_local uint64_t version{0};
    static thread_local uint64_t overdue_time_ms{0};
    uint64_t now{comm::utils::Time::GetSteadyClockMS()};
    bool need_acquire_lock{false};
    {
        comm::proto::GetLockInfoRequest req;
        comm::proto::GetLockInfoResponse resp;
        req.set_topic_id(topic_id);
        req.set_lock_id(lock_id);
        req.set_lock_key(lock_key);
        ret = impl_->scheduler->RawGetLockInfo(req, resp);
        if (comm::RetCode::RET_ERR_KEY_NOT_EXIST != ret && comm::RetCode::RET_OK != ret) {
            QLErr("MASTERSTAT: RawGetLockInfo err %d", as_integer(ret));
            comm::SchedulerKeepMasterBP::GetThreadInstance()->OnRawGetLockInfoFail(topic_id, req);

            return ret;
        }
        comm::SchedulerKeepMasterBP::GetThreadInstance()->OnRawGetLockInfoSucc(topic_id, req, resp);

        QLInfo("MASTERSTAT: RawGetLockInfo ret %d lock_key %s old_version %" PRIu64 " old_overdue_time_ms %" PRIu64, as_integer(ret), lock_key.c_str(), version, overdue_time_ms);
        if (comm::RetCode::RET_ERR_KEY_NOT_EXIST == ret) {
            version = 0;
            overdue_time_ms = now;
        } else if (version != resp.lock_info().version()) {
            version = resp.lock_info().version();
            overdue_time_ms = now + resp.lock_info().lease_time_ms();
        }