LT(水平触发)和ET(边缘触发)

100 阅读 0 评论 66 点赞

我是靠谱客的博主干净牛排，这篇文章主要介绍LT(水平触发)和ET(边缘触发)，现在分享给大家，希望可以做个参考。

以下内容均为本人学习笔记，若有不当，感谢指出
上一篇学习了epoll 的基本概念和使用：epoll
今天学习了解epoll 的两种工作模式：水平触发和 边缘触发

一、LT模式(Level Triggered)

epoll默认为LT工作模式，所以event.events选项中EPOLLET而没有LT

工作模式如下：
- 若缓冲区中有10k数据，我们可以多次进行读取
- 比如先读取2k数据，再次调用epoll_wait()，并且会立刻通知socket读事件就绪，可以再次读取剩余的数据
- 再次调用epoll_wait

一、ET模式（Edge Triggered）

ET工作模式即我们将添加进epoll描述符时候使用EPOLLET标志，epoll进入工作模式
工作模式如下：
- 若缓冲区中有10k数据，第一次只读取了1k
- 再次调用epoll_wait，已经不是就绪状态了，
- ET模式下，只有当缓冲区中数据由无到有，由少变多时才会进行读取数据
- 支持阻塞和非阻塞的读写

ET模式带来的问题
1. 因为只有当缓冲区中数据由无到有，由少变多时才会区读取数据，
所以一次要将缓冲区中的数据读完，否则剩下的数据可能就读不到了。
正常的读取数据时，我们若是要保证一次把缓冲区的数据读完，意为本次读被阻塞时即缓冲区中没有数据了，可是我们 epoll 服务器要处理多个用户的请求，read()不能被阻塞，所以采用非阻塞轮询的方式读取数据。

2.若轮询的将数据读完，对方给我们发9.5k的数据，我们采取每次读取1k的方式进行轮询读取，在读完9k的时候，下一次我们读到的数据为0.5k，我们就知道缓冲区中数据已经读完了就停止本次轮询。
但还有一种情况，对方给我们发的数据为10k,我们采取每次读取1k的方式轮询的读取数据，当我们已经读取了10k的时候，并不知道有没有数据了，我们仍旧还要尝试读取数据，这时read()就被阻塞了。

结论：epoll在ET模式下必须以非阻塞轮询的方式进行读取数据

三、epoll应用场景

适合用epoll的应用场景：对于连接特别多，活跃的连接特别少，这种情况等的时间特别久，典型的应用场景为一个需要处理上万的连接服务器，例如各种app的入口服务器，例如qq

不适合epoll的场景：连接比较少，数据量比较大，例如ssh

epoll 的惊群问题：因为epoll 多用于多个连接，只有少数活跃的场景，但是万一某一时刻，epoll 等的上千个文件描述符都就绪了，这时候epoll 要进行大量的I/O,此时压力太大。

上一片篇中实现epoll的LT模式

四、epoll的ET模式举例

注意文件描述符设为非阻塞，加入epoll描述符中时，加EPOLLET标志

复制代码

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>

#include <sys/epoll.h>
#include <fcntl.h>

//使用epoll实现多路复用

//将文件描述符设置为非阻塞
void SetNoBlock(int fd)
{
  int flag =  fcntl(fd,F_GETFL);
  fcntl(fd,F_SETFL, flag | O_NONBLOCK);
}
//启动服务器
int server_start(const char * ip,const short port)
{
  int sock = socket(AF_INET,SOCK_STREAM,0);
  if(sock < 0)
  {
    perror("socket");
    return -1;
  }

sockaddr_in addr;
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = inet_addr(ip);
  addr.sin_port = htons(port);

int ret = bind(sock,(sockaddr *)&addr,sizeof(addr));
  if(ret < 0)
  {
    perror("bind");
    return -1; 
  }

ret = listen(sock,5);
  if(ret < 0)
  {
    perror("listen");
    return -1;
  }
  return sock;
}

//处理连接socket就绪
//封装处理非阻塞轮询accept()
void Process_listen_socket(int epfd,int listen_socket,sockaddr_in  * peer,socklen_t peer_len)
{
  //就进行非阻塞式轮询accept()

//我们看到这里返回的events数据中并没有文件描述符
  //所以我们最开始将data.fd赋值为文件描述符就是在这里用到
  while(1)
  {
    int new_socket = accept(listen_socket,(sockaddr *)peer,&peer_len);

if(new_socket < 0 && errno ==EAGAIN)
    {
      //因为将文件描述符都设置为非阻塞，这里的accpt()为非阻塞的
      //这里就需要轮询式的进行accpet()
      //这里的listen()函数的第二个参数为5，表示排队等待连接的客户端最多有5个
      //说明已经将所有的文件描述符进行accpet
      perror("accpet");
      return;
    }
    //如果创建new_socket成功之后
    //就将new_socket 加入到epfd中，让epoll_wait()再去监视new_socket的状态

//将文件描述符设置为非阻塞
    SetNoBlock(new_socket);

epoll_event event ;
    event.data.fd = new_socket;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl(epfd,EPOLL_CTL_ADD,new_socket,&event);

}
}

//封装Read,实现非阻塞的轮询式的read()
ssize_t Read(int sock,char * buf,ssize_t max_size)
{
  if(buf == NULL || max_size <= 0)
  {
    return -1;
  }
  ssize_t total_size = 0;
  while(total_size < max_size)
  {
    //控制每次读取都不能使buf越界
    //剩余空间如果大于1024，就一次性读取1024，否则就读取剩余的空间大小
    int len = (max_size - total_size) > 1024 ? 1024 :(max_size - total_size);
    ssize_t read_size = read(sock,buf+total_size,len);
    if(read_size < 0 && errno == EAGAIN)
    {
      //这里的read()为非阻塞的
      //说明为缓冲区中没有数据资源
      //非阻塞轮询就结束了
      printf("data not readyn");
      break;
    }

//如果这里为读取失败，read_size < 0 errno != EAGAIN
    //让其再次尝试读取

if(read_size == 0)
    {
      printf("rad donen");
      break;
    }
    else
    {
      //正常读取的情况
      //修改total_size
      total_size += read_size; 
    }
  }
  return total_size;
}

//处理连接上socket,进行读取数据
void Process_accept_socket(int epfd, int acc_socket,sockaddr_in * peer)
{
  char buf[1024 * 10] = {0};
  //非阻塞轮询进行读取
  ssize_t read_size = Read(acc_socket,buf,sizeof(buf)-1);

if(read_size <= 0 )
  {
    close(acc_socket);
    epoll_ctl(epfd,EPOLL_CTL_DEL,acc_socket,NULL);
    printf("client[%s:%d] disconnect!n",inet_ntoa(peer->sin_addr),peer->sin_port);
    return;
  }
  else
  {
    //正常读取的情况
    buf[read_size] = '';
    printf("[client %s:%d.%d]say:%s",inet_ntoa(peer->sin_addr),ntohs(peer->sin_port),acc_socket,buf);
    //回显服务
    //将收到的内容发送给服务器
    write(acc_socket,buf,strlen(buf));
  }
}

//主函数
int main(int argc,char * argv[])
{
  //检验命令行参数是否正确
  if(argc != 3)
  {
    printf("Usage : ./server ip portn");
    return 1;
  }

//一、启动服务器
  int listen_socket = server_start(argv[1],atoi(argv[2]));

if(listen_socket < 0)
  {
    printf("server start failedn");
    return 2;
  }
  //将文件描述符设置为非阻塞
  SetNoBlock(listen_socket);

printf("server start okn");

//1.创建epoll对象
  int epfd = epoll_create(256);
  if(epfd < 0)
  {
    perror("epoll_create");
    return 3;
  }
  epoll_event event;
  event.data.fd = listen_socket;//此处将要监听的文件描述符放在data中，后面取值要用到
  //event.events = EPOLLIN;//关心的事件为读事件
  event.events = EPOLLIN | EPOLLET;//关心的事件为读，并且设置为边缘触发
  //2.将listen_socket添加到epfd中
  epoll_ctl(epfd,EPOLL_CTL_ADD,listen_socket,&event);

//二、进行事件循环
  while(true)
  {
    //创建事件数组，返回已经就绪的文件
    epoll_event events[2];
    int size = epoll_wait(epfd,events,sizeof(events)/sizeof(events[0]),-1);
    if(size < 0)
    {
      perror("epoll_wait");
      continue;
    }

if(size == 0)
    {
      printf("time outn");
      continue;
    }

//epoll_wait()成功返回后
    int i = 0;
    for(i = 0; i < size ;++i)
    {
      sockaddr_in peer;
      socklen_t peer_len =  sizeof(peer);
      if(events[i].data.fd == listen_socket)
      {
      //(a)listen_socket就绪
      //连接事件就绪
        Process_listen_socket(epfd,listen_socket,&peer,peer_len);
        continue;
      }//end if(events[i].data.fd == listen_socket)
      else
      {
     //(b)new_sock就绪
     //读事件就绪
        Process_accept_socket(epfd,events[i].data.fd,&peer);
      }//end else
    }//end for()
  }//end while(true)

close(epfd);
  close(listen_socket);

}// end main

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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>

#include <sys/epoll.h>
#include <fcntl.h>


//使用epoll实现多路复用


//将文件描述符设置为非阻塞
void SetNoBlock(int fd)
{
  int flag =  fcntl(fd,F_GETFL);
  fcntl(fd,F_SETFL, flag | O_NONBLOCK);
}
//启动服务器
int server_start(const char * ip,const short port)
{
  int sock = socket(AF_INET,SOCK_STREAM,0);
  if(sock < 0)
  {
    perror("socket");
    return -1;
  }

  sockaddr_in addr;
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = inet_addr(ip);
  addr.sin_port = htons(port);

  int ret = bind(sock,(sockaddr *)&addr,sizeof(addr));
  if(ret < 0)
  {
    perror("bind");
    return -1; 
  }

  ret = listen(sock,5);
  if(ret < 0)
  {
    perror("listen");
    return -1;
  }
  return sock;
}


//处理连接socket就绪
//封装处理非阻塞轮询accept()
void Process_listen_socket(int epfd,int listen_socket,sockaddr_in  * peer,socklen_t peer_len)
{
  //就进行非阻塞式轮询accept()

  //我们看到这里返回的events数据中并没有文件描述符
  //所以我们最开始将data.fd赋值为文件描述符就是在这里用到
  while(1)
  {
    int new_socket = accept(listen_socket,(sockaddr *)peer,&peer_len);

    if(new_socket < 0 && errno ==EAGAIN)
    {
      //因为将文件描述符都设置为非阻塞，这里的accpt()为非阻塞的
      //这里就需要轮询式的进行accpet()
      //这里的listen()函数的第二个参数为5，表示排队等待连接的客户端最多有5个
      //说明已经将所有的文件描述符进行accpet
      perror("accpet");
      return;
    }
    //如果创建new_socket成功之后
    //就将new_socket 加入到epfd中，让epoll_wait()再去监视new_socket的状态

    //将文件描述符设置为非阻塞
    SetNoBlock(new_socket);

    epoll_event event ;
    event.data.fd = new_socket;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl(epfd,EPOLL_CTL_ADD,new_socket,&event); 

  }
}


//封装Read,实现非阻塞的轮询式的read()
ssize_t Read(int sock,char * buf,ssize_t max_size)
{
  if(buf == NULL || max_size <= 0)
  {
    return -1;
  }
  ssize_t total_size = 0;
  while(total_size < max_size)
  {
    //控制每次读取都不能使buf越界
    //剩余空间如果大于1024，就一次性读取1024，否则就读取剩余的空间大小
    int len = (max_size - total_size) > 1024 ? 1024 :(max_size - total_size);
    ssize_t read_size = read(sock,buf+total_size,len);
    if(read_size < 0 && errno == EAGAIN)
    {
      //这里的read()为非阻塞的
      //说明为缓冲区中没有数据资源
      //非阻塞轮询就结束了
      printf("data not readyn");
      break;
    }

    //如果这里为读取失败，read_size < 0 errno != EAGAIN
    //让其再次尝试读取

    if(read_size == 0)
    {
      printf("rad donen");
      break;
    }
    else
    {
      //正常读取的情况
      //修改total_size
      total_size += read_size; 
    }
  }
  return total_size;
}


//处理连接上socket,进行读取数据
void Process_accept_socket(int epfd, int acc_socket,sockaddr_in * peer)
{
  char buf[1024 * 10] = {0};
  //非阻塞轮询进行读取
  ssize_t read_size = Read(acc_socket,buf,sizeof(buf)-1); 

  if(read_size <= 0 )
  {
    close(acc_socket);
    epoll_ctl(epfd,EPOLL_CTL_DEL,acc_socket,NULL);
    printf("client[%s:%d] disconnect!n",inet_ntoa(peer->sin_addr),peer->sin_port);
    return;
  }
  else
  {
    //正常读取的情况
    buf[read_size] = '';
    printf("[client %s:%d.%d]say:%s",inet_ntoa(peer->sin_addr),ntohs(peer->sin_port),acc_socket,buf);
    //回显服务
    //将收到的内容发送给服务器
    write(acc_socket,buf,strlen(buf));
  }
}

//主函数
int main(int argc,char * argv[])
{
  //检验命令行参数是否正确
  if(argc != 3)
  {
    printf("Usage : ./server ip portn");
    return 1;
  }

  //一、启动服务器
  int listen_socket = server_start(argv[1],atoi(argv[2]));

  if(listen_socket < 0)
  {
    printf("server start failedn");
    return 2;
  }
  //将文件描述符设置为非阻塞
  SetNoBlock(listen_socket);

  printf("server start okn");

  //1.创建epoll对象
  int epfd = epoll_create(256);
  if(epfd < 0)
  {
    perror("epoll_create");
    return 3;
  }
  epoll_event event;
  event.data.fd = listen_socket;//此处将要监听的文件描述符放在data中，后面取值要用到
  //event.events = EPOLLIN;//关心的事件为读事件
  event.events = EPOLLIN | EPOLLET;//关心的事件为读，并且设置为边缘触发
  //2.将listen_socket添加到epfd中
  epoll_ctl(epfd,EPOLL_CTL_ADD,listen_socket,&event);


  //二、进行事件循环
  while(true)
  {
    //创建事件数组，返回已经就绪的文件
    epoll_event events[2];
    int size = epoll_wait(epfd,events,sizeof(events)/sizeof(events[0]),-1);
    if(size < 0)
    {
      perror("epoll_wait");
      continue;
    }

    if(size == 0)
    {
      printf("time outn");
      continue;
    }

    //epoll_wait()成功返回后
    int i = 0;
    for(i = 0; i < size ;++i)
    {
      sockaddr_in peer;
      socklen_t peer_len =  sizeof(peer);
      if(events[i].data.fd == listen_socket)
      {
      //(a)listen_socket就绪
      //连接事件就绪
        Process_listen_socket(epfd,listen_socket,&peer,peer_len);
        continue;
      }//end if(events[i].data.fd == listen_socket)
      else
      {
     //(b)new_sock就绪
     //读事件就绪
        Process_accept_socket(epfd,events[i].data.fd,&peer);
      }//end else
    }//end for()
  }//end while(true)

  close(epfd);
  close(listen_socket);

}// end main