#include <sys/types.h>#include <sys/socket.h>#include <sys/epoll.h>#include <stdio.h>#include <stdlib.h>#include <unistd.h>#include <fcntl.h>#include <arpa/inet.h>#include <string.h>
#define BUFFER_LENGTH 1024#define MAX_EPOLL_EVENTS 1024#define SERVER_PORT 9105#define PORT_COUNT 100
typedef int (*NCALLBACK)(int fd, void *arg);
typedef struct _nevent {
int fd; int events; void *arg; NCALLBACK callback;
int status; // whether fd is in epoll now. char buffer[BUFFER_LENGTH]; int length;
} nevent;
typedef struct _eventblock {
struct _eventblock *next; nevent *events; // 每一个block 1024个event
} eventblock;
typedef struct _nreactor {
int epfd; int blkcnt; eventblock *evblk;
} nreactor;
int recv_cb(int client_fd, void *arg);int send_cb(int client_fd, void *arg);int accept_cb(int listen_fd, void *arg);nevent *nreactor_idx(nreactor *reactor, int sockfd);
int init_sock(unsigned short port) {
int listen_fd = socket(AF_INET, SOCK_STREAM, 0); fcntl(listen_fd, F_SETFL, O_NONBLOCK);
struct sockaddr_in server_addr; memset(&server_addr, 0, sizeof(server_addr)); server_addr.sin_family = AF_INET; server_addr.sin_addr.s_addr = htonl(INADDR_ANY); server_addr.sin_port = htons(port);
bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (listen(listen_fd, 20) < 0) { perror("listen"); }
return listen_fd;}
int nreactor_alloc(nreactor *reactor) {
if (reactor == NULL) return -1; if (reactor->evblk == NULL) return -1;
eventblock *blk = reactor->evblk;
while (blk->next != NULL) {
blk = blk->next; }
nevent *evs = (nevent *)malloc(MAX_EPOLL_EVENTS * sizeof(nevent)); if (evs == NULL) { perror("nreactor_alloc malloc events failed"); return -2; } memset(evs, 0, MAX_EPOLL_EVENTS * sizeof(nevent));
eventblock *block = (eventblock *)malloc(sizeof(eventblock));
if (block == NULL) { perror("nreactor_alloc malloc block failed"); return -2; } memset(block, 0, sizeof(eventblock));
block->events = evs; block->next = NULL;
blk->next = block; reactor->blkcnt++;
return 0;}
int nreactor_init(nreactor *reactor) {
if (reactor == NULL) return -1; memset(reactor, 0, sizeof(nreactor));
reactor->epfd = epoll_create(1); if (reactor->epfd < 0) { perror("epoll_create"); return -2; }
nevent *evs = (nevent *)malloc(MAX_EPOLL_EVENTS * sizeof(nevent)); if (evs == NULL) { perror("nreactor_init malloc events failed"); return -2; } memset(evs, 0, MAX_EPOLL_EVENTS * sizeof(nevent));
eventblock *block = (eventblock *)malloc(sizeof(eventblock));
if (block == NULL) { perror("nreactor_init malloc block failed"); return -2; } memset(block, 0, sizeof(eventblock));
block->events = evs; block->next = NULL;
reactor->evblk = block; reactor->blkcnt = 1;
return 0;
}
int nreactor_destroy(nreactor *reactor) {
close(reactor->epfd); eventblock *blk = reactor->evblk; eventblock *blk_next = NULL;
while (blk != NULL) {
blk_next = blk->next;
free(blk->events); free(blk); blk = blk_next; }
return 0;}
void nreactor_event_set(nevent *ev, int fd, NCALLBACK callback, void *arg) {
ev->fd = fd; ev->callback = callback; ev->arg = arg; ev->events = 0;
}
int nreactor_event_add(int epfd, nevent *ev, int events) {
struct epoll_event ep_ev = {0, {0}}; ep_ev.events = ev->events = events; ep_ev.data.ptr = ev;
int op; if (ev->status == 1) { op = EPOLL_CTL_MOD; } else { ev->status = 1; op = EPOLL_CTL_ADD; }
if (epoll_ctl(epfd, op, ev->fd, &ep_ev) < 0) { perror("epoll_ctl"); return -1; }
return 0;
}
int nreactor_event_del(int epfd, nevent *ev) {
struct epoll_event ep_ev = {0, {0}}; if (ev->status != 1) { return -1; }
ev->status = 0;
epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, NULL);
return 0;
}
int recv_cb(int client_fd, void *arg) {
nreactor *reactor = (nreactor *)arg; if (reactor == NULL) return -1;
nevent *ev = nreactor_idx(reactor, client_fd);
int len = recv(client_fd, ev->buffer, BUFFER_LENGTH, 0); nreactor_event_del(reactor->epfd, ev);
if (len > 0) {
ev->length = len; ev->buffer[len] = '\0';
printf("client_fd[%d]:%s\n", client_fd, ev->buffer);
nreactor_event_set(ev, client_fd, send_cb, reactor); nreactor_event_add(reactor->epfd, ev, EPOLLOUT);
} else if (len == 0) {
close(client_fd); // printf("[client_fd=%d] pos[%ld], closed\n", client_fd, ev - reactor->events);
} else {
close(client_fd); perror("recv");
}
return len;
}
int send_cb(int client_fd, void *arg) {
nreactor *reactor = (nreactor *)arg; if (reactor == NULL) return -1;
nevent *ev = nreactor_idx(reactor, client_fd);
int len = send(client_fd, ev->buffer, ev->length, 0); nreactor_event_del(reactor->epfd, ev); if (len > 0) {
printf("send[client_fd=%d], [%d]%s\n", client_fd, len, ev->buffer);
nreactor_event_set(ev, client_fd, recv_cb, reactor); nreactor_event_add(reactor->epfd, ev, EPOLLIN);
} else {
close(ev->fd); perror("send"); }
return len;
}
int accept_cb(int listen_fd, void *arg) {
nreactor *reactor = (nreactor *)arg; if (reactor == NULL) return -1;
struct sockaddr_in client_addr; socklen_t len = sizeof(client_addr);
int client_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &len); if (client_fd < 0) { perror("accept"); return -1; }
int flag = fcntl(client_fd, F_SETFL, O_NONBLOCK); if (flag < 0) { perror("fcntl"); return -2; }
nevent *ev = nreactor_idx(reactor, client_fd); nreactor_event_set(ev, client_fd, recv_cb, reactor); nreactor_event_add(reactor->epfd, ev, EPOLLIN);
printf("new connect [%s:%d], client_fd[%d]\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), client_fd);}
nevent *nreactor_idx(nreactor *reactor, int sockfd) {
int blkidx = sockfd / MAX_EPOLL_EVENTS;
while (blkidx >= reactor->blkcnt) { nreactor_alloc(reactor); } int i = 0; eventblock *blk = reactor->evblk; while (i++ < blkidx && blk != NULL) { blk = blk->next; } return &blk->events[sockfd % MAX_EPOLL_EVENTS];
}
int nreactor_addlistener(nreactor *reactor, int listen_fd, NCALLBACK accept_cb) {
if (reactor == NULL || reactor->evblk == NULL) { return -1; }
nevent *event = nreactor_idx(reactor, listen_fd);
nreactor_event_set(event, listen_fd, accept_cb, reactor); nreactor_event_add(reactor->epfd, event, EPOLLIN);
return 0;
}
int nreactor_run(nreactor *reactor) {
if (reactor == NULL) return -1; if (reactor->evblk == NULL) return -1;
struct epoll_event events[MAX_EPOLL_EVENTS];
while (1) {
int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000); if (nready < 0) { perror("epoll_wait"); continue; }
int i; for (i = 0; i < nready; i++) {
nevent *ev = (nevent *)events[i].data.ptr;
if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) { ev->callback(ev->fd, ev->arg); } if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) { ev->callback(ev->fd, ev->arg); }
}
}
}
#if 1
int main(int argc, char *argv[]) {
unsigned short port = SERVER_PORT; if (argc == 2) { port = atoi(argv[1]); }
nreactor *reactor = (nreactor *)malloc(sizeof(nreactor)); nreactor_init(reactor);
int i = 0; int listen_fds[PORT_COUNT] = {0}; for (i = 0; i < PORT_COUNT; i++) { listen_fds[i] = init_sock(port + i); nreactor_addlistener(reactor, listen_fds[i], accept_cb); }
nreactor_run(reactor);
nreactor_destroy(reactor);
for (i = 0; i < PORT_COUNT; i++) { close(listen_fds[i]); }
free(reactor);
return 0;}
#endif
评论