在写epoll回显服务器代码之前，可以先看看上一篇文章：select poll epoll三者之间的比较。最近在继续学习网络编程中的服务端编程中，了解到很多网游服务器是在IOMP（IO完成端口）框架下写的，但是这种方式只能在 Windows 下使用，奇了怪了，这么好的东西为什么不在Linux下也实现一套呢？这个问题我继续学习IOMP再来谈一谈！
epoll是linux下高并发服务器的完美方案，因为是基于事件触发的，所以比select快的不只是一个数量级。单线程epoll，触发量可达到15000，但是加上业务后，因为大多数业务都与数据库打交道，所以就会存在阻塞的情况，这个时候就必须用多线程来提速。业务在线程池内，这里要加锁才行。测试结果2300个/s
在Linux下，一个高效率一点的多线程并发服务器可以使用epoll来实现，我记得有一个库 libevent 好像就是在 epoll 基础上实现的。
epoll的两种工作方式：
1. LT
LT(level triggered)是 epoll 缺省的工作方式，并且同时支持 block 和 no-block socket.
在这种做法中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的 fd 进行 IO 操作。如果你不作任何操作，内核还是会继续通知你的，所以，这种模式编程出错误可能性要小一点。传统的 select/poll 都是这种模型的代表。
2. ET
ET (edge-triggered)是高速工作方式，只支持no-block socket，它效率要比LT更高。
ET与LT的区别在于，当 一个新的事件到来时，ET模式下当然可以从 epoll_wait 调用中获取到这个事件，可是如果这次没有把这个事件对应的套接字缓冲区处理完，在这个套接字中没有新的事件再次到来时，在ET模式下是无法再次从epoll_wait调用中获取这个事件的。
而LT模式正好相反，只要一个事件对应的套接字缓冲区还有数据，就总能从epoll_wait中获取这个事件。
因此，LT模式下开发基于epoll的应用要简单些，不太容易出错。而在ET模式下事件发生时，如果没有彻底地将缓冲区数据处理完，则会导致缓冲区中的用户请求得不到响应。
好，废话不多说，为了方便以后使用基于 epoll 的服务器，把代码贴在这里，方便以后查找！
/*************************************************************************
> File Name: epoll_echosrv.c
> Author: huabo
> Mail: bohua1@126.com
> Created Time: Sun 05 Oct 2014 08:27:06 PM HKT
************************************************************************/ #include<stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h> #include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/epoll.h> #include <pthread.h> #define MAXSIZE 5000
#define MAXLINE 10
#define OPEN_MAX 10
#define LISTENQ 20
#define INFTIM 1000 #define ERR_EXIT(m)\
do\
{\
perror(m);\
exit(EXIT_FAILURE);\
}while() //record the socket descriptor which need to read or write.
struct task
{
int fd;
struct task *next;
}; struct user_data
{
int fd;
unsigned int n_size;
char line[MAXLINE];
}; void * readtask(void *args);
void * writetask(void *args); struct epoll_event ev, events[];
int epfd; pthread_mutex_t mutex;
pthread_cond_t cond1; struct task *readhead = NULL, *readtail = NULL, *writehead = NULL; void setnonblocking(int sockfd)
{
int opts;
opts = fcntl(sockfd, F_GETFL);
if(opts < )
{
ERR_EXIT("fcntl");
}
opts = (opts | O_NONBLOCK);
opts = fcntl(sockfd, F_SETFL, opts);
if(opts < )
{
ERR_EXIT("fcntl");
}
} int main()
{
int listenfd;
//read and write thread ID
pthread_t tid1, tid2;
struct task *new_task = NULL;
struct user_data *rdata = NULL; pthread_mutex_init(&mutex, NULL);
pthread_cond_init(&cond1, NULL); //initial id of read thread and write thread
pthread_create(&tid1, NULL, readtask, NULL);
pthread_create(&tid2, NULL, writetask, NULL); if( (listenfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < )
{
ERR_EXIT("socket");
}
setnonblocking(listenfd); struct sockaddr_in servaddr;
memset(&servaddr, , sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_port = htons();
servaddr.sin_addr.s_addr = htonl(INADDR_ANY); //reuse address
int on = ;
if(setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)))
{
ERR_EXIT("setsockopt");
} //bind port
if(bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < )
{
ERR_EXIT("bind");
} //listen
if(listen(listenfd, SOMAXCONN) < )
{
ERR_EXIT("listen");
} struct sockaddr_in peeraddr;
socklen_t peerlen = sizeof(peeraddr);
int connfd;
int sockfd; //epoll server
epfd = epoll_create(); //step 1: create descriptor for events
struct epoll_event ev, events[];
ev.data.fd = listenfd;
ev.events = EPOLLIN|EPOLLET;
epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev); //step 2: add descriptor and events int nfds;
int i;
while()
{
nfds = epoll_wait(epfd, events, , -);
for(i = ; i < nfds; ++i)
{
if(events[i].data.fd == listenfd)
{
printf("listen = %d\n", events[i].data.fd);
connfd = accept(listenfd, (struct sockaddr *)(&peeraddr), &peerlen);
if(connfd < )
{
ERR_EXIT("accept");
}
//set non-block when use epoll
setnonblocking(connfd); char *str = inet_ntoa(peeraddr.sin_addr);
printf("connect from >> %s %d\n", str, connfd); //add new file descriptor
ev.data.fd = connfd;
ev.events = EPOLLIN|EPOLLET;
epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev); //reset the listen descriptor
ev.data.fd = listenfd;
ev.events = EPOLLIN|EPOLLET; epoll_ctl(epfd, EPOLL_CTL_MOD, listenfd, &ev);
}
else if(events[i].events & EPOLLIN) //there is data to read.
{
if((sockfd = events[i].data.fd) <= )
{
printf("file descriptor error\n");
continue;
} new_task = (struct task*)malloc(sizeof(struct task));
new_task->fd = sockfd;
new_task->next = NULL; //add to the read task queue.
pthread_mutex_lock(&mutex);
if(readhead == NULL)
{
readhead = new_task;
readtail = new_task;
}
else
{
readtail->next = new_task;
readtail = new_task;
} //wake up the thread wait for cond1
pthread_cond_broadcast(&cond1);
pthread_mutex_unlock(&mutex);
}
else if(events[i].events & EPOLLOUT) //there is data to write
{
//something needed to be done here.
rdata = (struct user_data*)events[i].data.ptr;
sockfd = rdata->fd;
write(sockfd, rdata->line, rdata->n_size);
free(rdata); ev.data.fd = sockfd;
ev.events = EPOLLIN|EPOLLET;
epoll_ctl(epfd, EPOLL_CTL_MOD, sockfd, &ev);
}
else if( (events[i].events & EPOLLHUP) || (events[i].events & EPOLLERR) || !(events[i].events & EPOLLIN))
{
//An error has occured on this fd, or the socket is not ready for reading(why were we notified then?)
fprintf(stderr, "epoll error\n");
close(events[i].data.fd);
continue;
}
}
}
return ;
} static int count111 = ;
static time_t oldtime = , nowtime = ; void* readtask(void *args)
{
printf("read task begin\n");
int fd = -;
unsigned int n;
struct user_data *data = NULL;
while()
{
pthread_mutex_lock(&mutex);
while(readhead == NULL)
pthread_cond_wait(&cond1, &mutex); fd = readhead->fd; struct task *tmp = readhead;
readhead = readhead->next;
free(tmp); pthread_mutex_unlock(&mutex);
data = (struct user_data*)malloc(sizeof(struct user_data));
data->fd = fd; char recvBuf[] = {};
int ret = ;
int rs = ; while(rs)
{
ret = recv(fd, recvBuf, , );
if(ret < )
{
if(errno == EAGAIN)
{
printf("EAGAIN\n");
break;
}
else
{
printf("recv error!\n");
close(fd);
break;
}
}
else if(ret == )
{
printf("client close.\n");
rs = ;
} if(ret == sizeof(recvBuf))
{
//need to recv again.
rs = ;
write(fd, recvBuf, ret);
fputs(recvBuf, stdout);
}
else
{
write(fd, recvBuf, ret);
fputs(recvBuf, stdout);
rs = ;
} } if(ret > )
{
data->n_size = n;
count111++; struct tm *today;
time_t ltime;
time(&nowtime); if(nowtime != oldtime)
{
printf("%d\n", count111);
oldtime = nowtime;
count111 = ;
} char buf[] = {};
sprintf(buf, "HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s", "Hello world!\n");
send(fd, buf, strlen(buf), );
close(fd);
}
}
} void * writetask(void *args)
{ }