3高并发服务器:多路IO之epoll

简介:  1 epoll epoll是Linux下多路复用IO接口select/poll的增强版本,它能显著提高程序在大量并、发连接中只有少量活跃的情况下的系统CPU利用率,因为它会复用文件描述符集合来传递结果而不用迫使开发者每次等待事件之前都必须重新准备要被侦听的文件描述符集合,另一点原因就是获取事件的时候,它无须遍历整个被侦听的描述符集,只要遍历那些被内核IO事件异步唤醒而加入


1 epoll

epollLinux下多路复用IO接口select/poll的增强版本,它能显著提高程序在大量并、发连接中只有少量活跃的情况下的系统CPU利用率,因为它会复用文件描述符集合来传递结果而不用迫使开发者每次等待事件之前都必须重新准备要被侦听的文件描述符集合,另一点原因就是获取事件的时候,它无须遍历整个被侦听的描述符集,只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。

目前epelllinux大规模并发网络程序中的热门首选模型。

epoll除了提供select/ poll那种IO事件的电平触发(Level Triggered)外,还提供了边沿触发(Edge Triggered),这就使得用户空间程序有可能缓存IO状态,减少epoll_wait/epoll_pwait的调用,提高应用程序效率。

一个进程打开大数目的socket描述符

cat/proc/sys/fs/file-max

设置最大打开文件描述符限制

sudo vi /etc/security/limits.conf

    epollAPI

1.创建一个epoll句柄,参数size用来告诉内核监听的文件描述符个数,跟内存大小有关。

依赖的头文件

#include <sys/epoll.h>

函数声明

int epoll_create(int size);

函数说明:

size:告诉内核监听的数目

 

2 控制某个epoll监听的文件描述符上的事件:注册、修改、删除

依赖的头文件

#include <sys/epoll.h>

函数声明

int epoll_ctl(int epfd,int op,int fd,structepoll_event);

函数说明:

epfd:epoll_create的句柄

op:表示动作,用3个宏来表示

EPOLL_CTL_ADD(注册新的fdepfd)

EPOLL_CTL_MOD(修改已经注册的fd的监听事件)

EPOLL_CTL_DEL(epfd删除一个fd)

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;

 

EPOLLIN :表示对应的文件描述符可以读(包括对端SOCKET正常关闭)

EPOLLOUT:表示对应的文件描述符可以写

EPOLLPRI:表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来)

EPOLLERR:表示对应的文件描述符发生错误

EPOLLHUP:表示对应的文件描述符被挂断;

EPOLLETEPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的

EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里 

3 等待所监控文件描述符上有事件的产生,类似select()调用。

依赖的头文件

#include <sys/epoll.h>

函数声明:

int epoll_wait(int epfd, struct epoll_event*events, int maxevents, int timeout);

参数介绍:

events:用来从内核得到事件的集合。

maxevents:告之内核这个events有多大,这个maxevents的值不能大于创建epoll_create()时的size,timeout:是超时时间

-1:阻塞

0:立即返回,非阻塞

    >0:指定微秒

返回值:成功返回有多少文件描述符就绪,时间到时返回0,出错返回-1

案例说明:

server.c

#include<stdio.h>

#include<stdlib.h>

#include<string.h>

#include<netinet/in.h>

#include<arpa/inet.h>

#include<sys/epoll.h>

#include<errno.h>

#include<unistd.h>

#include<ctype.h>

#include"wrap.h"

 

#define MAXLINE 80

#define SERV_PORT 8000

#define OPEN_MAX 1024

 

int main(void)

{

    int i,j,maxi,listenfd,connfd,sockfd;

    int nready,efd,res;

    ssize_t n;

    char buf[MAXLINE],str[INET_ADDRSTRLEN];

    socklen_t clilen;

    int client[OPEN_MAX];

struct sockaddr_in cliaddr,servaddr;

//ep[OPEN_MAX]保存就绪的文件描述符

    struct epoll_event tep,ep[OPEN_MAX];

 

    listenfd = Socket(AF_INET,SOCK_STREAM,0);

 

    bzero(&servaddr,sizeof(servaddr));

    servaddr.sin_family = AF_INET;

    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    servaddr.sin_port = htons(SERV_PORT);

 

    Bind(listenfd,(struct sockaddr *)&servaddr,sizeof(servaddr));

    Listen(listenfd,20);

 

    for(i = 0;i < OPEN_MAX;i++) {

        client[i] = -1;

    }

    maxi = -1;

 

    efd = epoll_create(OPEN_MAX);

 

    if(efd == -1)

        perr_exit("epoll_create");

  

    //监听读属性

tep.events = EPOLLIN;

//data里面保存了就绪的文件的文件描述符。

    tep.data.fd = listenfd;

    res = epoll_ctl(efd,EPOLL_CTL_ADD,listenfd,&tep);

    if(res == -1)

        perr_exit("epoll_ctl");

 

    for(;;) {

        /*阻塞监听*/

        nready = epoll_wait(efd,ep,OPEN_MAX,-1);

        if(nready == -1) {

            perr_exit("epoll_wait");

        }

        for(i = 0;i< nready;i++) {

            if(!ep[i].events & EPOLLIN)

                continue;

            if(ep[i].data.fd == listenfd) {

                clilen = sizeof(cliaddr);

                connfd = Accept(listenfd,(struct sockaddr *)&cliaddr,&clilen);

                printf("received from %s at PORT %d\n",

                        inet_ntop(AF_INET,&cliaddr.sin_addr,str,sizeof(str)),

                        ntohs(cliaddr.sin_port));

 

                for(j = 0; j < OPEN_MAX; j++) {

                    if(client[j] < 0) {

                        client[j] = connfd;  /*save descriptor*/

                        break;

                    }

                }

 

                if(j==OPEN_MAX)

                    perr_exit("too many clients");

 

                if(j > maxi)

                    maxi = j;  /*max index in client[] array*/

                tep.events = EPOLLIN;

                tep.data.fd = connfd;

                res = epoll_ctl(efd,EPOLL_CTL_ADD,connfd,&tep);

                if(res == -1) {

                    perr_exit("epoll_ctl");

                } else {

                    sockfd = ep[i].data.fd;

                    n = Read(sockfd,buf,MAXLINE);

                    if(n==0) {

                        for(j = 0; j <= maxi;j++) {

                            if(client[j] == sockfd) {

                                client[j] = -1;

                                break;

                            }

                        }

                        res = epoll_ctl(efd,EPOLL_CTL_DEL,sockfd,NULL);

                        if(res == -1) {

                            perr_exit("epoll_ctl");

                        }

                        Close(sockfd);

                        printf("client[%d] closed connection",j);

                    } else {

                        for(j = 0;j < n;j++)

                            buf[j] = toupper(buf[j]);

                        Writen(sockfd,buf,n);

                    }

                }

            }

        }

    }

    Close(listenfd);

    Close(efd);

    return 0;

}

client.c

#include<stdio.h>

#include<string.h>

#include<unistd.h>

#include<arpa/inet.h>

#include<netinet/in.h>

#include"wrap.h"

 

#define MAXLINE 80

#define SERV_PORT 8000

 

int main(void)

{

    struct sockaddr_in servaddr;

    char buf[MAXLINE];

    int sockfd,n;

 

    sockfd = Socket(AF_INET,SOCK_STREAM,0);

   

    bzero(&servaddr,sizeof(servaddr));

    servaddr.sin_family = AF_INET;

    inet_pton(AF_INET,"127.0.0.1",&servaddr.sin_addr);

    servaddr.sin_port = htons(SERV_PORT);

 

    Connect(sockfd,(struct sockaddr *) &servaddr,sizeof(servaddr));

 

    while(fgets(buf,MAXLINE,stdin) != NULL) {

        Write(sockfd,buf,strlen(buf));

        n = Read(sockfd,buf,MAXLINE);

        if(n == 0)

            printf("the other side has been closed.\n");

        else

            Write(STDOUT_FILENO,buf,n);

    }

 

    Close(sockfd);

    return 0;

}

wrap.h

#ifndef __WRAP_H_

#define __WRAP_H_

 

void perr_exit(const char *s);

int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr);

void Bind(int fd, const struct sockaddr *sa, socklen_t salen);

void Connect(int fd, const struct sockaddr *sa, socklen_t salen);

void Listen(int fd, int backlog);

int Socket(int family, int type, int protocol);

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

ssize_t Write(int fd, const void *ptr, size_t nbytes);

void Close(int fd);

ssize_t Readn(int fd, void *vptr, size_t n);

ssize_t Writen(int fd, const void *vptr, size_t n);

static ssize_t my_read(int fd, char *ptr);

ssize_t Readline(int fd, void *vptr, size_t maxlen);

 

#endif

wrap.c

#include <stdlib.h>

#include <stdio.h>

#include <errno.h>

#include <unistd.h>

#include <sys/socket.h>

 

void perr_exit(const char *s)

{

         perror(s);

         exit(1);

}

 

int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr)

{

         int n;

 

again:

         if ( (n = accept(fd, sa, salenptr)) < 0) {

                   if ((errno == ECONNABORTED) || (errno == EINTR))

                            goto again;

                   else

                            perr_exit("accept error");

         }

         return n;

}

 

void Bind(int fd, const struct sockaddr *sa, socklen_t salen)

{

         if (bind(fd, sa, salen) < 0)

                   perr_exit("bind error");

}

 

void Connect(int fd, const struct sockaddr *sa, socklen_t salen)

{

         if (connect(fd, sa, salen) < 0)

                   perr_exit("connect error");

}

 

void Listen(int fd, int backlog)

{

         if (listen(fd, backlog) < 0)

                   perr_exit("listen error");

}

 

int Socket(int family, int type, int protocol)

{

         int n;

 

         if ( (n = socket(family, type, protocol)) < 0)

                   perr_exit("socket error");

         return n;

}

 

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

{

         ssize_t n;

 

again:

         if ( (n = read(fd, ptr, nbytes)) == -1) {

                   if (errno == EINTR)

                            goto again;

                   else

                            return -1;

         }

         return n;

}

 

ssize_t Write(int fd, const void *ptr, size_t nbytes)

{

         ssize_t n;

 

again:

         if ( (n = write(fd, ptr, nbytes)) == -1) {

                   if (errno == EINTR)

                            goto again;

                   else

                            return -1;

         }

         return n;

}

 

void Close(int fd)

{

         if (close(fd) == -1)

                   perr_exit("close error");

}

ssize_t Readn(int fd, void *vptr, size_t n)

{

         size_t  nleft;

         ssize_t nread;

         char   *ptr;

 

         ptr = vptr;

         nleft = n;

         while (nleft > 0) {

                   if ( (nread = read(fd, ptr, nleft)) < 0) {

                            if (errno == EINTR)

                                     nread = 0;

                            else

                                     return -1;

                   } else if (nread == 0)

                            break;

 

                   nleft -= nread;

                   ptr += nread;

         }

         return n - nleft;

}

 

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;

                            else

                                     return -1;

                   }

 

                   nleft -= nwritten;

                   ptr += nwritten;

         }

         return n;

}

static ssize_t my_read(int fd, char *ptr)

{

         static int read_cnt;

         static char *read_ptr;

         static char read_buf[100];

 

         if (read_cnt <= 0) {

again:

                   if ( (read_cnt = read(fd, read_buf, sizeof(read_buf))) < 0) {

                            if (errno == EINTR)

                                     goto again;

                            return -1;

                   } else if (read_cnt == 0)

                            return 0;

                   read_ptr = read_buf;

         }

         read_cnt--;

         *ptr = *read_ptr++;

         return 1;

}

 

ssize_t Readline(int fd, void *vptr, size_t maxlen)

{

         ssize_t n, rc;

         char    c, *ptr;

 

         ptr = vptr;

         for (n = 1; n < maxlen; n++) {

                   if ( (rc = my_read(fd, &c)) == 1) {

                            *ptr++ = c;

                            if (c  == '\n')

                                     break;

                   } else if (rc == 0) {

                            *ptr = 0;

                            return n - 1;

                   } else

                            return -1;

         }

         *ptr  = 0;

         return n;

}

目录
相关文章
|
1月前
|
网络协议 安全 测试技术
手撕测试tcp服务器效率工具——以epoll和io_uring对比为例
手撕测试tcp服务器效率工具——以epoll和io_uring对比为例
64 2
|
1月前
|
存储 网络协议
TCP服务器 IO多路复用的实现:select、poll、epoll
TCP服务器 IO多路复用的实现:select、poll、epoll
43 0
|
1月前
|
API C++
socket编程之常用api介绍与socket、select、poll、epoll高并发服务器模型代码实现(1)
前言   本文旨在学习socket网络编程这一块的内容,epoll是重中之重,后续文章写reactor模型是建立在epoll之上的。
51 0
|
1月前
|
监控 安全 Linux
socket编程之常用api介绍与socket、select、poll、epoll高并发服务器模型代码实现(3)
高并发服务器模型-poll poll介绍   poll跟select类似, 监控多路IO, 但poll不能跨平台。其实poll就是把select三个文件描述符集合变成一个集合了。
43 0
|
1月前
|
Java 关系型数据库 MySQL
Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
【2月更文挑战第33天】Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
122 2
|
1月前
|
存储 负载均衡 监控
epoll服务器百万并发测试
epoll服务器百万并发测试
36 1
|
1月前
|
前端开发 Java API
构建异步高并发服务器:Netty与Spring Boot的完美结合
构建异步高并发服务器:Netty与Spring Boot的完美结合
|
1月前
|
网络协议 Linux
【Linux C TCP服务器端-epoll案例】
本文主要介绍了linux下Select的TCP通信流程,实现了客户端和服务器的通信,主要实现了消息的回发,即服务器将消息原封不动的回发给客户端。
42 0
|
1月前
【基于C++HTTP 服务器的epoll 改造】
【基于C++HTTP 服务器的epoll 改造】

热门文章

最新文章