diff --git a/fec_manager.cpp b/fec_manager.cpp index 4c20c79..77db1e0 100644 --- a/fec_manager.cpp +++ b/fec_manager.cpp @@ -120,7 +120,7 @@ fec_encode_manager_t::fec_encode_manager_t() } timer_fd64=fd_manager.create(timer_fd); - re_init(4,2,1200,100,10000); + re_init(4,2,1200,100,10000,0); } fec_encode_manager_t::~fec_encode_manager_t() { @@ -130,13 +130,14 @@ u64_t fec_encode_manager_t::get_timer_fd64() { return timer_fd64; } -int fec_encode_manager_t::re_init(int data_num,int redundant_num,int mtu,int pending_num,int pending_time) +int fec_encode_manager_t::re_init(int data_num,int redundant_num,int mtu,int pending_num,int pending_time,int type) { fec_data_num=data_num; fec_redundant_num=redundant_num; fec_mtu=mtu; fec_pending_num=pending_num; fec_pending_time=pending_time; + this->type=type; counter=0; blob_encode.clear(); @@ -150,41 +151,136 @@ int fec_encode_manager_t::re_init(int data_num,int redundant_num,int mtu,int pen return 0; } +int fec_encode_manager_t::append(char *s,int len/*,int &is_first_packet*/) +{ + if(counter==0) + { + itimerspec its; + memset(&its.it_interval,0,sizeof(its.it_interval)); + my_time_t tmp_time=fec_pending_time+get_current_time_us(); + its.it_value.tv_sec=tmp_time/1000000llu; + its.it_value.tv_nsec=(tmp_time%1000000llu)*1000llu; + timerfd_settime(timer_fd,TFD_TIMER_ABSTIME,&its,0); + } + + if(type==0) + { + blob_encode.input(s,len); + } + else if(type==1) + { + assert(len<=65535&&len>=0); + write_u16(buf[counter]+sizeof(u32_t)+4*sizeof(char),(u16_t)((u32_t)len)); + memcpy(buf[counter]+sizeof(u32_t)+4*sizeof(char)+sizeof(u16_t),s,len);//remember to change this,if protocol is modified + buf_s_len[counter]=len+sizeof(u16_t); + } + else + { + assert(0==1); + } + counter++; + return 0; +} int fec_encode_manager_t::input(char *s,int len/*,int &is_first_packet*/) { - if(s==0 ||blob_encode.get_shard_len(fec_data_num,len)>=fec_mtu||counter>=fec_pending_num) + int about_to_fec=0; + int delayed_append=0; + if(type==0&& s!=0 &&blob_encode.get_num()==0&&blob_encode.get_shard_len(fec_data_num,len)>=fec_mtu) + { + mylog(log_warn,"message too long len=%d,ignored\n",len); + return -1; + } + if(type==1&&s!=0&&len>=fec_mtu) + { + mylog(log_warn,"message too long len=%d,ignored\n",len); + return -1; + } + if(s==0) about_to_fec=1;//now + assert(type==0||type==1); + 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(counter0); + if(buf_s_len[i]>fec_len) fec_len=buf_s_len[i]; + } + + } + + for(int i=0;imax_fec_packet_num) { return -1; @@ -274,15 +369,16 @@ int fec_decode_manager_t::input(char *s,int len) ok=0; if(fec_data[first_idx].redundant_num!=redundant_num) ok=0; - if(fec_data[first_idx].len!=len) - 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) { return 0; } } + if(fec_data[index].used!=0) { int tmp_seq=fec_data[index].seq; @@ -303,9 +399,6 @@ int fec_decode_manager_t::input(char *s,int 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 &inner_mp=mp[seq]; assert((int)inner_mp.size()<=data_num); if((int)inner_mp.size()==data_num) @@ -324,14 +417,81 @@ int fec_decode_manager_t::input(char *s,int len) blob_decode.input(fec_tmp_arr[i],len); } blob_decode.output(output_n,output_s_arr,output_len_arr); + assert(ready_for_output==0); ready_for_output=1; anti_replay.set_invaild(seq); } else { + int max_len=-1; + int fec_ok=1; + //outupt_s_arr_buf[max_fec_packet_num+5]={0}; + 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>=0); + if(fec_data[it->second].len > max_len) + max_len=fec_data[it->second].len; + } + 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 missed_packet[max_fec_packet_num+5]; + int missed_packet_counter=0; + for(int i=0;imax_data_len) + { + fec_ok=0; + break; + } + } + if(fec_ok) + { + output_n=missed_packet_counter; + for(int i=0;i