从select引起的bug聊聊多路复用一

作者: 明翼 | 来源:发表于2022-09-13 11:08 被阅读0次

一前言

首先祝大家双节过的开心，平安喜乐！很久没写文章了，主要自己还在沉淀，学习类的分享总觉得为了分享而分享，多几天可能自己都记不清细节了，所以一直没有再去写，这次遇到一个比较有意思的bug，多路复用的一个bug，这个领域那，虽然自己也学习过，但是一直也没写过代码练习，就这个机会就一并练习下，可能对高手来说这是稀松平常的问题，却耗费了我们一天左右的时间进行问题的排查。

二问题描述和排查步骤

我们有个跑了很久的c开发的系统，在新版本测试中，发现一直会core，core的位置飘忽不定，而且core的有点莫名其妙，根本不该core的地方却core了，开始从现象看来很像是多线程引起的问题，排查了下却没有发现问题所在。

由于代码量很多，我们排查步骤是：

利用ascan库定位core的位置，我们根据core的地方开始关闭相关的功能。
减少了core的地方后，接下来还是会core，core的位置在一个unix socket 通信线程的创建上，这个线程本该早就创建好的，但是为什么运行5-10分钟才开始创建，线程创建没有做父子进程的监控，所以不存在重启可能而且如果是这个线程挂了，引起的重新创建也是不可能的，因为线程挂了，必然会导致进程都挂了，结果整个进程的其他线程仍然是正常运行的。（这个至今无解）
由于是线程创建问题，同事注意到了此进程的由于新增写kafka的功能，导致线程过多，遂代码上注释掉这些功能，继续排查。
由于这个线程主要用来执行一些程序交互命令的，所以就用客户端工具连着去测试，发现经常连不上，有时候连上也会core，ascan的报错信息：

ASAN:SIGSEGV
=================================================================
==316088== ERROR: AddressSanitizer: SEGV on unknown address 0x00000366650b (pc 0x00000366650b sp 0x7f6e7db81fa0 bp 0x7f6e7db82820 T236)
AddressSanitizer can not provide additional info.
    #0 0x366650a (+0xbd650a)

从报错信息利用add2line命令查到具体的堆栈，这个命令以前文章有聊过,执行起来是：

addr2line -a -C -e bin/可执行程序 pc对应的地址

如是经过gdb调试，发现core的时候在unix socket的处理函数的返回上，也就是说栈信息被破坏了，百思不得其解啊，甚至汇编每次跟踪地址也没查到谁破坏的。

正常连接的时候，客户端进程卡死，通过strace 跟踪客户端的系统调用，如下：

socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0) = 3
connect(3, {sa_family=AF_UNIX, sun_path="./run/xxxx.socket"}, 33) = 0
ioctl(3, FIONBIO, [1])                  = 0
poll([{fd=3, events=POLLOUT}], 1, 10000) = 1 ([{fd=3, revents=POLLOUT}])
sendto(3, "{\"version\": \"0.2\"}", 18, 0, NULL, 0) = 18
select(4, [3], [], [], {tv_sec=600, tv_usec=0}) = 1 (in [3], left {tv_sec=599, tv_usec=999794})
poll([{fd=3, events=POLLIN}], 1, 10000) = 1 ([{fd=3, revents=POLLIN}])
recvfrom(3, "{\"return\":\"OK\"}\n", 1024, 0, NULL, NULL) = 16
poll([{fd=3, events=POLLOUT}], 1, 10000) = 1 ([{fd=3, revents=POLLOUT}])
sendto(3, "{\"command\": \"command-list\"}\n", 28, 0, NULL, 0) = 28
select(4, [3], [], [], {tv_sec=600, tv_usec=0}

通过客户端的日志打印信息，发送command-list命令后服务器端没有返回，sendto命令是成功的，返回28，来看看服务器端怎么说：

[23331] 9/9/2022 -- 22:01:47 - (xxx.c:403) <Info> (xx) -- Unix socket: recv msg: {"version": "0.2"}
[23331] 9/9/2022 -- 22:01:47 - (xxx.c:449) <Info> (xx) -- Unix socket: send to client:(null)
[23331] 9/9/2022 -- 22:01:47 - (xxx.c:343) <Info> (xx) -- Unix socket: send content:{"return":"OK"}
[23331] 9/9/2022 -- 22:01:47 - (xxx.c:345) <Info> (xx) -- Unix socket:sent message of size 16 to client socket 1118

通过服务器端的日志来看，只收到了初次的版本信息，后续的command-list命令并没有收到。
这就很奇怪了。

交互图

百思不得其解，是难道是内核bug？通过gdb调试并没有发现什么问题，接着通过lsof 查看socket文件的连接数，当我们通过客户端去连接的时候，连接数递增了，这没啥问题，如下图：

[root@localhost xx]# lsof ./run/command.socket
COMMAND     PID USER   FD   TYPE             DEVICE SIZE/OFF    NODE NAME
xxx   30894 root  101u  unix 0xffff8810e7e99800      0t0  300947 ./run/command.socket
xxx  30894  root  1172u  unix 0xffff8802b42a4000      0t0 1446065 ./run/command.socket

开始没有注意到这个1172，这个文件描述符有什么特别的地方，也知道select做多路复用的时候，有一定的局限，只能处理1024个连接，我在想，我们就只有一个连接没有超过1024这个限制啊，
也许有朋友知道了原因，是1172超过了1024，也就是说select的FD的数量不能超过1024，且大小也不能超过，那么就是这么简单嘛，继续实践吧。

The behavior of these macros is undefined if a descriptor value is less than zero or
greater than or equal to FD_SETSIZE, which is normally at least equal to the maximum num-
ber of descriptors supported by the system.

三多路复用

在高性能的服务器上，多采用多路复用技术，多路其实就是多个连接，复用就是复用此服务器进程，那么何在一起多路复用，就是用一个进程进行多个连接的处理。

对于服务器来说，开放端口等待客户端连接，开始多采用多进程或多线程编程的方式，即每个连接采用单独的进程或线程进行处理，但是每台计算机因为内存等资源限制，可以开的进程或线程数有限，而且过多的线程会导致线程切换的成本过大，缓存失效等一系列问题，根本无法做到单机处理十万、百万连接。

如果采用非阻塞，在用户进程里面轮询方式那？这样会占用很高的cpu资源，所以后来发展出多路复用技术，即采用一个进程处理多个连接，一个引用怎么处理多个连接那，不可能采用阻塞的方式，一旦阻塞在一个连接的IO上，其他连接有事件过来了也没办法处理，那只能轮询查看各个连接上是否有可读、可写消息，从而达到多路复用的目的，linux内核提供select、poll、epoll三种多路复用机制。

3.1 select 机制实现多路复用

3.1.1 基本使用说明

/* According to POSIX.1-2001 */
#include <sys/select.h>

int select(int nfds, fd_set *readfds, fd_set *writefds,
        fd_set *exceptfds, struct timeval *timeout);

void FD_CLR(int fd, fd_set *set);           // 从fdset中删除fd
int  FD_ISSET(int fd, fd_set *set);         // 判断fd是否已存在fdset
void FD_SET(int fd, fd_set *set);           // 将fd添加到fdset
void FD_ZERO(fd_set *set);              // fdset所有位清0

1.nfds 表示监视的文件描述符中，待测的最大描述符+1.

readfds：监视有读数据到达的文件描述符集合。
writefds：监视有写数据到达的文件描述符集合。
exceptfds：监视有异常发生的文件描述符集合。
这三个集合每次都要传入，每当要监视的事件发生时候，都会被复制出来。
timeout 设置为NULL，则select阻塞，直到事件发生；如果不为NULL，且值不为0，则等待固定时间，如果这个事件没有监视事件来的话，也仍然会返回；如果不为NULL，且值为0，则不等待，立刻返回。

下面四个为宏，含义如后面的注释，在linux内核的中的实现如下（不同的版本实现稍微有差异）：

#define __NFDBITS (8 * sizeof(unsigned long))                // 每个ulong型可以表示多少个bit,
#define __FD_SETSIZE 1024                                          // socket最大取值为1024
#define __FDSET_LONGS (__FD_SETSIZE/__NFDBITS)     // bitmap一共有1024个bit，共需要多少个ulong
 
#define __FD_ELT(d) ((d) / __NFDBITS)
#define __FD_MASK(d)    ((__fd_mask) 1 << ((d) % __NFDBITS))
 
typedef struct {
    unsigned long fds_bits [__FDSET_LONGS];                 //用ulong数组来表示bitmap
} __kernel_fd_set;
 
typedef __kernel_fd_set   fd_set;

#define __FD_SET(d, set) \
  ((void) (__FDS_BITS (set)[__FD_ELT (d)] |= __FD_MASK (d)))
#define __FD_CLR(d, set) \
  ((void) (__FDS_BITS (set)[__FD_ELT (d)] &= ~__FD_MASK (d)))
#define __FD_ISSET(d, set) \
  ((__FDS_BITS (set)[__FD_ELT (d)] & __FD_MASK (d)) != 0)

fd_set是由unsigned long 的类型组构成的位图， FD_SET 操作即找到哪个unsigned long的哪个位，通过((__fd_mask) 1 << ((d) % __NFDBITS)) 来定位具体的位信息，将那一位设置为1，取反即设置为0.

问题出在是FD_SET地方，即在__FD_ELT(d) ((d) / __NFDBITS) 如果d的值大于1024，那么fds_bits 就越界了，就会破坏栈数据，从而导致返回异常。

简化下我们的程序，写的如下：

#服务器端 
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>

#define CLIENT_SIZE 100
#define SOCK_FILE "command.socket"
#define TOO_MANY "Too many client."

typedef struct unix_socket_infos_ {
  int socket;
  int select_max;
  struct sockaddr_un client_addr;
  int clients[CLIENT_SIZE];
} unix_socket_infos_t;

static int create_unix_socket(unix_socket_infos_t *this) {
  struct sockaddr_un addr;
  addr.sun_family = AF_UNIX;
  strncpy(addr.sun_path, SOCK_FILE, sizeof(addr.sun_path));
  addr.sun_path[sizeof(addr.sun_path) - 1] = 0;
  int len = strlen(addr.sun_path) + sizeof(addr.sun_family) + 1;

  int listen_socket = socket(AF_UNIX, SOCK_STREAM, 0);
  if (listen_socket == -1) {
    perror("create socket error.\n");
    return -1;
  }
  // fcntl (socket, F_SETFL,SOCK_NONBLOCK) ;
  int on = 1;
  /* set reuse option */
  int ret = setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&on,
                       sizeof(on));
  unlink(SOCK_FILE);
  /* bind socket */
  ret = bind(listen_socket, (struct sockaddr *)&addr, len);
  if (ret == -1) {
    perror("bind error.\n");
    return -1;
  }
  printf("start to listen\n");
  ret = listen(listen_socket, 1);
  if (ret == -1) {
    perror("listen error\n");
    return -1;
  }
  ret = chmod(SOCK_FILE, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
  if (ret == -1) {
    perror("chmod error\n");
    return -1;
  }
  this->socket = listen_socket;
  this->select_max = listen_socket;
  return 1;
}

static int set_max(unix_socket_infos_t *this) {
  for (int i = 0; i < CLIENT_SIZE; i++) {
    if (this->clients[i] >= this->select_max) {
      this->select_max = this->clients[i];
    }
    if (this->clients[i] < 0) {
      break;
    }
  }
  fprintf(stderr, "max is:%d\n", this->select_max);
  return 0;
}

static int close_client(unix_socket_infos_t *this, int index) {
  int client = this->clients[index];
  close(client);
  this->clients[index] = -1;
  set_max(this);
}

static int deal_client(unix_socket_infos_t *this, int index) {
  char buffer[1024] = {0};
  int ret = recv(this->clients[index], buffer, sizeof(buffer) - 1, 0);
  if (ret <= 0) {
    if (ret == 0) {
      printf("lost connect.\n");
    } else {
      printf("recv error:%s \n", strerror(errno));
    }
    close_client(this, index);
    return -1;
  }
  if (ret < sizeof(buffer)-2) {
    buffer[ret] = '\n';
    buffer[ret+1] = 0;
  }
  fprintf(stderr, "client[%d]:%s", this->clients[index],buffer);
  ret = send(this->clients[index], buffer, strlen(buffer), MSG_NOSIGNAL);
  if (ret < 0) {
    perror("send error:");
  } else {
    fprintf(stderr, "server:%s", buffer);
  }
  return 1;
}

static int accept_client(unix_socket_infos_t *this) {
  socklen_t len = sizeof(this->client_addr);
  char buffer[1024] = {0};
  int client = accept(this->socket, (struct sockaddr *)&(this->client_addr), &len);
  printf("client to comming:%d\n", client);
  if (client < 0) {
    perror("accept error\n");
    return -1;
  }
  memset(buffer,0x0,1024);
  int ret = recv(client, buffer, sizeof(buffer) - 1, 0);
  if (ret < 0) {
    perror("recv error\n");
    return -1;
  }
  if (ret < sizeof(buffer)-2) {
    buffer[ret] = '\n';
    buffer[ret+1] = 0;
  }
  fprintf(stderr, "client[%d][first]:%s",client,buffer);
  ret = send(client, buffer, strlen(buffer), MSG_NOSIGNAL);
  if (ret < 0) {
    perror("send error\n");
  } else {
    fprintf(stderr, "server[first]:%s", buffer);
  }
  int is_set = 0;
  for (int i = 0; i < CLIENT_SIZE; i++) {
    if (this->clients[i] < 0) {
      this->clients[i] = client;
      is_set = 1;
      break;
    }
  }
  set_max(this);
  if (is_set == 0) {
    fputs(TOO_MANY, stdout);
    close(client);
    return -1;
  }
  return 1;
}

static int run_select(unix_socket_infos_t *this) {
  struct timeval tv;
  int ret;
  fd_set select_set;
  FD_ZERO(&select_set);
  FD_SET(this->socket, &select_set);

  for (int i = 0; i < CLIENT_SIZE; i++) {
    if (this->clients[i] > 0) {
      FD_SET(this->clients[i], &select_set);
    } else {
      break;
    }
  }

  tv.tv_sec = 0;
  tv.tv_usec = 200 * 1000;
  int select_max = this->select_max + 1;
  ret = select(select_max, &select_set, NULL, NULL, &tv);
  if (ret == -1) {
    if (errno == EINTR) {
      return 1;
    }
    return -1;
  }
  if (ret == 0) {
    return 1;
  }
  if (FD_ISSET(this->socket, &select_set)) {
    accept_client(this);
  }
  for (int i = 0; i < CLIENT_SIZE; i++) {
    if (this->clients[i] <= 0) {
      break;
    }
    if (FD_ISSET(this->clients[i], &select_set)) {
      deal_client(this, i);
    }
  }
  return 1;
}

int main(int argc, char **argv) {
  unix_socket_infos_t unix_socket_infos;
  for (int i = 0; i < CLIENT_SIZE; i++) {
    unix_socket_infos.clients[i] = -1;
  }
  int ret = create_unix_socket(&unix_socket_infos);
  printf("start to loop\n");
  while (1) {
    int run_ret = run_select(&unix_socket_infos);
    if (run_ret == -1) {
      break;
    }
  }
  return 0;
}

客户端连接代码：

#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include <string.h>

#define SOCK_FILE "command.socket"

int main(int argc, char **argv) {
  struct sockaddr_un un;
  int sock_fd;
  char buffer[1024] = "hello unix socket server";
  char recv_buffer[1024];

  un.sun_family = AF_UNIX;
  strcpy(un.sun_path, SOCK_FILE);
  sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
  if (sock_fd < 0) {
    perror("socket error.\n");
    return -1;
  }
  if (connect(sock_fd, (struct sockaddr *)&un, sizeof(un)) < 0) {
    perror("connect error.\n");
    return -1;
  }
  while (1) {
    memset(recv_buffer,0,1024);
    memset(buffer,0,1024);
    fprintf(stderr,"\nmy[%d]:",sock_fd);
    fgets(buffer,sizeof(buffer)-1,stdin);
    if (strncmp(buffer, "quit", 4) == 0) {
      break;
    }
    int ret = send(sock_fd, buffer, strlen(buffer) - 1, 0);
    if (ret == -1) {
      perror("send error.\n");
    } else {
      ret = recv(sock_fd, recv_buffer, sizeof(recv_buffer) - 1, 0);
      if (ret <= 0) {
        perror("recv error.\n");
      }
      recv_buffer[ret - 1] = 0;
      fprintf(stderr,"server:%s",recv_buffer);
    }
  }

  close(sock_fd);
  return 0;
}

练习代码，写的比较挫，客户端通过unix socket 连接到服务器，然后接收用户输入发送给服务器，服务器回送消息，直到用户输入quit退出。
示意下效果：

root@ubuntu-lab:/home/miao/c-test/select# ./a.out 
start to listen
start to loop
client to comming:4
client[4][first]:123
server[first]:123
max is:4
client[4]:456
server:456
client[4]:abc
server:abc
lost connect.
max is:4
.....

客户端显示：

root@ubuntu-lab:/home/miao/c-test/select# ./client 

my[3]:123
server:123
my[3]:456
server:456
my[3]:abc
server:abc
my[3]:quit

3.1.2 core模拟

在main的开始位置加上如下的代码：

  int files[1800] = {0};
  char fileName[256] = {0};
  for (int i = 0; i < 1800; i++) {
    memset(fileName, 0x0, sizeof(fileName));
    sprintf(fileName, "test_%d", i);
    files[i] = open(fileName, O_RDWR | O_CREAT);
    if (files[i] < 0) {
      close(files[i]);
    }
  }

会发现，程序会自动退出或core，偶尔也有成功的情况，还有的情况是发送到的命令没回复，也就是没监听起来。

3.2 select 缺点

虽然select也支持了IO多路复用，但是存在以下问题：

每次select返回后，监视的集合需要重新设置，比较麻烦。
限制1024个连接，如果想在应用上突破连接，采用malloc等动态申请内存方式也是可以，但是最好采用poll或epoll。
每次都要将监视的文件描述符复制到内核空间，有事件的发生的时候，需要再从内核空间复制到用户空间，比较占用cpu资源，
几种机制的性能比较如下

image.png

未完待续。。。

从select引起的bug聊聊多路复用一

一前言

二问题描述和排查步骤

三多路复用

3.1 select 机制实现多路复用

3.1.1 基本使用说明

3.1.2 core模拟

3.2 select 缺点

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

从select引起的bug聊聊多路复用一

一 前言

二 问题描述和排查步骤

三 多路复用

3.1 select 机制实现多路复用

3.1.1 基本使用说明

3.1.2 core模拟

3.2 select 缺点

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

一前言

二问题描述和排查步骤

三多路复用