Method DoRun

phxqueue/lock/cleanthread.cpp:77–128

}

void CleanThread::DoRun() {
    if (nullptr == impl_->lock || nullptr == impl_->lock->GetNode()) {
        QLErr("lock %p", impl_->lock);

        return;
    }

    while (true) {
        if (impl_->stop) return;

        int nr_key{0};
        int nr_clean_key{0};
        const uint64_t now{comm::utils::Time::GetSteadyClockMS()};

        bool need_clean_key{false};
        if (now > last_clean_lock_ms_ + impl_->lock->GetLockOption()->clean_interval_s * 1000) {
            need_clean_key = true;
            last_clean_lock_ms_ = now;
        }

        bool need_write_restart_checkpoint{false};
        if (now > last_write_restart_checkpoint_ms_ +
            impl_->lock->GetLockOption()->idle_write_checkpoint_interval_s * 1000) {
            need_write_restart_checkpoint = true;
            last_write_restart_checkpoint_ms_ = now;
        }

        for (int paxos_group_id{0}; impl_->lock->GetLockOption()->nr_group > paxos_group_id;
             ++paxos_group_id) {

            if (need_clean_key) {
                int nr_group_key{0};
                int nr_group_clean_key{0};
                CleanLock(paxos_group_id, now, nr_group_key, nr_group_clean_key);
                nr_key += nr_group_key;
                nr_clean_key += nr_group_clean_key;
            }

            if (need_write_restart_checkpoint) {
                WriteRestartCheckpoint(paxos_group_id, now);
            }
        }  // foreach group

        QLInfo("now %llu nr_key %d nr_clean_key %d", now, nr_key, nr_clean_key);
        comm::LockCleanThreadBP::GetThreadInstance()->
                OnNrKey(impl_->lock->GetTopicID(), nr_key);
        comm::LockCleanThreadBP::GetThreadInstance()->
                OnNrCleanKey(impl_->lock->GetTopicID(), nr_clean_key);

        sleep(1);
    }
}

void CleanThread::CleanLock(const int paxos_group_id, const uint64_t now,
                            int &nr_group_key, int &nr_group_clean_key) {