forked from wangyu-/UDPspeeder
-
Notifications
You must be signed in to change notification settings - Fork 0
/
fec_manager.h
202 lines (167 loc) · 3.83 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
/*
* 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;
struct anti_replay_t
{
u64_t replay_buffer[anti_replay_buff_size];
unordered_set<u32_t> st;
int index;
anti_replay_t()
{
memset(replay_buffer,-1,sizeof(replay_buffer));
st.rehash(anti_replay_buff_size*3);
index=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 type;
int fec_data_num,fec_redundant_num;
int fec_mtu;
int fec_pending_num;
int fec_pending_time;
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();
my_time_t get_first_packet_time()
{
return first_packet_time_for_output;
}
int get_type()
{
return type;
}
u64_t get_timer_fd64();
int re_init(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];
re_init();
}
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 re_init();
int input(char *s,int len);
int output(int &n,char ** &s_arr,int* &len_arr);
};
#endif /* FEC_MANAGER_H_ */