I/O多路转接之poll

承接上文：I/O多路转接之select-CSDN博客

简介

poll原型介绍

select代码改造成poll型

poll优缺点

在前文中我们得知了，select的诸多缺点，接下来这个poll就可以解决上述出现的问题

poll也是一种Linux中的多路转接的方案，主要解决下面这两个问题

1. select的fd有上限的问题

2. 每次调用都要重新设置关心的fd

poll原型介绍

events和revents的取值

纯大写，全是系统内定义出来的宏

其中比较重要并且常用的就那么几个，比如POLLIN(读)，POLLOUT(写)，POLLERR(错误),还有一个POLLPRI(就是TCP中的那个带外数据)

代码改造

poll在实际上其实在代码书写上是比select要简单的，只要稍加改造就可以了

err.hpp

#pragma once

#include <iostream>

enum
{
    USAGE_ERR = 1,
    SOCKET_ERR,
    BIND_ERR,
    LISTEN_ERR
};

log.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>

#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4

const char * to_levelstr(int level)
{
    switch(level)
    {
        case DEBUG : return "DEBUG";
        case NORMAL: return "NORMAL";
        case WARNING: return "WARNING";
        case ERROR: return "ERROR";
        case FATAL: return "FATAL";
        default : return nullptr;
    }
}

void logMessage(int level, const char *format, ...)
{
#define NUM 1024
    char logprefix[NUM];
    snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",
        to_levelstr(level), (long int)time(nullptr), getpid());

    char logcontent[NUM];
    va_list arg;
    va_start(arg, format);
    vsnprintf(logcontent, sizeof(logcontent), format, arg);

    std::cout << logprefix << logcontent << std::endl;
}

main.cc

#include "pollServer.hpp"
#include "err.hpp"
#include <memory>

using namespace std;
using namespace poll_ns;

static void usage(std::string proc)
{
    std::cerr << "Usage:\n\t" << proc << " port" << "\n\n";
}

std::string transaction(const std::string &request)
{
    return request;
}

// ./select_server 8081
int main(int argc, char *argv[])
{
    // if(argc != 2)
    // {
    //     usage(argv[0]);
    //     exit(USAGE_ERR);
    // }

    // unique_ptr<SelectServer> svr(new SelectServer(atoi(argv[1])));

    // std::cout << "test: " << sizeof(fd_set) * 8 << std::endl;
    unique_ptr<PollServer> svr(new PollServer(transaction));

    svr->initServer();

    svr->start();

    return 0;
}

makefile

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

pollServer.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <poll.h>
#include "sock.hpp"

namespace poll_ns
{
    static const int defaultport = 8081;
    static const int num = 2048;
    static const int defaultfd = -1;

    using func_t = std::function<std::string (const std::string&)>;

    class PollServer
    {
    public:
        PollServer(func_t f, int port = defaultport) : _func(f), _port(port), _listensock(-1), _rfds(nullptr)
        {
        }
        void initServer()
        {
            _listensock = Sock::Socket();
            Sock::Bind(_listensock, _port);
            Sock::Listen(_listensock);
            _rfds = new struct pollfd[num];
            for (int i = 0; i < num; i++) ResetItem(i);
            _rfds[0].fd = _listensock; // 不变了
            _rfds[0].events = POLLIN;
        }
        void Print()
        {
            std::cout << "fd list: ";
            for (int i = 0; i < num; i++)
            {
                if (_rfds[i].fd != defaultfd)
                    std::cout << _rfds[i].fd << " ";
            }
            std::cout << std::endl;
        }
        void ResetItem(int i)
        {
            _rfds[i].fd = defaultfd;
            _rfds[i].events = 0;
            _rfds[i].revents = 0;
        }
        void Accepter(int listensock)
        {
            logMessage(DEBUG, "Accepter in");
            // 走到这里，accept 函数，会不会阻塞？？？1 0
            // select 告诉我， listensock读事件就绪了
            std::string clientip;
            uint16_t clientport = 0;
            int sock = Sock::Accept(listensock, &clientip, &clientport); // accept = 等 + 获取
            if (sock < 0)
                return;
            logMessage(NORMAL, "accept success [%s:%d]", clientip.c_str(), clientport);
            // sock我们能直接recv/read 吗？不能，整个代码，只有select有资格检测事件是否就绪
            // 将新的sock 托管给select！
            // 将新的sock托管给select的本质，其实就是将sock，添加到fdarray数组中即可！
            int i = 0;
            for (; i < num; i++)
            {
                if (_rfds[i].fd != defaultfd)
                    continue;
                else
                    break;
            }
            if (i == num)
            {
                logMessage(WARNING, "server if full, please wait");
                close(sock);
            }
            else
            {
                _rfds[i].fd = sock;
                _rfds[i].events = POLLIN;
                _rfds[i].revents = 0;
            }
            Print();
            logMessage(DEBUG, "Accepter out");
        }
        void Recver(int pos)
        {
            logMessage(DEBUG, "in Recver");

            // 1. 读取request
            // 这样读取是有问题的！
            char buffer[1024];
            ssize_t s = recv(_rfds[pos].fd, buffer, sizeof(buffer) - 1, 0); // 这里在进行读取的时候，会不会被阻塞？1， 0
            if (s > 0)
            {
                buffer[s] = 0;
                logMessage(NORMAL, "client# %s", buffer);
            }
            else if (s == 0)
            {
                close(_rfds[pos].fd);
                ResetItem(pos);
                logMessage(NORMAL, "client quit");
                return;
            }
            else
            {
                close(_rfds[pos].fd);
                ResetItem(pos);
                logMessage(ERROR, "client quit: %s", strerror(errno));
                return;
            }

            // 2. 处理request
            std::string response = _func(buffer);

            // 3. 返回response
            // write bug
            write(_rfds[pos].fd, response.c_str(), response.size());

            logMessage(DEBUG, "out Recver");
        }
        // 1. handler event rfds 中，不仅仅是有一个fd是就绪的，可能存在多个
        // 2. 我们的select目前只处理了read事件
        void HandlerReadEvent()
        {
            for (int i = 0; i < num; i++)
            {
                // 过滤掉非法的fd
                if (_rfds[i].fd == defaultfd)
                    continue;
                if (!(_rfds[i].events & POLLIN)) continue;
                // 正常的fd
                // 正常的fd不一定就绪了
                // 目前一定是listensock，只有这一个
                if (_rfds[i].fd== _listensock && (_rfds[i].revents & POLLIN))
                    Accepter(_listensock);
                else if(_rfds[i].revents & POLLIN)
                    Recver(i);
                else{}
            }
        }

        void start()
        {
            int timeout = 1000;
            for (;;)
            {
                int n = poll(_rfds, num, timeout);
                switch (n)
                {
                case 0:
                    logMessage(NORMAL, "timeout...");
                    break;
                case -1:
                    logMessage(WARNING, "poll error, code: %d, err string: %s", errno, strerror(errno));
                    break;
                default:
                    // 说明有事件就绪了,目前只有一个监听事件就绪了
                    logMessage(NORMAL, "have event ready!");
                    HandlerReadEvent();
                    // HandlerWriteEvent(wfds);
                    break;
                }
            }
        }
        ~PollServer()
        {
            if (_listensock < 0)
                close(_listensock);
            if (_rfds)
                delete[] _rfds;
        }

    private:
        int _port;
        int _listensock;
        struct pollfd *_rfds;
        func_t _func;
    };
}

sock.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "log.hpp"
#include "err.hpp"

class Sock
{
    const static int backlog = 32;

public:
    static int Socket()
    {
        // 1. 创建socket文件套接字对象
        int sock = socket(AF_INET, SOCK_STREAM, 0);
        if (sock < 0)
        {
            logMessage(FATAL, "create socket error");
            exit(SOCKET_ERR);
        }
        logMessage(NORMAL, "create socket success: %d", sock);

        int opt = 1;
        setsockopt(sock, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT, &opt, sizeof(opt));
        return sock;
    }

    static void Bind(int sock, int port)
    {
        // 2. bind绑定自己的网络信息
        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;
        if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0)
        {
            logMessage(FATAL, "bind socket error");
            exit(BIND_ERR);
        }
        logMessage(NORMAL, "bind socket success");
    }

    static void Listen(int sock)
    {
        // 3. 设置socket 为监听状态
        if (listen(sock, backlog) < 0) // 第二个参数backlog后面在填这个坑
        {
            logMessage(FATAL, "listen socket error");
            exit(LISTEN_ERR);
        }
        logMessage(NORMAL, "listen socket success");
    }

    static int Accept(int listensock, std::string *clientip, uint16_t *clientport)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int sock = accept(listensock, (struct sockaddr *)&peer, &len);
        if (sock < 0)
            logMessage(ERROR, "accept error, next");
        else
        {
            logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
            *clientip = inet_ntoa(peer.sin_addr);
            *clientport = ntohs(peer.sin_port);
        }

        return sock;
    }
};

测试结果

Poll的特点

优点

不同与select使用三个位图来表示三个fdset的方式， poll使用一个pollfd的指针实现

pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式. 接口使用比select更方便；

poll并没有最大数量限制 (但是数量过大后性能也是会下降)；

缺点

poll中监听的文件描述符数目增多时，因为它是以线性遍历的方式进行查看，所以我们就有了下一个接口epoll，这个接口才是集百家之长处

        和select函数一样， poll返回后，需要轮询pollfd来获取就绪的描述符；
        每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中；
        同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降；