#pragma once /** * @file RuleStatShm.h * @brief 共享内存 map,用于 mon↔cron 进程间冷数据交换 * * 热数据(display)改走本地缓存 → Memcached,不经过共享内存。 * 冷数据(stat_values/running_time/shear_times 等)保留在共享内存中, * 使用 boost::interprocess::interprocess_mutex 做真正的进程间同步。 * * @author your name (you@domain.com) * @version 0.2 * @date 2023-10-18 * * Copyright: Baosight Co. Ltd. * DO NOT COPY/USE WITHOUT PERMISSION */ #include #include #include "shm_header.h" namespace RuleStatShm { using namespace ShmHeader; namespace { using ColdMutex = bipc::interprocess_mutex; ColdMutex llmtx{}; ///< 进程间互斥锁 const static std::string dir_path = "/users/dsc/shm"; const static std::string shm_file = "MapRuleStat"; ///<映射文件名 const double data_size = 1024; ///< 数据大小 MB const size_t stat_size_min = 1000; ///<统计样本最小批处理量 const size_t stat_size_max = 10000; ///<统计样本最大储存量 static managed_mapped_file_t obj_mapped_file(bipc::open_or_create, (dir_path + "/" + shm_file + "_boost.mmap").c_str(), mix_cc::data_size::MB(data_size)); static void_allocator default_allocator(obj_mapped_file.get_segment_manager()); ///<默认分配器 static vec_allocator_s items_allocator(obj_mapped_file.get_segment_manager()); ///< vector_s分配器 thread_local static char_string tl_key_object("", default_allocator); thread_local static char_string tl_key_delete("", default_allocator); char_string &get_thread_local_key(const std::string &key) { tl_key_object = key.c_str(); return tl_key_object; } char_string &get_thread_local_delete_key(const std::string &key) { tl_key_delete = key.c_str(); return tl_key_delete; } } // namespace /// 本地完整数据(AlgBase 使用,标准分配器) struct RuleStatLocal { double alarm_value = 0; double current_value = 0; double limit_up = 0; double limit_down = 0; std::vector items; std::vector stat_values; bool fetch_mark = false; double running_time = 0; int64_t shear_times = 0; int64_t alarm_times = 0; std::string last_alarm_time = "无报警"; std::string dev_coder = "无"; std::string unit; }; /// 共享内存冷数据(mon↔cron 交换用) struct RuleStatCold { vector_d stat_values; ///<统计值,mon攒样本,cron消费 bool fetch_mark = false; ///<取数据标记 double running_time = 0; ///<累积的运行时间 int64_t shear_times = 0; ///<剪切次数 int64_t alarm_times = 0; ///<报警次数 bipc::string last_alarm_time = "无报警"; ///<上次报警时间 bipc::string dev_coder = "无"; ///<设备编码 RuleStatCold() : stat_values(default_allocator) {} double limit_precision(double data, int precision = 2) const { double factor = std::pow(10, precision); return std::round(data * factor) / factor; } bool update_static(const RuleStatCold &value) { try { running_time = value.running_time; shear_times = value.shear_times; alarm_times = value.alarm_times; last_alarm_time = value.last_alarm_time; dev_coder = value.dev_coder; } catch (const std::exception &e) { return false; } return true; } bool update_cold(const RuleStatCold &value) { try { stat_values = value.stat_values; fetch_mark = value.fetch_mark; running_time = value.running_time; shear_times = value.shear_times; } catch (...) { return false; } return true; } }; ///< key 是string的情况 typedef std::pair pair_s; typedef bipc::node_allocator allocator_s; typedef std::less less_s; typedef bipc::map MapRuleStat_s; typedef MapRuleStat_s::iterator map_iter_s; /// 共享内存 map 操作(仅冷数据) struct MapRuleStat { private: MapRuleStat_s *p_msg_map = obj_mapped_file.find_or_construct( shm_file.c_str())(less_s(), obj_mapped_file.get_segment_manager()); public: // ── mon 高频调用 ── bool add_stat_value(const std::string &key, const double &value) { try { bipc::scoped_lock guard(llmtx); if (p_msg_map->operator[](get_thread_local_key(key)).fetch_mark) { p_msg_map->operator[](get_thread_local_key(key)).stat_values.clear(); p_msg_map->operator[](get_thread_local_key(key)).fetch_mark = false; } if (p_msg_map->operator[](get_thread_local_key(key)).stat_values.size() < stat_size_max) { p_msg_map->operator[](get_thread_local_key(key)) .stat_values.push_back(value); } return true; } catch (const std::exception &e) { return false; } } /// mon 写入累积的冷字段(stat_values, running_time, shear_times) bool update_cold_fields(const std::string &key, const RuleStatCold &value) { bipc::scoped_lock guard(llmtx); try { if (p_msg_map->find(get_thread_local_key(key)) != p_msg_map->end()) { p_msg_map->operator[](get_thread_local_key(key)).update_cold(value); } else { p_msg_map->operator[](get_thread_local_key(key)) = value; } return true; } catch (const std::exception &e) { return false; } } /// cron 写入静态字段(running_time, alarm_times, last_alarm_time, dev_coder 等) bool update_static_fields(const std::string &key, const RuleStatCold &value) { bipc::scoped_lock guard(llmtx); try { if (p_msg_map->find(get_thread_local_key(key)) != p_msg_map->end()) { p_msg_map->operator[](get_thread_local_key(key)).update_static(value); return true; } p_msg_map->operator[](get_thread_local_key(key)) = value; return true; } catch (const std::exception &e) { return false; } } // ── cron 调用 ── bool get_stat_value(const std::string &key, RuleStatCold &value) { try { bipc::scoped_lock guard(llmtx); if (!p_msg_map->operator[](get_thread_local_key(key)).fetch_mark && p_msg_map->operator[](get_thread_local_key(key)).stat_values.size() > stat_size_min) { value.stat_values = p_msg_map->operator[](get_thread_local_key(key)).stat_values; p_msg_map->operator[](get_thread_local_key(key)).fetch_mark = true; return true; } return false; } catch (const std::exception &e) { return false; } } // ── 管理操作 ── bool delete_data(const std::string &key) { bipc::scoped_lock guard(llmtx); try { if (p_msg_map->find(get_thread_local_delete_key(key)) != p_msg_map->end()) { p_msg_map->erase(get_thread_local_delete_key(key)); } return true; } catch (const std::exception &e) { return false; } } size_t size() { bipc::scoped_lock guard(llmtx); return p_msg_map->size(); } bool empty() { bipc::scoped_lock guard(llmtx); return p_msg_map->empty(); } std::vector find_rule_id(const std::vector &ruleid) { bipc::scoped_lock guard(llmtx); std::vector res; if (p_msg_map->empty()) { return {}; } for (auto iter1 = p_msg_map->begin(); iter1 != p_msg_map->end(); iter1++) { auto resV = std::find(ruleid.begin(), ruleid.end(), iter1->first); if (resV != ruleid.end()) { res.push_back(iter1->first); } } return res; } std::vector find_no_rule_id(const std::vector &ruleid) { bipc::scoped_lock guard(llmtx); std::vector res; if (p_msg_map->empty()) { return {}; } for (auto iter1 = p_msg_map->begin(); iter1 != p_msg_map->end(); iter1++) { auto resV = std::find(ruleid.begin(), ruleid.end(), iter1->first); if (resV == ruleid.end()) { res.push_back(iter1->first); } } return res; } }; } // namespace RuleStatShm