【计算机网络】IO多路转接之poll

一、poll函数接口

#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
// pollfd结构
struct pollfd {int fd; /* file descriptor */short events; /* requested events */short revents; /* returned events */
};

参数说明

fds是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返回的事件集合.

nfds表示fds数组的长度.

timeout表示poll函数的超时时间, 单位是毫秒(ms)

时间单位:ms

1.>0：在timeout以内，阻塞，否则非阻塞返回一次

2.=0：非阻塞等待

3.<0：阻塞等待

events和revents的取值:

用户 -> 内核 ：要帮我关心一下fd。输入看 ：fd + events

内核告诉用户：你要关心fd上面的events中有哪些事件已经就绪了。输出时看：fd + revents

这样达到了输入和输出分离，poll不需要对参数进行重新设定，以及解决了select等待fd有上限的问题

返回结果

返回值小于0, 表示出错;

返回值等于0, 表示poll函数等待超时;

返回值大于0, 表示poll由于监听的文件描述符就绪而返回.

二、socket就绪条件

读就绪

socket内核中, 接收缓冲区中的字节数, 大于等于低水位标记SO_RCVLOWAT. 此时可以无阻塞的读该文件描述符, 并且返回值大于0;

socket TCP通信中, 对端关闭连接, 此时对该socket读, 则返回0;

监听的socket上有新的连接请求;

socket上有未处理的错误;

写就绪

socket内核中, 发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小), 大于等于低水位标记

SO_SNDLOWAT, 此时可以无阻塞的写, 并且返回值大于0;

socket的写操作被关闭(close或者shutdown). 对一个写操作被关闭的socket进行写操作, 会触发SIGPIPE信号;

socket使用非阻塞connect连接成功或失败之后;

socket上有未读取的错误;

异常就绪

socket上收到带外数据. 关于带外数据, 和TCP紧急模式相关(TCP协议头中, 有一个紧急指针的字段),

三、poll的优点

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

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

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

四、poll的缺点

poll中监听的文件描述符数目增多时,和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符.

每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.

同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降.

五、poll使用案例–只读取数据的server服务器

1.err.hpp

#pragma onceenum
{USAGE_ERR = 1,SOCKET_ERR,BIND_ERR,LISTEN_ERR
};

2.log.hpp

#pragma once#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <stdarg.h>#define NORMAL 0
#define DEBUG 1
#define WARNING 2
#define ERROR 3
#define FATAL 4#define LOG_NORMAL "./log.txt"
#define LOG_ERR "./err.txt"#define NUM 1024const 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, ...)
{// [日志等级] [时间戳/时间] [pid] [messge]char logprofix[NUM];snprintf(logprofix, sizeof logprofix, "[%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 << logprofix << logcontent << std::endl;FILE *log = fopen(LOG_NORMAL, "a");FILE *error = fopen(LOG_ERR, "a");if (log && error){FILE *cur = nullptr;if (level == DEBUG || level == NORMAL || level == WARNING)cur = log;if (level == ERROR || level == FATAL)cur = error;if (cur)fprintf(cur, "%s%s\n", logprofix, logcontent);fclose(log);fclose(error);}
}

3.sock.hpp

#pragma once#include <iostream>
#include <cstring>
#include <string>
#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
{static const int backlog = 32;public:// 1. 创建socket文件套接字对象static int 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;}// 2.bind自己的网络信息static void Bind(int sock, const 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;int n = bind(sock, (struct sockaddr *)&local, sizeof local);if (n < 0){LogMessage(FATAL, "socket bind error");exit(BIND_ERR);}LogMessage(NORMAL, "socket bind success");}// 3. 设置socket 为监听状态static void Listen(int sock){int n = listen(sock, backlog);if (n < 0){LogMessage(FATAL, "socket listen error");exit(LISTEN_ERR);}LogMessage(NORMAL, "socket listen success");}static int Accept(int listensock, std::string *clientip, uint16_t *clientport){struct sockaddr_in peer;memset(&peer, 0, sizeof peer);socklen_t len = sizeof(peer);int sock = accept(listensock, (struct sockaddr *)&peer, &len);if (sock < 0){LogMessage(ERROR, "socket 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;}
};

4.pollServer.hpp

#pragma once#include <iostream>
#include <cstring>
#include <cstdlib>
#include <cerrno>
#include <string>
#include <functional>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <poll.h>#include "sock.hpp"
#include "log.hpp"
#include "err.h"namespace poll_ns
{const static int defaultport = 8080;const static int num = 2048;const static int defaultfd = -1;typedef std::function<std::string(const std::string &)> func_t;class pollServer{public:pollServer(const func_t &func, const uint16_t &port = defaultport): _port(port), _func(func), _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;_rfds[0].revents = 0;}void ResetItem(int pos){_rfds[pos].fd = defaultfd;_rfds[pos].events = 0;_rfds[pos].revents = 0;}void Print(){std::cout << "fd list: ";for (int i = 0; i < num; i++){if (_rfds[i].fd != defaultfd){std::cout << _rfds[i].fd << " ";}}}void Accepter(int listensock){std::string clientip;uint16_t clientport;int sock = Sock::Accept(listensock, &clientip, &clientport);if (sock < 0)return;LogMessage(NORMAL, "accept success[%s:%d]", clientip.c_str(), clientport);int i = 0;for (; i < num; i++){if (_rfds[i].fd == defaultfd)continue;elsebreak;}if (i == num){LogMessage(WARNING, "server is full,please wait");close(sock);}else{_rfds[i].fd = sock;_rfds[i].events = POLLIN;_rfds[i].revents = 0;}Print();}void Recver(int pos){// 1. 读取requestchar buffer[1024];ssize_t s = recv(_rfds[pos].fd, buffer, sizeof(buffer) - 1, 0); // 这里在进行读取的时候，会不会被阻塞？1， 0if (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 error:%s", strerror(errno));return;}// 2. 处理requeststd::string response = _func(buffer);// 3. 返回responsewrite(_rfds[pos].fd, response.c_str(), response.size());}void HandlerEvent(){for (int i = 0; i < num; i++){// 过滤掉非法的fdif (_rfds[i].fd == defaultfd)continue;if (!(_rfds[i].events & POLLIN))continue;if (_rfds[i].fd == _listensock && (_rfds[i].events & POLLIN))Accepter(_listensock);else if (_rfds[i].events & POLLIN)Recver(i);else{}}}void start(){for (;;){int timeout = -1;int n = poll(_rfds, num, timeout);switch (n){case 0:LogMessage(NORMAL, "timeout...");break;case -1:LogMessage(WARNING, "select error,code: %d, err string: %s", errno, strerror(errno));break;default:// 说明有事件就绪了,目前只有一个监听事件就绪了LogMessage(NORMAL, "have event ready!");HandlerEvent();break;}}}~pollServer(){if (_listensock > 0)close(_listensock);if (_rfds)delete[] _rfds;}private:int _port;int _listensock;struct pollfd *_rfds;func_t _func;};
}

5.main.cc

#include "pollServer.hpp"
#include "err.hpp"
#include <memory>using namespace std;
using namespace poll_ns;static void Usage(const string proc)
{std::cerr << "Usage:\n\t" << proc << " port\n\n";
}string transaction(const string &request)
{return "pollServer# " + request;
}int main(int argc, char *argv[])
{// if (argc != 2)// {//     Usage(argv[0]);//     exit(USAGE_ERR);// }std::unique_ptr<pollServer> svr(new pollServer(transaction));svr->initServer();svr->start();return 0;
}