操作系统小玩具

---

主程序部分

hello.py

def main():
    x = 0
    for _ in range(10):
        b = sys_read()
        x = x * 2 + b

    sys_write(f'x = {x:010b}b')

proc.py

def Process(name):
    for _ in range(5):
        sys_write(name)

def main():
    sys_spawn(Process, 'A')
    sys_spawn(Process, 'B')

Without CommentsWith comments

#!/usr/bin/env python3

import sys
import random
from pathlib import Path

class OS:
    SYSCALLS = ['read', 'write', 'spawn']

    class Process:

        def __init__(self, func, *args):
            self._func = func(*args)

            self.retval = None

        def step(self):
            syscall, args, *_ = self._func.send(self.retval)
            self.retval = None
            return syscall, args

    def __init__(self, src):
        exec(src, globals())
        self.procs = [OS.Process(main)]
        self.buffer = ''

    def run(self):
        while self.procs:

            current = random.choice(self.procs)

            try:
                match current.step():
                    case 'read', _:
                        current.retval = random.choice([0, 1])
                    case 'write', s:
                        self.buffer += s
                    case 'spawn', (fn, *args):
                        self.procs += [OS.Process(fn, *args)]
                    case _:
                        assert 0

            except StopIteration:
                self.procs.remove(current)

        return self.buffer

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} file')
        exit(1)

    src = Path(sys.argv[1]).read_text()

    for syscall in OS.SYSCALLS:
        src = src.replace(f'sys_{syscall}',
                          f'yield "{syscall}", ')

    stdout = OS(src).run()
    print(stdout)

#!/usr/bin/env python3

import sys
import random
from pathlib import Path

class OS:
    '''
    A minimal executable operating system model. Processes
    are state machines (Python generators) that can be paused
    or continued with local states being saved.
    '''

    '''
    We implement four system calls:

    - read: read a random bit value.
    - write: write a string to the buffer.
    - spawn: create a new state machine (process).
    '''
    SYSCALLS = ['read', 'write', 'spawn']

    class Process:
        '''
        A "freezed" state machine. The state (local variables,
        program counters, etc.) are stored in the generator
        object.
        '''

        def __init__(self, func, *args):
            # func should be a generator function. Calling
            # func(*args) returns a generator object.
            self._func = func(*args)

            # This return value is set by the OS's main loop.
            self.retval = None

        def step(self):
            '''
            Resume the process with OS-written return value,
            until the next system call is issued.
            '''
            syscall, args, *_ = self._func.send(self.retval)
            self.retval = None
            return syscall, args

    def __init__(self, src):
        # This is a hack: we directly execute the source
        # in the current Python runtime--and main is thus
        # available for calling.
        exec(src, globals())
        self.procs = [OS.Process(main)]
        self.buffer = ''

    def run(self):
        # Real operating systems waste all CPU cycles
        # (efficiently, by putting the CPU into sleep) when
        # there is no running process at the moment. Our model
        # terminates if there is nothing to run.
        while self.procs:

            # There is also a pointer to the "current" process
            # in today's operating systems.
            current = random.choice(self.procs)

            try:
                # Operating systems handle interrupt and system
                # calls, and "assign" CPU to a process.
                match current.step():
                    case 'read', _:
                        current.retval = random.choice([0, 1])
                    case 'write', s:
                        self.buffer += s
                    case 'spawn', (fn, *args):
                        self.procs += [OS.Process(fn, *args)]
                    case _:
                        assert 0

            except StopIteration:
                # The generator object terminates.
                self.procs.remove(current)

        return self.buffer

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} file')
        exit(1)

    src = Path(sys.argv[1]).read_text()

    # Hack: patch sys_read(...) -> yield "sys_read", (...)
    for syscall in OS.SYSCALLS:
        src = src.replace(f'sys_{syscall}',
                          f'yield "{syscall}", ')

    stdout = OS(src).run()
    print(stdout)

---

对程序的思考

OS 对象：

• OS.procs 是一个对象
• OS.buffer 是一个空字符串
• OS.run() 是一个方法，匹配 OS.procs 中任意一个对象的 OS.procs.step() 方法返回值

Process 对象（也就是 OS.procs ）：

• Process._func 是一个生成器
• Process.retval 是返回值
• Process.step() 是一个方法，传送 Process.retval 参数的值后，重置 Process.retval 参数，并返回 yield 后的 sycall 和 args。

Tip

对于函数生成器 gen，首次 send() 必须是 send(None) 或 next(gen)

参数 src 被修改为以下内容后导入：

def Process(name):
    for _ in range(5):
        yield "write", (name)

def main():
    yield "spawn", (Process, 'A')
    yield "spawn", (Process, 'B')

def main():
    x = 0
    for _ in range(10):
        b = yield "read", ()
        x = x * 2 + b

    yield "write", (f'x = {x:010b}b')