Epoll-ET模式读写之用户连接的封装

前言

服务端而言，对于每一个新的连接我们都需要去保存其基本信息，如ip地址，套接字fd，也需要赋予其唯一标识如用户名。 这里，我们来谈谈对用户连接的封装。

用户连接需要哪些数据

1. 套接字描述符 sockfd 执行读写操作时当然不可缺 2. 连接信息 sockaddr 基本信息的保存 3. 用户缓存区 Buffer 非阻塞读写不可缺 4. 唯一标识 服务端进行消息转发时不可能根据sockfd去进行查找，所以我们需要对每一个连接有一个唯一标识，用于对其进行检索

关于Buffer细节可看Epoll-ET模式下非阻塞读写之Buffer的封装

用户连接需要哪些操作

1. 读 epoll-Et模式下，读必须将缓冲区全部读完，否则不会再次触发 2. 写 epoll-Et模式下，写必须将所写数据全部写入或是写到缓冲区满，否则不会再次触发 3. 断开 断开时，如果是异步读写，需要需要确认写事件已全部完成，再关闭套接字 4. 初始化 保证连接可复用，否则每来一个连接，都要开辟空间，太过低效。

关于读写的细节

本身 send和recv都是线程安全的。 因为读操作，我们可以确定只会有一个线程来对读操作进行处理(通过设置EPOLLONESHOT)，也就是说对读缓冲区的写入(接受数据)和取出(处理数据)都是在一个线程内，所以读缓冲区本质上是线程安全的。 但是写操作，写缓冲区可能会有多个线程对其进行写入和更改，所以我们用一个原子变量，来标志写缓冲区是否被操作，来保证数据读写不会冲突。

代码

本代码数据处理只是简单的进行转发到原本socket里

// // Connection.h // QuoridorServer // // Created by shiyi on 2016/12/2. // Copyright © 2016年 shiyi. All rights reserved. // #ifndef Connection_H #define Connection_H #include <stdio.h> #include <iostream> #include <atomic> #include <fcntl.h> #include <sys/socket.h> #include <sys/types.h> #include <signal.h> #include <string.h> #include <sys/epoll.h> #include <unistd.h> #include <assert.h> #include <stdlib.h> #include <arpa/inet.h> #include <errno.h> #include "Buffer.h" #include "Util.h" #include "BaseFunc.h" const size_t BUFFER_SIZE = 65535; class Connection { public: Connection() : m_writeing(true), m_epollfd(-1), m_sockfd(-1) {} ~Connection(){} //初始化连接 void init(int epollfd, int sockfd, const sockaddr_in& client_addr) { m_epollfd = epollfd; m_sockfd = sockfd; m_writeing = true; m_address = client_addr; //初始化读写缓冲区 m_inBuff.init(); m_outBuff.init(); } void HandleRead() { while(true) { char buf[BUFFER_SIZE]; int ret = recv(m_sockfd, buf, BUFFER_SIZE, 0); if(ret < 0) { //缓冲区内容已读完 if((errno == EAGAIN) || (errno == EWOULDBLOCK)) { break; } //其他错误直接断开连接 Util::outError("HandleRead"); shutdown(); } //断开连接 else if(ret == 0) { shutdown(); } else { //将读取的内容加入缓冲区 m_inBuff.PutData(buf, ret); } //printf("[%s]\n", buf); } worker(); } void HandleWrite() { //更改临界值 if(!m_writeing) return; m_writeing = false; //取出数据 char buf[BUFFER_SIZE]; int len = m_outBuff.GetData(buf, BUFFER_SIZE); int n = len; while (n > 0) { int ret = send(m_sockfd, buf+len-n, n, 0); if (ret < n) { if (ret == -1 && errno != EAGAIN) { Util::outError("write error"); } break; } n -= ret; } //n=0表示数据全部写完，删除写事件 if(n == 0) modFd(m_epollfd, m_sockfd, EPOLLIN | EPOLLET | EPOLLONESHOT); //恢复临界值 m_writeing = true; } void worker() { //解析 //取出数据 char buf[BUFFER_SIZE]; int len = m_inBuff.GetData(buf, BUFFER_SIZE); while(!m_writeing) usleep(1000); m_writeing = false; m_outBuff.PutData(buf, len); modFd(m_epollfd, m_sockfd, EPOLLIN | EPOLLOUT | EPOLLET | EPOLLONESHOT); m_writeing = true; } void shutdown() { //等待写事件完成后关闭 while(!m_writeing) usleep(1000); m_writeing = false; removeFd(m_epollfd, m_sockfd); } private: int m_epollfd; //epoll描述符 int m_sockfd; //套接字描述符 sockaddr_in m_address; //套接字地址 Buffer m_inBuff; //读缓冲 Buffer m_outBuff; //写缓冲 std::atomic_bool m_writeing; //是否正在写 }; #endif /* Connection_H */