Reactor 反应堆模式：事件驱动的服务器设计范式

// thread_per_conn.c —— 传统"一连接一线程"，注意观察它的问题
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
 
// 每个连接一个线程的处理函数：死循环读、回声发回去
static void *handle_conn(void *arg)
{
    int cfd = *(int *)arg;          // 取出客户端 socket
    free(arg);                      // 释放拷贝过来的参数（指针指向堆上空间）
    char buf[1024];
    ssize_t n;
    while ((n = read(cfd, buf, sizeof(buf))) > 0)   // 阻塞读
    {
        write(cfd, buf, (size_t)n);                 // 原样写回
    }
    close(cfd);                     // 客户端断开，读返回 0，关闭连接
    return NULL;
}
 
int main(void)
{
    int lfd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
 
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(8888);
    addr.sin_addr.s_addr = INADDR_ANY;
    bind(lfd, (struct sockaddr *)&addr, sizeof(addr));
    listen(lfd, 128);
 
    printf("server listen on 8888 ...\n");
    while (1)
    {
        int cfd = accept(lfd, NULL, NULL);   // 阻塞等待新连接
        int *p = malloc(sizeof(int));        // 参数要拷贝到堆上，避免传栈上指针
        *p = cfd;
        pthread_t tid;
        pthread_create(&tid, NULL, handle_conn, p);  // 每个连接开一个线程
        pthread_detach(tid);                 // 分离线程，用完自动回收
    }
    close(lfd);
    return 0;
}

// event_loop_skeleton.c —— 事件循环的通用骨架（伪结构化示意）
while (1)
{
    // 1. 阻塞等待：把"关注的事件"交给内核，内核就绪了才返回
    int n = epoll_wait(epfd, events, MAX_EV, -1);
 
    // 2. 逐个分发：每一个就绪的 fd，找到它的"处理函数"去调用
    for (int i = 0; i < n; i++)
    {
        Handler *h = fd_to_handler(events[i].data.fd);  // 由 fd 找回处理者
        h->handle(events[i].events);                    // 把事件类型交给它
    }
    // 3. 一圈结束，继续回到等待——这就是"永动的循环"
}

// Connection.hpp —— 一个连接对象：socket + 缓冲区 + 事件 + 回调
#pragma once
#include <functional>     // std::function：用来装"处理函数"
#include <string>
#include <sys/socket.h>   // socket 相关
#include <netinet/in.h>   // sockaddr_in
#include <unistd.h>       // close
 
class Connection;         // 前置声明，下面回调签名要引用它
class TcpServer;          // 前置声明，Connection 需要回指服务器
 
// func_t：一个"接收 Connection* 并处理它"的函数类型
// std::function 让我们能把成员函数 / lambda / 绑定器都存成一个对象
using func_t = std::function<void(Connection *)>;
 
class Connection
{
public:
    // 构造：记住这个连接的 socket、关注的事件、以及它属于哪个服务器
    Connection(int sockfd, uint32_t events, TcpServer *R)
        : _sockfd(sockfd), _events(events), _R(R) {}
 
    // 注册三个回调：收到数据怎么处理（recver）、可写时怎么发（sender）、异常/断开怎么办（excepter）
    void RegisterCallback(func_t recver, func_t sender, func_t excepter)
    {
        _recver = recver;
        _sender = sender;
        _excepter = excepter;
    }
 
    // 把读到的字节追加进接收缓冲区。追加而非覆盖，是为了攒"半包"。
    void AddInBuffer(const std::string &buffer) { _inbuffer += buffer; }
 
    // 把要发的数据追加进发送缓冲区。这里是"攒着等 EPOLLOUT"，是 Reactor 写事件的核心。
    void AddOutBuffer(const std::string &buffer) { _outbuffer += buffer; }
 
    bool OutBufferEmpty() const { return _outbuffer.empty(); }  // 发送缓冲区是否空
 
    int SockFd() const { return _sockfd; }          // 取出这个连接的 socket
    uint32_t Events() const { return _events; }     // 取出关注的事件
    void SetEvents(uint32_t events) { _events = events; }  // 更新关注的事件
 
    std::string &InBuffer()  { return _inbuffer; }  // 取出接收缓冲区（引用，允许外部改）
    std::string &OutBuffer() { return _outbuffer; } // 取出发送缓冲区（引用）
    void Close() { ::close(_sockfd); }              // 关闭 socket
 
private:
    int _sockfd;              // 这个连接对应的 socket fd
    std::string _inbuffer;    // 接收缓冲区（先用 string 顶替，工程上会用更精细的 buffer）
    std::string _outbuffer;   // 发送缓冲区
    uint32_t _events;         // 这个连接当前关注的事件（EPOLLIN/EPOLLOUT/...）
    struct sockaddr_in _client;  // 对端 ip/port，便于日志
 
public:
    func_t _recver;           // 收数据时的回调
    func_t _sender;           // 可写时的回调
    func_t _excepter;         // 出错/断开时的回调
    TcpServer *_R;            // 回指服务器：方便调用服务器提供的"注册/移除事件"接口
};

// ConnectionFactory —— 工厂：屏蔽"怎么构造不同类型 Connection"的细节
class ConnectionFactory
{
public:
    // 造一个"监听连接"：只有 recver（连接到来时用），sender/excepter 暂时为空
    static Connection *BuildListenConnection(int listensock, func_t recver,
                                             uint32_t events, TcpServer *R)
    {
        Connection *conn = new Connection(listensock, events, R);
        conn->RegisterCallback(recver, nullptr, nullptr);
        return conn;
    }
 
    // 造一个"普通数据连接"：三个回调都有
    static Connection *BuildNormalConnection(int sockfd,
                                             func_t recver, func_t sender, func_t excepter,
                                             uint32_t events, TcpServer *R)
    {
        Connection *conn = new Connection(sockfd, events, R);
        conn->RegisterCallback(recver, sender, excepter);
        return conn;
    }
};

// TcpServer.hpp —— Reactor 的主控类
#pragma once
#include "Connection.hpp"
#include <sys/epoll.h>
#include <unordered_map>
 
class TcpServer
{
public:
    TcpServer();                // 建 socket、bind、listen、建 epoll，注册监听连接
    ~TcpServer();
 
    void InitServer();          // 初始化 + 启动事件循环
    void Loop();                // 事件循环本体
 
    // 对外接口：添加/移除一个连接；开启/关闭某 fd 对读写事件的关注
    void AddConnection(Connection *conn);              // 新的数据连接进来
    void RemoveConnection(int sockfd);                 // 从 epoll 与映射中移除
    void EnableReadWrite(int sockfd, bool rd, bool wr); // 调整某 fd 关注的事件位
 
private:
    int _listensock;              // 监听 socket
    int _epfd;                    // epoll 实例句柄
    std::unordered_map<int, Connection *> _conns;  // fd → 连接对象 的映射
    void SetNonBlock(int fd);     // 把 fd 设为非阻塞，Reactor 的读写必须在非阻塞下进行
    void BuildListen();           // 创建监听 socket 并注册到 epoll
};

void TcpServer::Loop()            // 事件循环：监听 → 分发 → 再监听
{
    const int MAX_EVENTS = 64;                    // 每次 epoll_wait 最多取回的就绪事件数
    struct epoll_event events[MAX_EVENTS];        // 就绪事件数组
    while (1)                                     // 永动循环
    {
        // 阻塞等待就绪事件。-1 表示永久等待，直到有事件返回
        int n = epoll_wait(_epfd, events, MAX_EVENTS, -1);
        if (n < 0)                                // 出错
        {
            if (errno == EINTR) break;            // 被信号打断：通常 continue，此处简化直接退出
            continue;
        }
 
        for (int i = 0; i < n; i++)               // 逐个分发就绪的 fd
        {
            int fd = events[i].data.fd;           // 取出就绪的 fd（我们注册时塞的是 data.fd）
            uint32_t ev = events[i].events;       // 取出就绪的事件位
 
            auto it = _conns.find(fd);            // 从映射反查这一个 fd 对应哪个连接
            if (it == _conns.end()) continue;     // 找不到（可能已被移除），跳过
 
            Connection *conn = it->second;        // 取到连接对象
 
            // ---- 分发阶段：根据事件位调用对应的回调 ----
            if (ev & EPOLLIN)                     // 可读：包含"新连接到来"和"数据到达"两种情况
            {
                if (conn->Events() & EPOLLIN)     // 看它注册时是不是带了可读关注
                    conn->_recver(conn);          // 调用它的"可读处理回调"（Accepter 或 Recver）
            }
            if (ev & EPOLLOUT)                    // 可写：一般意味着发送缓冲区可以被写入
            {
                if (conn->Events() & EPOLLOUT)
                    conn->_sender(conn);          // 调用"可写处理回调"（Sender）
            }
            // 出错或对端关闭：EPOLLERR 一定会带；EPOLLHUP 表示对方半关闭
            if ((ev & EPOLLERR) || (ev & EPOLLHUP))
            {
                if (conn->_excepter)              // 有异常回调才调用（监听连接没有 excepter）
                    conn->_excepter(conn);
                continue;                         // 异常后连接该清理或跳过后续处理
            }
        }
    }
}

// Accepter —— 连接管理器：处理"新连接到来"这一事件
class Accepter
{
public:
    Accepter() {}
 
    // conn 是监听 socket 对应的那个 Connection
    void AccepterConnection(Connection *conn)
    {
        while (1)                                  // 得一直 accept 到没有为止
        {
            struct sockaddr_in peer;               // 对端地址
            socklen_t len = sizeof(peer);
            int sockfd = ::accept(conn->SockFd(),
                                  (struct sockaddr *)&peer, &len);
            if (sockfd > 0)                        // 成功 accept 出一个新连接
            {
                SetNonBlock(sockfd);               // 立刻设为非阻塞，后面 epoll 才能照管它
 
                // 为这个普通连接准备三个回调（绑定到 HandlerConnection 的静态方法）
                auto recver   = std::bind(&HandlerConnection::Recver,   _1);
                auto sender   = std::bind(&HandlerConnection::Sender,   _1);
                auto excepter = std::bind(&HandlerConnection::Excepter, _1);
 
                // 用工厂造出一个普通连接对象：关注 可读 + ET（边缘触发）
                Connection *normal_conn =
                    ConnectionFactory::BuildNormalConnection(sockfd, recver, sender, excepter,
                                                             EPOLLIN | EPOLLET, conn->_R);
 
                conn->_R->AddConnection(normal_conn);  // 交给服务器：注册进 epoll 和映射表
            }
            else
            {
                // accept 失败：根据 errno 判断该怎么处理
                if (errno == EAGAIN) break;        // 没有更多连接了，正常退出循环
                else if (errno == EINTR) continue; // 被信号打断，重试
                else break;                        // 其它真正的错误，退出
            }
        }
    }
};

void TcpServer::AddConnection(Connection *conn)
{
    struct epoll_event ev;
    ev.events = conn->Events();          // 关注的事件（EPOLLIN | EPOLLET）
    ev.data.fd = conn->SockFd();         // 关键：把 fd 塞进 data.fd，epoll_wait 后靠它反查
    epoll_ctl(_epfd, EPOLL_CTL_ADD, conn->SockFd(), &ev);  // 加进监听
    _conns[conn->SockFd()] = conn;       // 维护映射：fd → 连接对象
}

// Recver —— 处理"普通连接可读"：把读到的字节追加进 inbuffer，再交给上层 HandlerRequest
static void Recver(Connection *conn)
{
    while (1)                                  // ET 模式：一次把能读的都读光
    {
        // 注意：是非阻塞读。读到 EAGAIN 说明暂时没数据了，退出循环。
        char buffer[1024];
        ssize_t n = recv(conn->SockFd(), buffer, sizeof(buffer) - 1, 0);
        if (n > 0)                             // 读到 n 个字节
        {
            buffer[n] = 0;                     // 转成字符串好追加（安全性看具体协议）
            conn->AddInBuffer(buffer);         // 追加进接收缓冲区，而不是覆盖！
        }
        else if (n == 0)
        {
            // 对端正常关闭（收到 FIN）：这是一个"断开事件"，走异常/清理回调
            conn->_excepter(conn);
            return;
        }
        else                                   // 出错：根据 errno 区分
        {
            if (errno == EAGAIN) break;        // 暂时没数据了，正常退出
            else if (errno == EINTR) continue; // 被信号打断，重试
            else { conn->_excepter(conn); return; } // 真正错误，清理连接
        }
    }
 
    // 读到这儿，说明这轮读完了。把整个 inbuffer 交给上层去解析成"一条条完整报文"
    HandlerRequest(conn);
}

// Sender —— 处理"可写事件"：尝试把 outbuffer 里的数据尽量发出去
static void Sender(Connection *conn)
{
    std::string &outbuffer = conn->OutBuffer();   // 拿到发送缓冲区
    while (1)
    {
        // 非阻塞 send：把 outbuffer 尽量发出去
        ssize_t n = send(conn->SockFd(), outbuffer.c_str(), outbuffer.size(), 0);
        if (n >= 0)
        {
            outbuffer.erase(0, (size_t)n);        // 已经交给内核的，从我们缓冲区里删掉
            if (outbuffer.empty()) break;         // 全发完了，退出
        }
        else
        {
            if (errno == EAGAIN) break;           // 内核发送缓冲满了，这次发不完，退出等下次
            else if (errno == EINTR) continue;    // 被信号打断，重试
            else { conn->_excepter(conn); return; }  // 真正错误，清理
        }
    }
 
    // 走到这说明本轮"能发的都发了"，但可能还没发完
    if (!conn->OutBufferEmpty())
    {
        // 还有没发完的 → 开启对这个 fd EPOLLOUT 的关注，等下一次可写事件再来发
        conn->_R->EnableReadWrite(conn->SockFd(), true, true);
    }
    else
    {
        // 发完了 → 关闭 EPOLLOUT 关注，否则会一直"可写"狂触发，空转烧 CPU
        conn->_R->EnableReadWrite(conn->SockFd(), true, false);
    }
}

// Excepter —— 异常/断开收尾：把连接从各个地方干净地移除
static void Excepter(Connection *conn)
{
    // 清理三部曲的顺序很重要：
    // 1. 先从 epoll 的关注里移除这个 fd，防止它继续触发事件导致悬空回调
    // 2. 从服务器的映射表里移除"fd → conn"
    // 3. 关闭 socket，释放内存
    conn->_R->RemoveConnection(conn->SockFd());   // 内部会做 epoll_ctl(del) 和 map.erase
    conn->Close();                                // close(fd)
    delete conn;                                  // 释放这个 Connection 对象
    // 注意：delete 之后绝不能再碰 conn 了，所有清理动作都要在 delete 之前完成
}

// HandlerRequest —— 上层的协议解析 + 业务处理
static void HandlerRequest(Connection *conn)
{
    std::string &inbuffer = conn->InBuffer();
    std::string message;                      // 一条"符合协议边界的完整报文"
 
    // while (Decode(inbuffer, &message))：从 inbuffer 里抠一条完整报文
    // 抠不出来（半包没凑齐）就返回，剩下的留在 inbuffer 里等下一次读事件
    while (Decoder::Decode(inbuffer, &message))
    {
        // message 一定是完整的报文了
        // 剩下的：反序列化 → 业务计算 → 序列化 → 加边界封装 → 追加进 outbuffer
        // 这里 elided 业务细节，落点在"把应答追加到 outbuffer，交给 Sender 发"
        conn->AddOutBuffer(build_response(message));
    }
 
    // 攒了一批要发的应答，就主动尝试"直接发一次"（而不是干等 EPOLLOUT）
    // —— 因为通常这里 outbuffer 里有货，直接调 Sender 走一轮"尽量发"更高效
    if (!conn->OutBufferEmpty())
    {
        conn->_sender(conn);    // 注意：这只是"尽量发"，不代表能全部发完！
        // 发不完的部分，Sender 会自己注册 EPOLLOUT，等可写事件继续发
    }
}

// reactor_echo.cpp —— 单线程 Reactor 回声服务器（完整可编译运行）
// 用法：g++ -std=c++17 -O2 -pthread reactor_echo.cpp -o reactor_echo && ./reactor_echo
// 客户端：nc 127.0.0.1 8888  （或 telnet），随便发什么，原样回给你
#include <iostream>
#include <string>
#include <functional>
#include <unordered_map>
#include <cstring>
#include <cerrno>
#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
 
// ============ 工具函数 ============
void SetNonBlock(int fd)                       // 把 fd 设为非阻塞
{
    int fl = fcntl(fd, F_GETFL);               // 取当前标志
    fcntl(fd, F_SETFL, fl | O_NONBLOCK);       // 加上非阻塞位
}
 
// 一个简单的消息边界协议：每条消息前面用 4 字节大端表示长度（length-prefix）
// 这能彻底解决"半包/粘包"，因为有了长度就能精确切出每个完整报文
struct ByteStream
{
    std::string data;                          // 累积的未解析字节
 
    // 尝试从 data 里取出一条完整报文（前面 4 字节是长度）
    // 成功返回 true 并把报文装进 out；长度不够（半包）返回 false，data 里的不动
    bool Decode(std::string &out)
    {
        if (data.size() < 4) return false;     // 连长度头都没凑齐，半包，等下次
        uint32_t len = 0;
        for (int i = 0; i < 4; i++)            // 组装大端长度
            len = (len << 8) | (unsigned char)data[i];
        if (data.size() < 4u + len) return false;    // 长度够了，但数据还没到齐，仍是半包
        out = data.substr(4, len);             // 切出这条完整报文
        data.erase(0, 4u + len);               // 已消费的部分从缓冲区删除
        return true;
    }
 
    static std::string Encode(const std::string &msg)  // 给报文加上 4 字节长度头
    {
        uint32_t len = (uint32_t)msg.size();
        std::string head(4, '0');
        for (int i = 0; i < 4; i++)            // 大端写长度
            head[i] = (char)((len >> (8 * (3 - i))) & 0xFF);
        return head + msg;
    }
};
 
// ============ 前置声明 ============
class TcpServer;
class Connection;
using func_t = std::function<void(Connection *)>;   // 回调类型：吃一个 Connection*
 
// ============ Connection：一个连接 ============
class Connection
{
public:
    Connection(int fd, uint32_t ev, TcpServer *t)
        : fd_(fd), events_(ev), server_(t) {}
 
    void Reg(func_t r, func_t w, func_t e) { on_read_ = r; on_write_ = w; on_err_ = e; }
    void AddIn(const std::string &d)    { recv_buf_.data += d; }   // 追加进接收侧
    void AddOut(const std::string &d)   { send_buf_ += d; }        // 累积到发送缓冲
    int  fd() const { return fd_; }
    uint32_t events() const { return events_; }
    void SetEvents(uint32_t e) { events_ = e; }
    std::string &OutBuf() { return send_buf_; }
    ByteStream &InBuf()  { return recv_buf_; }
    void Close() { ::close(fd_); }
 
    func_t on_read_;      // 可读回调：监听连接→Accepter，数据连接→Recver
    func_t on_write_;     // 可写回调：Sender
    func_t on_err_;       // 异常回调：Excepter
    TcpServer *server_;   // 回指服务器，方便调用它的注册/移除接口
private:
    int      fd_;         // socket fd
    uint32_t events_;     // 关注的事件位
    ByteStream recv_buf_; // 接收字节流缓冲
    std::string send_buf_;// 发送缓冲
};
 
// ============ 事件处理器（集中的"处理逻辑"） ============
struct Handler
{
    // Accepter：监听 socket 可读 → accept 新连接
    static void Accepter(Connection *lconn)
    {
        while (1)
        {
            struct sockaddr_in peer; socklen_t len = sizeof(peer);
            int cfd = ::accept(lconn->fd(), (struct sockaddr *)&peer, &len);
            if (cfd > 0)
            {
                SetNonBlock(cfd);
                char ip[INET_ADDRSTRLEN];
                inet_ntop(AF_INET, &(peer.sin_addr), ip, sizeof(ip));
                std::cout << "[+] new conn " << cfd << " from " << ip
                          << ":" << ntohs(peer.sin_port) << std::endl;
 
                // 造一个数据连接：只关心 可读（ET），回调都指向 Handler 的静态方法
                Connection *c = new Connection(cfd, EPOLLIN | EPOLLET, lconn->server_);
                c->Reg(Recver, Sender, Excepter);
                lconn->server_->AddConnection(c);   // 注册进 epoll 和映射表
            }
            else
            {
                if (errno == EAGAIN) break;         // 新连接收完了
                else if (errno == EINTR) continue;
                else break;
            }
        }
        std::cout << "[accepter] done for listen " << lconn->fd() << std::endl;
    }
 
    // Recver：数据连接可读 → 读入缓冲，再尝试解析成一条条报文
    static void Recver(Connection *c)
    {
        char buf[4096];
        while (1)
        {
            ssize_t n = ::recv(c->fd(), buf, sizeof(buf), 0);
            if (n > 0) { c->AddIn(std::string(buf, (size_t)n)); }   // 追加，不是覆盖
            else if (n == 0) { DeferClose(c); return; }             // 对端 FIN：断开
            else {
                if (errno == EAGAIN) break;   // 读完一轮
                else if (errno == EINTR) continue;
                else { DeferClose(c); return; }
            }
        }
        // 收完一轮，解析 + 业务 + 攒应答
        std::string msg;
        while (c->InBuf().Decode(msg))                       // 一条条完整报文
        {
            std::cout << "[recv " << c->fd() << "] " << msg << std::endl;
            std::string reply = ByteStream::Encode(msg);     // 回声：原样回
            c->AddOut(reply);                                // 先攒进发送缓冲
        }
        if (!c->OutBuf().empty()) c->on_write_(c);           // 有货就立刻"尽量发"一轮
    }
 
    // Sender：尝试把发送缓冲发出去；发不完注册 EPOLLOUT
    static void Sender(Connection *c)
    {
        std::string &ob = c->OutBuf();
        while (1)
        {
            ssize_t n = ::send(c->fd(), ob.data(), ob.size(), 0);
            if (n >= 0) {
                ob.erase(0, (size_t)n);
                if (ob.empty()) break;                       // 全发完
            } else {
                if (errno == EAGAIN) break;                  // 内核缓冲满，等下次 EPOLLOUT
                else if (errno == EINTR) continue;
                else { DeferClose(c); return; }
            }
        }
        // 发完与否，决定是否还关心 EPOLLOUT
        c->server_->EnableReadWrite(c->fd(), true, !ob.empty());
    }
 
    // Excepter：统一收尾（这里用"延迟关闭"避免悬垂）
    static void Excepter(Connection *c) { DeferClose(c); }
 
    // 延迟关闭：只打标记，真正的 delete 交给事件循环在安全时刻做
    static void DeferClose(Connection *c) { c->server_->DeferClose(c); }
};
 
// ============ TcpServer：Reactor 主控 ============
class TcpServer
{
public:
    TcpServer(uint16_t port) : port_(port) {}
 
    void Start()
    {
        // 1. 建监听 socket
        lfd_ = ::socket(AF_INET, SOCK_STREAM, 0);
        int opt = 1;
        setsockopt(lfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
        struct sockaddr_in a; std::memset(&a, 0, sizeof(a));
        a.sin_family = AF_INET; a.sin_port = htons(port_); a.sin_addr.s_addr = INADDR_ANY;
        ::bind(lfd_, (struct sockaddr *)&a, sizeof(a));
        ::listen(lfd_, 512);
        SetNonBlock(lfd_);
 
        // 2. 建 epoll
        epfd_ = ::epoll_create1(0);
 
        // 3. 用工厂造"监听连接"，回调指向 Accepter，只关心可读（ET）
        Connection *lc = new Connection(lfd_, EPOLLIN | EPOLLET, this);
        lc->Reg(Handler::Accepter, nullptr, nullptr);
        AddConnection(lc);
 
        std::cout << "[reactor] listening on " << port_ << std::endl;
        Loop();
    }
 
    void AddConnection(Connection *c)   // 把一个连接纳入 epoll + 映射
    {
        struct epoll_event ev{};
        ev.events   = c->events();
        ev.data.fd  = c->fd();          // 用 data.fd 携带回查线索
        ::epoll_ctl(epfd_, EPOLL_CTL_ADD, c->fd(), &ev);
        conns_[c->fd()] = c;
    }
 
    void RemoveConnection(int fd)       // 从 epoll 和映射里摘掉（不 delete）
    {
        ::epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
        auto it = conns_.find(fd);
        if (it != conns_.end()) { conns_.erase(it); }
    }
 
    void EnableReadWrite(int fd, bool rd, bool wr)  // 动态调整某 fd 关注的事件位
    {
        auto it = conns_.find(fd);
        if (it == conns_.end()) return;
        Connection *c = it->second;
        uint32_t ev = EPOLLET;          // 永远保持边缘触发
        if (rd) ev |= EPOLLIN;
        if (wr) ev |= EPOLLOUT;
        c->SetEvents(ev);
        struct epoll_event ee{};
        ee.events = ev; ee.data.fd = fd;
        ::epoll_ctl(epfd_, EPOLL_CTL_MOD, fd, &ee);   // MOD 修改，不是 ADD
    }
 
    void DeferClose(Connection *c)      // 记进待清理列表，稍后统一释放
    {
        dead_.push_back(c);
    }
 
private:
    void Loop()                          // 事件循环本体
    {
        const int MAXE = 64;
        struct epoll_event evs[MAXE];
        while (1)
        {
            int n = ::epoll_wait(epfd_, evs, MAXE, -1);
            if (n < 0) { if (errno == EINTR) continue; break; }
 
            for (int i = 0; i < n; i++)
            {
                int fd = evs[i].data.fd;
                uint32_t ev = evs[i].events;
                auto it = conns_.find(fd);
                if (it == conns_.end()) continue;   // 已被清理，跳过
                Connection *c = it->second;
 
                if (ev & EPOLLIN)  { if (c->on_read_)  c->on_read_(c);  }
                if (ev & EPOLLOUT) { if (c->on_write_) c->on_write_(c); }
                if (ev & (EPOLLERR | EPOLLHUP))
                {
                    if (c->on_err_) c->on_err_(c);
                    continue;
                }
            }
 
            // ---- 本轮分发结束，此刻安全，统一释放"待关闭"连接 ----
            for (Connection *c : dead_)
            {
                std::cout << "[-] close conn " << c->fd() << std::endl;
                RemoveConnection(c->fd());   // 1 先从 epoll/map 摘掉
                c->Close();                  // 2 关闭 fd
                delete c;                    // 3 释放内存
            }
            dead_.clear();
        }
    }
 
    uint16_t port_;
    int lfd_ = -1, epfd_ = -1;
    std::unordered_map<int, Connection *> conns_;  // fd → 连接对象
    std::vector<Connection *> dead_;               // 本轮回调节点产生的"待清理连接"
};
 
int main()
{
    TcpServer s(8888);     // 监听 8888 端口
    s.Start();             // 启动，进入永不返回的事件循环
    return 0;
}

# 编译（用 C++17 及以上）
g++ -std=c++17 -O2 -pthread reactor_echo.cpp -o reactor_echo
# 启动服务器
./reactor_echo
# 另开一个终端，多开几个客户端连上来试试
nc 127.0.0.1 8888

// eventfd_demo.c —— eventfd 跨进程通知最小示例（可编译运行：gcc eventfd_demo.c -o efd）
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdint.h>
#include <sys/eventfd.h>
 
int main(void)
{
    // 创建 eventfd：初始计数 0，EFD_CLOEXEC 保证 exec 时自动关闭
    int efd = eventfd(0, EFD_CLOEXEC);
    if (efd == -1) { perror("eventfd"); return 1; }
 
    pid_t pid = fork();
    if (pid == -1) { perror("fork"); return 1; }
 
    if (pid == 0)                     /* 子进程 */
    {
        uint64_t value;
        // 阻塞读：等父进程写，读到事件值
        if (read(efd, &value, sizeof(value)) < 0) perror("read");
        printf("Child received event, counter was %llu\n",
               (unsigned long long)value);
        close(efd);
        return 0;
    }
    else                              /* 父进程 */
    {
        uint64_t value = 1;
        // 写 1 到计数器，相当于"发一次事件"；子进程(或阻塞在 epoll_Wait 的事件循环)会被唤醒
        if (write(efd, &value, sizeof(value)) < 0) perror("write");
        printf("Parent sent event\n");
        wait(NULL);                   /* 等子进程结束 */
        close(efd);
        return 0;
    }
}

模式一（默认）:                 模式二（EFD_SEMAPHORE）:
 写 1 → 计数器 1                写 1 → 计数器 1
 写 2 → 计数器 3                写 2 → 计数器 3
 read → 读出 3，计数器变 0      read → 读出 1，计数器变 2
 read → 读出 0，计数器 0        read → 读出 1，计数器变 1
 read → 读出 0，计数器 0        read → 读出 1，计数器变 0