当前位置：首页 > news >正文

Linux——多路复用之poll

news 来源：原创 2024/9/20 14:29:05

前言

一、poll的认识

二、poll的接口

三、poll的使用

前言

前面我们学习了多路复用的select，知道多路复用的原理与select的使用方法，但是select也有许多缺点，导致他的效率不算高。今天我们来学习poll的使用，看看poll较于select的优势。

一、poll的认识

poll与select一样，只负责IO的等的过程，只不过一次可以等待多个文件描述符，他的作用是让read和write不再阻塞。

是用来监视多个文件描述符的状态变化的
程序会停在poll这里等待，直到被监视的文件描述符有一个或多个发生了状态改变

二、poll的接口

poll的接口如下，比select要轻量化很多，只有三个参数

参数1：struct pollfd *fds，pollfd数组首元素地址，

pollfd是操作系统给我们提供的结构体，主要成员如下

fd：文件描述符

events：用户告诉内核，需要关心的fd，上面的事件

revents：poll返回，内核告诉用户，关心的fd，那些事件就绪

参数2：nfds_t nfds，数组元素个数

参数3：int timeout，毫秒级的等待时间

timeout > 0 等待timeout毫秒或者有fd就绪再返回。

timeout == 0 非阻塞轮询。

timeout == -1 阻塞等待，直到有fd就绪。

返回值：

ret > 0 ：poll等待的多个fd中，已经就需要的fd个数
ret == 0 ：poll超时返回
ret < 0 ：poll出错

poll的事件如下，这些值是bit位，可以通过 |（或运算）的方式写入到events中，我们着重学习POLLIN和POLLOUT，

我们来思考一下这样设计的好处

poll的调用将输入和输出分离，这样就不用一直设置参数。
只要系统资源足够，就能一直创建pollfd，解决了等待fd的上限问题。
不用再自己组织结构，将fd放入其中，现在维护好这个pollfd的结构体数组即可。
参数变少了，通过或运算就可以添加自己关心的事件
时间参数timeout使用也很简单。

三、poll的使用

Log.hpp

#pragma once#include <iostream>
#include <cstdarg>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <pthread.h>
using namespace std;enum
{Debug = 0,Info,Warning,Error,Fatal
};enum
{Screen = 10,OneFile,ClassFile
};string LevelToString(int level)
{switch (level){case Debug:return "Debug";case Info:return "Info";case Warning:return "Warning";case Error:return "Error";case Fatal:return "Fatal";default:return "Unknown";}
}const int default_style = Screen;
const string default_filename = "Log.";
const string logdir = "log";class Log
{
public:Log(int style = default_style, string filename = default_filename): _style(style), _filename(filename){if (_style != Screen)mkdir(logdir.c_str(), 0775);}// 更改打印方式void Enable(int style){_style = style;if (_style != Screen)mkdir(logdir.c_str(), 0775);}// 时间戳转化为年月日时分秒string GetTime(){time_t currtime = time(nullptr);struct tm *curr = localtime(&currtime);char time_buffer[128];snprintf(time_buffer, sizeof(time_buffer), "%d-%d-%d %d:%d:%d",curr->tm_year + 1900, curr->tm_mon + 1, curr->tm_mday, curr->tm_hour, curr->tm_min, curr->tm_sec);return time_buffer;}// 写入到文件中void WriteLogToOneFile(const string &logname, const string &message){FILE *fp = fopen(logname.c_str(), "a");if (fp == nullptr){perror("fopen failed");exit(-1);}fprintf(fp, "%s\n", message.c_str());fclose(fp);}// 打印日志void WriteLogToClassFile(const string &levelstr, const string &message){string logname = logdir;logname += "/";logname += _filename;logname += levelstr;WriteLogToOneFile(logname, message);}pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;void WriteLog(const string &levelstr, const string &message){pthread_mutex_lock(&lock);switch (_style){case Screen:cout << message << endl; // 打印到屏幕中break;case OneFile:WriteLogToClassFile("all", message); // 给定all，直接写到all里break;case ClassFile:WriteLogToClassFile(levelstr, message); // 写入levelstr里break;default:break;}pthread_mutex_unlock(&lock);}// 提供接口给运算符重载使用void _LogMessage(int level, const char *file, int line, char *rightbuffer){char leftbuffer[1024];string levelstr = LevelToString(level);string currtime = GetTime();string  idstr = to_string(getpid());snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%s][%s][%s:%d]", levelstr.c_str(), currtime.c_str(), idstr.c_str(), file, line);string messages = leftbuffer;messages += rightbuffer;WriteLog(levelstr, messages);}// 运算符重载void operator()(int level, const char *file, int line, const char *format, ...){char rightbuffer[1024];va_list args;                                              // va_list 是指针va_start(args, format);                                    // 初始化va_list对象，format是最后一个确定的参数vsnprintf(rightbuffer, sizeof(rightbuffer), format, args); // 写入到rightbuffer中va_end(args);_LogMessage(level, file, line, rightbuffer);}~Log(){}private:int _style;string _filename;
};Log lg;class Conf
{
public:Conf(){lg.Enable(Screen);}~Conf(){}
};Conf conf;// 辅助宏
#define lg(level, format, ...) lg(level, __FILE__, __LINE__, format, ##__VA_ARGS__)

Socket.hpp

#pragma once#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cstring>
#include <unistd.h>
using namespace std;
namespace Net_Work
{static const int default_backlog = 5;static const int default_sockfd = -1;using namespace std;enum{SocketError = 1,BindError,ListenError,ConnectError,};// 封装套接字接口基类class Socket{public:// 封装了socket相关方法virtual ~Socket() {}virtual void CreateSocket() = 0;virtual void BindSocket(uint16_t port) = 0;virtual void ListenSocket(int backlog) = 0;virtual bool ConnectSocket(string &serverip, uint16_t serverport) = 0;virtual int AcceptSocket(string *peerip, uint16_t *peerport) = 0;virtual int GetSockFd() = 0;virtual void SetSockFd(int sockfd) = 0;virtual void CloseSocket() = 0;virtual bool Recv(string *buff, int size) = 0;virtual void Send(string &send_string) = 0;// 方法的集中在一起使用public:void BuildListenSocket(uint16_t port, int backlog = default_backlog){CreateSocket();BindSocket(port);ListenSocket(backlog);}bool BuildConnectSocket(string &serverip, uint16_t serverport){CreateSocket();return ConnectSocket(serverip, serverport);}void BuildNormalSocket(int sockfd){SetSockFd(sockfd);}};class TcpSocket : public Socket{public:TcpSocket(int sockfd = default_sockfd): _sockfd(sockfd){}~TcpSocket() {}void CreateSocket() override{_sockfd = socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0)exit(SocketError);}void BindSocket(uint16_t port) override{int opt = 1;setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));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(_sockfd, (struct sockaddr *)&local, sizeof(local));if (n < 0)exit(BindError);}void ListenSocket(int backlog) override{int n = listen(_sockfd, backlog);if (n < 0)exit(ListenError);}bool ConnectSocket(string &serverip, uint16_t serverport) override{struct sockaddr_in addr;memset(&addr, 0, sizeof(addr));addr.sin_family = AF_INET;addr.sin_port = htons(serverport);// addr.sin_addr.s_addr = inet_addr(serverip.c_str());inet_pton(AF_INET, serverip.c_str(), &addr.sin_addr);int n = connect(_sockfd, (sockaddr *)&addr, sizeof(addr));if (n == 0)return true;return false;}int AcceptSocket(string *peerip, uint16_t *peerport) override{struct sockaddr_in addr;socklen_t len = sizeof(addr);int newsockfd = accept(_sockfd, (sockaddr *)&addr, &len);if (newsockfd < 0)return -1;// *peerip = inet_ntoa(addr.sin_addr);// INET_ADDRSTRLEN 是一个定义在头文件中的宏，表示 IPv4 地址的最大长度char ip_str[INET_ADDRSTRLEN];inet_ntop(AF_INET, &addr.sin_addr, ip_str, INET_ADDRSTRLEN);*peerip = ip_str;*peerport = ntohs(addr.sin_port);return newsockfd;}int GetSockFd() override{return _sockfd;}void SetSockFd(int sockfd) override{_sockfd = sockfd;}void CloseSocket() override{if (_sockfd > default_sockfd)close(_sockfd);}bool Recv(string *buff, int size) override{char inbuffer[size];ssize_t n = recv(_sockfd, inbuffer, size - 1, 0);if (n > 0){inbuffer[n] = 0;*buff += inbuffer;return true;}elsereturn false;}void Send(string &send_string) override{send(_sockfd, send_string.c_str(),send_string.size(),0);}private:int _sockfd;string _ip;uint16_t _port;};
}

PollServer.hpp

#pragma once
#include <iostream>
#include <string>
#include <poll.h>
#include <memory>
#include "Log.hpp"
#include "Socket.hpp"using namespace Net_Work;
const static int gdefaultport = 8888;
const static int gbacklog = 8;
const static int gnum = 1024;
class PollServer
{
public:PollServer(int port) : _port(port), _num(gnum), _listensock(new TcpSocket()){}void HandlerEvent(){for (int i = 0; i < _num; i++){if (_rfds[i].fd == -1)continue;int fd = _rfds[i].fd;short revents = _rfds[i].revents;// 判断事件是否就绪if (revents & POLLIN){// 读事件分两类，一类是新链接到来，一类是新数据到来if (fd == _listensock->GetSockFd()){// 新链接到来lg(Info, "get a new link");// 获取连接std::string clientip;uint16_t clientport;int sockfd = _listensock->AcceptSocket(&clientip, &clientport);if (sockfd == -1){lg(Error, "accept error");continue;}lg(Info, "get a client,client info is# %s:%d,fd: %d", clientip.c_str(), clientport, sockfd);// 此时获取连接成功了，但是不能直接read write,sockfd仍需要交给poll托管 -- 添加到数组_rfds中int pos = 0;for (; pos < _num; pos++){if (_rfds[pos].fd == -1){_rfds[pos].fd = sockfd;_rfds[pos].events = POLLIN;lg(Info, "get a new link, fd is : %d", sockfd);break;}}if (pos == _num){// 1.扩容// 2.关闭close(sockfd);lg(Warning, "server is full, be carefull...");}}else{// 普通的读事件就绪char buffer[1024];ssize_t n = recv(fd, buffer, sizeof(buffer-1), 0);if (n > 0){buffer[n] = 0;lg(Info, "client say# %s", buffer);std::string message = "你好,同志";message += buffer;send(fd, message.c_str(), message.size(), 0);}else{lg(Warning, "client quit ,maybe close or error,close fd: %d", fd);close(fd);// 还要取消poll的关心_rfds[i].fd = -1;_rfds[i].events = 0;_rfds[i].revents = 0;}}}}}void InitServer(){_listensock->BuildListenSocket(_port, gbacklog);_rfds = new struct pollfd[_num];for (int i = 0; i < _num; i++){_rfds[i].fd = -1;_rfds[i].events = 0;_rfds[i].revents = 0;}// 最开始的时候，只有一个文件描述符，Listensock_rfds[0].fd = _listensock->GetSockFd();_rfds[0].events |= POLLIN;}void Loop(){_isrunning = true;// 循环重置select需要的rfdswhile (_isrunning){// 定义时间int timeout = 1000;//PrintDebug();// rfds是输入输出型参数，rfds是在select调用返回时，不断被修改，所以每次需要重置rfdsint n = poll(_rfds, _num, timeout);switch (n){case 0:lg(Info, "select timeout...");break;case -1:lg(Error, "select error!!!");default:// 正常就绪的fdlg(Info, "select success,begin event handler");HandlerEvent();break;}}_isrunning = false;}void Stop(){_isrunning = false;}void PrintDebug(){// std::cout << "current select rfds list is :";// for (int i = 0; i < num; i++)// {//     if (_rfds_array[i] == nullptr)//         continue;//     else//         std::cout << _rfds_array[i]->GetSockFd() << " ";// }// std::cout << std::endl;}private:std::unique_ptr<Socket> _listensock;int _port;bool _isrunning;struct pollfd *_rfds;int _num;
};

Main.cc

#include <iostream>
#include <memory>
#include "PollServer.hpp"void Usage(char* argv)
{std::cout<<"Usage: \n\t"<<argv<<" port\n"<<std::endl;
}
// ./select_server 8080
int main(int argc,char* argv[])
{// std::cout<<num<<std::endl;       1024if(argc!=2){Usage(argv[0]);return -1;}uint16_t localport = std::stoi(argv[1]);std::unique_ptr<PollServer> svr = std::make_unique<PollServer>(localport);svr->InitServer();svr->Loop();return 0;
}

运行结果如下，由于我们poll第三个参数设置的是1000ms，因此每一秒poll都会返回，当发现有新链接的时候，就回去执行函数，在函数中调用write或者read变不会再阻塞了。