打开导航 打开导航 Open menu
量化交易系统开发（C++）/ 期权策略与风控 / 行业研究分析
量化交易系统開發（C++）/ 期权策略与风控 / 行业研究分析
C++ trading systems / options strategy & risk controls / equity research
adrian@adrianxv.cn
文章目录

交易系统：TCP server 类 交易系统：TCP server 类 交易系统：TCP server 类

之所以书中只定义了 TCPServer 而没有 TCPClient，是因为服务端需要处理监听、accept()、多个连接 socket、epoll 事件分发的逻辑；而客户端只需要管理一个连接。

理解 TCPServer 的行为最好从 epoll 唤醒 -> TCPServer::sendAndRecv() -> TCPSocket::sendAndRecv() 这条链条入手

TCPServer 的成员变量

struct TCPServer {
public:
    int epoll_fd_ = -1; // epoll 由 TCPServer 对象持有
    TCPSocket listener_socket_; // 监听 socket，只有一个

    epoll_event events_[1024];
    
    // 管理 socket 的容器，包含 3 个：所有 socket，以及 读/写 就绪的 socket
    std::unordered_map<int, std::unique_ptr<TCPSocket>> sockets_;
    std::vector<TCPSocket*> receive_sockets_;
    std::vector<TCPSocket*> send_sockets_;
    
    // 为每一个连接 socket 注册的回调函数
    std::function<void(TCPSocket* s, Nanos rx_time)> recv_callback_ = nullptr;

    // 这一轮 epoll 接收事件处理完毕后，调用的函数对象。
    std::function<void()> recv_finished_callback_ = nullptr;

    std::string time_str_;
    Logger& logger_; // 注意 logger 是用引用传入的，而不是指针。为什么？
}

epoll 要用 fd 来管理的原因是，在 Linux 中 epoll 本身也是一个内核对象，此处 epoll_fd_ 就是管理 epoll 实例的句柄，后续调用都需要它。这也符合 Linux 当中 “一切皆文件描述符” 的抽象。

receive_sockets_ 保存当前这一轮需要读取数据的客户端 socket 指针；

send_sockets_ 保存当前这一轮需要发送数据的客户端 socket 指针

有关回调函数：回调函数本身可以由很多事件触发，比如 epoll、定时器、信号、GUI 等等。但是在交易系统当中的回调函数，基本上都是 epoll + 函数指针。所谓注册回调函数，就是让这个函数指针指向某一个我们希望被事件触发的函数对象。

epoll_wait() 会把已经就绪的事件写入调用者提供的 events_ 数组当中。这些事件包括：EPOLLIN（表示可读）、EPOLLOUT（表示可写）、EPOLLERR（表示发生错误）、EPOLLHUB（表示连接挂起或关闭）

int n = epoll_wait(epoll_fd_, events_, max_events, 0);

这里的 n 存储 epoll_wait() 本次返回的就绪事件数量，如果 n == -1 表示调用失败。

如果当前没有任何事件就绪，epoll_wait(epoll_fd_, events_, max_events_, timeout) 会阻塞；反之，如果有至少一个事件就绪，一般会直接把这些事件写入 events_ 并立刻返回（实际上，timeout 参数可以指定最多阻塞的毫秒数。但是交易系统中设为 0）。

TCPServer::listen()

auto TCPServer::listen() {
    // 先创建 epoll
    epoll_fd_ = epoll_create1(EPOLL_CLOEXEC); // 为什么这里有个 1？这是 Linux API 吗
    // 由于无法启动 epoll 属于致命错误，因此要抛出 std::system_error
    if (epoll_fd_ == -1) {
        throw std::system_error(errno, std::generic_categroy(), "epoll_create1");
    }

    // 如果 connect 失败。注意这里的 connect 更像创建并配置监听 socket
    if (listener_socket_.connect("", iface, port, true) < 0) {
        const int error = errno;
        close(epoll_fd_);
        epoll_fd_ = -1;

        throw std::system_error(
            error,
            std::generic_category(),
            "failed to create listener on " + iface + ":" + std::string(port)
        );
    }

    // 把监听 socket 加入 epoll。
    if (!addToEpollList(&listener_socket_)) {  // 如果注册失败
        // addToEpollList 内部要调用 epoll_ctl
        const int error = errno; // 先保存错误码
        close(epoll_fd_);        // 再关闭已经创建的 epoll fd
        epoll_fd_ = -1;          // 表示 epoll_fd_ 无效

        throw std::system_error(
            error, std::generic_category(), "epoll_ctl"
        ); // 这里的第二个参数 std::generic_category 告诉 C++ 这是 POSIX/通用系统错误类别；第三个参数 "epoll_ctl" 说明哪个操作失败了。
    }
}

之所以在 listen() 内部创建 epoll，是因为 listen() 是 TCPServer 的初始化入口。

在 Order Server 组件上，会由 OrderServer::start() 调用 TCPServer::listen()（随后会进入 OrderServer::run() 的工作循环）

TCPServer::addToEpollList()

TCPServer::addToEpollList(TCPSocket* socket) {
    epoll_event ev{EPOLLET | EPOLLIN, {reinterpret_cast<void *>(socket)}};
    return !epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket->socket_fd_, &env);
}

什么是 epoll_event？为什么需要存在？

这里什么是 epoll_ctl？我知道这是一个 Linux/Unix 系统调用。

TCPServer::poll()

  1. 调用 epoll_wait() 这个 Linux API
  2. 遍历返回的 epoll_event
  • 判断事件属于监听 socket 还是已连接 socket
  • 对已连接 socket：
    • 把可读 socket 加入 receive_sockets_（注意，这里只是加入队列，并没唤醒操作）
    • 把可写 socket 加入 send_sockets_
  • 对监听 socket
    • 调用 accept()，创建新的 TCPSocket 并注册到 epoll
  1. 处理错误和断开

它在外层 TCPServer::run() 的无限循环中被调用，每被调用依次就 epoll 一次。

auto TCPServer::poll() noexcept {
    // events_ 当中最多存放的元素个数
    const int max_events = 1 + send_sockets_.size() + receive_sockets_.size();

    const int n = epoll_wait(epoll_fd_, events_, max_events, 0);

    // 用来表示监听 socket 是否有新连接请求
    bool have_new_connection = false;

    for (int i = 0; i < n; i++) {// 遍历 events_ 处理所有 epoll 事件
        const auto& event = events_[i];
        // 这个 data.ptr 是什么？
        auto socket = reinterpret_cast<TCPSocket*>(event.data.ptr);

        // 可读，或有新连接可接受
        if (events.events & EPOLLIN) { // 这里是按位与
        // 表示检查 event.events 中代表 EPOLLIN 的那一位是否被设置。
        // 结果非零表示包含该事件
            // 是监听 socket，表明有新连接
            if (socket == &listener_socket_) {
                logger_.log("%:% %() % EPOLLIN listener_socket:%\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);

                // 对于连接 socket 的连接事件，只是把这个布尔值置为 true，然后最后集中处理连接。这里没有把任何 socket 加入队列
                have_new_connection = true;
                continue;
            }

            // 是连接 socket，表明已建立的连接可读
            else {
                logger_.log("%:% %() % EPOLLIN socket:%\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);

                if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end()) { 
                    // 在 receive_sockets_ 当中没找到对应的连接 socket，就把这个连接 socket 加进去。
                    receive_sockets_.push_back(socket);
                }
            }
        }

        // 有可写事件
        if (event.events & EPOLLOUT) {
            logger_.log("%:% %() % EPOLLOUT socket:%\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);

            if (std::find(send_sockets_.begin(), send_sockets_.end(), socket) == send_sockets_.end()) {
                send_sockets_.push_back(socket);
            }
        }

        // 处理 socket 错误或挂断事件：EPOLLERR 表 socket 错误；EPOLLHUP 表对端挂断
        if (event.events & (EPOLLERR | EPOLLHUP)) {  // HUP 是 hang up 的缩写
            logger_.log("%:% %() % EPOLLERR socket:%\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);

            if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end()) {
                receive_sockets_.push_back(socket);
                // 错误或对端挂断，还是要加入 receive_sockets_，处理可能的剩余数据
            }
        }
    }
            

    // 集中处理当前轮 poll() 需要处理的所有连接请求，直到所有请求都被处理，即 fd == -1
    while (have_new_connection) {
        logger_.log("%:% %() % have_new_connection\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
    
        sockaddr_storage addr; // 什么是 sockaddr_storage？

        socklen_t addr_len = sizeof(addr);

        // 这里调用 accept() 让监听 socket 接受连接请求，获得连接 socket 的 fd
        int fd = accept(listener_socket_.socket_fd, reinterpret_cast<sockaddr*>(&addr), &addr_len);

        // 如果 fd == -1 说明把所有连接请求都处理完毕了，此时退出 while 循环
        if (fd == -1) {
            break;
        }

        // 对新创建的连接 socket 进行设置
        // 等下，我们什么时候把这两个设置写成函数了？
        ASSERT(setNonBlocking(fd) && disableNagle(fd), "Failed to set non-blocking or no-delay on socket: " + std::to_string(fd)); 

        logger_.log("%:% %() % accepted socket:%\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_), fd)
    
        // 为新创建的连接 socket 创建应用层对象 TCPSocket，注册其回调
        auto socket = new TCPSocket(logger_);
        socket->socket_fd_ = fd;
        socket->recv_callback_ = recv_callback_;

        // 把新创建的连接 socket 加入该 TCPServer 的 epoll 队列
        ASSERT(addToEpollList(socket), "Unable to add socket. error:" + std::string(std::strerror(errno)));

        // 客户端可以在服务端调用 accept() 前就发送数据，数据会进入 socket 的内核缓冲区
        // 因此这里直接把新连接 socket 加入 receive_sockets_ 尝试读取，若没有数据也不影响。
        if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end()) {
            receive_sockets_.push_back(socket);
        }
    }
}

在 poll() 当中，只有连接 socket 会被放入 receive_sockets_ 或者 send_sockets_ 当中，而监听 socket 直接在 poll 内部处理掉了。后续 TCPServer::sendAndRecv() 遍历 receive_sockets_ 和 send_sockets_，进而调用 TCPSocket::sendAndRecv()，进而调用回调函数。因此监听 socket 是不会调用回调函数的。

TCPServer::sendAndRecv()

用于处理 “一轮 epoll” 唤醒的 socket 事件

遍历 receive_sockets_ 当中的所有 socket，进而调用 TCPSocket::sendAndRecv()。TCPSocket::sendAndRecv() 进而会调用该 TCPSocket 注册的回调函数来处理

遍历 receive_sockets_ 和 send_sockets_ 当中的所有 socket，进而调用 TCPSocket::sendAndRecv()。TCPSocket::sendAndRecv() 进而会对 调用该 TCPSocket 注册的回调函数。

auto TCPServer::sendAndRecv() {
    // 这个 recv 变量表示什么？
    auto recv = false;

    // 遍历 receive_sockets_ 当中的所有 socket，进而调用 TCPSocket::sendAndRecv()
    // 我们还未学 std::for_each() 的语法
    std::for_each(receive_sockets_.begin(), receive_sockets_.end(), [&recv](auto socket) {
        recv |= socket->sendAndRecv();
    });

    if (recv) { // 表示本轮 socket 接收的字节流解析结束
        recv_finished_callback_();
    }
    
    // 为什么这里的 lambda 写得和上面那个 lambda 也不同？
    std::for_each(send_sockets_.begin(), send_sockets_.end(), [](auto sockets) {
        socket->sendAndRecv();
    });
}

// TODO：发现一个问题：书中的 poll() 当中把 EPOLLIN 和 EPOLLOUT 事件分开，把可读/可写 socket 分别放入 receive_sockets_ 和 send_sockets_，然后在 TCPServer::sendAndRecv 当中再对每个 socket->sendAndRecv()，且 TCPSockets::sendAndRecv() 同时进行读和写。那么其实可以在 poll() 当中不区分 EPOLLIN 和 EPOLLOUT 事件，将这些 socket 放入一个队伍。然后在 TCPServer::sendAndRecv() 当中对每个 socket 一视同仁地 socket->sendAndRecv()。