?(...)?

UnionFS OverlayFS

是什么 UnionFS 即聚合文件系统，是一种虚拟文件系统，用于将多个文件系统聚合为一个挂载点主要实现有 AUFS 和 OverlayFS 两种，其中前者已被主流抛弃，目前 OverlayFS 才是绝对主流（已并入 Linux 主线内核，是 Docker 镜像的默认文件系统） OverlayFS 分为上下两层（upper, lower）其中 upper 为可写层，lower 为只读层 # 准备实验环境 $ mkdir {upper,lower,merged,work} $ echo 1 > ./lower/a $ tree . . ├── lower │ └── a ├── merged ├── upper └── work $ sudo mount -t overlay overlay -o lowerdir=./lower,upperdir=./upper,workdir=./work ./merged $ tree . . ├── lower │ └── a ├── merged │ └── a ├── upper └── work ├── index [error opening dir] └── work [error opening dir] $ cat lower/a 1 $ cat merged/a 1 可以看到 merged 后的目录与 lower 目录完全一致，如果这个时候我们修改 merged/a 的值呢？ ...

My Docker 2

今日目标基于 Cgroup 实现资源控制如何实现参考 k8s 官方文档, 有以下两种驱动 cgroupfs systemd cgroupfs 直接操作 /sys/fs/cgroup 下的虚拟文件接口, 而 systemd 则通过 D-Bus 总线与 systemd 通信，将操作 /sys/fs/cgroup 的任务委托给它为了降低项目复杂度，目前只计划实现 cgroupfs 驱动且不考虑适配 Cgroup v1 将 Cgroup 控制的资源抽象为接口 type Resource interface { // CgroupFile 返回 cgroup v2 控制器文件名（如 "cpu.weight"）。 CgroupFile() string // CgroupValue 返回要写入文件的格式化字符串。 CgroupValue() string // Validate 做边界检查，失败返回 ErrInvalidResource。 Validate() error } 每一种资源最终都会变为字符流(CgroupValue())写入 cgroup 叶子节点对应的文件(CgroupFile()) 内存 // MemoryLimit 单位是字节，对应 cgroup v2 的 memory.max。 // 不允许 0（v2 中 0 = "max" 即无限制），用户想解除限制就别传 -m。 type MemoryLimit int64 func (m MemoryLimit) CgroupFile() string { return "memory.max" } func (m MemoryLimit) CgroupValue() string { return strconv.FormatInt(int64(m), 10) } func (m MemoryLimit) Validate() error { if m <= 0 { return fmt.Errorf("%w: --memory must be > 0 bytes, got %d", ErrInvalidResource, m) } return nil } CPU占用率 // CPUMax 描述 cpu.max 的一行：每周期的 CPU 时间配额和周期长度。 // 对应 cgroup v2 cpu.max 文件格式 "$MAX $PERIOD"（单位均为微秒）。 type CPUMax struct { Quota uint64 // 0 表示 "max"（不限制） Period uint64 // 周期长度（微秒） } func (c CPUMax) CgroupFile() string { return "cpu.max" } func (c CPUMax) CgroupValue() string { if c.Quota == 0 { return fmt.Sprintf("max %d", c.Period) } return fmt.Sprintf("%d %d", c.Quota, c.Period) } func (c CPUMax) Validate() error { if c.Period == 0 { return fmt.Errorf("%w: cpu-max period must be > 0", ErrInvalidResource) } return nil } 驱动接口 // Driver 是 cgroup 后端抽象。cgroupfs 和 systemd 各自实现一份。 // // 接口维度（不是底层路径）： // - Create: 申请容器所属的 cgroup（scope 对 systemd 来说是 lazy：在 AddPID 时才创建） // - Apply: 写入 Resource 指定的资源限额 // - AddPID: 把进程加进 cgroup；host PID 而不是容器内 PID // - Remove: 销毁 cgroup；可恢复错误（残留进程）记 warning 不 fatal type Driver interface { Create(id, parent string) error Apply(id, parent string, resources ...Resource) error AddPID(id, parent string, pid int) error Remove(id, parent string) error Name() string } cgroupfs 实现 ...

My Docker 1

今日任务 init 传参改用匿名管道 init 传参改用匿名管道匿名管道为什么要使用匿名管道原方案通过 /proc/self/exe init [args...] 将所有参数直接塞进 exec.Command() 调用中，如果参数过长或者有特殊符号就会发生错误。匿名管道的读和写直接走文件io，可以加强对参数的控制和校验 // run func NewParentProcess(interactive, tty bool) (*exec.Cmd, *os.File) { // 返回 Cmd 对象和管道的写对象 fr, fw, err := os.Pipe() // 创建匿名管道 if err != nil { l.Error("failed to create pipe", "detail", err) return nil, nil } cmd := exec.Command("/proc/self/exe", "init") // 不需要在此处将 args 传入 cmd.SysProcAttr = &syscall.SysProcAttr{ Cloneflags: syscall.CLONE_NEWUTS | syscall.CLONE_NEWPID | syscall.CLONE_NEWNS | syscall.CLONE_NEWNET | syscall.CLONE_NEWIPC, } if interactive { cmd.Stdin = os.Stdin } if tty { cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr } cmd.ExtraFiles = append(cmd.ExtraFiles, fr) // 将管道的读文件描述符放进新创建进程的文件描述符数组中, 新的进程通过这个fd可以打开对应的管道文件 return cmd, fw } // init const readingFd uintptr = 3 // 不固定 func Init() error { l.Info("initiating") l.Info("opening pipe") f := os.NewFile(readingFd, "pipe") // 打开对应的管道 if f == nil { l.Error("failed to create FILE with the given fd", "fd", readingFd) return fmt.Errorf("failed to create FILE with the given fd: %d", readingFd) } l.Info("reading from pipe") data, err := io.ReadAll(f) // 直接走文件io if err != nil { l.Error("failed to read from pipe", "detail", err) return err } // 将参数反序列化 // // 用 NUL 分隔符还原 cmds,避免含空格的参数(如 sh -c "echo hi")被腰斩 // 使用管道传参可以完成更多的操作 args := strings.Split(strings.TrimRight(string(data), "\x00"), "\x00") if len(args) < 1 || args[0] == "" { return fmt.Errorf("command is nil") } l.Info("remounting `/`") if err := syscall.Mount("", "/", "", syscall.MS_PRIVATE|syscall.MS_REC, ""); err != nil { l.Error("failed to remount `/` with `MS_PRIVATE`", "detail", err) return err } mountFlags := syscall.MS_NOEXEC | syscall.MS_NOSUID | syscall.MS_NODEV if err := syscall.Mount("proc", "/proc", "proc", uintptr(mountFlags), ""); err != nil { l.Error("failed to mount proc", "detail", err) return err } l.Info("executing", "argv0", args[0], "argv", args) if err := syscall.Exec(args[0], args, os.Environ()); err != nil { l.Error("failed to init", "cmds", args, "detail", err) return err } return nil }

匿名管道

本质存在于内核内存中的虚拟文件系统，属于进程的资源，内部使用引用计数特性数据单向流动可跨进程大小可以通过以下命令查看 $ cat /proc/sys/fs/pipe-max-size 1048576 # 单位 byte, 1 MB 可以动态调整阻塞/非阻塞阻塞模式下，读/写一个写/读端被关闭的管道会直接返回非阻塞模式下，读/写立即返回如何创建 A -> B A 进程创建管道， fork 出子进程 B fork 会让主进程和子进程共用同一个fd数组，此时A和B实现了进程间通信为了保证管道中数据的单向流动, A 关闭 fdr(读), B 关闭 fdw(写) go 中如何创建 fr, fw, err := os.Pipe() // 创建 if n, err := fw.WriteString(strings.Join(cmds, "\x00")); err != nil {} // 写入 _ = fw.Close() // 写完一定要及时关闭，防止接收方堵塞 // 读取 // // 需要打开管道 // readingFd 为进程中管道文件描述符在fd数组中的索引 f := os.NewFile(readingFd, "pipe") if f == nil { } data, err := io.ReadAll(f) if err != nil { }

My Docker 0

仿造一个自己的 docker 今日目标 mydocker run -i -t <command> 基础隔离启动流程 mydocker run -i -t <command> -> 解析 -> 启动 init 进程 -> 重新挂载 / 为 PRIVATE -> 挂载 proc -> 执行(Exec 系统调用) <command> 为什么需要 init 进程 init 进程有独立的命名空间(namespace)和控制组(cgroup)，执行系统调用(Exec)后可以保证 <command> pid 为 1并且与宿主机隔离为什么要重新挂载 / 为 PRIVATE 使用以下代码验证 func Init(cmdPath string) error { mountFlags := syscall.MS_NOEXEC | syscall.MS_NOSUID | syscall.MS_NODEV if err := syscall.Mount("proc", "/proc", "proc", uintptr(mountFlags), ""); err != nil { l.Error("failed to mount proc", "detail", err) return err } if err := syscall.Exec(cmdPath, []string{cmdPath}, os.Environ()); err != nil { l.Error("failed to init", "cmdPath", cmdPath, "detail", err) return err } return nil } $ sudo ./cli run -i -t /bin/sh sh-5.3# ps PID TTY TIME CMD 1 pts/5 00:00:00 sh 6 pts/5 00:00:00 ps sh-5.3# 可以看到容器内 ps 命令执行没有报错并且返回内容符合预期(<command>作为容器内的 init 进程) ...

发信箱模式 Outbox Pattern

解决了什么问题？分布式系统通过消息队列等手段互相解耦带来的状态不一致, 如数据落库成功但是kafka消息发送失败等情况是什么？利用本地数据库事务的原子性，将数据入库和消息入库(outbox-records)变为原子性操作。异步启动一个 poller() worker 定时从 outbox-records 中取出消息尝试发送，消息至少被成功发送一次, 保证了分布式系统数据的一致性。问题高并发场景下会导致更高的延迟消息接收端需要做幂等性校验发送失败/超时不能自动回退数据库操作使用 PostgreSQL 的行锁通过给 Outbox 表的查询语句加上行锁 SELECT id, event, payload, topic FROM outbox FOR UPDATE SKIP LOCKED 就可以让多个 poller 同时抢占发信任务，提高效率的同时也不会导致竞争 SKIP LOCKED 是 pg 独有的拓展，语义是跳过已经上锁的行

Golang 多路复用

快速路径运用 select 中的 default 分支可以实现无锁（非阻塞）的快速检查 select { case v, ok := <- ch: if !ok { // ch 已经被关闭, 直接返回 return io.EOF } return v // ch 缓冲区中有内容, 直接取出返回 default: // ch 缓冲区中无内容，select 走 default 分支防止阻塞 } select { // 阻塞等待 case v, ok := <- ch: if !ok { // ch 已经被关闭, 直接返回 return io.EOF } return v // ch 缓冲区中有内容, 直接取出返回 case <- ctx.Done(): // 等待超时/取消 return ctx.Err() } 在阻塞等待之前加入快速路径主要是因为 golang 中的 select 具有随机性，在高并发场景下可能会出现 <- ctx.Done() 分支永不被进入。详细的源码解析见下文 select 语句源码 #filepath: src/runtime/select.go // case 对应的结构体 type scase struct { // chan -> case 关联的 channel c *hchan // data element // 发送: 待发送数据的源地址 (case ch <- v elem 指向 v 源地址) // 接收: 目标变量的地址（堆、栈）(case v := <- ch elem 指向 v 的地址) elem unsafe.Pointer } #filepath: src/runtime/select.go // select 语句经编译器编译后被转化为 selectgo 的函数调用 func selectgo( cas0 *scase, // scase 数组的首地址，存放在 goroutine 栈上 order0 *uint16, // [2*ncases]uint16 数组首地址，也分配在栈上 // 为什么有了 cas0 还需要 order0? 详见下文 pc0 *uintptr, // (race detector 构建) [ncases]uintptr 数组首地址；否则 nil；与本章节无关 nsends int, // 发送 case 的数量 nrecvs int, // 接收 case 的数量 block bool, // true=阻塞（无 default），false=非阻塞（有 default） ) (int, bool) // 返回值：选中的 case 索引 + 是否成功接收到值 // NOTE: In order to maintain a lean stack size, the number of scases // is capped at 65536. // 将裸指针转换为具有最大合法边界的数组 // 防止后续写入栈外内存 cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0)) order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0)) ncases := nsends + nrecvs scases := cas1[:ncases:ncases] // array[:end:max] 限制新切片最大容量为 ncases pollorder := order1[:ncases:ncases] // 轮询顺序注意：轮询顺序是随机的，并非顺序轮询，详见下文 lockorder := order1[ncases:][:ncases:ncases] // channel 加锁顺序 selecctgo 如何决定轮询顺序？ ...

Linux Cgroup

新版内核普遍采用 v2 版本的 cgroup 是什么 cgroup 既 Controll group, 是 linux 对外提供的文件接口，用来控制进程资源。 $ mount -t cgroup2 cgroup2 on /sys/fs/cgroup type cgroup2 (rw,nosuid,nodev,noexec,relatime,nsdelegate,memory_recursiveprot,memory_hugetlb_accounting) 使用新建 group 直接在 /sys/fs/cgroup 下新建目录 $ cd /sys/fs/cgroup/ $ sudo mkdir test $ ls ./test cgroup.controllers cpu.stat memory.peak cgroup.events cpu.stat.local memory.pressure cgroup.freeze cpu.uclamp.max memory.reclaim cgroup.kill cpu.uclamp.min memory.stat cgroup.max.depth cpu.weight memory.swap.current cgroup.max.descendants cpu.weight.nice memory.swap.events cgroup.procs ... 重点关注 cgroup.procs 这个文件，其中的内容(pid)既我们想要控制的进程号。控制cpu占用率 $ while true; do : ; done & [1] 50274 # 50274 为对应pid $ top -b -n 1 -p 50274 | grep "50274" | awk '{print $9}' 99.7 # 此时没有限制，死循环占用100%的cpu $ echo 50274 | sudo tee cgroup.procs # 直接使用当前shell的进程号, 这样子随后开启的子进程也会受到控制 $ echo 2000 10000 | sudo tee cpu.max # 10000 微秒内 `$$` 进程只能占用 2000 微秒 $ top -b -n 1 -p 50274 | grep "50274" | awk '{print $9}' 19.9 # 20% 控制内存测试程序 ...