forked from k1995/UDPspeeder
-
Notifications
You must be signed in to change notification settings - Fork 0
/
fec_manager.h
254 lines (208 loc) · 4.71 KB
/
fec_manager.h
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
/*
* fec_manager.h
*
* Created on: Sep 27, 2017
* Author: root
*/
#ifndef FEC_MANAGER_H_
#define FEC_MANAGER_H_
#include "common.h"
#include "log.h"
#include "lib/rs.h"
const int max_blob_packet_num=30000;//how many packet can be contain in a blob_t ,can be set very large
const u32_t anti_replay_buff_size=30000;//can be set very large
const int max_fec_packet_num=255;// this is the limitation of the rs lib
extern u32_t fec_buff_num;
/*begin for first time init or dynamic update*/
extern int g_fec_data_num;
extern int g_fec_redundant_num;
extern int g_fec_mtu;
extern int g_fec_queue_len;
extern int g_fec_timeout; //8ms
extern int g_fec_mode;
extern int dynamic_update_fec;
/*end for first time init or dynamic update*/
struct anti_replay_t
{
u64_t replay_buffer[anti_replay_buff_size];
unordered_set<u32_t> st;
int index;
anti_replay_t()
{
clear();
}
int clear()
{
memset(replay_buffer,-1,sizeof(replay_buffer));
st.clear();
st.rehash(anti_replay_buff_size*3);
index=0;
return 0;
}
void set_invaild(u32_t seq)
{
if(st.find(seq)!=st.end() )
{
mylog(log_trace,"seq %u exist\n",seq);
return;
//return 0;
}
if(replay_buffer[index]!=u64_t(i64_t(-1)))
{
assert(st.find(replay_buffer[index])!=st.end());
st.erase(replay_buffer[index]);
}
replay_buffer[index]=seq;
st.insert(seq);
index++;
if(index==int(anti_replay_buff_size)) index=0;
//return 1; //for complier check
}
int is_vaild(u32_t seq)
{
return st.find(seq)==st.end();
}
};
struct blob_encode_t
{
char input_buf[(max_fec_packet_num+5)*buf_len];
int current_len;
int counter;
char *output_buf[max_fec_packet_num+100];
blob_encode_t();
int clear();
int get_num();
int get_shard_len(int n);
int get_shard_len(int n,int next_packet_len);
int input(char *s,int len); //len=use len=0 for second and following packet
int output(int n,char ** &s_arr,int & len);
};
struct blob_decode_t
{
char input_buf[(max_fec_packet_num+5)*buf_len];
int current_len;
int last_len;
int counter;
char *output_buf[max_blob_packet_num+100];
int output_len[max_blob_packet_num+100];
blob_decode_t();
int clear();
int input(char *input,int len);
int output(int &n,char ** &output,int *&len_arr);
};
class fec_encode_manager_t
{
private:
u32_t seq;
int fec_mode;
int fec_data_num,fec_redundant_num;
int fec_mtu;
int fec_queue_len;
int fec_timeout;
my_time_t first_packet_time;
my_time_t first_packet_time_for_output;
blob_encode_t blob_encode;
char input_buf[max_fec_packet_num+5][buf_len];
int input_len[max_fec_packet_num+100];
char *output_buf[max_fec_packet_num+100];
int output_len[max_fec_packet_num+100];
int counter;
int timer_fd;
u64_t timer_fd64;
int ready_for_output;
u32_t output_n;
int append(char *s,int len);
public:
fec_encode_manager_t();
~fec_encode_manager_t();
int clear()
{
counter=0;
blob_encode.clear();
ready_for_output=0;
itimerspec zero_its;
memset(&zero_its, 0, sizeof(zero_its));
timerfd_settime(timer_fd, TFD_TIMER_ABSTIME, &zero_its, 0);
seq=(u32_t)get_true_random_number(); //TODO temp solution for a bug.
return 0;
}
my_time_t get_first_packet_time()
{
return first_packet_time_for_output;
}
int get_pending_time()
{
return fec_timeout;
}
int get_type()
{
return fec_mode;
}
u64_t get_timer_fd64();
int reset_fec_parameter(int data_num,int redundant_num,int mtu,int pending_num,int pending_time,int type);
int input(char *s,int len/*,int &is_first_packet*/);
int output(int &n,char ** &s_arr,int *&len);
};
struct fec_data_t
{
int used;
u32_t seq;
int type;
int data_num;
int redundant_num;
int idx;
char buf[buf_len];
int len;
};
struct fec_group_t
{
int type=-1;
int data_num=-1;
int redundant_num=-1;
int len=-1;
//int data_counter=0;
map<int,int> group_mp;
};
class fec_decode_manager_t
{
anti_replay_t anti_replay;
fec_data_t *fec_data;
unordered_map<u32_t, fec_group_t> mp;
blob_decode_t blob_decode;
int index;
int output_n;
char ** output_s_arr;
int * output_len_arr;
int ready_for_output;
char *output_s_arr_buf[max_fec_packet_num+100];//only for type=1,for type=0 the buf inside blot_t is used
int output_len_arr_buf[max_fec_packet_num+100];//same
public:
fec_decode_manager_t()
{
fec_data=new fec_data_t[fec_buff_num+5];
clear();
}
fec_decode_manager_t(const fec_decode_manager_t &b)
{
assert(0==1);//not allowed to copy
}
~fec_decode_manager_t()
{
delete fec_data;
}
int clear()
{
anti_replay.clear();
mp.clear();
mp.rehash(fec_buff_num*3);
for(int i=0;i<(int)fec_buff_num;i++)
fec_data[i].used=0;
ready_for_output=0;
index=0;
return 0;
}
//int re_init();
int input(char *s,int len);
int output(int &n,char ** &s_arr,int* &len_arr);
};
#endif /* FEC_MANAGER_H_ */