epoll 回显服务器源码

/*************************************************************************
    > File Name: epoll_echosrv.c
    > Author: huabo
    > Mail: bohua1@126.com 
    > Created Time: Sun 05 Oct 2014 08:27:06 PM HKT
 ************************************************************************/

#include<stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>

#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/epoll.h>

#include <pthread.h>

#define MAXSIZE 5000
#define MAXLINE 10
#define OPEN_MAX 10
#define LISTENQ 20
#define INFTIM 1000

#define ERR_EXIT(m)        do        {            perror(m);            exit(EXIT_FAILURE);        }while(0)

//record the socket descriptor which need to read or write.
struct task
{
    int fd;
    struct task *next;
};


struct user_data
{
    int fd;
    unsigned int n_size;
    char line[MAXLINE];
};

void * readtask(void *args);
void * writetask(void *args);

struct epoll_event ev, events[20];
int epfd;

pthread_mutex_t mutex;
pthread_cond_t cond1;

struct task *readhead = NULL, *readtail = NULL, *writehead = NULL;

void setnonblocking(int sockfd)
{
    int opts;
    opts = fcntl(sockfd, F_GETFL);
    if(opts < 0)
    {
        ERR_EXIT("fcntl");
    }
    opts = (opts | O_NONBLOCK);
    opts = fcntl(sockfd, F_SETFL, opts);
    if(opts < 0)
    {
        ERR_EXIT("fcntl");
    }
}


int main()
{
    int listenfd;
    //read and write thread ID
    pthread_t tid1, tid2;
    struct task *new_task = NULL;
    struct user_data *rdata = NULL;

    pthread_mutex_init(&mutex, NULL);
    pthread_cond_init(&cond1, NULL);

    //initial id of read thread and write thread
    pthread_create(&tid1, NULL, readtask, NULL);
    pthread_create(&tid2, NULL, writetask, NULL);

    if( (listenfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0)
    {
        ERR_EXIT("socket");
    }
    setnonblocking(listenfd);

    struct sockaddr_in servaddr;
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(8189);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    //reuse address
    int on = 1;
    if(setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)))
    {
        ERR_EXIT("setsockopt");
    }

    //bind port
    if(bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
    {
        ERR_EXIT("bind");
    }

    //listen
    if(listen(listenfd, SOMAXCONN) < 0)
    {
        ERR_EXIT("listen");
    }

    struct sockaddr_in peeraddr;
    socklen_t peerlen = sizeof(peeraddr);
    int connfd;
    int sockfd;

    //epoll server
    epfd = epoll_create(256); //step 1: create descriptor for events
    struct epoll_event ev, events[5000];
    ev.data.fd = listenfd;
    ev.events = EPOLLIN|EPOLLET;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev); //step 2: add descriptor and events

    int nfds;
    int i;
    while(1)
    {
        nfds = epoll_wait(epfd, events, 20, -1);
        for(i = 0; i < nfds; ++i)
        {
            if(events[i].data.fd == listenfd)
            {
                printf("listen = %d\n", events[i].data.fd);
                connfd = accept(listenfd, (struct sockaddr *)(&peeraddr), &peerlen);
                if(connfd < 0)
                {
                    ERR_EXIT("accept");
                }
                //set non-block when use epoll
                setnonblocking(connfd);

                char *str = inet_ntoa(peeraddr.sin_addr);
                printf("connect from >> %s %d\n", str, connfd);

                //add new file descriptor
                ev.data.fd = connfd;
                ev.events = EPOLLIN|EPOLLET;
                epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);

                //reset the listen descriptor
                ev.data.fd = listenfd;
                ev.events = EPOLLIN|EPOLLET;

                epoll_ctl(epfd, EPOLL_CTL_MOD, listenfd, &ev);
            }
            else if(events[i].events & EPOLLIN) //there is data to read.
            {
                if((sockfd = events[i].data.fd) <= 0)
                {
                    printf("file descriptor error\n");
                    continue;
                }
                
                new_task = (struct task*)malloc(sizeof(struct task));
                new_task->fd = sockfd;
                new_task->next = NULL;

                //add to the read task queue.
                pthread_mutex_lock(&mutex);
                if(readhead == NULL)
                {
                    readhead = new_task;
                    readtail = new_task;
                }
                else
                {
                    readtail->next = new_task;
                    readtail = new_task;
                }
                
                //wake up the thread wait for cond1
                pthread_cond_broadcast(&cond1);
                pthread_mutex_unlock(&mutex);
            }
            else if(events[i].events & EPOLLOUT) //there is data to write 
            {
                //something needed to be done here.
                rdata = (struct user_data*)events[i].data.ptr;
                sockfd = rdata->fd;
                write(sockfd, rdata->line, rdata->n_size);
                free(rdata);

                ev.data.fd = sockfd;
                ev.events = EPOLLIN|EPOLLET;
                epoll_ctl(epfd, EPOLL_CTL_MOD, sockfd, &ev);
            }
            else if( (events[i].events & EPOLLHUP) || (events[i].events & EPOLLERR) || !(events[i].events & EPOLLIN))
            {
                //An error has occured on this fd, or the socket is not ready for reading(why were we notified then?)
                fprintf(stderr, "epoll error\n");
                close(events[i].data.fd);
                continue;
            }
        }
    }
    return 0;
}




static int count111 = 0;
static time_t oldtime = 0, nowtime = 0;


void* readtask(void *args)
{
    printf("read task begin\n");
    int fd = -1;
    unsigned int n;
    struct user_data *data = NULL;
    while(1)
    {
        pthread_mutex_lock(&mutex);
        while(readhead == NULL)
            pthread_cond_wait(&cond1, &mutex);

        fd = readhead->fd;

        struct task *tmp = readhead;
        readhead = readhead->next;
        free(tmp);

        pthread_mutex_unlock(&mutex);
        data = (struct user_data*)malloc(sizeof(struct user_data));
        data->fd = fd;

        char recvBuf[1024] = {0};
        int ret = 999;
        int rs = 1;

        while(rs)
        {
            ret = recv(fd, recvBuf, 1024, 0);
            if(ret < 0)
            {
                if(errno == EAGAIN)
                {
                    printf("EAGAIN\n");
                    break;
                }
                else
                {
                    printf("recv error!\n");
                    close(fd);
                    break;
                }
            }
            else if(ret == 0)
            {
                printf("client close.\n");
                rs = 0;
            }

            if(ret == sizeof(recvBuf))
            {
                //need to recv again.
                rs = 1;
                write(fd, recvBuf, ret);
                fputs(recvBuf, stdout);
            }
            else
            {
                write(fd, recvBuf, ret);
                fputs(recvBuf, stdout);
                rs = 0;
            }

        }

        if(ret > 0)
        {
            data->n_size = n;
            count111++;

            struct tm *today;
            time_t ltime;
            time(&nowtime);

            if(nowtime != oldtime)
            {
                printf("%d\n", count111);
                oldtime = nowtime;
                count111 = 0;
            }

            char buf[1000] = {0};
            sprintf(buf, "HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s", "Hello world!\n");
            send(fd, buf, strlen(buf), 0);
            close(fd);
        }
    }
}


void * writetask(void *args)
{

}