读走ab以后，cde去哪了？
之前读pipe.c，我在截图旁边记了这些话：从读端读出多少字节、向管道追加多少字节、关闭时清理、poll检查是否就绪。
这些说明能帮我们认出函数，却还不能回答一个很具体的问题：写入abcde，第一次只读走ab，第二次还能不能读到cde？
先沿着这五个字节往下走。原稿截图用的是线性缓冲区；本次固定两个历史提交对照，不把当前工作树的分页缓冲实现混进来。
- 修订版：
9fbddc5a547580462f38ed20c95c1df23b183782，本文读写主线使用这个版本。 - 旧读函数对照：
97c816e9e62c57f3ccb6636993aaa9c5135f1633，包含与旧图相同的那条错误搬移语句，但不是在宣称它就是所有截图的精确版本。
两者路径都是kernel/src/fs/vfs/pipe.c，对应头文件为同目录的pipe.h。本文从本地 Git 对象读取固定提交，不依赖当前工作树；附录直接保留测试所用原函数。我们做的是宿主进程中的函数验证，没有启动这两个版本的 naos 内核。
两个端点，数据放在哪边？
先看创建管道的sys_pipe。这个版本会分配共享的pipe_info_t，再创建读、写两个pipe_specific_t，二者都指向同一个info。
不要把两个文件描述符理解为各有一份独立数据：
读端fd -> 读端spec --+ | v 同一个info buf + ptr ^ | 写端fd -> 写端spec --+这里的buf指向存放字节的内存，ptr表示当前有效字节数，不是C指针。开始ptr=0。写函数从buf[ptr]处追加，读取则从buf[0]开始。
修订版pipe.h的PIPE_BUFF是512 * 1024，原图旁注里的128KiB不能继续当它的容量。这个宏也不是 POSIX 中表示写入原子性界限的PIPE_BUF，二者不要因为名字像就混用。
写进去不难，读一半以后怎么办？
初始状态和第一次写入：
开始：ptr=0 buf: [ 空闲区域...... ] 写入abcde后：ptr=5 下标: 0 1 2 3 4 buf: [a][b][c][d][e] ...pipe_write_inner在空间足够时做的就是两步：
memcpy(&pipe->buf[pipe->ptr], addr, size); pipe->ptr += size;现在请求读取2字节。实际读取量取请求长度和已有数据量的较小值，所以to_read=2。把前两个字节拷给调用者，拿到ab。
但事情没结束。这个实现规定下一次还是从buf[0]读，如果只把ptr从5改成3，却不搬数据，下次读到的仍然从a开始。
所以未读部分必须前移：
读前：ptr=5 [a][b] | [c][d][e] 已读 | 未读 交给调用者：ab 把右侧cde搬到开头： 读后：ptr=3 [c][d][e] | 后面不算有效数据修订版的核心顺序是：
memcpy(addr, pipe->buf, to_read); memmove(pipe->buf, &pipe->buf[to_read], pipe->ptr - to_read); pipe->ptr -= to_read;搬移源地址是buf + to_read，长度是原有效长度减去读走的长度。memmove允许源区域和目标区域重叠，这里剩余字节往前移动，正好需要这一点。
ptr减少以后，只把前3字节视为有效数据；后面的旧字节不需要全部清零。再写入xy时，从下标3追加，逻辑数据就变成cdexy。
原图里的那一行，为什么不能照着解释？
旧图和旧提交里出现的是：
memmove(pipe->buf, &pipe->buf[pipe->ptr], pipe->ptr - to_read);代入刚才的数，源地址不是下标2，而是下标5：
下标: 0 1 2 3 4 5 6 7 buf: [a][b][c][d][e][0][0][0] 应搬：[c][d][e]，从下标2开始 旧式：[0][0][0]，从下标5开始 旧式搬3字节：[0][0][0] ptr仍然减到3，cde却已经不在前面这里把实验中的后续缓冲区初始化成了0，所以看到的是三个零字节；不是说真实系统中那里永远为0。关键是它越过了我们要保留的未读部分。
测试实际调用旧读函数后，第一次确实得到ab，但剩余区检查失败：
PARTIAL_READ_LOST_TAIL: ptr=3 bytes=00,00,00仅检查第一次返回值是2，甚至检查第一次内容是ab，都发现不了这个问题。必须接着检查未读字节，或者再读一次。
这也是读源码时不能只把memmove翻译成“移动数据”的原因：需要把源下标、目标下标和长度放回同一组数里，看它究竟移了哪一段。
有锁，不代表检查就可以放在锁外
修订版的读取大致按这个顺序：取得pipe->lock，检查有没有数据，有就完成拷贝、搬移与计数更新，再释放锁返回。
如果先在锁外看到ptr=5，等真正拿到锁时，另一个读者可能已经把5字节都读走了。此时还拿之前算好的数量拷贝，依据的就不是当前状态。
所以“看有没有数据”和“根据这个数量取走数据”不能随意拆开。锁不仅保护那条memcpy，还要保护决定这次操作是否合法的状态。
空的时候又不能一直拿着锁等。我们希望写端往里放数据，写端也需要同一把锁；读者若占着它不放，写者就进不来。
这个版本的空读分支在放锁后检查非阻塞标志，普通等待路径再调用调度函数，之后回到循环，重新加锁、重新检查：
读者：加锁 -> 空 -> 放锁 | v 读者让出执行 写者：加锁 -> 写abc -> 放锁 | v 读者：加锁 -> 重查 -> 读abc本次测试把“让出执行”替换成一个受控钩子：检查锁已释放，调用原写函数写入abc，再让原读函数继续，最后确认读到了它。这检验了这个顺序能否走通，不能代替真实多核调度的竞争测试。
这里是让出执行后再检查的历史实现，不应直接把它描述成已经实现了完整等待队列和定向唤醒。
写满以后，大请求怎么继续？
写函数判断的是PIPE_BUFF - ptr能不能装下这一块。如果装不下，就先放锁、让出执行，之后重试。
外面的pipefs_write会把超过容量的请求拆块。设容量为 B，请求 B+3字节：
先写B字节：ptr=B，空间满 还有3字节：装不下，放锁并让出执行 读者取走B字节：ptr=0 写者再检查：现在能放3字节 最后写完：ptr=3测试按原容量512KiB实际运行这条路径，不把图里的B偷偷改成一个小数组再声称测过原函数。受控钩子用原读函数取出第一块，检查前B字节完整；写函数继续后，再读末尾3字节，检查整段顺序没有乱。
内层如果返回负错误码，外层必须先判断，不能直接把负值加进“已经写了多少”的计数。本文固定的修订版在增加cycle前有wrote < 0分支。不过它并不自动等于完整的 POSIX 部分写入语义；跨块已经写了一部分后又遇到错误怎样返回，仍需单独审查。
关闭写端，为什么不马上抹掉缓冲区？
info还记录读、写端计数。静态看这个版本的关闭函数：关闭一端会减少对应计数；两边都为0时才释放共享缓冲区和info。这解释了为什么不能一关写端就把读者还没取走的数据删掉。
读函数也是先处理现有数据，空了再判断是否还有写端。测试里设置“缓冲区有xy，写端计数为0”，先读到xy，下一次才返回0。poll同时报告可读与挂断，也不矛盾：还剩数据可读，但写端已消失。
这里验证的是读和poll在给定计数下的分支；没有在宿主里执行真实的VFS关闭、dup、fork引用计数链。不能因为这几个函数通过，就宣布整个端点生命周期已正确。
Linux对外的EOF、EPIPE规则可查 pipe手册。本文测试的函数直接返回内核负错误码，不是用Linux的read()来替代naos实现，也不把这些返回值冒充成 libc 的-1 + errno。
附：完整验证程序
下面的fixed-lab.c包含固定修订版原样提取的四个函数：pipefs_read、pipe_write_inner、pipefs_write、pipefs_poll。外围结构和锁、调度钩子是宿主适配，不是完整内核。源码项目使用 GPL-3.0，原函数来源及提交已列在前面，实验目录保留原始文件、摘要与LICENSE；复制和分发衍生测试应保留来源并遵守该许可。
原工作树已有的RISC-V与系统调用修改没有被本次实验改动。若想在仓库里核对版本，可运行git show 提交号:kernel/src/fs/vfs/pipe.c，不要为了读历史代码直接切换或清理自己的工作树。
本次宿主环境为Arch Linux、x86_64、GCC16.1.1。在有GCC和Python3的Linux环境，把下面三份文件放到同一专用目录，执行python3 run.py。使用 GNU C11，因为历史函数含GNU风格的指针运算。编译启用 AddressSanitizer 与 UBSan；环境不支持时不要简单删掉选项后声称同样覆盖。
fixed-lab.c：
/* SPDX-License-Identifier: GPL-3.0-only * Historical core functions: na-kernel, commit 9fbddc5a547580462f38ed20c95c1df23b183782. * Host-side single-thread harness. Scheduling is replaced by explicit hooks. */ #include <assert.h> #include <errno.h> #include <fcntl.h> #include <stdbool.h> #include <stdint.h> #include <stdio.h> #include <stdlib.h> #include <string.h> #include <sys/epoll.h> #include <sys/types.h> #define MIN(a, b) ((a) < (b) ? (a) : (b)) #define SCHED_FLAG_YIELD 1 #define PIPE_BUFF (512 * 1024) typedef struct { int held; } spinlock_t; typedef struct { uint32_t ptr; char *buf; int write_fds, read_fds; spinlock_t lock; } pipe_info_t; typedef struct node { void *handle; } *vfs_node_t; typedef struct { bool write; pipe_info_t *info; vfs_node_t node; } pipe_specific_t; typedef struct { vfs_node_t node; int flags; } fd_t; static pipe_info_t state; static pipe_specific_t spec; static struct node node; static fd_t fd; static int action, yields; static char *drained; static void spin_lock(spinlock_t *p) { assert(!p->held); p->held = 1; } static void spin_unlock(spinlock_t *p) { assert(p->held); p->held = 0; } static void arch_enable_interrupt(void) {} static void arch_disable_interrupt(void) {} static void schedule(int flags); static void arch_yield(void) { schedule(1); } ssize_t pipefs_read(fd_t *fd, void *addr, size_t offset, size_t size) { if (size > PIPE_BUFF) size = PIPE_BUFF; void *file = fd->node->handle; pipe_specific_t *spec = (pipe_specific_t *)file; if (!spec) return -EINVAL; pipe_info_t *pipe = spec->info; if (!pipe) return -EINVAL; if (!size) return 0; while (true) { spin_lock(&pipe->lock); if (pipe->ptr > 0) { // 实际读取量 uint32_t to_read = MIN(size, pipe->ptr); memcpy(addr, pipe->buf, to_read); memmove(pipe->buf, &pipe->buf[to_read], pipe->ptr - to_read); pipe->ptr -= to_read; spin_unlock(&pipe->lock); return to_read; } if (pipe->write_fds == 0) { spin_unlock(&pipe->lock); return 0; } spin_unlock(&pipe->lock); if (fd->flags & O_NONBLOCK) { return -EWOULDBLOCK; } arch_enable_interrupt(); schedule(SCHED_FLAG_YIELD); arch_disable_interrupt(); } } ssize_t pipe_write_inner(void *file, const void *addr, size_t size) { pipe_specific_t *spec = (pipe_specific_t *)file; pipe_info_t *pipe = spec->info; while (true) { spin_lock(&pipe->lock); if (pipe->read_fds == 0) { spin_unlock(&pipe->lock); return -EPIPE; } if ((PIPE_BUFF - pipe->ptr) >= size) { memcpy(&pipe->buf[pipe->ptr], addr, size); pipe->ptr += size; spin_unlock(&pipe->lock); return size; } spin_unlock(&pipe->lock); schedule(SCHED_FLAG_YIELD); } } ssize_t pipefs_write(fd_t *fd, const void *addr, size_t offset, size_t size) { int ret = 0; void *file = fd->node->handle; const char *data = addr; size_t chunks = size / PIPE_BUFF; size_t remainder = size % PIPE_BUFF; if (chunks) for (size_t i = 0; i < chunks; i++) { int cycle = 0; while (cycle < PIPE_BUFF) { ssize_t wrote = pipe_write_inner( file, data + i * PIPE_BUFF + cycle, PIPE_BUFF - cycle); if (wrote < 0) return wrote; cycle += wrote; } ret += cycle; } if (remainder) { size_t cycle = 0; while (cycle < remainder) { ssize_t wrote = pipe_write_inner( file, data + chunks * PIPE_BUFF + cycle, remainder - cycle); if (wrote < 0) return wrote; cycle += wrote; } ret += cycle; } return ret; } int pipefs_poll(void *file, size_t events) { pipe_specific_t *spec = (pipe_specific_t *)file; pipe_info_t *pipe = spec->info; int out = 0; spin_lock(&pipe->lock); if (events & EPOLLIN) { if (!pipe->write_fds) out |= EPOLLHUP; if (pipe->ptr > 0) out |= EPOLLIN; } if (events & EPOLLOUT) { if (!pipe->read_fds) out |= EPOLLHUP; if (pipe->ptr < PIPE_BUFF) out |= EPOLLOUT; } spin_unlock(&pipe->lock); return out; } static void reset(void) { memset(state.buf, 0, PIPE_BUFF); state.ptr = 0; state.write_fds = state.read_fds = 1; state.lock.held = 0; fd.flags = 0; action = yields = 0; } static void schedule(int flags) { (void)flags; assert(!state.lock.held); assert(++yields <= 2); #ifndef LEGACY_ONLY if (action == 1) { action = 0; assert(pipe_write_inner(&spec, "abc", 3) == 3); } else if (action == 2) { action = 0; assert(pipefs_read(&fd, drained, 0, PIPE_BUFF) == PIPE_BUFF); } else { assert(!"unexpected scheduling hook"); } #else assert(!"legacy partial-read test must not wait"); #endif } int main(void) { state.buf = calloc(1, PIPE_BUFF); drained = malloc(PIPE_BUFF); assert(state.buf && drained); spec.info = &state; spec.node = &node; node.handle = &spec; fd.node = &node; reset(); memcpy(state.buf, "abcde", 5); state.ptr = 5; char out[40] = {0}; assert(pipefs_read(&fd, out, 0, 2) == 2); assert(!memcmp(out, "ab", 2)); if (state.ptr != 3 || memcmp(state.buf, "cde", 3)) { fprintf(stderr, "PARTIAL_READ_LOST_TAIL: ptr=%u bytes=%02x,%02x,%02x\n", state.ptr, (unsigned char)state.buf[0], (unsigned char)state.buf[1], (unsigned char)state.buf[2]); free(drained); free(state.buf); return 2; } assert(pipefs_read(&fd, out, 0, 3) == 3); assert(!memcmp(out, "cde", 3)); #ifndef LEGACY_ONLY unsigned cases = 0; for (unsigned n = 0; n < 32; n++) { for (unsigned request = 0; request <= n + 3; request++) { reset(); for (unsigned i = 0; i < n; i++) state.buf[i] = (char)(i + 1); state.ptr = n; state.write_fds = 0; unsigned taken = MIN(n, request); assert(pipefs_read(&fd, out, 0, request) == taken); assert(state.ptr == n - taken && !state.lock.held); for (unsigned i = 0; i < taken; i++) assert(out[i] == (char)(i + 1)); for (unsigned i = 0; i < state.ptr; i++) assert(state.buf[i] == (char)(taken + i + 1)); cases++; } } reset(); fd.flags = O_NONBLOCK; assert(pipefs_read(&fd, out, 0, 1) == -EWOULDBLOCK); state.write_fds = 0; assert(pipefs_read(&fd, out, 0, 1) == 0); reset(); action = 1; assert(pipefs_read(&fd, out, 0, 3) == 3); assert(!memcmp(out, "abc", 3) && yields == 1 && !state.lock.held); reset(); state.read_fds = 0; assert(pipe_write_inner(&spec, "x", 1) == -EPIPE); reset(); char *input = malloc(PIPE_BUFF + 3); assert(input); for (unsigned i = 0; i < PIPE_BUFF + 3; i++) input[i] = (char)(i % 251); action = 2; assert(pipefs_write(&fd, input, 0, PIPE_BUFF + 3) == PIPE_BUFF + 3); assert(yields == 1 && state.ptr == 3); assert(!memcmp(drained, input, PIPE_BUFF)); assert(pipefs_read(&fd, out, 0, 3) == 3); assert(!memcmp(out, input + PIPE_BUFF, 3)); free(input); reset(); memcpy(state.buf, "xy", 2); state.ptr = 2; state.write_fds = 0; assert((pipefs_poll(&spec, EPOLLIN) & (EPOLLIN | EPOLLHUP)) == (EPOLLIN | EPOLLHUP)); assert(pipefs_read(&fd, out, 0, 2) == 2); assert(pipefs_read(&fd, out, 0, 1) == 0); assert(pipefs_poll(&spec, EPOLLIN) == EPOLLHUP); printf("HOST_CHECKS_PASSED partial_read_cases=%u capacity=%u\n", cases, PIPE_BUFF); #endif free(drained); free(state.buf); return 0; }old-read.c，只取旧版本读函数来复现尾部丢失，不拿它代表旧内核全部行为：
ssize_t pipefs_read(fd_t *fd, void *addr, size_t offset, size_t size) { if (size > PIPE_BUFF) size = PIPE_BUFF; void *file = fd->node->handle; pipe_specific_t *spec = (pipe_specific_t *)file; if (!spec) return -EINVAL; pipe_info_t *pipe = spec->info; if (!pipe) return -EINVAL; while (pipe->ptr == 0) { if (fd->flags & O_NONBLOCK) { return -EWOULDBLOCK; } if (pipe->write_fds == 0) { return 0; } arch_yield(); } // 实际读取量 uint32_t to_read = MIN(size, pipe->ptr); if (to_read == 0) { return 0; } spin_lock(&pipe->lock); memcpy(addr, pipe->buf, to_read); memmove(pipe->buf, &pipe->buf[pipe->ptr], pipe->ptr - to_read); pipe->ptr -= to_read; spin_unlock(&pipe->lock); return to_read; }run.py：
import json import os import resource from pathlib import Path import subprocess import tempfile base = Path(__file__).resolve().parent resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) run = Path(tempfile.mkdtemp(prefix='host-run-', dir=base)) env = dict(os.environ, ASAN_OPTIONS='detect_leaks=1:abort_on_error=1') flags = ['gcc', '-std=gnu11', '-Wall', '-Wextra', '-Werror', '-Wno-unused-parameter', '-Wno-unused-function', '-g', '-O1', '-fsanitize=address,undefined', '-fno-omit-frame-pointer'] records = [] def check(name, source, expected, extra=()): file = run / (name + '.c') file.write_text(source) exe = run / name compile = subprocess.run(flags + list(extra) + [str(file), '-o', str(exe)], capture_output=True, text=True, timeout=30) assert compile.returncode == 0, compile.stderr result = subprocess.run([str(exe)], capture_output=True, text=True, env=env, timeout=15) if expected == 'lock-assertion': assert result.returncode != 0 and 'Assertion' in result.stderr, result assert '!p->held' in result.stderr, result.stderr else: assert result.returncode == expected, result.stderr if expected == 0: assert 'partial_read_cases=624' in result.stdout assert not result.stderr, result.stderr if expected == 2: assert 'PARTIAL_READ_LOST_TAIL' in result.stderr assert 'bytes=00,00,00' in result.stderr records.append({'name': name, 'exit': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr}) fixed = (base / 'fixed-lab.c').read_text() old_read = (base / 'old-read.c').read_text().strip() start = fixed.index('ssize_t pipefs_read(') end = fixed.index('\nssize_t pipe_write_inner(', start) old = (fixed[:start] + old_read + '\n' + fixed[end:]).replace( '#define PIPE_BUFF (512 * 1024)', '#define PIPE_BUFF 131072', 1) check('fixed', fixed, 0) check('historical-read', old, 2, ['-DLEGACY_ONLY']) wrong_tail = fixed.replace('&pipe->buf[to_read]', '&pipe->buf[pipe->ptr]') assert wrong_tail != fixed check('wrong-tail-mutant', wrong_tail, 2) locked_return = fixed.replace('spin_unlock(&pipe->lock);\n\n if (fd->flags', '/* intentionally broken: lock retained */\n\n if (fd->flags', 1) assert locked_return != fixed check('retained-lock-mutant', locked_return, 'lock-assertion') result = {'passed': True, 'cases': 624, 'records': records, 'gcc': subprocess.check_output(['gcc', '--version'], text=True).splitlines()[0], 'architecture': os.uname().machine, 'scope': 'Unmodified selected kernel functions; host stubs and scripted scheduling, no kernel boot.'} (run / 'results.json').write_text(json.dumps(result, indent=2)) print('RESULTS=' + str(run / 'results.json'))除了abcde这个反例，测试还逐一检查已有0至31字节、不同读取长度的624组组合；检查非阻塞空读、无写端的空读、受控读等待、无读端写入、跨容量分块和poll状态。旧读函数应被反例拒绝，两个故意破坏的修订版也应失败，不能把每个进程都返回0当成通过条件。
这些检查回答的是选定函数在受控状态下有没有保住字节与检查顺序。真实内核的用户地址拷贝、信号、抢占、多读写者和完整资源回收，还要在对应内核环境中继续验证。
