网络学习笔记：基于 I/O 复用 epoll 的并发服务器

本文迁移并整理自个人之前在博客园的博客。

本文承接自上一篇基于 I/O 复用 select 函数的并发服务器，也是并发服务器系列学习笔记的第 4 篇。

参考资料：《UNIX网络编程 卷1》

epoll 介绍

epoll 是在 2.6 内核中提出的。和 select 类似，它也是一种 I/O 复用技术，是之前的 select 和 poll 的增强版本。

Linux 下设计并发网络程序，向来不缺少方法，比如典型的 Apache 模型（Process Per Connection，简称 PPC），TPC（Thread PerConnection）模型，以及 select 模型和 poll 模型，那为何还要再引入 epoll 呢？我们先来看一下常用模型的缺点：

PPC/TPC模型
这两种模型思想类似，就是让每一个到来的连接一边自己做事去，别再来烦我。只是 PPC 是为它开了一个进程，而 TPC 开了一个线程。可是别烦我是有代价的，它要时间和空间啊，连接多了之后，那么多的进程/线程切换，这开销就上来了；因此这类模型能接受的最大连接数都不会高，一般在几百个左右。

select 模型
1）最大并发数限制，因为一个进程所打开的 FD（文件描述符）是有限制的，由 FD_SETSIZE 设置，默认值是 1024/2048，因此 select 模型的最大并发数就被相应限制了。自己改改这个 FD_SETSIZE？想法虽好，可是先看看下面吧…
2）效率问题，select 每次调用都会线性扫描全部的 FD 集合，这样效率就会呈现线性下降，把 FD_SETSIZE 改大的后果就是，大家都慢慢来，什么？都超时了？？！！
3）内核/用户空间内存拷贝问题，如何让内核把 FD 消息通知给用户空间呢？在这个问题上 select 采取了内存拷贝方法。

poll 模型
基本上效率和 select 是相同的，select 缺点的 2 和 3 它都没有改掉。

epoll 的提升
其实把 select 的缺点反过来那就是 epoll 的优点了：
1）epoll 没有最大并发连接的限制，上限是最大可以打开文件的数目（一般远大于2048），一般跟系统内存关系很大。具体数目可以 ‘cat /proc/sys/fs/file-max’ 察看。
2）效率提升，epoll 最大的优点就在于它只管“活跃”的连接，而跟连接总数无关，因此在实际的网络环境中，epoll 的效率就会远远高于 select 和 poll。
3）内存拷贝，epoll 在这点上使用了“共享内存”，所以这个内存拷贝也省略了。

关于 select、poll 和 epoll 的详细区别可参考博文select、poll、epoll 之间的区别总结。

epoll 接口

epoll 操作过程用到的三个接口如下：

1
2
3
4
5
6
7
#include <sys/epoll.h>
int epoll_create(int size);    
// 返回：若成功返回非负 epoll 描述符，否则返回 -1，并设置 errno 的值
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *events);    
// 返回：若成功返回 0，否则返回 -1，并设置 errno 的值
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);    
// 返回：若成功返回在 timeout 时间内就绪的文件描述符数，否则返回 -1，并设置 errno 的值

epoll_create

该函数返回一个 epoll 的描述符（一个整数）。size 用来告诉内核这个监听的数目一共有多大，不同于 select 函数中的第一个参数给出最大监听的 fd+1 的值。需要注意的是，当创建好 epoll 句柄后，它就是会占用一个 fd 值，在 Linux 下如果查看 /proc/进程id号/fd/，是能够看到这个 fd 的，所以在使用完 epoll 后，必须调用 close 函数关闭，否则可能导致 fd 被耗尽。

如下图是刚打开服务端程序（采用 epoll ET 模式）的截图：

上图中，前三个 /dev/pts/1 表示系统输入、输出及异常，socket 表示监听套接字，eventepoll 表示进程创建的 epoll。

另外，当有新的客户连接到服务端时，截图如下：

epoll_ctl

该函数是epoll的事件注册函数。它不同于 select 函数是在监听事件时告诉内核要监听什么类型的事件，而是在这里注册要监听的事件类型。

该函数的参数说明如下（fd 是 file descriptor 的缩写，表示文件描述符）：
1）第一个参数是 epoll_create 函数的返回值。
2）第二个参数表示动作：

  • EPOLL_CTL_ADD：注册新的fd到epfd中
  • EPOLL_CTL_MOD：修改已经注册的fd的监听事件
  • EPOLL_CTL_DEL：从epfd中删除一个fd
    3）第三个参数表示需要监听的 fd。
    4）第四个参数告诉内核需要监听什么事。struct epoll_event的结构如下：
    1
    2
    3
    4
    struct epoll_event {
        uint32_t events;    /* Epoll events */
        epoll_data_t data;  /* User data variable */
    };

events可以是以下几个宏的集合：

  • EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
  • EPOLLOUT：表示对应的文件描述符可以写；
  • EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
  • EPOLLERR：表示对应的文件描述符发生错误；
  • EPOLLHUP：表示对应的文件描述符被挂断；
  • EPOLLET： 将EPOLL设为边缘触发（Edge Triggered）模式，这是相对于水平触发（Level Triggered）来说的。
  • EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

epoll_wait

该函数等待事件的产生，类似于 select 调用。参数 events 用来从内核得到事件的集合；maxevents 告诉内核返回的 events 的最大大小，这个 maxevents 的值不能大于创建 epoll_create 时的 size，也必须大于 0；参数 timeout 是超时时间（毫秒，0 会立即返回，-1 将永久阻塞）。该函数返回需要处理的数目，如返回 0 表示已超时。

epoll 工作模式

需重点参考资料或博文：

epoll 对文件描述符的操作有两种模式：LT（level trigger）和 ET（edge trigger）。LT 模式是默认模式，LT 模式与 ET 模式的区别如下：
LT 模式：只要某个监听中的文件描述符处于 readable/writable 状态，无论什么时候进行 epoll_wait 都会返回该描述符；
ET 模式：只有某个监听中的文件描述符从 unreadable 变为 readable 或从 unwritable 变为 writable 时，epoll_wait 才会返回该描述符。

非阻塞 Socket 读写

在一个非阻塞的 socket 上调用 read/write 函数, 返回 EAGAIN 或者 EWOULDBLOCK (注: EAGAIN 就是 EWOULDBLOCK)
从字面上看，意思是：

  • EAGAIN：再试一次
  • EWOULDBLOCK：如果这是一个阻塞 socket，操作将被 block
  • perror 输出：Resource temporarily unavailable

总结：
这个错误表示资源暂时不够，可能read时，读缓冲区没有数据，或者，write 时，写缓冲区满了。
遇到这种情况，如果是阻塞 socket，read/write 就要阻塞掉。而如果是非阻塞 socket，read/write 立即返回 -1，同时 errno 设置为 EAGAIN。所以，对于阻塞 socket, read/write 返回 -1 代表网络出错了。但对于非阻塞 socket, read/write 返回 -1 不一定网络真的出错了，可能是 Resource temporarily unavailable。 这时我们应该再试，直到 Resource available。
　　
综上，对于 non-blocking 的 socket，正确的读写操作为：
读：忽略掉 errno = EAGAIN 的错误，下次继续读。
写：忽略掉 errno = EAGAIN 的错误，下次继续写。

对于 select 和 epoll 的 LT 模式，这种读写方式是没有问题的。 但对于 epoll 的 ET 模式，这种方式还有漏洞。

LT 及 ET 模式

我们先来看一下 Linux 官方给出的手册关于 LT 和 ET 的一个例子：

The epoll event distribution interface is able to behave both as edge-triggered (ET) and as level-triggered (LT). The difference between the two mechanisms can be described as follows. Suppose that this scenario happens:

  1. The file descriptor that represents the read side of a pipe (rfd) is registered on the epoll instance.
  2. A pipe writer writes 2 kB of data on the write side of the pipe.
  3. A call to epoll_wait(2) is done that will return rfd as a ready file descriptor.
  4. The pipe reader reads 1 kB of data from rfd.
  5. A call to epoll_wait(2) is done.

If the rfd file descriptor has been added to the epoll interface using the EPOLLET (edge-triggered) flag, the call to epoll_wait(2) done in step 5 will probably hang despite the available data still present in the file input buffer; meanwhile the remote peer might be expecting a response based on the data it already sent. The reason for this is that edge-triggered mode delivers events only when changes occur on the monitored file descriptor. So, in step 5 the caller might end up waiting for some data that is already present inside the input buffer. In the above example, an event on rfd will be generated because of the write done in 2 and the event is consumed in 3. Since the read operation done in 4 does not consume the whole buffer data, the call to epoll_wait(2) done in step 5 might block indefinitely.

An application that employs the EPOLLET flag should use nonblocking file descriptors to avoid having a blocking read or write starve a task that is handling multiple file descriptors. The suggested way to use epoll as an edge-triggered (EPOLLET) interface is as follows:

i with nonblocking file descriptors; and
ii by waiting for an event only after read(2) or write(2) return EAGAIN.

By contrast, when used as a level-triggered interface (the default, when EPOLLET is not specified), epoll is simply a faster poll(2), and can be used wherever the latter is used since it shares the same semantics.

从上述例子，我们可以看出：

  1. 在 LT 模式下，只要某个监听中的文件描述符处于 readable/writable 状态，无论什么时候进行 epoll_wait 都会返回该描述符。所以，只要读缓冲区有数据或者写缓冲区仍有空间，那就可读或可写， 都不会因套接字设定为 blocking 或者 non-blocking 而导致 epoll_wait 进入无限期等待；
  2. 在 ET 模式下，只有某个监听中的文件描述符从 unreadable 变为 readable 或从 unwritable 变为 writable 时，epoll_wait 才会返回该描述符。所以当我们没有把读缓冲区的数据全部读完或者没有把写缓冲区的空间写满就返回，套接字就会一直处于不可读或者不可写的状态，这样 read/write 会阻塞直到套接字可读或可写，但在这种情况下，套接字不可能变为可读或可写，所以 read/write 会一直阻塞下去，从而导致 epoll_wait 无限期阻塞于该套接字而无法返回。所以，要将套接字设定为 non-blocking，让 epoll_waite 可以及时返回。

下边两图可形象展示 LT 和 ET 的区别：
从 socket 读数据:

往 socket 写数据:

所以，在 epoll 的 ET 模式下，正确的读写方式为:
读：只要可读，就一直读，直到返回 0，或者 errno = EAGAIN
写：只要可写，就一直写，直到数据发送完，或者 errno = EAGAIN

示例

示例1：回射程序（LT模式）

这里服务器端程序主要摘自博文IO多路复用之epoll总结，不过修复了部分 Bug。服务端代码如下：

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
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>

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

#define PORT        9877
#define BUFFERSIZ    1024
#define LISTENQ     1024
#define FDSIZE      1024
#define EPOLLEVENTS 100

//函数声明
//创建套接字并进行绑定
static int socket_bind(int port);
//IO多路复用epoll
static void do_epoll(int listenfd);
//事件处理函数
static void
handle_events(int epollfd, struct epoll_event *events, int num, int listenfd, char *buf);
//处理接收到的连接
static void handle_accpet(int epollfd, int listenfd);
//读处理
static void do_read(int epollfd, int fd, char *buf);
//写处理
static void do_write(int epollfd, int fd, char *buf);
//添加事件
static void add_event(int epollfd, int fd, int state);
//修改事件
static void modify_event(int epollfd, int fd, int state);
//删除事件
static void delete_event(int epollfd, int fd, int state);
//close socket
void Close(int fd);

int main(int argc, char *argv[])
{
    int  listenfd;
    listenfd = socket_bind(PORT);
    listen(listenfd, LISTENQ);
    do_epoll(listenfd);
    exit(0);
}

static int socket_bind(int port)
{
    int  listenfd;
    struct sockaddr_in servaddr;
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd == -1)
    {
        perror("socket error:");
        exit(1);
    }
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(port);
    if (bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) == -1)
    {
        perror("bind error: ");
        exit(1);
    }
    return listenfd;
}

static void do_epoll(int listenfd)
{
    int epollfd;
    struct epoll_event events[EPOLLEVENTS];
    int num;
    char buf[BUFFERSIZ];
    memset(buf, 0, BUFFERSIZ);
    //创建一个描述符
    epollfd = epoll_create(FDSIZE);
    //添加监听描述符事件
    add_event(epollfd, listenfd, EPOLLIN);
    for ( ; ; )
    {
        //获取已经准备好的描述符事件
        if ((num = epoll_wait(epollfd, events, EPOLLEVENTS, -1)) == -1) 
        {
            perror("epoll_pwait");
            exit(1);
        }
        handle_events(epollfd, events, num, listenfd, buf);
    }
    Close(epollfd);
}

static void
handle_events(int epollfd, struct epoll_event *events, int num, int listenfd, char *buf)
{
    int i;
    int fd;
    // 遍历
    for (i = 0; i < num; i++)
    {
        fd = events[i].data.fd;
        //根据描述符的类型和事件类型进行处理
        if ((fd == listenfd) && (events[i].events & EPOLLIN))
            handle_accpet(epollfd, listenfd);
        else if (events[i].events & EPOLLIN)
            do_read(epollfd, fd, buf);
        else if (events[i].events & EPOLLOUT)
            do_write(epollfd, fd, buf);
    }
}
static void handle_accpet(int epollfd, int listenfd)
{
    int clifd;
    struct sockaddr_in cliaddr;
    socklen_t  cliaddrlen;
    clifd = accept(listenfd, (struct sockaddr*)&cliaddr, &cliaddrlen);
    if (clifd == -1)
        perror("accpet error:");
    else
    {
        printf("accept a new client: %s:%d\n", inet_ntoa(cliaddr.sin_addr), cliaddr.sin_port);
        //添加一个客户描述符和事件
        add_event(epollfd, clifd, EPOLLIN);
    }
}

static void do_read(int epollfd, int fd, char *buf)
{
    int nread;
    nread = read(fd, buf, BUFFERSIZ);
    if (nread == -1)
    {
        perror("read error:");
        //Close(fd);
        delete_event(epollfd, fd, EPOLLIN);
    }
    else if (nread == 0)
    {
        fprintf(stderr, "client close.\n");
        //Close(fd);
        delete_event(epollfd, fd, EPOLLIN);
    }
    else
    {
        printf("read message is : %s", buf);
        //修改描述符对应的事件，由读改为写
        modify_event(epollfd, fd, EPOLLOUT);
    }
}

static void do_write(int epollfd, int fd, char *buf)
{
    int nwrite;
    nwrite = write(fd, buf, strlen(buf));
    if (nwrite == -1)
    {
        perror("write error:");
        //Close(fd);
        delete_event(epollfd, fd, EPOLLOUT);
    }
    else
        modify_event(epollfd, fd, EPOLLIN);
    memset(buf, 0, BUFFERSIZ);
}

static void add_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if((epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev)) == -1)
    {
        perror("epoll_ctl: add");
    }
}

static void delete_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if((epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev)) == -1)
    {
        perror("epoll_ctl: del");
    }
    Close(fd);    
    // 如果描述符fd已关闭，再从epoll中删除fd，则会出现epoll failed: Bad file descriptor问题
    // 所以要先从epoll中删除fd，在关闭fd. 具体可参考博文http://www.cnblogs.com/scw2901/p/3907657.html
}

static void modify_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if((epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &ev)) == -1)
    {
        perror("epoll_ctl: mod");
    }
}

void Close(int fd)
{
    if((close(fd)) < 0)
    {
        perror("close socket error");
        exit(1);
    }
}

客户端程序：

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
#include <netinet/in.h>
#include <sys/socket.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <time.h>
#include <unistd.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <errno.h>

#define BUFFERSIZ    1024
#define SERV_PORT   9877
#define FDSIZE        1024
#define EPOLLEVENTS 100

void handle_connection(int sockfd);
void handle_events(int epollfd, struct epoll_event *events, int num, int sockfd, char *buf);
void add_event(int epollfd, int fd, int state);
void delete_event(int epollfd, int fd, int state);
void modify_event(int epollfd, int fd, int state);

ssize_t Read(int fd, void *ptr, size_t nbytes);
void Write(int fd, void *ptr, size_t nbytes);
ssize_t writen(int fd, const void *vptr, size_t n);
void Writen(int fd, void *ptr, size_t nbytes);

void Close(int fd);

int main(int argc, char **argv)
{
    if(argc != 2)
        perror("usage: tcpcli <IPaddress>");
    
    int sockfd;
    struct sockaddr_in  servaddr;
    if( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) 
    {
        perror("socket error\n");
        exit(1);
    }
    
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(SERV_PORT);
    if((inet_pton(AF_INET, argv[1], &servaddr.sin_addr)) <= 0)
    {
        perror("inet_pton error");
        exit(1);
    }

    if((connect(sockfd, (struct sockaddr*) &servaddr, sizeof(servaddr))) < 0)
    {
        perror("connect error\n");
        exit(1);
    }
    
    handle_connection(sockfd);
    
    Close(sockfd);
    exit(0);
}

void handle_connection(int sockfd)
{
    int epollfd;
    struct epoll_event events[EPOLLEVENTS];
    char buf[BUFFERSIZ];
    int num;
    epollfd = epoll_create(FDSIZE);
    add_event(epollfd, sockfd, EPOLLIN);
    add_event(epollfd, STDIN_FILENO, EPOLLIN);
    for ( ; ; )
    {
        num = epoll_wait(epollfd, events, EPOLLEVENTS, -1);
        handle_events(epollfd, events, num, sockfd, buf);
    }
    Close(epollfd);
}

void handle_events(int epollfd, struct epoll_event *events, int num, int sockfd, char *buf)
{
    int fd;
    int i;
    int stdineof = 0;
    for (i = 0; i < num; i++)
    {
        fd = events[i].data.fd;
        if(fd == sockfd)
        {
            int nread;
            nread = Read(sockfd, buf, BUFFERSIZ);
            if (nread == 0)
            {
                if(stdineof == 1)
                    return;
                fprintf(stderr, "server close\n");
                Close(sockfd);
                Close(epollfd);
                exit(1);
            }
            Write(fileno(stdout), buf, nread);
        }
        else
        {
            int nread;
            nread = Read(fd, buf, BUFFERSIZ);
            if (nread == 0)
            {
                stdineof = 1;
                fprintf(stderr, "no inputs\n");
                Close(fd);
                continue;
            }
            modify_event(epollfd, sockfd, EPOLLOUT);
            Writen(sockfd, buf, nread);
            modify_event(epollfd, sockfd, EPOLLIN);
            memset(buf, 0, BUFFERSIZ);
        }
    }
}

void add_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev) == -1)
    {
        perror("epoll_ctl: add");
    }
}

void delete_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if(epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev) == -1)
    {
        perror("epoll_ctl: del");
    }
}

void modify_event(int epollfd, int fd, int state)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    if(epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &ev) == -1)
    {
        perror("epoll_ctl: mod");
    }
}

ssize_t Read(int fd, void *ptr, size_t nbytes)
{
    ssize_t n;

    if ( (n = read(fd, ptr, nbytes)) == -1)
        perror("read error");
    return(n);
}

void Write(int fd, void *ptr, size_t nbytes)
{
    if (write(fd, ptr, nbytes) != nbytes)
        perror("write error");
}

/* Write "n" bytes to a descriptor. */
ssize_t writen(int fd, const void *vptr, size_t n)
{
    size_t        nleft;
    ssize_t        nwritten;
    const char    *ptr;

    ptr = vptr;
    nleft = n;
    while (nleft > 0) {
        if ( (nwritten = write(fd, ptr, nleft)) <= 0) {
            if (nwritten < 0 && errno == EINTR)
                nwritten = 0;        /* and call write() again */
            else
                return(-1);            /* error */
        }

        nleft -= nwritten;
        ptr   += nwritten;
    }
    return(n);
}

void Writen(int fd, void *ptr, size_t nbytes)
{
    if (writen(fd, ptr, nbytes) != nbytes)
        perror("writen error");
}

void Close(int fd)
{
    if((close(fd)) < 0)
    {
        perror("close error");
        exit(1);
    }
}

程序运行截图如下：

1）客户端主动连接主动关闭
客户端：

服务端：

2）客户端主动连接被动关闭
客户端：

服务端：

示例2：回射程序（ET模式）

这里我们只给出服务器端的程序。这部分程序部分参考自博文Epoll在LT和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
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>

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

#define PORT        9877
#define BUFFERSIZ    1024
#define LISTENQ     1024
#define FDSIZE      1024
#define EPOLLEVENTS 100

//函数声明
//创建套接字并进行绑定
int socket_bind(int port);
//IO多路复用epoll
void do_epoll(int listenfd);
//事件处理函数
void handle_events(int epollfd, struct epoll_event *events, int num, int listenfd, char *buf);
//处理接收到的连接
void handle_accpet(int epollfd, int listenfd);
//读处理
void do_read(int epollfd, int fd, char *buf);
//写处理
void do_write(int epollfd, int fd, char *buf);

//设置socket连接为非阻塞模式
void setnonblocking(int sockfd);
//close socket
void Close(int fd);

int main(int argc, char *argv[])
{
    int  listenfd;
    listenfd = socket_bind(PORT);
    listen(listenfd, LISTENQ);
    do_epoll(listenfd);
    exit(0);
}

int socket_bind(int port)
{
    int  listenfd;
    struct sockaddr_in servaddr;
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd == -1)
    {
        perror("socket error:");
        exit(1);
    }
    setnonblocking(listenfd);
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(port);
    if (bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) == -1)
    {
        perror("bind error: ");
        exit(1);
    }
    return listenfd;
}

void do_epoll(int listenfd)
{
    int epollfd;
    struct epoll_event ev;
    struct epoll_event events[EPOLLEVENTS];
    int num;
    char buf[BUFFERSIZ];
    memset(buf, 0, BUFFERSIZ);
    //创建一个描述符
    epollfd = epoll_create(FDSIZE);
    //添加监听描述符事件
    ev.events = EPOLLIN;
    ev.data.fd = listenfd;
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev) == -1) 
    {
        perror("epoll_ctl: add");
        exit(1);
    }

    for ( ; ; )
    {
        //获取已经准备好的描述符事件
        if ((num = epoll_wait(epollfd, events, EPOLLEVENTS, -1)) == -1) 
        {
            perror("epoll_wait");
            exit(1);
        }
        handle_events(epollfd, events, num, listenfd, buf);
    }
    Close(epollfd);
}

void handle_events(int epollfd, struct epoll_event *events, int num, int listenfd, char *buf)
{
    int i;
    int fd;
    // 遍历
    for (i = 0; i < num; i++)
    {
        fd = events[i].data.fd;
        //根据描述符的类型和事件类型进行处理
        if ((fd == listenfd) && (events[i].events & EPOLLIN))
            handle_accpet(epollfd, listenfd);
        else if (events[i].events & EPOLLIN)
            do_read(epollfd, fd, buf);
        else if (events[i].events & EPOLLOUT)
            do_write(epollfd, fd, buf);
    }
}

void handle_accpet(int epollfd, int listenfd)
{
    int connfd;
    struct sockaddr_in cliaddr;
    socklen_t  cliaddrlen;
    struct epoll_event ev;
    while ((connfd = accept(listenfd, (struct sockaddr *) &cliaddr, &cliaddrlen)) > 0) 
    {
        printf("accept a new client: %s:%d\n", inet_ntoa(cliaddr.sin_addr), cliaddr.sin_port);
        setnonblocking(connfd);
        ev.events = EPOLLIN | EPOLLET;
        ev.data.fd = connfd;
        if (epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev) == -1) 
        {
            perror("epoll_ctl: add");
            exit(1);
        }
    }
    if (connfd == -1) 
    {
        if (errno != EAGAIN && errno != ECONNABORTED 
                && errno != EPROTO && errno != EINTR) 
            perror("accept");
    }
}

void do_read(int epollfd, int fd, char *buf)
{
    int nread;
    int n = 0;
    struct epoll_event ev;
    while ((nread = read(fd, buf + n, BUFFERSIZ-1)) > 0) 
    {
        n += nread;
    }
    if (nread == -1 && errno != EAGAIN) 
    {
        perror("read error");
        
        ev.data.fd = fd;
        ev.events = EPOLLIN | EPOLLET;
        if (epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev) == -1) 
        {
            perror("epoll_ctl: mod");
        }
        Close(fd);
    }
    else if(nread == 0)
    {
        perror("client close");
        
        ev.data.fd = fd;
        ev.events = EPOLLIN | EPOLLET;
        if (epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev) == -1) 
        {
            perror("epoll_ctl: mod");
        }
        Close(fd);
    }
    else
    {
        printf("read message is : %s", buf); 
        ev.data.fd = fd;
        ev.events = EPOLLOUT | EPOLLET;
        if (epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &ev) == -1) 
        {
            perror("epoll_ctl: mod");
        }
    }
}

void do_write(int epollfd, int fd, char *buf)
{
    int nwrite, data_size = strlen(buf);
    int n = data_size;
    struct epoll_event ev;
    int flag = 0;
    while (n > 0) 
    {
        nwrite = write(fd, buf + data_size - n, n);
        if (nwrite < n) 
        {
            if (nwrite == -1 && errno != EAGAIN) 
            {
                flag = 1;
                perror("write error");
    
                ev.data.fd = fd;
                ev.events = EPOLLOUT | EPOLLET;
                if (epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev) == -1) 
                {
                    perror("epoll_ctl: mod");
                }
                Close(fd);
            }
            break;
        }
        n -= nwrite;
    }
    
    if(flag != 1)
    {
        ev.data.fd = fd;
        ev.events =  EPOLLIN | EPOLLET;
        if (epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &ev) == -1) 
        {
            perror("epoll_ctl: mod");
        }
    }
    // 这句很重要
    memset(buf, 0, BUFFERSIZ);
}

void setnonblocking(int sockfd) 
{
    int opts;
    opts = fcntl(sockfd, F_GETFL);
    if(opts < 0) 
    {
        perror("Error: fcntl(F_GETFL)\n");
        exit(1);
    }
    opts = (opts | O_NONBLOCK);
    if(fcntl(sockfd, F_SETFL, opts) < 0) 
    {
        perror("Error: fcntl(F_SETFL)\n");
        exit(1);
    }
}

void Close(int fd)
{
    if((close(fd)) < 0)
    {
        perror("close socket error");
        exit(1);
    }
}

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

谢宏峰，毕业于暨南大学（2009 - 2013）、中国科学院（2013 - 2016），现就职于深信服科技，从事 OpenStack 开发。