webserver.cpp

#include "webserver.h"

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

  // root文件夹路径
  char server_path[200];
  getcwd(server_path, 200);
  char root[6] = "/root";
  m_root = (char *)malloc(strlen(server_path) + strlen(root) + 1);
  strcpy(m_root, server_path);
  strcat(m_root, root);

  // 定时器
  users_timer = new client_data[MAX_FD];
}

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) //  参数1的话就是异步写入日志，最后的800代表异步写入的阻塞队列的大小为800
      Log::get_instance()->init("./ServerLog", m_close_log, 2000, 800000, 800);
    else
      Log::get_instance()->init("./ServerLog", m_close_log, 2000, 800000, 0);
  }
}

void WebServer::sql_pool() {
  // 初始化数据库连接池
  m_connPool = connection_pool::GetInstance();
  m_connPool->init("localhost", m_user, m_passWord, m_databaseName, 3306,
                   m_sql_num, m_close_log);

  // 初始化数据库读取表
  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() {
  // 网络编程基础步骤
  //  m_listenfd 存储了监听套接字的文件描述符
  m_listenfd = socket(PF_INET, SOCK_STREAM, 0);
  // assert断言来检查套接字创建是否成功，若创建失败，则程序终止
  assert(m_listenfd >= 0);

  // 优雅关闭连接
  if (0 == m_OPT_LINGER) {
    // 不启用优雅关闭连接，设置so_linger套接字选项，立即关闭连接
    struct linger tmp = {0, 1};
    setsockopt(m_listenfd, SOL_SOCKET, SO_LINGER, &tmp, sizeof(tmp));
  } else if (1 == m_OPT_LINGER) {
    // 启用优雅关闭连接，设置so_linger套接字选项，等待未发送的数据发送完毕后再关闭连接
    struct linger tmp = {1, 1};
    setsockopt(m_listenfd, SOL_SOCKET, SO_LINGER, &tmp, sizeof(tmp));
  }

  int ret = 0;
  // 创建并配置服务器地址结构体address
  struct sockaddr_in address;
  bzero(&address, sizeof(address));
  address.sin_family = AF_INET; // 设置地址族为AF_INET，表示使用IPv4地址
  address.sin_addr.s_addr =
      htonl(INADDR_ANY); // 设置IP地址为INADDR_ANY，表示绑定到所有可用的网络接口
  address.sin_port = htons(m_port);

  int flag = 1;
  // 启用SO_REUSEADDR套接字选项，以便在服务器重启时能够重新使用绑定的端口
  setsockopt(m_listenfd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));
  // 绑定套接字到指定的地址结构体，这将服务器的监听套接字与指定的端口绑定在一起
  ret = bind(m_listenfd, (struct sockaddr *)&address, sizeof(address));
  assert(ret >= 0);
  // 调用listen函数开始监听连接请求，设置最大等待连接队列的长度为5
  ret = listen(m_listenfd, 5);
  assert(ret >= 0);

  // 初始化实例utils
  utils.init(TIMESLOT);

  // epoll创建内核事件表
  //  声明了一个存储 epoll 事件的数组 events，数组的大小是 MAX_EVENT_NUMBER
  epoll_event events[MAX_EVENT_NUMBER];
  // 创建了一个 epoll 内核事件表，并将其文件描述符存储在 m_epollfd 成员变量中
  m_epollfd = epoll_create(5);
  assert(m_epollfd != -1);

  // 调用utils对象的addfd函数，将监听套接字m_listenfd添加到epoll事件表中，指定事件监听方式为
  // m_LISTENTrigmode
  utils.addfd(m_epollfd, m_listenfd, false, m_LISTENTrigmode);
  // 设置http_conn类的成员变量
  http_conn::m_epollfd = m_epollfd;

  // 创建了一个UNIX域的全双工套接字对，其中m_pipefd数组中存储了两个套接字描述符，用于进程间通信。
  ret = socketpair(PF_UNIX, SOCK_STREAM, 0, m_pipefd);
  assert(ret != -1);
  // 将m_pipefd[1]设置为非阻塞模式，以提高管道写入的效率。
  utils.setnonblocking(m_pipefd[1]);
  // 将m_pipefd[0]添加到epoll事件表中，用于监听管道读事件。
  utils.addfd(m_epollfd, m_pipefd[0], false, 0);

  // 设置了忽略SIGPIPE信号，这是为了防止在向已关闭的套接字写入数据时触发SIGPIPE信号导致程序退出
  utils.addsig(SIGPIPE, SIG_IGN);
  // 设置了SIGALRM信号的处理函数为utils::sig_handler，并指定false参数表示不重新启用信号处理。
  utils.addsig(SIGALRM, utils.sig_handler, false);
  // 设置了SIGALRM信号的处理函数为utils::sig_handler，并指定false参数表示不重新启用信号处理。
  // utils.addsig(SIGTERM, utils.sig_handler, false);

  // 设置一个定时器，每隔TIMESLOT秒触发一次SIGALRM信号，这将用于定时任务的处理。
  alarm(TIMESLOT);

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

// 传递了一个连接fd 和一个连接客户地址
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::dealclientdata() {
  struct sockaddr_in client_address;
  socklen_t client_addrlength = sizeof(client_address);
  if (0 == m_LISTENTrigmode) {
    //  接收客户端连接，返回一个新的套接字文件描述符
    int connfd = accept(m_listenfd, (struct sockaddr *)&client_address,
                        &client_addrlength);
    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 {
    while (1) {
      int connfd = accept(m_listenfd, (struct sockaddr *)&client_address,
                          &client_addrlength);
      if (connfd < 0) {
        LOG_ERROR("%s:errno is:%d", "accept error", errno);
        break;
      }
      if (http_conn::m_user_count >= MAX_FD) {
        utils.show_error(connfd, "Internal server busy");
        LOG_ERROR("%s", "Internal server busy");
        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));

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

      if (timer) {
        adjust_timer(timer);
      }
    } else {
      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);
    }
  }
}

// 该事件循环不断地等待各种事件的发生,然后根据事件类型进行相应的处理,包括处理新的客户端连接、处理信号、处理读写事件等
// 在定时器超时时，也会触发定时器事件的处理。循环会一直运行，直到stop_server变为true,通常由外部机制出发的,表示服务器要
// 停止运行
void WebServer::eventLoop() {
  // 是否发生了定时器超时
  bool timeout = false;
  // 是否要停止服务器
  bool stop_server = false;

  while (!stop_server) {
    // epoll_wait函数等待事件的发生,m_epollfd是一个epoll文件描述符，events是用于存储发生事件的数组，MAX_EVENT_NUMBER是最大
    // 事件数量，-1表示一直等待直到有事件发生。number保存了发生的事件数量
    int number = epoll_wait(m_epollfd, events, MAX_EVENT_NUMBER, -1);

    // 如果返回值小于0且错误码不是EINTR（表示被信号中断），则打印错误信息，并退出循环
    if (number < 0 && errno != EINTR) {
      LOG_ERROR("%s", "epoll failure");
      break;
    }

    // 遍历发生的事件
    for (int i = 0; i < number; i++) {
      // 获取事件的文件描述符
      int sockfd = events[i].data.fd;

      // 处理新到的客户连接
      //  如果事件是由服务器监听套接字m_listenfd触发的，表示有新的客户端连接请求，进入处理新客户端连接的分支
      if (sockfd == m_listenfd) {
        // 这个函数的目的是处理新的客户端连接数据，并返回一个布尔值flag，表示处理结果。
        bool flag = dealclientdata();
        // 如果处理结果flag为false，则跳过当前循环迭代，继续处理下一个事件。
        if (false == flag)
          continue;
      }
      // 套接字关闭或发生错误的情况，移除对应的定时器
      else if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {
        // 服务器端关闭连接，移除对应的定时器
        util_timer *timer = users_timer[sockfd].timer;
        deal_timer(timer, sockfd);
      }

      // 处理从管道读取的信号事件，用于处理信号
      else if ((sockfd == m_pipefd[0]) && (events[i].events & EPOLLIN)) {
        bool flag = dealwithsignal(timeout, stop_server);
        // 处理信号事件时出现了错误，通过日志记录错误信息。
        if (false == flag)
          LOG_ERROR("%s", "dealclientdata failure");
      }
      // 处理客户连接上接收到的数据
      else if (events[i].events & EPOLLIN) {
        dealwithread(sockfd);
      } else if (events[i].events & EPOLLOUT) {
        dealwithwrite(sockfd);
      }
    }
    // timeout为true，表示发生了定时器超时，会调用utils.timer_handler()处理定时器事件，并打印相关信息
    if (timeout) {
      utils.timer_handler();

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

      timeout = false;
    }
  }
}