阻塞IO

服务器一般为了处理客户端代码，有以下操作

socket()   //打开一个网络通信端口               
bind()     //绑定                 
listen()   //监听            
accept()   //阻塞建立连接        
recv()     //阻塞读取数据                   
send()     //给客户端回放数据
close()    //关闭连接
1
2
3
4
5
6
7

客户端则需要以下操作

socket()  //打开一个网络通信端口
connect() //与客户端进行连接
send()    //给客户端发送数据
recv()    //从客户端那里读取数据
close()   //关闭连接
1
2
3
4
5

这个时候，服务端的线程往往阻塞在两个地方

一个是accpet函数

等到有客户端来连接才会解除阻塞

一个是read函数

read函数会一直阻塞，直到数据从网卡到达内核缓冲区，接着文件描述符准备就绪，接着解除阻塞，把数据从内核缓冲区读到数据缓冲区

所以当连接的客户端一直不发数据，那么服务端线程就会一直阻塞在read函数，无法接受其它用户连接。

有一种解决办法，就是我们每次可以创建一个新的线程或进程，去调用read函数，并做业务处理，让主函数中的执行路径可以一直执行下去，而不被阻塞。

多线程

#include
#include
#include
#include
#include
#include
#include
#include
#include
int sock_init()
{
  int socketfd = socket(AF_INET,SOCK_STREAM,0);
  if(socketfd == -1)
  {
   printf("socket erro\n");
   exit(1);
  }
  struct sockaddr_in saddr,caddr;
  memset(&saddr,0,sizeof(saddr));
  saddr.sin_family = AF_INET;
  saddr.sin_port=htons(6000);
  saddr.sin_addr.s_addr=inet_addr("127.0.0.1");
  int res = bind(socketfd,(struct sockaddr *)&saddr,sizeof(saddr));
  if(res == -1)
  {
   printf("bind erro \n");
   exit(1);
  }
  res = listen(socketfd,5);
  if(res == -1)
  {
    printf("listen erro\n");
    exit(1);
  }
 return socketfd;

}
void * fun(void *);
int main()
{
 int socketfd = sock_init();
 while(1)
 {

   struct sockaddr_in caddr;
   int len = sizeof(caddr);
   int c = accept(socketfd,(struct sockaddr *)&caddr,&len);
   if(c<=0)
   {
     printf("accept erro\n");
     break;
   
   }
   printf("c=%d,ip=%s,port=%d",c,inet_ntoa(caddr.sin_addr),ntohs(caddr.sin_port));
   pthread_t id;
   pthread_create(&id,NULL,fun,(void *)c);
 
   }
}
void * fun(void * arg)
{
 int c = (int)arg;
  while(1)
  {
   char buff[128]={0};
   int n = recv(c,buff,127,0);
   if(n<=0)
   {
    break;
   
   }
   printf("recive len = %d %s\n",n,buff);
   send(c,"ok",2,0);
  }
  printf("c(%d)close\n",c);
  close(c);

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

多进程

#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
void fun(int sig)
{
  wait(NULL);

}
int sock_init()
{
  int socketfd = socket(AF_INET,SOCK_STREAM,0);
  if(socketfd == -1)
  {
   printf("socket erro\n");
   exit(1);
  }
  struct sockaddr_in saddr,caddr;
  memset(&saddr,0,sizeof(saddr));
  saddr.sin_family = AF_INET;
  saddr.sin_port=htons(6000);
  saddr.sin_addr.s_addr=inet_addr("127.0.0.1");
  int res = bind(socketfd,(struct sockaddr *)&saddr,sizeof(saddr));
  if(res == -1)
  {
   printf("bind erro \n");
   exit(1);
  }
  res = listen(socketfd,5);
  if(res == -1)
  {
    printf("listen erro\n");
    exit(1);
  }
 return socketfd;

}
int main()
{
 signal(SIGCHLD,fun);
 int socketfd = sock_init();
 while(1)
 {

   struct sockaddr_in caddr;
   int len = sizeof(caddr);
   int c = accept(socketfd,(struct sockaddr *)&caddr,&len);
   if(c<=0)
   {
     printf("accept erro\n");
     break;
   
   }
   printf("c=%d,ip=%s,port=%d\n",c,inet_ntoa(caddr.sin_addr),ntohs(caddr.sin_port));
   pid_t pid = fork();
  if(pid ==0)
  {
  
    while(1)
   {
     char buff[128]={0};
     int n = recv(c,buff,127,0);
     if(n<=0)
    {
      break;
   
    }
     printf("c(%d) recive len = %d %s\n",c,n,buff);
     send(c,"ok",2,0);
  
   } 
  printf("c(%d)close\n",c);
  close(c);
  }
 
 }
  exit(0);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

非阻塞IO

我们需要操作系统为我们提供一个非阻塞的read函数

这个read函数的效果是，如果没有数据到达时（到达网卡并到达内核缓冲区），立刻返回一个错误值 -1 ，而不是阻塞地等待。

操作系统提供了这样的功能，只需要在调用read前，将文件描述符设置为非阻塞即可。

fcntl(c,F_SETFL,O_NONBLOCK);
int n = read((c,buffer) != SUCCESS);
1
2

这个时候，我们可以当一个客户端连接后，将这个文件描述符放到一个数组中，然后弄出一个新的线程去不断的遍历这个数组，调用每一个元素的非阻塞read方法

while(1)
{
 for(int fd;fdarry)
 {
   if(read(fd) ！= -1)
   {
   }
 }
}
1
2
3
4
5
6
7
8
9

但是这样每次遍历就要不停的调用read，当某个连接上没有数据时，read返回为-1时，这就白白的执行了一次系统调用，这种极大的浪费了系统资源。

IO多路复用

IO复用方法有 ：
1.select
2. poll
3. epoll（Linux上特有的） Windows上与之对应的方法(iocp)

select

select是系统调用函数，通过它，我们可以把一个文件描述符的数组发给操作系统，让操作系统去遍历，确定那个文件描述符可以读写，然后告诉我们去处理

函数原型

int select(
        int nfds, 
         fd_set * readfds,
         fd_set * wirtefds,
         fd_set * exceptfds,
         struct timeval * timeout);
         /* nfds : 监控文件描述符集李最大文件描述符加1
         *  readfds: 监控有读数据到达文件描述符集合
         *  writefds: 监控些数据到达文件描述符集合
         * exceptfds: 监控异常发送到达文件描述符集合
         * timeout:   定时阻塞监控时间，3种情况
         * 1.NULL,永远等下去
         * 2.设置timeval,等待固定时间
         * 3.设置timevals里时间均为0，检查描述字后立马返回 
         * struct timeval
         * {
         *   long tv_sec;//秒数
         *   long tv_usec;//微妙数
         * };
         */
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

select的返回值：

select在成功时****返回就绪描述符的总数
如果在超时时间内没有任何描述符准备就绪，那么select将返回0，
如果在select期间程序收到信号，则select将返回-1，代表select失败，并设置errno为EINTR。

通过以下描述符可以访问fd_set中的位

      FD_ZERO(fd_set * fdset) //清除fdset中的所有位
      FD_SET(int fd,fd_set * fdset)//设置fdset的位fd
      FD_CLR(int fd,fd_set * fdset)//清除fdset的位fd
      int FD_ISSET(int fd_set * fdset) //检测fdset中的位fd是否被设置
1
2
3
4

fd_set实际是一个为1024位的数据，它可以存放值为0~1023的描述符
当描述符的值为0时，它就把fd_set的第0位的值置为0，当值为1时，就把fd_set的第1位的值置为1。

所以它只能监听1024个文件描述符。

代码实现

#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define MAX 10
void accept_client(int sockfd,int fds[])
{
 struct sockaddr_in caddr;
 int len = sizeof(caddr);
 int  c =accept(sockfd,(struct sockaddr *)&caddr,&len);

 if(c < 0 )
 {
  return ;
 }
 printf("accept c =%d\n",c);
 fds_add(fds,c);
}
void fds_init(int fds[])
{
 
    for(int i=0;i<MAX;++i)
    {
     fds[i]=-1;
    }

}
void fds_add(int fds[],int fd)
{
 for(int i=0;i<MAX;++i)
 {
   if(fds[i] == -1)
   {
    fds[i]=fd;
    break;
   }
 }

}
void fds_del(int fds,int fd)
{
 for(int i=0;i<MAX;++i)
 {
  if(fds [i] == fd)
  {
   fds[i] =-1;
   break;
  }
 
 }
}
int sockinit()
{
  int sockfd = socket(AF_INET,SOCK_STREAM,0);
  if(sockfd == -1)
  {
    printf("sock erro\n");
   return -1;
  }
 struct  sockaddr_in saddr;
 memset(&saddr,0,sizeof(saddr));
 saddr.sin_familm = AF_INET;
 saddr.sin_port = htons(6000);
 saddr.sin_addr.s_addr = inet_addr("127.0.0.1");
 
 int res = bind(sockfd,0,(struct saddr *)&saddr,sizeof(saddr));
 if(res == -1)
 {
  printf("bind erro\n");
  return -1;
 }
 res = listen(sockfd,5);
 if(res  == -1)
 {
  printf("listen erro\n");
  return -1;
 }
 return res;
}
void recv_data(int fd,int fds[])
{
 char buff[128] ={0};
 int num = recv(fd,buff,127,0);
 if(num < 0)
 {
  close(fd);
  fds_del(fds,fd);
  printf("close\n");
  return ;
 }
 printf("buff = %s\n",buff);
 send(fd,"ok",2,0);


}
int main()
{
   int sockfd = sockinit();
   if(sockfd == -1)
   {
     exit(1);
  }
   int fds[MAX];
   fds_init(fds);
   fds_add(fds,sockfd);
   fd_set fdset;
   while(1)
  {
    FD_ZERO(&fdset);//FD_ZERO:将文件描述符集合清空
    int maxfd =-1;
    for(int i=0;i<MAX;++i)
   {
      if(fds[i]==-1)
      {
        continue;
      }
       FD_SET(fds[i],&fdset);//将文件描述符添加到文件描述符集合中
       if(maxfd < fds[i])
      {
        masfd = fds[i];
      }
  
    }
 
  }
  struct timaval tv = {5,0}; //设置定时时间
  int n = select(maxfd+1,&fdset,NULL,NULL,&tv);//select的返回值为就绪的描述符数目
  if(n == -1)
   {
    printf("select erro\n");

   }else if(n ==0)
   {
      printf("time out \n");
   }else
   {
     for(int i=0;i<MAX;++i)
    {
      if(fds[i] == -1)
     {
      continue;
     }
     if(FD_ISSET(fds[i],&fdset))//判断该描述符是否就绪
     {
      if(fds[i] == sockfd)
      {
       accept_client(sockfd,fds);
      }else
      {
        recv_data(fds[i].fds);
      }
     
     }
   }
 
   }

   return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

注意：
1.select调用需要传入fd数组，而这个数组会被拷贝到内核空间，当高并发下这样会消耗大量的资源
2. select在内核层依然是通过遍历方式检查文件描述符的就绪状态，是同步过程，只不过少了系统调用上下文切换的资源开销。
3. select仅仅返回就绪文件符的数目，具体哪个就绪还需要用户自己遍历

select的开销， 一次select + n 次就绪的文件描述符的read 开销

poll

它与select不同是，它没有最多文件描述符的限制

函数原型

int poll 
   (
       struct pollfd * fds,
       nfds_t nfds, 
       int timeout)
       /*
       * fds 指定用户感兴趣的文件描述符上发生的可读，可写，异常事件 pollfd的结构体定义如下
       * struct pollfd
       * {
       *   int fd;//文件描述符
       *   short events;//注册的关注事件类型
       *   short revents;//实际发生的事件类型，由内核进行填充。
       * };
       * nfds  指定被监听事件集合fds的大小
       * timeout指定poll的超时值，单位是毫秒
       * timeout为-1时，poll调用将永久阻塞，直到某个事件发生
       * timeout为0时，poll调用将立即返回
       * 
       * 
       */
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

事件类型

代码实现

#include
#include
#include
#include
#include
#include
#include
#include
#include
#include

#define MAX 10

void evs_init(struct pollfd evs[])
{
 for(int i=0;i<MAX;++i)
 {
     evs[i].fd =-1;
     evs[i].events =0;
     evs[i].revents =0; 
 }
}
void evs_add(struct pollfd evs [],int fd)
{
 for(int i=0;i<MAX;++i)
 {
  if(evs[i].fd == -1)
  {
   evs[i].fd = fd;
   evs[i].events = POLLIN;
   evs[i].revents = 0;
   break;
  }
 
 }
}
void evs_del(struct pollfd evs[], int fd)
{
  for(int i=0;i<MAX;++i)
  {
   if(evs[i].fd == fd)
   {
    evs[i].fd =-1;
    evs[i].events  = 0;
    evs[i].revents = 0;
    break;
   }
  
  
  }
}
void accept_client(int sockfd,struct pollfd evs[])
{
    struct sockaddr_in caddr;
    int len = sizeof(caddr);
    int c = accept(sockfd,(struct sockaddr *)&caddr,&len);
    if(c <0)
    {
     printf("accept erro\n");
     return ;
    }
    printf("accept %d\n",c);
    evs_add(evs,c);


}
void recv_data(int c,struct pollfd evs[])
{
 char buff[128]={0};
 int num = recv(c,buff,127,0);
 if(num <0)
 {
  evs_del(evs,c);
  close(c);
  printf("%d close\n",c);
  return ;

 }
 printf("data = %s\n",buff);
 send(c,"ok",2,0);

}
int socket_init();
int main()
{
    int sockfd = socket_init();
    if(sockfd == -1)
    {
      printf("socketfd init erro\n");
      exit(1);
    }
    struct pollfd evs[MAX];
    evs_init(evs);

    evs_add(evs,sockfd);
    
    while(1)
    {
      int n = poll(evs,MAX,5000);
      if(n == -1)
      {
       printf("poll erro\n");
      }else if(n == 0)
      {
       printf("time out\n");
      }else
      {
        for(int i=0;i<MAX;++i)
       {
         if(evs[i].fd ==-1)
         {
          continue;
         }
         if(evs[i].revents & POLLIN)
         {
          if(evs[i].fd == sockfd)
          {
        
            accept_client(sockfd,evs);
          }else
          {
           recv_data(evs[i].fd,evs);
          }
         }
        }
     
       }
    }


    return 0;
}
int socket_init()
{

    int sockfd = socket(AF_INET,SOCK_STREAM,0);
    if(sockfd  == 0)
    {
     printf("sockfd erro\n");
     return -1;
    }
    struct sockaddr_in saddr;
    memset(&saddr,0,sizeof(saddr));
    saddr.sin_family = AF_INET;
    saddr.sin_port = htons(6001);
    saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

    int res = bind(sockfd,(struct sockaddr *)&saddr,sizeof(saddr));
    if(res == -1)
    {
     printf("bind erro\n");
     return -1;
    }
    res = listen(sockfd,5);
    if(res == -1)
    {
     printf("listen erro\n");
     return -1;
    }
    return sockfd;

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

epoll

特点

1.内核中保存一份文件描述符集合，无需用户每次都重新传入，只需告诉内核要修改的部分即可
2.内核不再通过轮询方式找到就绪的文件描述符，而是通过异步IO事件唤醒
3.内核仅会将有IO事件的文件描述符返回给用户，用户无需遍历整个文件描述符集合
 4.与select和poll相比它还有边沿模式
1
2
3
4

epoll有EPOLLLT（水平触发）和EPOLLET（边沿触发）两种触发模式
EPOLLLT和select和poll的触发方式一样，都是水平触发，即只要内核缓冲区有数据，就会让该连接的文件描述符有绪
而对应边沿触发来说，当数据从网卡到达内核缓冲区后，它只提示一次，所以在EPOLLET模式下，当read进行读取数据时，一定要把所有数据一次性读完。

epoll的三个函数各自的原型

int epoll(int size);
/*
*  epoll成功返回内核事件表的文件描述符，失败返回-1
* size参数并不起作用，只是给内核一个提示，告诉它事件表需要多大
*/
1
2
3
4
5

int epoll_ctl
(  
    int epfd,
    int op , 
    int fd , 
    struct epoll_event * event);
    /*
    * epoll_ctl 成功返回0 ，失败返回-1
    * epfd 参数指定要操作的内核事件表的文件描述符
    * fd 参数指定要操作的文件描述符
    * op 指定要操作的数据类型：
    * EPOLL_CTL_ADD  往内核事件表中注册fd上的事件
    * EPOLL_CTL_MOD  修改fd上面的注册事件
    * EPOLL_CTL_DEL  删除fd上面的注册事件
    * event参数指定事件，它是epoll_event结构指针类型，epoll_event的定义如下
    * struct epoll_event
    * {
    *   _uint32_t events;//epoll事件
    *   epoll_data_t data;//用户数据
    * };
    *  events成员描述事件类型,epoll支持的事件类与poll基本相同，表示epoll事件的宏是在poll对应的宏前       
    * 面加‘E’,它有两个额外的事件EPOLLET和EPOLLONESHOT
    * struct epoll_evnet中的data是一个联合体，用于存储用户数据
    *  typedef union epoll_data
    * {
    *   void *ptr;
    *   int fd; //指定事件所属的目标文件描述符
    *   uint32_t u32;
    *   uint64_t u64; 
    * }
    */
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

int epoll_wait
(   
   int epfd,
   struct epoll_event *events,
   int maxevents,
   int timeout);
   /* 
   * epoll_wait返回就绪文件描述符个数，失败返回-1，超时返回0
   *  epfd参数指定要操作的内核事件表的文件描述符
   * events是一个用户数组，在eopll_wait返回时保存内核检测到的所有就绪事件
   * maxevents 参数指定用户数组大小，即最多监听多少个事件，必须大于0
   * timeout指定超时时间，单位为毫秒,
   * timeout为0，则epoll_wait会立即返回
   * timeout为-1，则epoll_wait会一直阻塞，直到有事件就绪
   */
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

代码实现

#include
#include
#include
#include
#include
#include
#include
#include
#include
#include

#define MAX 10
void epoll_add(int epfd,int fd)
{
 struct epoll_event ev;
 ev.data.fd =fd;
 ev.events = EPOLLIN;
 if(epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev) == -1)
 {
  printf("epoll_ctl erro\n");
 
 }
}
void epoll_del(int epfd,int fd)
{
  if(epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL) == -1)
  {
   printf("epoll ctl del erro\n");
  
  }
}
void accept_client(int sockfd,int epfd)
{
    struct sockaddr_in caddr;
    int len = sizeof(caddr);
    int c = accept(sockfd,(struct sockaddr *)&caddr,&len);
    if(c <0)
    {
     printf("accept erro\n");
     return ;
    }
   epoll_add(epfd,c);

}
void recv_data(int c,int epfd)
{
  char buff[128] ={0};
  int num = recv(c,buff,127,0);
  if(num <= 0)
  {
   epoll_del(epfd,c);
   close(c);
   printf("%d close\n",c);
   return ;
  
  }
  printf("(%d),buff=%s\n",c,buff);
  send(c,"ok",2,0);

}
int socket_init();
int main()
{
    int sockfd = socket_init();
    if(sockfd == -1)
    {
      printf("socketfd init erro\n");
      exit(1);
    }
    int epfd = epoll_create(MAX);
    if(epfd == -1)
    {
     printf("epoll_create erro\n");
     close(sockfd);
     exit(1);
    }

    epoll_add(epfd,sockfd);
    struct epoll_event evs[MAX];

    while(1)
    {
     int n = epoll_wait(epfd,evs,MAX,5000);
     if(n == -1)
     {
      printf("epoll erro\n");
     }else if(n == 0)
     {
      printf("time out\n");
     }else
     {
      for(int i =0;i<n;++i)
      {
        int fd = evs[i].data.fd;
       if(evs[i].events & EPOLLIN)
       {
        if(fd == sockfd)
        {
         accept_client(fd,epfd);
        }else
        {
        
            recv_data(fd,epfd);
        
        }
       }
      
      }
     
     }
    }

    return 0;
}
int socket_init()
{

    int sockfd = socket(AF_INET,SOCK_STREAM,0);
    if(sockfd  == 0)
    {
     printf("sockfd erro\n");
     return -1;
    }
    struct sockaddr_in saddr;
    memset(&saddr,0,sizeof(saddr));
    saddr.sin_family = AF_INET;
    saddr.sin_port = htons(6001);
    saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

    int res = bind(sockfd,(struct sockaddr *)&saddr,sizeof(saddr));
    if(res == -1)
    {
     printf("bind erro\n");
     return -1;
    }
    res = listen(sockfd,5);
    if(res == -1)
    {
     printf("listen erro\n");
     return -1;
    }
    return sockfd;

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145