reactor模式

reactor模式是对epoll的一层封装，将网络io转换成event。redis/nginx/libevent对网络io的处理，都采用了reactor模式。

参考下面的代码, 看完这段代码就会明白什么是reactor

#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <fcntl.h>

#define BUFFER_LENGTH 4096
#define MAX_EPOLL_EVENTS 1024*512
#define LISTEN_PORT_COUNT   100

typedef int NCALLBACK(int, int, void*);

typedef struct ntyevent {
    int fd;
    int events;
    void *arg;
    int (*callback)(int fd, int events, void *arg);

    int status;
    char buffer[BUFFER_LENGTH];
    int length;
} ntyevent;

typedef struct ntyreactor {
    int epfd;
    ntyevent *events;
} ntyreactor;

int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);

void nty_event_set(struct ntyevent *ev, int fd, NCALLBACK callback, void *arg) {
    ev->fd = fd;
    ev->events = 0;
    ev->arg = arg;
    ev->callback = callback;
}

void nty_event_add(int epfd, int events, ntyevent *ev) {
    struct epoll_event ep_ev = {0};
    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events = events;

    int op;
    if (ev->status == 1) {
        op = EPOLL_CTL_MOD;
    } else {
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }

    if (epoll_ctl(epfd, op, ev->fd, &ep_ev)) {
        printf("fail to add fd[%d], events[%d] into epoll\n", ev->fd, events);
    }
}

void nty_event_del(int epfd, ntyevent *ev) {
    struct epoll_event ep_ev = {0};
    
    if (ev->status != 1) {
        return;
    }

    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events;
    ev->status = 0;
    epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);
}

int recv_cb(int fd, int events, void *arg) {
    ntyreactor *reactor = (ntyreactor*)arg;
    ntyevent *ev = reactor->events + fd;

    int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0);
    nty_event_del(reactor->epfd, ev);

    if (len > 0) {
        ev->length = len;
        ev->buffer[len] = '\0';

        printf("Recv from [%d]: %s, %d Bytes\n", fd, ev->buffer, len);
        nty_event_set(ev, fd, send_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLOUT, ev);
    } else if (len == 0) {
        close(fd);
        printf("[fd=%d] pos[%ld], closed\n", fd, ev-reactor->events);
    } else {
        close(fd);
        perror("recv");
    }
    return len;
}

int send_cb(int fd, int events, void *arg) {
    ntyreactor *reactor = (ntyreactor*)arg;
    ntyevent *ev = reactor->events + fd;
    int len = send(fd, ev->buffer, ev->length, 0);

    nty_event_del(reactor->epfd, ev);

    if (len > 0) {
        printf("send[fd=%d], [%d]%s\n", fd, len, ev->buffer);
        memset(ev->buffer, 0, BUFFER_LENGTH);

        nty_event_set(ev, fd, recv_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLIN, ev);
    } else {
        close(fd);
        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }

    return len;
}

int accept_cb(int fd, int events, void *arg) {
    ntyreactor *reactor = (ntyreactor*)arg;
    if (reactor == NULL) {
        return -1;
    }

    struct sockaddr_in client_addr;
    socklen_t len = sizeof(client_addr);

    int client_fd = accept(fd, (struct sockaddr*)&client_addr, &len);
    if (client_fd < 0) {
        perror("accept");
        return -1;
    }

    if (client_fd == MAX_EPOLL_EVENTS) {
        printf("%s: max connect limit[%d]\n", __func__, MAX_EPOLL_EVENTS);
        return -1;
    }
    fcntl(client_fd, F_SETFL, O_NONBLOCK);
    nty_event_set(&reactor->events[client_fd], client_fd, recv_cb, reactor);
    nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[client_fd]);

    
    return 0;
}

int init_listen_fd(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, 100) < 0) {
        printf("listen failed : %s\n", strerror(errno));
    }
    return listen_fd;
}

int ntyreactor_init(ntyreactor *reactor) {
    if (reactor == NULL) {
        return -1;
    }
    memset(reactor, 0, sizeof(ntyreactor));

    reactor->epfd = epoll_create(1);
    if (reactor->epfd < 0) {
        perror("epoll_create");
        return -2;
    }

    reactor->events = (ntyevent*)malloc(MAX_EPOLL_EVENTS * sizeof(ntyevent));
    if (reactor->events == NULL) {
        printf("fail to malloc reactor events\n");
        close(reactor->epfd);
        return -3;
    }
    return 0;
}

void ntyreactor_destroy(ntyreactor *reactor) {
    close(reactor->epfd);
    free(reactor->events);
    free(reactor);
}

int ntyreactor_addlistener(ntyreactor *reactor, int listen_fd, NCALLBACK *acceptor) {
    if (reactor == NULL || reactor->events == NULL) {
        return -1;
    }

    nty_event_set(&reactor->events[listen_fd], listen_fd, accept_cb, reactor);
    nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[listen_fd]);

    return 0;
}

int ntyreactor_run(ntyreactor *reactor) {
    if (reactor == NULL || reactor->epfd < 0 || reactor->events == 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++) {
            ntyevent *ev = (ntyevent*)events[i].data.ptr;

            if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
            if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
        }
    }
}


int main(int argc, char* argv[]) {

    if (argc < 2) {
        printf("Parameter error\n");
        return -1;
    }
    int port = atoi(argv[1]);
    
    ntyreactor *reactor = (ntyreactor*)malloc(sizeof(ntyreactor));
    ntyreactor_init(reactor);
    
    int listen_fd[LISTEN_PORT_COUNT] = {0};
    int i = 0;
    for (int i = 0; i < LISTEN_PORT_COUNT; i++) {
        listen_fd[i] = init_listen_fd(port + i);
        ntyreactor_addlistener(reactor, listen_fd[i], accept_cb);
    }
    ntyreactor_run(reactor);

    ntyreactor_destroy(reactor);
    for (int i = 0; i < LISTEN_PORT_COUNT; i++) {
        close(listen_fd[i]);
    }


    return 0;
}