text: /*
   * 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;
  
  const int rs_str_len=max_fec_packet_num*10+100;
  extern int header_overhead;
  extern int debug_fec;
  
  struct fec_parameter_t
  {
  	int version=0;
  	int mtu=1250;
  	int queue_len=200;
  	int timeout=8*1000;
  	int mode=0;
  
  	int rs_cnt=0;
  	struct rs_parameter_t //parameters for reed solomon
  	{
  		unsigned char x;//AKA fec_data_num  (x should be same as <index of rs_par>+1 at the moment)
  		unsigned char y;//fec_redundant_num
  	}rs_par[max_fec_packet_num+10];
  
  	int rs_from_str(char * s)//todo inefficient
  	{
  		vector<string> str_vec=string_to_vec(s,",");
  		if(str_vec.size()<1)
  		{
  			mylog(log_warn,"failed to parse [%s]\n",s);
  			return -1;
  		}
  		vector<rs_parameter_t> par_vec;
  		for(int i=0;i<(int)str_vec.size();i++)
  		{
  			rs_parameter_t tmp_par;
  			string &tmp_str=str_vec[i];
  			int x,y;
  			if(sscanf((char *)tmp_str.c_str(),"%d:%d",&x,&y)!=2)
  			{
  				mylog(log_warn,"failed to parse [%s]\n",tmp_str.c_str());
  				return -1;
  			}
  			if(x<1||y<0||x+y>max_fec_packet_num)
  			{
  				mylog(log_warn,"invaild value x=%d y=%d, x should >=1, y should >=0, x +y should <%d\n",x,y,max_fec_packet_num);
  				return -1;
  			}
  			tmp_par.x=x;
  			tmp_par.y=y;
  			par_vec.push_back(tmp_par);
  		}
  		assert(par_vec.size()==str_vec.size());
  
  		int found_problem=0;
  		for(int i=1;i<(int)par_vec.size();i++)
  		{
  			if(par_vec[i].x<=par_vec[i-1].x)
  			{
  				mylog(log_warn,"error in [%s], x in x:y should be in ascend order\n",s);
  				return -1;
  			}
  			int now_x=par_vec[i].x;
  			int now_y=par_vec[i].y;
  			int pre_x=par_vec[i-1].x;
  			int pre_y=par_vec[i-1].y;
  
  			double now_ratio=double(par_vec[i].y)/par_vec[i].x;
  			double pre_ratio=double(par_vec[i-1].y)/par_vec[i-1].x;
  
  			if(pre_ratio+0.0001<now_ratio)
  			{
  				if(found_problem==0)
  				{
  					mylog(log_warn,"possible problems: %d/%d<%d/%d",pre_y,pre_x,now_y,now_x);
  					found_problem=1;
  				}
  				else
  				{
  					log_bare(log_warn,", %d/%d<%d/%d",pre_y,pre_x,now_y,now_x);
  				}
  			}
  		}
  		if(found_problem)
  		{
  			log_bare(log_warn," in %s\n",s);
  		}
  
  		{ //special treatment for first parameter
  			int x=par_vec[0].x;
  			int y=par_vec[0].y;
  			for(int i=1;i<=x;i++)
  			{
  				rs_par[i-1].x=i;
  				rs_par[i-1].y=y;
  			}
  		}
  
  		for(int i=1;i<(int)par_vec.size();i++)
  		{
  			int now_x=par_vec[i].x;
  			int now_y=par_vec[i].y;
  			int pre_x=par_vec[i-1].x;
  			int pre_y=par_vec[i-1].y;
  			rs_par[now_x-1].x=now_x;
  			rs_par[now_x-1].y=now_y;
  
  			double now_ratio=double(par_vec[i].y)/par_vec[i].x;
  			double pre_ratio=double(par_vec[i-1].y)/par_vec[i-1].x;
  
  			//double k= double(now_y-pre_y)/double(now_x-pre_x);
  			for(int j=pre_x+1;j<=now_x-1;j++)
  			{
  				int in_x=j;
  
  			////////	int in_y= double(pre_y) + double(in_x-pre_x)*k+ 0.9999;// round to upper
  
  			double distance=now_x-pre_x;
  			///////	double in_ratio=pre_ratio*(1.0-(in_x-pre_x)/distance)   +   now_ratio *(1.0- (now_x-in_x)/distance);
  			//////	int in_y= in_x*in_ratio + 0.9999;
  				int in_y= pre_y +(now_y-pre_y) *(in_x-pre_x)/distance +0.9999;
  
  				if(in_x+in_y>max_fec_packet_num)
  				{
  					in_y=max_fec_packet_num-in_x;
  					assert(in_y>=0&&in_y<=max_fec_packet_num);
  				}
  
  				rs_par[in_x-1].x=in_x;
  				rs_par[in_x-1].y=in_y;
  			}
  		}
  		rs_cnt=par_vec[par_vec.size()-1].x;
  
  		return 0;
  	}
  
  	char *rs_to_str()//todo inefficient
  	{
  		static char res[rs_str_len];
  		string tmp_string;
  		char tmp_buf[100];
  		assert(rs_cnt>=1);
  		for(int i=0;i<rs_cnt;i++)
  		{
  			sprintf(tmp_buf,"%d:%d",int(rs_par[i].x),int(rs_par[i].y));
  			if(i!=0)
  				tmp_string+=",";
  			tmp_string+=tmp_buf;
  		}
  		strcpy(res,tmp_string.c_str());
  		return res;
  	}
  
  	rs_parameter_t get_tail()
  	{
  		assert(rs_cnt>=1);
  		return rs_par[rs_cnt-1];
  	}
  
  
  	int clone(fec_parameter_t & other)
  	{
  		version=other.version;
  		mtu=other.mtu;
  		queue_len=other.queue_len;
  		timeout=other.timeout;
  		mode=other.mode;
  
  		assert(other.rs_cnt>=1);
  		rs_cnt=other.rs_cnt;
  		memcpy(rs_par,other.rs_par,sizeof(rs_parameter_t)*rs_cnt);
  
  		return 0;
  	}
  
  
  };
  
  extern fec_parameter_t g_fec_par;
  //extern int dynamic_update_fec;
  
  const int anti_replay_timeout=60*1000;// 60s
  
  struct anti_replay_t
  {
  
  	struct info_t
  	{
  		my_time_t my_time;
  		int index;
  	};
  
  	u64_t replay_buffer[anti_replay_buff_size];
  	unordered_map<u32_t,info_t> mp;
  	int index;
  	anti_replay_t()
  	{
  		clear();
  	}
  	int clear()
  	{
  		memset(replay_buffer,-1,sizeof(replay_buffer));
  		mp.clear();
  		mp.rehash(anti_replay_buff_size*3);
  		index=0;
  		return 0;
  	}
  	void set_invaild(u32_t seq)
  	{
  
  		if(is_vaild(seq)==0)
  		{
  			mylog(log_trace,"seq %u exist\n",seq);
  			//assert(mp.find(seq)!=mp.end());
  			//mp[seq].my_time=get_current_time_rough();
  			return;
  		}
  		if(replay_buffer[index]!=u64_t(i64_t(-1)))
  		{
  			assert(mp.find(replay_buffer[index])!=mp.end());
  			mp.erase(replay_buffer[index]);
  		}
  		replay_buffer[index]=seq;
  		assert(mp.find(seq)==mp.end());
  		mp[seq].my_time=get_current_time_rough();
  		mp[seq].index=index;
  		index++;
  		if(index==int(anti_replay_buff_size)) index=0;
  	}
  	int is_vaild(u32_t seq)
  	{
  		if(mp.find(seq)==mp.end()) return 1;
  		
  		if(get_current_time_rough()-mp[seq].my_time>anti_replay_timeout)
  		{
  			replay_buffer[mp[seq].index]=u64_t(i64_t(-1));
  			mp.erase(seq);
  			return 1;
  		}
  
  		return 0;
  	}
  };
  
  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:not_copy_able_t
  {
  
  private:
  	u32_t seq;
  
  	//int fec_mode;
  	//int fec_data_num,fec_redundant_num;
  	//int fec_mtu;
  	//int fec_queue_len;
  	//int fec_timeout;
  	fec_parameter_t fec_par;
  
  
  	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);
  
  	ev_timer timer;
  	struct ev_loop *loop=0;
  	void (*cb) (struct ev_loop *loop, struct ev_timer *watcher, int revents)=0;
  
  public:
  	fec_encode_manager_t();
  	~fec_encode_manager_t();
  
  	fec_parameter_t & get_fec_par()
  	{
  		return fec_par;
  	}
  	void set_data(void * data)
  	{
  		timer.data=data;
  	}
  
  
  	void set_loop_and_cb(struct ev_loop *loop,void (*cb) (struct ev_loop *loop, struct ev_timer *watcher, int revents))
  	{
  		this->loop=loop;
  		this->cb=cb;
  		ev_init(&timer,cb);
  	}
  
  	int clear_data()
  	{
  		counter=0;
  		blob_encode.clear();
  		ready_for_output=0;
  
  		seq=(u32_t)get_fake_random_number(); //TODO temp solution for a bug.
  
  		if(loop)
  		{
  			ev_timer_stop(loop,&timer);
  		}
  		return 0;
  	}
  	int clear_all()
  	{
  
  		//itimerspec zero_its;
  		//memset(&zero_its, 0, sizeof(zero_its));
  		//timerfd_settime(timer_fd, TFD_TIMER_ABSTIME, &zero_its, 0);
  
  		if(loop)
  		{
  			ev_timer_stop(loop,&timer);
  			loop=0;
  			cb=0;
  		}
  
  		clear_data();
  
  		return 0;
  	}
  
  	my_time_t get_first_packet_time()
  	{
  		return first_packet_time_for_output;
  	}
  
  	int get_pending_time()
  	{
  		return fec_par.timeout;
  	}
  
  	int get_type()
  	{
  		return fec_par.mode;
  	}
  	//u64_t get_timer_fd64();
  	int reset_fec_parameter(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:not_copy_able_t
  {
  	anti_replay_t anti_replay;
  	fec_data_t *fec_data=0;
  	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];
  		assert(fec_data!=0);
  		clear();
  	}
  	/*
  	fec_decode_manager_t(const fec_decode_manager_t &b)
  	{
  		assert(0==1);//not allowed to copy
  	}*/
  	~fec_decode_manager_t()
  	{
  		mylog(log_debug,"fec_decode_manager destroyed\n");
  		if(fec_data!=0)
  		{
  			mylog(log_debug,"fec_data freed\n");
  			delete fec_data;
  		}
  	}
  	int clear()
  	{
  		anti_replay.clear();
  		mp.clear();
  		mp.rehash(fec_buff_num*3);
  
  		for(int i=0;i<(int)fec_buff_num;i++)
  			fec_data[i].used=0;
  		ready_for_output=0;
  		index=0;
  
  		return 0;
  	}
  
  	//int re_init();
  	int input(char *s,int len);
  	int output(int &n,char ** &s_arr,int* &len_arr);
  };
  
  #endif /* FEC_MANAGER_H_ */
  