Anoii之UDP与多路复用

jhDfiPdP

代码连接：https://github.com/Afeather2017/anoii/blob/master/src/udp_peer.cc
以往写了TCP的多路复用，发现它还挺难写对的。现在写UDP的，发现似乎没有容易太多。
本人所在的公司，UDP用于本机通讯时（即loopback通讯），假设了一个UDP包总是能够完整的发送到对端，且对端总是能够回复一个完整的包，所以为了处理这个问题，费了些力气。
UDP基本用法

同步的UDP的使用过程如下，忽略了头文件与错误处理过程：
// 服务端#define PORT 12345#define BUFFER_SIZE 1024int main() {    int sockfd;    struct sockaddr_in server_addr, client_addr;    socklen_t client_len = sizeof(client_addr);    char buffer[BUFFER_SIZE];    // 创建 UDP 套接字    sockfd = socket(AF_INET, SOCK_DGRAM, 0);    // 设置服务器地址    memset(&server_addr, 0, sizeof(server_addr));    server_addr.sin_family = AF_INET;    server_addr.sin_addr.s_addr = INADDR_ANY;    server_addr.sin_port = htons(PORT);    // 绑定套接字    bind(sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0);    printf("UDP 服务器已启动，等待客户端连接...\n");    while (1) {        // 接收客户端消息        ssize_t recv_len = recvfrom(sockfd, buffer, BUFFER_SIZE - 1, 0,                                    (struct sockaddr *)&client_addr, &client_len);        if (recv_len > 0) {            buffer[recv_len] = '\0';            printf("收到来自 %s:%d 的消息：%s\n", inet_ntoa(client_addr.sin_addr),                   ntohs(client_addr.sin_port), buffer);            // 发送回复消息            const char *response = "服务器已收到消息";            sendto(sockfd, response, strlen(response), 0,                   (struct sockaddr *)&client_addr, client_len);        }    }    close(sockfd);    return 0;}// 客户端#define PORT 12345#define SERVER_IP "127.0.0.1"#define BUFFER_SIZE 1024int main() {    int sockfd;    struct sockaddr_in server_addr;    char buffer[BUFFER_SIZE];    // 创建 UDP 套接字    sockfd = socket(AF_INET, SOCK_DGRAM, 0);    // 设置服务器地址    memset(&server_addr, 0, sizeof(server_addr));    server_addr.sin_family = AF_INET;    server_addr.sin_addr.s_addr = inet_addr(SERVER_IP);    server_addr.sin_port = htons(PORT);    // 发送消息    const char *message = "你好，服务器！";    sendto(sockfd, message, strlen(message), 0,           (struct sockaddr *)&server_addr, sizeof(server_addr));    printf("已发送消息：%s\n", message);    // 接收服务器回复    ssize_t recv_len = recvfrom(sockfd, buffer, BUFFER_SIZE - 1, 0, NULL, NULL);    if (recv_len > 0) {        buffer[recv_len] = '\0';        printf("收到来自服务器的消息：%s\n", buffer);    }    close(sockfd);    return 0;}好吧，正常流程就这么多，但是也有许多坑。
坑1: 如何才能完整发送一个包？

UDP没有传输控制，所以无法确定发送的包被接收到，这里说的完整发送就只是交给了操作系统罢了。这里不考虑操作系统拿到整个包只有由于缓冲区不够而丢弃包的问题。
Linux中，一次成功的sendto就代表发送一个包，一次成功的recvfrom就代表接收了一个包。所以不用担心sendto和recvfrom成功后，你实际发送或接收的是半个包。如果sendto发送的包大小超过了MTU该怎么办？不用担心，UDP会进行分段发送（即IP层分片），分段发送后，如果对端完整接收了所有分段，就会把它们组装起来，否则全部丢弃。
理论能发送的包的最大大小是65527。它不是65535，原因是UDP头部中的长度字段的范围是0到65535，包括了头部8字节的大小。所以如果你尝试发送一个大小是65528的UDP包，会出现包过长的问题：

但是实际上还有其他地方进行了限制，实际能够发送的包的大小会比65527小许多。我的电脑可以发送35000的包，zerotier似乎只支持1472以内的包。
sendto出错时，可能有很多的情况。其errno值中有一项是EINTR，意思是被中断了。比如说发送的时候收到了SIGPIPE之类的信号，那么这个时候你这个包就没有成功提交到操作系统中，那么这个包将不可能进入到网络中。所以写了很长一段恶心的代码：
void UdpPeer::SendTo(const char *data, int size, InetAddr &addr) {  assert(size > 0);  auto *sock_addr = addr.GetSockAddr();  for (;;) {    int sent = ::sendto(fd_, data, size, 0, sock_addr, sizeof(*sock_addr));    if (sent == size) {      if (!binded_addr_) { // 如果没有绑定，那么sendto之后会系统偷偷绑定了一个ip:port        binded_addr_ = true;        addr_ = GetLocalAddr(fd_);      }      return;    }    if (sent > 0) {      Error("Wants to send {} but sent {} actually", size, sent);      return;    }    Error(strerror(errno));    // 根据man文档可以得知以下错误。但实际上似乎没有必要关心这么多的问题。    switch (errno) {      ......      case EINTR:        // 中断。再试试        continue;      ......    }  }}坑2: 如何确保完整接收了一个包？

这个问题实际上是如何确保已经到达操作系统的包被完整的接收
除了上文说的EINTR以外，还有一个关键的recvfrom的缓冲区参数的问题。
recvfrom的声明是：
ssize_t recvfrom(int socket, void *restrict buffer, size_t length,           int flags, struct sockaddr *restrict address,           socklen_t *restrict address_len);如果成功执行，那么返回值是填入缓冲区的数据量，否则返回-1。
当flags为0的时，如果这个包的大小超过了length，那么你只拿到了包的前半部分，后半部分拿不到了，直接丢弃；如果没超过，那么说明你的包被完整接收到程序里面了。
这里的超过与没超过，实际上是recvfrom的返回值size与length对比。如果length > size，表示这个包确确实实被完整接收了；如果length = size，那么这个包有可能没有被完整接收；length < size的情况不存在。
当flags为MSG_PEEK的时候，recvfrom就只是拿包出来看了一眼而已，并不会丢弃这个包。所以我们可以通过这个方式来试探一个包的大小。
所以，只有length > size的时候才可以把这个包交给回调。因此又写了一段恶心的代码：
void UdpPeer::OnMessage() {  InetAddr peer{};  auto *peer_addr = peer.GetSockAddr();  socklen_t len = sizeof(*peer.GetSockAddr());  int size;  if (!binded_addr_) {    Fatal("Tries recvfrom an unbinded UDP socket");  }  for (;;) {    size = ::recvfrom(fd_,                      buffer_.data(),                      buffer_.size(),                      auto_buffer_size_ ? MSG_PEEK : 0,                      peer_addr,                      &len);    if (size > 0) {      if (!auto_buffer_size_) {        if (size < buffer_.size()) break;        Error("Package corrupted, ignore it.");        return;      }      if (size >= buffer_.size()) {        // UDP的包的长度字段包括了首部的长度，所以不是65535        // 试探一个包的大小。        if (buffer_.size() * 2 >= 65527 + 1) {          buffer_.resize(65527 + 1);        } else {          buffer_.resize(buffer_.size() * 2);        }        continue;      }      ::recvfrom(fd_, nullptr, 0, 0, nullptr, nullptr);      break;    }    switch (size) {      ......      case EINTR:        if (!auto_buffer_size_) return;        // 中断，由于使用的是PEEK参数，所以中断之后这个数据还有救        continue;      ......    }  }  // 啧，坑真多……只有缓冲区比size大才可能表明接收的是整个包而不是半个。  // 如果没有保证尽量接收，即auto_buffer_size_=false，那么就有可能出现这种情况  assert(size >= 0);  assert(size < buffer_.size());  // Package corrupted  readable_cb_(this, peer, buffer_.data(), size);}坑3: 没有bind的时候进行了recvfrom

如果一个UDP socket没有进行bind，此时recvfrom，如果是阻塞IO，那么此时recvfrom会永远阻塞。非阻塞是否会有这个情况我不知道，也不想试，所以就加上了这样的判断。前文的binded_addr_就是做这个的。
在sendto调用的时候，操作系统会“隐式”地bind一个端口给socket，所以sendto的时候也会设置binded_addr_。
坑4: 与多路复用结合

poll有POLLIN, POLLOUT，而UDP中POLLOUT是没用的，因为它没有TCP那样的传输控制，没有发送缓冲区，所以我们只要设置一个POLLIN即可。其他多路复用函数操作类似。