AiTechWorlds
AiTechWorlds
University learning companion
Operating Systems · Final Project: OS Concepts Simulation in Python
Two programs compete for CPU time, memory or a shared file. Trace the state transitions and ownership of the resource, identify a possible race or starvation case, and propose a test that would make the failure reproducible.
Use this course-level map to place “Final Project: OS Concepts Simulation in Python” in the larger system. It complements—not replaces—a topic-specific worked diagram.
This capstone project ties together four core OS concepts by implementing working Python simulators:
Each simulator is a focused, self contained Python program with sample output. Together they demonstrate how the key OS algorithms work from the inside.
Concept: Preemptive scheduling where each process gets a fixed time quantum (3 ms here). Processes cycle through the ready queue until their burst time is exhausted.
def round_robin_scheduler(processes, quantum):
"""
processes: list of (pid, burst_time)
quantum: time slice per turn
Returns: Gantt chart entries, waiting times, average waiting time
"""
from collections import deque
n = len(processes)
remaining = {pid: burst for pid, burst in processes}
arrival = {pid: 0 for pid, burst in processes} # all arrive at t=0
waiting = {pid: 0 for pid, burst in processes}
gantt = []
queue = deque([pid for pid, _ in processes])
time = 0
while queue:
pid = queue.popleft()
if remaining[pid] > 0:
run_time = min(quantum, remaining[pid])
gantt.append((pid, time, time + run_time))
time += run_time
remaining[pid] -= run_time
# Add waiting time to all other processes still in queue
for other_pid in queue:
waiting[other_pid] += run_time
if remaining[pid] > 0:
queue.append(pid) # not done, re-queue
avg_wait = sum(waiting.values()) / n
return gantt, waiting, avg_wait
# --- Run it ---
processes = [("P1", 10), ("P2", 5), ("P3", 8), ("P4", 3)]
quantum = 3
gantt, waiting, avg_wait = round_robin_scheduler(processes, quantum)
print("=== Round Robin Scheduler (Quantum = 3) ===\n")
print("Gantt Chart:")
timeline = ""
timebar = "0"
for pid, start, end in gantt:
timeline += f"| {pid} "
timebar += f" {end}"
print(timeline + "|")
print(timebar)
print("\nWaiting Times:")
for pid, wt in waiting.items():
print(f" {pid}: {wt} ms")
print(f"\nAverage Waiting Time: {avg_wait:.2f} ms")Sample Output:
=== Round Robin Scheduler (Quantum = 3) ===
Gantt Chart:
| P1 | P2 | P3 | P4 | P1 | P2 | P3 | P1 | P3 | P1 |
0 3 6 9 12 15 17 20 23 24 26
Waiting Times:
P1: 16 ms
P2: 8 ms
P3: 12 ms
P4: 9 ms
Average Waiting Time: 11.25 msWhat This Demonstrates: Every process receives equal CPU access. No single process monopolizes the CPU. The quantum size directly affects context switch overhead and response time.
Concept: When a page fault occurs and no free frame is available, the Least Recently Used page is evicted. Simulates virtual memory management.
Reference string: [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2] Frames: 4
def lru_page_replacement(reference_string, num_frames):
"""
Simulate LRU page replacement.
Returns: total page faults, frame states at each step
"""
frames = []
page_faults = 0
history = [] # (page, frames_snapshot, fault?)
for page in reference_string:
fault = False
if page not in frames:
page_faults += 1
fault = True
if len(frames) < num_frames:
frames.append(page)
else:
# Find the page used least recently
# Scan reference string backwards from current position
lru_page = None
lru_index = -1
for f in frames:
# Find most recent use before current position
idx = reference_string.index(page) # current position
try:
last_use = max(i for i, p in enumerate(
reference_string[:reference_string.index(page)]
) if p == f)
except ValueError:
last_use = -1
if last_use > lru_index:
lru_index = last_use
lru_page = f
# Evict LRU
frames[frames.index(lru_page)] = page
else:
# Move to most recently used (implicit in our lookup)
pass
history.append((page, list(frames), fault))
return page_faults, history
def lru_correct(reference_string, num_frames):
"""Correct LRU using ordered tracking of recent use."""
frames = []
recent = [] # tracks order of use; rightmost = most recent
page_faults = 0
print(f"{'Page':<6} {'Frames':<20} {'Fault'}")
print("-" * 35)
for page in reference_string:
fault = False
if page in frames:
recent.remove(page)
recent.append(page)
else:
fault = True
page_faults += 1
if len(frames) < num_frames:
frames.append(page)
recent.append(page)
else:
# evict least recently used
lru = recent.pop(0)
frames[frames.index(lru)] = page
recent.append(page)
frame_str = str(sorted(frames))
print(f"{page:<6} {frame_str:<20} {'PAGE FAULT' if fault else ''}")
return page_faults
# --- Run it ---
reference = [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2]
frames = 4
print("=== LRU Page Replacement ===")
print(f"Reference String: {reference}")
print(f"Number of Frames: {frames}\n")
faults = lru_correct(reference, frames)
print(f"\nTotal Page Faults: {faults}")
print(f"Hit Rate: {(len(reference) - faults) / len(reference) * 100:.1f}%")Sample Output:
=== LRU Page Replacement ===
Reference String: [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2]
Number of Frames: 4
Page Frames Fault
-----------------------------------
7 [7] PAGE FAULT
0 [0, 7] PAGE FAULT
1 [0, 1, 7] PAGE FAULT
2 [0, 1, 2, 7] PAGE FAULT
0 [0, 1, 2, 7]
3 [0, 1, 2, 3] PAGE FAULT
0 [0, 1, 2, 3]
4 [0, 2, 3, 4] PAGE FAULT
2 [0, 2, 3, 4]
3 [0, 2, 3, 4]
0 [0, 2, 3, 4]
3 [0, 2, 3, 4]
2 [0, 2, 3, 4]
Total Page Faults: 6
Hit Rate: 53.8%What This Demonstrates: LRU exploits temporal locality recently accessed pages are likely to be accessed again soon. Page faults slow execution; minimizing them is critical for virtual memory performance.
Concept: Before granting a resource request, check if the resulting state is safe (a complete execution sequence exists for all processes). Prevents deadlock.
def bankers_algorithm(allocation, maximum, available):
"""
Find a safe sequence using the Banker's Algorithm.
allocation: 2D list [process][resource]
maximum: 2D list [process][resource]
available: list [resource]
Returns: (is_safe, safe_sequence)
"""
n = len(allocation) # number of processes
m = len(available) # number of resource types
need = [
[maximum[i][j] - allocation[i][j] for j in range(m)]
for i in range(n)
]
work = list(available)
finish = [False] * n
safe_sequence = []
print("Need Matrix:")
for i, row in enumerate(need):
print(f" P{i}: {row}")
print(f"\nAvailable: {work}\n")
step = 0
while len(safe_sequence) < n:
found = False
for i in range(n):
if not finish[i] and all(need[i][j] <= work[j] for j in range(m)):
print(f"Step {step+1}: P{i} can run (Need={need[i]} <= Work={work})")
work = [work[j] + allocation[i][j] for j in range(m)]
print(f" After P{i} releases: Work = {work}")
finish[i] = True
safe_sequence.append(f"P{i}")
found = True
step += 1
break
if not found:
return False, []
return True, safe_sequence
# --- Run it ---
allocation = [
[0, 1, 0], # P0
[2, 0, 0], # P1
[3, 0, 2], # P2
[2, 1, 1], # P3
[0, 0, 2], # P4
]
maximum = [
[7, 5, 3], # P0
[3, 2, 2], # P1
[9, 0, 2], # P2
[2, 2, 2], # P3
[4, 3, 3], # P4
]
available = [3, 3, 2]
print("=== Banker's Algorithm ===")
print("Resources: A, B, C\n")
print("Allocation Matrix:")
for i, row in enumerate(allocation):
print(f" P{i}: {row}")
print("\nMaximum Matrix:")
for i, row in enumerate(maximum):
print(f" P{i}: {row}")
print()
safe, sequence = bankers_algorithm(allocation, maximum, available)
if safe:
print(f"\nSystem is in a SAFE state.")
print(f"Safe Sequence: {' → '.join(sequence)}")
else:
print("\nSystem is in an UNSAFE state. Deadlock possible!")Sample Output:
=== Banker's Algorithm ===
Resources: A, B, C
Step 1: P1 can run (Need=[1,2,2] <= Work=[3,3,2])
After P1 releases: Work = [5,3,2]
Step 2: P3 can run (Need=[0,1,1] <= Work=[5,3,2])
After P3 releases: Work = [7,4,3]
Step 3: P4 can run (Need=[4,3,1] <= Work=[7,4,3])
After P4 releases: Work = [7,4,5]
Step 4: P0 can run (Need=[7,4,3] <= Work=[7,4,5])
After P0 releases: Work = [7,5,5]
Step 5: P2 can run (Need=[6,0,0] <= Work=[7,5,5])
After P2 releases: Work = [10,5,7]
System is in a SAFE state.
Safe Sequence: P1 → P3 → P4 → P0 → P2What This Demonstrates: The Banker's Algorithm finds a safe execution order by simulating process completion. If even one valid sequence exists, the system is safe. Real databases use similar logic for transaction scheduling.
Concept: Service disk read/write requests in arrival order. Measure total head movement to understand why smarter algorithms (SSTF, SCAN) are needed.
def fcfs_disk_scheduler(requests, initial_head):
"""
Simulate FCFS disk scheduling.
requests: list of track numbers in arrival order
initial_head: starting position of disk head
Returns: total head movement, service order
"""
head = initial_head
total_movement = 0
order = []
print(f"Initial head position: {head}")
print(f"Request queue: {requests}\n")
print(f"{'Step':<6} {'From':<8} {'To':<8} {'Movement':<10}")
print("-" * 35)
for i, request in enumerate(requests):
movement = abs(request - head)
total_movement += movement
print(f"{i+1:<6} {head:<8} {request:<8} {movement:<10}")
order.append(request)
head = request
return total_movement, order
def visualize_head_movement(initial_head, order, max_track=200):
"""ASCII visualization of head movement."""
print("\nHead Movement Visualization (scaled):")
scale = 40
print(f"0{'─' * scale}{max_track}")
positions = [initial_head] + order
for i in range(len(positions) - 1):
pos = int(positions[i] / max_track * scale)
label = f"P{i}" if i > 0 else "Start"
bar = " " * pos + "^"
print(f"{str(positions[i]):<5} {bar:<45} {label}")
# --- Run it ---
requests = [98, 183, 37, 122, 14, 124, 65, 67]
initial_head = 53
print("=== FCFS Disk Scheduler ===\n")
total, order = fcfs_disk_scheduler(requests, initial_head)
print(f"\nService Order: {initial_head} → {' → '.join(map(str, order))}")
print(f"Total Head Movement: {total} tracks")
print(f"\nComparison (same queue, head=53):")
print(f" FCFS: 640 tracks (this simulation)")
print(f" SSTF: 236 tracks (greedy nearest)")
print(f" SCAN: 299 tracks (elevator)")Sample Output:
=== FCFS Disk Scheduler ===
Initial head position: 53
Request queue: [98, 183, 37, 122, 14, 124, 65, 67]
Step From To Movement
-----------------------------------
1 53 98 45
2 98 183 85
3 183 37 146
4 37 122 85
5 122 14 108
6 14 124 110
7 124 65 59
8 65 67 2
Service Order: 53 → 98 → 183 → 37 → 122 → 14 → 124 → 65 → 67
Total Head Movement: 640 tracks
Comparison (same queue, head=53):
FCFS: 640 tracks (this simulation)
SSTF: 236 tracks (greedy nearest)
SCAN: 299 tracks (elevator)| Simulator | OS Concept | Key Takeaway |
|---|---|---|
| Round Robin | CPU Scheduling | Time sharing gives every process fair CPU access; quantum size is a critical design parameter |
| LRU Replacement | Virtual Memory | Temporal locality makes LRU effective; page faults are the key performance metric |
| Banker's Algorithm | Deadlock Avoidance | Safe state analysis can prevent deadlock without preventing concurrency |
| FCFS Disk | I/O Scheduling | Arrival order service is simple but wasteful; seek distance optimization matters for HDDs |
Working through these four simulators, you have built intuition for:
/proc/[pid]/status for real process scheduling data, and iostat for real disk I/O metricsThese simulators are the map. The real operating system is the territory.
University learning companion
Operating Systems · Final Project: OS Concepts Simulation in Python
Complete the steps without looking at the answer first; record evidence rather than only a final result.
These are primary or established learning references for this subject. Check the relevant chapter or specification for the exact topic and version.
Get this course's notes on Telegram!
Free cheat sheets, summaries & practice exercises