# 项目 3 · 有界文件统计流水线 Leon | 原创教学项目。[English guide](README.md) 米娜帮学校机器人社整理每天的笔记。她收到一组小文本文件，需要统计字节数、行数和单词数。有些文件是空的，有些文件末尾没有换行。某个文件不存在时，报告也必须说明问题，并保留其他文件的统计。她想尝试多个工作线程，但报告的顺序必须与输入文件列表一致。 她先做出能准确统计单个文件的工具，然后把编号任务放入只有四个位置的队列。工作线程取出编号，各自打开文件。队列满时，主线程等待空位，不再分配新位置。这就是背压。任务先完成不代表先输出。主线程等所有工作线程结束后，按编号打印结果。 这是原创的引导式工程项目，不是课程作业答案。需要 C11、POSIX 线程、Clang、Make 和 Python 3。源码面向 macOS 与 Linux，不需要网络，也不需要第三方 Python 库。 ## 编译与运行 在本项目目录中执行： ```sh make ./thread_pipeline 3 fixtures/morning.txt fixtures/evening.txt fixtures/empty.txt ``` 程序的标准输出如下。`make` 显示的编译命令不属于程序输出。 ```text job=1 bytes=37 lines=2 words=6 job=2 bytes=26 lines=2 words=5 job=3 bytes=0 lines=0 words=0 summary files=3 ok=3 failed=0 ``` 文件列表可以为空。主线程关闭空队列后，等待的线程会醒来并退出。 ```sh ./thread_pipeline 16 ``` ```text summary files=0 ok=0 failed=0 ``` 某个文件不存在，不会丢弃其他结果。 ```sh ./thread_pipeline 2 fixtures/morning.txt fixtures/missing.txt fixtures/evening.txt ``` ```text job=1 bytes=37 lines=2 words=6 job=2 error=open job=3 bytes=26 lines=2 words=5 summary files=3 ok=2 failed=1 ``` 前两次运行的退出码为 0。缺少文件的运行退出码为 1，还会向标准错误写入诊断。系统错误文字因平台而异。`job=2` 表示第二个文件参数，不是第二个工作线程。相同路径可以重复出现，每个参数都算独立任务。 ```sh ./thread_pipeline 0 fixtures/morning.txt ``` 该命令只向标准错误打印用法，不产生标准输出，退出码为 2。WORKERS 只能包含十进制数字，数值必须在 1 到 16 之间，可以有前导零。最多接受 128 个文件参数。这里不解析选项，WORKERS 后的每个参数都当作文件路径。 ## 先确定统计定义 * 字节数计算读到的每个字节，包括空白和零字节。它不是 Unicode 字符数。 * 行数等于 LF 字节（`\n`）的数量；若文件非空且最后一个字节不是 LF，再加一。空文件有 0 行，`a` 有 1 行，`a\n` 有 1 行，`\n\n` 有 2 行。末尾的换行不会凭空增加一行。这与只计算换行符数量的工具不同。 * 单词是由非分隔字节组成的最长连续段。分隔字节仅为空格、制表符、LF、CR、垂直制表符和换页符，定义不受地区设置影响。零字节和非 ASCII 字节都视为普通非分隔字节；不会当作字符串结尾，也不会解码成文字。 * 每个普通文件最多 8,388,608 字节，即 8 MiB。超过上限时输出 `error=too-large`，不把部分计数冒充完整统计。`in_word` 状态在每次 4096 字节的读取之间保留，因此跨缓冲区的一个单词仍只计算一次。 合成样例包括 37 字节的早间笔记、26 字节且末尾没有 LF 的晚间笔记、两行空行、零字节文件，以及六个空白字节。空白样例的 LF 后还有字节，所以按本项目定义有两行。测试必须以这些约定为准。 ## 分五步阅读实现 1. **定义单个文件的结果。** 先读 `stats.h`，再读 `stats.c` 中的 `separator` 和 `count_stream`。用 `a b\n` 手动追踪 `bytes`、`lines`、`words`、`in_word` 和 `last`。循环统计单词的起点。到达文件末尾后，最多补计最后一行。`read` 的返回类型可以表示负数，必须先检查错误，再转换成 `size_t`。 2. **建立有界队列。** 阅读 `queue.h` 和 `queue_push`。`head` 指向下一个要取出的元素，插入位置为 `(head + count) % 4`。不变量为 `0 <= count <= 4`。队列满且未关闭时，生产者等待；队列空且未关闭时，消费者等待。检查和修改共用同一把互斥锁。`pthread_cond_wait` 在等待时释放锁，返回前重新获得锁。通知只表示需要重新检查条件，并不预留位置，因此必须用 `while`，也要允许虚假唤醒。 3. **让每个结果只有一个写入者。** 阅读 `main.c` 中的 `WorkerContext` 和 `consume`。主线程给每个输入分配一个编号，并且只入队一次。取出编号时持有队列锁，因此只有一个工作线程能得到它。扫描期间，该线程独占 `results[index]`，并拥有自己打开的文件描述符。文件读取发生在队列锁之外，让其他线程能够继续取任务。 4. **关闭、排空、等待结束。** 阅读 `queue_close` 和 `main` 后半段。关闭只禁止新任务，不删除队列中的旧任务。队列既空又关闭时，`queue_pop` 才返回 false。广播使所有空闲消费者醒来，包括零任务时的十六个线程。主线程等待每个成功创建的线程结束后，才销毁共享对象并打印。 5. **验证成功和失败路径。** 阅读 `parse_workers`、启动清理、`print_results` 和 `test.py`。全部工作线程创建成功后才发布任务。若创建失败，关闭仍为空的队列，只等待已经成功创建的线程，然后销毁队列并返回失败。文件错误存入该文件自己的结果槽，其他任务继续。测试比较独立算出的正确结果，不依赖线程的速度或执行先后。 ## 所有权和执行阶段 ```text 主线程：初始化队列、上下文、结果数组 | +--> 创建全部工作线程 --> 发布编号 --> 关闭队列 | | | 工作线程： +--> 等待/取任务 -+--> 扫描 --> 取任务 --> 排空 --> 返回 | 主线程：等待每个已创建线程结束 --> 销毁队列 --> 按 0..J-1 打印结果 创建失败：关闭空队列 --> 等待已创建线程 --> 销毁 --> 退出码 1 ``` 队列、上下文和结果数组保存在主线程的栈帧中，所有 join 完成后才结束其生命周期。`argv` 中的路径在进程运行期间保持有效，线程只借用，不修改。队列元素只是编号，不拥有文件或动态分配。每个工作线程拥有自己的读取缓冲区和文件描述符。成功打开后，`scan_file` 的每条路径都会尝试关闭描述符。主线程拥有可等待的线程句柄，只 join 成功创建的线程。 `pthread_create` 之前的初始化使工作线程能够访问正确的上下文。互斥锁为队列的读写建立先后关系，也把编号安全地交给消费者。数组的不同元素是不同对象，工作线程不会同时写同一个 `FileStats`。主线程在所有 `pthread_join` 成功后才读取结果，这些同步操作让工作线程完成的写入对主线程可见。所有权和阶段限制共同省去了结果锁。仅仅期待主线程“大概会晚一点运行”不能保证正确。 ## 错误与有意保留的限制 `scan_file` 区分打开、元数据查询、非普通文件、读取、超出大小和关闭错误。出现多个问题时保留第一个。读取因 `EINTR` 中断会重试；短读取的有效字节会被处理，然后继续读取。非阻塞打开避免因 FIFO 没有写入方而无限等待，随后用 `fstat` 拒绝非普通文件。本项目用于所附普通样例和其他稳定的普通文件，不是任意设备、恶意路径、持续变化文件的处理器，也不是安全的文件系统沙箱。 队列初始化会检查错误并撤销已初始化的部件。线程创建失败有正常的回收路径。运行过程中若互斥锁、条件变量、join 或销毁操作意外失败，则采用明确的快速失败策略：打印错误并结束整个进程，不承诺完整报告或有序清理。pthread 函数直接返回错误码，不能用 `errno` 解读。关闭失败会报告，不会在描述符状态不确定时盲目重试。 本项目没有线程取消、信号驱动退出或单文件时间限制。慢文件系统仍可能拖延普通文件读取；连续信号也可能不断中断读取。测试给子进程设置超时，但程序本身不保证运行时间上限。要得到可重复的计数，输入文件在扫描期间不能改变。原生 Windows 不属于本 POSIX 目标。能检测到的标准输出错误会返回失败；管道读端关闭时，也可能按默认 SIGPIPE 行为直接结束进程。 计数工作量为 O(B + J)，B 为扫描字节数，J 为任务数。队列操作除等待外只需常数工作和固定存储。队列只放四个编号，结果数组最多 128 项，工作线程最多 16 个，每个线程有一个 4096 字节读取缓冲区以及系统管理的线程栈。内存不会随文件长度增长。判断文件超限时可能额外读取一个缓冲区。小样例只能验证正确性，不能证明多线程更快；存储竞争和线程启动开销反而可能让它更慢。 ## 测试和定位问题 ```sh make test python3 test.py --tsan make clean ``` `make test` 始终在临时副本中编译和运行，即使源码目录已有本地构建产物也一样。不会修改样例。每个子进程都有超时。可选的第二条命令要求 Clang 的 ThreadSanitizer 编译器支持和运行环境可用；不可用或检测失败都会报错，不会假装通过。 测试覆盖精确样例输出、零任务、空文件、单线程和十六线程、任务数少于与多于队列容量、任务数上限、非法线程数、缺失文件、目录与 FIFO、跨缓冲区单词、二进制字节，以及文件大小上限的两侧。独立的队列探针检查先进先出、关闭后排空和关闭后拒绝入队。另有 48 次重复运行，使用 37 个有界合成文件以及 1、2、4、16 个工作线程，分别与独立字节/正则表达式算法和单线程输出比较。 测试还在临时目录编译专用 C 替身，注入第一个线程创建前失败、成功创建两个线程后失败、读取失败，以及一次中断读取。这能检验错误路径而不耗尽系统资源。编译期替换只存在于测试副本，不是正式程序的运行开关。压力测试和 ThreadSanitizer 能发现问题，但都不能证明所有调度正确；仍需要不变量与同步关系的论证。 零任务运行卡住时，检查关闭广播和 `count == 0 && !closed` 条件。末尾无换行的行漏算时，检查 EOF 后补计。输出顺序变化时，检查是否在工作线程中打印，或是否漏了 join。4096 字节边界把单词拆开时，检查 `in_word` 是否错误地放到读取循环内部初始化。缺失文件被打印成零计数时，检查打印计数前是否判断了错误状态。 ## 文件导览与后续实验 `main.c` 管理命令行、线程启动、任务发布、join 和报告。`queue.h` 声明有界状态，`queue.c` 实现受锁保护的状态变化。`stats.h` 定义结果与错误，`stats.c` 扫描单个普通文件。`Makefile` 提供严格编译、测试和清理目标。`test.py` 包含独立算法、重复运行和故障注入。两个 README 用不同语言介绍同一个项目。`fixtures/` 的五个文件都是有意设计边界情况的合成输入。 第一个扩展可以把队列容量改成有界参数，并验证容量为 1、4、8 时结果不变。第二个扩展可以对固定、较大的本地样例做性能实验：记录总字节数、编译选项、线程数、耗时、重复次数和文件缓存条件，用字节数除以秒数得到吞吐量。只有测量支持时才能声称提速。时间信息应放在确定性统计输出之外。 ## 官方资料 教学解释均为原创，下列官方 POSIX 页面用于核对 API 约定： * [条件等待与条件重新检查](https://pubs.opengroup.org/onlinepubs/9799919799/functions/pthread_cond_clockwait.html) * [pthread_join](https://pubs.opengroup.org/onlinepubs/9799919799.2024edition/functions/pthread_join.html) * [第 4.15.2 节：内存同步](https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/V1_chap04.html) * [pthread_create：Issue 6 的 API 约定](https://pubs.opengroup.org/onlinepubs/000095399/functions/pthread_create.html) * [read](https://pubs.opengroup.org/onlinepubs/9799919799/functions/read.html)