forked from wangyu-/UDPspeeder
-
Notifications
You must be signed in to change notification settings - Fork 0
/
fec_manager.h
306 lines (251 loc) · 5.63 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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
/*
* 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*/
const int anti_replay_timeout=60*1000;// 60s
struct anti_replay_t
{
struct info_t
{
my_time_t my_time;
int index;
};
u64_t replay_buffer[anti_replay_buff_size];
unordered_map<u32_t,info_t> mp;
int index;
anti_replay_t()
{
clear();
}
int clear()
{
memset(replay_buffer,-1,sizeof(replay_buffer));
mp.clear();
mp.rehash(anti_replay_buff_size*3);
index=0;
return 0;
}
void set_invaild(u32_t seq)
{
if(is_vaild(seq)==0)
{
mylog(log_trace,"seq %u exist\n",seq);
//assert(mp.find(seq)!=mp.end());
//mp[seq].my_time=get_current_time_rough();
return;
}
if(replay_buffer[index]!=u64_t(i64_t(-1)))
{
assert(mp.find(replay_buffer[index])!=mp.end());
mp.erase(replay_buffer[index]);
}
replay_buffer[index]=seq;
assert(mp.find(seq)==mp.end());
mp[seq].my_time=get_current_time_rough();
mp[seq].index=index;
index++;
if(index==int(anti_replay_buff_size)) index=0;
}
int is_vaild(u32_t seq)
{
if(mp.find(seq)==mp.end()) return 1;
if(get_current_time_rough()-mp[seq].my_time>anti_replay_timeout)
{
replay_buffer[mp[seq].index]=u64_t(i64_t(-1));
mp.erase(seq);
return 1;
}
return 0;
}
};
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);
ev_timer timer;
struct ev_loop *loop=0;
void (*cb) (struct ev_loop *loop, struct ev_timer *watcher, int revents)=0;
public:
fec_encode_manager_t();
~fec_encode_manager_t();
void set_data(void * data)
{
timer.data=data;
}
void set_loop_and_cb(struct ev_loop *loop,void (*cb) (struct ev_loop *loop, struct ev_timer *watcher, int revents))
{
this->loop=loop;
this->cb=cb;
ev_init(&timer,cb);
}
int clear_data()
{
counter=0;
blob_encode.clear();
ready_for_output=0;
seq=(u32_t)get_fake_random_number(); //TODO temp solution for a bug.
if(loop)
{
ev_timer_stop(loop,&timer);
}
return 0;
}
int clear_all()
{
//itimerspec zero_its;
//memset(&zero_its, 0, sizeof(zero_its));
//timerfd_settime(timer_fd, TFD_TIMER_ABSTIME, &zero_its, 0);
if(loop)
{
ev_timer_stop(loop,&timer);
loop=0;
cb=0;
}
clear_data();
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_ */