一、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;
}