多路IO轉(zhuǎn)接
?
#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: 監(jiān)控的文件描述符集里最大文件描述符加1, 因?yàn)榇藚?shù)會(huì)告訴內(nèi)核檢測(cè)前多少個(gè)文件描述符的狀態(tài)
2. readfds: 所監(jiān)聽的文件描述符"可讀"事件, 傳入傳出參數(shù)
3. writefds: 所監(jiān)聽的文件描述符"可寫"事件, 傳入傳出參數(shù)
4. exceptfds: 所監(jiān)聽的文件描述符"異常"事件, 傳入傳出參數(shù)
5. timeout: 定時(shí)阻塞監(jiān)控時(shí)間, 3種情況
1> NULL, 永遠(yuǎn)等下去
2> 設(shè)置timeval, 等待固定時(shí)間
3> 設(shè)置timeval 里時(shí)間均為0, 檢查描述字后立即返回, 輪詢
返回值: 成功 - 返回的是監(jiān)聽的 所有的 集合中, 滿足條件的總數(shù).
失敗 - 返回-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 設(shè)置到set集合中
void FD_SET(int fd, fd_set* set);
// exp:
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(fd1, &readfds);
FD_SET(fd2, &readfds);
三個(gè)集合中一共有幾個(gè)是有效的: 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]; /** 自定義數(shù)組client, 防止遍歷1024個(gè)文件描述符, FD_SETSIZE默認(rèn)為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[]的下標(biāo), 初始值指向0個(gè)元素之前的下標(biāo)位置 */
for (i = 0; i < FD_SETSIZE; i++) {
client[i] = -1; /** 用 -1 初始化client[] */
}
FD_ZERO(&allset);
FD_SET(listenfd, &allset); /** 構(gòu)造select監(jiān)控文件描述符集 */
while (1) {
rset = allset; /** 每次循環(huán)時(shí)都從新設(shè)置select監(jiān)控信號(hào)集 */
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 不會(huì)阻塞 */
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) { /** 達(dá)到select能監(jiān)控的文件個(gè)數(shù)上限1024 */
fputs("too many clients\n", stderr);
exit(1);
}
FD_SET(connfd, &allset); /** 向文件描述符集合allset添加新的文件描述符connfd */
if (connfd > maxfd) {
maxfd = connfd; /** select 第一個(gè)參數(shù)需要 */
}
if (i > maxi) {
maxi = i; /** 保證maxi存的總是client[]最后一個(gè)元素的下標(biāo) */
}
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: 結(jié)構(gòu)體數(shù)組 的首地址
fds[0].fd = listen_fd;
fds[0].events = POLLIN/POLLOUT/POLLERR
fds[0].revents = 0; // 該值在設(shè)置時(shí)沒有用, 當(dāng)監(jiān)聽的fd有事件返回的時(shí)候, 該值會(huì)被操作系統(tǒng)賦值為對(duì)應(yīng)的事件;POLLIN/POLLOUT/POLLERR
nfds: 數(shù)組中元素的個(gè)數(shù)
timeout: 毫秒級(jí)等待
-1 阻塞等, #define INFTIM -1, Linux中沒有定義此宏
0 立即返回, 不阻塞進(jìn)程
>0 等待指定毫秒數(shù), 如當(dāng)前系統(tǒng)時(shí)間精度不夠毫秒, 向上取值
poll(fds, 5, -1);
struct pollfd {
int fd; /* file descriptor */ 描述符
short events; /* requested events */ 描述符對(duì)應(yīng)的事件
short revents; /* returned events */ 返回的事件
};
??1> poll可以突破1024個(gè)文件描述符的限制
??2> 與select(傳入傳出參數(shù))相比, 監(jiān)聽集合與返回集合分離
??3> 搜索的返回變小, 不是固定1024個(gè)
??4> 查看一個(gè)進(jìn)程可以打開的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);
}
// 端口復(fù)用
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; /** 要監(jiān)聽的第一個(gè)文件描述符, 存入client[0] */
client[0].events = POLLIN; /** listen_fd監(jiān)聽普通讀事件 */
for (i = 1; i < OPEN_MAX; i++) {
client[i].fd = -1; /** 用-1初始化client里剩余的元素, 因?yàn)?也是文件描述符, 不能使用0*/
}
maxi = 0; /** client[]有效元素中最大元素的下標(biāo) */
for (;;) {
nready = poll(client, maxi + 1, -1); /** 阻塞監(jiān)聽是否有連接請(qǐng)求 */
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) { /** 達(dá)到了最大客戶端數(shù)量 */
printf("too many clients\n");
exit(1);
}
client[i].events = POLLIN; /** 設(shè)置剛剛返回的conn_fd, 監(jiān)控讀事件 */
if (i > maxi) {
maxi = i; /** 更新client[]中 最大元素下標(biāo) */
}
if (--nready <= 0) {
continue; /** 沒有更多就緒事件時(shí), 繼續(xù)回到poll阻塞*/
}
}
for (i = 1; i < maxi; i++) {
/** 前面的if沒有滿足, 說明沒有client_fd滿足, 檢測(cè)client[], 看是哪個(gè)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標(biāo)志
printf("client[%d] aborted connection\n", i);
close(sock_fd);
client[i].fd = -1; /** poll中不監(jiān)控該描述符, 直接置為-1即可, 不用像select那樣移除 */
} else {
printf("read error\n");
exit(1);
}
} else if (n == 0) { /** 說明客戶端先關(guān)閉了鏈接*/
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) 創(chuàng)建一個(gè)epoll句柄, 參數(shù)size用來告訴內(nèi)核監(jiān)聽的文件描述符的個(gè)數(shù)(建議值), 跟內(nèi)存大小有關(guān), 返回一個(gè)文件描述符.
????int epoll_create(int size);?//size監(jiān)聽數(shù)目
??2) 控制某個(gè)epoll監(jiān)控的文件描述符上的事件: 注冊(cè)、修改咽安、刪除.
????int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
??????epfd: 為epoll_create的句柄
??????op: 表示動(dòng)作, 用三個(gè)宏來表示
???????EPOLL_CTL_ADD (注冊(cè)新的fd到epfd);
???????EPOLL_CTL_MOD (修改已經(jīng)注冊(cè)的fd的監(jiān)聽事件);
???????EPOLL_CTL_DEL (從epfd刪除一個(gè)fd);
??????fd: 要操作的描述符
??????event: 告訴內(nèi)核需要監(jiān)聽的事件
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) 等待所監(jiān)控文件描述符上有事件的產(chǎn)生, 類似于select()調(diào)用
????int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);
??????events: 傳出參數(shù), 用來存內(nèi)核得到事件的集合;
??????maxevents: 告知內(nèi)核這個(gè)events有多大, 這個(gè)maxevents的值不能大于創(chuàng)建epoll_create()時(shí)的size;
??????timeout: 超時(shí)時(shí)間
???????? -1??阻塞
???????? 0??立即返回, 非阻塞
???????? >0??指定毫秒
??????返回值: 成功返回有多少文件描述符就緒, 時(shí)間到時(shí)返回0, 出錯(cuò)返回-1 errno;
??3) 邊沿觸發(fā)(epoll ET)&水平觸發(fā)(epoll LT)
????邊沿觸發(fā): 高電頻->低電頻 || 低點(diǎn)頻->高電頻 (0->1 || 1->0)
????水平觸發(fā): 高電頻->高電頻 || 低電頻->低電頻 (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標(biāo)志
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;
}
邊沿觸發(fā)&非阻塞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; // 邊沿觸發(fā)
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設(shè)置
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) 反應(yīng)堆模型
????
/*
*epoll基于非阻塞I/O事件驅(qū)動(dòng)
*/
#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 //監(jiān)聽上限數(shù)
#define BUFLEN 4096
#define SERV_PORT 8080
void recvdata(int fd, int events, void *arg);
void senddata(int fd, int events, void *arg);
/* 描述就緒文件描述符相關(guān)信息 */
struct myevent_s {
int fd; //要監(jiān)聽的文件描述符
int events; //對(duì)應(yīng)的監(jiān)聽事件
void *arg; //泛型參數(shù)
void (*call_back)(int fd, int events, void *arg); //回調(diào)函數(shù)
int status; //是否在監(jiān)聽:1->在紅黑樹上(監(jiān)聽), 0->不在(不監(jiān)聽)
char buf[BUFLEN];
int len;
long last_active; //記錄每次加入紅黑樹 g_efd 的時(shí)間值
};
int g_efd; //全局變量, 保存epoll_create返回的文件描述符
struct myevent_s g_events[MAX_EVENTS+1]; //自定義結(jié)構(gòu)體類型數(shù)組. +1-->listen fd
/*將結(jié)構(gòu)體 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); //調(diào)用eventset函數(shù)的時(shí)間
return;
}
/* 向 epoll監(jiān)聽的紅黑樹 添加一個(gè) 文件描述符 */
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) { //已經(jīng)在紅黑樹 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) //實(shí)際添加/修改
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 監(jiān)聽的 紅黑樹中刪除一個(gè) 文件描述符*/
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; //修改狀態(tài)
epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv); //從紅黑樹 efd 上將 ev->fd 摘除
return ;
}
/* 當(dāng)有文件描述符就緒, epoll返回, 調(diào)用該函數(shù) 與客戶端建立鏈接 */
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) {
/* 暫時(shí)不做出錯(cuò)處理 */
}
printf("%s: accept, %s\n", __func__, strerror(errno));
return ;
}
do {
for (i = 0; i < MAX_EVENTS; i++) //從全局?jǐn)?shù)組g_events中找一個(gè)空閑元素
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) 不執(zhí)行后續(xù)代碼
}
int flag = 0;
if ((flag = fcntl(cfd, F_SETFL, O_NONBLOCK)) < 0) { //將cfd也設(shè)置為非阻塞
printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
break;
}
/* 給cfd設(shè)置一個(gè) myevent_s 結(jié)構(gòu)體, 回調(diào)函數(shù) 設(shè)置為 recvdata */
eventset(&g_events[i], cfd, recvdata, &g_events[i]);
eventadd(g_efd, EPOLLIN, &g_events[i]); //將cfd添加到紅黑樹g_efd中,監(jiān)聽讀事件
} 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); //讀文件描述符, 數(shù)據(jù)存入myevent_s成員buf中
eventdel(g_efd, ev); //將該節(jié)點(diǎn)從紅黑樹上摘除
if (len > 0) {
ev->len = len;
ev->buf[len] = '\0'; //手動(dòng)添加字符串結(jié)束標(biāo)記
printf("C[%d]:%s\n", fd, ev->buf);
eventset(ev, fd, senddata, ev); //設(shè)置該 fd 對(duì)應(yīng)的回調(diào)函數(shù)為 senddata
eventadd(g_efd, EPOLLOUT, ev); //將fd加入紅黑樹g_efd中,監(jiān)聽其寫事件
} 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); //直接將數(shù)據(jù) 回寫給客戶端。未作處理
/*
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的 回調(diào)函數(shù)改為 recvdata
eventadd(g_efd, EPOLLIN, ev); //從新添加到紅黑樹上， 設(shè)為監(jiān)聽讀事件
} else {
close(ev->fd); //關(guān)閉鏈接
eventdel(g_efd, ev); //從紅黑樹g_efd中移除
printf("send[fd=%d] error %s\n", fd, strerror(errno));
}
return ;
}
/*創(chuàng)建 socket, 初始化lfd */
void initlistensocket(int efd, short port)
{
int lfd = socket(AF_INET, SOCK_STREAM, 0);
fcntl(lfd, F_SETFL, O_NONBLOCK); //將socket設(shè)為非阻塞
/* 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]); //使用用戶指定端口.如未指定,用默認(rèn)端口
g_efd = epoll_create(MAX_EVENTS+1); //創(chuàng)建紅黑樹,返回給全局 g_efd
if (g_efd <= 0)
printf("create efd in %s err %s\n", __func__, strerror(errno));
initlistensocket(g_efd, port); //初始化監(jiān)聽socket
struct epoll_event events[MAX_EVENTS+1]; //保存已經(jīng)滿足就緒事件的文件描述符數(shù)組
printf("server running:port[%d]\n", port);
int checkpos = 0, i;
while (1) {
/* 超時(shí)驗(yàn)證醇疼，每次測(cè)試100個(gè)鏈接，不測(cè)試listenfd 當(dāng)客戶端60秒內(nèi)沒有和服務(wù)器通信，則關(guān)閉此客戶端鏈接 */
long now = time(NULL); //當(dāng)前時(shí)間
for (i = 0; i < 100; i++, checkpos++) { //一次循環(huán)檢測(cè)100個(gè)。 使用checkpos控制檢測(cè)對(duì)象
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); //關(guān)閉與該客戶端鏈接
printf("[fd=%d] timeout\n", g_events[checkpos].fd);
eventdel(g_efd, &g_events[checkpos]); //將該客戶端 從紅黑樹 g_efd移除
}
}
/*監(jiān)聽紅黑樹g_efd, 將滿足的事件的文件描述符加至events數(shù)組中, 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++) {
/*使用自定義結(jié)構(gòu)體myevent_s類型指針, 接收 聯(lián)合體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;
}
