TCP 服务器：recv() 具有动态增加缓冲区大小

Question

我试图读取客户端发送到 TCP 服务器的所有数据，但事先不知道数据可能有多大，并且不限制接收的缓冲区大小。 TCP 服务器基于

epoll()

并且套接字是非阻塞的。服务器是多线程的：主事件循环检查 epoll 事件是否是一个新连接，即套接字文件描述符，并接受它，否则它将向队列添加一个任务，一旦线程可用，该任务将由线程池处理处理与客户端的 I/O。这一切都有效，但是，我无法实现的是动态增加接收的缓冲区大小。

此实现的问题是，每当接收到的输入的长度等于给定给 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) {
      // each client when the connection is accepted gets a NULL buffer to recieve their data
      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

）程序打印出以下内容：

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

如果我发送另一条长度为奇数的消息，我只会收到消息

1234

，例如发送

o

然后我现在得到

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 shows up here now
o

有没有办法以不同的方式做到这一点？

Answer 1

我想我找到了解决方案，而且它正在发挥作用。非常感谢评论和提示。这个想法是期望一个分隔符，在本例中为

\r\n

并修改

extend_rbuf()

函数：

static void do_recv(int fd) {
  char rbuf[MAX_RECV_BUFFER_SIZE] = {0};
  int n = -1;
  while (1) {
    memset(rbuf, 0, sizeof(rbuf)); // reset the receiver char
    n = recv(fd, rbuf, sizeof(rbuf), MSG_DONTWAIT);
    if (n > 0) {
      extend_rbuf(fd, rbuf);
      if (n == sizeof(rbuf) && rbuf[sizeof(rbuf) - 1] != '\n' &&
          rbuf[sizeof(rbuf) - 2] != '\r') {
        continue;
      }
      std::cout << "read ended here, read from client: "
                << clients.get_client(fd)->rbuf << '\n';
      // ... some send logic ...
      clients.get_client(fd)->rbuf.clear();
      if (errno == EINTR) {
        continue;
      }
      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()

是

void extend_rbuf(int fd, const char *data) {
  std::string *client_rbuf = &clients.get_client(fd)->rbuf;
  (*client_rbuf).append(data);
}

这适用于任何输入长度，即使初始缓冲区的大小为 2。如果我遇到任何问题或找到更好的解决方案，我将更新此答案。