1
0

znet.cpp 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667
  1. #include <log.h>
  2. #include <unistd.h>
  3. #include <signal.h>
  4. #include <sys/time.h>
  5. #include <stdio.h>
  6. #include <list>
  7. #include <vector>
  8. #include <set>
  9. #include <thread>
  10. #include <mutex>
  11. #include <atomic>
  12. #include <algorithm>
  13. #include <fstream>
  14. #include <time.h>
  15. #include <zio.h>
  16. #include <znet.h>
  17. #include <zloop.h>
  18. #include <clock.h>
  19. #include <worker.h>
  20. #include "config_file.h"
  21. #include "crc.h"
  22. extern config_file config;
  23. struct client_ex:client
  24. {
  25. virtual void on_notify()=0;
  26. virtual void close_impl()=0;
  27. };
  28. struct io_context: zloop<std::shared_ptr<client>> ,service_handle
  29. {
  30. private:
  31. service_callback&m_serv;
  32. ev::timer m_timer;
  33. public:
  34. std::vector<std::shared_ptr<client>> m_thread_clts;
  35. io_context(service_callback&serv)
  36. :m_serv(serv)
  37. ,m_timer(*this)
  38. {
  39. m_thread_clts.reserve(2048);
  40. m_timer.set<io_context,&io_context::on_timer>(this);
  41. m_timer.start(0,1);
  42. }
  43. virtual ~io_context()
  44. {
  45. }
  46. void boardcast(const std::vector<char>&msg)
  47. {
  48. for(const auto&i:m_thread_clts)
  49. {
  50. std::vector<char> tmp(msg);
  51. i->send(std::move(tmp));
  52. }
  53. }
  54. void on_timer()
  55. {
  56. m_serv.on_timer();
  57. }
  58. void on_connect(const std::shared_ptr<client> &clt)
  59. {
  60. if(clt->handle()>=(int)m_thread_clts.size())
  61. {
  62. m_thread_clts.resize(clt->handle()+1);
  63. }
  64. m_thread_clts[clt->handle()]=clt;
  65. m_serv.on_connect(clt);
  66. }
  67. void on_close(const std::shared_ptr<client> &clt)
  68. {
  69. m_serv.on_close(clt);
  70. m_thread_clts[clt->handle()].reset();
  71. }
  72. void on_send_timeout(const std::shared_ptr<client> &clt)
  73. {
  74. m_serv.on_send_timeout(clt);
  75. }
  76. void on_recv_timeout(const std::shared_ptr<client> &clt)
  77. {
  78. m_serv.on_recv_timeout(clt);
  79. }
  80. void on_message(const std::shared_ptr<client>& clt,const char*data,size_t len)
  81. {
  82. m_serv.on_message(clt,data,len);
  83. }
  84. void close_all()
  85. {
  86. for(const auto&clt:m_thread_clts)
  87. {
  88. if(!clt)
  89. continue;
  90. ((client_ex*)clt.get())->close_impl();
  91. }
  92. }
  93. void on_async(const std::list<std::shared_ptr<client>>&notify_clts)
  94. {
  95. for(const auto&clt:notify_clts)
  96. {
  97. ((client_ex*)clt.get())->on_notify();
  98. }
  99. }
  100. void stop()
  101. {
  102. async_stop();
  103. }
  104. };
  105. struct fd_io:ev::io
  106. {
  107. io_context&m_ic;
  108. int m_fd;
  109. fd_io(io_context&ic,int fd,int ev_flag=EV_READ)
  110. :ev::io(ic)
  111. ,m_ic(ic)
  112. ,m_fd(fd)
  113. {
  114. zio::setblocking(fd,false);
  115. this->set(this);
  116. this->set(m_fd,ev_flag);
  117. this->start();
  118. }
  119. void stop () throw () \
  120. {
  121. ev::io::stop();
  122. }
  123. io_context&context()
  124. {
  125. return m_ic;
  126. }
  127. virtual void operator()(ev::io &w, int)=0;
  128. virtual ~fd_io()
  129. {
  130. stop();
  131. zio::close(m_fd);
  132. }
  133. };
  134. struct stdin_io:fd_io
  135. {
  136. stdin_io(io_context&ic)
  137. :fd_io(ic,0)
  138. {}
  139. virtual void operator()(ev::io &w, int)
  140. {
  141. #if 0
  142. char buf[256];
  143. if(!gets(buf))
  144. return;
  145. if(strchr(buf,'X'))
  146. {
  147. log_info("stdin input 'X', exiting...");
  148. worker::instance()->stop();
  149. stop();
  150. m_ic.stop();
  151. }
  152. #endif
  153. }
  154. };
  155. struct sock_client:fd_io,client_ex
  156. {
  157. io_context&m_ic;
  158. std::string m_name;
  159. std::atomic<bool> m_close_flag{false};
  160. int m_type=1;//site
  161. char *m_b{0};
  162. int m_clen{0};
  163. int m_size{1<<16};
  164. int m_max_package_size{2048};
  165. int m_recv_time_out;
  166. ev::timer m_recv_timer;
  167. ev::timer m_sync_timer;
  168. // ev::timer m_send_timer;
  169. std::mutex m_mutex;
  170. std::list<std::vector<char>> m_olist;
  171. std::vector<char> m_obuf;
  172. size_t m_opos=0;
  173. bool m_can_write{false};
  174. int m_site_id{-1};
  175. sock_client(io_context&ic,const char*name,int fd,int recv_time_out,int max_package_size)
  176. :fd_io(ic,fd,EV_READ|EV_WRITE)
  177. ,m_ic(ic)
  178. ,m_name(name)
  179. ,m_recv_timer(ic)
  180. ,m_sync_timer(ic)
  181. {
  182. m_max_package_size=max_package_size;
  183. m_recv_time_out=recv_time_out;
  184. m_recv_timer.set<sock_client,&sock_client::on_recv_timeout>(this);
  185. int recv_timeout_first=config.get("service.recv_timeout_first",10);
  186. m_recv_timer.start(recv_timeout_first,0);
  187. int site_sync=config.get("site_sync",0);//分站时间同步,考虑到双IP双机情况,缺省关闭
  188. int site_sync_freq=config.get("site_sync.freq",10*60);//分站时间同步间隔
  189. m_sync_timer.set<sock_client,&sock_client::on_sync_timeout>(this);
  190. if(site_sync)
  191. {
  192. log_info("启动分站同步定时器:%s,%d",name,site_sync_freq);
  193. struct timeval tv;
  194. gettimeofday(&tv,0);
  195. m_sync_timer.start(2-tv.tv_usec/1000000.+0.01,site_sync_freq);
  196. }
  197. // m_send_timer.set(ic);
  198. // m_send_timer.set(5,0);
  199. // m_send_timer.set<sock_client,&sock_client::on_send_timeout>(this);
  200. // m_send_timer.start();
  201. m_b=(char*)malloc(m_size);
  202. // m_timestamp[0]=0;
  203. }
  204. ~sock_client()
  205. {
  206. free(m_b);
  207. }
  208. #if 0
  209. bool check_timestamp(const char*time)
  210. {
  211. //秒、分、时、天、周、月、年, 脑残的设计
  212. char buf[6];
  213. buf[0]=time[6];
  214. buf[1]=time[5];
  215. buf[2]=time[3];
  216. buf[3]=time[2];
  217. buf[4]=time[1];
  218. buf[5]=time[0];
  219. if(memcmp(m_timestamp,buf,6)<=0)
  220. {
  221. memcpy(m_timestamp,buf,6);
  222. return true;
  223. }
  224. return false;
  225. }
  226. #endif
  227. int type()
  228. {
  229. return m_type;
  230. }
  231. void set_conn_type(int t)
  232. {
  233. m_type=t;
  234. }
  235. void close()
  236. {
  237. m_close_flag.store(true);
  238. m_ic.async_request(shared_from_this());
  239. }
  240. void send(std::vector<char>&&b)
  241. {
  242. {
  243. std::unique_lock<std::mutex> _lock(m_mutex);
  244. m_olist.push_back(std::move(b));
  245. }
  246. m_ic.async_request(shared_from_this());
  247. }
  248. void on_sync_timeout()
  249. {
  250. // 从第一个字节开始,分别表示毫秒(2字节)、秒、分、时、天、月、年
  251. unsigned char buf[20]={0,13,0x78,0x3b};
  252. struct timeval tv;
  253. gettimeofday(&tv,0);
  254. struct tm buff={0};
  255. const struct tm*t=localtime_r(&tv.tv_sec,&buff);
  256. int p=4;
  257. buf[p++]=(tv.tv_usec/1000)&0xFF;
  258. buf[p++]=((tv.tv_usec/1000)>>8)&0xFF;
  259. buf[p++]=t->tm_sec;
  260. buf[p++]=t->tm_min;
  261. buf[p++]=t->tm_hour;
  262. buf[p++]=t->tm_mday;
  263. buf[p++]=t->tm_wday;
  264. buf[p++]=t->tm_mon+1;
  265. buf[p++]=t->tm_year-100;
  266. uint16_t ccrc=do_crc(buf+2,11);
  267. buf[p++]=ccrc>>8;
  268. buf[p++]=ccrc&0xff;
  269. std::vector<char> tmp((char*)buf,(char*)buf+15);
  270. send(std::move(tmp));
  271. logn_info(1,"分站同步:ip=%s,time=%d-%02d-%02d %02d:%02d:%02d.%03d",m_name.c_str(),buf[12]+2000,buf[11],buf[9],buf[8],buf[7],buf[6],buf[5]*256+buf[4]);
  272. }
  273. void on_send_timeout()
  274. {
  275. m_ic.on_send_timeout(shared_from_this());
  276. }
  277. void on_recv_timeout()
  278. {
  279. m_ic.on_recv_timeout(shared_from_this());
  280. logn_warn(1,"socket %s recv timeout.",m_name.c_str());
  281. close_impl();
  282. }
  283. std::string name()
  284. {
  285. return m_name;
  286. }
  287. int handle()
  288. {
  289. return m_fd;
  290. }
  291. void grow_buf(int len)
  292. {
  293. if(m_size-m_clen>=len)
  294. return;
  295. int size=m_size;
  296. while(m_size-m_clen<len)
  297. size<<=1;
  298. if(size!=m_size)
  299. {
  300. m_b=(char*)realloc(m_b,m_size=size);
  301. }
  302. }
  303. int read_clt()
  304. {
  305. for(;m_clen<m_size;)
  306. {
  307. int rc=zio::read(m_fd,m_b+m_clen,m_size-m_clen);
  308. if(rc>0)
  309. {
  310. m_clen+=rc;
  311. continue;
  312. }
  313. if(rc==-2)
  314. return 0;
  315. if(rc==0)
  316. {
  317. logn_info(1,"socket %d(%s) close by remote",m_fd,m_name.c_str());
  318. }
  319. else if(rc==-1)
  320. {
  321. logn_errno(1,"hava a error on socket %d(%s)",m_fd,m_name.c_str());
  322. }
  323. return -1;
  324. }
  325. return 0;
  326. }
  327. void close_impl()
  328. {
  329. m_recv_timer.stop();
  330. fd_io::stop();
  331. logn_info(1,"socket %s closed.",m_name.c_str());
  332. m_ic.on_close(shared_from_this());
  333. }
  334. size_t calc_length(uint8_t*b)const
  335. {
  336. return (b[0]<<8)|b[1];
  337. }
  338. int io_read()
  339. {
  340. if(read_clt()<0)
  341. return -1;
  342. try
  343. {
  344. int msg_len;
  345. for(;m_clen>=2;)
  346. {
  347. msg_len=calc_length((uint8_t*)m_b)+2;
  348. if(msg_len>m_max_package_size)
  349. {
  350. logn_error(1,"package too big:%d/%d,close socket. site=%s.",msg_len,m_max_package_size,m_name.c_str());
  351. return -1;
  352. }
  353. if(msg_len<=2)
  354. {
  355. logn_error(1,"package too small:%d,close socket. site=%s.",msg_len,m_name.c_str());
  356. return -1;
  357. }
  358. if(m_clen<msg_len)
  359. break;
  360. if(check_crc(m_b,msg_len))
  361. {
  362. on_message(m_b,msg_len);
  363. logn_bin(1,name().c_str(),m_b,msg_len);//输出二进制日志
  364. }
  365. else
  366. {
  367. logn_bin(1,name().c_str(),m_b,msg_len);//输出二进制日志
  368. logn_error(1,"check_crc_error,close socket. site=%s.",m_name.c_str());
  369. return -1;
  370. }
  371. memmove(m_b,&m_b[msg_len],m_clen-msg_len);
  372. m_clen-=msg_len;
  373. }
  374. }
  375. catch(const std::exception&e)
  376. {
  377. logn_error(1,"package error,close socket. site=%s,err_info=%s",m_name.c_str(),e.what());
  378. return -1;
  379. }
  380. return 0;
  381. }
  382. int io_write()
  383. {
  384. if(! m_can_write)
  385. {
  386. set(EV_READ|EV_WRITE);
  387. return 0;
  388. }
  389. for(;;)
  390. {
  391. if(m_obuf.size()==m_opos)
  392. {
  393. std::unique_lock<std::mutex> _lock(m_mutex);
  394. if(m_olist.empty())
  395. break;
  396. m_obuf.swap(m_olist.front());
  397. m_olist.pop_front();
  398. m_opos=0;
  399. }
  400. int left=m_obuf.size()-m_opos;
  401. int rc=zio::write(m_fd,&*m_obuf.begin()+m_opos,left);
  402. if(rc>0)
  403. {
  404. m_opos+=rc;
  405. if(m_opos==m_obuf.size())
  406. continue;
  407. }
  408. else if(rc==0||rc==-2)//缓冲区满
  409. {
  410. m_can_write=false;
  411. set(EV_READ|EV_WRITE);
  412. break;
  413. }
  414. else
  415. {
  416. logn_errno(1,"zio::write(%d,ptr,%d)",m_fd,m_obuf.size()-m_opos);
  417. return -1;
  418. }
  419. }
  420. if(m_olist.empty() && m_can_write)
  421. {
  422. set(EV_READ);
  423. }
  424. return 0;
  425. }
  426. void operator()(ev::io &w, int flag)
  427. {
  428. if(flag & EV_WRITE)
  429. {
  430. logn_debug(1,"socket %d(%s) can write,flag=%d." ,m_fd,m_name.c_str(),flag);
  431. m_can_write=true;
  432. if(io_write()<0)
  433. {
  434. close_impl();
  435. return;
  436. }
  437. }
  438. if(flag & EV_READ)
  439. {
  440. if(io_read()<0)
  441. {
  442. close_impl();
  443. return;
  444. }
  445. m_recv_timer.start(m_recv_time_out);
  446. }
  447. }
  448. bool check_crc(void *b0,size_t mlen)
  449. {
  450. uint8_t*b=(uint8_t*)b0;
  451. uint16_t crc=b[mlen-2];
  452. crc<<=8;
  453. crc|=b[mlen-1];
  454. uint16_t ccrc=do_crc((unsigned char*)b+2,mlen-4);
  455. return ccrc==crc;
  456. }
  457. void on_message(const char*m,int mlen)
  458. {
  459. m_ic.on_message(shared_from_this(),m,mlen);
  460. }
  461. void on_notify()
  462. {
  463. if(m_close_flag)
  464. {
  465. close_impl();
  466. return;
  467. }
  468. io_write();
  469. }
  470. };
  471. struct sock_listen: fd_io
  472. {
  473. sock_listen(io_context&ic,int fd):fd_io(ic,fd){}
  474. int recv_time_out=config.get("service.recv_timeout",3);
  475. int max_package_size=config.get("service.max_package",2048);
  476. void operator()(ev::io &w, int)
  477. {
  478. char name[32];
  479. int fd=zio::accept(m_fd,name);
  480. if(fd<0)
  481. {
  482. logn_errno(1,"socket %s connect error.",name);
  483. return;
  484. }
  485. zio::setiobuf(fd,32<<10,32<<10);
  486. m_ic.on_connect(std::make_shared<sock_client>(m_ic,name,fd,recv_time_out,max_package_size));
  487. logn_info(1,"socket %s connected.",name);
  488. }
  489. };
  490. struct signal_w:ev::sig
  491. {
  492. io_context&m_ic;
  493. signal_w(io_context&ic, int s)
  494. :ev::sig(ic)
  495. ,m_ic(ic)
  496. {
  497. this->set(this);
  498. this->set(s);
  499. this->start();
  500. }
  501. void operator()(ev::sig &w, int s)
  502. {
  503. log_info("recved signal %d",s);
  504. worker::instance()->stop();
  505. stop();
  506. m_ic.stop();
  507. }
  508. };
  509. struct main_loop:io_context
  510. {
  511. main_loop(service_callback&sc)
  512. :io_context(sc)
  513. {
  514. }
  515. virtual int run(int port)
  516. {
  517. int fd=zio::listen_on(port);
  518. if(fd<0)
  519. {
  520. log_errno("try listen_on %d",port);
  521. return -1;
  522. }
  523. sock_listen _1(*this,fd);
  524. block_sig(SIGPIPE);
  525. signal_w sint(*this,SIGINT),term(*this,SIGTERM);
  526. while(!check_stop_flag())
  527. {
  528. try
  529. {
  530. ev::dynamic_loop::run(0);
  531. }
  532. catch(const std::exception&e)
  533. {
  534. log_error("捕获到异常:%s",e.what());
  535. }
  536. catch(...)
  537. {
  538. log_error("捕获到未知异常");
  539. }
  540. }
  541. _1.stop();
  542. close_all();
  543. return 0;
  544. }
  545. void block_sig(int sig)
  546. {
  547. sigset_t signal_mask;
  548. sigemptyset(&signal_mask);
  549. sigaddset(&signal_mask, sig);
  550. if(pthread_sigmask(SIG_BLOCK, &signal_mask, NULL) == -1)
  551. {
  552. log_errno("block signal:%d",sig);
  553. }
  554. }
  555. };
  556. service_handle*service_handle::instance(service_callback*sc)
  557. {
  558. static main_loop _impl(*sc);
  559. sc->set_handle(&_impl);
  560. return &_impl;
  561. }