compile success

common.h

@@ -7,7 +7,7 @@
 
 #ifndef COMMON_H_
 #define COMMON_H_
-#define __STDC_FORMAT_MACROS 1
+//#define __STDC_FORMAT_MACROS 1
 #include <inttypes.h>
 
 #include<stdio.h>
@@ -64,9 +64,7 @@ const int max_data_len=1600;
 const int buf_len=max_data_len+200;
 
 const u32_t conv_clear_interval=200;
-const u32_t timer_interval=400;
-const int conv_clear_ratio=40;
-const int conv_clear_min=5;
+//const u32_t timer_interval=400;
 ////const u32_t conv_timeout=180000;
 const u32_t conv_timeout=40000;//for test
 const int max_conv_num=10000;
@@ -80,7 +78,7 @@ const u32_t max_ready_conn_num=1000;
 const u32_t client_handshake_timeout=5000;
 const u32_t client_retry_interval=1000;
 
-const u32_t server_handshake_timeout=10000;// this should be much longer than clients. client retry initially ,server retry passtively
+const u32_t server_handshake_timeout=10000;// this should be much longer than clients. client retry initially ,server retry passtively*/
 
 const int conv_clear_ratio=10;  //conv grabage collecter check 1/10 of all conv one time
 const int conn_clear_ratio=10;
@@ -98,14 +96,14 @@ const u32_t heartbeat_interval=1000;
 const u32_t timer_interval=400;//this should be smaller than heartbeat_interval and retry interval;
 
 //const uint32_t conv_timeout=120000; //120 second
-const u32_t conv_timeout=120000; //for test
+//const u32_t conv_timeout=120000; //for test
 
 const u32_t client_conn_timeout=10000;
 const u32_t client_conn_uplink_timeout=client_conn_timeout+2000;
 
 //const uint32_t server_conn_timeout=conv_timeout+60000;//this should be 60s+ longer than conv_timeout,so that conv_manager can destruct convs gradually,to avoid latency glicth
 const u32_t server_conn_timeout=conv_timeout+60000;//for test
-*/
+
 
 extern int about_to_exit;
 

connection.cpp

@@ -162,25 +162,17 @@ conv_manager_t::~conv_manager_t()
 		}
 		return 0;
 	}
-
-	conn_manager_t::conn_manager_t()
+ conn_manager_t::conn_manager_t()
  {
 	 ready_num=0;
-	 mp.reserve(10007);
+	 mp.reserve(100007);
+	 fd64_mp.reserve(100007);
 	 clear_it=mp.begin();
-	 timer_fd_mp.reserve(10007);
-	 const_id_mp.reserve(10007);
-	 udp_fd_mp.reserve(100007);
 	 last_clear_time=0;
-	 //current_ready_ip=0;
-	// current_ready_port=0;
  }
- int conn_manager_t::exist(u32_t ip,uint16_t port)
+ int conn_manager_t::exist_ip_port(ip_port_t ip_port)
  {
-	 u64_t u64=0;
-	 u64=ip;
-	 u64<<=32u;
-	 u64|=port;
+	 u64_t u64=ip_port.to_u64();
 	 if(mp.find(u64)!=mp.end())
 	 {
 		 return 1;
@@ -197,12 +189,9 @@ conv_manager_t::~conv_manager_t()
 	 mp[u64];
 	 return 0;
  }*/
- conn_info_t *& conn_manager_t::find_insert_p(u32_t ip,uint16_t port)  //be aware,the adress may change after rehash
+ conn_info_t *& conn_manager_t::find_insert_p(ip_port_t ip_port)  //be aware,the adress may change after rehash
  {
-	 u64_t u64=0;
-	 u64=ip;
-	 u64<<=32u;
-	 u64|=port;
+	 u64_t u64=ip_port.to_u64();
 	 unordered_map<u64_t,conn_info_t*>::iterator it=mp.find(u64);
 	 if(it==mp.end())
 	 {
@@ -210,12 +199,9 @@ conv_manager_t::~conv_manager_t()
 	 }
 	 return mp[u64];
  }
- conn_info_t & conn_manager_t::find_insert(u32_t ip,uint16_t port)  //be aware,the adress may change after rehash
+ conn_info_t & conn_manager_t::find_insert(ip_port_t ip_port)  //be aware,the adress may change after rehash
  {
-	 u64_t u64=0;
-	 u64=ip;
-	 u64<<=32u;
-	 u64|=port;
+	 u64_t u64=ip_port.to_u64();
 	 unordered_map<u64_t,conn_info_t*>::iterator it=mp.find(u64);
 	 if(it==mp.end())
 	 {
@@ -223,8 +209,26 @@ conv_manager_t::~conv_manager_t()
 	 }
 	 return *mp[u64];
  }
+ int conn_manager_t::exist_fd64(fd64_t fd64)
+ {
+	 return fd64_mp.find(fd64)!=fd64_mp.end();
+ }
+ void conn_manager_t::insert_fd64(fd64_t fd64,ip_port_t ip_port)
+ {
+	 assert(exist_ip_port(ip_port));
+	 u64_t u64=ip_port.to_u64();
+	 fd64_mp[fd64]=u64;
+ }
+ ip_port_t conn_manager_t::find_by_fd64(fd64_t fd64)
+ {
+	 assert(exist_fd64(fd64));
+	 ip_port_t res;
+	 res.from_u64(fd64_mp[fd64]);
+	 return res;
+ }
  int conn_manager_t::erase(unordered_map<u64_t,conn_info_t*>::iterator erase_it)
  {
+	 /*
 		if(erase_it->second->state.server_current_state==server_ready)
 		{
 			ready_num--;
@@ -241,14 +245,10 @@ conv_manager_t::~conv_manager_t()
 			delete(erase_it->second);
 			mp.erase(erase_it->first);
 		}
-		else
-		{
-			assert(erase_it->second->blob==0);
-			assert(erase_it->second->timer_fd ==0);
-			assert(erase_it->second->oppsite_const_id==0);
-			delete(erase_it->second);
-			mp.erase(erase_it->first);
-		}
+		else*/
+	 ////////todo  close and erase timer_fd ,check fd64 empty
+		delete(erase_it->second);
+		mp.erase(erase_it->first);
 		return 0;
  }
 int conn_manager_t::clear_inactive()
@@ -288,22 +288,13 @@ int conn_manager_t::clear_inactive0()
 			it=mp.begin();
 		}
 
-		if(it->second->state.server_current_state==server_ready &&current_time - it->second->last_hb_recv_time  <=server_conn_timeout)
-		{
-				it++;
-		}
-		else if(it->second->state.server_current_state!=server_ready&& current_time - it->second->last_state_time  <=server_handshake_timeout )
+		else if(it->second->conv_manager.get_size() >0)
 		{
 			it++;
 		}
-		else if(it->second->blob!=0&&it->second->blob->conv_manager.get_size() >0)
-		{
-			assert(it->second->state.server_current_state==server_ready);
-			it++;
-		}
 		else
 		{
-			mylog(log_info,"[%s:%d]inactive conn cleared \n",my_ntoa(it->second->raw_info.recv_info.src_ip),it->second->raw_info.recv_info.src_port);
+			mylog(log_info,"[%s:%d]inactive conn cleared \n",my_ntoa(get_u64_h(it->first)),get_u64_l(it->first));
 			old_it=it;
 			it++;
 			erase(old_it);
@@ -317,29 +308,19 @@ int conn_manager_t::clear_inactive0()
 void server_clear_function(u64_t u64)//used in conv_manager in server mode.for server we have to use one udp fd for one conv(udp connection),
 //so we have to close the fd when conv expires
 {
-	int fd=int(u64);
+	int fd64=u64;
 	int ret;
-	assert(fd!=0);
-	/*
-	epoll_event ev;
-
-	ev.events = EPOLLIN;
-	ev.data.u64 = u64;
+	assert(fd_manager.fd64_exist(fd64));
+	int fd=fd_manager.fd64_to_fd(fd64);
 
-	ret = epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev);
-	if (ret!=0)
-	{
-		mylog(log_fatal,"fd:%d epoll delete failed!!!!\n",fd);
-		myexit(-1);   //this shouldnt happen
-	}*/                //no need
+	fd_manager.remove_fd64(fd64);
 	ret= close(fd);  //closed fd should be auto removed from epoll
-
 	if (ret!=0)
 	{
 		mylog(log_fatal,"close fd %d failed !!!!\n",fd);
 		myexit(-1);  //this shouldnt happen
 	}
 	//mylog(log_fatal,"size:%d !!!!\n",conn_manager.udp_fd_mp.size());
-	assert(conn_manager.udp_fd_mp.find(fd)!=conn_manager.udp_fd_mp.end());
-	conn_manager.udp_fd_mp.erase(fd);
+	assert(conn_manager.fd64_mp.find(fd)!=conn_manager.fd64_mp.end());
+	conn_manager.fd64_mp.erase(fd);
 }

connection.h

@@ -14,9 +14,10 @@ extern int disable_anti_replay;
 #include "common.h"
 #include "log.h"
 #include "delay_manager.h"
+#include "fd_manager.h"
 
 
-
+/*
 struct anti_replay_t  //its for anti replay attack,similar to openvpn/ipsec 's anti replay window
 {
 	u64_t max_packet_received;
@@ -28,7 +29,7 @@ struct anti_replay_t  //its for anti replay attack,similar to openvpn/ipsec 's a
 
 	int is_vaild(u64_t seq);
 };//anti_replay;
-
+*/
 
 struct conv_manager_t  // manage the udp connections
 {
@@ -68,7 +69,9 @@ struct conn_info_t     //stores info for a raw connection.for client ,there is o
 //handle multiple clients
 {
 	conv_manager_t conv_manager;
-	anti_replay_t anti_replay;
+	//anti_replay_t anti_replay;
+	fd64_t timer_fd;
+	ip_port_t ip_port;
 };//g_conn_info;
 
 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
@@ -76,31 +79,22 @@ struct conn_manager_t  //manager for connections. for client,we dont need conn_m
 
  u32_t ready_num;
 
- unordered_map<int,conn_info_t *> udp_fd_mp;  //a bit dirty to used pointer,but can void unordered_map search
- unordered_map<int,conn_info_t *> timer_fd_mp;//we can use pointer here since unordered_map.rehash() uses shallow copy
-
- unordered_map<id_t,conn_info_t *> const_id_mp;
-
- unordered_map<u64_t,conn_info_t*> mp; //put it at end so that it de-consturcts first
+ unordered_map<fd64_t,u64_t> fd64_mp;
+ unordered_map<u64_t,conn_info_t*> mp;//<ip,port> to conn_info_t;
+ 	 	 	 	 	 	 	 	 	  //put it at end so that it de-consturcts first
 
  unordered_map<u64_t,conn_info_t*>::iterator clear_it;
 
  long long last_clear_time;
 
  conn_manager_t();
- int exist(u32_t ip,uint16_t port);
- /*
- int insert(uint32_t ip,uint16_t port)
- {
-	 uint64_t u64=0;
-	 u64=ip;
-	 u64<<=32u;
-	 u64|=port;
-	 mp[u64];
-	 return 0;
- }*/
- conn_info_t *& find_insert_p(u32_t ip,uint16_t port);  //be aware,the adress may change after rehash
- conn_info_t & find_insert(u32_t ip,uint16_t port) ; //be aware,the adress may change after rehash
+ int exist_ip_port(ip_port_t);
+ conn_info_t *& find_insert_p(ip_port_t);  //be aware,the adress may change after rehash
+ conn_info_t & find_insert(ip_port_t) ; //be aware,the adress may change after rehash
+ int exist_fd64(fd64_t fd64);
+ void insert_fd64(fd64_t fd64,ip_port_t);
+ ip_port_t find_by_fd64(fd64_t fd64);
+
 
  int erase(unordered_map<u64_t,conn_info_t*>::iterator erase_it);
 int clear_inactive();

main.cpp

@@ -88,10 +88,10 @@ int new_connected_socket(int &fd,u32_t ip,int port)
 		mylog(log_warn, "[%s]create udp_fd error\n", ip_port);
 		return -1;
 	}
-	setnonblocking(new_udp_fd);
+	setnonblocking(fd);
 	set_buf_size(fd, socket_buf_size);
 
-	mylog(log_debug, "[%s]created new udp_fd %d\n", ip_port, new_udp_fd);
+	mylog(log_debug, "[%s]created new udp_fd %d\n", ip_port, fd);
 	int ret = connect(fd, (struct sockaddr *) &remote_addr_in, slen);
 	if (ret != 0) {
 		mylog(log_warn, "[%s]fd connect fail\n",ip_port);
@@ -185,7 +185,7 @@ int client_event_loop()
 				int new_len;
 				get_conv(conv,data,data_len,new_data,new_len);
 				if(!conn_info.conv_manager.is_conv_used(conv))continue;
-				u64_t u64=conn_info.conv_manager.conv_to_u64(conv);
+				u64_t u64=conn_info.conv_manager.find_conv_by_u64(conv);
 				u32_t ip=get_u64_h(u64);
 				int port=get_u64_l(u64);
 				dest_t dest;
@@ -221,7 +221,7 @@ int client_event_loop()
 
 				if(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 ",recv_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);
 				}
 				mylog(log_trace,"Received packet from %s:%d,len: %d\n", inet_ntoa(udp_new_addr_in.sin_addr),
 						ntohs(udp_new_addr_in.sin_port),data_len);
@@ -343,19 +343,20 @@ int server_event_loop()
 
 				if(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 ",recv_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);
 				}
 				mylog(log_trace,"Received packet from %s:%d,len: %d\n", inet_ntoa(udp_new_addr_in.sin_addr),
 						ntohs(udp_new_addr_in.sin_port),data_len);
 
-				uint32_t ip=udp_new_addr_in.sin_addr.s_addr;
-				int port=udp_new_addr_in.sin_port;
-				if(!conn_manager.exist(ip,port))
+				ip_port_t ip_port;
+				ip_port.ip=udp_new_addr_in.sin_addr.s_addr;
+				ip_port.port=udp_new_addr_in.sin_port;
+				if(!conn_manager.exist_ip_port(ip_port))
 				{
-					conn_info_t &conn_info=conn_manager.find_insert(ip,port);
+					conn_info_t &conn_info=conn_manager.find_insert(ip_port);
 					conn_info.conv_manager.reserve();
 				}
-				conn_info_t &conn_info=conn_manager.find_insert(ip,port);
+				conn_info_t &conn_info=conn_manager.find_insert(ip_port);
 
 				u32_t conv;
 				char *new_data;
@@ -373,7 +374,7 @@ int server_event_loop()
 					new_connected_socket(new_udp_fd,remote_ip_uint32,remote_port);
 
 					if (ret != 0) {
-						mylog(log_warn, "[%s:%d]add udp_fd error\n",my_ntoa(ip),port);
+						mylog(log_warn, "[%s:%d]add udp_fd error\n",my_ntoa(ip_port.ip),ip_port.port);
 						close(new_udp_fd);
 						return -1;
 					}
@@ -384,9 +385,9 @@ int server_event_loop()
 					ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_udp_fd, &ev);
 
 					conn_info.conv_manager.insert_conv(conv, fd64);
-					assert(conn_manager.udp_fd_mp.find(new_udp_fd)==conn_manager.udp_fd_mp.end());
+					assert(!conn_manager.exist_fd64(fd64));
 
-					conn_manager.udp_fd_mp[new_udp_fd] = &conn_info;
+					conn_manager.insert_fd64(fd64,ip_port);
 				}
 				fd64_t fd64= conn_info.conv_manager.find_u64_by_conv(conv);
 				//int fd=fd_manager.fd64_to_fd(fd64);
@@ -434,22 +435,24 @@ int server_event_loop()
 			}*/
 			else if (events[idx].data.u64 >u32_t(-1))
 			{
-				//uint32_t conv_id=events[n].data.u64>>32u;
-
-				int fd=int((events[idx].data.u64<<32u)>>32u);
-
-				if(conn_manager.udp_fd_mp.find(fd)==conn_manager.udp_fd_mp.end()) //this can happen,when fd is a just closed fd
+				char data[buf_len];
+				int data_len;
+				fd64_t fd64=events[idx].data.u64;
+				if(!fd_manager.fd64_exist(fd64))
+				{
+					continue;
+				}
+				int fd=fd_manager.fd64_to_fd(fd64);
+				if(!conn_manager.exist_fd64(fd64)) //this can happen,when fd is a just closed fd
 				{
-					mylog(log_debug,"fd no longer exists in udp_fd_mp,udp fd %d\n",fd);
+					mylog(log_debug,"fd no longer exists in udp_fd_mp,udp fd64 %lld\n",fd64);
 					recv(fd,0,0,0);
 					continue;
 				}
-				conn_info_t* p_conn_info=conn_manager.udp_fd_mp[fd];
+				ip_port_t ip_port=conn_manager.find_by_fd64(fd64);
+				conn_info_t* p_conn_info=conn_manager.find_insert_p(ip_port);
 
-				u64_t u64=conn_manager.udp_fd_mp[fd];
-				u32_t ip=get_u64_h(u64);
-				u32_t port=get_u64_l(u64);
-				if(!conn_manager.exist(ip,port))//TODO remove this for peformance
+				if(!conn_manager.exist_ip_port(ip_port))//TODO remove this for peformance
 				{
 					mylog(log_fatal,"ip port no longer exits 2!!!this shouldnt happen\n");
 					myexit(-1);
@@ -466,20 +469,20 @@ int server_event_loop()
 
 				u32_t conv_id=conn_info.conv_manager.find_conv_by_u64(fd);
 
-				int recv_len=recv(fd,buf,max_data_len,0);
+				data_len=recv(fd,data,max_data_len,0);
 
-				mylog(log_trace,"received a packet from udp_fd,len:%d\n",recv_len);
+				mylog(log_trace,"received a packet from udp_fd,len:%d\n",data_len);
 
-				if(recv_len<0)
+				if(data_len<0)
 				{
 					mylog(log_debug,"udp fd,recv_len<0 continue,%s\n",strerror(errno));
 
 					continue;
 				}
 
-				if(recv_len>=mtu_warn)
+				if(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 ",recv_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);
 				}
 
 				////////todo send data

makefile

@@ -8,7 +8,7 @@ cc_arm= /toolchains/arm-2014.05/bin/arm-none-linux-gnueabi-g++
 #cc_bcm2708=/home/wangyu/raspberry/tools/arm-bcm2708/gcc-linaro-arm-linux-gnueabihf-raspbian/bin/arm-linux-gnueabihf-g++ 
 FLAGS= -std=c++11   -Wall -Wextra -Wno-unused-variable -Wno-unused-parameter -Wno-missing-field-initializers
 
-SOURCES=main.cpp log.cpp common.cpp lib/fec.c lib/rs.c packet.cpp conn_manager.cpp delay_manager.cpp
+SOURCES=main.cpp log.cpp common.cpp lib/fec.c lib/rs.c packet.cpp delay_manager.cpp fd_manager.cpp connection.cpp
 
 NAME=speeder
 TARGETS=amd64 arm mips24kc_be x86  mips24kc_le

packet.cpp

@@ -238,16 +238,22 @@ int my_send(dest_t &dest,char *data,int len)
 	switch(dest.type)
 	{
 		case type_ip_port:
+		{
 			return sendto_ip_port(dest.inner.ip_port.ip,dest.inner.ip_port.port,data,len,0);
 			break;
+		}
 		case type_fd64:
+		{
 			if(!fd_manager.fd64_exist(dest.inner.fd64)) return -1;
 			int fd=fd_manager.fd64_to_fd(dest.inner.fd64);
 			return send_fd(fd,data,len,0);
 			break;
+		}
 		case type_fd:
+		{
 			send_fd(dest.inner.fd,data,len,0);
 			break;
+		}
 		default:
 			assert(0==1);
 	}

packet.h

@@ -31,6 +31,16 @@ struct ip_port_t
 {
 	u32_t ip;
 	int port;
+	void from_u64(u64_t u64)
+	{
+		ip=get_u64_h(u64);
+		port=get_u64_l(u64);
+	}
+	u64_t to_u64()
+	{
+		return pack_u64(ip,port);
+	}
+
 };
 union inner_t
 {
@@ -57,4 +67,7 @@ int de_obscure(const char * input, int in_len,char *output,int &out_len);
 int sendto_ip_port (u32_t ip,int port,char * buf, int len,int flags);
 int send_fd (int fd,char * buf, int len,int flags);
 
+int put_conv(u32_t conv,char * input,int len_in,char *&output,int &len_out);
+int get_conv(u32_t &conv,char *input,int len_in,char *&output,int &len_out );
+
 #endif /* PACKET_H_ */