fast_able/

high_speed_cache.rs

// 高频数据

use crossbeam::atomic::AtomicCell;
use crate::unsafe_cell_type::U;
use std::ops::Deref;
use std::{default::Default, usize};
use chrono::NaiveTime;
use crate::fasttime::b2_time::TimeExt;
use crate::fasttime::a8_micros;

pub trait CacheData: Default + Clone {
    fn time(&self) -> NaiveTime;
}

pub struct HighSpeedCache<T, const LEN: usize> {
    item: U<UnsafeData<T, LEN>>,
}

impl<T, const LEN: usize> Deref for HighSpeedCache<T, LEN> {
    type Target = UnsafeData<T, LEN>;
    fn deref(&self) -> &Self::Target {
        self.item.as_ref()
    }
}

pub struct UnsafeData<T, const LEN: usize> {
    items: [T; LEN],
    index: usize,
    defalut_val: T,
    pub count: usize,
    待补数据: AtomicCell<Option<(usize, T)>>,
}

impl<T, const LEN: usize> HighSpeedCache<T, LEN>
where
    T: CacheData,
{
    pub fn new() -> Self {
        // TODO: 生成 [T; LEN]
        // 当前报错: error[E0277]: the trait bound `T: std::marker::Copy` is not satisfied
        // 请在不实现 copy 的情况下, 帮我修改这里的代码, 让它编译通过
        let items: [T; LEN] = std::array::from_fn(|_i| T::default());
        let item = UnsafeData {
            items,
            index: 0,
            count: 0,
            defalut_val: T::default(),
            待补数据: AtomicCell::new(None),
        };
        Self { item: item.into() }
    }

    fn item_mut(&self) -> &mut UnsafeData<T, LEN> {
        self.item.as_mut()
    }

    /// 添加1项
    /// 请控制此方法不可并发访问
    pub fn push(&self, item_new: T) {
        let self_item = self.item_mut();
        if let Some((i, d)) = self.待补数据.swap(None) {
            self_item.items[i] = d;
        }

        if self_item.count == 0 {
            // 此时 index = 0, 对index不操作
        } else {
            self_item.index += 1;
            if self_item.index == LEN {
                self_item.index = 0;
            }
        }
        self_item.items[self.index] = item_new;
        self_item.count += 1;
    }

    /// 当前项
    pub fn current(&self) -> &T {
        if self.count == 0 {
            return &self.item.as_ref().defalut_val;
        }
        &self.items[self.index]
    }

    /// 请控制此方法不可并发访问
    pub fn clear(&self) {
        let self_item = self.item_mut();
        self_item.index = 0;
        self_item.count = 0;
        // 第1条数据要清空，并无实质数据。
        self_item.items[0] = T::default();
    }

    /// 传参：毫秒
    /// 如果找不到精确定位的数据, 则返回最新的那条数据
    pub fn find_找之前的项(&self, 毫秒数: i64, is_找不到即取最新数据: bool) -> Option<&T> {
        if self.count == 0 {
            return None;
        }
        let capacity = self.items.len();
        // 将微秒数转为 u64，但只在非负当前情况才如此
        let 微秒数 = if 毫秒数 >= 0 {
            毫秒数 as u64 * a8_micros::MICROS_PER_MILLIS
        } else {
            0
        };

        let ticks_当前项 = self.items[self.index].time().micros_of_day();
        let index_当前索引 = self.index;

        let mut find_最新 = Option::<&T>::None;
        for idx in (0..index_当前索引).rev() {
            let ticks_往前项 = self.items[idx].time().micros_of_day();
            if ticks_往前项 == 0 {
                continue;
            }
            let delta_往前微秒数 = ticks_当前项 - ticks_往前项;
            if delta_往前微秒数 > 微秒数 {
                return Some(&self.items[idx]);
            }

            // 修复: 如果找不到精确定位的数据, 则返回最新的那条数据; 从vec后面开始循环, 第一条数据就是最新的
            if is_找不到即取最新数据 {
                match &find_最新 {
                    None => {
                        find_最新 = Some(&self.items[idx]);
                    }
                    _ => (),
                }
            }
        }

        if self.count <= LEN {
            // 尚未走出一轮(填满 CAPACTIY 个项)，后面没有值
            return Some(find_最新.unwrap_or_else(|| &self.item.as_ref().defalut_val));
        }

        // 从数组后面开始找
        for idx in ((index_当前索引 + 1)..capacity).rev() {
            let item = &self.items[idx];
            let ticks_往前项 = item.time().micros_of_day();
            let delta_往前微秒数 = ticks_当前项 - ticks_往前项;
            if ticks_往前项 > 0 && delta_往前微秒数 > 微秒数 {
                return Some(item);

                // 修复: 找前封单时刚好在前一段找不到数据, 再从后面开始找, 匹配到最后一个数据, 但却拿到第一个下标的初始值
                // if idx + 1 == capacity {
                //     return &self.items[0];
                // } else {
                //     return item;
                // }
            }
        }

        return find_最新;
    }

    /// 如果十档或者前封单数据如果出现断档的情况，那么把前一个数据的时间戳改成现在数据并且减10毫秒
    /// 下次找数据时可以直接找到这一条
    // fn 向前补一个数据(&self) -> Option<T> {
    //     let idx = if self.index == 0 && self.count >= LEN {
    //         LEN - 1
    //     } else if self.index > 0 {
    //         self.index - 1
    //     } else {
    //         // index 等于零, 没有数据的情况
    //         return None;
    //     };

    //     let mut item = self.items[idx].clone();
    //     let millis = self.current().time().micros_of_day() - 10 * 1000;
    //     let time = NaiveTime::from_micros_day_unsafe(millis);
    //     item.set_time_发生时间(time);
    //     self.待补数据.store(Some((idx, item.clone())));
    //     return Some(item);
    // }

    pub fn find<F: Fn(&T) -> bool>(&self, f: F) -> Option<&T> {
        if self.count == 0 {
            return None;
        }

        let 当前项 = &self.items[self.index];
        if f(当前项) {
            return Some(当前项);
        }

        let index_当前索引 = self.index;
        for idx in (0..index_当前索引).rev() {
            if f(&self.items[idx]) {
                return Some(&self.items[idx]);
            }
        }
        if self.count <= LEN {
            // 尚未走出一轮(填满 CAPACTIY 个项)，后面没有值
            return None;
        }

        for idx in ((index_当前索引 + 1)..self.items.len()).rev() {
            if f(&self.items[idx]) {
                return Some(&self.items[idx]);
            }
        }
        None
    }

    /// 找到范围内最大的值
    /// n: 毫秒数
    /// is_return_last: true; 两秒内最大值找不到, 即找一个最新的数据
    /// compare: 比较两个值, 返回true则使用新的值
    /// 
    /// TODO: 不要使用补数据的逻辑, 找不到数据即取一个最新的数据
    pub fn find_limit_max<F: Fn(&T, &T) -> bool>(
        &self,
        n: i64,
        is_return_last: bool,
        compare: F,
    ) -> T {
        let items = self.list_items(n);
        let 补一个数据 = None;

        if (items.is_empty() || items.len() == 1) && is_return_last {
            // TODO: 注释掉不存在的方法调用
            // 补一个数据 = self.向前补一个数据();
        }

        if items.is_empty() {
            return 补一个数据.unwrap_or_else(|| self.defalut_val.clone());
        }

        let mut find_v = items[0];
        for item in items.iter().skip(1) {
            if compare(&find_v, item) {
                find_v = item;
            }
        }

        if let Some(v) = &补一个数据 {
            if compare(&find_v, v) {
                find_v = v;
            }
        }

        return find_v.clone();
    }

    /// 传参：找多少个
    pub fn list_items_len(&self, mut count: usize) -> Vec<&T> {
        if self.count == 0 {
            return vec![];
        }
        let mut items = vec![];

        let 当前项 = &self.items[self.index];
        items.push(当前项);

        let index_当前索引 = self.index;
        for idx in (0..index_当前索引).rev() {
            if count == 0 {
                return items;
            }
            items.push(&self.items[idx]);
            count -= 1;
        }

        if self.count <= LEN {
            // 尚未走出一轮(填满 CAPACTIY 个项)，后面没有值
            return items;
        }

        for idx in ((index_当前索引 + 1)..self.items.len()).rev() {
            if count == 0 {
                return items;
            }
            items.push(&self.items[idx]);
            count -= 1;
        }
        items
    }

    /// 传参：毫秒
    pub fn list_items(&self, 毫秒数: i64) -> Vec<&T> {
        if self.count == 0 {
            return vec![];
        }
        let mut items = vec![];

        // 将微秒数转为 u64，但只在非负当前情况才如此
        let 微秒数 = if 毫秒数 as i64 >= 0 {
            毫秒数 as u64 * a8_micros::MICROS_PER_MILLIS
        } else {
            0
        };

        let 当前项 = &self.items[self.index];
        items.push(当前项);

        let ticks_当前项 = 当前项.time().micros_of_day();
        let index_当前索引 = self.index;
        for idx in (0..index_当前索引).rev() {
            let ticks_往前项 = self.items[idx].time().micros_of_day();
            let delta_往前微秒数 = ticks_当前项 - ticks_往前项;
            if delta_往前微秒数 > 微秒数 {
                // return items;
                break;
            }
            items.push(&self.items[idx]);
        }
        if self.count <= LEN {
            // 尚未走出一轮(填满 CAPACTIY 个项)，后面没有值
            return items;
        }

        for idx in ((index_当前索引 + 1)..self.items.len()).rev() {
            let ticks_往前项 = self.items[idx].time().micros_of_day();
            let delta_往前微秒数 = ticks_当前项 - ticks_往前项;
            if delta_往前微秒数 > 微秒数 {
                return items;
            }
            // 是不是少了这句代码? todo 志松
            items.push(&self.items[idx]);
        }
        items
    }

    // 补数据, 十档数据有可能在N秒后才有新数据, 逐笔同理, 把之前缺失的数据补上(10毫秒一个)
    /* pub fn update_补数据(&mut self, last: TimeFT) {
        let last_ticks = last.micros_of_day();

        if self.current().time().micros_of_day() + 10 >= last_ticks {
            return;
        }

        let mut 最新时间的数据 = Option::<T>::None;
        for ele in &self.items {
            match 最新时间的数据{
                Some(v) => {
                    if v.time().micros_of_day() <= ele.time().micros_of_day() {
                        最新时间的数据 = Some(ele.clone());
                    }
                },
                None => {
                    最新时间的数据 = Some(ele.clone());
                }
            }
        }

        let mut 内存_最新数据 = match 最新时间的数据 {
            Some(v) => v,
            None => return,
        };

        let mut 内存_最新数据时间 = 内存_最新数据.time().micros_of_day();

        while last_ticks > 内存_最新数据时间 + 10{
            内存_最新数据时间 += 10;
            内存_最新数据.set_接收时间(TimeFT::from_micros_day_unsafe(内存_最新数据时间));
            self.push(&内存_最新数据);
        }

    } */
}

impl<T: CacheData, const LEN: usize> Default for HighSpeedCache<T, LEN> {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod test_高频数据 {
    use chrono::NaiveTime;
    use crate::fasttime::b2_time::TimeExt;

    use super::{CacheData, HighSpeedCache};

    #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
    struct Demo_高频数据 {
        pub time_发生时间: NaiveTime,
        pub time_接收时间: NaiveTime,
        pub val: i32,
    }

    impl CacheData for Demo_高频数据 {
        fn time(&self) -> NaiveTime {
            self.time_发生时间
        }
    }

    #[test]
    fn test_demo_高频数据() {
        let HighSpeedCache = HighSpeedCache::<Demo_高频数据, 1000>::new();

        // 使用固定增量代替随机数，避免依赖 rand
        let mut micros = NaiveTime::from_hmsi_friendly_unsafe(93000_000).micros_of_day();
        for val in 1..1000 {
            let time_发生时间 = NaiveTime::from_micros_day_unsafe(micros);
            micros += 100; // 固定增加100微秒

            HighSpeedCache.push(Demo_高频数据 {
                time_发生时间,
                time_接收时间: time_发生时间,
                val,
            });
            _ = HighSpeedCache.find_找之前的项(30_000, false);
        }
    }
}