前言
作为公司的公共产品,经常有这样的需求:就是新建一个本地服务,产品线作为客户端通过 tcp 接入本地服务,来获取想要的业务能力。与印象中动辄处理成千上万连接的 tcp 网络服务不同,这个本地服务是跑在客户机器上的,Win32 上作为开机自启动的 windows 服务运行;Linux 上作为 daemon 在后台运行。总的说来就是用于接收几个产品进程的连接,因此轻量化是其最重要的要求,在这个基础上要能兼顾跨平台就可以了。其实主要就是 windows,再兼顾一点儿 linux。
考察了几个现有的开源网络框架,从 ACE 、boost::asio 到 libevent,都有不尽于人意的地方:
a) ACE:太重,只是想要一个网络框架,结果它扒拉扒拉一堆全提供了,不用还不行;
b) boost::asio:太复杂,牵扯到 boost 库,并且引入了一堆 c++ 模板,需要高版本 c++ 编译器支持;
c) libevent:这个看着不错,还实际用这个做底层封装了一版,结果发版后发现一个比较致命的问题,导致在一些机器上服务启动失败,这个后面我会详细提到。
其它的就更不用说了,之前也粗略看过陈硕的 muddo,总的感觉吧,它是基于其它开源框架不足地方改进的一个库,有相当可取的地方,但是这个改进的方向也主要是解决更大并发、更多连接,不是我的痛点,所以没有继续深入研究。
与其在不同开源框架之间纠结,不如自己动手写一个。我的场景比较固定,不用那么面面俱到,简单罗列一下这个框架需要支持基本的功能:
- 同步写、异步读;
- 可同时监听多路事件,基于 1) 这里只针对异步 READ 事件 (包含连接进入、连接断开),写数据是同步的,因而不需要处理异步 WRITE 事件;
- 要有设置一次性和周期性定时器的能力 (业务决定的);
- 不需要处理信号 (windows 上也没信号这一说,linux 自己搞搞 sigaction 就好啦);
- ……
虽然这个框架未来只会运行在单机上,但我们不希望它一出生就带有性能缺陷,所以性能平平的 select 没能进入法眼,最终决定给它装上最强大的心脏:
Windows 平台: iocp
Linux 平台:epoll
ok,从需求到底层技术路线,貌似都讲清楚了,依照 libevent 我给它取名为 gevent,下面我们从代码级别看下这个框架是怎么简化 tcp 服务搭建这类工作的。
gevent 示例
gevent server
首先看一下服务侧的 sample:
1 #include "EventBase.h" 2 #include "EventHandler.h" 3 4 class GMyEventBase : public GEventBase 5 { 6 public: 7 GEventHandler* create_handler (); 8 }; 9 10 11 class svc_handler : public GJsonEventHandler 12 { 13 public: 14 virtual ~svc_handler () {} 15 virtual void on_read_msg (Json::Value const& val); 16 };
这个服务的核心是 GMyEventBase 类,它使用了框架中的 GEventBase 类,从后者派生而来,只改写了一个 create_handler 接口来提供我们的事件处理对象 svc_handler,它是从框架中的 GEventHandler 派生而来,svc_handler 只改写了一个 on_read_msg 来处理 Json 格式的消息输入。
1 #include <stdio.h> 2 #include "svc_handler.h" 3 #include <signal.h> 4 5 GMyEventBase g_base; 6 GEventHandler* GMyEventBase::create_handler () 7 { 8 return new svc_handler; 9 } 10 11 void sig_int (int signo) 12 { 13 printf ("%d caught\n", signo); 14 g_base.exit (1); 15 printf ("exit ok\n"); 16 } 17 18 int main (int argc, char *argv[]) 19 { 20 if (argc < 2) 21 { 22 printf ("usage: epoll_svc port\n"); 23 return -1; 24 } 25 26 unsigned short port = atoi (argv[1]); 27 28 #ifndef WIN32 29 struct sigaction act; 30 act.sa_handler = sig_int; 31 sigemptyset(&act.sa_mask); 32 act.sa_flags = SA_RESTART; 33 if (sigaction (SIGINT, &act, NULL) < 0) 34 { 35 printf ("install SIGINT failed, errno %d\n", errno); 36 return -1; 37 } 38 else 39 printf ("install SIGINT ok\n"); 40 #endif 41 42 // to test small message block 43 if (g_base.init (/*8, 10*/) < 0) 44 return -1; 45 46 printf ("init ok\n"); 47 do 48 { 49 if (!g_base.listen (port)) 50 { 51 g_base.exit (0); 52 printf ("exit ok\n"); 53 break; 54 } 55 56 printf ("listen ok\n"); 57 g_base.run (); 58 printf ("run over\n"); 59 } while (0); 60 61 g_base.fini (); 62 printf ("fini ok\n"); 63 64 g_base.cleanup (); 65 printf ("cleanup ok\n"); 66 return 0; 67 }
程序的运行就是分别调用 GMyEventBase(实际上是GEventBase) 的 init / listen / run / fini / cleaup 方法。而与业务相关的代码,都在 svc_handler 中处理:
1 #include "svc_handler.h" 2 3 void svc_handler::on_read_msg (Json::Value const& val) 4 { 5 int key = val["key"].asInt (); 6 std::string data = val["data"].asString (); 7 printf ("got %d:%s\n", key, data.c_str ()); 8 9 Json::Value root; 10 Json::FastWriter writer; 11 root["key"] = key + 1; 12 root["data"] = data; 13 14 int ret = 0; 15 std::string resp = writer.write(root); 16 resp = resp.substr (0, resp.length () - 1); // trim tailing \n 17 if ((ret = send (resp)) <= 0) 18 printf ("send response failed, errno %d\n", errno); 19 else 20 printf ("response %d\n", ret); 21 }
它期待 Json 格式的数据,并且有两个字段 key(int) 与 data (string),接收数据后将 key 增 1 后返回给客户端。
gevent client
再来看下客户端 sample:
1 #include "EventBaseAR.h" 2 #include "EventHandler.h" 3 4 class GMyEventBase : public GEventBaseWithAutoReconnect 5 { 6 public: 7 GEventHandler* create_handler (); 8 }; 9 10 11 class clt_handler : public GJsonEventHandler 12 { 13 public: 14 virtual ~clt_handler () {} 15 #ifdef TEST_TIMER 16 virtual bool on_timeout (GEV_PER_TIMER_DATA *gptd); 17 #endif 18 virtual void on_read_msg (Json::Value const& val); 19 };
客户端同样使用了 GEventBase 的派生类 GMyEventBase 来作为事件循环的核心,所不同的是 (注意并非之前例子里的那个类,虽然同名),它提供了 clt_handler 来处理自己的业务代码。另外为了提供连接中断后自动向服务重连的功能,这里 GMyEventBase 派生自 GEventBase 类的子类 GEventBaseWithAutoReconnect (位于 EventBaseAR.h/cpp 中)。
1 #include <stdio.h> 2 #include "clt_handler.h" 3 #include <signal.h> 4 5 //#define TEST_READ 6 //#define TEST_CONN 7 //#define TEST_TIMER 8 9 GMyEventBase g_base; 10 GEventHandler* GMyEventBase::create_handler () 11 { 12 return new clt_handler; 13 } 14 15 16 int sig_caught = 0; 17 void sig_int (int signo) 18 { 19 sig_caught = 1; 20 printf ("%d caught\n", signo); 21 g_base.exit (0); 22 printf ("exit ok\n"); 23 } 24 25 void do_read (GEventHandler *eh, int total) 26 { 27 char buf[1024] = { 0 }; 28 int ret = 0, n = 0, key = 0, err = 0; 29 char *ptr = nullptr; 30 while ((total == 0 || n++ < total) && fgets (buf, sizeof(buf), stdin) != NULL) 31 { 32 // skip \n 33 buf[strlen(buf) - 1] = 0; 34 //n = sscanf (buf, "%d", &key); 35 key = strtol (buf, &ptr, 10); 36 if (ptr == nullptr) 37 { 38 printf ("format: int string\n"); 39 continue; 40 } 41 42 Json::Value root; 43 Json::FastWriter writer; 44 root["key"] = key; 45 // skip space internal 46 root["data"] = *ptr == ' ' ? ptr + 1 : ptr; 47 48 std::string req = writer.write (root); 49 req = req.substr (0, req.length () - 1); // trim tailing \n 50 if ((ret = eh->send (req)) <= 0) 51 { 52 err = 1; 53 printf ("send %d failed, errno %d\n", req.length (), errno); 54 break; 55 } 56 else 57 printf ("send %d\n", ret); 58 } 59 60 if (total == 0) 61 printf ("reach end\n"); 62 63 if (!err) 64 { 65 eh->disconnect (); 66 printf ("call disconnect to notify server\n"); 67 } 68 69 // wait receiving thread 70 //sleep (3); 71 // if use press Ctrl+D, need to notify peer our break 72 } 73 74 #ifdef TEST_TIMER 75 void test_timer (unsigned short port, int period_msec, int times) 76 { 77 int n = 0; 78 GEventHandler *eh = nullptr; 79 80 do 81 { 82 eh = g_base.connect (port); 83 if (eh == nullptr) 84 break; 85 86 printf ("connect ok\n"); 87 void* t = g_base.timeout (1000, period_msec, eh, NULL); 88 if (t == NULL) 89 { 90 printf ("timeout failed\n"); 91 break; 92 } 93 else 94 printf ("set timer %p ok\n", t); 95 96 // to wait timer 97 do 98 { 99 sleep (400); 100 printf ("wake up from sleep\n"); 101 } while (!sig_caught && n++ < times); 102 103 g_base.cancel_timer (t); 104 } while (0); 105 } 106 #endif 107 108 #ifdef TEST_CONN 109 void test_conn (unsigned short port, int per_read, int times) 110 { 111 # ifdef WIN32 112 srand (GetCurrentProcessId()); 113 # else 114 srand (getpid ()); 115 # endif 116 int n = 0, elapse = 0; 117 clt_handler *eh = nullptr; 118 119 do 120 { 121 eh = (clt_handler *)g_base.connect (port); 122 if (eh == nullptr) 123 break; 124 125 printf ("connect ok\n"); 126 127 do_read (eh, per_read); 128 # ifdef WIN32 129 elapse = rand() % 1000; 130 Sleep(elapse); 131 printf ("running %d ms\n", elapse); 132 # else 133 elapse = rand () % 1000000; 134 usleep (elapse); 135 printf ("running %.3f ms\n", elapse/1000.0); 136 # endif 137 138 } while (!sig_caught && n++ < times); 139 } 140 #endif 141 142 #ifdef TEST_READ 143 void test_read (unsigned short port, int total) 144 { 145 int n = 0; 146 GEventHandler *eh = nullptr; 147 148 do 149 { 150 eh = g_base.connect (port); 151 if (eh == nullptr) 152 break; 153 154 printf ("connect ok\n"); 155 do_read (eh, total); 156 } while (0); 157 } 158 #endif 159 160 int main (int argc, char *argv[]) 161 { 162 if (argc < 2) 163 { 164 printf ("usage: epoll_clt port\n"); 165 return -1; 166 } 167 168 unsigned short port = atoi (argv[1]); 169 170 #ifndef WIN32 171 struct sigaction act; 172 act.sa_handler = sig_int; 173 sigemptyset(&act.sa_mask); 174 // to ensure read be breaked by SIGINT 175 act.sa_flags = 0; //SA_RESTART; 176 if (sigaction (SIGINT, &act, NULL) < 0) 177 { 178 printf ("install SIGINT failed, errno %d\n", errno); 179 return -1; 180 } 181 #endif 182 183 if (g_base.init (2) < 0) 184 return -1; 185 186 printf ("init ok\n"); 187 188 #if defined(TEST_READ) 189 test_read (port, 0); // 0 means infinite loop until user break 190 #elif defined(TEST_CONN) 191 test_conn (port, 10, 100); 192 #elif defined (TEST_TIMER) 193 test_timer (port, 10, 1000); 194 #else 195 # error please define TEST_XXX macro to do something! 196 #endif 197 198 if (!sig_caught) 199 { 200 // Ctrl + D ? 201 g_base.exit (0); 202 printf ("exit ok\n"); 203 } 204 else 205 printf ("has caught Ctrl+C\n"); 206 207 g_base.fini (); 208 printf ("fini ok\n"); 209 210 g_base.cleanup (); 211 printf ("cleanup ok\n"); 212 return 0; 213 }
程序的运行是分别调用 GEventBase 的 init / connect / fini / cleaup 方法以及 GEventHandler 的 send / disconnect 来测试读写与连接。
- 定义宏 TEST_READ 用来测试读写;
- 定义宏 TEST_CONN 可以测试连接的通断及读写;
- 定义宏 TEST_TIMER 来测试周期性定时器及读写。
注意它们是互斥的。clt_handler 主要用来异步接收服务端的回送数据并打印:
1 #include "clt_handler.h" 2 3 #ifdef TEST_TIMER 4 extern void do_read (clt_handler *, int); 5 bool clt_handler::on_timeout (GEV_PER_TIMER_DATA *gptd) 6 { 7 printf ("time out ! id %p, due %d, period %d\n", gptd, gptd->due_msec, gptd->period_msec); 8 do_read ((clt_handler *)gptd->user_arg, 1); 9 return true; 10 } 11 #endif 12 13 void clt_handler::on_read_msg (Json::Value const& val) 14 { 15 int key = val["key"].asInt (); 16 std::string data = val["data"].asString (); 17 printf ("got %d:%s\n", key, data.c_str ()); 18 }
集成测试
这个测试程序可以通过在控制台手工输入数据来驱动,也可以通过测试数据文件来驱动,下面的 awk 脚本用来制造符合格式的测试数据:
1 #! /bin/awk -f 2 BEGIN { 3 WORDNUM = 1000 4 for (i = 1; i <= WORDNUM; i++) { 5 printf("%d %s\n", randint(WORDNUM), randword(20)) 6 } 7 } 8 9 # randint(n): return a random integer number which is >= 1 and <= n 10 function randint(n) { 11 return int(n *rand()) + 1 12 } 13 14 # randlet(): return a random letter, which maybe upper, lower or number. 15 function randlet() { 16 return substr("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", randint(62), 1) 17 } 18 19 # randword(LEN): return a rand word with a length of LEN 20 function randword(LEN) { 21 randw="" 22 for( j = 1; j <= LEN; j++) { 23 randw=randw randlet() 24 } 25 return randw 26 }
生成的测试文件格式如下:
238 s0jKlYkEjwE4q3nNJugF 568 0cgNaSgDpP3VS45x3Wum 996 kRF6SgmIReFmrNBcCecj 398 QHQqCrB5fC61hao1BV2x 945 XZ6KLtA4jZTEnhcAugAM 619 WE95NU7FnsYar4wz279j 549 oVCTmD516yvmtuJB2NG3 840 NDAaL5vpzp8DQX0rLRiV 378 jONIm64AN6UVc7uTLIIR 251 EqSBOhc40pKXhCbCu8Ey
整个工程编译的话就是一个 CMakeLists 文件,可以通过 cmake 生成对应的 Makefile 或 VS solution 来编译代码:
1 cmake_minimum_required(VERSION 3.0) 2 project(epoll_svc) 3 include_directories(../core ../include) 4 set(CMAKE_CXX_FLAGS "-std=c++11 -pthread -g -Wall ${CMAKE_CXX_FLAGS}") 5 link_directories(${PROJECT_SOURCE_DIR}/../lib) 6 set(EXECUTABLE_OUTPUT_PATH ${PROJECT_SOURCE_DIR}/../bin) 7 8 add_executable (epoll_svc epoll_svc.cpp svc_handler.cpp ../core/EventBase.cpp ../core/EventHandler.cpp ../core/log.cpp) 9 IF (WIN32) 10 target_link_libraries(epoll_svc jsoncpp ws2_32) 11 ELSE () 12 target_link_libraries(epoll_svc jsoncpp rt) 13 ENDIF () 14 15 add_executable (epoll_clt epoll_clt.cpp clt_handler.cpp ../core/EventBase.cpp ../core/EventBaseAR.cpp ../core/EventHandler.cpp ../core/log.cpp) 16 target_compile_definitions(epoll_clt PUBLIC -D TEST_READ) 17 IF (WIN32) 18 target_link_libraries(epoll_clt jsoncpp ws2_32) 19 ELSE () 20 target_link_libraries(epoll_clt jsoncpp rt) 21 ENDIF () 22 23 add_executable (epoll_local epoll_local.cpp) 24 IF (WIN32) 25 target_link_libraries(epoll_local jsoncpp ws2_32) 26 ELSE () 27 target_link_libraries(epoll_local jsoncpp rt) 28 ENDIF ()
这个项目包含三个编译目标,分别是 epoll_svc 、epoll_clt 与 epoll_local,其中前两个可以跨平台编译,后一个只能在 Linux 平台编译,用来验证 epoll 的一些特性。编译完成后,首先运行服务端:
>./epoll_svc 1025
然后运行客户端:
>./epoll_clt 1025 < demo
测试多个客户端同时连接,可以使用下面的脚本:
1 #! /bin/bash 2 # /bin/sh -> /bin/dash, do not recognize our for loop 3 4 for((i=0;i<10;i=i+1)) 5 do 6 ./epoll_clt 1025 < demo & 7 echo "start $i" 8 done
可以同时启动 10 个客户端。通过 Ctrl+C 退出服务端;通过 Ctrl+C 或 Ctrl+D 退出单个客户端;通过下面的脚本来同时停止多个客户端与服务端:
1 #! /bin/sh 2 pkill -INT epoll_clt 3 sleep 1 4 pkill -INT epoll_svc
gevent 接口说明
框架的用法介绍完之后,再简单游览一下这个库的各层级对外接口。
GEventBase
先来看事件引擎的接口:
1 class IEventBase 2 { 3 public: 4 #ifdef WIN32 5 virtual HANDLE iocp () const = 0; 6 #else 7 virtual int epfd () const = 0; 8 #endif 9 10 virtual void* timeout(int due_msec, int period_msec, void *arg, GEventHandler *exist_handler) = 0; 11 virtual bool cancel_timer(void* tid) = 0; 12 virtual bool post_timer(GEV_PER_TIMER_DATA *gptd) = 0; 13 }; 14 15 struct GEV_PER_HANDLE_DATA 16 { 17 SOCKET so; 18 SOCKADDR_IN laddr; 19 SOCKADDR_IN raddr; 20 21 GEV_PER_HANDLE_DATA(SOCKET s, SOCKADDR_IN *l, SOCKADDR_IN *r); 22 virtual ~GEV_PER_HANDLE_DATA(); 23 }; 24 25 struct GEV_PER_IO_DATA 26 { 27 SOCKET so; 28 #ifdef WIN32 29 GEV_IOCP_OP op; 30 OVERLAPPED ol; 31 WSABUF wsa; // wsa.len is buffer length 32 DWORD bytes; // after compeleted, bytes trasnfered 33 #else 34 char *buf; 35 int len; 36 #endif 37 38 GEV_PER_IO_DATA( 39 #ifdef WIN32 40 GEV_IOCP_OP o, 41 #endif 42 SOCKET s, int l); 43 virtual ~GEV_PER_IO_DATA(); 44 }; 45 46 struct GEV_PER_TIMER_DATA 47 #ifdef WIN32 48 : public GEV_PER_IO_DATA 49 #endif 50 { 51 IEventBase *base; 52 int due_msec; 53 int period_msec; 54 void *user_arg; 55 bool cancelled; 56 #ifdef WIN32 57 HANDLE timerque; 58 HANDLE timer; 59 #else 60 timer_t timer; 61 #endif 62 63 GEV_PER_TIMER_DATA(IEventBase *base, int due, int period, void *arg 64 #ifdef WIN32 65 , HANDLE tq); 66 #else 67 , timer_t tid); 68 #endif 69 70 virtual ~GEV_PER_TIMER_DATA(); 71 void cancel (); 72 }; 73 74 75 class GEventBase : public IEventBase 76 { 77 public: 78 GEventBase(); 79 ~GEventBase(); 80 81 #ifdef WIN32 82 virtual HANDLE iocp () const; 83 #else 84 virtual int epfd () const; 85 #endif 86 87 virtual bool post_timer(GEV_PER_TIMER_DATA *gptd); 88 virtual GEventHandler* create_handler() = 0; 89 90 // thr_num : 91 // =0 - no default thread pool, user provide thread and call run 92 // <0 - use max(|thr_num|, processer_num) 93 // >0 - use thr_num 94 bool init(int thr_num = -8, int blksize = GEV_MAX_BUF_SIZE 95 #ifndef WIN32 96 , int timer_sig = SIGUSR1 97 #endif 98 ); 99 100 bool listen(unsigned short port, unsigned short backup = 10); 101 GEventHandler* connect(unsigned short port, char const* host = "127.0.0.1", GEventHandler* exist_handler = NULL); 102 // PARAM 103 // due_msec: first timeout milliseconds 104 // period_msec: later periodically milliseconds 105 // arg: user provied argument 106 // exist_handler: reuse the timer handler 107 // 108 // RETURN 109 // NULL: failed 110 void* timeout(int due_msec, int period_msec, void *arg, GEventHandler *exist_handler); 111 bool cancel_timer(void* tid); 112 void fini(); 113 void run(); 114 void exit(int extra_notify = 0); 115 void cleanup(); 116 117 void disconnect(); 118 int broadcast(std::string const& msg); 119 int foreach(std::function<int(GEventHandler *h, void *arg)> func, void *arg); 120 121 protected: 122 virtual bool on_accept(GEV_PER_HANDLE_DATA *gphd); 123 virtual bool on_read(GEventHandler *h, GEV_PER_IO_DATA *gpid); 124 virtual void on_error(GEventHandler *h); 125 virtual bool on_timeout (GEV_PER_TIMER_DATA *gptd); 126 // whether this handler should be processed in foreach, 127 // true - process; false - skip 128 virtual bool filter_handler(GEventHandler *h); 129 130 …… 131 };
出于突出接口的目的,一些与实现相关的细节这里略去了。下面对 GEventBase 的主要接口做个简单说明:
- init,它在底层启动 thr_num 个线程来跑 run 方法;每次 IO 的块缓冲区大小由 blksize 指定;它内部还创建了对应的 iocp 或 epoll 对象,便于之后加入 socket 句柄进行处理;在 unix like 平台上支持信号方式触发的定时器 (timer_sig),时间间隔到达后,通过发送信号来通知调用者。
- exit,它通知线程池中的所有线程退出等待,windows 上是通过 PostQueuedCompletionStatus,Linux 上是通过在自建的一个 pipe 上写数据以触发 epoll 退出 (这个 pipe 在 init 中创建并加入 epoll);
- fini,它在所有工作线程退出后,关闭之前创建的对象,清理事件循环用到的资源;
- cleanup,它清理之前建立的 fd-handler 映射,清理遗留的处理器并释放资源;
- run,它是线程池运行函数,windows 上是通过 GetQueuedCompletionStatus 在 iocp 上等待;在 linux 上是通过 epoll_wait 在 epoll 上等待事件。当有事件产生后,根据事件类型,分别调用 do_accept / on_accept、do_recv / on_read、do_error / on_error 回调来分派事件;
- listen,创建侦听 socket 并加入到 iocp 或 epoll 中;
- connect,连接到远程服务并将成功连接的 socket 加入到 iocp 或 epoll 中;
- timeout,设置定时器事件,windows 上是通过 CreateTimerQueueTimer 实现定时器超时;linux 则是通过 timer_create 实现的,都是系统现成的东西,在系统定时器到期后,再给对应的 iocp 或 epoll 对象发送一个通知,在 linux 上这个通知机制是上面提到过的 pipe 来实现的,因而有一定延迟,不能指定精度太小的定时器;
- cancel_timer,取消之前设置的定时器。
GEventHanlder
然后看下事件处理器提供的回调接口,应用可以从它派生来完成业务相关代码:
1 class GEventHandler 2 { 3 public: 4 GEventHandler(); 5 virtual ~GEventHandler(); 6 7 GEV_PER_HANDLE_DATA* gphd(); 8 GEV_PER_TIMER_DATA* gptd(); 9 bool connected(); 10 void disconnect(); 11 void clear(); 12 SOCKET fd(); 13 14 int send_raw(char const* buf, int len); 15 int send_raw(std::string const& str); 16 int send(char const* buf, int len); 17 int send(std::string const& str); 18 19 virtual bool reuse(); 20 virtual bool auto_reconnect(); 21 virtual void arg(void *param) = 0; 22 virtual void reset(GEV_PER_HANDLE_DATA *gphd, GEV_PER_TIMER_DATA *gptd, IEventBase *base); 23 virtual bool on_read(GEV_PER_IO_DATA *gpid) = 0; 24 virtual void on_error(GEV_PER_HANDLE_DATA *gphd); 25 // note when on_timeout called, handler's base may cleared by cancel_timer, use gptd->base instead if it is not null. 26 virtual bool on_timeout(GEV_PER_TIMER_DATA *gptd) = 0; 27 virtual void cleanup(bool terminal); 28 void close(bool terminal); 29 30 protected: 31 // think like this: 32 // on_read (pre_read (msg)); 33 // send (pre_write (msg)); 34 // 35 // to give user a chance to modify input before send/handle msg. 36 virtual bool has_preread() const; 37 virtual bool has_prewrite() const; 38 virtual std::string pre_read (char const* buf, int len); 39 virtual std::string pre_write (char const* buf, int len); 40 41 …… 42 };
除了在连接上发送数据 (send 函数),还有其它一些重要的回调接口,列明如下:
- on_read,连接上有数据到达;
- on_error,连接断开;
- on_tmeout,定时器事件;
- ……
如果有新的事件需要处理 ,也可以在这个类里扩展。GEventHandler 处理基于流的数据,它并不关心底层消息的格式,具体是二进制、文本,还是 json / xml / protobuf … 通通留给调用者实现,保证通信框架与数据格式的解耦。
GJsonEventHandler
1 // a common handler to process json protocol. 2 class GJsonEventHandler : public GEventHandler 3 { 4 public: 5 //virtual void on_read(); 6 virtual void arg(void *param); 7 virtual void reset(GEV_PER_HANDLE_DATA *gphd, GEV_PER_TIMER_DATA *gptd, IEventBase *base); 8 virtual bool on_read(GEV_PER_IO_DATA *gpid); 9 virtual void on_read_msg(Json::Value const& root) = 0; 10 virtual bool on_timeout(GEV_PER_TIMER_DATA *gptd); 11 virtual void cleanup(bool terminal); 12 13 …… 14 };
为了演示的便利性,这里基于 GEventHandler 派生了专门处理 json 格式数据的处理器 GJsonEventHandler。与 GEventHandler 不同的是,前者需要重写 on_read 方法来处理流数据;后者需要重写 on_read_msg 方法来处理 json 消息。目前 json 的解析是通过 jsoncpp 库完成的,这个库本身是开源跨平台的 ,不过这里的仅提供 64 位 Linux 静态链接库及 windows 32 位 Release 版本静态库,其余平台需要用户自己编译。
在前面的 sample 中,svc_handler 与 clt_handler 均从 GJsonEventHandler 派生。如果要处理新的数据格式 ,只需要从 GEventHandler 类派生新的处理类即可,记得有次为了在项目中实现 websocket 协议,就从 GEventHandler 派生了一个 GWSEventHandler 来专门处理 websocket 的协议层。
GEventBaseWithAutoReconnect
在实际工程中,连接中断是经常发生的事,如果没有自动重连机制是不可想象的。下面看看带自动重连机制的事件引擎接口,它可以在检测到连接断开时,自动尝试重新建立连接:
1 class GEventBaseWithAutoReconnect : public GEventBase 2 { 3 public: 4 GEventBaseWithAutoReconnect(int reconn_min = GEV_RECONNECT_TIMEOUT, int reconn_max = GEV_RECONNECT_TIMEOUT_MAX, int max_retry = GEV_RECONNECT_TRY_MAX); 5 ~GEventBaseWithAutoReconnect(); 6 7 GEventHandler* connector(); 8 bool do_connect(unsigned short port, char const* host = "127.0.0.1", void *arg = nullptr); 9 // whether this handler should be processed in foreach, 10 // true - process; false - skip 11 virtual bool filter_handler(GEventHandler *h); 12 13 protected: 14 virtual void on_error(GEventHandler *h); 15 virtual bool on_timeout(GEV_PER_TIMER_DATA *gptd); 16 17 virtual void on_connect_break(); 18 virtual bool on_connected(GEventHandler *app); 19 virtual void on_retry_max(void *arg); 20 21 protected: 22 void do_reconnect(void *arg); 23 24 …… 25 };
比较简单,只比 GEventBase 类多了一个 do_connect 接口,来扩展 connect 接口不能自动重连的问题。底层的话,是通过定时器来实现指数后退重连算法的。不过这个类也有一个限制,就是只能对一个主动连出的连接进行自动重连,限于实际需求和人力,没有再扩展它,有感兴趣的同学可以加入进来,把它搞成不限制连接数量的重连。
类图
最后,上面这些类的关系可以参考下面这张图:
黑色标注的是框架提供的类,红色是服务端派生的类,蓝色是客户端派生的类,从图中可以看到,GMyEventBase 的唯一作用就是将 svc_handler 与 clt_handler 分别引入各自的框架中,所以作为用户,关注点主要还是在派生自己的 GEventHandler 类,并通过回调接口处理数据。
后记
这个框架已经应用到我司的公共产品中,并为数个 tcp 服务提供底层支撑,经过百万级别用户机器验证,运行稳定性还是可以的,当得起“工业级”这三个字。
行文至此,我想再提一下前面在说到开源库的选型为何没有选择 libevent,而是另起炉灶。其实第一次重构的版本确实是使用 libevent 来实现的,但是发现它在 windows 上使用的是低效的 select,而且为了增加、删除句柄,它又使用了一种 self-pipe-trick 的技巧,简单说来的就是下面的代码序列:
listen (listen_fd, 1); …… connect (connect_fd, &addr, size); …… accept_fd = accept (listen_fd, &addr, &size);
在缺乏 pipe 调用的 win32 环境制造了一个 socket 自连接 (Windows 也有管道的机制,但是那个不能 select),从而进行一些通知。这一步是必要的,如果不能成功连接就会导致整个 libevent 初始化失败,从而运行不起来。
不巧的是,在一些 windows 机器上 ,由于防火墙设置严格,上述代码片段中 listen 与 connect 调用可以成功,但 accept 会失败返回,从而导致整个服务退出 (防火墙会严格禁止不在白名单上侦听的端口的连接),对于已知端口,可以通过在防火墙上设置白名单来避免,但是对于这种随机 listen 的端口,真的是太难了,基本无解。而这部分用户占比还不小,约 10%,属于不能忽略的那种。
那其它开源库有没有类似的问题呢?回头考察了一下 asio,windows 上使用的是 iocp,自然不需要这个自连接;ACE 有多种实现可供选择,如果使用 ACE_Select_Reactor / ACE_TP_Reactor 是会有这个自连接,但是你可以选择其它实现,如基于 WaitForMultipleEvents 的 ACE_WFMO_Reactor,或基于 iocp 的 ACE_Proactor 就没有这个自连接,不过前者最大只支持 62 个句柄,很容易达到上限,放弃;后者为前摄式,与反应式在编程上稍有不同,更接近于 asio。
再往前介绍一下,其实公司最早的网络库使用的就是基于 boost 的 asio,它大量的使用了 c++ 模板,有时候产生了崩溃,但是根据 dump 完全无法定位崩溃点 ——产生各种冗长的模板展开名称,且对库内部机制一无所知——导致了一些顽固的已知 bug 一直找不到崩溃点,所以才有了要去重新选型网络库以及 gevent 的诞生。
本来一开始我是想用 ACE 的,因为读过这个库的源码,对里面所有的东西都非常熟悉,但是看看 ACE 小 5 MB 的 dll 尺寸,还是放弃了 (产品本身安装包也就这么大吧),对于一个公司底层的公共组件,被各种产品携带,需要严格控制“体重”。不过后来听说 ACE 按功能拆分了代码模块,不过我没有试过。
使用这个库代替之前的 boost::asio 后,我还有一个意外收获,就是编译出来的 dll 尺寸明显小了很多,700 K -> 500 K 的样子,看来所谓模板膨胀是真有其事……
下载
最后奉上 gevent 的 github 链接,欢迎有相同需求的小伙伴前来“复刻” :