diff --git a/fec_manager.cpp b/fec_manager.cpp index 6fc215a..f301661 100644 --- a/fec_manager.cpp +++ b/fec_manager.cpp @@ -257,12 +257,21 @@ int fec_encode_manager_t::input(char *s,int len/*,int &is_first_packet*/) for(int i=0;i=1); + assert(counter<=255); + int buf_idx=counter-1; + assert(ready_for_output==0); + ready_for_output=1; + output_n=1; + + int tmp_idx=0; + write_u32(buf[buf_idx]+tmp_idx,seq); + tmp_idx+=sizeof(u32_t); + buf[buf_idx][tmp_idx++]=(unsigned char)type; + buf[buf_idx][tmp_idx++]=(unsigned char)0; + buf[buf_idx][tmp_idx++]=(unsigned char)0; + buf[buf_idx][tmp_idx++]=(unsigned char)((u32_t)counter); + + output_len[0]=buf_s_len[buf_idx]+tmp_idx; + output_buf[0]=buf[buf_idx]; + } + } if(s!=0&&delayed_append) { @@ -356,6 +413,7 @@ int fec_decode_manager_t::input(char *s,int len) int inner_index=(unsigned char)s[tmp_idx++]; len=len-tmp_idx; + mylog(log_info,"input\n"); if(len<0) { mylog(log_warn,"len<0\n"); @@ -377,28 +435,50 @@ int fec_decode_manager_t::input(char *s,int len) if(data_num+redundant_num>max_fec_packet_num) { + mylog(log_info,"failed here\n"); return -1; } if(!anti_replay.is_vaild(seq)) { + mylog(log_info,"failed here2\n"); return 0; } - if(!mp[seq].empty()) + + int ok=1; + if(mp[seq].type==-1) + mp[seq].type=type; + else { - 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].type!=type) - ok=0; - if(type==0&&fec_data[first_idx].len!=len) - ok=0; - if(ok==0) + if(mp[seq].type!=type) ok=0; + } + + if(type==0) assert(data_num!=0); + + if(data_num!=0) + { + mp[seq].data_counter++; + + if(mp[seq].data_num==-1) { - return -1; + mp[seq].data_num=data_num; + mp[seq].redundant_num=redundant_num; + mp[seq].len=len; } + else + { + if(mp[seq].data_num!=data_num||mp[seq].redundant_num!=redundant_num||mp[seq].len!=len) + ok=0; + } + } + + if(ok==0) + { + mylog(log_info,"ok=0\n"); + return -1; + } + else + { + mylog(log_info,"ok=1\n"); } if(fec_data[index].used!=0) @@ -423,12 +503,70 @@ int fec_decode_manager_t::input(char *s,int len) 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; + mp[seq].group_mp[inner_index]=index; - map &inner_mp=mp[seq]; - assert((int)inner_mp.size()<=data_num); - if((int)inner_mp.size()==data_num) + map &inner_mp=mp[seq].group_mp; + + + int about_to_fec=0; + if(type==0) { + assert((int)inner_mp.size()<=data_num); + if((int)inner_mp.size()==data_num) + about_to_fec=1; + } + else + { + if(mp[seq].data_num!=-1) + { + data_num=mp[seq].data_num; + redundant_num=mp[seq].redundant_num; + + if(mp[seq].data_counter>data_num) + { + mylog(log_warn,"unexpected mp[seq].data_counter>mp[seq].data_num\n"); + anti_replay.set_invaild(seq); + return -1; + } + else if(mp[seq].data_counter==data_num) + { + 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+1) + { + 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+1) + { + anti_replay.set_invaild(seq); + if(data_num==0) + return 0; + else + { + //mylog(log_warn,"data_num!=0\n"); //this is possible + return -1; + } + } + + if((int)inner_mp.size()==data_num) + about_to_fec=1; + } + } + else + { + + } + //for() + } + + if(about_to_fec) + { + mylog(log_error,"fec here!\n"); if(type==0) { char *fec_tmp_arr[max_fec_packet_num+5]={0}; @@ -533,7 +671,7 @@ int fec_decode_manager_t::input(char *s,int len) if(decode_fast_send) { - if(type==1&&inner_index group_mp; +}; class fec_decode_manager_t { anti_replay_t anti_replay; fec_data_t fec_data[fec_buff_size]; int index; - unordered_map > mp; + unordered_map mp; blob_decode_t blob_decode; diff --git a/main.cpp b/main.cpp index ca0f2cb..2d8dd4a 100644 --- a/main.cpp +++ b/main.cpp @@ -553,7 +553,7 @@ int server_event_loop() { uint64_t value; read(timer.get_timer_fd(), &value, 8); - conn_manager.clear_inactive(); + ///conn_manager.clear_inactive();///////////////////////////remember to uncomment //conn_info.conv_manager.clear_inactive(); } else if (events[idx].data.u64 == (u64_t)local_listen_fd) @@ -614,6 +614,7 @@ int server_event_loop() int out_n;char **out_arr;int *out_len;int *out_delay; from_fec_to_normal(conn_info,data,data_len,out_n,out_arr,out_len,out_delay); + mylog(log_info,"out_n= %d\n",out_n); for(int i=0;i