基于epoll的非阻塞多线程TCP Server动态缓冲区接收异常问题

我正在实现一个TCP服务器，要求能读取客户端发送的全部数据，不需要提前知道数据大小，也不限制接收缓冲区的大小。这个Server基于epoll实现，套接字是非阻塞模式，采用多线程架构：主事件循环检测epoll事件，新连接就接受，否则把任务加入队列，由线程池里的空闲线程处理客户端I/O。目前这部分功能正常，但没法实现接收缓冲区的动态增长，还存在一个问题：当接收的输入长度等于recv()给定的最大缓冲区大小的倍数时，循环会一直等待下一条消息。

接收函数do_recv实现如下：

static void do_recv(int fd) {
  char rbuf[MAX_RECV_BUFFER_SIZE] = {0};
  int n = -1;
  while (1) {
    n = recv(fd, rbuf, sizeof(rbuf), MSG_DONTWAIT);
    if (n > 0) {
      // 每个客户端连接建立时会分配一个空缓冲区用于接收数据
      extend_rbuf(fd, rbuf, n, sizeof(rbuf)); 
      std::cout << "bytes read: " << n << " current read: " << rbuf << "\n";
      std::cout << "size: " << clients.get_client(fd)->rbuf.size
                << " len: " << clients.get_client(fd)->rbuf.len << "\n";
      if (n == sizeof(rbuf)) {
        continue;
      }
      std::cout << "read ended here, read from client\n"
                << clients.get_client(fd)->rbuf.buf << '\n';
      break;
    }
    if (n == 0) {
      return close_fd(fd);
    }
    if (errno == EINTR) {
      continue;
    } else if (errno == EAGAIN) {
      modify_fd_event(fd, EPOLLIN);
      break;
    } else {
      return close_fd(fd);
    }
  }

  if (n <= 0) {
    return;
  }
}

extend_rbuf函数定义如下：

int extend_rbuf(int fd, const char *data, int rlen, int rsize) {
  Clients::client_buffer *client_rbuf = &(clients.get_client(fd)->rbuf);
  if (client_rbuf->buf == NULL) {
    client_rbuf->buf = (char *)malloc(sizeof(char) * rlen);
    client_rbuf->size = rsize;
  }
  if (client_rbuf->size - client_rbuf->len > rlen) {
    client_rbuf->len = rlen;
    memcpy(client_rbuf->buf + client_rbuf->len, data, rlen);
  } else {
    client_rbuf->size = client_rbuf->len + rlen + 1;
    char *new_rbuf = (char *)realloc(client_rbuf->buf, client_rbuf->size * sizeof(char));
    if (!new_rbuf) {
      perror("realloc new_rbuf: out of memory");
    }
    client_rbuf->buf = new_rbuf;
    memcpy(client_rbuf->buf + client_rbuf->len, data, rlen);
    client_rbuf->len += rlen;
  }
  return 0;
}

问题触发点是这段条件判断：

if (n == sizeof(rbuf)) {
  continue;
}

测试时将MAX_RECV_BUFFER_SIZE设为2：

• 发送长度l满足l mod 2 = 1的消息（比如telnet发送"123"，telnet会自动添加\r\n），程序正常触发"read ended here"，输出：

bytes read: 2 current read: 12
size: 3 len: 2
bytes read: 2 current read: 3
size: 5 len: 4
bytes read: 1 current read: 

size: 6 len: 5
read ended here, read from client
123

• 发送长度为4的消息如"1234"时，程序无法触发输出，只会循环等待：

bytes read: 2 current read: 12
size: 3 len: 2
bytes read: 2 current read: 34
size: 5 len: 4
bytes read: 2 current read: 

size: 7 len: 6

只有再发送一条长度为奇数的消息（比如"o"），才能看到之前的"1234"：

bytes read: 2 current read: o
size: 9 len: 8
bytes read: 1 current read: 

size: 10 len: 9
read ended here, read from client
1234 # 1234 此时才会显示
o

有没有其他实现方式可以解决这个问题？

核心问题分析 #

1. extend_rbuf的长度计算bug：当缓冲区有剩余空间时，错误地将client_rbuf->len直接赋值为当前读取长度rlen，而非累加，导致缓冲区长度统计混乱。
2. 错误的读取终止判断：用n == sizeof(rbuf)判断是否继续读取是完全错误的——TCP是流式协议，没有消息边界，无法通过单次recv()的返回值判断是否还有数据，必须依赖EAGAIN（非阻塞模式下）确定当前无更多数据可读。

具体修改步骤 #

1. 修复extend_rbuf的缓冲区长度与内存分配逻辑

int extend_rbuf(int fd, const char *data, int rlen, int rsize) {
  Clients::client_buffer *client_rbuf = &(clients.get_client(fd)->rbuf);
  if (client_rbuf->buf == NULL) {
    // 首次初始化：分配与当前读取长度匹配的内存
    client_rbuf->buf = (char *)malloc(sizeof(char) * rlen);
    if (!client_rbuf->buf) {
      perror("malloc failed");
      return -1;
    }
    client_rbuf->size = rlen;
    client_rbuf->len = rlen;
    memcpy(client_rbuf->buf, data, rlen);
    client_rbuf->buf[client_rbuf->len] = '\0'; // 确保字符串终止
    return 0;
  }
  // 检查剩余空间是否足够
  if (client_rbuf->size - client_rbuf->len >= rlen) {
    memcpy(client_rbuf->buf + client_rbuf->len, data, rlen);
    client_rbuf->len += rlen;
  } else {
    // 空间不足时扩容，额外+1预留字符串终止位
    size_t new_size = client_rbuf->len + rlen + 1;
    char *new_rbuf = (char *)realloc(client_rbuf->buf, new_size);
    if (!new_rbuf) {
      perror("realloc new_rbuf: out of memory");
      return -1;
    }
    client_rbuf->buf = new_rbuf;
    client_rbuf->size = new_size;
    memcpy(client_rbuf->buf + client_rbuf->len, data, rlen);
    client_rbuf->len += rlen;
  }
  client_rbuf->buf[client_rbuf->len] = '\0';
  return 0;
}

2. 修改do_recv的读取循环逻辑

移除错误的n == sizeof(rbuf)判断，改为只要recv()返回>0就继续读取，直到遇到EAGAIN、错误或连接关闭，最后统一输出读取到的数据：

static void do_recv(int fd) {
  char rbuf[MAX_RECV_BUFFER_SIZE] = {0};
  int n = -1;
  bool has_data = false;
  while (1) {
    n = recv(fd, rbuf, sizeof(rbuf), MSG_DONTWAIT);
    if (n > 0) {
      has_data = true;
      extend_rbuf(fd, rbuf, n, sizeof(rbuf)); 
      // 用std::string避免空字符导致的输出乱码
      std::cout << "bytes read: " << n << " current read: " << std::string(rbuf, n) << "\n";
      std::cout << "size: " << clients.get_client(fd)->rbuf.size
                << " len: " << clients.get_client(fd)->rbuf.len << "\n";
      continue;
    }
    if (n == 0) {
      std::cout << "client closed connection\n";
      return close_fd(fd);
    }
    if (errno == EINTR) {
      continue;
    } else if (errno == EAGAIN) {
      modify_fd_event(fd, EPOLLIN);
      break;
    } else {
      perror("recv error");
      return close_fd(fd);
    }
  }

  // 循环结束后，若读取到数据则输出完整内容
  if (has_data) {
    std::cout << "read ended here, read from client\n"
              << clients.get_client(fd)->rbuf.buf << '\n';
  }
}

修改后的效果 #

无论发送的数据长度是缓冲区大小的整数倍还是奇数，都会在当前数据读取完毕（recv()返回EAGAIN）后，立即输出完整的接收内容，不会出现等待下一条消息的情况。

内容的提问来源于stack exchange，提问作者mino