UDPspeeder/connection.h

/*
 * connection.h
 *
 *  Created on: Sep 23, 2017
 *      Author: root
 */

#ifndef CONNECTION_H_
#define CONNECTION_H_

extern int disable_anti_replay;

#include "connection.h"
#include "common.h"
#include "log.h"
#include "delay_manager.h"
#include "fd_manager.h"
#include "fec_manager.h"

extern int report_interval;

const int disable_conv_clear=0;

void server_clear_function(u64_t u64);

template  <class T>
struct conv_manager_t  // manage the udp connections
{
	//typedef hash_map map;
	unordered_map<T,u32_t> data_to_conv;  //conv and u64 are both supposed to be uniq
	unordered_map<u32_t,T> conv_to_data;

	lru_collector_t<u32_t> lru;
	//unordered_map<u32_t,u64_t> conv_last_active_time;

	//unordered_map<u32_t,u64_t>::iterator clear_it;

	void (*additional_clear_function)(T data) =0;

	long long last_clear_time;

	conv_manager_t()
		{
			//clear_it=conv_last_active_time.begin();
			long long last_clear_time=0;
			additional_clear_function=0;
		}
	~conv_manager_t()
		{
			clear();
		}
		int get_size()
		{
			return conv_to_data.size();
		}
		void reserve()
		{
			data_to_conv.reserve(10007);
			conv_to_data.reserve(10007);
			//conv_last_active_time.reserve(10007);

			lru.mp.reserve(10007);
		}
		void clear()
		{
			if(disable_conv_clear) return ;

			if(additional_clear_function!=0)
			{
				for(auto it=conv_to_data.begin();it!=conv_to_data.end();it++)
				{
					//int fd=int((it->second<<32u)>>32u);
					additional_clear_function(  it->second);
				}
			}
			data_to_conv.clear();
			conv_to_data.clear();

			lru.clear();
			//conv_last_active_time.clear();

			//clear_it=conv_last_active_time.begin();

		}
		u32_t get_new_conv()
		{
			u32_t conv=get_fake_random_number_nz();
			while(conv_to_data.find(conv)!=conv_to_data.end())
			{
				conv=get_fake_random_number_nz();
			}
			return conv;
		}
		int is_conv_used(u32_t conv)
		{
			return conv_to_data.find(conv)!=conv_to_data.end();
		}
		int is_data_used(T data)
		{
			return data_to_conv.find(data)!=data_to_conv.end();
		}
		u32_t find_conv_by_data(T data)
		{
			return data_to_conv[data];
		}
		T find_data_by_conv(u32_t conv)
		{
			return conv_to_data[conv];
		}
		int update_active_time(u32_t conv)
		{
			//return conv_last_active_time[conv]=get_current_time();
			lru.update(conv);
			return 0;
		}
		int insert_conv(u32_t conv,T data)
		{
			data_to_conv[data]=conv;
			conv_to_data[conv]=data;
			//conv_last_active_time[conv]=get_current_time();
			lru.new_key(conv);
			return 0;
		}
		int erase_conv(u32_t conv)
		{
			if(disable_conv_clear) return 0;
			T data=conv_to_data[conv];
			if(additional_clear_function!=0)
			{
				additional_clear_function(data);
			}
			conv_to_data.erase(conv);
			data_to_conv.erase(data);
			//conv_last_active_time.erase(conv);
			lru.erase(conv);
			return 0;
		}
		int clear_inactive(char * info=0)
		{
			if(get_current_time()-last_clear_time>conv_clear_interval)
			{
				last_clear_time=get_current_time();
				return clear_inactive0(info);
			}
			return 0;
		}
		int clear_inactive0(char * info)
		{
			if(disable_conv_clear) return 0;


			unordered_map<u32_t,u64_t>::iterator it;
			unordered_map<u32_t,u64_t>::iterator old_it;

			//map<uint32_t,uint64_t>::iterator it;
			int cnt=0;
			//it=clear_it;
			int size=lru.size();
			int num_to_clean=size/conv_clear_ratio+conv_clear_min;   //clear 1/10 each time,to avoid latency glitch

			num_to_clean=min(num_to_clean,size);

			my_time_t current_time=get_current_time();
			for(;;)
			{
				if(cnt>=num_to_clean) break;
				if(lru.empty()) break;

				u32_t conv;
				my_time_t ts=lru.peek_back(conv);

				if(current_time- ts < conv_timeout) break;

				erase_conv(conv);
				if(info==0)
				{
					mylog(log_info,"conv %x cleared\n",conv);
				}
				else
				{
					mylog(log_info,"[%s]conv %x cleared\n",info,conv);
				}
				cnt++;
			}
			return 0;
		}


		/*
	conv_manager_t();
	~conv_manager_t();
	int get_size();
	void reserve();
	void clear();
	u32_t get_new_conv();
	int is_conv_used(u32_t conv);
	int is_u64_used(T u64);
	u32_t find_conv_by_u64(T u64);
	T find_u64_by_conv(u32_t conv);
	int update_active_time(u32_t conv);
	int insert_conv(u32_t conv,T u64);
	int erase_conv(u32_t conv);
	int clear_inactive(char * ip_port=0);
	int clear_inactive0(char * ip_port);*/
};//g_conv_manager;



struct inner_stat_t
{
	u64_t input_packet_num;
	u64_t input_packet_size;
	u64_t output_packet_num;
	u64_t output_packet_size;
};
struct stat_t
{
	u64_t last_report_time;
	inner_stat_t normal_to_fec;
	inner_stat_t fec_to_normal;
	stat_t()
	{
		memset(this,0,sizeof(stat_t));
	}
	void report_as_client()
	{
		if(report_interval!=0 &&get_current_time()-last_report_time>u64_t(report_interval)*1000)
		{
			last_report_time=get_current_time();
			inner_stat_t &a=normal_to_fec;
			inner_stat_t &b=fec_to_normal;
			mylog(log_info,"[report]client-->server:(original:%llu pkt;%llu byte) (fec:%llu pkt,%llu byte)  server-->client:(original:%llu pkt;%llu byte) (fec:%llu pkt;%llu byte)\n",
					a.input_packet_num,a.input_packet_size,a.output_packet_num,a.output_packet_size,
					b.output_packet_num,b.output_packet_size,b.input_packet_num,b.input_packet_size
					);
		}
	}
	void report_as_server(address_t &addr)
	{
		if(report_interval!=0 &&get_current_time()-last_report_time>u64_t(report_interval)*1000)
		{
			last_report_time=get_current_time();
			inner_stat_t &a=fec_to_normal;
			inner_stat_t &b=normal_to_fec;
			mylog(log_info,"[report][%s]client-->server:(original:%llu pkt;%llu byte) (fec:%llu pkt;%llu byte)  server-->client:(original:%llu pkt;%llu byte) (fec:%llu pkt;%llu byte)\n",
					addr.get_str(),
					a.output_packet_num,a.output_packet_size,a.input_packet_num,a.input_packet_size,
					b.input_packet_num,b.input_packet_size,b.output_packet_num,b.output_packet_size
					);
		}
	}
};


struct conn_info_t     //stores info for a raw connection.for client ,there is only one connection,for server there can be thousand of connection since server can
//handle multiple clients
{
	union  tmp_union_t
	{
		conv_manager_t<address_t> c;
		conv_manager_t<u64_t> s;
		//avoid templates here and there, avoid pointer and type cast
		tmp_union_t()
		{
			if(program_mode==client_mode)
			{
				new( &c ) conv_manager_t<address_t>();
			}
			else
			{
				assert(program_mode==server_mode);
				new( &s ) conv_manager_t<u64_t>();
			}
		}
		~tmp_union_t()
		{
			if(program_mode==client_mode)
			{
				c.~conv_manager_t<address_t>();
			}
			else
			{
				assert(program_mode==server_mode);
				s.~conv_manager_t<u64_t>();
			}
		}
	}conv_manager;


	fec_encode_manager_t fec_encode_manager;
	fec_decode_manager_t fec_decode_manager;
	ev_timer timer;
	//my_timer_t timer;

	u64_t last_active_time;
	stat_t stat;

	ev_loop* loop=0;
	int local_listen_fd;

	int remote_fd;  //only used for client
	fd64_t remote_fd64;//only used for client

	//ip_port_t ip_port;
	address_t  addr;//only used for server

	conn_info_t()
	{
		if(program_mode==server_mode)
		{
			conv_manager.s.additional_clear_function=server_clear_function;
		}
		else
		{
			assert(program_mode==client_mode);
		}
	}

	~conn_info_t()
	{
		if(loop)
			ev_timer_stop(loop,&timer);
	}
	void update_active_time()
	{
		last_active_time=get_current_time();
	}
	conn_info_t(const conn_info_t &b)
	{
		assert(0==1);
	}
};
/*
struct conn_manager_t  //manager for connections. for client,we dont need conn_manager since there is only one connection.for server we use one conn_manager for all connections
{

	unordered_map<u64_t,conn_info_t*> mp;//<ip,port> to conn_info_t;
	unordered_map<u64_t,conn_info_t*>::iterator clear_it;
	long long last_clear_time;

	conn_manager_t();
	conn_manager_t(const conn_info_t &b)
	{
		assert(0==1);
	}
	int exist(ip_port_t);
	conn_info_t *& find_p(ip_port_t);  //be aware,the adress may change after rehash
	conn_info_t & find(ip_port_t) ; //be aware,the adress may change after rehash
	int insert(ip_port_t);

	int erase(unordered_map<u64_t,conn_info_t*>::iterator erase_it);
	int clear_inactive();
	int clear_inactive0();

};*/

struct conn_manager_t  //manager for connections. for client,we dont need conn_manager since there is only one connection.for server we use one conn_manager for all connections
{


 unordered_map<address_t,conn_info_t*> mp; //put it at end so that it de-consturcts first
 unordered_map<address_t,conn_info_t*>::iterator clear_it;

 long long last_clear_time;

 conn_manager_t();
 int exist(address_t addr);
 conn_info_t *& find_insert_p(address_t addr);  //be aware,the adress may change after rehash //not true?
 conn_info_t & find_insert(address_t addr) ; //be aware,the adress may change after rehash

 int erase(unordered_map<address_t,conn_info_t*>::iterator erase_it);
int clear_inactive();
int clear_inactive0();

};



extern conn_manager_t conn_manager;


#endif /* CONNECTION_H_ */