UNIX环境高级编程学习之第十六章网络IPC:套接字 – 非阻塞的Socket通信EPoll模型(多路复用), 实用Socket通信模板
[code lang=”cpp”]/* User:Lixiujie
* Date:20101207
* Desc:Unix(Linux)非阻塞的Socket通信EPoll模型,多路复用,TCP服务器端, 向客户端发送响应信息。
* File:tcp_server_epoll.c
* System:Ubuntu 64bit
* gcc tcp_server_epoll.c -o tcp_server_epoll
* tcp_server_epoll 7878
*
EPoll 函数介绍
epoll是Linux内核为处理大批量句柄而作了改进的poll,是Linux下多路复用IO接口select/poll的增强版本,它能显著减少程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。

EPoll 优点
1、它保留了poll的两个相对与select的优点
2、epoll_wait的参数events作为出参,直接返回了有事件发生的fd,epoll_wait的返回值既是发生事件的个数,省略了poll中返回之后的循环操作。
3、不再象select、poll一样将标识符局限于fd,epoll中可以将标识符扩大为指针,大大增加了epoll模型下的灵活性。

EPoll 使用说明事项
1、如果fd被注册到两个epoll中时,如果有时间发生则两个epoll都会触发事件。
2、如果注册到epoll中的fd被关闭,则其会自动被清除出epoll监听列表。注意:关闭自动清除,不用手机清除
3、如果多个事件同时触发epoll,则多个事件会被联合在一起返回。
4、epoll_wait会一直监听epollhup事件发生,所以其不需要添加到events中。
5、为了避免大数据量io时,et模式下只处理一个fd,其他fd被饿死的情况发生。linux建议可以在fd联系到的结构中增加ready位,然后epoll_wait触发事件之后仅将其置位为ready模式,然后在下边轮询ready fd列表。
6、epoll_ctl epoll的事件注册函数,其events参数可以是以下宏的集合:
EPOLLIN: 表示对应的文件描述符可以读(包括对端SOCKET正常关闭);
EPOLLOUT: 表示对应的文件描述符可以写;
EPOLLPRI: 表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来);
EPOLLERR: 表示对应的文件描述符发生错误;写已关闭socket pipe broken
EPOLLHUP: 表示对应的文件描述符被挂断;譬如收到RST包。在注册事件的时候这个事件是默认添加。
EPOLLRDHUP: 表示对应的文件描述符对端socket关闭事件,主要应用于ET模式下。
在水平触发模式下,如果对端socket关闭,则会一直触发epollin事件,驱动去处理client socket。
在边沿触发模式下,如果client首先发送协议然后shutdown写端。则会触发epollin事件。但是如果处理程序只进行一次recv操作时,根据recv收取到得数据长度来判读后边是否还有需要处理的协议时,将丢失客户端关闭事件。

EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里

EPoll 工作模式
1、水平触发Level Triggered (LT) 是EPoll的缺省的工作方式,并且同时支持block和no-block socket.在这种做法中,内核告诉你一个文件描述符是否就绪了,然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作,内核还是会继续通知你 的,所以,这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表.

2、边缘触发Edge Triggered (ET) 是EPoll的高速工作方式,只支持no-block socket。在这种模式下,当描述符从未就绪变为就绪时,内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪,并且不会再为那个文件描述 符发送更多的就绪通知,直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如,你在发送,接收或者接收请求,或者发送接收的数据少于一定量时导致 了一个EWOULDBLOCK 错误)。但是请注意,如果一直不对这个fd作IO操作(从而导致它再次变成未就绪),内核不会发送更多的通知(only once),不过在TCP协议中,ET模式的加速效用仍需要更多的benchmark确认。
内核的读buffer有内核态主动变化时,内核会通知你, 无需再去mod。写事件是给用户使用的,最开始add之后,内核都不会通知你了,你可以强制写数据(直到EAGAIN或者实际字节数小于 需要写的字节数),当然你可以主动mod OUT,此时如果句柄可以写了(send buffer有空间),内核就通知你。
这里内核态主动的意思是:内核从网络接收了数据放入了读buffer(会通知用户IN事件,即用户可以recv数据)
并且这种通知只会通知一次,如果这次处理(recv)没有到刚才说的两种情况(EAGIN或者实际字节数小于 需要读写的字节数),则该事件会被丢弃,直到下次buffer发生变化。
与LT的差别就在这里体现,LT在这种情况下,事件不会丢弃,而是只要读buffer里面有数据可以让用户读,则不断的通知你。

EPoll 函数使用介绍
1、EPoll 创建epoll句柄函数。
int epoll_create(int size);
参数size:用来告诉内核要监听的数目一共有多少个。
返回值:成功时,返回一个非负整数的文件描述符,作为创建好的epoll句柄。调用失败时,返回-1,错误信息可以通过errno获得。
说明:创建一个epoll句柄,size用来告诉内核这个监听的数目一共有多大。这个参数不同于select()中的第一个参数,给出最大监听的fd+1的值。需要注意的是,当创建好epoll句柄后,它就是会占用一个fd值,所以在使用完epoll后,必须调用close()关闭,否则可能导致fd被耗尽。

2、EPoll 注册修改删除文件描述符到epoll句柄函数
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
参数epfd:epoll_create()函数返回的epoll句柄。
参数op:操作选项。
参数fd:要进行操作的目标文件描述符。
参数event:struct epoll_event结构指针,将fd和要进行的操作关联起来。
返回值:成功时,返回0,作为创建好的epoll句柄。调用失败时,返回-1,错误信息可以通过errno获得。
说明:epoll的事件注册函数,它不同与select()是在监听事件时告诉内核要监听什么类型的事件,而是在这里先注册要监听的事件类型。
参数op的可选值有以下3个:
EPOLL_CTL_ADD:注册新的fd到epfd中;
EPOLL_CTL_MOD:修改已经注册的fd的监听事件;
EPOLL_CTL_DEL:从epfd中删除一个fd;

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

3、EPoll 等待事件的产生函数
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
参数epfd:epoll_create()函数返回的epoll句柄。
参数events:struct epoll_event结构指针,用来从内核得到事件的集合。
参数 maxevents:告诉内核这个events有多大
参数 timeout: 等待时的超时时间,以毫秒为单位。
返回值:成功时,返回需要处理的事件数目。调用失败时,返回0,表示等待超时。
说明:等待事件的产生。
timeout值 说明
-1 永远等待
0 立即返回,不阻塞进程
>0 等待指定数目的毫秒数

EPoll 注意事项
建立连接的时候epoll_add IN和OUT事件, 后面就不需要管了
每次read/write的时候,到两种情况下结束:
1 发生EAGAIN
2 read/write的实际字节数小于 需要读写的字节数
对于第二点需要注意两点:
A:如果是UDP服务,处理就不完全是这样,必须要recv到发生EAGAIN为止,否则就丢失事件了
因为UDP和TCP不同,是有边界的,每次接收一定是一个完整的UDP包,当然recv的buffer需要至少大于一个UDP包的大小
随便再说一下,一个UDP包到底应该多大?
对于internet,由于MTU的限制,UDP包的大小不要超过576个字节,否则容易被分包,对于公司的IDC环境,建议不要超过1472,否则也比较容易分包。

B 如果发送方发送完数据以后,就close连接,这个时候如果recv到数据是实际字节数小于读写字节数,根据开始所述就认为到EAGIN了从而直接返回,等待下一次事件,这样是有问题的,close事件丢失了!
因此如果依赖这种关闭逻辑的服务,必须接收数据到EAGIN为止。

*/
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h> /* socket bind listen connect accept send recv */
#include <arpa/inet.h> /* htons ntohs htonl ntohl inet_addr inet_ntoa */
#include <netinet/in.h> /* sockaddr_in */

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

#define BUFLEN 1024
#define QLEN 20

/* 传送信息结构体 */
typedef struct _MyMsg{
char szCmd[16];/* message command
* RE_LINK:test link request
* RESP_LINK:test link response
* RE_EXIT: exit request
* RESP_TEXT: exit response
* RE_DATA: data request
* RESP_DATA: data response
*/
int iLen; /* message data length*/
char szData[0];/* message data */
}MyMsg;

/* 信息处理 */
MyMsg * MsgProcess(MyMsg *pMsgIn){
char szBuf[BUFLEN] = { 0x00 };
char szTmp[BUFLEN] = { 0x00 };
MyMsg *pMsg = NULL;
FILE *fp;
if (strcmp(pMsgIn->szCmd, "RE_LINK") == 0){
pMsg = (MyMsg *)malloc(sizeof(MyMsg));
memset(pMsg, 0, sizeof(MyMsg));
strcpy(pMsg->szCmd, "RESP_LINK");
}else if (strcmp(pMsgIn->szCmd, "RESP_LINK") == 0){
}else if (strcmp(pMsgIn->szCmd, "RE_EXIT") == 0){
pMsg = (MyMsg *)malloc(sizeof(MyMsg));
memset(pMsg, 0, sizeof(MyMsg));
strcpy(pMsg->szCmd, "RESP_TEXT");
}else if (strcmp(pMsgIn->szCmd, "RESP_TEXT") == 0){
}else if (strcmp(pMsgIn->szCmd, "RE_DATA") == 0){
memset(szBuf, 0, BUFLEN);
strncpy(szBuf, pMsgIn->szData, pMsgIn->iLen);
if ((fp = popen(szBuf, "r")) == NULL){
memset(szBuf, 0, BUFLEN);
sprintf(szBuf, "error: %s\n", strerror(errno));
}else{
memset(szTmp, 0, BUFLEN);
while (fgets(szTmp, BUFLEN, fp) != NULL){
strcat(szBuf, szTmp);
memset(szTmp, 0, BUFLEN);
}
pclose(fp);
}
pMsg = (MyMsg *)malloc(sizeof(MyMsg)+ strlen(szBuf)+1);
memset(pMsg, 0, sizeof(MyMsg));
strcpy(pMsg->szCmd, "RESP_DATA");
pMsg->iLen = strlen(szBuf)+1;
strcpy(pMsg->szData, szBuf);
}else if (strcmp(pMsgIn->szCmd, "RESP_DATA") == 0){
}
return pMsg;
}

/* Socket结构体 */
typedef struct _SockNode{
int sock;
struct sockaddr_in addr;
struct _SockNode *pNext;
}SockNode;

/* create SockNode */
SockNode* createSockNode(int sock, struct sockaddr_in *pAddr){
SockNode *p = NULL;
if ((p = (SockNode *)malloc(sizeof(SockNode))) != NULL){
p->sock = sock;
memcpy(&(p->addr), pAddr, sizeof(struct sockaddr_in));
p->pNext = NULL;
}
return p;
}

/* add SockNode from list */
void addSockNodeList(SockNode *head, SockNode *node){
SockNode *p = head;
while (p->pNext != NULL){
p = p->pNext;
}
p->pNext = node;
}

/* delete SockNode from list
* return head
*/
SockNode* deleteSockNodeList(SockNode *head, int sock){
SockNode *p = head, *pPrevious = p;
while (p != NULL){
if (p->sock == sock){
if (p != pPrevious){
pPrevious->pNext = p->pNext;
}else{
head = p->pNext;
}
free(p);
break;
}
pPrevious = p;
p = p->pNext;
}
return head;
}
/* select SockNode from list
* return head
*/
SockNode* selectSockNodeList(SockNode *head, int sock){
SockNode *p = head, *pPrevious = p;
while (p != NULL){
if (p->sock == sock){
return p;
}
p = p->pNext;
}
return NULL;
}

/* maximumly sock from list */
int maxSockNodeList(SockNode *head){
SockNode *p = head;
int maxsock = -1;
while (p != NULL){
maxsock = maxsock > p->sock ? maxsock : p->sock;
p = p->pNext;
}
return maxsock;
}

/* create tcp server */
int initserver(int type, const struct sockaddr *addr, socklen_t alen, int qlen){
int fd;
int err = 0, iSockAttrOn = 1;

/* 创建 */
if ((fd = socket(addr->sa_family, type, 0)) < 0){
return -1;
}
/* 端口复用 */
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &iSockAttrOn, sizeof(iSockAttrOn) ) < 0){
err = errno;
goto errout;
}
/* 绑定 */
if (bind(fd, addr, alen) < 0){
err = errno;
goto errout;
}
/* 监听数 */
if (SOCK_STREAM == type || SOCK_SEQPACKET == type){
if (listen(fd, qlen) < 0) {
err = errno;
goto errout;
}
}
return fd;
errout:
close(fd);
errno = err;
return -1;
}
/* 设置为非阻塞模式 */
void setnonblocking(int sock){
int opts;
opts=fcntl(sock,F_GETFL);
if(opts<0){
perror("fcntl(sock,GETFL)");
exit(1);
}
opts = opts|O_NONBLOCK;
if(fcntl(sock,F_SETFL,opts)<0){
perror("fcntl(sock,SETFL,opts)");
exit(1);
}
}
/* EPoll ET模式下读取数据函数, 保证 szBuf空间足够大, 否则数据丢失 */
int epoolRecv(int fd, char *szBuf, int nBuflen){
int recvTotalLen = 0, recvLen = 0;
char szTmp[1024] = { 0x00 };
while (1) {
memset(szTmp, 0x00, sizeof(szTmp));
recvLen = recv(fd, szTmp, sizeof(szTmp), 0);
if (recvLen < 0){
if (EAGAIN == errno) {
return recvTotalLen; /* 读取结束,正常返回 */
} else {
perror("Err:epoolRecv recv err!");
return -1;
}
}else if (0 == recvLen) {
return 0; /* 对方Socket正常关闭 */
}
if (recvTotalLen + recvLen <= nBuflen){
memcpy(szBuf + recvTotalLen, szTmp, recvLen);
}
recvTotalLen += recvLen;
if (recvLen < sizeof(szTmp)){ /* TCP 可以用这种模式, UDP 可能报EAGAIN才能结束 */
return recvTotalLen; /* 读取结束,正常返回 */
}
}
return recvTotalLen;
}
/* EPoll ET模式下发送数据函数 */
int epollSend(int fd, const char *szBuf, int nBuflen){
int sendTotalLen = 0, sendLen = 0;
char szTmp[1024] = { 0x00 };
while (1) {
sendLen = send(fd, szBuf + sendTotalLen, nBuflen – sendTotalLen, 0);
if (sendLen < 0) {
/* 当socket是非阻塞时,如返回此错误,表示写缓冲队列已满,在这里做延时后再重试. */
if (errno == EAGAIN) {
usleep(1000);
continue;
}
return -1;
}else if (0 == sendLen) {
return 0;
}
sendTotalLen += sendLen;
if(sendTotalLen == nBuflen)
return sendTotalLen;
}
return sendTotalLen;
}

int main(int argc, char *argv[]){
if (argc != 2){
printf("arg err!\n");
return -1;
}
int sefd, clfd, ret, len;
char szBuf[BUFLEN];
SockNode *head = NULL,*node = NULL; /* socket 监听链表 */

struct sockaddr_in se_addr,cl_addr;
socklen_t alen = sizeof(struct sockaddr);
/* 设置服务IP和端口 */
memset((void *)&se_addr, 0, alen);
se_addr.sin_family = AF_INET;
se_addr.sin_addr.s_addr = INADDR_ANY;// inet_addr("0.0.0.0");
se_addr.sin_port = htons(atoi(argv[1]));

if ((sefd = initserver(SOCK_STREAM, (struct sockaddr *)&se_addr, alen, QLEN)) < 0){
printf("initserver err=%s!\n", strerror(errno));
return -1;
}
printf("initserver OK !\n");
head = createSockNode(sefd, &se_addr);

int epfd = epoll_create(256);/* 创建一个epoll句柄,size用来告诉内核这个监听的数目一共有多大 */
struct epoll_event ev, events[QLEN];/* 声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件 */
/* 注册Server socket事件到EPoll */
ev.data.fd = sefd;
ev.events = EPOLLIN|EPOLLET; /* 读事件、ET模式 */
epoll_ctl(epfd, EPOLL_CTL_ADD, sefd, &ev); /* 注册epoll事件 */
int i, nevs;
while (1){
printf("epoll_wait before OK !\n");
nevs = epoll_wait(epfd, events, QLEN, -1);
printf("epoll_wait after OK !ret = %d\n", ret);
if (nevs < 0){
if(errno == EINTR && epfd > 0){
usleep(10*1000);
continue;
}
printf("epoll_wait err=%s!\n", strerror(errno));
while (head != NULL){
node = head;
head = head->pNext;
close(node->sock);
free(node);
}
return -1;
}else if (0 == nevs) { /* 不可能出现 */
printf("epoll_wait timeout!\n");
sleep(1);
continue;
}
for (i = 0; i < nevs; i++){
if (events[i].data.fd == sefd){ /* Server Socket */
alen = sizeof(struct sockaddr);
memset((void *)&cl_addr, 0 , alen);
clfd = accept(events[i].data.fd, (struct sockaddr*)&cl_addr, &alen);
if (clfd < 0){
printf("accept err=%s!\n", strerror(errno));
while (head != NULL){
node = head;
head = head->pNext;
close(node->sock);
free(node);
}
return -1;
}
printf("Client connect:ip=%s, port=%d \n", inet_ntoa(cl_addr.sin_addr),
ntohs(cl_addr.sin_port));
addSockNodeList(head, createSockNode(clfd, &cl_addr));

setnonblocking(clfd); /* 设置客户端为非阻塞模式 */
ev.data.fd = clfd;
ev.events = EPOLLIN|EPOLLET; /* 读事件、ET模式 */
epoll_ctl(epfd, EPOLL_CTL_ADD, clfd, &ev); /* 注册epoll事件 */

}else if (events[i].events | EPOLLIN){
node = selectSockNodeList(head, events[i].data.fd);
if (NULL == node){
continue;
}
memset(szBuf, 0, BUFLEN);
/* ret = recv(node->sock, szBuf, BUFLEN, 0); */
ret = epoolRecv(node->sock, szBuf, BUFLEN);
if (ret < 0){
printf("epoolRecv Client err=%s, ip=%s, port=%d!\n", strerror(errno),
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
close(node->sock);
events[i].data.fd = -1;
head = deleteSockNodeList(head, node->sock);
} else if (0 == ret){
printf("epoolRecv Client exit, ip=%s, port=%d!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
close(node->sock);
events[i].data.fd = -1;
head = deleteSockNodeList(head, node->sock);

} else {
MyMsg *msgRecv = (MyMsg *)szBuf;
msgRecv->iLen = ntohl(msgRecv->iLen);/* 转换成本机字节序 */
MyMsg *msgSend = NULL;
/* 预处理 */
if (strcmp(msgRecv->szCmd, "RE_LINK") == 0){
printf("epoolRecv Client RE_LINK, ip=%s, port=%d!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
msgSend = MsgProcess(msgRecv); /* 实际处理 */
if (msgSend != NULL){
len = msgSend->iLen;
msgSend->iLen = htonl(msgSend->iLen); /* 转换成网络字节序 */
memset(szBuf, 0, BUFLEN);
memcpy(szBuf, msgSend, sizeof(MyMsg) + len);
epollSend(node->sock, szBuf, sizeof(MyMsg) + len);
printf("epollSend Client %s, ip=%s, port=%d!\n", msgSend->szCmd,
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
free(msgSend);
msgSend = NULL;
}
}else if (strcmp(msgRecv->szCmd, "RESP_LINK") == 0){
printf("epoolRecv Client RESP_LINK, ip=%s, port=%d!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
}else if (strcmp(msgRecv->szCmd, "RE_EXIT") == 0){
printf("epoolRecv Client RE_EXIT, ip=%s, port=%d!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
msgSend = MsgProcess(msgRecv); /* 实际处理 */
if (msgSend != NULL){
len = msgSend->iLen;
msgSend->iLen = htonl(msgSend->iLen); /* 转换成网络字节序 */
memset(szBuf, 0, BUFLEN);
memcpy(szBuf, msgSend, sizeof(MyMsg) + len);
epollSend(node->sock, szBuf, sizeof(MyMsg) + len);
printf("epollSend Client %s, ip=%s, port=%d!\n", msgSend->szCmd,
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
free(msgSend);
msgSend = NULL;

}
close(node->sock);
events[i].data.fd = -1;
head = deleteSockNodeList(head, node->sock);
}else if (strcmp(msgRecv->szCmd, "RESP_TEXT") == 0){
printf("epoolRecv Client RESP_TEXT, ip=%s, port=%d!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port));
close(node->sock);
events[i].data.fd = -1;
head = deleteSockNodeList(head, node->sock);
}else if (strcmp(msgRecv->szCmd, "RE_DATA") == 0){
printf("epoolRecv Client RE_DATA, ip=%s, port=%d, data:%s!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port), msgRecv->szData);
msgSend = MsgProcess(msgRecv); /* 实际处理 */
if (msgSend != NULL){
len = msgSend->iLen;
msgSend->iLen = htonl(msgSend->iLen); /* 转换成网络字节序 */
memset(szBuf, 0, BUFLEN);
memcpy(szBuf, msgSend, sizeof(MyMsg) + len);
epollSend(node->sock, szBuf, sizeof(MyMsg) + len);
printf("epollSend Client %s, ip=%s, port=%d, data:%s!\n", msgSend->szCmd,
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port),
len > 0 ? msgSend->szData : "");
free(msgSend);
msgSend = NULL;
}
}else if (strcmp(msgRecv->szCmd, "RESP_DATA") == 0){
printf("epoolRecv Client RESP_DATA, ip=%s, port=%d, data:%s!\n",
inet_ntoa(node->addr.sin_addr), ntohs(node->addr.sin_port), msgRecv->szData);
}
}/* recv */
}/* if i = 0 */
}/*for */
}/* while 1 */

while (head != NULL){
node = head;
head = head->pNext;
close(node->sock);
free(node);
}
return 0;
}

[/code]

发表评论