【在Linux世界中追寻巨大的One Piece】多路转接epoll(续) - ToB企服应用市场:ToB评测及商务社交产业平台

///
// 封装一个 Epoll 服务器, 只考虑读就绪的情况
///
#pragma once
#include <vector>
#include <functional>
#include <iostream>
#include <sys/epoll.h>
#include "tcp_socket.hpp"
typedef std::function<void(const std::string&, std::string*resp)> Handler;
class Epoll
{
public:
Epoll()
{
epoll_fd_ = epoll_create(10);
}
~Epoll()
{
close(epoll_fd_);
}
bool Add(const TcpSocket& sock) const
{
int fd = sock.GetFd();
printf("[Epoll Add] fd = %d\n", fd);
epoll_event ev;
ev.data.fd = fd;
ev.events = EPOLLIN;
int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);
if (ret < 0)
{
perror("epoll_ctl ADD");
return false;
}
return true;
}
bool Del(const TcpSocket& sock) const
{
int fd = sock.GetFd();
printf("[Epoll Del] fd = %d\n", fd);
int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);
if (ret < 0)
{
perror("epoll_ctl DEL");
return false;
}
return true;
}
bool Wait(std::vector<TcpSocket>* output) const
{
output->clear();
epoll_event events[1000];
int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);
if (nfds < 0)
{
perror("epoll_wait");
return false;
}
// [注意!] 此处必须是循环到 nfds, 不能多循环
for (int i = 0; i < nfds; ++i)
{
TcpSocket sock(events[i].data.fd);
output->push_back(sock);
}
return true;
}
private:
int epoll_fd_;
};
class TcpEpollServer
{
public:
TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip),
port_(port)
{
}
bool Start(Handler handler)
{
// 1. 创建 socket
TcpSocket listen_sock;
CHECK_RET(listen_sock.Socket());
// 2. 绑定
CHECK_RET(listen_sock.Bind(ip_, port_));
// 3. 监听
CHECK_RET(listen_sock.Listen(5));
// 4. 创建 Epoll 对象, 并将 listen_sock 加入进去
Epoll epoll;
epoll.Add(listen_sock);
// 5. 进入事件循环
for (;;)
{
// 6. 进行 epoll_wait
std::vector<TcpSocket> output;
if (!epoll.Wait(&output))
{
continue;
}
// 7. 根据就绪的文件描述符的种类决定如何处理
for (size_t i = 0; i < output.size(); ++i)
{
if (output[i].GetFd() == listen_sock.GetFd())
{
// 如果是 listen_sock, 就调用 accept
TcpSocket new_sock;
listen_sock.Accept(&new_sock);
epoll.Add(new_sock);
}
else
{
// 如果是 new_sock, 就进行一次读写
std::string req, resp;
bool ret = output[i].Recv(&req);
if (!ret)
{
// [注意!!] 需要把不用的 socket 关闭
// 先后顺序别搞反. 不过在 epoll 删除的时候其实就已经关闭 socket 了
epoll.Del(output[i]);
output[i].Close();
continue;
}
handler(req, &resp);
output[i].Send(resp);
} // end for
} // end for (;;)
}
return true;
}
private:
std::string ip_;
uint16_t port_;
};

复制代码

// 以下代码添加在 TcpSocket 类中
// 非阻塞 IO 接口
bool SetNoBlock()
{
int fl = fcntl(fd_, F_GETFL);
if (fl < 0)
{
perror("fcntl F_GETFL");
return false;
}
int ret = fcntl(fd_, F_SETFL, fl | O_NONBLOCK);
if (ret < 0)
{
perror("fcntl F_SETFL");
return false;
}
return true;
}
bool RecvNoBlock(std::string* buf) const
{
// 对于非阻塞 IO 读数据, 如果 TCP 接受缓冲区为空, 就会返回错误
// 错误码为 EAGAIN 或者 EWOULDBLOCK, 这种情况也是意料之中, 需要重试
// 如果当前读到的数据长度小于尝试读的缓冲区的长度, 就退出循环
// 这种写法其实不算特别严谨(没有考虑粘包问题)
buf->clear();
char tmp[1024 * 10] = { 0 };
for (;;)
{
ssize_t read_size = recv(fd_, tmp, sizeof(tmp) - 1, 0);
if (read_size < 0)
{
if (errno == EWOULDBLOCK || errno == EAGAIN)
{
continue;
}
perror("recv");
return false;
}
if (read_size == 0)
{
// 对端关闭, 返回 false
return false;
}
tmp[read_size] = '\0';
*buf += tmp;
if (read_size < (ssize_t)sizeof(tmp) - 1)
{
break;
}
}
return true;
}
bool SendNoBlock(const std::string& buf) const
{
// 对于非阻塞 IO 的写入, 如果 TCP 的发送缓冲区已经满了, 就会出现出错的情况
// 此时的错误号是 EAGAIN 或者 EWOULDBLOCK. 这种情况下不应放弃治疗
// 而要进行重试
ssize_t cur_pos = 0; // 记录当前写到的位置
ssize_t left_size = buf.size();
for (;;)
{
ssize_t write_size = send(fd_, buf.data() + cur_pos,
left_size, 0);
if (write_size < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
// 重试写入
continue;
}
return false;
}
cur_pos += write_size;
left_size -= write_size;
// 这个条件说明写完需要的数据了
if (left_size <= 0)
{
break;
}
}
return true;
}

复制代码

///
// 封装一个 Epoll ET 服务器
// 修改点:
// 1. 对于 new sock, 加上 EPOLLET 标记
// 2. 修改 TcpSocket 支持非阻塞读写
// [注意!] listen_sock 如果设置成 ET, 就需要非阻塞调用 accept 了
///
#pragma once
#include <vector>
#include <functional>
#include <sys/epoll.h>
#include "tcp_socket.hpp"
typedef std::function<void(const std::string&, std::string* resp)> Handler;
class Epoll
{
public:
Epoll()
{
epoll_fd_ = epoll_create(10);
}
~Epoll()
{
close(epoll_fd_);
}
bool Add(const TcpSocket& sock, bool epoll_et = false) const
{
int fd = sock.GetFd();
printf("[Epoll Add] fd = %d\n", fd);
epoll_event ev;
ev.data.fd = fd;
if (epoll_et)
{
ev.events = EPOLLIN | EPOLLET;
}
else
{
ev.events = EPOLLIN;
}
int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);
if (ret < 0)
{
perror("epoll_ctl ADD");
return false;
}
return true;
}
bool Del(const TcpSocket& sock) const
{
int fd = sock.GetFd();
printf("[Epoll Del] fd = %d\n", fd);
int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, NULL);
if (ret < 0)
{
perror("epoll_ctl DEL");
return false;
}
return true;
}
bool Wait(std::vector<TcpSocket>* output) const
{
output->clear();
epoll_event events[1000];
int nfds = epoll_wait(epoll_fd_, events, sizeof(events) / sizeof(events[0]), -1);
if (nfds < 0)
{
perror("epoll_wait");
return false;
}
// [注意!] 此处必须是循环到 nfds, 不能多循环
for (int i = 0; i < nfds; ++i)
{
TcpSocket sock(events[i].data.fd);
output->push_back(sock);
}
return true;
}
private:
int epoll_fd_;
};
class TcpEpollServer
{
public:
TcpEpollServer(const std::string& ip, uint16_t port) : ip_(ip),
port_(port)
{
}
bool Start(Handler handler)
{
// 1. 创建 socket
TcpSocket listen_sock;
CHECK_RET(listen_sock.Socket());
// 2. 绑定
CHECK_RET(listen_sock.Bind(ip_, port_));
// 3. 监听
CHECK_RET(listen_sock.Listen(5));
// 4. 创建 Epoll 对象, 并将 listen_sock 加入进去
Epoll epoll;
epoll.Add(listen_sock);
// 5. 进入事件循环
for (;;)
{
// 6. 进行 epoll_wait
std::vector<TcpSocket> output;
if (!epoll.Wait(&output))
{
continue;
}
// 7. 根据就绪的文件描述符的种类决定如何处理
for (size_t i = 0; i < output.size(); ++i)
{
if (output[i].GetFd() == listen_sock.GetFd())
{
// 如果是 listen_sock, 就调用 accept
TcpSocket new_sock;
listen_sock.Accept(&new_sock);
epoll.Add(new_sock, true);
}
else
{
// 如果是 new_sock, 就进行一次读写
std::string req, resp;
bool ret = output[i].RecvNoBlock(&req);
if (!ret)
{
// [注意!!] 需要把不用的 socket 关闭
// 先后顺序别搞反. 不过在 epoll 删除的时候其实就已经关闭 socket 了
epoll.Del(output[i]);
output[i].Close();
continue;
}
handler(req, &resp);
output[i].SendNoBlock(resp);
printf("[client %d] req: %s, resp: %s\n",
output[i].GetFd(),
req.c_str(), resp.c_str());
} // end for
} // end for (;;)
}
return true;
}
private:
std::string ip_;
uint16_t port_;
};

复制代码