webserver.cpp

#include "webserver.h"

WebServer::WebServer()
{
    //http_conn类对象
    users = new http_conn[MAX_FD]; //MAX_FD是65536

    /*
    //root文件夹路径
    char server_path[200];
    //getcwd(server_path, 200); //获取当前的目录
    server_path = 'home/ccpang/web/';
    printf("%s\n",server_path);
    char root[6] = "/root";
    m_root = (char *)malloc(strlen(server_path) + strlen(root) + 1);
    strcpy(m_root, server_path); //将server_path拷贝到m_root
    strcat(m_root, root); //连接m_root和root
    */

    char root[29] = "/home/ccpang/web/root/";
    m_root = (char*)malloc(strlen(root)+1);
    strcpy(m_root,root);
    //定时器
    users_timer = new client_data[MAX_FD];  //65536
}

WebServer::~WebServer()
{
    close(m_epollfd);
    close(m_listenfd);
    close(m_pipefd[1]);
    close(m_pipefd[0]);
    delete[] users;
    delete[] users_timer;
    delete m_pool;
}

void WebServer::init(int port, string user, string passWord, string databaseName, int log_write,
                     int opt_linger, int trigmode, int sql_num, int thread_num, int close_log, int actor_model)
{
    m_port = port;
    m_user = user;
    m_passWord = passWord;
    m_databaseName = databaseName;
    m_sql_num = sql_num;
    m_thread_num = thread_num;
    m_log_write = log_write;
    m_OPT_LINGER = opt_linger;
    m_TRIGMode = trigmode;
    m_close_log = close_log;
    m_actormodel = actor_model;
}


//配置监听套接字和已连接套接字的触发模式
void WebServer::trig_mode()
{
    //LT + LT
    if (0 == m_TRIGMode)
    {
        m_LISTENTrigmode = 0;
        m_CONNTrigmode = 0;
    }
    //LT + ET
    else if (1 == m_TRIGMode)
    {
        m_LISTENTrigmode = 0;
        m_CONNTrigmode = 1;
    }
    //ET + LT
    else if (2 == m_TRIGMode)
    {
        m_LISTENTrigmode = 1;
        m_CONNTrigmode = 0;
    }
    //ET + ET
    else if (3 == m_TRIGMode)
    {
        m_LISTENTrigmode = 1;
        m_CONNTrigmode = 1;
    }
}

void WebServer::log_write()
{
    if (0 == m_close_log)
    {
        //初始化日志
        if (1 == m_log_write)
            Log::get_instance()->init("./ServerLog", m_close_log, 2000, 80000, 800);  //异步队列中的值为800
        else{

              Log::get_instance()->init("./ServerLog", m_close_log, 2000, 80000, 0);  //同步队列中的值就为0
        }

    }

}

//数据库连接池初始化
void WebServer::sql_pool()
{
    //初始化数据库连接池
    cout << "初始化数据库连接池" <<endl;

    m_connPool = connection_pool::GetInstance();

    //这个地方创建了8个数据库连接
    m_connPool->init("122.112.162.222", m_user, m_passWord, m_databaseName, 3306, m_sql_num, m_close_log);

    cout << "数据库连接成功"<<endl;
    //初始化数据库读取表
    users->initmysql_result(m_connPool);


}

void WebServer::thread_pool()
{
    //线程池
    m_pool = new threadpool<http_conn>(m_actormodel, m_connPool, m_thread_num);
}

void WebServer::eventListen()
{
    int ret = 0;

    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    address.sin_port = htons(m_port);


    //网络编程基础步骤
    m_listenfd = socket(PF_INET, SOCK_STREAM, 0);
    if(m_listenfd < 0) {
        LOG_ERROR("Create socket error!", m_port);
    }

    //优雅关闭连接
    struct linger tmp;
    //如果选择此选项, close或 shutdown将等到所有套接字里排队的消息成功发送或到达延迟时间后>才会返回. 否则, 调用将立即返回。
    if (0 == m_OPT_LINGER)
         tmp = {0, 1};
    else if (1 == m_OPT_LINGER)
         tmp = {1, 1};
    ret = setsockopt(m_listenfd, SOL_SOCKET, SO_LINGER, &tmp, sizeof(tmp)); //第二个参数是通用套接字选项,第三个是延迟关闭连接
    if(ret < 0) {
        close(m_listenfd);
        LOG_ERROR("Init linger error!", m_port);
    }

    //允许重用本地地址和端口
    int flag = 1;
    ret = setsockopt(m_listenfd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));      //允许重用本地地址和端口
    if(ret == -1) {
        LOG_ERROR("set socket setsockopt error !");
        close(m_listenfd);
    }




    ret = bind(m_listenfd, (struct sockaddr *)&address, sizeof(address));
    //cout << "ret的值: " << ret<< endl;
    if(ret < 0) {
        LOG_ERROR("Bind Port:%d error!", m_port);
        close(m_listenfd);
    }
    ret = listen(m_listenfd, 5);
    if(ret < 0) {
        LOG_ERROR("Listen port:%d error!", m_port);
        close(m_listenfd);
    }

    cout << "监听端口:"<< m_port << endl;

    //工具类,信号和描述符基础操作
    Utils::u_pipefd = m_pipefd;
    Utils::u_epollfd = m_epollfd;

    utils.init(TIMESLOT);

    //epoll创建内核事件表
    epoll_event events[MAX_EVENT_NUMBER];
    m_epollfd = epoll_create(5);
    assert(m_epollfd != -1);

    //使用IO复用处理监听套接字
    utils.addfd(m_epollfd, m_listenfd, false, m_LISTENTrigmode);
    http_conn::m_epollfd = m_epollfd;

    //统一事件源:使用I/O复用监听管道的读端,
    ret = socketpair(PF_UNIX, SOCK_STREAM, 0, m_pipefd);
    assert(ret != -1);
    utils.setnonblocking(m_pipefd[1]);
    utils.addfd(m_epollfd, m_pipefd[0], false, 0);

    utils.addsig(SIGPIPE, SIG_IGN);
    utils.addsig(SIGALRM, utils.sig_handler, false);
    utils.addsig(SIGTERM, utils.sig_handler, false);


    LOG_INFO("Server port:%d", m_port);
    alarm(TIMESLOT);

    LOG_INFO("Listen Mode: %s, OpenConn Mode: %s",
            (m_LISTENTrigmode == 1 ? "ET": "LT"),
            (m_CONNTrigmode == 1 ? "ET": "LT"));
    LOG_INFO("Port:%d, OpenLinger: %s, IO Mode: %s",
            m_port, m_OPT_LINGER? "true":"false", m_actormodel?"Reactor":"Proctor");
    LOG_INFO("srcDir: %s", m_root);
    LOG_INFO("SqlConnPool num: %d, ThreadPool num: %d", m_sql_num, m_thread_num);
}



void WebServer::timer(int connfd, struct sockaddr_in client_address)
{
    users[connfd].init(connfd, client_address, m_root, m_CONNTrigmode, m_close_log, m_user, m_passWord, m_databaseName);

    //初始化client_data数据
    //创建定时器，设置回调函数和超时时间，绑定用户数据，将定时器添加到链表中
    users_timer[connfd].address = client_address;
    users_timer[connfd].sockfd = connfd;
    util_timer *timer = new util_timer;
    timer->user_data = &users_timer[connfd];
    timer->cb_func = cb_func;
    time_t cur = time(NULL);
    timer->expire = cur + 3 * TIMESLOT;
    users_timer[connfd].timer = timer;
    utils.m_timer_lst.add_timer(timer);
}

//若有数据传输，则将定时器往后延迟3个单位
//并对新的定时器在链表上的位置进行调整
void WebServer::adjust_timer(util_timer *timer)
{
    time_t cur = time(NULL);
    timer->expire = cur + 3 * TIMESLOT;
    utils.m_timer_lst.adjust_timer(timer);

    LOG_INFO("%s", "adjust timer once");
}

void WebServer::deal_timer(util_timer *timer, int sockfd)
{
    timer->cb_func(&users_timer[sockfd]);
    if (timer)
    {
        utils.m_timer_lst.del_timer(timer);
    }

    LOG_INFO("close fd %d", users_timer[sockfd].sockfd);
}

bool WebServer::dealclinetdata()
{
    struct sockaddr_in client_address;
    socklen_t client_addrlength = sizeof(client_address);
    if (0 == m_LISTENTrigmode)  //监听套接字使用LT模式
    {
        //
        //cout << "开始处理三次握手完成的连接" <<endl;
        int connfd = accept(m_listenfd, (struct sockaddr *)&client_address, &client_addrlength);
        //cout << "accept调用有返回" <<endl;
        if (connfd < 0)
        {
            LOG_ERROR("%s:errno is:%d", "accept error", errno);
            return false;
        }
        //大于连接
        if (http_conn::m_user_count >= MAX_FD)
        {
            utils.show_error(connfd, "Internal server busy");
            LOG_ERROR("%s", "Internal server busy");
            return false;
        }
        timer(connfd, client_address);
    }

    else
    {
        //如果accept调用有返回,除了建立当前这个连接外,不能马上就epoll_wait,还需要继续循环accept,直到返回-1
        while (1)
        {
            int connfd = accept(m_listenfd, (struct sockaddr *)&client_address, &client_addrlength);
            if (connfd < 0)
            {
                LOG_ERROR("Failed accept Client(%s:%d).", inet_ntoa(client_address.sin_addr), client_address.sin_port);
                break;
            }
            if (http_conn::m_user_count >= MAX_FD)
            {
                utils.show_error(connfd, "Internal server busy");
                LOG_ERROR("%s", "Clients is full");
                break;
            }
            timer(connfd, client_address);
        }
        return false;
    }
    return true;
}

bool WebServer::dealwithsignal(bool &timeout, bool &stop_server)
{
    int ret = 0;
    int sig;
    char signals[1024];
    //从管道中接收数据
    ret = recv(m_pipefd[0], signals, sizeof(signals), 0);
    if (ret == -1)
    {
        return false;
    }
    else if (ret == 0)
    {
        return false;
    }
    else
    {
        //一个一个字节接受
        for (int i = 0; i < ret; ++i)
        {
            switch (signals[i])
            {
            case SIGALRM:   //定时器终止时发送给进程的信号
            {
                timeout = true;
                break;
            }
            case SIGTERM: //终止进程,软件终止信号
            {
                stop_server = true;
                break;
            }
            }
        }
    }
    return true;
}

void WebServer::dealwithread(int sockfd)
{
    util_timer *timer = users_timer[sockfd].timer;

    //reactor
    if (1 == m_actormodel)
    {
        if (timer)
        {
            adjust_timer(timer);
        }

        //若监测到读事件，将该事件放入请求队列
        m_pool->append(users + sockfd, 0);

        while (true)
        {
            if (1 == users[sockfd].improv)
            {
                if (1 == users[sockfd].timer_flag)
                {
                    deal_timer(timer, sockfd);
                    users[sockfd].timer_flag = 0;
                }
                users[sockfd].improv = 0;
                break;
            }
        }
    }
    else
    {
        //proactor
        if (users[sockfd].read_once())
        {
            LOG_INFO("deal with the client(%s)", inet_ntoa(users[sockfd].get_address()->sin_addr));


            //若监测到读事件，将该事件放入请求队列
            //cout << "开一个工作线程去处理已连接套接字上的数据"<<endl;
            m_pool->append_p(users + sockfd);


            if (timer)
            {
                adjust_timer(timer); //有数据传输，则将定时器往后延迟3个单位
            }
        }
        else
        {
            //cout << "将已连接的套接字从epoll中删除"<<endl;
            deal_timer(timer, sockfd);
        }
    }
}

void WebServer::dealwithwrite(int sockfd)
{
    util_timer *timer = users_timer[sockfd].timer;
    //reactor
    if (1 == m_actormodel)
    {
        if (timer)
        {
            adjust_timer(timer);
        }

        m_pool->append(users + sockfd, 1);

        while (true)
        {
            if (1 == users[sockfd].improv)
            {
                if (1 == users[sockfd].timer_flag)
                {
                    deal_timer(timer, sockfd);
                    users[sockfd].timer_flag = 0;
                }
                users[sockfd].improv = 0;
                break;
            }
        }
    }
    else
    {
        //proactor
        if (users[sockfd].write())
        {
            LOG_INFO("send data to the client(%s)", inet_ntoa(users[sockfd].get_address()->sin_addr));

            if (timer)
            {
                adjust_timer(timer);
            }
        }
        else
        {
            deal_timer(timer, sockfd);
        }
    }
}


void WebServer::eventLoop()
{
    bool timeout = false;
    bool stop_server = false;
    if(m_close_log == 0) //不关闭日志
        LOG_INFO("========== Server start ==========");
    while (!stop_server)
    {
        int number = epoll_wait(m_epollfd, events, MAX_EVENT_NUMBER, -1);
        if (number < 0 && errno != EINTR)
        {
            LOG_ERROR("%s", "epoll failure");
            break;
        }

        for (int i = 0; i < number; i++)
        {
            int sockfd = events[i].data.fd;

            //处理新到的客户连接
            if (sockfd == m_listenfd)
            {
                //cout << "处理客户上的请求:" <<endl;
                bool flag = dealclinetdata(); //一定要把这个dealclinetdata改为deaclinerequest()函数
                if (false == flag)
                    continue;
            }
            //处理错误:EPOLLRDHUP:对端正常关闭,EPOLLHUP:文件描述符挂断,本端server出错 EPOLLERR:本端server出错,文件描述符发生错误
            else if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))
            {
                //服务器端关闭连接，移除对应的定时器
                //cout << "EPOLLRDHUP:对端正常关闭,EPOLLHUP:文件描述符挂断,本端server出错 EPOLLERR:本端server出错,文件描述符发生错误"<<endl;
                util_timer *timer = users_timer[sockfd].timer;
                deal_timer(timer, sockfd);
            }
            //处理信号:写管道上有数据来,有普通数据可读
            else if ((sockfd == m_pipefd[0]) && (events[i].events & EPOLLIN))
            {
                //cout << "信号中有数据可读" <<endl;
                bool flag = dealwithsignal(timeout, stop_server);
                if (false == flag)
                    LOG_ERROR("%s", "dealclientdata failure");
            }
            //处理客户连接上接收到的数据
            else if (events[i].events & EPOLLIN)
            {
                //cout << "读客户端的数据"<<endl;
                dealwithread(sockfd);
            }
            //向客户上写数据
            else if (events[i].events & EPOLLOUT)
            {
                //cout << "向客户端写数据"<<endl;
                dealwithwrite(sockfd);
            }
            else
                LOG_ERROR("Unexpected event");
        }

        if (timeout)
        {
            //cout << "定时器超时"<<endl;
            utils.timer_handler();

            LOG_INFO("%s", "timer tick");

            timeout = false;
        }
    }
}