情景分析
在网络编程中，通常异步比同步处理更为复杂，但由于异步的事件通知机制，避免了同步方式中的忙等待，提高了吞吐量，因此效率较高，在高性能应用开发中，经常被用到。而在处理异步相关的问题时，状态机模式是一种典型的有效方法，这在libevent、memcached、nginx等开源软件(库)中多次被使用而得到见证。据此，为抛砖引玉，本文展示了使用此方法异步接收变长数据包的实现，这里的变长是指在某一种网络协议中，具有完整意义的数据包长度是不固定的。为了描述方便，我们以一个TCP C/S模式的简单例子为场景分析说明，服务端在某知名端口监听，使用异步IO复用机制epoll ET模式接受连接、接收分析请求、存取数据到内存缓冲中，这种内存缓冲类似于数据库，数据按键值对存取；客户端的请求包括增加、修改和删除三种，每种请求对应的数据包长度不一样。
协议封包
如下图所示，请求封包各字段从左到右依次为：type字段表示请求类型，占1个字节，值域为{1,2,3}，1为增加，2为修改，3为删除；key字段表示键名，占16个字节；val字段表示值，占256个字节；expire表示生存期，占4个字节，单位为秒。显而易见，对于每种类型的请求，其封包长度是固定的。
状态转换
由于在一条连接上客户端可以先后发送多种不同类型的请求，因此服务端需要接收完整某种请求的包后，才能解析处理。当处于某种特定类型的请求时，接收完它的包，这很容易实现。但当存在多种不同类型的请求时，就需要先识别当前的请求类型，再在这种类型中接收完整的包，然后再识别新的请求类型，继续循环这样的一个过程。因此，这就自然而然地对应到了状态机，如下图所示，有4个状态：1个起始状态
prepare，在此状态中识别当前请求类型，转到下一中间状态；3个中间状态
add、
set、
del，分别对应
增加、
修改和
删除请求，在此状态中不断接收数据，直至接收完整，再转到起始状态。由于从中间状态能转到起始状态，因此就没必要存在结束状态。e1、e2、e3、e4表示不同状态间转化的触发事件。
代码实现
数据结构定义
connection类表示一条在客户端和服务端间建立的连接，静态成员函数handle_read被epoll模型当有数据可读时回调，普通成员函数handle_read则做实际的接收处理。
1 enum read_state { prepare, add, set, del };
2
3static const char MSG_TYPE_ADD = 1;
4static const char MSG_TYPE_SET = 2;
5static const char MSG_TYPE_DEL = 3;
6
7#pragma pack(1)
8struct msg_add
9
{
10
char key[16];
11 char val[256];
12 uint32_t expire;
13
};
14
15struct msg_set
16{
17 char key[16];
18 char val[256];
19};
20
21struct msg_del
22{
23 char key[16];
24};
25#pragma pack()
26
27static const size_t MSG_MAX_SIZE = sizeof(msg_add);
28
29class connection
30{
31public:
32 connection();
33
34 void recv_add_msg(msg_add* msg);
35 void recv_set_msg(msg_set* msg);
36 void recv_del_msg(msg_get* msg);
37
38 bool send_add_msg(const char* key,const char* val,uint32_t expire);
39 bool send_set_msg(const char* key,const char* val);
40 bool send_del_msg(const char* key);
41
42private:
43 void reset_state()
44
{ tran_ = 0, size_ = 1; s_ = prepare;}
45
46 void handle_read();
47 static void handle_read(int fd,short ev,void* arg);
48
49private:
50 int sock_;
51 char buf_[MSG_MAX_SIZE];
52 size_t tran_;
53 size_t size_;
54 read_state s_;
55};
服务端异步接收
最初时处于
起始状态
prepare，在这个状态中：先接收1个字节，分析请求类型，更新状态，然后继续接收数据。当收到数据read返回时，那么这时已经处于3种中间状态
add、
set、
del之一了，在这个状态中：只要继续收完这种类型的请求包即可解析处理，最后再重设，返回到状态
prepare，继续接收下一个请求包。
1connection::connection()
2:sock(-1)
3,tran_(0)
4,size_(1)
5,s_(prepare)
6{
7}
8
9void connection::recv_add_msg(msg_add* msg)
10{
11}
12
13void connection::recv_set_msg(msg_set* msg)
14{
15}
16
17void connection::recv_del_msg(msg_del* msg)
18{
19}
20
21void connection::handle_read(int fd,short ev,void* arg)
22{
23 static_cast<connection*>(arg)->handle_read();
24}
25
26void connection::state_machine()
27{
28 switch(s_){
29 case prepare:
30 if(MSG_TYPE_ADD==buf_[0]){
31 tran_ = 0, size_ = sizeof(msg_add);
32 s_ = add;
33 }else if(MSG_TYPE_SET==buf_[0]){
34 tran_ = 0, size_ = sizeof(msg_set);
35 s_ = set;
36 }else if(MSG_TYPE_DEL==buf_[0]){
37 tran_ = 0, size_ = sizeof(msg_del);
38 s_ = del;
39
}else
40 assert(false);
41 break;
42
43 case add:
44 if(tran_ == size_){
45 recv_add_msg(reinterpret_cast<msg_add*>(buf_));
46 reset_state();
47 }
48 break;
49
50 case set:
51 if(tran_ == size_){
52 recv_set_msg(reinterpret_cast<msg_set*>(buf_));
53 reset_state();
54 }
55 break;
56
57 case del:
58 if(tran_ == size_){
59 recv_del_msg(reinterpret_cast<msg_del*>(buf_));
60 reset_state();
61 }
62 break;
63 }
64}
65
66void connection::handle_read()
67{
68 ssize_t ret;
69 for(;;){
70 ret = read(sock_,buf_+tran_,size_-tran_);
71 if (ret > 0){
72 tran_ += ret;
73 state_machine();
74 }else if(ret < 0 && errno == EAGAIN){
75 break;
76 }else{
77 close(sock_); break;
78 }
79 }
80}
客户端同步发送
由于一般大多数的客户端不像服务端要求高性能高并发，因此使用同步方式来发送数据。下面代码忽略了错误处理，为简单方便，发送请求的实现也写在了类connection内，依次为send_add_msg、send_set_msg、send_del_msg成员函数。
1
bool connection::send_iovec(char type,void* msg,size_t len)
2
{
3
struct iovec iov[2];
4 iov[0].iov_base = &type;
5 iov[0].iov_len = 1;
6 iov[1].iov_base = msg;
7 iov[1].iov_len = len;
8
9 return writev(sock_,iov,NUM_ELEMENTS(iov))==1+len;
10
}
11
12bool connection::send_add_msg(const char* key,const char* val,uint32_t expire)
13{
14 msg_add msg;
15 strcpy(msg.key,key);
16 strcpy(msg.val,val);
17 msg.expire = expire;
18 return send_iovec(MSG_TYPE_ADD,&msg,sizeof(msg));
19}
20
21bool connection::send_set_msg(const char* key,const char* val)
22{
23 msg_set msg;
24 strcpy(msg.key,key);
25 strcpy(msg.val,val);
26 return send_iovec(MSG_TYPE_SET,&msg,sizeof(msg));
27}
28
29bool connection::send_del_msg(const char* key)
30{
31 msg_del msg;
32 strcpy(msg.key,key);
33 return send_iovec(MSG_TYPE_GET,&msg,sizeof(msg));
34}
小结
虽然以上所述的场景是网络通信，但对于进程间使用管道和字节流套接字的通信，也同样适合。
posted on 2012-09-20 15:48
春秋十二月 阅读(2768)
评论(2) 编辑 收藏 引用 所属分类:
Network