tunnel.cpp 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752
  1. /*
  2. * tunnel.cpp
  3. *
  4. * Created on: Oct 26, 2017
  5. * Author: root
  6. */
  7. #include "tunnel.h"
  8. int init_listen_socket()
  9. {
  10. local_listen_fd =socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
  11. int yes = 1;
  12. //setsockopt(udp_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
  13. struct sockaddr_in local_me={0};
  14. socklen_t slen = sizeof(sockaddr_in);
  15. //memset(&local_me, 0, sizeof(local_me));
  16. local_me.sin_family = AF_INET;
  17. local_me.sin_port = htons(local_port);
  18. local_me.sin_addr.s_addr = local_ip_uint32;
  19. if (bind(local_listen_fd, (struct sockaddr*) &local_me, slen) == -1) {
  20. mylog(log_fatal,"socket bind error\n");
  21. //perror("socket bind error");
  22. myexit(1);
  23. }
  24. setnonblocking(local_listen_fd);
  25. set_buf_size(local_listen_fd,socket_buf_size);
  26. mylog(log_debug,"local_listen_fd=%d\n,",local_listen_fd);
  27. return 0;
  28. }
  29. int new_connected_socket(int &fd,u32_t ip,int port)
  30. {
  31. char ip_port[40];
  32. sprintf(ip_port,"%s:%d",my_ntoa(ip),port);
  33. struct sockaddr_in remote_addr_in = { 0 };
  34. socklen_t slen = sizeof(sockaddr_in);
  35. //memset(&remote_addr_in, 0, sizeof(remote_addr_in));
  36. remote_addr_in.sin_family = AF_INET;
  37. remote_addr_in.sin_port = htons(port);
  38. remote_addr_in.sin_addr.s_addr = ip;
  39. fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
  40. if (fd < 0) {
  41. mylog(log_warn, "[%s]create udp_fd error\n", ip_port);
  42. return -1;
  43. }
  44. setnonblocking(fd);
  45. set_buf_size(fd, socket_buf_size);
  46. mylog(log_debug, "[%s]created new udp_fd %d\n", ip_port, fd);
  47. int ret = connect(fd, (struct sockaddr *) &remote_addr_in, slen);
  48. if (ret != 0) {
  49. mylog(log_warn, "[%s]fd connect fail\n",ip_port);
  50. close(fd);
  51. return -1;
  52. }
  53. return 0;
  54. }
  55. int client_event_loop()
  56. {
  57. //char buf[buf_len];
  58. int i, j, k;int ret;
  59. int yes = 1;
  60. int epoll_fd;
  61. int remote_fd;
  62. fd64_t remote_fd64;
  63. conn_info_t *conn_info_p=new conn_info_t;
  64. conn_info_t &conn_info=*conn_info_p; //huge size of conn_info,do not allocate on stack
  65. //conn_info.conv_manager.reserve();
  66. //conn_info.fec_encode_manager.re_init(fec_data_num,fec_redundant_num,fec_mtu,fec_pending_num,fec_pending_time,fec_type);
  67. init_listen_socket();
  68. epoll_fd = epoll_create1(0);
  69. assert(epoll_fd>0);
  70. const int max_events = 4096;
  71. struct epoll_event ev, events[max_events];
  72. if (epoll_fd < 0) {
  73. mylog(log_fatal,"epoll return %d\n", epoll_fd);
  74. myexit(-1);
  75. }
  76. ev.events = EPOLLIN;
  77. ev.data.u64 = local_listen_fd;
  78. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, local_listen_fd, &ev);
  79. if (ret!=0) {
  80. mylog(log_fatal,"add udp_listen_fd error\n");
  81. myexit(-1);
  82. }
  83. assert(new_connected_socket(remote_fd,remote_ip_uint32,remote_port)==0);
  84. remote_fd64=fd_manager.create(remote_fd);
  85. mylog(log_debug,"remote_fd64=%llu\n",remote_fd64);
  86. ev.events = EPOLLIN;
  87. ev.data.u64 = remote_fd64;
  88. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, remote_fd, &ev);
  89. if (ret!= 0) {
  90. mylog(log_fatal,"add raw_fd error\n");
  91. myexit(-1);
  92. }
  93. ev.events = EPOLLIN;
  94. ev.data.u64 = delay_manager.get_timer_fd();
  95. mylog(log_debug,"delay_manager.get_timer_fd()=%d\n",delay_manager.get_timer_fd());
  96. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, delay_manager.get_timer_fd(), &ev);
  97. if (ret!= 0) {
  98. mylog(log_fatal,"add delay_manager.get_timer_fd() error\n");
  99. myexit(-1);
  100. }
  101. u64_t fd64=conn_info.fec_encode_manager.get_timer_fd64();
  102. ev.events = EPOLLIN;
  103. ev.data.u64 = fd64;
  104. mylog(log_debug,"conn_info.fec_encode_manager.get_timer_fd64()=%llu\n",conn_info.fec_encode_manager.get_timer_fd64());
  105. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd_manager.to_fd(fd64), &ev);
  106. if (ret!= 0) {
  107. mylog(log_fatal,"add fec_encode_manager.get_timer_fd64() error\n");
  108. myexit(-1);
  109. }
  110. //my_timer_t timer;
  111. conn_info.timer.add_fd_to_epoll(epoll_fd);
  112. conn_info.timer.set_timer_repeat_us(timer_interval*1000);
  113. mylog(log_debug,"conn_info.timer.get_timer_fd()=%d\n",conn_info.timer.get_timer_fd());
  114. int fifo_fd=-1;
  115. if(fifo_file[0]!=0)
  116. {
  117. fifo_fd=create_fifo(fifo_file);
  118. ev.events = EPOLLIN;
  119. ev.data.u64 = fifo_fd;
  120. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fifo_fd, &ev);
  121. if (ret!= 0) {
  122. mylog(log_fatal,"add fifo_fd to epoll error %s\n",strerror(errno));
  123. myexit(-1);
  124. }
  125. mylog(log_info,"fifo_file=%s\n",fifo_file);
  126. }
  127. while(1)////////////////////////
  128. {
  129. if(about_to_exit) myexit(0);
  130. int nfds = epoll_wait(epoll_fd, events, max_events, 180 * 1000);
  131. if (nfds < 0) { //allow zero
  132. if(errno==EINTR )
  133. {
  134. mylog(log_info,"epoll interrupted by signal\n");
  135. myexit(0);
  136. }
  137. else
  138. {
  139. mylog(log_fatal,"epoll_wait return %d\n", nfds);
  140. myexit(-1);
  141. }
  142. }
  143. int idx;
  144. for (idx = 0; idx < nfds; ++idx) {
  145. if(events[idx].data.u64==(u64_t)conn_info.timer.get_timer_fd())
  146. {
  147. uint64_t value;
  148. read(conn_info.timer.get_timer_fd(), &value, 8);
  149. conn_info.conv_manager.clear_inactive();
  150. mylog(log_trace,"events[idx].data.u64==(u64_t)conn_info.timer.get_timer_fd()\n");
  151. conn_info.stat.report_as_client();
  152. if(debug_force_flush_fec)
  153. {
  154. int out_n;char **out_arr;int *out_len;my_time_t *out_delay;
  155. dest_t dest;
  156. dest.type=type_fd64;
  157. dest.inner.fd64=remote_fd64;
  158. dest.cook=1;
  159. from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
  160. for(int i=0;i<out_n;i++)
  161. {
  162. delay_send(out_delay[i],dest,out_arr[i],out_len[i]);
  163. }
  164. }
  165. }
  166. else if (events[idx].data.u64 == (u64_t)fifo_fd)
  167. {
  168. char buf[buf_len];
  169. int len=read (fifo_fd, buf, sizeof (buf));
  170. if(len<0)
  171. {
  172. mylog(log_warn,"fifo read failed len=%d,errno=%s\n",len,strerror(errno));
  173. continue;
  174. }
  175. buf[len]=0;
  176. handle_command(buf);
  177. }
  178. else if (events[idx].data.u64 == (u64_t)local_listen_fd||events[idx].data.u64 == conn_info.fec_encode_manager.get_timer_fd64())
  179. {
  180. char data[buf_len];
  181. int data_len;
  182. ip_port_t ip_port;
  183. u32_t conv;
  184. int out_n;char **out_arr;int *out_len;my_time_t *out_delay;
  185. dest_t dest;
  186. dest.type=type_fd64;
  187. dest.inner.fd64=remote_fd64;
  188. dest.cook=1;
  189. if(events[idx].data.u64 == conn_info.fec_encode_manager.get_timer_fd64())
  190. {
  191. mylog(log_trace,"events[idx].data.u64 == conn_info.fec_encode_manager.get_timer_fd64()\n");
  192. //mylog(log_info,"timer!!!\n");
  193. uint64_t value;
  194. if(!fd_manager.exist(fd64)) //fd64 has been closed
  195. {
  196. mylog(log_trace,"!fd_manager.exist(fd64)");
  197. continue;
  198. }
  199. if((ret=read(fd_manager.to_fd(fd64), &value, 8))!=8)
  200. {
  201. mylog(log_trace,"(ret=read(fd_manager.to_fd(fd64), &value, 8))!=8,ret=%d\n",ret);
  202. continue;
  203. }
  204. if(value==0)
  205. {
  206. mylog(log_debug,"value==0\n");
  207. continue;
  208. }
  209. assert(value==1);
  210. from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
  211. //from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
  212. }
  213. else//events[idx].data.u64 == (u64_t)local_listen_fd
  214. {
  215. mylog(log_trace,"events[idx].data.u64 == (u64_t)local_listen_fd\n");
  216. struct sockaddr_in udp_new_addr_in={0};
  217. socklen_t udp_new_addr_len = sizeof(sockaddr_in);
  218. if ((data_len = recvfrom(local_listen_fd, data, max_data_len, 0,
  219. (struct sockaddr *) &udp_new_addr_in, &udp_new_addr_len)) == -1) {
  220. mylog(log_error,"recv_from error,this shouldnt happen,err=%s,but we can try to continue\n",strerror(errno));
  221. continue;
  222. //mylog(log_error,"recv_from error,this shouldnt happen at client\n");
  223. //myexit(1);
  224. };
  225. if(!disable_mtu_warn&&data_len>=mtu_warn)
  226. {
  227. 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);
  228. }
  229. mylog(log_trace,"Received packet from %s:%d,len: %d\n", inet_ntoa(udp_new_addr_in.sin_addr),
  230. ntohs(udp_new_addr_in.sin_port),data_len);
  231. ip_port.ip=udp_new_addr_in.sin_addr.s_addr;
  232. ip_port.port=ntohs(udp_new_addr_in.sin_port);
  233. u64_t u64=ip_port.to_u64();
  234. if(!conn_info.conv_manager.is_u64_used(u64))
  235. {
  236. if(conn_info.conv_manager.get_size() >=max_conv_num)
  237. {
  238. mylog(log_warn,"ignored new udp connect bc max_conv_num exceed\n");
  239. continue;
  240. }
  241. conv=conn_info.conv_manager.get_new_conv();
  242. conn_info.conv_manager.insert_conv(conv,u64);
  243. mylog(log_info,"new packet from %s:%d,conv_id=%x\n",inet_ntoa(udp_new_addr_in.sin_addr),ntohs(udp_new_addr_in.sin_port),conv);
  244. }
  245. else
  246. {
  247. conv=conn_info.conv_manager.find_conv_by_u64(u64);
  248. mylog(log_trace,"conv=%d\n",conv);
  249. }
  250. conn_info.conv_manager.update_active_time(conv);
  251. char * new_data;
  252. int new_len;
  253. put_conv(conv,data,data_len,new_data,new_len);
  254. mylog(log_trace,"data_len=%d new_len=%d\n",data_len,new_len);
  255. //dest.conv=conv;
  256. from_normal_to_fec(conn_info,new_data,new_len,out_n,out_arr,out_len,out_delay);
  257. }
  258. mylog(log_trace,"out_n=%d\n",out_n);
  259. for(int i=0;i<out_n;i++)
  260. {
  261. delay_send(out_delay[i],dest,out_arr[i],out_len[i]);
  262. }
  263. //my_send(dest,data,data_len);
  264. }
  265. else if (events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()) {
  266. uint64_t value;
  267. read(delay_manager.get_timer_fd(), &value, 8);
  268. mylog(log_trace,"events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()\n");
  269. //printf("<timerfd_triggered, %d>",delay_mp.size());
  270. //fflush(stdout);
  271. }
  272. else if(events[idx].data.u64>u32_t(-1) )
  273. {
  274. char data[buf_len];
  275. if(!fd_manager.exist(events[idx].data.u64)) //fd64 has been closed
  276. {
  277. mylog(log_trace,"!fd_manager.exist(events[idx].data.u64)");
  278. continue;
  279. }
  280. assert(events[idx].data.u64==remote_fd64);
  281. int fd=fd_manager.to_fd(remote_fd64);
  282. int data_len =recv(fd,data,max_data_len,0);
  283. mylog(log_trace, "received data from udp fd %d, len=%d\n", remote_fd,data_len);
  284. if(data_len<0)
  285. {
  286. if(errno==ECONNREFUSED)
  287. {
  288. //conn_manager.clear_list.push_back(udp_fd);
  289. mylog(log_debug, "recv failed %d ,udp_fd%d,errno:%s\n", data_len,remote_fd,strerror(errno));
  290. }
  291. mylog(log_warn, "recv failed %d ,udp_fd%d,errno:%s\n", data_len,remote_fd,strerror(errno));
  292. continue;
  293. }
  294. if(!disable_mtu_warn&&data_len>mtu_warn)
  295. {
  296. 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);
  297. }
  298. if(de_cook(data,data_len)!=0)
  299. {
  300. mylog(log_debug,"de_cook error");
  301. continue;
  302. }
  303. int out_n;char **out_arr;int *out_len;my_time_t *out_delay;
  304. from_fec_to_normal(conn_info,data,data_len,out_n,out_arr,out_len,out_delay);
  305. mylog(log_trace,"out_n=%d\n",out_n);
  306. for(int i=0;i<out_n;i++)
  307. {
  308. u32_t conv;
  309. char *new_data;
  310. int new_len;
  311. if(get_conv(conv,out_arr[i],out_len[i],new_data,new_len)!=0)
  312. {
  313. mylog(log_debug,"get_conv(conv,out_arr[i],out_len[i],new_data,new_len)!=0");
  314. continue;
  315. }
  316. if(!conn_info.conv_manager.is_conv_used(conv))
  317. {
  318. mylog(log_trace,"!conn_info.conv_manager.is_conv_used(conv)");
  319. }
  320. conn_info.conv_manager.update_active_time(conv);
  321. u64_t u64=conn_info.conv_manager.find_u64_by_conv(conv);
  322. dest_t dest;
  323. dest.inner.ip_port.from_u64(u64);
  324. dest.type=type_ip_port;
  325. //dest.conv=conv;
  326. delay_send(out_delay[i],dest,new_data,new_len);
  327. }
  328. //mylog(log_trace,"[%s] send packet\n",dest.inner.ip_port.to_s());
  329. }
  330. else
  331. {
  332. mylog(log_fatal,"unknown fd,this should never happen\n");
  333. myexit(-1);
  334. }
  335. }
  336. delay_manager.check();
  337. }
  338. return 0;
  339. }
  340. int server_event_loop()
  341. {
  342. //char buf[buf_len];
  343. int i, j, k;int ret;
  344. int yes = 1;
  345. int epoll_fd;
  346. int remote_fd;
  347. // conn_info_t conn_info;
  348. init_listen_socket();
  349. epoll_fd = epoll_create1(0);
  350. assert(epoll_fd>0);
  351. const int max_events = 4096;
  352. struct epoll_event ev, events[max_events];
  353. if (epoll_fd < 0) {
  354. mylog(log_fatal,"epoll return %d\n", epoll_fd);
  355. myexit(-1);
  356. }
  357. ev.events = EPOLLIN;
  358. ev.data.u64 = local_listen_fd;
  359. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, local_listen_fd, &ev);
  360. if (ret!=0) {
  361. mylog(log_fatal,"add udp_listen_fd error\n");
  362. myexit(-1);
  363. }
  364. ev.events = EPOLLIN;
  365. ev.data.u64 = delay_manager.get_timer_fd();
  366. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, delay_manager.get_timer_fd(), &ev);
  367. if (ret!= 0) {
  368. mylog(log_fatal,"add delay_manager.get_timer_fd() error\n");
  369. myexit(-1);
  370. }
  371. mylog(log_debug," delay_manager.get_timer_fd() =%d\n", delay_manager.get_timer_fd());
  372. mylog(log_info,"now listening at %s:%d\n",my_ntoa(local_ip_uint32),local_port);
  373. my_timer_t timer;
  374. timer.add_fd_to_epoll(epoll_fd);
  375. timer.set_timer_repeat_us(timer_interval*1000);
  376. mylog(log_debug," timer.get_timer_fd() =%d\n",timer.get_timer_fd());
  377. int fifo_fd=-1;
  378. if(fifo_file[0]!=0)
  379. {
  380. fifo_fd=create_fifo(fifo_file);
  381. ev.events = EPOLLIN;
  382. ev.data.u64 = fifo_fd;
  383. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fifo_fd, &ev);
  384. if (ret!= 0) {
  385. mylog(log_fatal,"add fifo_fd to epoll error %s\n",strerror(errno));
  386. myexit(-1);
  387. }
  388. mylog(log_info,"fifo_file=%s\n",fifo_file);
  389. }
  390. while(1)////////////////////////
  391. {
  392. if(about_to_exit) myexit(0);
  393. int nfds = epoll_wait(epoll_fd, events, max_events, 180 * 1000);
  394. if (nfds < 0) { //allow zero
  395. if(errno==EINTR )
  396. {
  397. mylog(log_info,"epoll interrupted by signal\n");
  398. myexit(0);
  399. }
  400. else
  401. {
  402. mylog(log_fatal,"epoll_wait return %d\n", nfds);
  403. myexit(-1);
  404. }
  405. }
  406. int idx;
  407. for (idx = 0; idx < nfds; ++idx)
  408. {
  409. /*
  410. if ((events[idx].data.u64 ) == (u64_t)timer_fd)
  411. {
  412. conn_manager.clear_inactive();
  413. u64_t dummy;
  414. read(timer_fd, &dummy, 8);
  415. //current_time_rough=get_current_time();
  416. }
  417. else */
  418. if(events[idx].data.u64==(u64_t)timer.get_timer_fd())
  419. {
  420. uint64_t value;
  421. read(timer.get_timer_fd(), &value, 8);
  422. conn_manager.clear_inactive();
  423. mylog(log_trace,"events[idx].data.u64==(u64_t)timer.get_timer_fd()\n");
  424. //conn_info.conv_manager.clear_inactive();
  425. }
  426. else if (events[idx].data.u64 == (u64_t)fifo_fd)
  427. {
  428. char buf[buf_len];
  429. int len=read (fifo_fd, buf, sizeof (buf));
  430. if(len<0)
  431. {
  432. mylog(log_warn,"fifo read failed len=%d,errno=%s\n",len,strerror(errno));
  433. continue;
  434. }
  435. buf[len]=0;
  436. handle_command(buf);
  437. }
  438. else if (events[idx].data.u64 == (u64_t)local_listen_fd)
  439. {
  440. mylog(log_trace,"events[idx].data.u64 == (u64_t)local_listen_fd\n");
  441. //int recv_len;
  442. char data[buf_len];
  443. int data_len;
  444. struct sockaddr_in udp_new_addr_in={0};
  445. socklen_t udp_new_addr_len = sizeof(sockaddr_in);
  446. if ((data_len = recvfrom(local_listen_fd, data, max_data_len, 0,
  447. (struct sockaddr *) &udp_new_addr_in, &udp_new_addr_len)) == -1) {
  448. mylog(log_error,"recv_from error,this shouldnt happen,err=%s,but we can try to continue\n",strerror(errno));
  449. continue;
  450. //myexit(1);
  451. };
  452. mylog(log_trace,"Received packet from %s:%d,len: %d\n", inet_ntoa(udp_new_addr_in.sin_addr),
  453. ntohs(udp_new_addr_in.sin_port),data_len);
  454. if(!disable_mtu_warn&&data_len>=mtu_warn)///////////////////////delete this for type 0 in furture
  455. {
  456. 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);
  457. }
  458. if(de_cook(data,data_len)!=0)
  459. {
  460. mylog(log_debug,"de_cook error");
  461. continue;
  462. }
  463. ip_port_t ip_port;
  464. ip_port.ip=udp_new_addr_in.sin_addr.s_addr;
  465. ip_port.port=ntohs(udp_new_addr_in.sin_port);
  466. mylog(log_trace,"ip_port= %s\n",ip_port.to_s());
  467. if(!conn_manager.exist(ip_port))
  468. {
  469. if(conn_manager.mp.size() >=max_conn_num)
  470. {
  471. mylog(log_warn,"new connection %s ignored bc max_conn_num exceed\n",ip_port.to_s());
  472. continue;
  473. }
  474. conn_manager.insert(ip_port);
  475. conn_info_t &conn_info=conn_manager.find(ip_port);
  476. //conn_info.fec_encode_manager.re_init(fec_data_num,fec_redundant_num,fec_mtu,fec_pending_num,fec_pending_time,fec_type);
  477. //conn_info.conv_manager.reserve(); //already reserved in constructor
  478. u64_t fec_fd64=conn_info.fec_encode_manager.get_timer_fd64();
  479. mylog(log_debug,"fec_fd64=%llu\n",fec_fd64);
  480. ev.events = EPOLLIN;
  481. ev.data.u64 = fec_fd64;
  482. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd_manager.to_fd(fec_fd64), &ev);
  483. fd_manager.get_info(fec_fd64).ip_port=ip_port;
  484. conn_info.timer.add_fd64_to_epoll(epoll_fd);
  485. conn_info.timer.set_timer_repeat_us(timer_interval*1000);
  486. mylog(log_debug,"conn_info.timer.get_timer_fd64()=%llu\n",conn_info.timer.get_timer_fd64());
  487. u64_t timer_fd64=conn_info.timer.get_timer_fd64();
  488. fd_manager.get_info(timer_fd64).ip_port=ip_port;
  489. mylog(log_info,"new connection from %s\n",ip_port.to_s());
  490. }
  491. conn_info_t &conn_info=conn_manager.find(ip_port);
  492. conn_info.update_active_time();
  493. int out_n;char **out_arr;int *out_len;my_time_t *out_delay;
  494. from_fec_to_normal(conn_info,data,data_len,out_n,out_arr,out_len,out_delay);
  495. mylog(log_trace,"out_n= %d\n",out_n);
  496. for(int i=0;i<out_n;i++)
  497. {
  498. u32_t conv;
  499. char *new_data;
  500. int new_len;
  501. if(get_conv(conv,out_arr[i],out_len[i],new_data,new_len)!=0)
  502. {
  503. mylog(log_debug,"get_conv failed");
  504. continue;
  505. }
  506. /*
  507. id_t tmp_conv_id;
  508. memcpy(&tmp_conv_id,&data_[0],sizeof(tmp_conv_id));
  509. tmp_conv_id=ntohl(tmp_conv_id);*/
  510. if (!conn_info.conv_manager.is_conv_used(conv))
  511. {
  512. if(conn_info.conv_manager.get_size() >=max_conv_num)
  513. {
  514. mylog(log_warn,"ignored new udp connect bc max_conv_num exceed\n");
  515. continue;
  516. }
  517. int new_udp_fd;
  518. ret=new_connected_socket(new_udp_fd,remote_ip_uint32,remote_port);
  519. if (ret != 0) {
  520. mylog(log_warn, "[%s]new_connected_socket failed\n",ip_port.to_s());
  521. continue;
  522. }
  523. fd64_t fd64 = fd_manager.create(new_udp_fd);
  524. ev.events = EPOLLIN;
  525. ev.data.u64 = fd64;
  526. ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_udp_fd, &ev);
  527. conn_info.conv_manager.insert_conv(conv, fd64);
  528. fd_manager.get_info(fd64).ip_port=ip_port;
  529. mylog(log_info,"[%s]new conv %x,fd %d created,fd64=%llu\n",ip_port.to_s(),conv,new_udp_fd,fd64);
  530. //assert(!conn_manager.exist_fd64(fd64));
  531. //conn_manager.insert_fd64(fd64,ip_port);
  532. }
  533. conn_info.conv_manager.update_active_time(conv);
  534. fd64_t fd64= conn_info.conv_manager.find_u64_by_conv(conv);
  535. //int fd=fd_manager.fd64_to_fd(fd64);
  536. dest_t dest;
  537. dest.type=type_fd64;
  538. dest.inner.fd64=fd64;
  539. //dest.conv=conv;
  540. delay_send(out_delay[i],dest,new_data,new_len);
  541. }
  542. }
  543. else if (events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()) {
  544. uint64_t value;
  545. read(delay_manager.get_timer_fd(), &value, 8);
  546. mylog(log_trace,"events[idx].data.u64 == (u64_t)delay_manager.get_timer_fd()\n");
  547. }
  548. else if (events[idx].data.u64 >u32_t(-1))
  549. {
  550. char data[buf_len];
  551. int data_len;
  552. u32_t conv;
  553. fd64_t fd64=events[idx].data.u64;
  554. mylog(log_trace,"events[idx].data.u64 >u32_t(-1),%llu\n",(u64_t)events[idx].data.u64);
  555. if(!fd_manager.exist(fd64)) //fd64 has been closed
  556. {
  557. mylog(log_trace,"!fd_manager.exist(fd64)\n");
  558. continue;
  559. }
  560. assert(fd_manager.exist_info(fd64));
  561. ip_port_t ip_port=fd_manager.get_info(fd64).ip_port;
  562. assert(conn_manager.exist(ip_port));
  563. conn_info_t &conn_info=conn_manager.find(ip_port);
  564. //conn_info.update_active_time(); //cant put it here
  565. int out_n=-2;char **out_arr;int *out_len;my_time_t *out_delay;
  566. dest_t dest;
  567. dest.type=type_ip_port;
  568. //dest.conv=conv;
  569. dest.inner.ip_port=ip_port;
  570. dest.cook=1;
  571. if(fd64==conn_info.fec_encode_manager.get_timer_fd64())
  572. {
  573. //mylog(log_infol,"timer!!!\n");
  574. uint64_t value;
  575. if((ret=read(fd_manager.to_fd(fd64), &value, 8))!=8)
  576. {
  577. mylog(log_trace,"fd_manager.to_fd(fd64), &value, 8)!=8 ,%d\n",ret);
  578. continue;
  579. }
  580. if(value==0)
  581. {
  582. mylog(log_trace,"value==0\n");
  583. continue;
  584. }
  585. assert(value==1);
  586. from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
  587. }
  588. else if(fd64==conn_info.timer.get_timer_fd64())
  589. {
  590. uint64_t value;
  591. read(conn_info.timer.get_timer_fd(), &value, 8);
  592. conn_info.conv_manager.clear_inactive();
  593. if(debug_force_flush_fec)
  594. {
  595. from_normal_to_fec(conn_info,0,0,out_n,out_arr,out_len,out_delay);
  596. }
  597. conn_info.stat.report_as_server(ip_port);
  598. continue;
  599. }
  600. else
  601. {
  602. assert(conn_info.conv_manager.is_u64_used(fd64));
  603. conv=conn_info.conv_manager.find_conv_by_u64(fd64);
  604. conn_info.conv_manager.update_active_time(conv);
  605. conn_info.update_active_time();
  606. int fd=fd_manager.to_fd(fd64);
  607. data_len=recv(fd,data,max_data_len,0);
  608. mylog(log_trace,"received a packet from udp_fd,len:%d,conv=%d\n",data_len,conv);
  609. if(data_len<0)
  610. {
  611. mylog(log_debug,"udp fd,recv_len<0 continue,%s\n",strerror(errno));
  612. continue;
  613. }
  614. if(!disable_mtu_warn&&data_len>=mtu_warn)
  615. {
  616. 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);
  617. }
  618. char * new_data;
  619. int new_len;
  620. put_conv(conv,data,data_len,new_data,new_len);
  621. from_normal_to_fec(conn_info,new_data,new_len,out_n,out_arr,out_len,out_delay);
  622. }
  623. mylog(log_trace,"out_n=%d\n",out_n);
  624. for(int i=0;i<out_n;i++)
  625. {
  626. delay_send(out_delay[i],dest,out_arr[i],out_len[i]);
  627. }
  628. //mylog(log_trace,"[%s] send packet\n",ip_port.to_s());
  629. }
  630. else
  631. {
  632. mylog(log_fatal,"unknown fd,this should never happen\n");
  633. myexit(-1);
  634. }
  635. }
  636. delay_manager.check();
  637. }
  638. return 0;
  639. }