状态机模式之应用--异步接收变长数据包.doc

情景分析 在网络编程中，通常异步比同步处理更为复杂，但由于异步的事件通知机制，避免了同步方式中的忙等待，提高了吞吐量，因此效率较高，在高性能应用开发中，经常被用到。而在处理异步相关的问题时，状态机模式是一种典型的有效方法，这在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 910 char key16;11 char val256;12 uint32_t expire;13;1415struct msg_set1617 char key16;18 char val256;19;2021struct msg_del2223 char key16;24;25#pragma pack()2627static const size_t MSG_MAX_SIZE = sizeof(msg_add);2829class connection3031public: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);3738 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)10111213void connection:recv_set_msg(msg_set* msg)14151617void connection:recv_del_msg(msg_del* msg)181920 21void connection:handle_read(int fd,short ev,void* arg)2223 connection* conn = static_cast(arg);24 conn-handle_read(); 252627void connection:state_machine()2829 switch(s_)30 case prepare: 31 if(MSG_TYPE_ADD=buf_0)32 tran_ = 0, size_ = sizeof(msg_add);33 s_ = add;34 else if(MSG_TYPE_SET=buf_0)35 tran_ = 0, size_ = sizeof(msg_set);36 s_ = set;37 else if(MSG_TYPE_DEL=buf_0)38 tran_ = 0, size_ = sizeof(msg_del);39 s_ = del;40 else 41 assert(false);42 break;43 44 case add:45 if(tran_ = size_)46 msg_add* msg = reinterpret_cast(buf_);47 recv_add_msg(msg);48 reset_state();49 50 break;51 52 case set:53 if(tran_ = size_)54 msg_set* msg = reinterpret_cast(buf_);55 recv_set_msg(msg);56 reset_state();57 58 break;59 60 case del:61 if(tran_ = size_)62 msg_del* msg = reinterpret_cast(buf_);63 recv_del_msg(msg);64 reset_state();65 66 break;67 686970void connection:handle_read()7172 ssize_t ret;73 for(;)74 ret = read(sock_,buf_+tran_,size_-tran_);75 if (ret 0)76 tran_ += ret;77 state_machine();78 else if(ret 0 & errno = EAGAIN)79 break;80 else81 close(sock_); break;82 83 84 客户端同步发送 由于一般大多数的客户端不像服务端要求高性能高并发，因此使用同步方式来发送数据。下面代码忽略了错误处理，为简单方便，发送请求的实现也写在了类connection内，依次为send_add_msg、send_set_msg、send_del_msg成员函数。 1bool connection:send_iovec(char type,void* msg,size_t len) 2 3 struct iovec iov2; 4 iov0.iov_base = &type; 5 iov0.iov_len = 1; 6 iov1.iov_base = msg; 7 iov1.iov_len = len; 8 9 return writev(sock_,iov,NUM_ELEMENTS(iov)=1+len;101112bool connection:send_add_msg(const char* key,const char* val,uint32_t expire)1314 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);192021bool connection:send_set_msg(const char* key,const char* val)2223 msg_set msg; 24 strcpy(msg.key,key);25 strcpy(msg.val,val); 26 return send_io