IT博客汇 | 基于ENet实现可靠UDP通信的同步模型

基于ENet实现可靠UDP通信的同步模型

春秋十二月发表于 2020-05-04 11:08:00

场景说明
选择ENet代替TCP用于弱网环境(通常丢包率高)的数据传输，提高可靠性及传输效率。为了说明怎样正确有效地应用ENet，本文按照TCP C/S同步通信的流程作了对应的接口封装实现，取库名为rudp。

接口对照
左边为rudp库的API，右边为标准的Berkeley套接字API。rudp库所有API前缀为rudp，rudp_listen和rudp_accept仅用于服务端，rudp_connect和rudp_disconnect仅用于客户端；其它接口共用于两端，其中rudp_send调用rudp_sendmsg实现，rudp_recv调用rudp_recvmsg实现。

具体实现
所有接口遵循Berkeley套接字接口的语义，为简单起见，错误描述输出到标准错误流。
◆ 监听，成功返回0，失败返回-1

1 int rudp_listen(const char *ip, int port, ENetHost **host)
2 {
3     ENetAddress address;
4
5     if(!strcmp(ip, "*"))
6            ip = "0.0.0.0";
7
8     if(enet_address_set_host_ip(&address, ip)){
9           fprintf(stderr, "enet_address_set_host_ip %s fail", ip);
10         return -1;
11     }
12
13     address.port = port;
14
15     assert(host);
16     *host = enet_host_create(&address, 1, 1, 0, 0);
17     if(NULL==*host){
18           fprintf(stderr, "enet_host_create %s:%d fail", address.host, address.port);
19         return -1;
20     }
21
22     int size = 1024*1024*1024;
23     if(enet_socket_set_option((*host)->socket, ENET_SOCKOPT_RCVBUF, size)){
24            fprintf(stderr, "enet set server socket rcvbuf %d bytes fail", size);
25     }
26
27     return 0;
28 }

◆ 接受连接，成功返回0，失败返回-1

1 int rudp_accept(ENetHost *host, unsigned int timeout, ENetPeer **peer)
2 {
3     int ret;
4     ENetEvent event;
5
6     ret = enet_host_service(host, &event, timeout);
7     if(ret > 0){
8         if(event.type != ENET_EVENT_TYPE_CONNECT){
9             if(event.type == ENET_EVENT_TYPE_RECEIVE)
10                 enet_packet_destroy(event.packet);
11             fprintf(stderr, "enet_host_service event type %d is not connect", event.type);
12             return -1;
13         }
14
15         assert(peer);
16         *peer = event.peer;
17
18     }else if(0==ret){
19         fprintf(stderr, "enet_host_service timeout %d", timeout);
20         return -1;
21
22     }else{
23         fprintf(stderr, "enet_host_service fail");
24         return -1;
25     }
26
27     return 0;
28 }

◆ 建立连接，成功返回0，失败返回-1，conn_timeout是连接超时，rw_timeout是收发超时，单位为毫秒

1 int rudp_connect(const char *srv_ip, int srv_port, unsigned int conn_timeout, unsigned int rw_timeout, ENetHost **host, ENetPeer **peer)
2 {
3     assert(host);
4     *host = enet_host_create(NULL, 1, 1, 0, 0);
5     if(NULL==*host){
6         fprintf(stderr, "enet_host_create fail");
7         goto fail;
8     }
9     if(enet_socket_set_option((*host)->socket, ENET_SOCKOPT_RCVBUF, 1024*1024*1024)){
10            fprintf(stderr, "enet set server socket rcvbuf 1M bytes fail");
11     }
12
13     ENetAddress srv_addr;
14     if(enet_address_set_host_ip(&srv_addr, srv_ip)){
15         fprintf(stderr, "enet_address_set_host_ip %s fail", srv_ip);
16         goto fail;
17     }
18     srv_addr.port = srv_port;
19
20     assert(peer);
21     *peer = enet_host_connect(*host, &srv_addr, 1, 0);
22     if(*peer==NULL){
23             fprintf(stderr, "enet_host_connect %s:%d fail", srv_ip, srv_port);
24         goto fail;
25     }
26
27     enet_peer_timeout(*peer, 0, rw_timeout, rw_timeout);
28
29     int cnt = 0;
30     ENetEvent event;
31
32     while(1){
33         ret = enet_host_service(*host, &event, 1);
34         if(ret == 0){
35             if(++cnt >= conn_timeout){
36                    fprintf(stderr, "enet_host_service timeout %d", conn_timeout);
37                 goto fail;
38             }
39
40         }else if(ret > 0){
41             if(event.type != ENET_EVENT_TYPE_CONNECT){
42                     fprintf(stderr, "enet_host_service event type %d is not connect", event.type);
43                     goto fail;
44             }
45             break; //connect successfully
46
47         }else{
48                 fprintf(stderr, "enet_host_service fail");
49             goto fail;
50         }
51 }
52
53 #ifdef _DEBUG
54 char local_ip[16], foreign_ip[16];
55 ENetAddress local_addr;
56
57 enet_socket_get_address((*host)->socket, &local_addr);
58 enet_address_get_host_ip(&local_addr, local_ip, sizeof(local_ip));
59 enet_address_get_host_ip(&(*peer)->address, foreign_ip, sizeof(foreign_ip));
60
61 printf("%s:%d connected to %s:%d", local_ip, loca_addr.port, foreign_ip, (*peer)->address.port);
62 #endif
63
64 return 0;
65
66 fail:
67 if(*host) enet_host_destroy(*host);
68 return -1;
69 }

◆ 断开连接，若成功则返回0，超时返回1，出错返回-1。先进行优雅关闭，如失败再强制关闭

1 int rudp_disconnect(ENetHost *host, ENetPeer *peer)
2 {
3     int ret;
4
5 #ifdef _DEBUG
6     char local_ip[16], foreign_ip[16];
7     ENetAddress local_addr;
8
9     enet_socket_get_address(host->socket, &local_addr);
10     enet_address_get_host_ip(&local_addr, local_ip, sizeof(local_ip));
11     enet_address_get_host_ip(&peer->address, foreign_ip, sizeof(foreign_ip));
12
13     printf("%s:%d is disconnected from %s:%d", local_ip, local_addr.port, foreign_ip, peer->address.port);
14 #endif
15
16     ENetEvent event;
17     enet_peer_disconnect(peer, 0);
18
19     while((ret = enet_host_service(host, &event, peer->roundTripTime)) > 0){
20         switch (event.type){
21         case ENET_EVENT_TYPE_RECEIVE:
22             enet_packet_destroy (event.packet);
23             break;
24
25         case ENET_EVENT_TYPE_DISCONNECT:
26             ret = 0;
27             goto disconn_ok;
28         }
29     }
30
31     ret = 0==ret ? 1 : -1;
32
33     fprintf(stderr, "enet_host_service with timeout %d %s", peer->roundTripTime, 1==ret?"timeout":"failure");
34
35     enet_peer_reset(conn->peer);
36
37 disconn_ok:
38     enet_host_destroy(host);
39     return ret;
40 }

◆ 发送数据，若成功则返回已发送数据的长度，否则返回-1

1 int rudp_sendmsg(ENetHost *host, ENetPeer *peer, ENetPacket *packet)
2 {
3     int ret;
4
5     if(enet_peer_send(peer, 0, packet)){
6         fprintf(stderr, "enet send packet %lu bytes to peer fail", packet->dataLength);
7         return -1;
8     }
9
10     ret = enet_host_service(host, NULL, peer->roundTripTime);
11     if(ret >= 0){
12         if(peer->state == ENET_PEER_STATE_ZOMBIE){
13             fprintf(stderr, "enet peer state is zombie");
14             return -1;
15         }
16         return packet->dataLength;
17
18     }else{
19         fprintf(stderr, "enet host service %u millsecond failure", peer->roundTripTime);
20         return -1;
21     }
22 }
23
24 int rudp_send(ENetHost *host, ENetPeer *peer, const void *buf, size_t len)
25 {
26     int ret;
27
28     ENetPacket *packet = enet_packet_create(buf, len, ENET_PACKET_FLAG_RELIABLE);
29     if(NULL==packet){
30         fprintf(stderr, "enet create packet %lu bytes fail", sizeof(int)+len);
31         return -1;
32     }
33
34     return rudp_sendmsg(host, peer, packet);
35 }

发送数据时需根据对端状态判断是否断线，并且packet标志设为可靠

◆ 接收数据，若成功则返回已接收数据的长度，否则返回-1

1 int rudp_recvmsg(ENetHost *host, ENetPeer *peer, ENetPacket **packet, unsigned int timeout)
2 {
3     int ret;
4     ENetEvent event;
5
6     ret = enet_host_service(host, &event, timeout);
7     if(ret > 0){
8         if(event.peer != peer){
9             fprintf(stderr, "enet receive peer is not matched");
10             goto fail;
11         }
12         if(event.type != ENET_EVENT_TYPE_RECEIVE){
13             fprintf(stderr, "enet receive event type %d is not ENET_EVENT_TYPE_RECEIVE", event.type);
14             goto fail;
15         }
16
17         *packet = event.packet;
18         return （*packet)->dataLength;
19
20 fail:
21         enet_packet_destroy(event.packet);
22         return -1;
23
24     }else {
25         fprintf(stderr, "enet receive %u millsecond %s", timeout, ret?"failure":"timeout");
26         return -1;
27     }
28 }
29
30 int rudp_recv(ENetHost *host, ENetPeer *peer, void *buf, size_t maxlen, unsigned int timeout)
31 {
32     ENetPacket *packet;
33
34     if(-1==rudp_recvmsg(host, peer, &packet, timeout))
35         return -1;
36
37     if(packet->dataLength > maxlen) {
38         fprintf(stderr, "enet packet data length %d is greater than maxlen %lu", packet->dataLength, maxlen);
39         return -1;
40     }
41
42     memcpy(buf, packet->data, packet->dataLength);
43     enet_packet_destroy(packet);
44
45     return packet->dataLength;
46 }

◆ 等待所有确认，若成功返回0，超时返回1，失败返回-1

1 int rudp_wait_allack(ENetHost *host, ENetPeer *peer, unsigned int timeout)
2 {
3     int ret, cnt = 0;
4
5     while((ret = enet_host_service(host, NULL, 1)) >= 0){
6         if(enet_peer_is_empty_sent_reliable_commands(peer, 0,
7             ENET_PROTOCOL_COMMAND_SEND_RELIABLE|ENET_PROTOCOL_COMMAND_SEND_FRAGMENT))
8             return 0;
9
10         if(peer->state == ENET_PEER_STATE_ZOMBIE){
11             fprintf(stderr, "enet peer state is zombie");
12             return -1;
13         }
14
15         if(0==ret && ++cnt>=timeout){
16             return 1;
17         }
18     }
19
20     fprintf(stderr, "enet host service fail");
21     return -1;
22 }

等待已发送数据的所有确认时，需根据对端状态判断是否断线

示例流程
左边为客户端，压缩并传输文件；右边为服务端，接收并解压存储文件。

客户端【读文件块并压缩】这个环节，需要显式创建可靠packet，并将压缩后的块拷贝到其中

春秋十二月 2020-05-04 19:08 发表评论