前言

• 掌握select編程模型,能夠?qū)崿F(xiàn)select版本的TCP服務(wù)器.
• 掌握poll編程模型,能夠?qū)崿F(xiàn)poll版本的TCP服務(wù)器.
• 掌握epoll的編程模型,能夠?qū)崿F(xiàn)epoll版本的TCP服務(wù)器.
• epoll的LT模式和ET模式.
• 理解select和epoll的優(yōu)缺點(diǎn)對(duì)比.

提示：以下是本篇文章正文內(nèi)容，下面案例可供參考

一、IO多路轉(zhuǎn)接select

多路轉(zhuǎn)接天然的是讓我們可以依次等待多個(gè)文件描述符.

什么叫做文件描述符狀態(tài)的變化 —>1.可讀 2.可寫 3.異常

初始select

系統(tǒng)提供select函數(shù)來(lái)實(shí)現(xiàn)多路復(fù)用輸入/輸出模型.

• select系統(tǒng)調(diào)用是用來(lái)讓我們的程序堅(jiān)實(shí)多個(gè)文件描述符的狀態(tài)變化的;
• 程序會(huì)停在select這里等待,直到被監(jiān)視的文件描述符有一個(gè)或者多個(gè)發(fā)生了狀態(tài)變化;

select函數(shù)原型

如下所示:

參數(shù)解釋:

• 參數(shù)ndfs是需要監(jiān)視的最大的文件描述符值+1;
• rdset,wrset,exset分別對(duì)應(yīng)與需要檢測(cè)的可讀文件描述符的集合,可寫文件描述符的集合和異常文件描述符的集合;
• 參數(shù)timeout為結(jié)構(gòu)timeval,用來(lái)設(shè)置select()的等待時(shí)間.

參數(shù)timeout取值:

• NULL:表示select()沒(méi)有timeout,select將一直被阻塞,直到某個(gè)文件描述符上發(fā)生了事件;
• 0:僅檢測(cè)描述符集合的狀態(tài),然后立即返回,并不等待外部事件的發(fā)生;
• 特定的時(shí)間值:如果在指定的時(shí)間段里沒(méi)有事件發(fā)生,select將超時(shí)返回.

關(guān)于fd_set結(jié)構(gòu)

fd_set是一種位圖結(jié)構(gòu).**比特位的位置代表fd的編號(hào),比特位的內(nèi)容代表(就緒/未就緒)**的概念.

提供了一組fd_set的接口,來(lái)比較方便的操作位圖.

void FD_CLR(int fd, fd_set *set); // 用來(lái)清除描述詞組set中相關(guān)fd 的位
 int FD_ISSET(int fd, fd_set *set); // 用來(lái)測(cè)試描述詞組set中相關(guān)fd 的位是否為真
 void FD_SET(int fd, fd_set *set); // 用來(lái)設(shè)置描述詞組set中相關(guān)fd的位
 void FD_ZERO(fd_set *set); // 用來(lái)清除描述詞組set的全部位

查看fd_set的大小

因?yàn)閒d_set是位圖結(jié)構(gòu),求出來(lái)的結(jié)果是128字節(jié),所以對(duì)于bit為就是1024.

關(guān)于timeval結(jié)構(gòu)

timeval結(jié)構(gòu)用于描述一段時(shí)間長(zhǎng)度,如果在這個(gè)時(shí)間內(nèi),需要監(jiān)視的描述符沒(méi)有事件發(fā)生則函數(shù)返回,返回值為0.

函數(shù)返回值

• 執(zhí)行成功則返回文件描述詞狀態(tài)已改變的個(gè)數(shù)
• 如果返回0代表在描述詞狀態(tài)改變前已超過(guò)timeout時(shí)間,沒(méi)有返回
• 當(dāng)有錯(cuò)誤發(fā)生時(shí)則返回-1,錯(cuò)誤原因存于errno,此時(shí)參數(shù)readfds,writefds,exceptfds和timeout的值變成不可預(yù)測(cè).

錯(cuò)誤值可能為:

• EBADF文件描述詞為無(wú)效的或該文件已關(guān)閉
• EINTR此調(diào)用被信號(hào)所中斷
• EINVAL參數(shù)n為負(fù)值.
• ENOMEM核心內(nèi)存不足,

select使用示例

makefile

SelectServer:SelectServer.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm SelectServer

Sock.hpp

#include <iostream>
#include <string>
#include <cassert>
#include <unistd.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <fstream>
using namespace std;

class Sock
{
public:
    static const int gBackLog = 20;
    static int Socket()
    {
        // 1.創(chuàng)建socket
        int _listenSock = socket(AF_INET, SOCK_STREAM, 0);
        if (_listenSock < 0)
        {
            exit(1);
        }
        int opt = 1;
        setsockopt(_listenSock,SOL_SOCKET,SO_REUSEADDR | SO_REUSEPORT, &opt,sizeof opt);
    }
    static void Bind(int socket, uint16_t _port)
    {
        // 2.bind綁定
        // 2.1填充服務(wù)器
        struct sockaddr_in local; // 用戶棧
        memset(&local, 0, sizeof local);
        local.sin_family = AF_INET;
        local.sin_port = htons(_port);
        local.sin_addr.s_addr = INADDR_ANY;
        // 2.2本地socket信息，寫入_sock對(duì)應(yīng)的內(nèi)核區(qū)域
        if (bind(socket, (const sockaddr *)&local, sizeof local) < 0)
        {
            exit(2);
        }
    }
    static void Listen(int socket)
    {
        // 3.監(jiān)聽(tīng)socket，為何要監(jiān)聽(tīng)呢？tcp是面向連接的！
        if (listen(socket, gBackLog) < 0)
        {
            exit(3);
        }
        // 允許別人來(lái)連接你了
    }
    static int Accept(int socket,string* clientip,uint16_t* clientport)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int serviceSock = accept(socket, (struct sockaddr *)&peer, &len);
        if (serviceSock < 0)
        {
            return -1;
        }
        if(clientport) *clientport = ntohs(peer.sin_port);
        if(clientip) *clientip = inet_ntoa(peer.sin_addr);
        return serviceSock;
    }
};

SelectServer.cc

#include <iostream>
#include <sys/select.h>
#include <string>
#include "Sock.hpp"

using namespace std;

// 保存歷史上所有的合法fd
#define NUM (sizeof(fd_set) * 8)
int fdsArray[NUM] = {0};
#define DFL -1

static void Usage(string process)
{
    cerr << "\nUsage: " << process << "port\n"
         << endl;
}

static void showArray(int arr[], int num)
{
    cout << "當(dāng)前合法sock list # ";
    for (int i = 0; i < num; ++i)
    {
        if (arr[i] == DFL)
            continue;
        else
            cout << arr[i] << " ";
    }
    cout << endl;
}

// redfds : 現(xiàn)在包含的就是已經(jīng)就緒的sock
static void HandlerEvent(int listensock, fd_set &readfds)
{
    for (int i = 0; i < NUM; ++i)
    {
        if (fdsArray[i] == DFL)
            continue;
        if (i == 0 && fdsArray[i] == listensock)
        {
            // 如何得知那些fd上面的事件就緒了呢?
            if (FD_ISSET(listensock, &readfds))
            {
                // 具有了一個(gè)新鏈接
                cout << "已經(jīng)有一個(gè)新鏈接到來(lái)了,需要進(jìn)行獲取了..." << endl;
                string ip;
                uint16_t port;
                int sock = Sock::Accept(listensock, &ip, &port); // 這里不會(huì)阻塞
                if (sock < 0)
                    return;
                cout << "獲取新鏈接成功 : " << ip << " : " << port << " sock: " << sock << endl;
                // read/wirte --- 不能調(diào)用,因?yàn)槟鉹ead不知道底層數(shù)據(jù)是否就緒!!!
                // secelt知道! 想辦法把新的fd托管給select
                int i = 0;
                for (; i < NUM; ++i)
                {
                    if (fdsArray[i] == DFL)
                        break;
                }
                if (i == NUM)
                {
                    cerr << "我的服務(wù)器已經(jīng)到了最大的上限了,無(wú)法在承載更多的連接了..." << endl;
                    close(sock);
                }
                else
                {
                    // 將sock添加到select中,進(jìn)一步的監(jiān)聽(tīng)就緒事件了!
                    fdsArray[i] = sock;
                    showArray(fdsArray, NUM);
                }
            }
        }
        else
        {
            // 處理普通sock的IO事件!
            if (FD_ISSET(fdsArray[i], &readfds))
            {
                // 一定是一個(gè)合法的普通的IO類sock就緒了
                // read/recv讀取即可
                char buffer[1024];
                ssize_t s = recv(fdsArray[i], buffer, sizeof buffer, 0);
                if (s > 0)
                {
                    buffer[s] = 0;
                    cout << "client[" << fdsArray[i] << "]# " << buffer << endl;
                }
                else if (s == 0)
                {
                    cout << "client[" << fdsArray[i] << "] quit,server close: " << fdsArray[i] << endl;
                    close(fdsArray[i]);
                    fdsArray[i] = DFL; // 取出對(duì)該文件描述符的select的事件監(jiān)聽(tīng)
                }
                else
                {
                    cout << "client[" << fdsArray[i] << "] quit,server close: " << fdsArray[i] << endl;
                    close(fdsArray[i]);
                    fdsArray[i] = DFL; // 取出對(duì)該文件描述符的select的事件監(jiān)聽(tīng)
                    showArray(fdsArray, NUM);
                }
            }
        }
    }
}

// ./SelectServer 8080
// 只關(guān)心讀事件
int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        exit(1);
    }
    // fd_set fds; // fa_set是用位圖表示多個(gè)fd的.
    // cout<<sizeof(fd_set)<<endl;

    int listensock = Sock::Socket();
    Sock::Bind(listensock, 8080);
    Sock::Listen(listensock);

    for (int i = 0; i < NUM; ++i)
        fdsArray[i] = DFL;
    fdsArray[0] = listensock;

    while (true)
    {
        int maxFd = -1;
        fd_set readfds;
        FD_ZERO(&readfds);
        for (int i = 0; i < NUM; ++i)
        {
            if (fdsArray[i] == DFL) // 過(guò)濾不合法的fd
                continue;
            FD_SET(fdsArray[i], &readfds); // 添加所有的合法的fd到readfds中,方便select統(tǒng)一進(jìn)行就緒監(jiān)聽(tīng)
            if (maxFd < fdsArray[i])       // 更新最大值
                maxFd = fdsArray[i];
        }

        struct timeval timeout = {5, 0};
        // 如何看待監(jiān)聽(tīng)socket, 獲取新鏈接的, 本質(zhì)是需要先三次握手!
        // 前提是給我發(fā)送syn -? 建立連接的本質(zhì),其實(shí)也是IO,一個(gè)建立好的連接,我們成為讀時(shí)間就緒!
        // accept: 等+"數(shù)據(jù)拷貝"
        // string ip;
        // uint16_t port;
        // int sock = Sock::Accept(listensock,&ip,&port);

        // 編寫多路轉(zhuǎn)接代碼時(shí),必須先保證條件就緒了,才能調(diào)用IO類函數(shù)!
        int n = select(maxFd + 1, &readfds, nullptr,
                       nullptr, &timeout);

        switch (n)
        {
        case 0:
            cout << "time out ..." << (unsigned long long)time(nullptr) << endl;
            break;
        case -1:
            cerr << errno << " : " << strerror(errno) << endl;
            break;
        default:
            // 等待成功
            // 1.剛啟動(dòng)的時(shí)候,只有一個(gè)fd,listenscok
            // 2.server 運(yùn)行的時(shí)候,sock才會(huì)慢慢變多
            // 3.select 使用位圖,采用輸入輸出型參數(shù)的方式,來(lái)進(jìn)行 內(nèi)核<->用戶 信息的傳遞.
            // 每一次調(diào)用select,都需要對(duì)歷史數(shù)據(jù)和sock進(jìn)行重新設(shè)置!
            // 4.listensock, 永遠(yuǎn)都要被設(shè)置進(jìn)readfds中!
            // 5.select就緒的時(shí)候,可能是listen就緒,也可能是普通的IO sock就緒!!
            HandlerEvent(listensock, readfds);
            break;
        }
    }

    return 0;
}

select編碼特征

1. select之前要進(jìn)行所有參數(shù)的重置.之后要遍歷所有合法fd進(jìn)行事件監(jiān)測(cè).
2. select需要用戶自己維護(hù)第三方數(shù)組,來(lái)保存所有的合法fd,方便select批量處理.
3. 一旦特點(diǎn)的fd事件就緒,本次的讀取或者寫入不會(huì)被阻塞.

select優(yōu)缺點(diǎn)

優(yōu)點(diǎn):占用資源少,并且高效.對(duì)比之前的多線程,多進(jìn)程.

缺點(diǎn):每一次都要進(jìn)行大量的重置工作,效率比較低.
每一次能夠檢測(cè)的fd數(shù)量是有上限的.
每一次都需要內(nèi)核到用戶,用戶到內(nèi)核傳遞位圖參數(shù),有較為大量的數(shù)據(jù)拷貝
select編碼特別不方便,需要用戶自己維護(hù)數(shù)組
select底層需要遍歷的方式,檢測(cè)所需要檢測(cè)的fd

二、IO多路轉(zhuǎn)接poll

poll函數(shù)接口

參數(shù)說(shuō)明:

• fds是一個(gè)poll函數(shù)監(jiān)聽(tīng)的結(jié)構(gòu)列表.每一個(gè)元素中,包含了三部分內(nèi)容:文件描述符,監(jiān)聽(tīng)事件集合,返回的事件集合.
• nfds表示fds數(shù)組的長(zhǎng)度.
• timeout表示poll函數(shù)的超時(shí)時(shí)間,單位是毫秒(ms).

events和revents的取值

返回結(jié)果

• 返回值小于0,表示出錯(cuò).
• 返回值等于0,表示poll函數(shù)等待超時(shí).
• 返回值大于0,表示poll由于監(jiān)聽(tīng)的文件描述符就緒而返回.

socket就緒的條件

和select相同.

poll的優(yōu)點(diǎn)

不同于select使用三個(gè)位圖來(lái)表示三個(gè)fdset的方式,poll使用一個(gè)pollfd的指針實(shí)現(xiàn).

• pollfd結(jié)構(gòu)包含了要監(jiān)視的event和發(fā)生的event,不再使用select"參數(shù)-值"傳遞的方式.接口使用比select更方便.
• poll并沒(méi)有最大文件描述符數(shù)量限制(但是數(shù)量過(guò)大后性能還是會(huì)下降).

poll的缺點(diǎn)

poll監(jiān)聽(tīng)的文件描述符數(shù)目增多時(shí)

• 和select函數(shù)一樣,poll返回后,需要輪詢pollfd來(lái)獲取就緒的文件描述符.
• 每次調(diào)用poll都需要把大量的pollfd結(jié)構(gòu)從用戶態(tài)拷貝到內(nèi)核中.
• 同時(shí)連接的大量客戶端在同一時(shí)刻可能只有很少的處于就緒狀態(tài),因此隨著監(jiān)視的描述符數(shù)量的增長(zhǎng),其效率也會(huì)線性下降.

poll示例

makefile

PollServer:PollServer.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm PollServer

#include <iostream>
#include <poll.h>
#include <string>
#include "Sock.hpp"

using namespace std;

// 保存歷史上所有的合法fd
#define NUM 1024
struct pollfd fdsArray[NUM] = {0};
#define DFL -1


// 意義:不光是網(wǎng)絡(luò)sock,本地的文件描述符也可以被托管給多路轉(zhuǎn)接,那么后面的文件操作,管道等也可以直接對(duì)接到多路轉(zhuǎn)接!!!


static void Usage(string process)
{
    cerr << "\nUsage: " << process << "port\n"
         << endl;
}

static void showArray(int arr[], int num)
{
    cout << "當(dāng)前合法sock list # ";
    for (int i = 0; i < num; ++i)
    {
        if (arr[i] == DFL)
            continue;
        else
            cout << arr[i] << " ";
    }
    cout << endl;
}

// redfds : 現(xiàn)在包含的就是已經(jīng)就緒的sock
static void HandlerEvent(int listensock)
{
    for (int i = 0; i < NUM; ++i)
    {
        if (fdsArray[i].fd == DFL)
            continue;
        if (i == 0 && fdsArray[i].fd == listensock)
        {
            // 如何得知那些fd上面的事件就緒了呢?
            if (fdsArray[i].revents & POLLIN)
            {
                // 具有了一個(gè)新鏈接
                cout << "已經(jīng)有一個(gè)新鏈接到來(lái)了,需要進(jìn)行獲取了..." << endl;
                string ip;
                uint16_t port;
                int sock = Sock::Accept(listensock, &ip, &port); // 這里不會(huì)阻塞
                if (sock < 0)
                    return;
                cout << "獲取新鏈接成功 : " << ip << " : " << port << " sock: " << sock << endl;
                // read/wirte --- 不能調(diào)用,因?yàn)槟鉹ead不知道底層數(shù)據(jù)是否就緒!!!
                // secelt知道! 想辦法把新的fd托管給select
                int i = 0;
                for (; i < NUM; ++i)
                {
                    if (fdsArray[i].fd == DFL)
                        break;
                }
                if (i == NUM)
                {
                    cerr << "我的服務(wù)器已經(jīng)到了最大的上限了,無(wú)法在承載更多的連接了..." << endl;
                    close(sock);
                }
                else
                {
                    // 將sock添加到select中,進(jìn)一步的監(jiān)聽(tīng)就緒事件了!
                    fdsArray[i].fd = sock;
                    fdsArray[i].events = POLLIN;
                    fdsArray[i].revents = 0;
                }
            }
        }
        else
        {
            // 處理普通sock的IO事件!
            if (fdsArray[i].revents & POLLIN)
            {
                // 一定是一個(gè)合法的普通的IO類sock就緒了
                // read/recv讀取即可
                char buffer[1024];
                ssize_t s = recv(fdsArray[i].fd, buffer, sizeof buffer, 0);
                if (s > 0)
                {
                    buffer[s] = 0;
                    cout << "client[" << fdsArray[i].fd << "]# " << buffer << endl;
                }
                else if (s == 0)
                {
                    cout << "client[" << fdsArray[i].fd << "] quit,server close: " << fdsArray[i].fd << endl;
                    close(fdsArray[i].fd);
                    fdsArray[i].fd = DFL; // 取出對(duì)該文件描述符的select的事件監(jiān)聽(tīng)
                }
                else
                {
                    cout << "client[" << fdsArray[i].fd << "] quit,server error: " << fdsArray[i].fd << endl;
                    close(fdsArray[i].fd);
                    fdsArray[i].fd = DFL; // 取出對(duì)該文件描述符的select的事件監(jiān)聽(tīng)
                }
            }
        }
    }
}

// ./SelectServer 8080
// 只關(guān)心讀事件
int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        exit(1);
    }
    // fd_set fds; // fa_set是用位圖表示多個(gè)fd的.
    // cout<<sizeof(fd_set)<<endl;

    int listensock = Sock::Socket();
    Sock::Bind(listensock, 8080);
    Sock::Listen(listensock);

    for (int i = 0; i < NUM; ++i)
    {
        fdsArray[i].fd = DFL;
        fdsArray[i].events = 0;
        fdsArray[i].revents = 0;
    }
    fdsArray[0].fd = listensock;
    fdsArray[0].events = POLLIN; // 只關(guān)心讀事件
    int timeout = 1000;
    while (true)
    {
        int n = poll(fdsArray,NUM,timeout);

        switch (n)
        {
        case 0:
            cout << "time out ..." << (unsigned long long)time(nullptr) << endl;
            break;
        case -1:
            cerr << errno << " : " << strerror(errno) << endl;
            break;
        default:
            HandlerEvent(listensock);
            break;
        }
    }

    return 0;
}

Sock.hpp的代碼和select當(dāng)時(shí)的代碼相同!!!

那么現(xiàn)在的多路轉(zhuǎn)接還有什么問(wèn)題?

1. 鏈接多的時(shí)候,select和poll都是基于對(duì)多個(gè)fd進(jìn)行遍歷檢測(cè),來(lái)識(shí)別事件的,鏈接多的時(shí)候,一定會(huì)引起遍歷周期的增加.
2. 對(duì)于事件(用戶告訴內(nèi)核,內(nèi)核通知用戶),需要使用的數(shù)據(jù)結(jié)構(gòu)(數(shù)組),需要由程序員自己維護(hù).

三、IO多路轉(zhuǎn)接之epoll

epoll初始

按照man手冊(cè)的說(shuō)法:是為處理大批量句柄而作了改進(jìn)的poll.

他是在2.5.44內(nèi)核中引進(jìn)的(epoll(4) is a nwe API introduced in Linux kernel 2.5.44).

他幾乎具備了之前所說(shuō)的一切優(yōu)點(diǎn),被公認(rèn)為L(zhǎng)inux2.6下性能最好的多路IO就緒通知方法.

epoll的相關(guān)系統(tǒng)調(diào)用

epoll有三個(gè)相關(guān)的系統(tǒng)調(diào)用.
無(wú)論有多個(gè)接口,核心工作:只負(fù)責(zé)等!

epoll_create

創(chuàng)建一個(gè)epoll的句柄.

• 自從linux2.6.8以后,size參數(shù)是被忽略的,參數(shù)只要大于0即可.
• 用完之后,必須調(diào)用close關(guān)閉.

epoll_ctl

epoll的事件注冊(cè)函數(shù).

• 他不同于select()是在監(jiān)聽(tīng)事件時(shí)告訴內(nèi)核要監(jiān)聽(tīng)什么類型的事件,而是在這里先注冊(cè)要監(jiān)聽(tīng)的事件類型.
• 第一個(gè)參數(shù)是epoll_create()的返回值(epoll的句柄).
• 第二個(gè)參數(shù)表示動(dòng)作,用三個(gè)宏來(lái)表示.
• 第三個(gè)參數(shù)是需要監(jiān)聽(tīng)的fd.
• 第四個(gè)參數(shù)是告訴內(nèi)核需要監(jiān)聽(tīng)什么事.

第二個(gè)參數(shù)的取值:

• EPOLL_CTL_ADD:注冊(cè)新的fd到epfd中;
• EPOLL_CTL_MOD:修改已經(jīng)注冊(cè)的fd的監(jiān)聽(tīng)事件;
• EPOLL_CTL_DEL:從epfd中刪除一個(gè)fd;

struct epoll_event的結(jié)構(gòu)如下:

epoll_data_t的結(jié)構(gòu)如下:

events可以是以下幾個(gè)宏的集合:

• EPOLLIN : 表示對(duì)應(yīng)的文件描述符可以讀(包括對(duì)端SOCKET正常關(guān)閉);
• EPOLLOUT : 表示對(duì)應(yīng)的文件描述符可以寫;
• EPOLLPRI : 表示對(duì)應(yīng)的文件描述符有緊急的數(shù)據(jù)可讀(這里應(yīng)該表示有外帶數(shù)據(jù)的到來(lái));
• EPOLLERR : 表示對(duì)應(yīng)的文件描述符發(fā)生錯(cuò)誤;
• EPOLLLHUP : 表示對(duì)應(yīng)的文件描述符被掛斷;
• EPOLLET : 將EPOLL設(shè)為邊緣觸發(fā)(Edge Triggered)模式,這是相對(duì)于水平觸發(fā)(Level Triggered)來(lái)說(shuō)的.
• EPOLLONESHOT : 只監(jiān)聽(tīng)一次事件,當(dāng)監(jiān)聽(tīng)完這次事件事件之后,如果還需要監(jiān)聽(tīng)這個(gè)socket的話,需要再次把這個(gè)socket加入到EPOLL隊(duì)列中.

epoll_wait

收集在epoll監(jiān)控的事件中已經(jīng)發(fā)送的事件:

• 參數(shù)events是分配好的epoll_event結(jié)構(gòu)體數(shù)組.
• epoll將會(huì)把發(fā)生的事件賦值到events數(shù)組中(events不可以是空指針,內(nèi)核只負(fù)責(zé)把數(shù)據(jù)復(fù)制到這個(gè)events數(shù)組中,不會(huì)取幫助我們?cè)谟脩魬B(tài)中分配內(nèi)存).
• maxevents告知內(nèi)核這個(gè)even ts有多大,這個(gè)maxevents的值不能大于創(chuàng)建epoll_create()時(shí)的size.
• 參數(shù)timeout是超時(shí)時(shí)間(毫秒,0會(huì)立即返回,-1是永久阻塞).
• 如果函數(shù)調(diào)用成功,返回對(duì)應(yīng)IO上已準(zhǔn)備好的文件描述符數(shù)目,如返回0表示已超時(shí),返回小于0表示函數(shù)調(diào)用失敗.

epoll工作原理

操作系統(tǒng)如何得知,網(wǎng)絡(luò)中的數(shù)據(jù)到來(lái)了呢?

網(wǎng)卡先得到數(shù)據(jù)會(huì)向CPU發(fā)送硬件中斷,調(diào)用OS預(yù)設(shè)的中斷函數(shù),負(fù)責(zé)從外設(shè)進(jìn)行數(shù)據(jù)拷貝.

epoll函數(shù)針對(duì)特定的一個(gè)或者多個(gè)fd,設(shè)定對(duì)應(yīng)的回調(diào)機(jī)制;當(dāng)fd緩沖區(qū)有數(shù)據(jù)的時(shí)候,進(jìn)行回調(diào).

當(dāng)某一進(jìn)程調(diào)用epoll_create方法時(shí),Linux內(nèi)核會(huì)創(chuàng)建一個(gè)eventpoll結(jié)構(gòu)體,這個(gè)結(jié)構(gòu)體中有兩個(gè)成員與epoll的使用方式密切相關(guān).

....
/* 紅黑樹(shù)的根節(jié)點(diǎn),這棵樹(shù)中存儲(chǔ)著所有添加到epoll中的需要監(jiān)控的事件*/
struct rb_root rbr;
/* 雙鏈表中則存放著將要通過(guò)epoll_wait返回給用戶的滿足條件的事件*/
struct list_head rdlist;
....

• 每一個(gè)epoll對(duì)象都有一個(gè)獨(dú)立的eventpoll結(jié)構(gòu)體,用于存放通過(guò)epoll_ctl方法向epoll對(duì)象中添加進(jìn)來(lái)的事件.
• 這些事件都會(huì)掛載在紅黑樹(shù)中,如此重復(fù)添加的事件就可以通過(guò)紅黑樹(shù)而高效的識(shí)別出來(lái)(紅黑樹(shù)的插入時(shí)間效率是O(lgn),其中n為樹(shù)的高度).
• 而所有添加到epoll的事件都會(huì)與設(shè)備(網(wǎng)卡)驅(qū)動(dòng)程序建立回調(diào)關(guān)系,也就是說(shuō),當(dāng)響應(yīng)的事件發(fā)生時(shí)會(huì)調(diào)用這個(gè)回調(diào)方法.
• 這個(gè)回調(diào)方法在內(nèi)核中叫ep_poll_callback,他會(huì)將發(fā)生的時(shí)間添加到rdlist雙鏈表中.
• 在epoll中,對(duì)于每一個(gè)事件,都會(huì)建立一個(gè)epitem的結(jié)構(gòu)體.

struct epitem{ 
 struct rb_node rbn;//紅黑樹(shù)節(jié)點(diǎn) 
 struct list_head rdllink;//雙向鏈表節(jié)點(diǎn) 
 struct epoll_filefd ffd; //事件句柄信息 
 struct eventpoll *ep; //指向其所屬的eventpoll對(duì)象 
 struct epoll_event event; //期待發(fā)生的事件類型 
}

• 當(dāng)調(diào)用epoll_wait檢查是否有事件發(fā)生時(shí),只需要檢查eventpoll對(duì)象中的rdlist雙鏈表中是否有epitem元素即可.
• 如果rdlist不為空,則把發(fā)生的時(shí)間復(fù)制到用戶態(tài),同時(shí)將事件數(shù)量返回給用戶,這個(gè)操作的時(shí)間復(fù)雜度是O(1).

總結(jié)一下,epoll的使用過(guò)程就是三部曲:

• 調(diào)用epoll_create創(chuàng)建一個(gè)epoll句柄.
• 調(diào)用epoll__ctl,將監(jiān)控的文件描述符進(jìn)行注冊(cè).
• 調(diào)用epoll_wait,等待文件描述符就緒.

epoll的優(yōu)點(diǎn)(對(duì)比于select)

• 接口使用方便:雖然拆分成三個(gè)函數(shù),但是反而使用起來(lái)更方便和更高效.不需要每次循環(huán)都設(shè)置關(guān)注的文件描述符,也做到了輸入輸出參數(shù)分離開(kāi).
• 數(shù)據(jù)拷貝輕量:只在合適的時(shí)候調(diào)用EPOLL_CTL_ADD將文件描述符結(jié)構(gòu)拷貝到內(nèi)核中,這個(gè)操作并不頻繁(而select/poll都是每次循環(huán)都要進(jìn)行拷貝).
• 事件回調(diào)機(jī)制:避免使用遍歷,而是使用回調(diào)函數(shù)的方式,將就緒的文件描述符結(jié)構(gòu)加入到就緒隊(duì)列中,epoll_wait返回直接訪問(wèn)就緒隊(duì)列就知道哪些文件描述符就緒.這個(gè)操作時(shí)間復(fù)雜度O(1).即使文件描述符數(shù)目很多,效率也不會(huì)受到影響.
• 沒(méi)有數(shù)量限制:文件描述符數(shù)目無(wú)上限.

注意!!!
網(wǎng)上有些博客說(shuō),epoll中使用了內(nèi)存映射機(jī)制

• 內(nèi)存映射機(jī)制:內(nèi)核直接將就緒隊(duì)列通過(guò)mmap的方式映射到用戶態(tài),避免了拷貝內(nèi)存這樣的額外性能開(kāi)銷.

這種說(shuō)法是不準(zhǔn)確的.我們定義的struct epoll_event是我們?cè)谟脩艨臻g中分配好的內(nèi)存.勢(shì)必還是需要將內(nèi)核的數(shù)據(jù)拷貝到這個(gè)用戶空間的內(nèi)存中的.

epoll的工作方式

epoll有兩種工作方式-水平觸發(fā)(LT)和邊緣觸發(fā)(ET).

加入有這樣一個(gè)例子

• 我們已經(jīng)把一個(gè)tcp socket添加到epoll描述符.
• 這個(gè)時(shí)候socket的另一端被寫入了2KB的數(shù)據(jù).
• 調(diào)用epoll_wait,并且他會(huì)返回,說(shuō)明他已經(jīng)準(zhǔn)備好讀取操作.
• 然后調(diào)用read,制度去了1KB數(shù)據(jù).
• 繼續(xù)調(diào)用epoll_wait.

水平觸發(fā)Level Triggered 工作模式

epoll默認(rèn)狀態(tài)下就是LT工作模式.

• 當(dāng)epoll檢測(cè)到socket上事件就緒的時(shí)候,可以不立刻進(jìn)行處理,或者只處理一部分.
• 如上面的例子,由于只讀了1K數(shù)據(jù),緩沖區(qū)中還剩1K數(shù)據(jù),在第二次調(diào)用epoll_wait時(shí),epoll_wait仍會(huì)立刻返回并通知socket讀事件就緒.
• 直到緩沖區(qū)上的所有數(shù)據(jù)都被處理完,epoll_wait才不會(huì)立刻返回.
• 支持阻塞讀寫和非阻塞讀寫.

邊緣觸發(fā)Edge Triggered工作模式

如果我們?cè)诘谝徊綄ocket添加到epoll描述符的時(shí)候使用了EPOLLET表示,epoll進(jìn)入ET工作模式.

• 當(dāng)epoll檢測(cè)到socket上事件就緒時(shí),必須立刻處理.
• 如上面的例子,雖然只讀了1K的數(shù)據(jù),緩沖區(qū)還剩1K的數(shù)據(jù),在第二次調(diào)用epoll_wait的時(shí)候,epoll_wait不會(huì)再返回了.
• 也就是說(shuō)ET模式下,文件描述符上的事件就緒后,只有一次處理機(jī)會(huì).
• ET的性能比LT性能更高(epoll_Wait返回的次數(shù)少了很多).Nginx默認(rèn)采用ET模式使用epoll.
• 只支持非阻塞的讀寫

select和poll其實(shí)也是工作在LT模式下,epoll既可以支持LT,也可以支持ET.

對(duì)比LT和ET

LT是epoll的默認(rèn)行為.使用ET能夠減少epoll觸發(fā)的次數(shù).但是代價(jià)就是強(qiáng)逼著程序猿一次響應(yīng)就緒過(guò)程中就把所有的數(shù)據(jù)都處理完.

相當(dāng)于一個(gè)文件描述符就緒以后,不會(huì)反復(fù)被提示就緒,看起來(lái)就比LT更高效一些.但是在LT情況下如果也能做到每次就緒的文件描述符都立即處理,不讓這個(gè)就緒被重復(fù)提示的話,其實(shí)性能也是一樣的.

另一方面,ET的代碼復(fù)雜程度更高了!

理解ET模式和非阻塞文件描述符

使用ET模式的epoll,需要將文件描述設(shè)置為非阻塞.這個(gè)不是接口上的要求,而是"工程實(shí)踐"上的要求.

假設(shè)這樣的場(chǎng)景:服務(wù)器接收到一個(gè)10K的請(qǐng)求,會(huì)向客戶端返回一個(gè)應(yīng)答數(shù)據(jù).如果客戶端收不到應(yīng)答,不會(huì)發(fā)送第二個(gè)10K請(qǐng)求.

如果服務(wù)器寫的代碼是阻塞式的read,并且一次只read1K的數(shù)據(jù)的話(read不能保證一次就把所有的數(shù)據(jù)都讀出來(lái),參考man手冊(cè)的說(shuō)明,可能信號(hào)被打斷),剩下的9K數(shù)據(jù)就會(huì)待在緩沖區(qū)中.

此時(shí)由于epoll是ET模式,并不會(huì)認(rèn)為文件描述符就緒.epoll_wait就不會(huì)再次返回.剩下的9K數(shù)據(jù)會(huì)一直在緩沖區(qū)中.直到下一次客戶端再給服務(wù)器寫數(shù)據(jù).epoll_wait才能返回.

但是問(wèn)題來(lái)了.

• 服務(wù)器只讀到1K數(shù)據(jù),要10K讀完才會(huì)給客戶端返回響應(yīng)數(shù)據(jù).
• 客戶端要讀到服務(wù)器的響應(yīng),才會(huì)發(fā)送下一個(gè)請(qǐng)求.
• 客戶端發(fā)送下一個(gè)請(qǐng)求,epoll_wait才會(huì)返回,才能去讀緩沖區(qū)中剩余的數(shù)據(jù).

所以為了解決上述問(wèn)題(阻塞read不一定能一下把完整的請(qǐng)求讀完),于是就可以使用非阻塞輪詢的方式來(lái)讀緩沖區(qū),保證一定能把完整的請(qǐng)求都讀出來(lái).

如果是LT沒(méi)這個(gè)問(wèn)題.只要緩沖區(qū)中的數(shù)據(jù)沒(méi)讀完,就能夠讓epoll_wait返回文件描述符讀就緒.

epoll的使用場(chǎng)景

epoll的高性能,是有一定的特定場(chǎng)景的.如果場(chǎng)景選擇不適宜的話,epoll的性能可能適得其反.

• 對(duì)于多連接,且多個(gè)連接中只有一部分連接比較活躍時(shí),比較適合epoll.

例如,典型的一個(gè)需要處理上萬(wàn)個(gè)客戶端的服務(wù)器,例如各種互聯(lián)網(wǎng)APP的入口服務(wù)器,這樣的服務(wù)器就很適合epoll.

如果只是系統(tǒng)內(nèi)部,服務(wù)器和服務(wù)器之間進(jìn)行通信,只有少數(shù)的幾個(gè)連接,這種情況下用epoll就并不合適.具體要根據(jù)需求和場(chǎng)景特點(diǎn)來(lái)決定使用哪種IO模型.

epoll示例: epoll服務(wù)器(LT模式)

makefile

main:main.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm main

log.hpp

/*
 * @Author: hulu 2367598978@qq.com
 * @Date: 2022-11-28 16:18:12
 * @LastEditors: hulu 2367598978@qq.com
 * @LastEditTime: 2022-12-05 11:47:11
 * @FilePath: /udp/log.hpp
 * @Description: 這是默認(rèn)設(shè)置,請(qǐng)?jiān)O(shè)置`customMade`, 打開(kāi)koroFileHeader查看配置 進(jìn)行設(shè)置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#pragma once
#include<cstdio>
#include<ctime>
#include<cstdarg>
#include<cassert>
#include<cstdlib>
#include<cstring>
#include<cerrno>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#define DEBUG     0
#define NOTICE    1
#define WARINING  2
#define FATAL     3

const char* log_level[]={"DEBUG","NOTICE","WARINING","FATAL"};

#define LOGFIFE "ServerTcp.log"

class Log
{
public:
    Log():_logFd((-1))
    {}
    ~Log()
    {
        if(_logFd!=-1)
        {
            fsync(_logFd);//將操作系統(tǒng)中的數(shù)據(jù)盡快刷盤
            close(_logFd);
        }

    }
    void enable()
    {
        _logFd=open(LOGFIFE,O_WRONLY|O_APPEND|O_CREAT,0666);
        assert(_logFd!=-1);
        dup2(_logFd,0);
        dup2(_logFd,1);
        dup2(_logFd,2);
    }
private:
    int _logFd;
};

//logMessage(DEBUG,"%d",10);
void logMessage(int level,const char* format,...)
{
    assert(level>=DEBUG);
    assert(level<=FATAL);
    char logInfo[1024];
    char* name=getenv("USER");
    va_list ap; //ap--->char*
    va_start(ap,format);

    vsnprintf(logInfo,sizeof(logInfo)-1,format,ap);

    va_end(ap); //ap=NULL

    FILE* out=(level==FATAL)?stderr:stdout;
    fprintf(out,"%s | %u | %s | %s\n",\
    log_level[level],(unsigned int)time(nullptr),\
    name==nullptr?"unknow":name,logInfo);
    fflush(out);//將C緩沖區(qū)中的數(shù)據(jù)刷新到OS
    fsync(fileno(out)); // 將OS中的數(shù)據(jù)盡快刷盤
}

Sock.hpp

#include <iostream>
#include <string>
#include <cassert>
#include <unistd.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <fstream>
using namespace std;

class Sock
{
public:
    static const int gBackLog = 20;
    static int Socket()
    {
        // 1.創(chuàng)建socket
        int _listenSock = socket(AF_INET, SOCK_STREAM, 0);
        if (_listenSock < 0)
        {
            exit(1);
        }
        int opt = 1;
        setsockopt(_listenSock,SOL_SOCKET,SO_REUSEADDR | SO_REUSEPORT, &opt,sizeof opt);
        return _listenSock;
    }
    static void Bind(int socket, uint16_t _port)
    {
        struct sockaddr_in local; // 用戶棧
        memset(&local, 0, sizeof local);
        local.sin_family = AF_INET;
        local.sin_port = htons(_port);
        local.sin_addr.s_addr = INADDR_ANY;
        // 2.2本地socket信息，寫入_sock對(duì)應(yīng)的內(nèi)核區(qū)域
        if (bind(socket, (const sockaddr *)&local, sizeof local) < 0)
        {
            exit(2);
        }
    }
    static void Listen(int socket)
    {
        // 3.監(jiān)聽(tīng)socket，為何要監(jiān)聽(tīng)呢？tcp是面向連接的！
        if (listen(socket, gBackLog) < 0)
        {
            exit(3);
        }
        // 允許別人來(lái)連接你了
    }
    static int Accept(int socket,string* clientip,uint16_t* clientport)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int serviceSock = accept(socket, (struct sockaddr *)&peer, &len);
        if (serviceSock < 0)
        {
            return -1;
        }
        if(clientport) *clientport = ntohs(peer.sin_port);
        if(clientip) *clientip = inet_ntoa(peer.sin_addr);
        return serviceSock;
    }
};

EpollServer.hpp

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include <string>
#include<functional>
#include "log.hpp"
#include "Sock.hpp"
using namespace std;

#define NUM 1024

class EpollServer
{
public:
    using func_t = function<int(int)>;
    EpollServer(uint16_t port,func_t func)
        : _port(port),_func(func)
    {
    }
    ~EpollServer()
    {
        if (_listenSock != -1)
            close(_listenSock);
        if (_epFd != -1)
            close(_epFd);
    }
    void InitEpollServer()
    {
        _listenSock = Sock::Socket();

        Sock::Bind(_listenSock, _port);

        Sock::Listen(_listenSock);

        // 這里直接使用原生接口
        _epFd = epoll_create(NUM);
        if (_epFd < 0)
        {
            logMessage(FATAL, "%d:%s", errno, strerror(errno));
            exit(4);
        }

        logMessage(DEBUG, "創(chuàng)建監(jiān)聽(tīng)套接字成功:%d", _listenSock);
        logMessage(DEBUG, "創(chuàng)建epoll成功:%d", _epFd);
    }
    void Run()
    {
        // 1.先添加_listenSock
        struct epoll_event ev;
        ev.events = EPOLLIN;
        ev.data.fd = _listenSock;
        int n = epoll_ctl(_epFd, EPOLL_CTL_ADD, _listenSock, &ev);
        assert(n == 0);
        (void)n;

        struct epoll_event revs[NUM];
        int timeout = 1000;
        while (true)
        {
            // 關(guān)于n: 就緒的fd的個(gè)數(shù),只需要進(jìn)行將底層的就緒隊(duì)列中節(jié)點(diǎn),一次從0下標(biāo)放入到revs中即可
            int n = epoll_wait(_epFd, revs, NUM, timeout);
            switch (n)
            {
            case 0:
                cout << "time out ..." << (unsigned long long)time(nullptr) << endl;
                break;
            case -1:
                cerr << errno << " : " << strerror(errno) << endl;
                break;
            default:
                HandlerEvents(revs, n);
                break;
            }
        }
    }
    void HandlerEvents(struct epoll_event revs[], int n)
    {
        for (int i = 0; i < n; ++i)
        {
            int sock = revs[i].data.fd;
            uint32_t event = revs[i].events;
            if (event & EPOLLIN) // 讀就緒就緒
            {
                if (sock == _listenSock) // 監(jiān)聽(tīng)socket就緒,獲取新鏈接
                {
                    string ip;
                    uint16_t port;
                    int sockfd = Sock::Accept(_listenSock, &ip, &port);
                    if (sockfd < 0)
                    {
                        logMessage(WARINING, "%d:%s", errno, strerror(errno));
                        continue;
                    }
                    // 托管給Epoll
                    struct epoll_event ev;
                    ev.events = EPOLLIN;
                    ev.data.fd = sockfd;
                    int n = epoll_ctl(_epFd, EPOLL_CTL_ADD, sockfd, &ev);
                    assert(n == 0);
                    (void)n;
                }
                else // 普通socket就緒,進(jìn)行數(shù)據(jù)INPUT
                {
                    int n = _func(sock);
                    if(n == 0 || n<0)
                    {
                        // 文件描述符先移除在關(guān)閉
                        int x =epoll_ctl(_epFd,EPOLL_CTL_DEL,sock,nullptr);
                        assert(x == 0);
                        (void)x;
                        logMessage(DEBUG,"client quit: %d",sock);
                        close(sock);
                    }


                }
            }
            else
            {
            }
        }
    }

private:
    int _listenSock = -1;
    int _epFd = -1;
    uint16_t _port;
    func_t _func;
};

main.cc

#include "EpollServer.hpp"
#include <memory>

static void Usage(string process)
{
    cerr << "\nUsage: " << process << "\tport\n"
         << endl;
}

int myfunc(int sock)
{
    char buffer[NUM];
    ssize_t s = recv(sock, buffer,sizeof(buffer)-1,0); // 不會(huì)被阻塞
    if(s > 0)
    {
        buffer[s] = 0;
        logMessage(DEBUG,"client[%d] #:%s",sock,buffer);
    }
    return s;
}

// ./SelectServer 8080
// 只關(guān)心讀事件
int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        exit(1);
    }
    unique_ptr<EpollServer> epollServer(new EpollServer(atoi(argv[1]), myfunc));
    epollServer->InitEpollServer();
    epollServer->Run();
}

總結(jié)

對(duì)于IO多路復(fù)用的三個(gè)函數(shù)就介紹到這里了,下一篇博客我們基于ET模式下的epoll服務(wù)器,也加Reactor模式.
(本章完!)文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-406973.html

到了這里，關(guān)于IO多路復(fù)用之select/poll/epoll的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！