diff --git a/common.h b/common.h index 8c780fc..30ddb1c 100644 --- a/common.h +++ b/common.h @@ -7,7 +7,7 @@ #ifndef COMMON_H_ #define COMMON_H_ -#define __STDC_FORMAT_MACROS 1 +//#define __STDC_FORMAT_MACROS 1 #include #include @@ -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; diff --git a/connection.cpp b/connection.cpp index e6e8dd8..cd32e36 100644 --- a/connection.cpp +++ b/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::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::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::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 &¤t_time - it->second->last_hb_recv_time <=server_conn_timeout) + else if(it->second->conv_manager.get_size() >0) { - it++; - } - else if(it->second->state.server_current_state!=server_ready&& current_time - it->second->last_state_time <=server_handshake_timeout ) - { - 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; + assert(fd_manager.fd64_exist(fd64)); + int fd=fd_manager.fd64_to_fd(fd64); - ev.events = EPOLLIN; - ev.data.u64 = u64; - - 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); } diff --git a/connection.h b/connection.h index a4dfea7..36ca1e2 100644 --- a/connection.h +++ b/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 udp_fd_mp; //a bit dirty to used pointer,but can void unordered_map search - unordered_map timer_fd_mp;//we can use pointer here since unordered_map.rehash() uses shallow copy - - unordered_map const_id_mp; - - unordered_map mp; //put it at end so that it de-consturcts first + unordered_map fd64_mp; + unordered_map mp;// to conn_info_t; + //put it at end so that it de-consturcts first unordered_map::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::iterator erase_it); int clear_inactive(); diff --git a/main.cpp b/main.cpp index a868a2f..a63deab 100644 --- a/main.cpp +++ b/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)) { - mylog(log_debug,"fd no longer exists in udp_fd_mp,udp fd %d\n",fd); + 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 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 diff --git a/makefile b/makefile index 7737dfe..b7b846b 100755 --- a/makefile +++ b/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 diff --git a/packet.cpp b/packet.cpp index cd12c70..3db5ab1 100644 --- a/packet.cpp +++ b/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); } diff --git a/packet.h b/packet.h index 343541c..f30bfad 100644 --- a/packet.h +++ b/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_ */