tun_dev_server.cpp 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362
  1. #include "tun_dev.h"
  2. static dest_t udp_dest;
  3. static dest_t tun_dest;
  4. static void local_listen_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
  5. {
  6. char data[buf_len];
  7. int len;
  8. assert(!(revents&EV_ERROR));
  9. conn_info_t & conn_info= *((conn_info_t*)watcher->data);
  10. int local_listen_fd=watcher->fd;
  11. //struct sockaddr_in udp_new_addr_in={0};
  12. //socklen_t udp_new_addr_len = sizeof(sockaddr_in);
  13. address_t::storage_t udp_new_addr_in={0};
  14. socklen_t udp_new_addr_len = sizeof(address_t::storage_t);
  15. if ((len = recvfrom(local_listen_fd, data, max_data_len+1, 0,
  16. (struct sockaddr *) &udp_new_addr_in, &udp_new_addr_len)) < 0) {
  17. mylog(log_error,"recv_from error,this shouldnt happen,err=%s,but we can try to continue\n",strerror(errno));
  18. return;
  19. //myexit(1);
  20. };
  21. address_t new_addr;
  22. new_addr.from_sockaddr((struct sockaddr *) &udp_new_addr_in,udp_new_addr_len);
  23. if(len==max_data_len+1)
  24. {
  25. mylog(log_warn,"huge packet, data_len > %d,dropped\n",max_data_len);
  26. return;
  27. }
  28. if(de_cook(data,len)<0)
  29. {
  30. mylog(log_warn,"de_cook(data,len)failed \n");
  31. return;
  32. }
  33. char header=0;
  34. if(get_header(header,data,len)!=0)
  35. {
  36. mylog(log_warn,"get_header() failed\n");
  37. return;
  38. }
  39. if(udp_dest.inner.fd_addr.addr==new_addr)
  40. {
  41. if(header==header_keep_alive)
  42. {
  43. mylog(log_trace,"got keep_alive packet\n");
  44. return;
  45. }
  46. if(header!=header_new_connect&& header!=header_normal)
  47. {
  48. mylog(log_warn,"invalid header\n");
  49. return;
  50. }
  51. }
  52. else
  53. {
  54. if(header==header_keep_alive)
  55. {
  56. mylog(log_debug,"got keep_alive packet from unexpected client\n");
  57. return;
  58. }
  59. if(header==header_new_connect)
  60. {
  61. mylog(log_info,"new connection from %s\n", new_addr.get_str());
  62. udp_dest.inner.fd_addr.addr=new_addr;
  63. //udp_dest.inner.fd_ip_port.ip_port.ip=udp_new_addr_in.sin_addr.s_addr;
  64. //udp_dest.inner.fd_ip_port.ip_port.port=ntohs(udp_new_addr_in.sin_port);
  65. conn_info.fec_decode_manager.clear();
  66. conn_info.fec_encode_manager.clear_data();
  67. memset(&conn_info.stat,0,sizeof(conn_info.stat));
  68. }
  69. else if(header==header_normal)
  70. {
  71. mylog(log_debug,"rejected connection from %s\n", new_addr.get_str());
  72. len=1;
  73. data[0]=header_reject;
  74. do_cook(data,len);
  75. dest_t tmp_dest;
  76. tmp_dest.type=type_fd_addr;
  77. tmp_dest.inner.fd_addr.fd=local_listen_fd;
  78. tmp_dest.inner.fd_addr.addr=new_addr;
  79. delay_manager.add(0,tmp_dest,data,len);;
  80. return;
  81. }
  82. else
  83. {
  84. mylog(log_warn,"invalid header\n");
  85. }
  86. }
  87. mylog(log_trace,"Received packet from %s,len: %d\n", new_addr.get_str(),len);
  88. from_fec_to_normal2(conn_info,tun_dest,data,len);
  89. }
  90. static void tun_fd_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
  91. {
  92. char data[buf_len];
  93. int len;
  94. assert(!(revents&EV_ERROR));
  95. conn_info_t & conn_info= *((conn_info_t*)watcher->data);
  96. int tun_fd=watcher->fd;
  97. len=read(tun_fd,data,max_data_len+1);
  98. if(len==max_data_len+1)
  99. {
  100. mylog(log_warn,"huge packet, data_len > %d,dropped\n",max_data_len);
  101. return;
  102. }
  103. if(len<0)
  104. {
  105. mylog(log_warn,"read from tun_fd return %d,errno=%s\n",len,strerror(errno));
  106. return;
  107. }
  108. do_mssfix(data,len);
  109. mylog(log_trace,"Received packet from tun,len: %d\n",len);
  110. if(udp_dest.inner.fd_addr.addr.is_vaild()==0)
  111. {
  112. mylog(log_debug,"received packet from tun,but there is no client yet,dropped packet\n");
  113. return;
  114. }
  115. from_normal_to_fec2(conn_info,udp_dest,data,len,header_normal);
  116. }
  117. static void delay_manager_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
  118. {
  119. assert(!(revents&EV_ERROR));
  120. //do nothing
  121. }
  122. static void fec_encode_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
  123. {
  124. assert(!(revents&EV_ERROR));
  125. mylog(log_trace,"fec_encode_cb() called\n");
  126. conn_info_t & conn_info= *((conn_info_t*)watcher->data);
  127. assert(udp_dest.inner.fd_addr.addr.is_vaild()!=0);
  128. ///mylog(log_trace,"events[idx].data.u64 == conn_info.fec_encode_manager.get_timer_fd64()\n");
  129. ///uint64_t fd64=events[idx].data.u64;
  130. ///uint64_t value;
  131. ///if(!fd_manager.exist(fd64)) //fd64 has been closed
  132. ///{
  133. /// mylog(log_trace,"!fd_manager.exist(fd64)");
  134. /// continue;
  135. ///}
  136. ///if((ret=read(fd_manager.to_fd(fd64), &value, 8))!=8)
  137. ///{
  138. /// mylog(log_trace,"(ret=read(fd_manager.to_fd(fd64), &value, 8))!=8,ret=%d\n",ret);
  139. /// continue;
  140. ///}
  141. ///if(value==0)
  142. ///{
  143. /// mylog(log_debug,"value==0\n");
  144. /// continue;
  145. ///}
  146. ///assert(value==1);
  147. from_normal_to_fec2(conn_info,udp_dest,0,0,header_normal);
  148. }
  149. static void fifo_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
  150. {
  151. assert(!(revents&EV_ERROR));
  152. int fifo_fd=watcher->fd;
  153. char buf[buf_len];
  154. int len=read (fifo_fd, buf, sizeof (buf));
  155. if(len<0)
  156. {
  157. mylog(log_warn,"fifo read failed len=%d,errno=%s\n",len,strerror(errno));
  158. return;
  159. }
  160. buf[len]=0;
  161. handle_command(buf);
  162. }
  163. static void conn_timer_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
  164. {
  165. assert(!(revents&EV_ERROR));
  166. conn_info_t & conn_info= *((conn_info_t*)watcher->data);
  167. mylog(log_trace,"conn_timer_cb() called\n");
  168. //uint64_t value;
  169. //read(conn_info.timer.get_timer_fd(), &value, 8);
  170. if(udp_dest.inner.fd_addr.addr.is_vaild()==0)
  171. {
  172. return;
  173. }
  174. conn_info.stat.report_as_server(udp_dest.inner.fd_addr.addr);
  175. do_keep_alive(udp_dest);
  176. }
  177. static void prepare_cb(struct ev_loop *loop, struct ev_prepare *watcher, int revents)
  178. {
  179. assert(!(revents&EV_ERROR));
  180. delay_manager.check();
  181. }
  182. int tun_dev_server_event_loop()
  183. {
  184. int i,j,k,ret;
  185. int tun_fd;
  186. int local_listen_fd;
  187. conn_info_t *conn_info_p=new conn_info_t;
  188. conn_info_t &conn_info=*conn_info_p; //huge size of conn_info,do not allocate on stack
  189. tun_fd=get_tun_fd(tun_dev);
  190. assert(tun_fd>0);
  191. assert(new_listen_socket2(local_listen_fd,local_addr)==0);
  192. assert(set_tun(tun_dev,htonl((ntohl(sub_net_uint32)&0xFFFFFF00)|1),htonl((ntohl(sub_net_uint32)&0xFFFFFF00 )|2),tun_mtu)==0);
  193. udp_dest.cook=1;
  194. udp_dest.type=type_fd_addr;
  195. udp_dest.inner.fd_addr.fd=local_listen_fd;
  196. udp_dest.inner.fd_addr.addr.clear();
  197. tun_dest.type=type_write_fd;
  198. tun_dest.inner.fd=tun_fd;
  199. ///epoll_fd = epoll_create1(0);
  200. ///assert(epoll_fd>0);
  201. ///const int max_events = 4096;
  202. ///struct epoll_event ev, events[max_events];
  203. ///if (epoll_fd < 0) {
  204. /// mylog(log_fatal,"epoll return %d\n", epoll_fd);
  205. /// myexit(-1);
  206. ///}
  207. struct ev_loop * loop= ev_default_loop(0);
  208. assert(loop != NULL);
  209. conn_info.loop=loop;
  210. ///ev.events = EPOLLIN;
  211. ///ev.data.u64 = local_listen_fd;
  212. ///ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, local_listen_fd, &ev);
  213. ///if (ret!=0) {
  214. /// mylog(log_fatal,"add udp_listen_fd error\n");
  215. /// myexit(-1);
  216. ///}
  217. struct ev_io local_listen_watcher;
  218. local_listen_watcher.data=&conn_info;
  219. ev_io_init(&local_listen_watcher, local_listen_cb, local_listen_fd, EV_READ);
  220. ev_io_start(loop, &local_listen_watcher);
  221. ///ev.events = EPOLLIN;
  222. ///ev.data.u64 = tun_fd;
  223. ///ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, tun_fd, &ev);
  224. ///if (ret!=0) {
  225. /// mylog(log_fatal,"add tun_fd error\n");
  226. /// myexit(-1);
  227. ///}
  228. struct ev_io tun_fd_watcher;
  229. tun_fd_watcher.data=&conn_info;
  230. ev_io_init(&tun_fd_watcher, tun_fd_cb, tun_fd, EV_READ);
  231. ev_io_start(loop, &tun_fd_watcher);
  232. ///ev.events = EPOLLIN;
  233. ///ev.data.u64 = delay_manager.get_timer_fd();
  234. ///mylog(log_debug,"delay_manager.get_timer_fd()=%d\n",delay_manager.get_timer_fd());
  235. ///ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, delay_manager.get_timer_fd(), &ev);
  236. ///if (ret!= 0) {
  237. /// mylog(log_fatal,"add delay_manager.get_timer_fd() error\n");
  238. /// myexit(-1);
  239. ///}
  240. delay_manager.set_loop_and_cb(loop,delay_manager_cb);
  241. ///u64_t tmp_timer_fd64=conn_info.fec_encode_manager.get_timer_fd64();
  242. ///ev.events = EPOLLIN;
  243. ///ev.data.u64 = tmp_timer_fd64;
  244. ///mylog(log_debug,"conn_info.fec_encode_manager.get_timer_fd64()=%llu\n",conn_info.fec_encode_manager.get_timer_fd64());
  245. ///ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd_manager.to_fd(tmp_timer_fd64), &ev);
  246. ///if (ret!= 0) {
  247. /// mylog(log_fatal,"add fec_encode_manager.get_timer_fd64() error\n");
  248. /// myexit(-1);
  249. ///}
  250. conn_info.fec_encode_manager.set_data(&conn_info);
  251. conn_info.fec_encode_manager.set_loop_and_cb(loop,fec_encode_cb);
  252. ///conn_info.timer.add_fd_to_epoll(epoll_fd);
  253. ///conn_info.timer.set_timer_repeat_us(timer_interval*1000);
  254. conn_info.timer.data=&conn_info;
  255. ev_init(&conn_info.timer,conn_timer_cb);
  256. ev_timer_set(&conn_info.timer, 0, timer_interval/1000.0 );
  257. ev_timer_start(loop,&conn_info.timer);
  258. struct ev_io fifo_watcher;
  259. int fifo_fd=-1;
  260. if(fifo_file[0]!=0)
  261. {
  262. fifo_fd=create_fifo(fifo_file);
  263. ev_io_init(&fifo_watcher, fifo_cb, fifo_fd, EV_READ);
  264. ev_io_start(loop, &fifo_watcher);
  265. mylog(log_info,"fifo_file=%s\n",fifo_file);
  266. }
  267. ev_prepare prepare_watcher;
  268. ev_init(&prepare_watcher,prepare_cb);
  269. ev_prepare_start(loop,&prepare_watcher);
  270. mylog(log_info,"now listening at %s\n",local_addr.get_str());
  271. ev_run(loop, 0);
  272. mylog(log_warn,"ev_run returned\n");
  273. myexit(0);
  274. return 0;
  275. }