多路IO转接
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout);
1. nfds: 监控的文件描述符集里最大文件描述符加1, 因为此参数会告诉内核检测前多少个文件描述符的状态
2. readfds: 所监听的文件描述符"可读"事件, 传入传出参数
3. writefds: 所监听的文件描述符"可写"事件, 传入传出参数
4. exceptfds: 所监听的文件描述符"异常"事件, 传入传出参数
5. timeout: 定时阻塞监控时间, 3种情况
1> NULL, 永远等下去
2> 设置timeval, 等待固定时间
3> 设置timeval 里时间均为0, 检查描述字后立即返回, 轮询
返回值: 成功 - 返回的是监听的 所有的 集合中, 满足条件的总数.
失败 - 返回-1
struct timeval {
long tv_sec; 秒
long tv_usec; 微秒
};
// 将set清空 0
void FD_ZERO(fd_set* set);
// 将fd 从set中清除出去.
void FD_CLR(int fd, fd_set* set);
// 判断fd 是否在set集合中: == 1 存在
int FD_ISSET(int fd, fd_set* set);
// 将fd 设置到set集合中
void FD_SET(int fd, fd_set* set);
// exp:
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(fd1, &readfds);
FD_SET(fd2, &readfds);
三个集合中一共有几个是有效的: select(...);
for() {
FD_ISSET(fd1, &readfds); -- 1: 满足
}
#include "wrap.h"
#define SERV_PORT 6666
int main(int argc, char* argv[]) {
int i, j, n, maxi;
int nready, client[FD_SETSIZE]; /** 自定义数组client, 防止遍历1024个文件描述符, FD_SETSIZE默认为1024 */
int maxfd, listenfd, connfd, sockfd;
char buff[BUFSIZ], str[INET_ADDRSTRLEN]; /** INET_ADDRSTRLEN 16 */
struct sockaddr_in cli_addr, serv_addr;
socklen_t cli_addr_len, serv_addr_len;
fd_set rset, allset; /** rset 读事件文件描述符集合, allset用来暂存 */
listenfd = Socket(AF_INET, SOCK_STREAM, 0);
bzero(&serv_addr, sizeof(serv_addr));
bzero(&cli_addr, sizeof(cli_addr));
bzero(buff, sizeof(buff));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(SERV_PORT);
serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
serv_addr_len = sizeof(serv_addr);
Bind(listenfd, (struct sockaddr*)&serv_addr, serv_addr_len);
Listen(listenfd, 128);
maxfd = listenfd; /** 起初 listenfd 即为最大文件描述符 */
maxi = -1; /** 将来用作client[]的下标, 初始值指向0个元素之前的下标位置 */
for (i = 0; i < FD_SETSIZE; i++) {
client[i] = -1; /** 用 -1 初始化client[] */
}
FD_ZERO(&allset);
FD_SET(listenfd, &allset); /** 构造select监控文件描述符集 */
while (1) {
rset = allset; /** 每次循环时都从新设置select监控信号集 */
nready = select(maxfd + 1, &rset, NULL, NULL, NULL);
if (nready < 0) {
perr_exit("select error");
}
if (FD_ISSET(listenfd, &rset)) {
cli_addr_len = sizeof(cli_addr);
connfd = Accept(listenfd, (struct sockaddr*)&cli_addr, &cli_addr_len); /** Accept 不会阻塞 */
printf("recive from %s at Port %d\n", inet_ntop(AF_INET, &cli_addr.sin_addr, str, sizeof(str)), ntohs(cli_addr.sin_port));
for (i = 0; i < FD_SETSIZE; i++) {
if (client[i] < 0) { /** 找到client[]中没有使用的位置 */
client[i] = connfd; /** 保存accept返回的文件描述符到client[]里 */
break;
}
}
if (i == FD_SETSIZE) { /** 达到select能监控的文件个数上限1024 */
fputs("too many clients\n", stderr);
exit(1);
}
FD_SET(connfd, &allset); /** 向文件描述符集合allset添加新的文件描述符connfd */
if (connfd > maxfd) {
maxfd = connfd; /** select 第一个参数需要 */
}
if (i > maxi) {
maxi = i; /** 保证maxi存的总是client[]最后一个元素的下标 */
}
if (--nready == 0) {
continue;
}
}
for (i = 0; i <= maxi; i++) {
if ((sockfd = client[i]) < 0) {
continue;
}
if (FD_ISSET(sockfd, & rset)) {
if ((n = Read(sockfd, buff, sizeof(buff))) == 0) {
Close(sockfd);
FD_CLR(sockfd, &allset);
client[i] = -1;
} else if (n > 0) {
for (j = 0; j < n; j++) {
buff[j] = toupper(buff[j]);
}
sleep(10);
Write(sockfd, buff, n);
}
if (--nready == 0) {
break;
}
}
}
}
return 0;
}
#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
fds: 结构体数组 的首地址
fds[0].fd = listen_fd;
fds[0].events = POLLIN/POLLOUT/POLLERR
fds[0].revents = 0; // 该值在设置时没有用, 当监听的fd有事件返回的时候, 该值会被操作系统赋值为对应的事件;POLLIN/POLLOUT/POLLERR
nfds: 数组中元素的个数
timeout: 毫秒级等待
-1 阻塞等, #define INFTIM -1, Linux中没有定义此宏
0 立即返回, 不阻塞进程
>0 等待指定毫秒数, 如当前系统时间精度不够毫秒, 向上取值
poll(fds, 5, -1);
struct pollfd {
int fd; /* file descriptor */ 描述符
short events; /* requested events */ 描述符对应的事件
short revents; /* returned events */ 返回的事件
};
1> poll可以突破1024个文件描述符的限制
2> 与select(传入传出参数)相比, 监听集合与返回集合分离
3> 搜索的返回变小, 不是固定1024个
4> 查看一个进程可以打开的socketm描述符上限
修改上限值
文件尾部写入以下配置
将用户注销后, 使其生效
#define MAXLINE 80
#define SERV_PORT 6666
#define OPEN_MAX 1024
int main(int argc, char* argv[]) {
int i, j, ret, maxi, listen_fd, conn_fd, sock_fd;
int nready;
ssize_t n;
char buff[MAXLINE], str[INET_ADDRSTRLEN];
socklen_t cli_len;
struct pollfd client[OPEN_MAX];
struct sockaddr_in serv_addr, cli_addr;
bzero(buff, sizeof(buff));
bzero(str, sizeof(str));
bzero(&serv_addr, sizeof(serv_addr));
bzero(&cli_addr, sizeof(cli_addr));
bzero(client, sizeof(client));
listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (listen_fd < 0) {
printf("%s\n", strerror(errno));
exit(1);
}
// 端口复用
int opt = 1;
setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(SERV_PORT);
ret = inet_pton(AF_INET, "127.0.0.1", (void*)&(serv_addr.sin_addr.s_addr));
if (ret < 0) {
printf("%s\n", strerror(errno));
exit(1);
} else if (ret == 0) {
printf("this host had been used\n");
exit(1);
}
ret = bind(listen_fd, (struct sockaddr*)&serv_addr, sizeof(serv_addr));
if (ret < 0) {
printf("%s\n", strerror(errno));
exit(1);
}
ret = listen(listen_fd, 128);
if (ret < 0) {
printf("%s\n", strerror(errno));
exit(1);
}
client[0].fd = listen_fd; /** 要监听的第一个文件描述符, 存入client[0] */
client[0].events = POLLIN; /** listen_fd监听普通读事件 */
for (i = 1; i < OPEN_MAX; i++) {
client[i].fd = -1; /** 用-1初始化client里剩余的元素, 因为0也是文件描述符, 不能使用0*/
}
maxi = 0; /** client[]有效元素中最大元素的下标 */
for (;;) {
nready = poll(client, maxi + 1, -1); /** 阻塞监听是否有连接请求 */
if (client[0].revents & POLLIN) { /** listen_fd有读事件就绪 */
cli_len = sizeof(cli_addr);
conn_fd = accept(listen_fd, (struct sockaddr*)&cli_addr, &cli_len);
printf("recived from %s at Port %d\n", inet_ntop(AF_INET, &cli_addr.sin_addr, str, sizeof(str)), ntohs(cli_addr.sin_port));
for (i = 1; i < OPEN_MAX; i++) {
if (client[i].fd < 0) {
client[i].fd = conn_fd; /** 找到client[]中空闲的位置, 存放accept返回的conn_fd*/
break;
}
}
if (i == OPEN_MAX) { /** 达到了最大客户端数量 */
printf("too many clients\n");
exit(1);
}
client[i].events = POLLIN; /** 设置刚刚返回的conn_fd, 监控读事件 */
if (i > maxi) {
maxi = i; /** 更新client[]中 最大元素下标 */
}
if (--nready <= 0) {
continue; /** 没有更多就绪事件时, 继续回到poll阻塞*/
}
}
for (i = 1; i < maxi; i++) {
/** 前面的if没有满足, 说明没有client_fd满足, 检测client[], 看是哪个conn_fd就绪 */
if ((sock_fd = client[i].fd) < 0) {
continue;
}
if (client[i].revents & POLLIN) {
if ((n = read(sock_fd, buff, sizeof(buff))) < 0) {
/** connection reset by client */
if (errno == ECONNRESET) {
// 收到RST标志
printf("client[%d] aborted connection\n", i);
close(sock_fd);
client[i].fd = -1; /** poll中不监控该描述符, 直接置为-1即可, 不用像select那样移除 */
} else {
printf("read error\n");
exit(1);
}
} else if (n == 0) { /** 说明客户端先关闭了链接*/
printf("client[%d] closed connection\n", i);
close(sock_fd);
client[i].fd = -1;
} else {
for (j = 0; j < n; j++) {
buff[j] = toupper(buff[j]);
}
write(sock_fd, buff, n);
}
if (--nready <= 0) {
break;
}
}
}
}
printf("\n");
return 0;
}
1> 原型
#include <sys/epoll.h>
1) 创建一个epoll句柄, 参数size用来告诉内核监听的文件描述符的个数(建议值), 跟内存大小有关, 返回一个文件描述符.
int epoll_create(int size); //size监听数目
2) 控制某个epoll监控的文件描述符上的事件: 注册、修改、删除.
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
epfd: 为epoll_create的句柄
op: 表示动作, 用三个宏来表示
EPOLL_CTL_ADD (注册新的fd到epfd);
EPOLL_CTL_MOD (修改已经注册的fd的监听事件);
EPOLL_CTL_DEL (从epfd删除一个fd);
fd: 要操作的描述符
event: 告诉内核需要监听的事件
typedef union epoll_data {
void* ptr;
int fd;
uint32_t u32;
uint64_t u64;
}epoll_data_t;
struct epoll_event {
__uint32_t events; /** Epoll events */
EPOLLIN
EPOLLOUT
EPOLLERR
epoll_data_t data; /** User data variable */
};
2) 等待所监控文件描述符上有事件的产生, 类似于select()调用
int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);
events: 传出参数, 用来存内核得到事件的集合;
maxevents: 告知内核这个events有多大, 这个maxevents的值不能大于创建epoll_create()时的size;
timeout: 超时时间
-1 阻塞
0 立即返回, 非阻塞
>0 指定毫秒
返回值: 成功返回有多少文件描述符就绪, 时间到时返回0, 出错返回-1 errno;
3) 边沿触发(epoll ET)&水平触发(epoll LT)
边沿触发: 高电频->低电频 || 低点频->高电频 (0->1 || 1->0)
水平触发: 高电频->高电频 || 低电频->低电频 (1->1 || 0->0)
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <pthread.h>
#include <semaphore.h>
#include <sys/socket.h>
#include <ctype.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include "wrap.h"
#define HOST "127.0.0.1"
#define PORT 6666
#define OPEN_MAX 5000
#define MAXLINE 1024
int main(int argc, char* argv[]) {
int listen_fd, conn_fd, sock_fd, epfd;
int i, j, n, ret, res_count;
struct sockaddr_in serv_addr, cli_addr;
socklen_t serv_addr_len, cli_addr_len;
char buff[MAXLINE], str[INET_ADDRSTRLEN];
struct epoll_event res_events[OPEN_MAX], listen_event;
bzero(&serv_addr, sizeof(serv_addr));
bzero(&cli_addr, sizeof(cli_addr));
bzero(buff, sizeof(buff));
bzero(str, sizeof(str));
listen_fd = Socket(AF_INET, SOCK_STREAM, 0);
int opt = 1;
ret = setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
if (ret == -1) {
perr_exit("set sock opt error");
}
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(PORT);
ret = inet_pton(AF_INET, HOST, (void*)&(serv_addr.sin_addr.s_addr));
if (ret <= 0) {
perr_exit("inet_pton() error");
}
serv_addr_len = sizeof(serv_addr);
Bind(listen_fd, (struct sockaddr*)&serv_addr, serv_addr_len);
Listen(listen_fd, 128);
epfd = epoll_create(10);
if(epfd < 0) {
perr_exit("epoll_create error");
}
struct epoll_event event;
event.events = EPOLLIN;
event.data.fd = listen_fd;
ret = epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event);
if (ret < 0) {
perr_exit("epoll_ctl error");
}
while(1) {
res_count = epoll_wait(epfd, res_events, OPEN_MAX, -1);
if (res_count < 0) {
perr_exit("epoll_wait error");
} else {
for(i = 0; i < res_count; i++) {
listen_event = res_events[i];
if (!(listen_event.events & EPOLLIN)) {
continue;
}
sock_fd = listen_event.data.fd;
if (sock_fd == listen_fd) {
bzero(&cli_addr, sizeof(cli_addr));
cli_addr_len = sizeof(cli_addr);
conn_fd = Accept(listen_fd, (struct sockaddr*)&cli_addr, &cli_addr_len);
printf("connection fd[%d]\n", conn_fd);
struct epoll_event new_event;
new_event.events = EPOLLIN;
new_event.data.fd = conn_fd;
ret = epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &new_event);
if (ret < 0) {
perr_exit("epoll_ctl error");
}
} else {
n = Read(sock_fd, buff, sizeof(buff));
if (n == 0) {
// close;
ret = epoll_ctl(epfd, EPOLL_CTL_DEL, sock_fd, &listen_event);
if (ret == -1) {
perr_exit("epoll_ctl error");
}
Close(sock_fd);
} else if (n < 0) {
if (errno == ECONNRESET) {
// 收到RST标志
printf("client[%d] aborted connection\n", i);
ret = epoll_ctl(epfd, EPOLL_CTL_DEL, sock_fd, &listen_event);
if (ret == -1) {
perr_exit("epoll_ctl error");
}
Close(sock_fd);
} else {
perr_exit("read error");
}
} else {
for (j = 0; j < n; j++) {
buff[j] = toupper(buff[j]);
}
Writen(sock_fd, buff, n);
}
}
}
}
}
return 0;
}
边沿触发&非阻塞IO
#define HOST "127.0.0.1"
#define PORT 6666
#define OPEN_MAX 5000
#define MAXLINE 5
int main(int argc, char* argv[]) {
int listen_fd, conn_fd, sock_fd, epfd;
int i, j, n, ret, res_count, flag;
struct sockaddr_in serv_addr, cli_addr;
socklen_t serv_addr_len, cli_addr_len;
char buff[MAXLINE], str[INET_ADDRSTRLEN];
struct epoll_event res_events[OPEN_MAX], listen_event;
bzero(&serv_addr, sizeof(serv_addr));
bzero(&cli_addr, sizeof(cli_addr));
bzero(buff, sizeof(buff));
bzero(str, sizeof(str));
listen_fd = Socket(AF_INET, SOCK_STREAM, 0);
int opt = 1;
ret = setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
if (ret == -1) {
perr_exit("set sock opt error");
}
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(PORT);
ret = inet_pton(AF_INET, HOST, (void*)&(serv_addr.sin_addr.s_addr));
if (ret <= 0) {
perr_exit("inet_pton() error");
}
serv_addr_len = sizeof(serv_addr);
Bind(listen_fd, (struct sockaddr*)&serv_addr, serv_addr_len);
Listen(listen_fd, 128);
epfd = epoll_create(10);
if(epfd < 0) {
perr_exit("epoll_create error");
}
struct epoll_event event;
event.events = EPOLLIN | EPOLLET; // 边沿触发
event.data.fd = listen_fd;
ret = epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event);
if (ret < 0) {
perr_exit("epoll_ctl error");
}
while(1) {
res_count = epoll_wait(epfd, res_events, OPEN_MAX, -1);
if (res_count < 0) {
perr_exit("epoll_wait error");
} else {
for(i = 0; i < res_count; i++) {
listen_event = res_events[i];
if (!(listen_event.events & EPOLLIN)) {
continue;
}
sock_fd = listen_event.data.fd;
if (sock_fd == listen_fd) {
bzero(&cli_addr, sizeof(cli_addr));
cli_addr_len = sizeof(cli_addr);
conn_fd = Accept(listen_fd, (struct sockaddr*)&cli_addr, &cli_addr_len);
printf("connection fd[%d]\n", conn_fd);
flag = fcntl(conn_fd, F_GETFL); /// flag设置
flag |= O_NONBLOCK;
fcntl(conn_fd, F_SETFL, flag);
struct epoll_event new_event;
new_event.events = EPOLLIN;
new_event.data.fd = conn_fd;
ret = epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &new_event);
if (ret < 0) {
perr_exit("epoll_ctl error");
}
} else {
while((n = Read(sock_fd, buff, sizeof(buff) / 2)) > 0) {
for (j = 0; j < n; j++) {
buff[j] = toupper(buff[j]);
}
Writen(sock_fd, buff, n);
}
/**
ret = epoll_ctl(epfd, EPOLL_CTL_DEL, sock_fd, &listen_event);
if (ret == -1) {
perr_exit("epoll_ctl error");
}
Close(sock_fd);
*/
}
}
}
}
return 0;
}
3) 反应堆模型
/*
*epoll基于非阻塞I/O事件驱动
*/
#include <stdio.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#define MAX_EVENTS 1024 //监听上限数
#define BUFLEN 4096
#define SERV_PORT 8080
void recvdata(int fd, int events, void *arg);
void senddata(int fd, int events, void *arg);
/* 描述就绪文件描述符相关信息 */
struct myevent_s {
int fd; //要监听的文件描述符
int events; //对应的监听事件
void *arg; //泛型参数
void (*call_back)(int fd, int events, void *arg); //回调函数
int status; //是否在监听:1->在红黑树上(监听), 0->不在(不监听)
char buf[BUFLEN];
int len;
long last_active; //记录每次加入红黑树 g_efd 的时间值
};
int g_efd; //全局变量, 保存epoll_create返回的文件描述符
struct myevent_s g_events[MAX_EVENTS+1]; //自定义结构体类型数组. +1-->listen fd
/*将结构体 myevent_s 成员变量 初始化*/
void eventset(struct myevent_s *ev, int fd, void (*call_back)(int, int, void *), void *arg)
{
ev->fd = fd;
ev->call_back = call_back;
ev->events = 0;
ev->arg = arg;
ev->status = 0;
//memset(ev->buf, 0, sizeof(ev->buf));
//ev->len = 0;
ev->last_active = time(NULL); //调用eventset函数的时间
return;
}
/* 向 epoll监听的红黑树 添加一个 文件描述符 */
void eventadd(int efd, int events, struct myevent_s *ev)
{
struct epoll_event epv = {0, {0}};
int op;
epv.data.ptr = ev;
epv.events = ev->events = events; //EPOLLIN 或 EPOLLOUT
if (ev->status == 1) { //已经在红黑树 g_efd 里
op = EPOLL_CTL_MOD; //修改其属性
} else { //不在红黑树里
op = EPOLL_CTL_ADD; //将其加入红黑树 g_efd, 并将status置1
ev->status = 1;
}
if (epoll_ctl(efd, op, ev->fd, &epv) < 0) //实际添加/修改
printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
else
printf("event add OK [fd=%d], op=%d, events[%0X]\n", ev->fd, op, events);
return ;
}
/* 从epoll 监听的 红黑树中删除一个 文件描述符*/
void eventdel(int efd, struct myevent_s *ev)
{
struct epoll_event epv = {0, {0}};
if (ev->status != 1) //不在红黑树上
return ;
epv.data.ptr = ev;
ev->status = 0; //修改状态
epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv); //从红黑树 efd 上将 ev->fd 摘除
return ;
}
/* 当有文件描述符就绪, epoll返回, 调用该函数 与客户端建立链接 */
void acceptconn(int lfd, int events, void *arg)
{
struct sockaddr_in cin;
socklen_t len = sizeof(cin);
int cfd, i;
if ((cfd = accept(lfd, (struct sockaddr *)&cin, &len)) == -1) {
if (errno != EAGAIN && errno != EINTR) {
/* 暂时不做出错处理 */
}
printf("%s: accept, %s\n", __func__, strerror(errno));
return ;
}
do {
for (i = 0; i < MAX_EVENTS; i++) //从全局数组g_events中找一个空闲元素
if (g_events[i].status == 0) //类似于select中找值为-1的元素
break; //跳出 for
if (i == MAX_EVENTS) {
printf("%s: max connect limit[%d]\n", __func__, MAX_EVENTS);
break; //跳出do while(0) 不执行后续代码
}
int flag = 0;
if ((flag = fcntl(cfd, F_SETFL, O_NONBLOCK)) < 0) { //将cfd也设置为非阻塞
printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
break;
}
/* 给cfd设置一个 myevent_s 结构体, 回调函数 设置为 recvdata */
eventset(&g_events[i], cfd, recvdata, &g_events[i]);
eventadd(g_efd, EPOLLIN, &g_events[i]); //将cfd添加到红黑树g_efd中,监听读事件
} while(0);
printf("new connect [%s:%d][time:%ld], pos[%d]\n",
inet_ntoa(cin.sin_addr), ntohs(cin.sin_port), g_events[i].last_active, i);
return ;
}
void recvdata(int fd, int events, void *arg)
{
struct myevent_s *ev = (struct myevent_s *)arg;
int len;
len = recv(fd, ev->buf, sizeof(ev->buf), 0); //读文件描述符, 数据存入myevent_s成员buf中
eventdel(g_efd, ev); //将该节点从红黑树上摘除
if (len > 0) {
ev->len = len;
ev->buf[len] = '\0'; //手动添加字符串结束标记
printf("C[%d]:%s\n", fd, ev->buf);
eventset(ev, fd, senddata, ev); //设置该 fd 对应的回调函数为 senddata
eventadd(g_efd, EPOLLOUT, ev); //将fd加入红黑树g_efd中,监听其写事件
} else if (len == 0) {
close(ev->fd);
/* ev-g_events 地址相减得到偏移元素位置 */
printf("[fd=%d] pos[%ld], closed\n", fd, ev-g_events);
} else {
close(ev->fd);
printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
}
return;
}
void senddata(int fd, int events, void *arg)
{
struct myevent_s *ev = (struct myevent_s *)arg;
int len;
len = send(fd, ev->buf, ev->len, 0); //直接将数据 回写给客户端。未作处理
/*
printf("fd=%d\tev->buf=%s\ttev->len=%d\n", fd, ev->buf, ev->len);
printf("send len = %d\n", len);
*/
if (len > 0) {
printf("send[fd=%d], [%d]%s\n", fd, len, ev->buf);
eventdel(g_efd, ev); //从红黑树g_efd中移除
eventset(ev, fd, recvdata, ev); //将该fd的 回调函数改为 recvdata
eventadd(g_efd, EPOLLIN, ev); //从新添加到红黑树上， 设为监听读事件
} else {
close(ev->fd); //关闭链接
eventdel(g_efd, ev); //从红黑树g_efd中移除
printf("send[fd=%d] error %s\n", fd, strerror(errno));
}
return ;
}
/*创建 socket, 初始化lfd */
void initlistensocket(int efd, short port)
{
int lfd = socket(AF_INET, SOCK_STREAM, 0);
fcntl(lfd, F_SETFL, O_NONBLOCK); //将socket设为非阻塞
/* void eventset(struct myevent_s *ev, int fd, void (*call_back)(int, int, void *), void *arg); */
eventset(&g_events[MAX_EVENTS], lfd, acceptconn, &g_events[MAX_EVENTS]);
/* void eventadd(int efd, int events, struct myevent_s *ev) */
eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS]);
struct sockaddr_in sin;
memset(&sin, 0, sizeof(sin)); //bzero(&sin, sizeof(sin))
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = INADDR_ANY;
sin.sin_port = htons(port);
bind(lfd, (struct sockaddr *)&sin, sizeof(sin));
listen(lfd, 20);
return ;
}
int main(int argc, char *argv[])
{
unsigned short port = SERV_PORT;
if (argc == 2)
port = atoi(argv[1]); //使用用户指定端口.如未指定,用默认端口
g_efd = epoll_create(MAX_EVENTS+1); //创建红黑树,返回给全局 g_efd
if (g_efd <= 0)
printf("create efd in %s err %s\n", __func__, strerror(errno));
initlistensocket(g_efd, port); //初始化监听socket
struct epoll_event events[MAX_EVENTS+1]; //保存已经满足就绪事件的文件描述符数组
printf("server running:port[%d]\n", port);
int checkpos = 0, i;
while (1) {
/* 超时验证，每次测试100个链接，不测试listenfd 当客户端60秒内没有和服务器通信，则关闭此客户端链接 */
long now = time(NULL); //当前时间
for (i = 0; i < 100; i++, checkpos++) { //一次循环检测100个。 使用checkpos控制检测对象
if (checkpos == MAX_EVENTS)
checkpos = 0;
if (g_events[checkpos].status != 1) //不在红黑树 g_efd 上
continue;
long duration = now - g_events[checkpos].last_active; //客户端不活跃的世间
if (duration >= 60) {
close(g_events[checkpos].fd); //关闭与该客户端链接
printf("[fd=%d] timeout\n", g_events[checkpos].fd);
eventdel(g_efd, &g_events[checkpos]); //将该客户端 从红黑树 g_efd移除
}
}
/*监听红黑树g_efd, 将满足的事件的文件描述符加至events数组中, 1秒没有事件满足, 返回 0*/
int nfd = epoll_wait(g_efd, events, MAX_EVENTS+1, 1000);
if (nfd < 0) {
printf("epoll_wait error, exit\n");
break;
}
for (i = 0; i < nfd; i++) {
/*使用自定义结构体myevent_s类型指针, 接收 联合体data的void *ptr成员*/
struct myevent_s *ev = (struct myevent_s *)events[i].data.ptr;
if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) { //读就绪事件
ev->call_back(ev->fd, events[i].events, ev->arg);
}
if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) { //写就绪事件
ev->call_back(ev->fd, events[i].events, ev->arg);
}
}
}
/* 退出前释放所有资源 */
return 0;
}
网友评论