diff --git a/delay_manager.cpp b/delay_manager.cpp index e610388..24e5da5 100644 --- a/delay_manager.cpp +++ b/delay_manager.cpp @@ -44,7 +44,7 @@ int delay_manager_t::add(my_time_t delay,const dest_t &dest,char *data,int len) { delay_data_t delay_data; delay_data.dest=dest; - delay_data.data=data; + //delay_data.data=data; delay_data.len=len; if(capacity!=0&&int(delay_mp.size()) >=capacity) @@ -54,6 +54,9 @@ int delay_manager_t::add(my_time_t delay,const dest_t &dest,char *data,int len) } if(delay==0) { + static char buf[buf_len]; + delay_data.data=buf; + memcpy(buf,data,len); int ret=delay_data.handle(); if (ret != 0) { mylog(log_trace, "handle() return %d\n", ret); diff --git a/fec_manager.cpp b/fec_manager.cpp index 47f2e54..8515273 100644 --- a/fec_manager.cpp +++ b/fec_manager.cpp @@ -174,10 +174,13 @@ int fec_encode_manager_t::append(char *s,int len/*,int &is_first_packet*/) { mylog(log_trace,"counter=%d\n",counter); assert(len<=65535&&len>=0); - char * p=input_buf[counter]+sizeof(u32_t)+4*sizeof(char); + assert(len<=fec_mtu); + char * p=input_buf[counter]+sizeof(u32_t)+4*sizeof(char);//copy directly to final position,avoid unnecessary copy. + //remember to change this,if protocol is modified + write_u16(p,(u16_t)((u32_t)len)); //TODO omit this u16 for data packet while sending p+=sizeof(u16_t); - memcpy(p,s,len);//remember to change this,if protocol is modified + memcpy(p,s,len); input_len[counter]=len+sizeof(u16_t); } else @@ -201,7 +204,7 @@ int fec_encode_manager_t::input(char *s,int len/*,int &is_first_packet*/) } if(type==1&&s!=0&&len>=fec_mtu) { - mylog(log_warn,"message too long len=%d,ignored\n",len); + mylog(log_warn,"message too long len=%d fec_mtu=%d,ignored\n",len,fec_mtu); return -1; } if(s==0&&counter==0) @@ -213,7 +216,7 @@ int fec_encode_manager_t::input(char *s,int len/*,int &is_first_packet*/) if(type==0&& blob_encode.get_shard_len(fec_data_num,len)>=fec_mtu) {about_to_fec=1; delayed_append=1;}//fec then add packet - if(type==0) assert(counter=max_fec_packet_num) { - mylog(log_info,"failed here\n"); + mylog(log_warn,"failed here\n"); return -1; } if(!anti_replay.is_vaild(seq)) @@ -460,7 +504,7 @@ int fec_decode_manager_t::input(char *s,int len) if(type==0&&data_num==0) { - mylog(log_warn,"unexpected here,data_num=0\n"); + mylog(log_warn,"unexpected type==0&&data_num==0\n"); return -1; } @@ -469,12 +513,16 @@ int fec_decode_manager_t::input(char *s,int len) mp[seq].type=type; else { - if(mp[seq].type!=type) ok=0; + if(mp[seq].type!=type) + { + mylog(log_warn,"type mismatch\n"); + ok=0; + } } if(data_num!=0) { - mp[seq].data_counter++; + //mp[seq].data_counter++; if(mp[seq].data_num==-1) { @@ -494,7 +542,7 @@ int fec_decode_manager_t::input(char *s,int len) if(ok==0) { - mylog(log_warn,"fec data invaild\n"); + mylog(log_warn,"fec packets invaild\n"); return -1; } @@ -538,64 +586,11 @@ int fec_decode_manager_t::input(char *s,int len) { if(mp[seq].data_num!=-1) { - //int old_data_num=data_num; - //int old_redundant_num=redundant_num; - - //data_num=mp[seq].data_num; - //redundant_num=mp[seq].redundant_num; - - /*if(mp[seq].data_counter>data_num) //invaild - { - mylog(log_warn,"unexpected mp[seq].data_counter>mp[seq].data_num\n"); - anti_replay.set_invaild(seq); - return -1; - }*/ - //assert((int)inner_mp.size()<=data_num); if((int)inner_mp.size()>=mp[seq].data_num) { about_to_fec=1; } - - - /* - else if(mp[seq].data_counter==data_num) //no need to fec . (received first redundant packet ,or received a data packet after redunant packet) - { - anti_replay.set_invaild(seq);// dont do fec,but still set invaild - } - else //mp[seq].data_counter < mp[seq].data_num - { - if((int)inner_mp.size()>data_num) - { - mylog(log_warn,"unexpected (int)inner_mp.size()>data_num+1\n"); - anti_replay.set_invaild(seq); - return -1; - }*/ - - //if((int)inner_mp.size()==data_num) - //about_to_fec=1; - - /* - if((int)inner_mp.size()==data_num+1) - { - anti_replay.set_invaild(seq); - - if(old_data_num==0) - return 0; - else - { - mylog(log_warn,"data_num=0\n"); - return -1; - } - }*/ - - - //} } - else - { - - } - //for() } @@ -628,8 +623,13 @@ int fec_decode_manager_t::input(char *s,int len) int max_len=-1; - int fec_ok=1; + int fec_result_ok=1; + int data_check_ok=1; int debug_num=inner_mp.size(); + + int missed_packet[max_fec_packet_num+5]; + int missed_packet_counter=0; + //outupt_s_arr_buf[max_fec_packet_num+5]={0}; //memset(output_s_arr_buf,0,sizeof(output_s_arr_buf));//in efficient @@ -641,17 +641,32 @@ int fec_decode_manager_t::input(char *s,int len) for(auto it=inner_mp.begin();it!=inner_mp.end();it++) { output_s_arr_buf[it->first]=fec_data[it->second].buf; - assert(fec_data[it->second].len>=(int)sizeof(u16_t)); + if(fec_data[it->second].len<(int)sizeof(u16_t)) + { + mylog(log_warn,"fec_data[it->second].len<(int)sizeof(u16_t)"); + data_check_ok=0; + } if(fec_data[it->second].len > max_len) max_len=fec_data[it->second].len; } + + if(max_len!=mp[seq].len) + { + data_check_ok=0; + mylog(log_warn,"max_len!=mp[seq].len"); + } + if(data_check_ok==0) + { + ready_for_output=0; + anti_replay.set_invaild(seq); + goto end; + } for(auto it=inner_mp.begin();it!=inner_mp.end();it++) { - memset(fec_data[it->second].buf+fec_data[it->second].len,0,max_len-fec_data[it->second].len); + int tmp_idx=it->second; + memset(fec_data[tmp_idx].buf+fec_data[tmp_idx].len,0,max_len-fec_data[tmp_idx].len); } - int missed_packet[max_fec_packet_num+5]; - int missed_packet_counter=0; for(int i=0;imax_data_len) { mylog(log_warn,"invaild len %d,seq= %u,data_num= %d r_num= %d,i= %d\n",output_len_arr_buf[i],seq,group_data_num,group_redundant_num,i); - fec_ok=0; + fec_result_ok=0; for(int i=0;i group_mp; }; class fec_decode_manager_t @@ -159,8 +161,8 @@ class fec_decode_manager_t int output_n; char ** output_s_arr; int * output_len_arr; - char *output_s_arr_buf[max_fec_pending_packet_num+100]; - int output_len_arr_buf[max_fec_pending_packet_num+100]; + char *output_s_arr_buf[max_fec_packet_num+100]; + int output_len_arr_buf[max_fec_packet_num+100]; int ready_for_output; public: fec_decode_manager_t(); diff --git a/main.cpp b/main.cpp index 6bd81a1..9b8be10 100644 --- a/main.cpp +++ b/main.cpp @@ -34,7 +34,7 @@ int debug_force_flush_fec=0; int fec_data_num=20; int fec_redundant_num=8; -int fec_mtu=1000; +int fec_mtu=1250; int fec_pending_num=30; int fec_pending_time=10000; int fec_type=0; @@ -171,7 +171,18 @@ int from_normal_to_fec(conn_info_t & conn_info,char *data,int len,int & out_n,ch } - mylog(log_debug,"from_normal_to_fec input_len=%d,output_n=%d\n",len,out_n); + mylog(log_info,"from_normal_to_fec input_len=%d,output_n=%d\n",len,out_n); + + if(out_n>0) + { + log_bare(log_info,"seq= %u ",read_u32(out_arr[0])); + } + for(int i=0;i",n); /* @@ -422,6 +434,7 @@ int client_event_loop() put_conv(conv,data,data_len,new_data,new_len); + mylog(log_trace,"data_len=%d new_len=%d\n",data_len,new_len); //dest.conv=conv; from_normal_to_fec(conn_info,new_data,new_len,out_n,out_arr,out_len,out_delay); @@ -964,6 +977,7 @@ int unit_test() 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()); @@ -995,6 +1009,70 @@ int unit_test() } } + for(int i=0;i<10;i++) + { + string a = "aaaaaaaaaaaaaaaaaaaaaaa"; + string b = "bbbbbbbbbbbbb"; + string c = "cccccccccccccccccc"; + + + + printf("======\n"); + int n; + char **s_arr; + int * len; + fec_decode_manager.output(n,s_arr,len); + + fec_encode_manager.re_init(3,2,fec_mtu,fec_pending_num,fec_pending_time,1); + + fec_encode_manager.input((char *) a.c_str(), a.length()); + fec_encode_manager.output(n,s_arr,len); + assert(n==1); + + fec_decode_manager.input(s_arr[0],len[0]); + + fec_decode_manager.output(n,s_arr,len); + assert(n==1); + printf("%s\n",s_arr[0]); + + fec_encode_manager.input((char *) b.c_str(), b.length()); + fec_encode_manager.output(n,s_arr,len); + assert(n==1); + //fec_decode_manager.input(s_arr[0],len[0]); + + + fec_encode_manager.input((char *) c.c_str(), c.length()); + fec_encode_manager.output(n,s_arr,len); + + assert(n==3); + + fec_decode_manager.input(s_arr[0],len[0]); + //printf("n=%d\n",n); + + + { + int n; + char **s_arr; + int * len; + fec_decode_manager.output(n,s_arr,len); + assert(n==1); + printf("%s\n",s_arr[0]); + } + + fec_decode_manager.input(s_arr[1],len[1]); + + { + int n; + char **s_arr; + int * len; + fec_decode_manager.output(n,s_arr,len); + assert(n==1); + printf("n=%d\n",n); + s_arr[0][len[0]]=0; + printf("%s\n",s_arr[0]); + } + + } return 0; }