IO复用epoll

发表于 | 分类于 | 阅读次数: |

epoll:Linux系统独有的

select:fd_set;

poll:struct pollfd fds[];

select和poll都是用户态存在，每一次调用select或者poll，都会存在两次数据拷贝，一次是在调用时，一次是在返回时。用户程序检索就绪事件的时间复杂度为O(n)。

epoll:用户程序检索就绪事件的时间复杂度为O(1)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//创建内核事件表（用户关注的所有文件描述符以及关注的事件类型）
//成功返回内核事件表的标识符，失败返回-1
int epoll_creat(int size);

//管理内核事件表：添加 删除 修改通过op这个值控制
//仅仅在调用ctl时拷贝一次
int epoll_ctl(int epfd,int op,int fd,struct epoll_event *event);
op:EPOLL_CTL_ADD EPOLL_CTL_SEL EPOLL_CTL_MOD三个宏
struct EPOLL_EVENT
{
	short events;//事件类型：在每一个poll的事件类型标识前加‘E’
    union epoll_data_t data; //-->fd文件描述符
};

//成功返回就绪的个数 失败返回-1 超时返回0
int epoll_wait(int epfd,struct epoll_event events[],int maxevents,int timeout);
epfd:监听的内核事件表
events:是一个用户数组，events的数据是内核在epoll_wait返回时填充的（有事件就绪的文件描述符和就绪的事件类型）
maxevents:数组的长度->一次epoll_wait最多返回多少个就绪的文件描述符

客户端

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
#define _GNU_SOURCE
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<assert.h>

#include<sys/types.h>
#include<sys/socket.h>
#include<sys/epoll.h>
#include<arpa/inet.h>
#include<netinet/in.h>
#define maxevents 100

int initsocket()
{
    int sockfd=socket(AF_INET,SOCK_STREAM,0);
    if(sockfd==-1) 
        return -1;
    struct sockaddr_in ser;
    memset(&ser,0,sizeof(ser));
    ser.sin_family=AF_INET;
    ser.sin_port=htons(6000);
    ser.sin_addr.s_addr=inet_addr("127.0.0.1");

    int res=bind(sockfd,(struct sockaddr*)&ser,sizeof(ser));
    if(res==-1)
        return -1;
    
    res=listen(sockfd,5);
    if(res==-1)
        return -1;
    return sockfd;
}
//获取新的客户端连接，并且将其添加到epfd指定的内核事件表中
void getnewclilink(int sockfd,int epfd)
{
    struct sockaddr_in cli;
    socklen_t len=sizeof(cli);
    int c=accept(sockfd,(struct sockaddr*)&cli,&len);
    if(c<0)
        return ;
    printf("one client link success\n");
    struct epoll_event event;
    event.events=EPOLLIN|EPOLLRDHUP;
    event.data.fd=c;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,c,&event);
    assert(res!=-1);
}
void dealclidata(int fd)
{
    char buff[128]={0};
    int n=recv(fd,buff,127,0);
    if(n<=0)
    {
        printf("%d recv error\n",fd);
        return ;
    }
    printf("%d: %s\n",fd,buff);
    send(fd,"OK",2,0);
}
void dealfinishevents(struct epoll_event *events,int n,int sockfd,int epfd)
{
    int i=0;
    for(;i<n;++i)
    {
        int fd=events[i].data.fd;
        if(fd==sockfd)
        {
            getnewclilink(sockfd,epfd);
        }
        else
        {
            if(events[i].events&EPOLLRDHUP)
            {
                close(fd);
                epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);//删除内核事件表中的fd
            }
            else
            {
                dealclidata(fd);
            }
        }
    }
}
int main()
{
    int sockfd=initsocket();
    assert(sockfd!=-1);

    int epfd=epoll_create(5);
    assert(epfd!=-1);

    struct epoll_event event;
    event.events=EPOLLIN;
    event.data.fd=sockfd;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&event);
    assert(res!=-1);

    while(1)
    {
        struct epoll_event events[maxevents];
        int n=epoll_wait(epfd,events,maxevents,-1);
        if(n<=0)
        {
            printf("epoll_wait error\n");
            continue;
        }
        dealfinishevents(events,n,sockfd,epfd);
    }
    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
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<assert.h>

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

int main()
{
    int sockfd=socket(AF_INET,SOCK_STREAM,0);
    assert(sockfd!=-1);

    struct sockaddr_in ser;
    memset(&ser,0,sizeof(ser));
    ser.sin_family=AF_INET;
    ser.sin_port=htons(6000);
    ser.sin_addr.s_addr=inet_addr("127.0.0.1");

    int res=connect(sockfd,(struct sockaddr*)&ser,sizeof(ser));
    assert(res!=-1);

    while(1)
    {
        printf("please input: ");
        char buff[128]={0};
        fgets(buff,127,stdin);
        if(strncmp(buff,"end",3)==0)
            break;
            
        send(sockfd,buff,strlen(buff)-1,0);

        char result[128]={0};
        int n=recv(sockfd,result,127,0);
        if(n<=0)
            break;
        printf("result: %s\n",result);
    }
    close(sockfd);
}

通信测试

不知道为啥在deepin系统下跑不出来，在centos下可以跑出来

LT和ET模式

LT模式是默认的工作模式，也就是上面的代码，但是ET模式是epoll的高效工作模式。对于ET模式的文件描述符，当epoll_wait检测到有事件发生时并将此事件通知应用程序后，应用程序必须立即处理该事件，因为后续的epoll_wait调用将不再向应用程序通知这一事件。所以ET模式很大程度降低了同一个epoll事件被重复触发的次数，因此ET模式比LT模式工作效率高。

LT代码

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
#define _GNU_SOURCE
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<assert.h>

#include<sys/types.h>
#include<sys/socket.h>
#include<sys/epoll.h>
#include<arpa/inet.h>
#include<netinet/in.h>
#define maxevents 100

int initsocket()
{
    int sockfd=socket(AF_INET,SOCK_STREAM,0);
    if(sockfd==-1) 
        return -1;
    struct sockaddr_in ser;
    memset(&ser,0,sizeof(ser));
    ser.sin_family=AF_INET;
    ser.sin_port=htons(6000);
    ser.sin_addr.s_addr=inet_addr("127.0.0.1");

    int res=bind(sockfd,(struct sockaddr*)&ser,sizeof(ser));
    if(res==-1)
        return -1;
    
    res=listen(sockfd,5);
    if(res==-1)
        return -1;
    return sockfd;
}
//获取新的客户端连接，并且将其添加到epfd指定的内核事件表中
void getnewclilink(int sockfd,int epfd)
{
    struct sockaddr_in cli;
    socklen_t len=sizeof(cli);
    int c=accept(sockfd,(struct sockaddr*)&cli,&len);
    if(c<0)
        return ;
    printf("one client link success\n");
    struct epoll_event event;
    event.events=EPOLLIN|EPOLLRDHUP;
    event.data.fd=c;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,c,&event);
    assert(res!=-1);
}
void dealclidata(int fd)
{
    char buff[128]={0};
    int n=recv(fd,buff,1,0);
    if(n<=0)
    {
        printf("%d recv error\n",fd);
        return ;
    }
    printf("%d: %s\n",fd,buff);
    send(fd,"OK",2,0);
}
void dealfinishevents(struct epoll_event *events,int n,int sockfd,int epfd)
{
    int i=0;
    for(;i<n;++i)
    {
        int fd=events[i].data.fd;
        if(fd==sockfd)
        {
            getnewclilink(sockfd,epfd);
        }
        else
        {
            if(events[i].events&EPOLLRDHUP)
            {
                close(fd);
                epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);//删除内核事件表中的fd
            }
            else
            {
                dealclidata(fd);
            }
        }
    }
}
int main()
{
    int sockfd=initsocket();
    assert(sockfd!=-1);

    int epfd=epoll_create(5);
    assert(epfd!=-1);

    struct epoll_event event;
    event.events=EPOLLIN;
    event.data.fd=sockfd;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&event);
    assert(res!=-1);
    int count=0;
    while(1)
    {
        struct epoll_event events[maxevents];
        int n=epoll_wait(epfd,events,maxevents,-1);
        printf("epoll_wait return %d\n",count++);
        if(n<=0)
        {
            printf("epoll_wait error\n");
            continue;
        }
        dealfinishevents(events,n,sockfd,epfd);
    }
    return 0;
}

ET代码

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
#define _GNU_SOURCE
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<assert.h>

#include<sys/types.h>
#include<sys/socket.h>
#include<sys/epoll.h>
#include<arpa/inet.h>
#include<netinet/in.h>
#define maxevents 100

int initsocket()
{
    int sockfd=socket(AF_INET,SOCK_STREAM,0);
    if(sockfd==-1) 
        return -1;
    struct sockaddr_in ser;
    memset(&ser,0,sizeof(ser));
    ser.sin_family=AF_INET;
    ser.sin_port=htons(6000);
    ser.sin_addr.s_addr=inet_addr("127.0.0.1");

    int res=bind(sockfd,(struct sockaddr*)&ser,sizeof(ser));
    if(res==-1)
        return -1;
    
    res=listen(sockfd,5);
    if(res==-1)
        return -1;
    return sockfd;
}
//获取新的客户端连接，并且将其添加到epfd指定的内核事件表中
void getnewclilink(int sockfd,int epfd)
{
    struct sockaddr_in cli;
    socklen_t len=sizeof(cli);
    int c=accept(sockfd,(struct sockaddr*)&cli,&len);
    if(c<0)
        return ;
    printf("one client link success\n");
    struct epoll_event event;
    event.events=EPOLLIN|EPOLLRDHUP|EPOLLET;
    event.data.fd=c;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,c,&event);
    assert(res!=-1);
}
void dealclidata(int fd)
{
    char buff[128]={0};
    int n=recv(fd,buff,1,0);
    if(n<=0)
    {
        printf("%d recv error\n",fd);
        return ;
    }
    printf("%d: %s\n",fd,buff);
    send(fd,"OK",2,0);
}
void dealfinishevents(struct epoll_event *events,int n,int sockfd,int epfd)
{
    int i=0;
    for(;i<n;++i)
    {
        int fd=events[i].data.fd;
        if(fd==sockfd)
        {
            getnewclilink(sockfd,epfd);
        }
        else
        {
            if(events[i].events&EPOLLRDHUP)
            {
                close(fd);
                epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);//删除内核事件表中的fd
            }
            else
            {
                dealclidata(fd);
            }
        }
    }
}
int main()
{
    int sockfd=initsocket();
    assert(sockfd!=-1);

    int epfd=epoll_create(5);
    assert(epfd!=-1);

    struct epoll_event event;
    event.events=EPOLLIN;
    event.data.fd=sockfd;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&event);
    assert(res!=-1);
    int count=0;
    while(1)
    {
        struct epoll_event events[maxevents];
        int n=epoll_wait(epfd,events,maxevents,-1);
        printf("epoll_wait return %d\n",count++);
        if(n<=0)
        {
            printf("epoll_wait error\n");
            continue;
        }
        dealfinishevents(events,n,sockfd,epfd);
    }
    return 0;
}

从图中可以看出只收到了一个字母，但是LT却接受了好几次来接受完整个数据。但是ET再次发送时，服务器却接受了上次发送端第二个字母，所以发送的数据全都在接收缓冲区存着。

那么怎么解决这个问题呢

改进的ET代码

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
#define _GNU_SOURCE
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<assert.h>

#include<sys/types.h>
#include<sys/socket.h>
#include<sys/epoll.h>
#include<arpa/inet.h>
#include<netinet/in.h>
#include<fcntl.h>
#include<errno.h>
#define maxevents 100

int initsocket()
{
    int sockfd=socket(AF_INET,SOCK_STREAM,0);
    if(sockfd==-1) 
        return -1;
    struct sockaddr_in ser;
    memset(&ser,0,sizeof(ser));
    ser.sin_family=AF_INET;
    ser.sin_port=htons(6000);
    ser.sin_addr.s_addr=inet_addr("127.0.0.1");

    int res=bind(sockfd,(struct sockaddr*)&ser,sizeof(ser));
    if(res==-1)
        return -1;
    
    res=listen(sockfd,5);
    if(res==-1)
        return -1;
    return sockfd;
}
//获取新的客户端连接，并且将其添加到epfd指定的内核事件表中
void getnewclilink(int sockfd,int epfd)
{
    struct sockaddr_in cli;
    socklen_t len=sizeof(cli);
    int c=accept(sockfd,(struct sockaddr*)&cli,&len);
    if(c<0)
        return ;
    printf("one client link success\n");
    struct epoll_event event;
    event.events=EPOLLIN|EPOLLRDHUP|EPOLLET;
    event.data.fd=c;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,c,&event);
    assert(res!=-1);
    //将c文件描述符设置为非阻塞
    int oldoption=fcntl(c,F_GETFL);
    int newoption=oldoption|O_NONBLOCK;//设置成非阻塞的
    fcntl(c,F_SETFL,newoption);
}
void dealclidata(int fd)
{
    while(1)
    {
        char buff[128]={0};
        int n=recv(fd,buff,1,0);//如果将fd设置为非阻塞方式，则recv不会阻塞，如果没有数据，则recv返回-1，并且会设置全局的errno
        if(n==0)
        {
            printf("%d recv error\n",fd);
            break;
        }
        else if(n==-1)
        {
            if(errno==EAGAIN||errno==EWOULDBLOCK)
            {
                printf("read later\n");
                send(fd,"OK",2,0);
                break;
            }
            else
            {
                printf("%d recv error\n",fd);
                break;
            }
        }
        else
        {
            printf("%d: %s\n",fd,buff);
        }
    }
}
void dealfinishevents(struct epoll_event *events,int n,int sockfd,int epfd)
{
    int i=0;
    for(;i<n;++i)
    {
        int fd=events[i].data.fd;
        if(fd==sockfd)
        {
            getnewclilink(sockfd,epfd);
        }
        else
        {
            if(events[i].events&EPOLLRDHUP)
            {
                close(fd);
                epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);//删除内核事件表中的fd
            }
            else
            {
                dealclidata(fd);
            }
        }
    }
}
int main()
{
    int sockfd=initsocket();
    assert(sockfd!=-1);

    int epfd=epoll_create(5);
    assert(epfd!=-1);

    struct epoll_event event;
    event.events=EPOLLIN;
    event.data.fd=sockfd;

    int res=epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&event);
    assert(res!=-1);
    int count=0;
    while(1)
    {
        struct epoll_event events[maxevents];
        int n=epoll_wait(epfd,events,maxevents,-1);
        printf("epoll_wait return %d\n",count++);
        if(n<=0)
        {
            printf("epoll_wait error\n");
            continue;
        }
        dealfinishevents(events,n,sockfd,epfd);
    }
    return 0;
}

测试

发一次之后全部接收完