任务 7：课后练习（打印客户端地址、群聊、输出缓冲区）

前提： 完成任务 6，epoll_echo.cpp 能跑通。

怎么用这篇： 每个练习先看“要做什么”和“提示”，自己动手改，卡住超过 30 分钟再展开参考答案。直接抄答案等于没做。

预计用时： 5 到 8 小时。

先把 epoll_echo.cpp 复制三份，每个练习改一份：

mkdir -p ~/code/cpp-practice/task7
cd ~/code/cpp-practice/task7
cp ../task6/epoll_echo.cpp echo_addr.cpp
cp ../task6/epoll_echo.cpp chat_server.cpp
cp ../task6/epoll_echo.cpp echo_buffered.cpp

练习 1：打印客户端的 IP 和端口

要做什么

有新连接时打印 new client 127.0.0.1:35340 fd=5，断开时打印 client 127.0.0.1:35340 fd=5 closed。

提示

accept 的第二、三个参数就是用来拿客户端地址的。准备一个 sockaddr_in peer{} 和一个 socklen_t len = sizeof(peer)，把 (sockaddr*)&peer 和 &len 传进去。
len 必须先设成结构体的大小。 它是一个“输入输出参数”：进去时告诉内核你的结构体有多大，出来时内核告诉你实际写了多少。不设的话里面是垃圾值，可能拿不到地址。
peer.sin_addr 里的 IP 是二进制的，要用 inet_ntop 转成 "127.0.0.1" 这样的字符串。
peer.sin_port 是网络字节序，要用 ntohs 转回来再打印。
断开时也要打印地址，但那时已经拿不到 peer 了，所以要在 accept 时把地址存起来。用一个 std::map<int, std::string>，fd 对应地址字符串。

参考答案

展开 echo_addr.cpp

echo_addr.cpp 下载

// 练习 1：在 epoll echo 的基础上，打印每个客户端的 IP 和端口
#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cerrno>
#include <cstdio>
#include <map>
#include <string>

static void set_nonblock(int fd) {
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}

// 把 sockaddr_in 转成 "127.0.0.1:41844" 这样的字符串
static std::string to_ip_port(const sockaddr_in& a) {
    char ip[INET_ADDRSTRLEN];
    inet_ntop(AF_INET, &a.sin_addr, ip, sizeof(ip));  // 二进制 IP -> 点分十进制字符串
    return std::string(ip) + ":" + std::to_string(ntohs(a.sin_port));  // 端口转回主机字节序
}

int main() {
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(8888);
    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0) { perror("bind"); return 1; }
    listen(listen_fd, 1024);
    set_nonblock(listen_fd);

    int epfd = epoll_create1(0);
    epoll_event ev{};
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);
    printf("listening on 8888\n");

    std::map<int, std::string> peers;  // fd -> "ip:port"，断开时还要打印
    epoll_event events[1024];
    char buf[4096];
    while (true) {
        int n = epoll_wait(epfd, events, 1024, -1);
        for (int i = 0; i < n; ++i) {
            int fd = events[i].data.fd;
            if (fd == listen_fd) {
                while (true) {
                    sockaddr_in peer{};
                    socklen_t len = sizeof(peer);  // 必须先设成结构体的大小，accept 会改写它
                    int conn_fd = accept(listen_fd, (sockaddr*)&peer, &len);
                    if (conn_fd < 0) break;
                    set_nonblock(conn_fd);
                    epoll_event cev{};
                    cev.events = EPOLLIN;
                    cev.data.fd = conn_fd;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &cev);
                    peers[conn_fd] = to_ip_port(peer);
                    printf("new client %s fd=%d\n", peers[conn_fd].c_str(), conn_fd);
                }
            } else {
                ssize_t r = read(fd, buf, sizeof(buf));
                if (r > 0) {
                    write(fd, buf, r);
                } else if (r == 0 || errno != EAGAIN) {
                    printf("client %s fd=%d closed\n", peers[fd].c_str(), fd);
                    peers.erase(fd);
                    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr);
                    close(fd);
                }
            }
        }
    }
}

运行结果

g++ -Wall -Wextra -o echo_addr echo_addr.cpp
./echo_addr

另一个终端跑 python3 ../task6/test_clients.py addr 3，服务器输出：

listening on 8888
new client 127.0.0.1:35340 fd=5
new client 127.0.0.1:35342 fd=6
new client 127.0.0.1:35356 fd=7
client 127.0.0.1:35342 fd=6 closed
client 127.0.0.1:35340 fd=5 closed
client 127.0.0.1:35356 fd=7 closed

为什么第一个连接是 fd 5 而不是 4？ 0、1、2 是标准输入输出，3 是 listen_fd，4 是 epfd（epoll 实例本身也是一个 fd）。

端口号 35340 是哪来的？ 客户端没有指定端口，系统随机分配的临时端口。这就是任务 4 里讲的四元组：三个连接的服务器 IP 和端口都一样，只有客户端端口不同。

练习 2：改成群聊服务器

要做什么

任何一个客户端发的消息，转发给其他所有在线的客户端（不发回给自己），并在前面加上发送者的编号，比如 [user 5] hello everyone。

提示

用一个 std::set<int> 记录所有在线的连接 fd。accept 时 insert，断开时 erase。
读到消息后，遍历这个 set，对每个不等于自己的 fd 调用 write。
用 std::string 拼接前缀："[user " + std::to_string(fd) + "] " + std::string(buf, r)。注意 std::string(buf, r) 要传长度，因为 buf 不是以 \0 结尾的。

参考答案

展开 chat_server.cpp

chat_server.cpp 下载

// 练习 2：群聊服务器。任何一个客户端发的消息，转发给其他所有客户端
#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cerrno>
#include <cstdio>
#include <set>
#include <string>

static void set_nonblock(int fd) {
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}

int main() {
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(8888);
    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0) { perror("bind"); return 1; }
    listen(listen_fd, 1024);
    set_nonblock(listen_fd);

    int epfd = epoll_create1(0);
    epoll_event ev{};
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);
    printf("chat server listening on 8888\n");

    std::set<int> clients;  // 当前在线的所有连接
    epoll_event events[1024];
    char buf[4096];
    while (true) {
        int n = epoll_wait(epfd, events, 1024, -1);
        for (int i = 0; i < n; ++i) {
            int fd = events[i].data.fd;
            if (fd == listen_fd) {
                while (true) {
                    int conn_fd = accept(listen_fd, nullptr, nullptr);
                    if (conn_fd < 0) break;
                    set_nonblock(conn_fd);
                    epoll_event cev{};
                    cev.events = EPOLLIN;
                    cev.data.fd = conn_fd;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &cev);
                    clients.insert(conn_fd);
                    printf("user %d joined, %zu online\n", conn_fd, clients.size());
                }
            } else {
                ssize_t r = read(fd, buf, sizeof(buf));
                if (r > 0) {
                    // 在消息前面加上发送者的编号，再发给除了自己以外的所有人
                    std::string msg = "[user " + std::to_string(fd) + "] " + std::string(buf, r);
                    for (int other : clients) {
                        if (other != fd) write(other, msg.data(), msg.size());
                    }
                } else if (r == 0 || errno != EAGAIN) {
                    clients.erase(fd);
                    printf("user %d left, %zu online\n", fd, clients.size());
                    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr);
                    close(fd);
                }
            }
        }
    }
}

运行结果

g++ -Wall -Wextra -o chat_server chat_server.cpp
./chat_server

开三个终端，各运行一个 nc 127.0.0.1 8888，在其中一个里输入 hello everyone，另外两个会看到：

[user 5] hello everyone

我用脚本让 A、B、C 三个客户端连上，A 发一条消息，实际结果：

B received: b'[user 5] hello everyone\n'
C received: b'[user 5] hello everyone\n'
A received nothing (its own message is not sent back)

服务器日志：

chat server listening on 8888
user 5 joined, 1 online
user 6 joined, 2 online
user 7 joined, 3 online
user 6 left, 2 online
user 5 left, 1 online
user 7 left, 0 online

想一想

这个群聊服务器有两个问题，阶段 1 和阶段 2 会解决：

一条消息可能被拆开或者粘在一起。 A 快速连发两条，服务器可能一次读到两条拼在一起的，也可能一条只读到一半。要真正区分“一条消息”，需要约定消息格式，比如每条以 \n 结尾，服务器攒够一整行再转发。这叫分帧，阶段 2 解析 Redis 协议时会专门处理。
write 可能写不完。 如果某个客户端接收得很慢，发给它的数据就会丢。这正是下一个练习要解决的。

练习 4：加输出缓冲区，解决“写不完”

（练习 3 是“用 nc 亲眼看阻塞版的问题”，已经在任务 5 做过了。）

先证明问题存在

下载这个测试脚本：

big_test.py 下载

# 测试“写不完”：先发 20 MB 数据，故意晚 1 秒才开始读回显，看最后收回来的是否完整
import socket
import threading
import time

SIZE = 20 * 1024 * 1024
data = bytes(i % 251 for i in range(SIZE))  # 有规律的内容，方便比对

s = socket.create_connection(("127.0.0.1", 8888))

def sender():
    s.sendall(data)

t = threading.Thread(target=sender)
t.start()
time.sleep(1)  # 这 1 秒里只发不收：服务器写回来的数据会把我们的接收缓冲区塞满

received = bytearray()
s.settimeout(2)
try:
    while len(received) < SIZE:
        chunk = s.recv(1 << 20)
        if not chunk:
            break
        received += chunk
except socket.timeout:
    pass
t.join(timeout=1)
print("sent", SIZE, "bytes, got back", len(received), "bytes,",
      "OK" if bytes(received) == data else "DATA LOST")

它发 20 MB 数据，但故意晚 1 秒才开始读回显。在这 1 秒里，服务器写回来的数据会塞满客户端的接收缓冲区，服务器的 write 就会写不完。

用任务 6 的原版 epoll_echo 跑一下：

./epoll_echo                 # 终端 1
python3 big_test.py          # 终端 2

实际结果：

sent 20971520 bytes, got back 4209610 bytes, DATA LOST

发了 20 MB，只收回来 4 MB，丢了 80% 的数据。原版代码 write(fd, buf, r) 只写进去一部分时，剩下的就被扔掉了；write 返回 -1（EAGAIN）时，整块数据都被扔掉了。

要做什么

改造 echo_buffered.cpp，让 big_test.py 输出 OK。

思路

write 的返回值是实际写进内核发送缓冲区的字节数，可能比你想写的少。 发送缓冲区满了（对方接收慢）时，非阻塞的 write 会返回 -1，errno 是 EAGAIN。

所以：

给每个连接准备一个输出缓冲区（std::map<int, std::string>）。
读到数据后，不直接 write，而是先追加到输出缓冲区，然后尝试把缓冲区里的数据写出去（写多少算多少，写出去的部分从缓冲区里删掉）。
如果缓冲区里还有剩的，说明内核发送缓冲区满了。这时告诉 epoll 我们关心 EPOLLOUT（可写）事件。等发送缓冲区有空间了，epoll 会通知我们，再接着写。
写完了，马上取消关心 EPOLLOUT。

为什么写完要取消 EPOLLOUT

在默认的 LT 模式下，只要发送缓冲区有空间，EPOLLOUT 就一直是“就绪”的。而发送缓冲区绝大多数时候都是有空间的。如果一直关心 EPOLLOUT，epoll_wait 会每次都立刻返回，程序变成死循环，CPU 跑到 100%。

所以规则是：只在有数据没写完的时候关心 EPOLLOUT，写完立刻取消。

提示

写一个 flush(fd) 函数：循环 write，写出去多少就从字符串开头 erase 多少；遇到 EAGAIN 就停；遇到其他错误返回 false。
写一个 watch_write(fd, on) 函数：用 EPOLL_CTL_MOD 把关心的事件改成 EPOLLIN | EPOLLOUT 或者只有 EPOLLIN。
处理事件时，要用 events[i].events 判断是哪种事件：if (e & EPOLLIN) 处理读，if (e & EPOLLOUT) 处理写。一次通知可能两个都有。
在 main 开头加上 signal(SIGPIPE, SIG_IGN);。往一个对方已经关闭的连接写数据，进程会收到 SIGPIPE 信号，默认处理方式是直接杀死进程。忽略它之后，write 只会返回 -1，errno 是 EPIPE，按出错处理关掉连接就行。

参考答案

展开 echo_buffered.cpp

echo_buffered.cpp 下载

// 练习 4：给每个连接加输出缓冲区，解决 write 写不完的问题
#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cerrno>
#include <csignal>
#include <cstdio>
#include <map>
#include <string>

static void set_nonblock(int fd) {
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}

static int epfd;
static std::map<int, std::string> outbuf;  // 每个连接还没发出去的数据

// 修改一个连接关心的事件：要不要关心 EPOLLOUT
static void watch_write(int fd, bool on) {
    epoll_event ev{};
    ev.events = on ? (EPOLLIN | EPOLLOUT) : EPOLLIN;
    ev.data.fd = fd;
    epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
}

static void close_conn(int fd) {
    outbuf.erase(fd);
    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr);
    close(fd);
}

// 尽量把输出缓冲区里的数据写出去；写不完的留在缓冲区里。出错返回 false
static bool flush(int fd) {
    std::string& out = outbuf[fd];
    while (!out.empty()) {
        ssize_t w = write(fd, out.data(), out.size());
        if (w > 0) {
            out.erase(0, w);  // 删掉已经写出去的部分
        } else if (w < 0 && errno == EAGAIN) {
            break;            // 内核发送缓冲区满了，等 EPOLLOUT 再写
        } else {
            return false;     // 真的出错了（比如对方已经关闭）
        }
    }
    return true;
}

int main() {
    signal(SIGPIPE, SIG_IGN);  // 往已关闭的连接写数据时，不要让 SIGPIPE 信号杀死进程

    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(8888);
    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0) { perror("bind"); return 1; }
    listen(listen_fd, 1024);
    set_nonblock(listen_fd);

    epfd = epoll_create1(0);
    epoll_event ev{};
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);
    printf("listening on 8888 (buffered)\n");

    epoll_event events[1024];
    char buf[65536];
    while (true) {
        int n = epoll_wait(epfd, events, 1024, -1);
        for (int i = 0; i < n; ++i) {
            int fd = events[i].data.fd;
            if (fd == listen_fd) {
                while (true) {
                    int conn_fd = accept(listen_fd, nullptr, nullptr);
                    if (conn_fd < 0) break;
                    set_nonblock(conn_fd);
                    epoll_event cev{};
                    cev.events = EPOLLIN;
                    cev.data.fd = conn_fd;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &cev);
                    outbuf[conn_fd];  // 创建一个空的输出缓冲区
                }
                continue;
            }

            uint32_t e = events[i].events;
            if (e & EPOLLIN) {
                ssize_t r = read(fd, buf, sizeof(buf));
                if (r > 0) {
                    outbuf[fd].append(buf, r);  // 先放进输出缓冲区
                    if (!flush(fd)) { close_conn(fd); continue; }
                    if (!outbuf[fd].empty()) watch_write(fd, true);  // 没写完，关心可写事件
                } else if (r == 0 || errno != EAGAIN) {
                    close_conn(fd);
                    continue;
                }
            }
            if (e & EPOLLOUT) {
                // 内核发送缓冲区又有空间了，接着写
                if (!flush(fd)) { close_conn(fd); continue; }
                if (outbuf[fd].empty()) watch_write(fd, false);  // 写完了，不再关心可写
            }
        }
    }
}

运行结果

g++ -Wall -Wextra -O2 -o echo_buffered echo_buffered.cpp
./echo_buffered              # 终端 1
python3 big_test.py          # 终端 2

实际结果：

sent 20971520 bytes, got back 20971520 bytes, OK

20 MB 一个字节都没丢。再跑一下 python3 ../task6/test_clients.py buffered 500，确认普通情况也没问题：

buffered echo ok: 500 / 500

这次加 -O2 编译也不会再有 ignoring return value of write 的警告了，因为我们检查了返回值。

这个方案还有什么问题

如果客户端永远不读，服务器给它的输出缓冲区会无限增长，最后把内存吃光。真实的服务器要给输出缓冲区设一个上限（叫高水位），超过就停止读这个客户端的新请求，或者直接断开它。Redis 的 client-output-buffer-limit 配置就是干这个的。阶段 1 的 Buffer 和 TcpConnection 会处理这个问题。

交作业

cd ~/code/cpp-practice
git add task7
git commit -m "Add echo exercises: client address, chat server, output buffer"
git push

检查清单

练习 1：能打印客户端地址，知道 len 为什么要先初始化
练习 2：群聊能用，能说出它的两个问题
练习 4：亲眼看到原版 DATA LOST，改完之后 OK
能解释为什么写完要取消 EPOLLOUT
知道 SIGPIPE 是什么，为什么要忽略它

自测

accept 的第三个参数为什么是指针，而且要先赋值？

它是输入输出参数：调用前告诉内核“我的地址结构体有多大”，内核写入地址后再把实际长度写回去。不初始化的话，内核读到的是一个随机值，可能写不进地址，或者行为不确定。

EPOLLOUT 是什么时候触发的？为什么不能一直关心它？

socket 的内核发送缓冲区有空闲空间时就是“可写”的。绝大多数时候发送缓冲区都有空间，所以在 LT 模式下一直关心 EPOLLOUT 的话，epoll_wait 每次都会立刻返回它，变成忙等，CPU 跑满。只有输出缓冲区里有没写完的数据时才需要关心它。

big_test.py 为什么要等 1 秒再开始读？

为了制造“对方接收慢”的情况。客户端不读，它的接收缓冲区很快被服务器发回来的数据填满，TCP 的流量控制会让服务器的发送缓冲区也跟着满，于是服务器的 write 只能写一部分或者返回 EAGAIN。如果客户端一边发一边读，这种情况很难出现，bug 就藏起来了。

面试常问

非阻塞 socket 上 write 返回的值比要写的少，怎么处理？

参考回答： 说明内核发送缓冲区满了，剩下的数据要先存进这个连接的应用层输出缓冲区，然后在 epoll 上注册 EPOLLOUT。等可写事件到来时，接着把输出缓冲区的数据写出去，写完后取消 EPOLLOUT，否则 LT 模式下会一直触发导致忙等。同时要给输出缓冲区设一个上限，防止慢客户端把服务器内存撑爆。

服务器往一个已经断开的连接 write，会发生什么？

参考回答： 第一次 write 通常能写进内核缓冲区，对方会回 RST；再写的话进程会收到 SIGPIPE 信号，默认行为是终止进程，表现为服务器莫名其妙退出。所以服务器启动时要 signal(SIGPIPE, SIG_IGN)，或者用 send 带 MSG_NOSIGNAL，这样 write 只会返回 -1、errno 为 EPIPE，按正常流程关闭连接。

如果一个客户端只发不收，服务器会怎样？

参考回答： 发给它的回复写不出去，会堆在服务器为它准备的输出缓冲区里，越积越多，最后可能耗尽内存。解决办法是设置输出缓冲区的高水位：超过之后暂停读取这个客户端的请求，或者直接断开它。Redis 的 client-output-buffer-limit 就是这么做的。