SPOOLing假脱机模拟：输入输出井与Python实现

简介：这是一份围绕Spooling假脱机输入输出模拟技术的操作系统课程设计资料包，面向正在学习设备管理、进程同步与虚拟设备机制的高校学生及课程实验者，帮助其把抽象概念落到可运行的程序实现中。包内共11个文件，以Java源程序与编译后的class文件为主体，辅以一份设计报告doc和4张运行截图jpg，压缩包约279KB，规模轻量但结构完整。目前已有442人学习，适合用作课程设计参考或实验复盘。读者可结合源代码观察请求输出进程与Spooling输出进程如何协作，理解输入井、输出井和后台处理程序在缓冲调度中的职责，并借助报告梳理设计思路、实现方法与性能分析；截图则呈现初始化、数据输入、文件为空及运行界面等关键状态，便于对照程序行为、定位问题并加深对共享慢速I/O设备优化思路的理解。

1. 从打印室排队到磁盘井：spooling假脱机输入输出模拟技术到底在模拟什么

spooling假脱机输入输出模拟技术，本质上是在用磁盘上的缓冲区把一台独占设备伪装成多用户共享设备。某公司打印室门口贴过一张纸：每人先把文件提交到一个目录，再由一台机器按队列出纸。很多人以为这就是排队，真正写模拟器时才发现，难点不在队列，而在“输入井、输出井、输入进程、输出进程”四者之间的状态同步和容量控制。这个方向适合操作系统课程设计、系统编程练手，也适合需要理解缓冲、调度和并发控制的工程师。下面从四个对象的文件系统落地讲起，再给一个能跑的Python骨架和参数边界。

2. 输入井和输出井怎么落地：用文件系统模拟磁盘缓冲区的四个关键对象

2.1 输入井、输出井、输入进程、输出进程分别对应什么

SPOOLing系统里，输入井和输出井是磁盘上的缓冲区域，输入进程负责把慢速输入设备的数据提前读入输入井，输出进程负责把输出井里的数据按设备节奏送出。用户进程不再直接碰独占设备，只和井打交道，于是打印机这类设备就从“一次只能一个进程用”变成“多个进程可以先后提交、由系统统一调度”。模拟技术要做的，是把这套抽象映射到普通文件系统：目录当井，文件当块，JSON当作业控制块，线程当外围进程。

我在模拟项目X里一般会先画一张映射表，把每个概念落到可检查的对象上，否则后面排查时很容易把“井满”和“队列满”混在一起。

这张表的价值在于：每个参数都能在目录里找到对应证据。比如“井满”不是看内存计数器，而是看input_well和output_well下*.bin的总大小；比如“输出进程卡住”不是猜线程，而是看meta.json里状态是否停在PRINTING。

2.2 用目录树和元数据模拟“井”：结构设计

目录结构不要平铺，一个作业一个目录，作业目录里再放块文件和元数据。这样移动作业时只需要一次目录重命名，比逐个文件搬移更接近“作业整体入井、整体出井”的语义。下面是我常用的最小结构，spool_root是模拟根目录，sink/printer.out代表最终输出设备。

spool_root/ ├── input_well/ # 输入井：等待输入进程处理的作业 ├── output_well/ # 输出井：等待输出进程送到设备的作业 ├── done/ # 已完成作业，保留现场便于校验 ├── tmp/ # 提交中的临时目录，保证原子性 ├── sink/ │ └── printer.out # 模拟独占输出设备 └── source_jobs/ # 用户提交的源文件，仅用于演示

每个作业目录里长这样：

<input_well|output_well|done>/<job_id>/ ├── meta.json ├── chunk_0000.bin ├── chunk_0001.bin └── ...

meta.json至少要有job_id、name、size、block_size、blocks、priority、state、submitted_at、owner这些字段。状态机建议固定成IN_WELL -> OUTPUT_WELL -> PRINTING -> DONE，失败时进FAILED。状态不要只放在内存里，因为模拟器一重启，内存状态就没了，而井里的文件还在；把状态写进元数据，才能做恢复和排查。

注意：meta.json不要最后才写。一个常见做法是先把块文件写进tmp/<job_id>/，最后写meta.json，再用一次目录重命名提交到input_well/。这样输出进程永远不会看到“只有一半块”的作业。

2.3 从独占到共享：一次作业在井里的完整路径

一次提交的路径可以拆成五步。第一步，用户进程把源文件读成字节流，按block_size切块，写入tmp/<job_id>/chunk_XXXX.bin，同时生成元数据。第二步，tmp/<job_id>原子重命名到input_well/<job_id>，此时作业才算真正入井。第三步，输入进程按调度策略从输入井取作业，把状态改成OUTPUT_WELL，再把整个目录移动到输出井；这一步模拟的是“输入设备的数据已经进入缓冲区，等待输出设备消费”。第四步，输出进程从输出井取作业，把状态改成PRINTING，按块顺序写入sink/printer.out。第五步，写完后状态改成DONE，目录移到done/，同时释放井容量。

这里有个容易被忽略的细节：输入井和输出井都可以叫“井”，但它们的容量统计口径要一致。我一般只统计*.bin的字节数，不把meta.json算进去，因为元数据开销相对固定，混进去会让“井大小”这个参数变得不可解释。调度时也不要只按文件名排序，文件名是随机ID，排序没有业务含义；应该按priority和submitted_at组合排序，优先级数字小的先出，同优先级按提交时间先入先出。

3. 用Python跑通最小SPOOLing模拟器：从submit到假脱机输出的可运行骨架

3.1 环境准备与目录初始化

环境只需要Python标准库，不依赖第三方包。下面命令创建演示目录，source_jobs放几个待提交文件，spool_root是模拟器的根。运行前把旧目录删掉，避免上一次的done/干扰观察。

mkdir -p spool_demo/source_jobs cd spool_demo python3 --version

代码放在spool_demo/spooler_demo.py。核心类叫Spooler，它负责容量控制、原子提交、输入进程和输出进程。下面这段代码可以直接运行，重点看submit里的cv.wait和os.replace，以及output_worker里的sink_lock。

import os import json import time import uuid import threading from pathlib import Path from dataclasses import dataclass, asdict @dataclass class JobMeta: job_id: str name: str size: int block_size: int blocks: int priority: int state: str submitted_at: float owner: str class Spooler: def __init__(self, root, block_size=4096, max_well_bytes=1024 * 1024): self.root = Path(root) self.block_size = block_size self.max_well_bytes = max_well_bytes self.input_well = self.root / "input_well" self.output_well = self.root / "output_well" self.done = self.root / "done" self.tmp = self.root / "tmp" self.sink = self.root / "sink" / "printer.out" for p in [self.input_well, self.output_well, self.done, self.tmp, self.sink.parent]: p.mkdir(parents=True, exist_ok=True) self.lock = threading.Lock() self.cv = threading.Condition(self.lock) self.sink_lock = threading.Lock() self.stop_flag = False self.used_bytes = self._well_used() def _well_used(self): total = 0 for well in (self.input_well, self.output_well): for f in well.rglob("*.bin"): total += f.stat().st_size return total def submit(self, src, priority=5, owner="user"): src = Path(src) data = src.read_bytes() size = len(data) with self.cv: while self.used_bytes + size > self.max_well_bytes and not self.stop_flag: self.cv.wait(timeout=0.5) if self.stop_flag: raise RuntimeError("spooler stopped") job_id = uuid.uuid4().hex[:12] tmp_job = self.tmp / job_id tmp_job.mkdir(parents=True) blocks = 0 for off in range(0, size, self.block_size): chunk = data[off:off + self.block_size] (tmp_job / f"chunk_{blocks:04d}.bin").write_bytes(chunk) blocks += 1 meta = JobMeta(job_id, src.name, size, self.block_size, blocks, priority, "IN_WELL", time.time(), owner) (tmp_job / "meta.json").write_text( json.dumps(asdict(meta), ensure_ascii=False), encoding="utf-8") os.replace(tmp_job, self.input_well / job_id) # 原子提交 self.used_bytes += size self.cv.notify_all() return job_id def _pick_job(self, well, expected_state): jobs = [] for d in well.iterdir(): if not d.is_dir(): continue mp = d / "meta.json" if not mp.exists(): continue meta = JobMeta(**json.loads(mp.read_text(encoding="utf-8"))) if meta.state == expected_state: jobs.append((meta.priority, meta.submitted_at, d, meta)) if not jobs: return None jobs.sort(key=lambda x: (x[0], x[1])) return jobs[0][3], jobs[0][2] def input_worker(self): while not self.stop_flag: with self.cv: picked = self._pick_job(self.input_well, "IN_WELL") if not picked: self.cv.wait(timeout=0.2) continue meta, d = picked meta.state = "OUTPUT_WELL" (d / "meta.json").write_text( json.dumps(asdict(meta), ensure_ascii=False), encoding="utf-8") os.replace(d, self.output_well / meta.job_id) self.cv.notify_all() def output_worker(self): while not self.stop_flag: with self.cv: picked = self._pick_job(self.output_well, "OUTPUT_WELL") if not picked: self.cv.wait(timeout=0.2) continue meta, d = picked meta.state = "PRINTING" (d / "meta.json").write_text( json.dumps(asdict(meta), ensure_ascii=False), encoding="utf-8") with self.sink_lock: with self.sink.open("ab") as out: for i in range(meta.blocks): chunk = (d / f"chunk_{i:04d}.bin").read_bytes() out.write(chunk) out.flush() time.sleep(0.01) # 模拟慢速设备 with self.cv: meta.state = "DONE" (d / "meta.json").write_text( json.dumps(asdict(meta), ensure_ascii=False), encoding="utf-8") os.replace(d, self.done / meta.job_id) self.used_bytes -= meta.size self.cv.notify_all()

这段代码的逻辑说明分三点。第一，submit在条件变量里等待井容量，used_bytes只统计*.bin，避免元数据把容量算乱；等待带timeout=0.5，是为了让停止标志能被及时看到。第二，os.replace(tmp_job, input_well/job_id)是原子提交的关键，输出进程要么看不到目录，要么看到完整目录，不会读到半个作业。第三，output_worker在锁内把状态改成PRINTING，再释放锁去写sink，这样另一个输出线程不会重复取同一个作业；sink_lock包住整个作业的写入，避免两个作业的块交错。

参数方面，block_size=4096对应常见的4KB块，块越小元数据文件越多，块越大单次写设备的时间越长；max_well_bytes=1024*1024是1MB井容量，演示够用，真实场景按磁盘配额和设备速率调。priority数字越小越先出，submitted_at用来在同优先级下做先入先出。

3.2 启动生产者和消费者：运行与观察

入口脚本负责创建源文件、启动输入进程和两个输出进程、提交作业，然后等几秒停止。两个输出进程会争sink_lock，所以输出内容不会交错，但完成顺序受调度和锁竞争影响。

if __name__ == "__main__": sp = Spooler("./spool_root", block_size=4096, max_well_bytes=1024 * 1024) threading.Thread(target=sp.input_worker, daemon=True).start() for _ in range(2): threading.Thread(target=sp.output_worker, daemon=True).start() src_dir = Path("./source_jobs") src_dir.mkdir(exist_ok=True) for i in range(5): (src_dir / f"job_{i}.txt").write_text( f"user job {i}\n" + ("A" * 5000), encoding="utf-8") for p in sorted(src_dir.glob("*.txt")): jid = sp.submit(p, priority=5) print("submitted", jid, p.name) time.sleep(3) sp.stop_flag = True with sp.cv: sp.cv.notify_all()

运行python3 spooler_demo.py后，先看spool_root的目录变化：input_well会短暂出现作业，随后移动到output_well，最后进done。再看spool_root/sink/printer.out的大小，应该等于5个源文件的字节数之和。如果大小对不上，优先查output_well里有没有卡在PRINTING的目录，以及done里meta.json的blocks和实际块文件数是否一致。

python3 spooler_demo.py find spool_root -maxdepth 2 -type d | sort wc -c spool_root/sink/printer.out cat spool_root/sink/printer.out | head -c 100

这里有个观察技巧：不要只看最终sink文件，done/里的元数据才是“黑匣子”。每个作业的state应该是DONE，blocks应该等于ceil(size/block_size)。如果某个作业的state是PRINTING但进程已经停了，说明输出线程在写sink时异常退出，或者stop_flag在写块中途被置位。模拟器不是生产系统，但状态机闭环这件事必须一开始就做对。

4. 参数怎么调：井大小、块大小、并发度与调度策略的边界

4.1 井大小和块大小：不是越大越好

max_well_bytes决定同时能容纳多少字节的作业。设得太小，提交线程会频繁等待，吞吐上不去；设得太大，磁盘占用和恢复时间都会增加，而且“井”失去了缓冲的意义。我一般先按“设备一分钟能输出多少字节”估算，再乘以2到3倍作为井容量。比如模拟打印机每秒写100KB，井容量设20MB到30MB就够，不需要把整个磁盘都当井。

块大小的影响更隐蔽。块小，元数据文件多，目录遍历和打开文件次数增加；块大，单次写设备的时间变长，输出进程持锁时间变长，其他作业等待更久。4KB到64KB是常见区间，演示用4KB，压测时可以试16KB和64KB。下面这组参数是我在模拟项目X里会先跑的对照，不是绝对标准，但能快速看出拐点。

提示：井容量统计一定要和清理逻辑成对出现。submit加了多少字节，作业完成时就要减多少字节；异常路径也要减，否则跑几十个作业后“井空但提交被拒”就是必然。

4.2 并发度：输入进程和输出进程各开几个

输入进程的工作是扫描目录、改状态、移动目录，瓶颈在目录锁和文件系统元数据操作，不在CPU。开多个输入线程，它们会争同一把self.cv锁，收益很小，反而增加状态竞争。我一般只开一个输入进程，把扫描间隔控制在0.1秒到0.5秒之间；间隔太短会空转烧CPU，太长会让作业在输入井里多躺一会儿。

输出进程的数量要看设备模型。真实打印机是串行设备，同一时刻只能打一份；模拟器里用sink_lock包住整个作业，开两个输出线程也只是让第二个线程在锁外等，并不会提高设备吞吐。真正能提高吞吐的是“把多个小作业合并成一次设备启动”，这属于调度技巧，不是加线程能解决的。压测时如果发现output_worker增加后总时间没降，先看sink_lock的等待时间，而不是继续加线程。

4.3 调度策略：FCFS、优先级、短作业优先在井里的差别

代码里的_pick_job用(priority, submitted_at)排序，等价于“优先级优先，同级先入先出”。这对演示够用，但真实井里还要考虑作业大小。短作业优先能降低平均等待时间，但大作业可能饥饿；优先级调度能保证重要作业先出，但低优先级作业可能一直排不上。常见做法是加老化：每等一轮，优先级数字减一，减到0为止。

我在模拟项目X里踩过一个坑：把priority设成0到9，但提交时全用默认5，结果调度看起来像FCFS，误以为优先级没生效。后来在提交入口按作业类型分配优先级，才看出差别。参数不是设了就有用，得让输入数据真的落到不同取值上。

5. 避坑与排查：spooling假脱机模拟里最常见的五类翻车

5.1 井满后提交线程一直等待：条件变量没有超时和回滚

现象是submit调用后进程不动，日志停在“waiting for well space”。原因通常有两个：cv.wait()没写超时，或者used_bytes只增不减。前者在停止标志置位后仍然醒不过来，后者会让容量永远回不到零。解决方式是把等待改成cv.wait(timeout=0.5)，每次醒来检查stop_flag和容量；提交失败时把已经写入tmp的目录删掉，不能留下半成品占空间。更稳妥的做法是把“申请容量”和“写块”放在同一个try/finally里，异常时统一回滚。

5.2 作业目录只写了一半就被输出进程看到：缺少原子提交

现象是输出进程报FileNotFoundError或JSONDecodeError，或者sink文件比预期短。原因是提交线程直接往input_well/<job_id>里写块，输出进程扫描时看到了不完整目录。解决方式是先写tmp/<job_id>，最后写meta.json，再用os.replace整体重命名到input_well。如果文件系统不支持目录原子重命名，就用一个ready标记文件，输出进程只处理带ready的目录。原子提交这件事没有后悔药，必须在第一版就做。

5.3 两个输出进程重复打印同一个作业：状态更新不在锁内

现象是printer.out里同一份内容出现两次，done/里却只有一个作业目录。原因是输出进程在锁外选作业，两个线程同时看到OUTPUT_WELL，都认为自己拿到了。解决方式是把“选作业+改状态为PRINTING”放在同一个锁内，改完再释放锁去写sink；或者用os.replace把作业目录先移动到printing/私有目录，谁移动成功谁负责。状态机的每一次迁移都要有互斥保护，不能靠“应该不会同时发生”来赌。

5.4 删除作业后used_bytes不归零：统计口径和清理顺序错

现象是井目录已经空了，但新作业提交仍然被拒，打印used_bytes发现是个正数。原因通常是减容量写在os.replace之前，移动失败后没有回滚；或者异常路径直接return，跳过了释放。解决方式是封装一个release_job(meta)，在作业进入DONE或FAILED时统一调用，内部先减容量再通知条件变量。统计口径也要固定：只减*.bin的字节数，不要把meta.json算进去，否则加减不对称。

5.5 输出文件交错：sink文件没有按作业加锁

现象是printer.out里作业A的前半段和作业B的后半段穿插。原因是多个输出线程各自打开sink追加写，块级写入不是原子的。解决方式是用sink_lock包住整个作业的写入循环，而不是只包住单次write；或者每个作业先写到done/<job_id>.out，最后再按完成顺序合并。真实设备本来就是串行的，模拟器里让输出进程排队等设备锁，反而更接近真实语义。

6. 从能跑到可信：spooling模拟器的压测、日志校验与一个调度技巧

6.1 压测脚本：用随机大小和优先级跑100个作业

演示跑5个作业看不出容量和调度的边界。我一般写一个压测入口，随机生成1KB到200KB的文件，优先级在0到9之间随机，提交100个作业后等所有作业进done，再校验sink总字节数是否等于源文件总字节数。下面脚本可以放在同一个目录，复用Spooler类。

import random import time from pathlib import Path def stress(sp, n=100): src_dir = Path("./stress_jobs") src_dir.mkdir(exist_ok=True) total = 0 for i in range(n): size = random.randint(1024, 200 * 1024) data = bytes([random.randint(65, 90)]) * size p = src_dir / f"stress_{i:03d}.bin" p.write_bytes(data) total += size sp.submit(p, priority=random.randint(0, 9)) deadline = time.time() + 30 while time.time() < deadline: done = len(list(sp.done.iterdir())) if done >= n: break time.sleep(0.2) sink_size = sp.sink.stat().st_size print("expected", total, "sink", sink_size, "done", len(list(sp.done.iterdir()))) assert sink_size == total, "sink size mismatch"

这段脚本的关键不是“跑完”，而是最后那个断言。sink总字节数必须等于所有源文件字节数之和，少了说明有作业丢块，多了说明重复输出。压测时把max_well_bytes调到刚好容纳十几个作业，观察submit的等待次数；再把block_size从4KB改到64KB，比较完成时间。日志里至少记录job_id、state、时间戳和字节数，出问题时按job_id重放状态机，比翻控制台输出快得多。

6.2 一个调度技巧：给输出井加“设备空闲窗口”

最后说一个我后来才想明白的技巧。输出进程不必来一个作业就立刻启动设备，可以在输出井里等一个很短的“空闲窗口”，比如20毫秒，把窗口内到达的小作业按优先级合并成一批，再一次性写sink。这样设备启动次数少了，小作业的等待时间也不会明显变长。实现上就是在output_worker取作业前加一个time.sleep(0.02)的窗口等待，或者用一个批量队列攒够N个块再写。参数别设太大，窗口超过100毫秒，交互型作业的延迟就会变得明显。

我现在的习惯是：先把状态机和原子提交做对，再谈参数和调度。spooling假脱机输入输出模拟技术看起来像操作系统课里的老概念，但真落到文件、锁、条件变量和异常回滚上，坑一个不少。每次调参前先看done/里的元数据，比盯着sink文件猜原因可靠得多。希望帮到你。

本文还有配套的精品资源，点击获取