poll案例

#include <iostream>
#include <sstream>
#include <map>

#include <stdio.h>
#include <errno.h>
#include <string.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <unistd.h>
#include <fcntl.h>
#include <poll.h>

#define LOCAL_AF            AF_INET             // 本地地址类型
#define LOCAL_IP_ADDR       "172.20.0.115"
#define LOCAL_PORT          8090

#define BACKLOG             10                  // 监听等待队列个数

#define MAX_FD              1024                // 最大打开文件描述符数目
#define MAX_USER            3                   // 最大用户数

#define FAILED_TEST_CNT     10                  // socket调用失败, 连续测试次数

// 统计连接socket和对应的socket连续调用失败次数, 失败 FAILED_TEST_CNT 次，关闭移除socket
typedef std::map<int, int> Clients;

using namespace std;

static void showErr()
{
printf("errno: %d, err_msg: %s\n", errno, strerror(errno));
}

// 设置socket为非阻塞
static bool setSocketNonBlock(int socketFd)
{
int flags = fcntl(socketFd, F_GETFL);
if (-1 == flags)
{
cout << "fcntl F_GETFL error" << endl;
return false;
}

int ret = fcntl(socketFd, F_SETFL, O_NONBLOCK | flags);
if (-1 == ret)
{
cout << "fcntl F_SETFL error" << endl;
return false;
}
return true;
}

// 设置socket关闭时不等待缓存区数据发送完毕(等待可能导致缓冲区数据无法发送导致一直占用端口: 连接一直处于FIN_WAIT1状态)
static int disableLinger(int socketFd)
{
struct linger lingerVal;
lingerVal.l_onoff = 1;
lingerVal.l_linger = 0;
int ret = setsockopt(socketFd, SOL_SOCKET, SO_LINGER, &lingerVal, sizeof(lingerVal));
if (0 != ret)
{
cout << "setsockopt linger error" << endl;
return false;
}
return true;
}

int main(int argc, char **argv)
{
int ret(0);

// 生成监听socket
int listenSocket(-1);
listenSocket = socket(LOCAL_AF, SOCK_STREAM, IPPROTO_TCP);
if (-1 == listenSocket)
{
cout << "socket error" << endl;
return 1;
}

if (!setSocketNonBlock(listenSocket))
{
return 1;
}

if (!disableLinger(listenSocket))
{
return 1;
}

// 绑定监听socket到指定网卡和端口
in_addr localAddr;
inet_pton(LOCAL_AF, LOCAL_IP_ADDR, &localAddr);
sockaddr_in localSockAddr;
localSockAddr.sin_family = LOCAL_AF;
localSockAddr.sin_addr = localAddr;
// localSockAddr.sin_addr.s_addr = htonl(INADDR_ANY);      // 绑定到所有网卡的指定端口(多网卡自适应)
localSockAddr.sin_port = htons(LOCAL_PORT);
ret = bind(listenSocket, (sockaddr *)&localSockAddr, sizeof(localSockAddr));
if (0 != ret)
{
cout << "bind error" << endl;
return 1;
}

ret = listen(listenSocket, BACKLOG);
if (0 != ret)
{
cout << "listen error" << endl;
return 1;
}

// 创建pollfd结构体数组, 使用poll函数管理监听socket和所有的连接socket
pollfd pollFds[MAX_FD];
memset(pollFds, 0, sizeof(pollFds));    // memset初始化内存只适用于POD类型
int usedFdIdx = -1;     // 已用到的pollfd数组索引

// 将监听socket纳入pollfd数组第一个元素
++usedFdIdx;
pollFds[usedFdIdx].fd = listenSocket;
pollFds[usedFdIdx].events = POLLIN;     // 客户端请求到达事件(connect等)

Clients clients;

bool done(false);
while (!done)
{
ret = poll(pollFds, usedFdIdx + 1, 1000);   // 等待1秒
switch(ret)
{
case -1:
showErr();
return 1;
case 0:
cout << "timeout" << endl;
break;
default:
// 遍历客户端连接socket
for (int idx = 1; idx <= usedFdIdx; ++idx)
{
Clients::iterator iter = clients.find(pollFds[idx].fd);

// 客户端可写
if (pollFds[idx].revents & POLLOUT)
{
string msg = "i'm client\r\n";
ret = send(pollFds[idx].fd, msg.c_str(), msg.length(), 0);
// send failed
if (-1 == ret && EAGAIN != errno)    // EAGAIN代表缓冲区满了, 不是错误需要排除
{
iter->second++;

// 连续
if (iter->second >= 10)
{
ret = close(pollFds[idx].fd);
if (0 == ret)
{
cout << "close socket :" << pollFds[idx].fd << " success" <<endl;
}
else
{
cout << "close socket :" << pollFds[idx].fd << " failed" <<endl;
}

clients.erase(iter);
for (int i = idx; i < usedFdIdx; ++i)
{
pollFds[i] = pollFds[i + 1];
}
--usedFdIdx;
}
}
else
{
iter->second = 0;
}

cout << "send ret: " << ret << endl;
}
else if (pollFds[idx].revents & POLLERR)
{
cout << pollFds[idx].fd << " error" << endl;

iter->second++;

// remove socket
if (iter->second >= 10)
{
ret = close(pollFds[idx].fd);
if (0 == ret)
{
cout << "close socket :" << pollFds[idx].fd << " success" <<endl;
}
else
{
cout << "close socket :" << pollFds[idx].fd << " failed" <<endl;
}

clients.erase(iter);
for (int i = idx; i < usedFdIdx; ++i)
{
pollFds[i] = pollFds[i + 1];
}
--usedFdIdx;
}
}

if (idx == usedFdIdx)
{
sleep(1);
}
}

// 客户端connect请求到达
if (pollFds[0].revents & POLLIN)
{
if (usedFdIdx < MAX_USER)
{
int connectSocket(-1);
connectSocket = accept(listenSocket, NULL, NULL);
if (-1 == connectSocket)
{
cout << "accept error" << endl;
break;
}
cout << "accept success" << endl;

if (!setSocketNonBlock(connectSocket))
{
break;
}

if (!disableLinger(connectSocket))
{
break;
}

// 连接socket使用poll管理
++usedFdIdx;
pollFds[usedFdIdx].fd = connectSocket;
pollFds[usedFdIdx].events = POLLOUT;

//
clients.insert(make_pair<int, int>(connectSocket, 0));
}
else    // fd too many
{

}
}
}
}

return 0;
}