just commit

connection.h

@@ -15,6 +15,7 @@ extern int disable_anti_replay;
 #include "log.h"
 #include "delay_manager.h"
 #include "fd_manager.h"
+#include "fec_manager.h"
 
 
 /*
@@ -67,7 +68,8 @@ struct conn_info_t     //stores info for a raw connection.for client ,there is o
 //handle multiple clients
 {
 	conv_manager_t conv_manager;
-	//anti_replay_t anti_replay;
+	fec_encode_manager_t fec_encode_manager;
+	fec_decode_manager_t fec_decode_manager;
 	fd64_t timer_fd;
 	ip_port_t ip_port;
 };//g_conn_info;

fec_manager.cpp

@@ -9,8 +9,6 @@
 #include "log.h"
 #include "common.h"
 #include "lib/rs.h"
-u32_t seq=0;
-
 
 
 blob_encode_t::blob_encode_t()
@@ -66,7 +64,6 @@ int blob_encode_t::output(int n,char ** &s_arr,int & len)
 blob_decode_t::blob_decode_t()
 {
 	clear();
-
 }
 int blob_decode_t::clear()
 {
@@ -118,46 +115,50 @@ int blob_decode_t::output(int &n,char ** &s_arr,int *&len_arr)
 
 fec_encode_manager_t::fec_encode_manager_t()
 {
-	re_init();
+	re_init(4,2,1200);
 }
-int fec_encode_manager_t::re_init()
+int fec_encode_manager_t::re_init(int data_num,int redundant_num,int mtu)
 {
-	fec_data_num=4;
+	fec_data_num=data_num;
-	fec_redundant_num=2;
+	fec_redundant_num=redundant_num;
-	fec_mtu=1200;
+	fec_mtu=mtu;
 
 	counter=0;
 	blob_encode.clear();
 	ready_for_output=0;
+	seq=0;
 	return 0;
 }
-int fec_encode_manager_t::input(char *s,int len,int &is_first_packet)
+int fec_encode_manager_t::input(char *s,int len/*,int &is_first_packet*/)
 {
-	is_first_packet=0;
     if(s==0 ||blob_encode.get_shard_len(fec_data_num,len)>=fec_mtu)
 	{
     	char ** blob_output;
     	int blob_len;
-    	assert(counter!=0);
+    	if(counter==0)
+    	{
+    		if(s==0) return 0;//relax this restriction temporarily
+    		else mylog(log_warn,"message too long,ignored\n");
+    	}
     	blob_encode.output(fec_data_num,blob_output,blob_len);
     	for(int i=0;i<fec_data_num+fec_redundant_num;i++)
     	{
     		int tmp_idx=0;
     		write_u32(buf[i]+tmp_idx,seq);
     		tmp_idx+=sizeof(u32_t);
+    		buf[i][tmp_idx++]=(unsigned char)0;
     		buf[i][tmp_idx++]=(unsigned char)fec_data_num;
     		buf[i][tmp_idx++]=(unsigned char)fec_redundant_num;
     		buf[i][tmp_idx++]=(unsigned char)i;
-    		buf[i][tmp_idx++]=(unsigned char)0;
+    		output_buf[i]=buf[i]+tmp_idx;
     		if(i<fec_data_num)
     		{
     			memcpy(buf[i]+tmp_idx,blob_output[i],blob_len);
     			tmp_idx+=blob_len;
     		}
-    		output_buf[i]=buf[i]+sizeof(u32_t)+3*sizeof(char);
-
     	}
-    	output_len=blob_len+sizeof(u32_t)+3*sizeof(char);
+    	output_len=blob_len+sizeof(u32_t)+4*sizeof(char);/////remember to change this 4,if modified the protocol
+
 		rs_encode2(fec_data_num,fec_data_num+fec_redundant_num,output_buf,blob_len);
 		for(int i=0;i<fec_data_num+fec_redundant_num;i++)
 		{
@@ -169,10 +170,9 @@ int fec_encode_manager_t::input(char *s,int len,int &is_first_packet)
     	counter=0;
     	blob_encode.clear();
 	}
-
     if(s!=0)
     {
-    	if(counter==0) is_first_packet=1;
+    	//if(counter==0) is_first_packet=1;
     	blob_encode.input(s,len);
     	counter++;
     }
@@ -198,14 +198,117 @@ int fec_encode_manager_t::output(int &n,char ** &s_arr,int &len)
 	return 0;
 }
 
-/*
+fec_decode_manager_t::fec_decode_manager_t()
-int fec_decode_manager_t::input(char *s,int l)
 {
+	re_init();
+}
+
+int fec_decode_manager_t::re_init()
+{
+	for(int i=0;i<(int)fec_buff_size;i++)
+		fec_data[i].used=0;
+	ready_for_output=0;
 	return 0;
 }
 
-int fec_decode_manager_t::output(int &n,char ** &s_arr,int* &l_arr)
+int fec_decode_manager_t::input(char *s,int len)
+{
+	assert(s!=0);
+	int tmp_idx=0;
+	u32_t seq=read_u32(s+tmp_idx);
+	tmp_idx+=sizeof(u32_t);
+	int type=(unsigned char)s[tmp_idx++];
+	int data_num=(unsigned char)s[tmp_idx++];
+	int redundant_num=(unsigned char)s[tmp_idx++];
+	int inner_index=(unsigned char)s[tmp_idx++];
+	len=len-tmp_idx;
+	if(len<0)
+	{
+		return -1;
+	}
+	if(data_num+redundant_num>255)
+	{
+		return -1;
+	}
+	if(!anti_replay.is_vaild(seq))
 	{
 		return 0;
-}*/
+	}
+	if(!mp[seq].empty())
+	{
+		int first_idx=mp[seq].begin()->second;
+		int ok=1;
+		if(fec_data[first_idx].data_num!=data_num)
+			ok=0;
+		if(fec_data[first_idx].redundant_num!=redundant_num)
+			ok=0;
+		if(fec_data[first_idx].len!=len)
+			ok=0;
+		if(ok==0)
+		{
+			return 0;
+		}
+	}
+	if(fec_data[index].used!=0)
+	{
+		int tmp_seq=fec_data[index].seq;
+		anti_replay.set_invaild(tmp_seq);
+		if(mp.find(tmp_seq)!=mp.end())
+		{
+			mp.erase(tmp_seq);
+		}
+	}
 
+	fec_data[index].used=1;
+	fec_data[index].seq=seq;
+	fec_data[index].type=type;
+	fec_data[index].data_num=data_num;
+	fec_data[index].redundant_num=redundant_num;
+	fec_data[index].idx=inner_index;
+	fec_data[index].len=len;
+	memcpy(fec_data[index].buf,s+tmp_idx,len);
+	mp[seq][inner_index]=index;
+
+	index++;
+	if(index==int(anti_replay_buff_size)) index=0;
+
+	map<int,int> &inner_mp=mp[seq];
+	assert((int)inner_mp.size()<=data_num);
+	if((int)inner_mp.size()==data_num)
+	{
+
+		char *fec_tmp_arr[256+5]={0};
+		for(auto it=inner_mp.begin();it!=inner_mp.end();it++)
+		{
+			fec_tmp_arr[it->first]=fec_data[it->second].buf;
+		}
+		rs_decode2(data_num,data_num+redundant_num,fec_tmp_arr,len); //the input data has been modified in-place
+		blob_decode.clear();
+		for(int i=0;i<data_num;i++)
+		{
+			blob_decode.input(fec_tmp_arr[i],len);
+		}
+		blob_decode.output(output_n,output_s_arr,output_len_arr);
+		ready_for_output=1;
+		anti_replay.set_invaild(seq);
+	}
+
+	return 0;
+}
+int fec_decode_manager_t::output(int &n,char ** &s_arr,int* &len_arr)
+{
+	if(!ready_for_output)
+	{
+		n=-1;
+		s_arr=0;
+		len_arr=0;
+	}
+	else
+	{
+		ready_for_output=0;
+		n=output_n;
+		s_arr=output_s_arr;
+		len_arr=output_len_arr;
+	}
+	return 0;
+}

fec_manager.h

@@ -87,7 +87,7 @@ struct blob_decode_t
 	int output(int &n,char ** &output,int *&len_arr);
 };
 
-struct fec_encode_manager_t
+class fec_encode_manager_t
 {
 	int fec_data_num,fec_redundant_num;
 	int fec_mtu;
@@ -95,27 +95,28 @@ struct fec_encode_manager_t
 	char *output_buf[256+5];
 	int output_len;
 	int ready_for_output;
-
+	u32_t seq;
 	int counter;
 
 	blob_encode_t blob_encode;
+public:
 	fec_encode_manager_t();
-	int re_init();
+	int re_init(int data_num,int redundant_num,int mtu);
-	int input(char *s,int len,int &is_first_packet);
+	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;
-	int type;
 	char buf[buf_len];
 	int len;
 };
-struct fec_decode_manager_t
+class fec_decode_manager_t
 {
 	anti_replay_t anti_replay;
 	fec_data_t fec_data[fec_buff_size];
@@ -123,114 +124,15 @@ struct fec_decode_manager_t
 	unordered_map<u32_t, map<int,int> > mp;
 	blob_decode_t blob_decode;
 
-	fec_decode_manager_t()
-	{
-		for(int i=0;i<(int)fec_buff_size;i++)
-			fec_data[i].used=0;
-		ready_for_output=0;
-	}
-
 	int output_n;
 	char ** output_s_arr;
 	int * output_len_arr;
-
 	int ready_for_output;
-	int input(char *s,int len)
+public:
-	{
+	fec_decode_manager_t();
-		char *ori_s=s;
+	int re_init();
-		u32_t seq=read_u32(s);
+	int input(char *s,int len);
-		s+=sizeof(u32_t);
+	int output(int &n,char ** &s_arr,int* &len_arr);
-		int data_num=(unsigned char)*(s++);
-		int redundant_num=(unsigned char)*(s++);
-		int innder_index=(unsigned char)*(s++);
-		int type=(unsigned char)*(s++);
-		len=len-int(s-ori_s);
-		if(len<0)
-		{
-			return -1;
-		}
-
-		if(!anti_replay.is_vaild(seq))
-		{
-			return 0;
-		}
-		if(!mp[seq].empty())
-		{
-			int tmp_idx=mp[seq].begin()->second;
-			int ok=1;
-			if(data_num+redundant_num>255)
-				ok=0;
-			if(fec_data[tmp_idx].data_num!=data_num||fec_data[tmp_idx].redundant_num!=redundant_num||fec_data[tmp_idx].len!=len)
-			{
-				ok=0;
-			}
-			if(ok==0)
-			{
-				return 0;
-			}
-		}
-		if(fec_data[index].used!=0)
-		{
-			int tmp_seq=fec_data[index].seq;
-			anti_replay.set_invaild(tmp_seq);
-			if(mp.find(tmp_seq)!=mp.end())
-			{
-				mp.erase(tmp_seq);
-			}
-		}
-
-		fec_data[index].used=1;
-		fec_data[index].seq=seq;
-		fec_data[index].data_num=data_num;
-		fec_data[index].redundant_num=redundant_num;
-		fec_data[index].idx=innder_index;
-		fec_data[index].type=type;
-		fec_data[index].len=len;
-		mp[seq][innder_index]=index;
-
-
-		map<int,int> &inner_mp=mp[seq];
-		if((int)inner_mp.size()>=data_num)
-		{
-			anti_replay.set_invaild(seq);
-			char *fec_tmp_arr[256+5]={0};
-			for(auto it=inner_mp.begin();it!=inner_mp.end();it++)
-			{
-				fec_tmp_arr[it->first]=fec_data[it->second].buf;
-			}
-			rs_decode2(data_num,redundant_num,fec_tmp_arr,len);
-			blob_decode.clear();
-			for(int i=0;i<data_num;i++)
-			{
-				blob_decode.input(fec_tmp_arr[i],len);
-			}
-			blob_decode.output(output_n,output_s_arr,output_len_arr);
-			ready_for_output=1;
-		}
-
-		index++;
-		if(index==int(anti_replay_buff_size)) index=0;
-
-
-		return 0;
-	}
-	int output(int &n,char ** &s_arr,int* &len_arr)
-	{
-		if(!ready_for_output)
-		{
-			n=-1;
-			s_arr=0;
-			len_arr=0;
-		}
-		else
-		{
-			ready_for_output=0;
-			n=output_n;
-			s_arr=output_s_arr;
-			len_arr=output_len_arr;
-		}
-		return 0;
-	}
 };
 
 #endif /* FEC_MANAGER_H_ */

main.cpp

@@ -26,6 +26,11 @@ int jitter_max=0;
 //int random_number_fd=-1;
 
 int mtu_warn=1350;
+
+int fec_data_num=3;
+int fec_redundant_num=2;
+int fec_mtu=30;
+
 u32_t local_ip_uint32,remote_ip_uint32=0;
 char local_ip[100], remote_ip[100];
 int local_port = -1, remote_port = -1;
@@ -105,15 +110,45 @@ int delay_send(my_time_t delay,const dest_t &dest,char *data,int len)
 {
 	return delay_manager.add(delay,dest,data,len);;
 }
-int from_normal_to_fec(const dest_t &dest,char *data,int len)
+int from_normal_to_fec(conn_info_t & conn_info,const dest_t &dest,char *data,int len)
 {
-	delay_send(0,dest,data,len);
+	static int counter=0;
-	delay_send(1000*1000,dest,data,len);
+	counter++;
+
+	conn_info.fec_encode_manager.input(data,len);
+	//if(counter%5==0)
+		//conn_info.fec_encode_manager.input(0,0);
+
+	int n;
+	char **s_arr;
+	int s_arr_len;
+
+	conn_info.fec_encode_manager.output(n,s_arr,s_arr_len);
+
+	for(int i=0;i<n;i++)
+	{
+		delay_send(0,dest,s_arr[i],s_arr_len);
+	}
+	//delay_send(0,dest,data,len);
+	//delay_send(1000*1000,dest,data,len);
 	return 0;
 }
-int from_fec_to_normal(const dest_t &dest,char *data,int len)
+int from_fec_to_normal(conn_info_t & conn_info,const dest_t &dest,char *data,int len)
 {
-	my_send(dest,data,len);
+	conn_info.fec_decode_manager.input(data,len);
+
+	int n;char ** s_arr;int* len_arr;
+	conn_info.fec_decode_manager.output(n,s_arr,len_arr);
+
+
+//	printf("<n:%d>",n);
+	for(int i=0;i<n;i++)
+	{
+		delay_send(0,dest,s_arr[i],len_arr[i]);
+		//s_arr[i][len_arr[i]]=0;
+		//printf("<%s>\n",s_arr[i]);
+	}
+	//my_send(dest,data,len);
 	return 0;
 }
 int client_event_loop()
@@ -125,10 +160,12 @@ int client_event_loop()
 	int remote_fd;
 	fd64_t remote_fd64;
 
-    conn_info_t conn_info;
+    conn_info_t *conn_info_p=new conn_info_t;
-
+    conn_info_t &conn_info=*conn_info_p;  //huge size of conn_info,do not allocate on stack
 	init_listen_socket();
 
+	conn_info.fec_encode_manager.re_init(fec_data_num,fec_redundant_num,fec_mtu);
+
 	epoll_fd = epoll_create1(0);
 
 	const int max_events = 4096;
@@ -231,8 +268,12 @@ int client_event_loop()
 				dest_t dest;
 				dest.type=type_fd64_conv;
 				dest.inner.fd64=remote_fd64;
+
+				//char * new_data;
+				//int new_len;
+				//put_conv(conv,data,data_len,new_data,new_len);
 				dest.conv=conv;
-				from_normal_to_fec(dest,data,data_len);
+				from_normal_to_fec(conn_info,dest,data,data_len);
 				//my_send(dest,data,data_len);
 			}
 		    else if (events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()) {
@@ -277,7 +318,7 @@ int client_event_loop()
 				dest_t dest;
 				dest.inner.ip_port.from_u64(u64);
 				dest.type=type_ip_port;
-				from_fec_to_normal(dest,new_data,new_len);
+				from_fec_to_normal(conn_info,dest,new_data,new_len);
 				mylog(log_trace,"[%s] send packet\n",dest.inner.ip_port.to_s());
 			}
 
@@ -311,7 +352,7 @@ int server_event_loop()
 	int epoll_fd;
 	int remote_fd;
 
-    conn_info_t conn_info;
+//    conn_info_t conn_info;
 
 	init_listen_socket();
 
@@ -397,10 +438,12 @@ int server_event_loop()
 				{
 					conn_manager.insert(ip_port);
 					conn_info_t &conn_info=conn_manager.find(ip_port);
+					conn_info.fec_encode_manager.re_init(fec_data_num,fec_redundant_num,fec_mtu);
 					conn_info.conv_manager.reserve();
 				}
 				conn_info_t &conn_info=conn_manager.find(ip_port);
 
+
 				u32_t conv;
 				char *new_data;
 				int new_len;
@@ -438,7 +481,7 @@ int server_event_loop()
 				dest_t dest;
 				dest.type=type_fd64;
 				dest.inner.fd64=fd64;
-				from_fec_to_normal(dest,new_data,new_len);
+				from_fec_to_normal(conn_info,dest,new_data,new_len);
 
 				//int fd = int((u64 << 32u) >> 32u);
 				//////////////////////////////todo
@@ -531,7 +574,12 @@ int server_event_loop()
 				dest.type=type_ip_port_conv;
 				dest.conv=conv;
 				dest.inner.ip_port=ip_port;
-				from_normal_to_fec(dest,data,data_len);
+
+				//char * new_data;
+				//int new_len;
+				//put_conv(conv,data,data_len,new_data,new_len);
+
+				from_normal_to_fec(conn_info,dest,data,data_len);
 				mylog(log_trace,"[%s] send packet\n",ip_port.to_s());
 
 			}
@@ -619,9 +667,83 @@ int unit_test()
 		printf("<%d:%s>",len_arr[i],buf);
 	}
 	printf("\n");
+	static fec_encode_manager_t fec_encode_manager;
+	static fec_decode_manager_t fec_decode_manager;
+
+	{
+
+		string a = "11111";
+		string b = "22";
+		string c = "33333333";
+
+		fec_encode_manager.input((char *) a.c_str(), a.length());
+		fec_encode_manager.input((char *) b.c_str(), b.length());
+		fec_encode_manager.input((char *) c.c_str(), c.length());
+		fec_encode_manager.input(0, 0);
+
+		int n;
+		char **s_arr;
+		int len;
+
+
+		fec_encode_manager.output(n,s_arr,len);
+		printf("<n:%d,len:%d>",n,len);
+
+		for(int i=0;i<n;i++)
+		{
+			fec_decode_manager.input(s_arr[i],len);
+		}
+
+		{
+			int n;char ** s_arr;int* len_arr;
+			fec_decode_manager.output(n,s_arr,len_arr);
+			printf("<n:%d>",n);
+			for(int i=0;i<n;i++)
+			{
+				s_arr[i][len_arr[i]]=0;
+				printf("<%s>\n",s_arr[i]);
+			}
+		}
+
+
+	}
+
+	{
+		string a = "aaaaaaa";
+		string b = "bbbbbbbbbbbbb";
+		string c = "ccc";
+
+		fec_encode_manager.input((char *) a.c_str(), a.length());
+		fec_encode_manager.input((char *) b.c_str(), b.length());
+		fec_encode_manager.input((char *) c.c_str(), c.length());
+		fec_encode_manager.input(0, 0);
+
+		int n;
+		char **s_arr;
+		int len;
+
+
+		fec_encode_manager.output(n,s_arr,len);
+		printf("<n:%d,len:%d>",n,len);
+
+		for(int i=0;i<n;i++)
+		{
+			if(i==1||i==3||i==5||i==0)
+			fec_decode_manager.input(s_arr[i],len);
+		}
+
+		{
+			int n;char ** s_arr;int* len_arr;
+			fec_decode_manager.output(n,s_arr,len_arr);
+			printf("<n:%d>",n);
+			for(int i=0;i<n;i++)
+			{
+				s_arr[i][len_arr[i]]=0;
+				printf("<%s>\n",s_arr[i]);
+			}
+		}
+	}
 
-	fec_encode_manager_t fec_encode_manager;
-	fec_decode_manager_t fec_decode_manager;
 
 	return 0;
 }

makefile

@@ -17,10 +17,7 @@ TAR=${NAME}_binaries.tar.gz `echo ${TARGETS}|sed -r 's/([^ ]+)/speeder_\1/g'`
 
 all:git_version
 	rm -f ${NAME}
-	${cc_local}   -o ${NAME}          -I. ${SOURCES} ${FLAGS} -lrt -ggdb -static -O3
+	${cc_local}   -o ${NAME}          -I. ${SOURCES} ${FLAGS} -lrt -ggdb -static
-fast: git_version
-	rm -f ${NAME}
-	${cc_local}   -o ${NAME}          -I. ${SOURCES} ${FLAGS} -lrt -ggdb
 debug: git_version
 	rm -f ${NAME}
 	${cc_local}   -o ${NAME}          -I. ${SOURCES} ${FLAGS} -lrt -Wformat-nonliteral -D MY_DEBUG