| 640 | } |
| 641 | |
| 642 | comm::RetCode SchedulerMgr::UpdateLive(const int topic_id, TopicData &topic_data, |
| 643 | TopicLock &topic_lock, const uint64_t now, bool &modified) { |
| 644 | comm::utils::RWLock rwlock_read(topic_lock.rwlock, comm::utils::RWLock::LockMode::READ); |
| 645 | |
| 646 | comm::RetCode ret; |
| 647 | |
| 648 | shared_ptr<const config::TopicConfig> topic_config; |
| 649 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()-> |
| 650 | GetTopicConfigByTopicID(topic_id, topic_config))) { |
| 651 | QLErr("GetTopicConfigByTopicID ret %d", as_integer(ret)); |
| 652 | |
| 653 | return ret; |
| 654 | } |
| 655 | |
| 656 | modified = false; |
| 657 | for (auto &&consumer_addr2info_kv : topic_data.consumer_addr2info_map) { |
| 658 | ConsumerInfo &consumer_info(consumer_addr2info_kv.second); |
| 659 | |
| 660 | // new die |
| 661 | if (consumer_info.live && consumer_info.last_active_time() + |
| 662 | topic_config->GetProto().topic().scheduler_hash_keep_time_s() * 1000 <= now) { |
| 663 | consumer_info.Reset(); |
| 664 | consumer_info.live = false; |
| 665 | // TODO: |
| 666 | //OssAttrInc(oss_attr_id(), 36u, 1u); |
| 667 | modified = true; |
| 668 | QLInfo("topic %d consumer %s new_die last_active %" PRIu64 " now %" PRIu64, topic_id, |
| 669 | comm::utils::EncodedAddrToIPString(consumer_addr2info_kv.first).c_str(), |
| 670 | consumer_info.last_active_time(), now); |
| 671 | |
| 672 | comm::proto::Addr addr; |
| 673 | comm::utils::DecodeAddr(consumer_addr2info_kv.first, addr); |
| 674 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnConsumerNewDie(topic_id, addr); |
| 675 | |
| 676 | continue; |
| 677 | } |
| 678 | |
| 679 | // new live |
| 680 | if (!consumer_info.live && consumer_info.last_active_time() + |
| 681 | topic_config->GetProto().topic().scheduler_hash_keep_time_s() * 1000 > now) { |
| 682 | consumer_info.live = true; |
| 683 | // TODO: |
| 684 | //OssAttrInc(oss_attr_id(), 35u, 1u); |
| 685 | modified = true; |
| 686 | QLInfo("topic %d consumer %s new_live last_active %" PRIu64 " now %" PRIu64, |
| 687 | topic_id, comm::utils::EncodedAddrToIPString(consumer_addr2info_kv.first).c_str(), |
| 688 | consumer_info.last_active_time(), now); |
| 689 | |
| 690 | comm::proto::Addr addr; |
| 691 | comm::utils::DecodeAddr(consumer_addr2info_kv.first, addr); |
| 692 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnConsumerNewLive(topic_id, addr); |
| 693 | |
| 694 | continue; |
| 695 | } |
| 696 | } |
| 697 | |
| 698 | return comm::RetCode::RET_OK; |
| 699 | } |
nothing calls this directly
no test coverage detected