UNIX环境高级编程学习之第十六章网络IPC：套接字 - 非阻塞的Socket通信EPoll模型(多路复用), 实用Socket通信模板

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]