rust_rsm/common/
indexring.rs

1//提供一个有序、有界的环，每个元素都有唯一的序号，按序号进行存取
2//主要场景：可靠传输的报文发送队列，确认、重传和查重
3#![allow(non_camel_case_types)]
4#![allow(non_snake_case)]
5#![allow(non_upper_case_globals)]
6use super::*;
7
8#[derive(Copy,Clone,PartialEq,Debug)]
9pub enum EItemState {
10    ITEM_STATE_IDLE = 0,
11    ITEM_STATE_USED = 1,
12    ITEM_STATE_DELETED = 2,
13}
14pub const INVALID_SEQ: u64 = 2 ^ 64 - 1;
15pub const INVALID_INDEX: usize = 0xFFFFFFFF;
16pub struct data_item<T> {
17    seq_no: u64,
18    state: EItemState,
19    item: Option<T>,
20}
21impl <T>data_item<T> {
22    pub fn new()->Self{
23        return Self{
24            seq_no:0,
25            state:EItemState::ITEM_STATE_IDLE,
26            item:None,
27        };
28    }
29}
30pub struct index_ring_t<T> {
31    data: Vec<data_item<T>>,
32    head: usize,
33    tail: usize,
34    min_seq: u64,
35    max_seq: u64,
36    capacity: usize,
37    expect_item:u64,
38    recv_item:u64,
39}
40
41impl<T> index_ring_t<T> {
42    pub fn new(ring_capacity: usize, start_seq: u64) -> Self {
43            let mut vItem:Vec<data_item<T>>=Vec::with_capacity(ring_capacity);
44            for _ in 0..ring_capacity {
45                vItem.push(data_item::new())
46            }
47                       
48            return Self {
49                data: vItem,
50                head: INVALID_INDEX,
51                tail: INVALID_INDEX,
52                min_seq: start_seq,
53                max_seq: start_seq,
54                capacity: ring_capacity,
55                expect_item:0,
56                recv_item:0,
57            };
58        }
59
60    fn is_valid_index(&self,idx:usize)->bool {
61        if idx>=self.capacity {
62            return false;
63        }
64        if self.head<=self.tail {
65            return idx>=self.head && idx<=self.tail
66        } else {
67            return (idx>=self.head && idx<self.capacity) || idx<=self.tail
68        }
69        
70       
71    }
72    fn set_item(&mut self, seq: u64, idx: usize, item: T) -> errcode::RESULT {
73        
74        if self.data[idx].state == EItemState::ITEM_STATE_USED && self.data[idx].seq_no == seq {
75            return errcode::ERROR_ALREADY_EXIST;
76        }
77        self.data[idx].state = EItemState::ITEM_STATE_USED;
78        self.data[idx].seq_no = seq;
79        self.data[idx].item = Some(item);
80        self.recv_item+=1;
81        errcode::RESULT_SUCCESS
82    }
83    //添加一条记录,如果序号超出当前范围，往前移动；但是如果序号超出当前最大值+capacity，则认为是非法序号，丢弃
84    pub fn add_item(&mut self, seq: u64, item: T) -> errcode::RESULT {
85        //序号小于min_seq，或者比max_seq大capacity，认为是非法序号
86        if seq<self.min_seq || seq>=self.max_seq+self.capacity as u64{
87            return errcode::ERROR_INVALID_INDEX;
88        }
89        let distance = if seq<=self.max_seq {0} else {seq-self.max_seq};
90        if self.get_ring_len() == 0 {
91            self.head = 0;
92            self.tail = distance as usize;
93            //self.min_seq = seq;
94            self.max_seq = seq;
95            self.expect_item+=1+distance;
96            return self.set_item(seq, self.tail, item);
97        }
98
99        if seq >= self.min_seq && seq <= self.max_seq {
100            let idx = self.get_index_by_seq(seq);
101            if idx < self.capacity {                
102                return self.set_item(seq, idx, item);
103            }
104        } else if seq > self.max_seq {            
105            self.expect_item+=distance;
106            self.max_seq = seq;
107            if self.get_ring_len() + distance as usize > self.capacity {
108                //self.min_seq = self.min_seq + distance;
109                self.remove_item_before_seq(self.min_seq + distance-1, true);
110                //self.head = (self.head + distance as usize) % self.capacity;
111            }
112            self.tail = (self.tail + distance as usize) % self.capacity;
113            return self.set_item(seq, self.tail, item);
114        }
115
116        return  errcode::ERROR_INVALID_INDEX;
117
118        
119    }
120
121    fn incr_hdr_index(&mut self) {
122        match self.get_ring_len() {
123            0 => (),
124            1 => {
125                self.head = INVALID_INDEX;
126                self.tail = INVALID_INDEX;
127                self.min_seq += 1;
128                self.max_seq=self.min_seq;
129            }
130            _ => {
131                self.min_seq += 1;
132                self.head = (self.head + 1) % self.capacity;
133            }
134        }
135    }
136
137    fn incr_tail_index(&mut self) {
138        match self.get_ring_len() {
139            0 => {
140                self.head = 0;
141                self.tail = 0;
142            }
143            _ => {
144                self.tail = (self.tail + 1) % self.capacity;
145                self.max_seq += 1;
146            }
147        }
148    }
149
150    fn inner_delete_item(&mut self,idx:usize) {
151        assert!(idx < self.capacity);
152            self.data[idx].state = EItemState::ITEM_STATE_DELETED;
153            self.data[idx].item = None;
154            if idx == self.head {
155                self.incr_hdr_index()
156            }
157    }
158    //删除一个元素，如果数据不一致，需要进行补救
159    pub fn remove_item(&mut self, seq: u64) -> errcode::RESULT {
160        let idx = self.get_index_by_seq(seq);
161        if idx < self.capacity {
162            let state = self.data[idx].state;
163            self.inner_delete_item(idx);
164            if  state == EItemState::ITEM_STATE_USED {                
165                errcode::RESULT_SUCCESS
166            } else {
167                return errcode::ERROR_NOT_FOUND;
168            }
169        } else {
170            return errcode::ERROR_NOT_FOUND;
171        }
172    }
173
174    //删除指定序号之前所有的数据，如果是强制删除，则无论序号状态如何均予以删除，否则判断状态是否是deleted
175    pub fn remove_item_before_seq(&mut self, seq: u64,force:bool) -> errcode::RESULT {
176        let idx = self.get_index_by_seq(seq);
177        assert!(idx==INVALID_INDEX || idx<self.capacity);
178        if idx >= self.capacity  {
179            return errcode::ERROR_INVALID_INDEX;
180        }
181        
182        let mut head=self.head;
183        while head!=((idx+1) % self.capacity) {
184            if self.data[head].state==EItemState::ITEM_STATE_DELETED || force {
185                self.inner_delete_item(head);
186            } else {
187                return errcode::ERROR_INVALID_STATE;
188            }
189            head = (head+1) % self.capacity;
190        }
191        
192        errcode::RESULT_SUCCESS
193    }
194    //根据序号获取下标
195    pub fn get_index_by_seq(&self, seq: u64) -> usize {
196        if seq < self.min_seq || seq > self.max_seq || self.head > self.capacity {
197            return INVALID_INDEX;
198        }
199        return ((seq - self.min_seq) as usize + self.head) % self.capacity;
200    }
201    //根据序号获取实际的数据
202    pub fn get_item_by_seq(&mut self, seq: u64) -> Option<&mut T> {
203        let idx = self.get_index_by_seq(seq);
204        if idx != INVALID_INDEX && self.data[idx].state == EItemState::ITEM_STATE_USED {
205            match &mut self.data[idx].item {
206                None => return None,
207                Some(ref mut s) => return Some(s),
208            }
209        } else {
210            return None;
211        }
212    }
213
214    pub fn get_head_item(&mut self) -> Option<&mut T> {
215        if self.get_ring_len() > 0 {
216            match &mut self.data[self.head].item {
217                None=>return None,
218                Some(item)=>return Some(item),                
219            }
220            
221        } else {
222            return None;
223        }
224    }
225
226    pub fn get_head_index(&self) -> usize {
227        self.head
228    }
229    pub fn get_tail_index(&self) -> usize {
230        self.tail
231    }
232
233    pub fn get_head_seq(&self) -> u64 {
234        self.min_seq
235    }
236    pub fn get_tail_seq(&self) -> u64 {
237        self.max_seq
238    }
239
240    pub fn get_ring_len(&self) -> usize {
241        if self.head == INVALID_INDEX || self.tail == INVALID_INDEX {
242            return 0;
243        }
244        if self.head <= self.tail {
245            return self.tail - self.head + 1;
246        } else {
247            return self.tail + self.capacity - self.head + 1;
248        }
249    }
250
251    pub fn get_ring_capacity(&self) -> usize {
252        self.capacity
253    }
254
255    pub fn get_head_item_state(&self)->EItemState {
256        if self.get_ring_len()==0 {
257            return EItemState::ITEM_STATE_IDLE;
258        } else {
259            return self.data[self.head].state;
260        }
261
262    }
263
264    pub fn get_item_state(&self,seq:u64)->EItemState {
265        let idx = self.get_index_by_seq(seq);
266
267        if idx==INVALID_INDEX {
268            return EItemState::ITEM_STATE_IDLE;
269        } else {
270            return self.data[idx].state;
271        }
272
273    }
274    ///返回收到的报文以及期望收到的报文数量，计算丢包率
275    pub fn get_recv_item_stats(&self)->(u64,u64) {
276        return (self.recv_item,self.expect_item);
277    }
278    //清除统计
279    pub fn clear_recv_stats(&mut self) {
280        self.recv_item = 0;
281        self.expect_item = 0;
282    }
283    pub fn get_loss_count(&self)->u64 {
284        return self.expect_item - self.recv_item;
285    }
286    pub fn get_loss_rate(&self)->f32 {
287        if self.expect_item==0 {
288            0.0
289        } else {
290            return ((self.expect_item-self.recv_item)*100) as f32 / self.expect_item as f32;
291        }
292    }
293    pub fn clear(&mut self) {
294        self.clear_recv_stats();
295        self.head = INVALID_INDEX;
296        self.tail = INVALID_INDEX;
297        self.min_seq = 1;
298        self.max_seq = 1;
299        for i in &mut self.data {
300            i.state = EItemState::ITEM_STATE_IDLE;
301            i.item=None;
302        }
303
304    }
305    pub fn to_string(&self)->String {
306        format!("capacity:{},len:{},head:{},tail:{},seq:<{}-{},expect:{},recved={}>",
307        self.capacity,self.get_ring_len(),self.head,self.tail,self.min_seq,self.max_seq,self.expect_item,self.recv_item)
308    }
309}
rust_rsm/common/indexring.rs

rust_rsm/common/
indexring.rs
