UDPspeeder/tunnel_server.cpp

/*
 * tunnel.cpp
 *
 *  Created on: Oct 26, 2017
 *      Author: root
 */

#include "tunnel.h"

static void conn_timer_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents);
static void fec_encode_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents);
static void remote_cb(struct ev_loop *loop, struct ev_io *watcher, int revents);

enum tmp_mode_t{is_from_remote=0,is_fec_timeout,is_conn_timer};

void data_from_remote_or_fec_timeout_or_conn_timer(conn_info_t & conn_info,fd64_t fd64,tmp_mode_t mode)
{
	int ret;

	char data[buf_len];
	int data_len;
	u32_t conv;
	//fd64_t fd64=events[idx].data.u64;
	//mylog(log_trace,"events[idx].data.u64 >u32_t(-1),%llu\n",(u64_t)events[idx].data.u64);


	//assert(fd_manager.exist_info(fd64));
	//ip_port_t ip_port=fd_manager.get_info(fd64).ip_port;


	//conn_info_t &conn_info=conn_manager.find(ip_port);
	address_t &addr=conn_info.addr;
	assert(conn_manager.exist(addr));

	int &local_listen_fd=conn_info.local_listen_fd;

	int  out_n=-2;char **out_arr;int *out_len;my_time_t *out_delay;

	dest_t dest;
	dest.inner.fd_addr.fd=local_listen_fd;
	dest.inner.fd_addr.addr=addr;
	dest.type=type_fd_addr;
	dest.cook=1;

	if(mode==is_fec_timeout)
	{
		assert(fd64==0);
		//uint64_t value;
		//if((ret=read(fd_manager.to_fd(fd64), &value, 8))!=8)
		//{
		//	mylog(log_trace,"fd_manager.to_fd(fd64), &value, 8)!=8 ,%d\n",ret);
		//	continue;
		//}
		//if(value==0)
		//{
		//	mylog(log_trace,"value==0\n");
		//	continue;
		//}
		//assert(value==1);
		from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
	}
	else if(mode==is_conn_timer)
	{
		assert(fd64==0);
		//uint64_t value;
		//read(conn_info.timer.get_timer_fd(), &value, 8);
		conn_info.conv_manager.s.clear_inactive();
		if(debug_force_flush_fec)
		{
		from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
		}

		conn_info.stat.report_as_server(addr);
		return;
	}
	else if(mode==is_from_remote)
	{
		if(!fd_manager.exist(fd64))   //fd64 has been closed
		{
			mylog(log_warn,"!fd_manager.exist(fd64)\n");
			return;
		}

		//fd64_t &fd64 =conn_info.remote_fd64;
		assert(conn_info.conv_manager.s.is_data_used(fd64));

		conv=conn_info.conv_manager.s.find_conv_by_data(fd64);
		conn_info.conv_manager.s.update_active_time(conv);
		conn_info.update_active_time();

		int fd=fd_manager.to_fd(fd64);
		data_len=recv(fd,data,max_data_len+1,0);

		if(data_len==max_data_len+1)
		{
			mylog(log_warn,"huge packet, data_len > %d, packet truncated, dropped\n",max_data_len);
			return ;
		}

		mylog(log_trace,"received a packet from udp_fd,len:%d,conv=%d\n",data_len,conv);

		if(data_len<0)
		{
			mylog(log_debug,"udp fd,recv_len<0 continue,%s\n",get_sock_error());

			return;
		}

		if(!disable_mtu_warn&&data_len>=mtu_warn)
		{
			mylog(log_warn,"huge packet,data len=%d (>=%d).strongly suggested to set a smaller mtu at upper level,to get rid of this warn\n ",data_len,mtu_warn);
		}

		char * new_data;
		int new_len;
		put_conv(conv,data,data_len,new_data,new_len);

		from_normal_to_fec(conn_info,new_data,new_len,out_n,out_arr,out_len,out_delay);
	}
	else
	{
		assert(0==1);
	}

	mylog(log_trace,"out_n=%d\n",out_n);
	for(int i=0;i<out_n;i++)
	{
		delay_send(out_delay[i],dest,out_arr[i],out_len[i]);
	}
}

static void local_listen_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	int local_listen_fd=watcher->fd;
	int ret;

	mylog(log_trace,"events[idx].data.u64 == (u64_t)local_listen_fd\n");
	char data[buf_len];
	int data_len;
	address_t::storage_t udp_new_addr_in={0};
	socklen_t udp_new_addr_len = sizeof(address_t::storage_t);
	if ((data_len = recvfrom(local_listen_fd, data, max_data_len+1, 0,
			(struct sockaddr *) &udp_new_addr_in, &udp_new_addr_len)) == -1) {
		mylog(log_error,"recv_from error,this shouldnt happen,err=%s,but we can try to continue\n",get_sock_error());
		return;
	};

	if(data_len==max_data_len+1)
        {
                mylog(log_warn,"huge packet, data_len > %d, packet truncated, dropped\n",max_data_len);
                return ;
        }

	address_t addr;
	addr.from_sockaddr((struct sockaddr *) &udp_new_addr_in,udp_new_addr_len);

	mylog(log_trace,"Received packet from %s,len: %d\n", addr.get_str(),data_len);

	if(!disable_mtu_warn&&data_len>=mtu_warn)///////////////////////delete this for type 0 in furture
	{
		mylog(log_warn,"huge packet,data len=%d (>=%d).strongly suggested to set a smaller mtu at upper level,to get rid of this warn\n ",data_len,mtu_warn);
	}


	if(de_cook(data,data_len)!=0)
	{
		mylog(log_debug,"de_cook error");
		return;
	}



	if(!conn_manager.exist(addr))
	{
		if(conn_manager.mp.size() >=max_conn_num)
		{
			mylog(log_warn,"new connection %s ignored bc max_conn_num exceed\n",addr.get_str());
			return;
		}

		//conn_manager.insert(addr);
		conn_info_t &conn_info=conn_manager.find_insert(addr);
		conn_info.addr=addr;
		conn_info.loop=ev_default_loop(0);
		conn_info.local_listen_fd=local_listen_fd;

		//u64_t fec_fd64=conn_info.fec_encode_manager.get_timer_fd64();
		//mylog(log_debug,"fec_fd64=%llu\n",fec_fd64);
		//ev.events = EPOLLIN;
		//ev.data.u64 = fec_fd64;
		//ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd_manager.to_fd(fec_fd64), &ev);

		//fd_manager.get_info(fec_fd64).ip_port=ip_port;

	    conn_info.timer.data=&conn_info;
	    ev_init(&conn_info.timer,conn_timer_cb);
	    ev_timer_set(&conn_info.timer, 0, timer_interval/1000.0 );
	    ev_timer_start(loop,&conn_info.timer);

		//conn_info.timer.add_fd64_to_epoll(epoll_fd);
		//conn_info.timer.set_timer_repeat_us(timer_interval*1000);

		//mylog(log_debug,"conn_info.timer.get_timer_fd64()=%llu\n",conn_info.timer.get_timer_fd64());

		//u64_t timer_fd64=conn_info.timer.get_timer_fd64();
		//fd_manager.get_info(timer_fd64).ip_port=ip_port;



	    conn_info.fec_encode_manager.set_data(&conn_info);
	    conn_info.fec_encode_manager.set_loop_and_cb(loop,fec_encode_cb);


		mylog(log_info,"new connection from %s\n",addr.get_str());

	}
	conn_info_t &conn_info=conn_manager.find_insert(addr);

	conn_info.update_active_time();
	int  out_n;char **out_arr;int *out_len;my_time_t *out_delay;
	from_fec_to_normal(conn_info,data,data_len,out_n,out_arr,out_len,out_delay);

	mylog(log_trace,"out_n= %d\n",out_n);
	for(int i=0;i<out_n;i++)
	{
		u32_t conv;
		char *new_data;
		int new_len;
		if(get_conv(conv,out_arr[i],out_len[i],new_data,new_len)!=0)
		{
			mylog(log_debug,"get_conv failed");
			continue;
		}


		if (!conn_info.conv_manager.s.is_conv_used(conv))
		{
			if(conn_info.conv_manager.s.get_size() >=max_conv_num)
			{
				mylog(log_warn,"ignored new udp connect bc max_conv_num exceed\n");
				continue;
			}

			int new_udp_fd;
			ret=new_connected_socket2(new_udp_fd,remote_addr,out_addr,out_interface);

			if (ret != 0) {
				mylog(log_warn, "[%s]new_connected_socket failed\n",addr.get_str());
				continue;
			}

			fd64_t fd64 = fd_manager.create(new_udp_fd);
			//ev.events = EPOLLIN;
			//ev.data.u64 = fd64;
			//ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_udp_fd, &ev);

			conn_info.conv_manager.s.insert_conv(conv, fd64);
			fd_manager.get_info(fd64).addr=addr;

			ev_io &io_watcher=fd_manager.get_info(fd64).io_watcher;
			io_watcher.u64=fd64;
			io_watcher.data=&conn_info;

			ev_init(&io_watcher,remote_cb);
			ev_io_set(&io_watcher,new_udp_fd,EV_READ);
			ev_io_start(conn_info.loop,&io_watcher);


			mylog(log_info,"[%s]new conv %x,fd %d created,fd64=%llu\n",addr.get_str(),conv,new_udp_fd,fd64);
		}
		conn_info.conv_manager.s.update_active_time(conv);
		fd64_t fd64= conn_info.conv_manager.s.find_data_by_conv(conv);
		dest_t dest;
		dest.type=type_fd64;
		dest.inner.fd64=fd64;
		delay_send(out_delay[i],dest,new_data,new_len);
	}
}

static void remote_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	conn_info_t & conn_info= *((conn_info_t*)watcher->data);
	fd64_t fd64=watcher->u64;

	data_from_remote_or_fec_timeout_or_conn_timer(conn_info,fd64,is_from_remote);
}

static void fifo_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	int fifo_fd=watcher->fd;

	char buf[buf_len];
	int len=read (fifo_fd, buf, sizeof (buf));
	if(len<0)
	{
		mylog(log_warn,"fifo read failed len=%d,errno=%s\n",len,get_sock_error());
		return;
	}
	buf[len]=0;
	handle_command(buf);
}

static void delay_manager_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	//uint64_t value;
	//read(delay_manager.get_timer_fd(), &value, 8);
	//mylog(log_trace,"events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()\n");

	//do nothing

}

static void fec_encode_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	conn_info_t & conn_info= *((conn_info_t*)watcher->data);

	data_from_remote_or_fec_timeout_or_conn_timer(conn_info,0,is_fec_timeout);
}

static void conn_timer_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	conn_info_t & conn_info= *((conn_info_t*)watcher->data);

	data_from_remote_or_fec_timeout_or_conn_timer(conn_info,0,is_conn_timer);
}

static void prepare_cb(struct ev_loop *loop, struct ev_prepare *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	delay_manager.check();
}

static void global_timer_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
{
	assert(!(revents&EV_ERROR));

	//uint64_t value;
	//read(timer.get_timer_fd(), &value, 8);
	conn_manager.clear_inactive();
	mylog(log_trace,"events[idx].data.u64==(u64_t)timer.get_timer_fd()\n");
}

int tunnel_server_event_loop()
{

	int i, j, k;int ret;
	int yes = 1;
	//int epoll_fd;
	//int remote_fd;

	int local_listen_fd;
    new_listen_socket2(local_listen_fd,local_addr);

	//epoll_fd = epoll_create1(0);
	//assert(epoll_fd>0);

	//const int max_events = 4096;
	//struct epoll_event ev, events[max_events];
	//if (epoll_fd < 0) {
	//	mylog(log_fatal,"epoll return %d\n", epoll_fd);
	//	myexit(-1);
	//}

	struct ev_loop * loop= ev_default_loop(0);
	assert(loop != NULL);

	//ev.events = EPOLLIN;
	//ev.data.u64 = local_listen_fd;
	//ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, local_listen_fd, &ev);
	//if (ret!=0) {
	//	mylog(log_fatal,"add  udp_listen_fd error\n");
	//	myexit(-1);
	//}
	struct ev_io local_listen_watcher;
    ev_io_init(&local_listen_watcher, local_listen_cb, local_listen_fd, EV_READ);
    ev_io_start(loop, &local_listen_watcher);

	//ev.events = EPOLLIN;
	//ev.data.u64 = delay_manager.get_timer_fd();
	//ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, delay_manager.get_timer_fd(), &ev);
	//if (ret!= 0) {
	//	mylog(log_fatal,"add delay_manager.get_timer_fd() error\n");
	//	myexit(-1);
	//}

    delay_manager.set_loop_and_cb(loop,delay_manager_cb);

	//mylog(log_debug," delay_manager.get_timer_fd() =%d\n", delay_manager.get_timer_fd());

	mylog(log_info,"now listening at %s\n",local_addr.get_str());

	//my_timer_t timer;
	//timer.add_fd_to_epoll(epoll_fd);
	//timer.set_timer_repeat_us(timer_interval*1000);

	ev_timer global_timer;
    ev_init(&global_timer,global_timer_cb);
    ev_timer_set(&global_timer, 0, timer_interval/1000.0 );
    ev_timer_start(loop,&global_timer);

	//mylog(log_debug," timer.get_timer_fd() =%d\n",timer.get_timer_fd());

    struct ev_io fifo_watcher;

	int fifo_fd=-1;

	if(fifo_file[0]!=0)
	{
		fifo_fd=create_fifo(fifo_file);
		//ev.events = EPOLLIN;
		//ev.data.u64 = fifo_fd;

		//ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fifo_fd, &ev);
		//if (ret!= 0) {
			//mylog(log_fatal,"add fifo_fd to epoll error %s\n",strerror(errno));
			//myexit(-1);
		//}
	    ev_io_init(&fifo_watcher, fifo_cb, fifo_fd, EV_READ);
	    ev_io_start(loop, &fifo_watcher);

		mylog(log_info,"fifo_file=%s\n",fifo_file);
	}

	ev_prepare prepare_watcher;
	ev_init(&prepare_watcher,prepare_cb);
	ev_prepare_start(loop,&prepare_watcher);


	ev_run(loop, 0);

	mylog(log_warn,"ev_run returned\n");
	myexit(0);

	/*
	while(1)////////////////////////
	{

		if(about_to_exit) myexit(0);

		int nfds = epoll_wait(epoll_fd, events, max_events, 180 * 1000);
		if (nfds < 0) {  //allow zero
			if(errno==EINTR  )
			{
				mylog(log_info,"epoll interrupted by signal,continue\n");
			}
			else
			{
				mylog(log_fatal,"epoll_wait return %d,%s\n", nfds,strerror(errno));
				myexit(-1);
			}
		}
		int idx;
		for (idx = 0; idx < nfds; ++idx)
		{
			if(events[idx].data.u64==(u64_t)timer.get_timer_fd())
			{

			}

			else if (events[idx].data.u64 == (u64_t)fifo_fd)
			{

			}

			else if (events[idx].data.u64 == (u64_t)local_listen_fd)
			{


			}
		    else if (events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()) {

			}
			else if (events[idx].data.u64 >u32_t(-1))
			{


			}
			else
			{
				mylog(log_fatal,"unknown fd,this should never happen\n");
				myexit(-1);
			}
		}

	}*/


	return 0;

}