1
0

znet.cpp 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694
  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. // char m_timestamp[8];
  176. sock_client(io_context&ic,const char*name,int fd,int recv_time_out,int max_package_size)
  177. :fd_io(ic,fd,EV_READ|EV_WRITE)
  178. ,m_ic(ic)
  179. ,m_name(name)
  180. ,m_recv_timer(ic)
  181. ,m_sync_timer(ic)
  182. {
  183. m_max_package_size=max_package_size;
  184. m_recv_time_out=recv_time_out;
  185. // m_recv_timer.set(ic);
  186. m_recv_timer.set<sock_client,&sock_client::on_recv_timeout>(this);
  187. int recv_timeout_first=config.get("service.recv_timeout_first",10);
  188. m_recv_timer.start(recv_timeout_first,0);
  189. int site_sync=config.get("site_sync",0);//分站时间同步,考虑到双IP双机情况,缺省关闭
  190. int site_sync_freq=config.get("site_sync.freq",10*60);//分站时间同步间隔
  191. m_sync_timer.set<sock_client,&sock_client::on_sync_timeout>(this);
  192. if(site_sync)
  193. {
  194. log_info("启动分站同步定时器:%s,%d",name,site_sync_freq);
  195. struct timeval tv;
  196. gettimeofday(&tv,0);
  197. m_sync_timer.start(2-tv.tv_usec/1000000.+0.01,site_sync_freq);
  198. }
  199. // m_send_timer.set(ic);
  200. // m_send_timer.set(5,0);
  201. // m_send_timer.set<sock_client,&sock_client::on_send_timeout>(this);
  202. // m_send_timer.start();
  203. m_b=(char*)malloc(m_size);
  204. // m_timestamp[0]=0;
  205. }
  206. ~sock_client()
  207. {
  208. free(m_b);
  209. }
  210. #if 0
  211. bool check_timestamp(const char*time)
  212. {
  213. //秒、分、时、天、周、月、年, 脑残的设计
  214. char buf[6];
  215. buf[0]=time[6];
  216. buf[1]=time[5];
  217. buf[2]=time[3];
  218. buf[3]=time[2];
  219. buf[4]=time[1];
  220. buf[5]=time[0];
  221. if(memcmp(m_timestamp,buf,6)<=0)
  222. {
  223. memcpy(m_timestamp,buf,6);
  224. return true;
  225. }
  226. return false;
  227. }
  228. #endif
  229. int type()
  230. {
  231. return m_type;
  232. }
  233. void set(int t)
  234. {
  235. m_type=t;
  236. }
  237. void close()
  238. {
  239. m_close_flag.store(true);
  240. m_ic.async_request(shared_from_this());
  241. }
  242. void send(std::vector<char>&&b)
  243. {
  244. {
  245. std::unique_lock<std::mutex> _lock(m_mutex);
  246. m_olist.push_back(std::move(b));
  247. }
  248. m_ic.async_request(shared_from_this());
  249. }
  250. void on_sync_timeout()
  251. {
  252. // 从第一个字节开始,分别表示毫秒(2字节)、秒、分、时、天、月、年
  253. unsigned char buf[20]={0,13,0x78,0x3b};
  254. struct timeval tv;
  255. gettimeofday(&tv,0);
  256. // tv.tv_sec+=365*86400;
  257. struct tm buff={0};
  258. const struct tm*t=localtime_r(&tv.tv_sec,&buff);
  259. int p=4;
  260. buf[p++]=(tv.tv_usec/1000)&0xFF;
  261. buf[p++]=((tv.tv_usec/1000)>>8)&0xFF;
  262. buf[p++]=t->tm_sec;
  263. buf[p++]=t->tm_min;
  264. buf[p++]=t->tm_hour;
  265. buf[p++]=t->tm_mday;
  266. buf[p++]=t->tm_wday;
  267. buf[p++]=t->tm_mon+1;
  268. buf[p++]=t->tm_year-100;
  269. uint16_t ccrc=do_crc(buf+2,11);
  270. buf[p++]=ccrc>>8;
  271. buf[p++]=ccrc&0xff;
  272. std::vector<char> tmp((char*)buf,(char*)buf+15);
  273. send(std::move(tmp));
  274. 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]);
  275. }
  276. void on_send_timeout()
  277. {
  278. m_ic.on_send_timeout(shared_from_this());
  279. }
  280. void on_recv_timeout()
  281. {
  282. m_ic.on_recv_timeout(shared_from_this());
  283. logn_warn(1,"socket %s recv timeout.",m_name.c_str());
  284. close_impl();
  285. }
  286. std::string name()
  287. {
  288. return m_name;
  289. }
  290. int handle()
  291. {
  292. return m_fd;
  293. }
  294. //void set_site_id(int sid)
  295. //{
  296. // m_site_id=id;
  297. //}
  298. void grow_buf(int len)
  299. {
  300. if(m_size-m_clen>=len)
  301. return;
  302. int size=m_size;
  303. while(m_size-m_clen<len)
  304. size<<=1;
  305. if(size!=m_size)
  306. {
  307. m_b=(char*)realloc(m_b,m_size=size);
  308. }
  309. }
  310. int read_clt()
  311. {
  312. for(;m_clen<m_size;)
  313. {
  314. int rc=zio::read(m_fd,m_b+m_clen,m_size-m_clen);
  315. if(rc>0)
  316. {
  317. m_clen+=rc;
  318. continue;
  319. }
  320. if(rc==-2)
  321. return 0;
  322. if(rc==0)
  323. {
  324. logn_info(1,"socket %d(%s) close by remote",m_fd,m_name.c_str());
  325. }
  326. else if(rc==-1)
  327. {
  328. logn_errno(1,"hava a error on socket %d(%s)",m_fd,m_name.c_str());
  329. }
  330. return -1;
  331. }
  332. return 0;
  333. }
  334. void close_impl()
  335. {
  336. m_recv_timer.stop();
  337. fd_io::stop();
  338. m_ic.on_close(shared_from_this());
  339. logn_info(1,"socket %s closed.",m_name.c_str());
  340. }
  341. size_t calc_length(uint8_t*b)const
  342. {
  343. return (b[0]<<8)|b[1];
  344. }
  345. int io_read()
  346. {
  347. if(read_clt()<0)
  348. return -1;
  349. try
  350. {
  351. int msg_len;
  352. for(;m_clen>=2;)
  353. {
  354. msg_len=calc_length((uint8_t*)m_b)+2;
  355. if(msg_len>m_max_package_size)
  356. {
  357. logn_error(1,"package too big:%d/%d,close socket. site=%s.",msg_len,m_max_package_size,m_name.c_str());
  358. return -1;
  359. }
  360. if(msg_len<=1)
  361. {
  362. logn_error(1,"package too small:%d,close socket. site=%s.",msg_len,m_name.c_str());
  363. return -1;
  364. }
  365. if(m_clen<msg_len)
  366. break;
  367. logn_bin(1,name().c_str(),m_b,msg_len);//输出二进制日志
  368. if(check_crc(m_b,msg_len))
  369. {
  370. on_message(m_b,msg_len);
  371. }
  372. else
  373. {
  374. logn_error(1,"check_crc_error,close socket. site=%s.",m_name.c_str());
  375. return -1;
  376. }
  377. memmove(m_b,&m_b[msg_len],m_clen-msg_len);
  378. m_clen-=msg_len;
  379. }
  380. }
  381. catch(const std::exception&e)
  382. {
  383. logn_error(1,"package error,close socket. site=%s,err_info=%s",m_name.c_str(),e.what());
  384. return -1;
  385. }
  386. return 0;
  387. }
  388. int io_write()
  389. {
  390. if(! m_can_write)
  391. {
  392. set(EV_READ|EV_WRITE);
  393. // m_send_timer.set(5);
  394. // m_send_timer.start();
  395. return 0;
  396. }
  397. for(;;)
  398. {
  399. if(m_obuf.size()==m_opos)
  400. {
  401. std::unique_lock<std::mutex> _lock(m_mutex);
  402. if(m_olist.empty())
  403. break;
  404. m_obuf.swap(m_olist.front());
  405. m_olist.pop_front();
  406. m_opos=0;
  407. }
  408. int left=m_obuf.size()-m_opos;
  409. int rc=zio::write(m_fd,&*m_obuf.begin()+m_opos,left);
  410. if(rc>0)
  411. {
  412. m_opos+=rc;
  413. if(m_opos==m_obuf.size())
  414. continue;
  415. }
  416. else if(rc==0||rc==-2)//缓冲区满
  417. {
  418. m_can_write=false;
  419. set(EV_READ|EV_WRITE);
  420. // m_send_timer.set(5);
  421. // m_send_timer.start();
  422. break;
  423. }
  424. else
  425. {
  426. logn_errno(1,"zio::write(%d,ptr,%d)",m_fd,m_obuf.size()-m_opos);
  427. return -1;
  428. }
  429. }
  430. if(m_olist.empty() && m_can_write)
  431. {
  432. set(EV_READ);
  433. // m_send_timer.stop();
  434. }
  435. return 0;
  436. }
  437. void operator()(ev::io &w, int flag)
  438. {
  439. if(flag & EV_WRITE)
  440. {
  441. logn_debug(1,"socket %d(%s) can write,flag=%d." ,m_fd,m_name.c_str(),flag);
  442. m_can_write=true;
  443. // m_send_timer.stop();
  444. if(io_write()<0)
  445. {
  446. close_impl();
  447. return;
  448. }
  449. }
  450. if(flag & EV_READ)
  451. {
  452. // log_debug("socket %d(%s) can read,flag=%d." ,m_fd,m_name.c_str(),flag);
  453. // zclock c;
  454. if(io_read()<0)
  455. {
  456. close_impl();
  457. return;
  458. }
  459. // log_info("use time %d ms.",c.count_us());
  460. m_recv_timer.start(m_recv_time_out);
  461. }
  462. }
  463. bool check_crc(void *b0,size_t mlen)
  464. {
  465. uint8_t*b=(uint8_t*)b0;
  466. uint16_t crc=b[mlen-2];
  467. crc<<=8;
  468. crc|=b[mlen-1];
  469. uint16_t ccrc=do_crc((unsigned char*)b+2,mlen-4);
  470. return ccrc==crc;
  471. }
  472. void on_message(const char*m,int mlen)
  473. {
  474. m_ic.on_message(shared_from_this(),m,mlen);
  475. }
  476. void on_notify()
  477. {
  478. if(m_close_flag)
  479. {
  480. close_impl();
  481. return;
  482. }
  483. io_write();
  484. }
  485. };
  486. struct sock_listen: fd_io
  487. {
  488. sock_listen(io_context&ic,int fd):fd_io(ic,fd){}
  489. int recv_time_out=config.get("service.recv_timeout",3);
  490. int max_package_size=config.get("service.max_package",2048);
  491. void operator()(ev::io &w, int)
  492. {
  493. char name[32];
  494. int fd=zio::accept(m_fd,name);
  495. if(fd<0)
  496. {
  497. logn_errno(1,"socket %s connect error.",name);
  498. return;
  499. }
  500. zio::setiobuf(fd,32<<10,32<<10);
  501. m_ic.on_connect(std::make_shared<sock_client>(m_ic,name,fd,recv_time_out,max_package_size));
  502. logn_info(1,"socket %s connected.",name);
  503. }
  504. };
  505. struct signal_w:ev::sig
  506. {
  507. io_context&m_ic;
  508. signal_w(io_context&ic, int s)
  509. :ev::sig(ic)
  510. ,m_ic(ic)
  511. {
  512. //this->set(m_ic);
  513. this->set(this);
  514. this->set(s);
  515. this->start();
  516. }
  517. void operator()(ev::sig &w, int s)
  518. {
  519. log_info("recved signal %d",s);
  520. worker::instance()->stop();
  521. stop();
  522. m_ic.stop();
  523. }
  524. };
  525. struct main_loop:io_context
  526. {
  527. main_loop(service_callback&sc)
  528. :io_context(sc)
  529. {
  530. }
  531. virtual int run(int port)
  532. {
  533. int fd=zio::listen_on(port);
  534. if(fd<0)
  535. {
  536. log_errno("try listen_on %d",port);
  537. return -1;
  538. }
  539. sock_listen _1(*this,fd);
  540. // stdin_io _2(*this);
  541. block_sig(SIGPIPE);
  542. signal_w sint(*this,SIGINT),term(*this,SIGTERM);
  543. while(!check_stop_flag())
  544. {
  545. try
  546. {
  547. ev::dynamic_loop::run(0);
  548. }
  549. catch(const std::exception&e)
  550. {
  551. log_error("捕获到异常:%s",e.what());
  552. }
  553. catch(...)
  554. {
  555. log_error("捕获到未知异常");
  556. }
  557. }
  558. _1.stop();
  559. close_all();
  560. return 0;
  561. }
  562. void block_sig(int sig)
  563. {
  564. sigset_t signal_mask;
  565. sigemptyset(&signal_mask);
  566. sigaddset(&signal_mask, sig);
  567. if(pthread_sigmask(SIG_BLOCK, &signal_mask, NULL) == -1)
  568. {
  569. log_errno("block signal:%d",sig);
  570. }
  571. }
  572. };
  573. service_handle*service_handle::instance(service_callback*sc)
  574. {
  575. static main_loop _impl(*sc);
  576. sc->set_handle(&_impl);
  577. return &_impl;
  578. }