Operating systems

process.py# MODEL: a process is really just a bundle of state the OS tracks in a
# "process control block" (PCB). We model that bundle as a dataclass.
from dataclasses import dataclass, field

@dataclass
class Process:
    pid: int
    arrival: int          # tick it becomes ready
    burst: int            # CPU ticks it needs
    priority: int = 0     # lower = more important (convention here)
    remaining: int = field(init=False)
    def __post_init__(self):
        self.remaining = self.burst

jobs = [Process(1, 0, 5), Process(2, 1, 3), Process(3, 2, 8)]
for j in jobs:
    print(f"pid={j.pid} arrival={j.arrival} burst={j.burst} remaining={j.remaining}")

switch_cost.py# MODEL: each context switch costs a fixed number of "overhead" ticks.
# Fewer switches -> less overhead, but (see round-robin) worse responsiveness.
def total_time(work_ticks, switches, switch_cost=1):
    return work_ticks + switches * switch_cost

print("2 switches:", total_time(100, 2))    # 102
print("50 switches:", total_time(100, 50))  # 150

scheduler_nonpre.py# MODEL of two non-preemptive schedulers. Each returns per-job (wait, turnaround).
# Non-preemptive = once a job starts it runs to completion.
from dataclasses import dataclass

@dataclass
class Job:
    pid: int; arrival: int; burst: int

def run_nonpreemptive(jobs, pick_next):
    """pick_next(ready, now) -> the Job to run next."""
    jobs = sorted(jobs, key=lambda j: j.arrival)
    now, done, metrics = 0, [], {}
    pending = list(jobs)
    while pending:
        ready = [j for j in pending if j.arrival <= now]
        if not ready:                      # CPU idle until next arrival
            now = min(j.arrival for j in pending); continue
        j = pick_next(ready, now)
        start = now
        now += j.burst                     # runs to completion
        turnaround = now - j.arrival
        wait = start - j.arrival
        metrics[j.pid] = (wait, turnaround)
        pending.remove(j)
    return metrics

def report(name, m):
    avg_w = sum(w for w, _ in m.values()) / len(m)
    avg_t = sum(t for _, t in m.values()) / len(m)
    print(f"{name:5s} avg_wait={avg_w:.2f} avg_turnaround={avg_t:.2f}")

jobs = [Job(1, 0, 7), Job(2, 2, 4), Job(3, 4, 1), Job(4, 5, 4)]
fcfs = run_nonpreemptive(jobs, lambda ready, now: min(ready, key=lambda j: j.arrival))
sjf  = run_nonpreemptive(jobs, lambda ready, now: min(ready, key=lambda j: j.burst))
report("FCFS", fcfs)
report("SJF", sjf)

round_robin.py# MODEL: round-robin gives each job a fixed time "quantum", then rotates.
# Preemptive = a running job can be interrupted before it finishes.
from collections import deque
from dataclasses import dataclass

@dataclass
class Job:
    pid: int; arrival: int; burst: int

def round_robin(jobs, quantum):
    jobs = sorted(jobs, key=lambda j: j.arrival)
    remaining = {j.pid: j.burst for j in jobs}
    finish = {}
    now, i, q = 0, 0, deque()
    def admit(t):                    # move arrivals up to time t into the queue
        nonlocal i
        while i < len(jobs) and jobs[i].arrival <= t:
            q.append(jobs[i].pid); i += 1
    admit(now)
    while q:
        pid = q.popleft()
        slice_ = min(quantum, remaining[pid])
        now += slice_
        remaining[pid] -= slice_
        admit(now)                   # new arrivals queue ahead of the re-added job
        if remaining[pid] > 0:
            q.append(pid)            # not done -> back of the line
        else:
            finish[pid] = now
        if not q:                    # CPU would idle: pull in a future arrival
            admit(now)
            if not q and i < len(jobs):
                now = jobs[i].arrival; admit(now)
    by_pid = {j.pid: j for j in jobs}
    metrics = {}
    for pid, f in finish.items():
        j = by_pid[pid]
        turnaround = f - j.arrival
        metrics[pid] = (turnaround - j.burst, turnaround)  # (wait, turnaround)
    return metrics

jobs = [Job(1, 0, 7), Job(2, 2, 4), Job(3, 4, 1), Job(4, 5, 4)]
for qn in (2, 4):
    m = round_robin(jobs, qn)
    avg_w = sum(w for w, _ in m.values()) / len(m)
    print(f"RR q={qn} avg_wait={avg_w:.2f}")

syscall_gate.py# MODEL: a syscall is a guarded entry point. In user mode you may NOT run a
# privileged op directly; you must request it and let the "kernel" validate + run it.
class Kernel:
    def __init__(self):
        self.mode = "user"
        self._files = {}
    def syscall(self, name, *args):
        self.mode = "kernel"            # trap: switch to kernel mode
        try:
            handler = getattr(self, f"_sys_{name}", None)
            if handler is None:
                return (-1, "ENOSYS")   # no such system call
            return handler(*args)
        finally:
            self.mode = "user"          # return: back to user mode
    def _sys_write(self, path, data):
        self._files.setdefault(path, "")
        self._files[path] += data
        return (0, len(data))
    def _sys_read(self, path):
        return (0, self._files.get(path, ""))

k = Kernel()
print(k.syscall("write", "/tmp/log", "hello"))   # (0, 5)
print(k.syscall("read", "/tmp/log"))             # (0, "hello")
print(k.syscall("launch_missiles"))              # (-1, "ENOSYS")
print("mode after return:", k.mode)              # back in user mode