tunnel.cpp 20 KB

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