Method ReloadTopicConfig

phxqueue/scheduler/schedulermgr.cpp:369–435

}

comm::RetCode SchedulerMgr::ReloadTopicConfig() {
    set<int> topic_ids;
    comm::RetCode ret{config::GlobalConfig::GetThreadInstance()->GetAllTopicID(topic_ids)};
    if (comm::RetCode::RET_OK != ret) {
        QLErr("GetAllTopic err %d", ret);

        return ret;
    }

    QLVerb("nr_topic %zu", topic_ids.size());

    // check map update
    bool modified{false};
    {
        comm::utils::RWLock rwlock_read(impl_->rwlock, comm::utils::RWLock::LockMode::READ);
        if (impl_->topic_id2data_map.size() != topic_ids.size()) {
            modified = true;
        } else {
            for (const int topic_id : topic_ids) {
                if (impl_->topic_id2data_map.end() == impl_->topic_id2data_map.find(topic_id)) {
                    modified = true;

                    break;
                }
            }
        }
    }

    // map updated
    if (modified) {
        comm::utils::RWLock rwlock_write(impl_->rwlock, comm::utils::RWLock::LockMode::WRITE);

        // add
        for (const int topic_id : topic_ids) {
            if (impl_->topic_id2data_map.end() == impl_->topic_id2data_map.find(topic_id)) {
                impl_->topic_id2lock_map.emplace(topic_id, move(TopicLock()));
                impl_->topic_id2data_map.emplace(topic_id, move(TopicData()));
            }
        }

        // remove
        if (topic_ids.size() != impl_->topic_id2data_map.size()) {
            set<int> topic_id_set;
            for (const auto &topic_id : topic_ids) {
                topic_id_set.emplace(topic_id);
            }
            for (auto it(impl_->topic_id2data_map.begin());
                 impl_->topic_id2data_map.end() != it; ++it) {
                if (topic_id_set.end() == topic_id_set.find(it->first)) {
                    const int topic_id{it->first};
                    auto topic_id2lock_it(impl_->topic_id2lock_map.find(topic_id));
                    if (impl_->topic_id2lock_map.end() == topic_id2lock_it) {
                        QLErr("topic %d lock lost", it->first);

                        continue;
                    }
                    // TODO: check if deadlock
                    comm::utils::RWLock rwlock_write(topic_id2lock_it->second.rwlock,