多路复用I/O模型epoll() 模型 代码实现

epoll模型

int epoll_create(int maxevent) //创建一个epoll的句柄 然后maxevent表示监听的数目的大小
int epoll_ctl(int epollfd,int op,int fd,struct epoll_event *event) //epoll的事件注册函数
/*
epollfd 为epoll_create()的返回值
op 表示为相应的操作
fd 为套接字
*event 表示监听什么事件
*/
OP 有
EPOLL_CTL_ADD：注册新的fd到epfd中；
EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
EPOLL_CTL_DEL：从epfd中删除一个fd；

int epoll_wait(int epollfd,struct epoll_event *event,int maxevents,int timeout);
/*
epollfd epoll_create的返回值
*event 从内核得到时间的集合
maxevents 告诉内核event多大 不能大于epoll_create()中的maxevent
timeout 表示等待的时间-1表示阻塞 0表示立即返回
*/

epoll_event结构体

struct epoll_event {
__uint32_t events;  /* Epoll events */
epoll_data_t data;  /* User data variable */
};

typedef union epoll_data
{
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;

其中epoll_event 结构体中的events参数的值是宏定义：

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

data.h

#ifndef DATA_H
#define DATA_H
#include <string.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <sys/stat.h>
#include <stdlib.h>
#include <errno.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <assert.h>
#include <sys/types.h>

#define IP "127.0.0.1"
#define PORT 4578
#define maxn 1100
#define MAXLINE 100
#define BACKLOG 5
#define EPOLLEVENTS 1120
#define _EVENTS 100
#endif

server.c

#include "data.h"
static void epoll_accept(int sockfd);
static void submit_to_check(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf);
static void submit_to_read(int epollfd,int conn,char *buf);
static void submit_to_write(int epollfd,int conn,char *buf);
static void submit_to_accept(int epollfd,int conn);
static void change_event(int epollfd,int conn,int  ser_event);
static void change_delete(int epollfd,int conn,int ser_event);
static void add_event(int epollfd,int sockfd,int  ser_event);

static int init()
{
int sockfd;
struct sockaddr_in server_in;
if((sockfd = socket(AF_INET,SOCK_STREAM,0)) == -1)
{
fprintf(stderr,"socket fail,error %s",strerror(errno));
return -1;
}
bzero(&server_in,sizeof(server_in));
server_in.sin_family = AF_INET;
server_in.sin_port = htons(PORT);
inet_pton(AF_INET,IP,&server_in.sin_addr);
if(bind(sockfd,(struct sockaddr*)&server_in,sizeof(server_in)) == -1)
{
fprintf(stderr,"bind fail,error %s",strerror(errno));
return -1;
}
listen(sockfd,BACKLOG);
return sockfd;
}

static void epoll_accept(int sockfd)
{
int epollfd;
char buf[maxn];
memset(buf,0,sizeof(buf));
struct epoll_event events[_EVENTS];
epollfd = epoll_create(EPOLLEVENTS);
int ret;
add_event(epollfd,sockfd,EPOLLIN);
for(;;)
{
ret = epoll_wait(epollfd,events,_EVENTS,-1);
submit_to_check(epollfd,events,ret,sockfd,buf);
}
close(epollfd);
}
static void submit_to_check(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf)
{
int i;
int fd;
for(i=0;i<num;i++)
{
fd = events[i].data.fd;
if((fd == sockfd) && (events[i].events&EPOLLIN))
{
submit_to_accept(epollfd,sockfd);
}
else if(events[i].events & EPOLLIN)
{
submit_to_read(epollfd,fd,buf);
}
else if(events[i].events & EPOLLOUT)
{
submit_to_write(epollfd,fd,buf);
}
}
}
static void submit_to_read(int epollfd,int conn,char *buf)
{
int n;
n = read(conn,buf,maxn);
if(n == 0)
{
fprintf(stderr,"client close!\n");
close(conn);
change_delete(epollfd,conn,EPOLLIN);
}
else if(n == -1)
{
fprintf(stderr,"read error\n");
close(conn);
change_delete(epollfd,conn,EPOLLIN);
}
else
{
fprintf(stderr,"read message is %s\n",buf);
change_event(epollfd,conn,EPOLLOUT);
}
}
static void submit_to_write(int epollfd,int conn,char *buf)
{
int n;
n = write(conn,buf,strlen(buf));
if(n == -1)
{
printf("server closed!\n");
close(conn);
change_delete(epollfd,conn,EPOLLOUT);
}
else
{
change_event(epollfd,conn,EPOLLIN);
}
memset(buf,'\0',sizeof(buf));
}
static void submit_to_accept(int epollfd,int conn)
{
struct sockaddr_in server_in;
bzero(&server_in,sizeof(server_in));
socklen_t socklen;
int afd;
socklen = 1;
if((afd = accept(conn,(struct sockaddr*)&server_in,&socklen)) == -1)
{
fprintf(stderr,"accept fail,error %s\n",strerror(errno));
return;
}
else
{
printf("accept a new client:%s :%d\n",inet_ntoa(server_in.sin_addr),server_in.sin_port);
add_event(epollfd,afd,EPOLLIN);
}
}
static void change_event(int epollfd,int conn,int  ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = conn;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_MOD,conn,&eventfd);
}
static void change_delete(int epollfd,int conn,int ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = conn;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_DEL,conn,&eventfd);
}
static void add_event(int epollfd,int sockfd,int  ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = sockfd;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_ADD,sockfd,&eventfd);
}

int main()
{

int sockfd;
sockfd = init();
epoll_accept(sockfd);
close(sockfd);
return 0;
}

client.c

#include "data.h"

static void connect_server(int sockfd);
static void submit_to_check(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf);
static void submit_to_read(int epollfd,int fd,int sockfd,char *buf);
static void submit_to_write(int epollfd,int fd,int sockfd,char *buf);
static void change_event(int epollfd,int sockfd,int ser_event);
static void change_delete(int epollfd,int sockfd,int ser_event);
static void add_event(int epollfd,int sockfd,int ser_event);

static int init()
{
struct sockaddr_in client_in;
int sockfd;
if((sockfd = socket(AF_INET,SOCK_STREAM,0)) == -1)
{
fprintf(stderr,"socket fail,error %s\n",strerror(errno));
return -1;
}
bzero(&client_in,sizeof(client_in));
client_in.sin_family = AF_INET;
client_in.sin_port = htons(PORT);
inet_pton(AF_INET,IP,&client_in.sin_addr);
connect(sockfd,(struct sockaddr*)&client_in,sizeof(client_in)) ;
return sockfd;
}
static void connect_server(int sockfd)
{
int epollfd;
struct epoll_event events[_EVENTS];
epollfd = epoll_create(EPOLLEVENTS);
char buf[maxn];
int conn;
memset(buf,'\0',sizeof(buf));
add_event(epollfd,STDIN_FILENO,EPOLLIN);
for(;;)
{
fprintf(stderr,"Please input message!\n");
conn = epoll_wait(epollfd,events,_EVENTS,-1);
submit_to_check(epollfd,events,conn,sockfd,buf);
}
close(epollfd);
}
static void submit_to_check(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf)
{
int i;
int fd;
for(i=0;i<num;i++)
{
fd = events[i].data.fd;
if(events[i].events & EPOLLIN)
{
submit_to_read(epollfd,fd,sockfd,buf);
}
else if(events[i].events & EPOLLOUT)
{
submit_to_write(epollfd,fd,sockfd,buf);
}
}
}
static void submit_to_read(int epollfd,int fd,int sockfd,char *buf)
{
memset(buf,'\0',sizeof(buf));
int n;
n = read(fd,buf,maxn);
if(n == 0)
{
fprintf(stderr,"server close.\n");
close(fd);
}
else if(n == -1)
{
perror("read error:");
close(fd);
}
else
{
if(fd == STDIN_FILENO)
add_event(epollfd,sockfd,EPOLLOUT);
else
{
change_delete(epollfd,sockfd,EPOLLIN);
add_event(epollfd,STDOUT_FILENO,EPOLLOUT);
}
}
}
static void submit_to_write(int epollfd,int fd,int sockfd,char *buf)
{
int n;
n = write(fd,buf,strlen(buf));
if(n == -1)
{
perror("write error:");
close(fd);
}
else
{
if(fd == STDOUT_FILENO)
{
change_delete(epollfd,fd,EPOLLOUT);
}
else
{
change_event(epollfd,fd,EPOLLIN);
}
}
memset(buf,0,sizeof(buf));
}
static void change_event(int epollfd,int sockfd,int ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = sockfd;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_MOD,sockfd,&eventfd);
}
static void change_delete(int epollfd,int sockfd,int ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = sockfd;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_DEL,sockfd,&eventfd);
}

static void add_event(int epollfd,int sockfd,int ser_event)
{
struct epoll_event eventfd;
eventfd.data.fd = sockfd;
eventfd.events = ser_event;
epoll_ctl(epollfd,EPOLL_CTL_ADD,sockfd,&eventfd);
}
int main()
{
int sockfd = init();
connect_server(sockfd);
close(sockfd);
return 0;
}