一個工業級、跨平臺、輕量級的 tcp 網路服務框架：gevent

作為公司的公共產品，經常有這樣的需求：就是新建一個本地服務，產品線作為客戶端通過 tcp 接入本地服務，來獲取想要的業務能力。

與印象中動輒處理成千上萬連接的 tcp 網路服務不同，這個本地服務是跑在客戶機器上的，Win32 上作為開機自啟動的 windows 服務運行；

Linux 上作為 daemon 在後臺運行。總的說來就是用於接收幾個產品進程的連接，因此輕量化是其最重要的要求，在這個基礎上要能兼顧跨平臺就可以了。

其實主要就是 windows，再兼顧一點兒 linux。

考察了幾個現有的開源網路框架，從 ACE 、boost::asio 到 libevent，都有不盡於人意的地方：

a) ACE：太重，只是想要一個網路框架，結果它扒拉扒拉一堆全提供了，不用還不行；

b) boost::asio：太複雜，牽扯到 boost 庫，並且引入了一堆 c++ 模板，需要高版本 c++ 編譯器支持；

c) libevent：這個看著不錯，當時確實用這個做底層封裝了一版，結果發版後發現一個比較致命的問題，導致在防火牆設置比較嚴格的機器上初始化失敗，這個後面我會詳細提到。

其它的就更不用說了，之前也粗略看過陳碩的 muddo，總的感覺吧，它是基於其它開源框架不足地方改進的一個庫，有相當可取的地方，但是這個改進的方向也主要是解決更大併發、更多連接，不是我的痛點，所以沒有繼續深入研究。

好了，與其在不同開源框架之間糾結，不如自己動手寫一個。

反正我的場景比較固定，不用像它們那樣面面俱，我給自己羅列了一些這個框架需要支持基本的功能：

1）同步寫、非同步讀；

2）可同時監聽多路事件，基於 1）這裡只針對非同步 READ 事件（包含連接進入、連接斷開），寫數據是同步的，因而不需要處理非同步 WRITE 事件；

3）要有設置一次性和周期性定時器的能力 （業務決定的）；

4）不需要處理信號 （windows 上也沒信號這一說，linux 自己搞搞 sigaction 就好啦）；

……

雖然這個框架未來只會運行在用戶的單機上，但是我不希望它一齣生就帶有性能缺陷，所以性能平平的 select 沒能進入我的法眼，我決定給它裝上最強大的心臟：

Windows 平臺： iocp

Linux 平臺：epoll

ok，從需求到底層技術路線，貌似都講清楚了，依照 libevent 我給它取名為 gevent，下麵我們從代碼級別看下這個框架是怎麼簡化 tcp 服務搭建這類工作的。

首先看一下這個 tcp 服務框架的 sample：

svc_handler.h

 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 };

epoll_svc.cpp

 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 類，從後者派生而來，

只改寫了一個 create_handler 介面來提供我們的事件處理對象 svc_handler，它是從框架中的 GEventHandler 派生而來，

svc_handler 只改寫了一個 on_read_msg 來處理 Json 格式的消息輸入。

程式的運行就是分別調用 GMyEventBase（實際上是GEventBase）  的 init / listen / run / fini / cleaup 方法。

而與業務相關的代碼，都在 svc_handler 中處理：

svc_handler.cpp

 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 後返回給客戶端。

再來看下客戶端 sample：

clt_handler.h

 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 };

epoll_clt.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 的派生類 GMyEventBase 來作為事件迴圈的核心，所不同的是（註意並非之前例子里的那個類，雖然同名），它提供了 clt_handler 來處理自己的業務代碼。

另外為了提供連接中斷後自動向服務重連的功能，這裡 GMyEventBase 派生自 GEventBase 類的子類 GEventBaseWithAutoReconnect （位於 EventBaseAR.h/cpp 中）。

程式的運行是分別調用 GEventBase 的 init / connect / fini / cleaup 方法以及 GEventHandler 的 send / disconnect 來測試讀寫與連接。

定義巨集 TEST_READ 用來測試讀寫；定義巨集 TEST_CONN 可以測試連接的通斷及讀寫；定義巨集 TEST_TIMER 來測試周期性定時器及讀寫。它們是互斥的。

clt_handler 主要用來非同步接收服務端的回送數據並列印：

clt_handler.cpp

 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 腳本用來製造符合格式的測試數據：

epoll_gen.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 來編譯代碼：

CMakeLists.txt

 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

測試多個客戶端同時連接，可以使用下麵的腳本：

epoll_start.sh

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 退出單個客戶端；

通過下麵的腳本來停止多個客戶端與服務端：

epoll_stop.sh

1 #! /bin/sh
2 pkill -INT epoll_clt
3 sleep 1
4 pkill -INT epoll_svc

框架的用法介紹完之後，再簡單游覽一下這個庫的各層級對外介面。

EventBase.h

  1 #pragma once
  2 
  3 
  4 #include "EventHandler.h" 
  5 #include <string>
  6 #include <map>
  7 #include <mutex> 
  8 #include <condition_variable>
  9 #include "thread_group.hpp"
 10 
 11 #define GEV_MAX_BUF_SIZE 65536
 12 
 13 class GEventBase : public IEventBase
 14 {
 15 public:
 16     GEventBase();
 17     ~GEventBase();
 18 
 19 #ifdef WIN32
 20     virtual HANDLE iocp () const; 
 21 #else
 22     virtual int epfd () const; 
 23 #endif
 24     virtual bool post_timer(GEV_PER_TIMER_DATA *gptd); 
 25     virtual GEventHandler* create_handler() = 0; 
 26 
 27     //