Key Code Examples
1. Process Scheduling Simulator (Round Robin)
from collections import deque
from dataclasses import dataclass
from typing import List
@dataclass
class Process:
pid: int
arrival_time: int
burst_time: int
remaining_time: int = 0
completion_time: int = 0
waiting_time: int = 0
turnaround_time: int = 0
def __post_init__(self):
self.remaining_time = self.burst_time
class RoundRobinScheduler:
"""Round Robin CPU scheduling algorithm."""
def __init__(self, time_quantum: int):
self.time_quantum = time_quantum
def schedule(self, processes: List[Process]) -> List[Process]:
# Sort by arrival time
processes.sort(key=lambda p: p.arrival_time)
ready_queue = deque()
completed = []
current_time = 0
process_idx = 0
while len(completed) < len(processes):
# Add newly arrived processes to ready queue
while process_idx < len(processes) and \
processes[process_idx].arrival_time <= current_time:
ready_queue.append(processes[process_idx])
process_idx += 1
if not ready_queue:
# CPU idle, jump to next arrival
if process_idx < len(processes):
current_time = processes[process_idx].arrival_time
continue
current = ready_queue.popleft()
execute_time = min(self.time_quantum, current.remaining_time)
current_time += execute_time
current.remaining_time -= execute_time
# Add processes that arrived during execution
while process_idx < len(processes) and \
processes[process_idx].arrival_time <= current_time:
ready_queue.append(processes[process_idx])
process_idx += 1
if current.remaining_time > 0:
ready_queue.append(current)
else:
current.completion_time = current_time
current.turnaround_time = current.completion_time - current.arrival_time
current.waiting_time = current.turnaround_time - current.burst_time
completed.append(current)
return completed
@staticmethod
def print_results(processes: List[Process]):
print(f"{'PID':<6}{'Arrival':<10}{'Burst':<8}{'Completion':<12}{'Turnaround':<12}{'Waiting':<8}")
print("-" * 60)
total_wt = total_tat = 0
for p in sorted(processes, key=lambda x: x.pid):
print(f"{p.pid:<6}{p.arrival_time:<10}{p.burst_time:<8}"
f"{p.completion_time:<12}{p.turnaround_time:<12}{p.waiting_time:<8}")
total_wt += p.waiting_time
total_tat += p.turnaround_time
n = len(processes)
print(f"\nAverage Waiting Time: {total_wt/n:.2f}")
print(f"Average Turnaround Time: {total_tat/n:.2f}")
# Example
processes = [
Process(pid=1, arrival_time=0, burst_time=8),
Process(pid=2, arrival_time=1, burst_time=4),
Process(pid=3, arrival_time=2, burst_time=9),
Process(pid=4, arrival_time=3, burst_time=5),
]
scheduler = RoundRobinScheduler(time_quantum=3)
result = scheduler.schedule(processes)
scheduler.print_results(result)
2. Banker's Algorithm (Deadlock Avoidance)
class BankersAlgorithm:
"""Banker's Algorithm for deadlock avoidance."""
def __init__(self, available, max_demand, allocation):
self.available = available[:] # Available resources
self.max_demand = max_demand[:] # Max demand matrix
self.allocation = allocation[:] # Allocation matrix
self.num_processes = len(allocation)
self.num_resources = len(available)
# Calculate need matrix
self.need = [
[self.max_demand[i][j] - self.allocation[i][j]
for j in range(self.num_resources)]
for i in range(self.num_processes)
]
def is_safe(self):
"""Check if system is in safe state."""
work = self.available[:]
finish = [False] * self.num_processes
safe_sequence = []
while len(safe_sequence) < self.num_processes:
found = False
for i in range(self.num_processes):
if not finish[i] and \
all(self.need[i][j] <= work[j] for j in range(self.num_resources)):
# Process can complete
for j in range(self.num_resources):
work[j] += self.allocation[i][j]
finish[i] = True
safe_sequence.append(i)
found = True
if not found:
# No safe sequence exists
return False, []
return True, safe_sequence
def request_resources(self, process_id, request):
"""Handle resource request from a process."""
# Check if request exceeds need
for j in range(self.num_resources):
if request[j] > self.need[process_id][j]:
raise ValueError("Request exceeds maximum claim")
# Check if resources are available
for j in range(self.num_resources):
if request[j] > self.available[j]:
return False, "Process must wait — resources unavailable"
# Pretend to allocate and check safety
old_available = self.available[:]
old_allocation = [row[:] for row in self.allocation]
old_need = [row[:] for row in self.need]
for j in range(self.num_resources):
self.available[j] -= request[j]
self.allocation[process_id][j] += request[j]
self.need[process_id][j] -= request[j]
safe, sequence = self.is_safe()
if safe:
return True, f"Safe state. Sequence: {sequence}"
else:
# Rollback
self.available = old_available
self.allocation = old_allocation
self.need = old_need
return False, "Unsafe state — request denied"
# Example: 3 resource types, 5 processes
available = [3, 3, 2]
max_demand = [
[7, 5, 3], # P0
[3, 2, 2], # P1
[9, 0, 2], # P2
[2, 2, 2], # P3
[4, 3, 3], # P4
]
allocation = [
[0, 1, 0], # P0
[2, 0, 0], # P1
[3, 0, 2], # P2
[2, 1, 1], # P3
[0, 0, 2], # P4
]
banker = BankersAlgorithm(available, max_demand, allocation)
safe, sequence = banker.is_safe()
print(f"Safe: {safe}, Sequence: {sequence}")
# Safe: True, Sequence: [P1, P3, P4, P0, P2]
# P1 requests (1, 0, 2)
result, msg = banker.request_resources(1, [1, 0, 2])
print(msg) # Safe state. Sequence: [...]
3. Memory Paging & Page Replacement (LRU)
from collections import OrderedDict
class LRUCache:
"""Least Recently Used page replacement algorithm."""
def __init__(self, capacity: int):
self.capacity = capacity
self.cache = OrderedDict() # page_num -> data
self.page_faults = 0
self.page_hits = 0
def access(self, page_num: int) -> bool:
"""Access a page. Returns True if hit, False if fault."""
if page_num in self.cache:
# Page hit — move to end (most recently used)
self.cache.move_to_end(page_num)
self.page_hits += 1
return True
else:
# Page fault
self.page_faults += 1
if len(self.cache) >= self.capacity:
# Evict least recently used (first item)
evicted = self.cache.popitem(last=False)
print(f" Page fault! Evicted page {evicted[0]}, loaded {page_num}")
else:
print(f" Page fault! Loaded {page_num}")
self.cache[page_num] = f"data_{page_num}"
return False
def get_stats(self):
total = self.page_hits + self.page_faults
hit_rate = self.page_hits / total * 100 if total > 0 else 0
return {
"hits": self.page_hits,
"faults": self.page_faults,
"hit_rate": f"{hit_rate:.1f}%",
"current_pages": list(self.cache.keys())
}
def display(self):
print(f"\nFrame: {' -> '.join(str(k) for k in self.cache.keys())}")
stats = self.get_stats()
print(f"Hits: {stats['hits']}, Faults: {stats['faults']}, Hit Rate: {stats['hit_rate']}")
class FIFOCache:
"""First In First Out page replacement."""
def __init__(self, capacity: int):
self.capacity = capacity
self.queue = [] # FIFO queue
self.page_set = set()
self.page_faults = 0
self.page_hits = 0
def access(self, page_num: int) -> bool:
if page_num in self.page_set:
self.page_hits += 1
return True
self.page_faults += 1
if len(self.queue) >= self.capacity:
evicted = self.queue.pop(0)
self.page_set.remove(evicted)
print(f" Page fault! Evicted {evicted}, loaded {page_num}")
else:
print(f" Page fault! Loaded {page_num}")
self.queue.append(page_num)
self.page_set.add(page_num)
return False
# Example: Simulate page references
print("=== LRU Page Replacement (3 frames) ===")
lru = LRUCache(capacity=3)
references = [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2, 1, 2, 0, 1, 7, 0, 1]
for page in references:
lru.access(page)
lru.display()
print("\n=== FIFO Page Replacement (3 frames) ===")
fifo = FIFOCache(capacity=3)
for page in references:
fifo.access(page)
print(f"FIFO: {fifo.page_hits} hits, {fifo.page_faults} faults")
4. Producer-Consumer with Semaphores
import threading
import time
import random
from collections import deque
class BoundedBuffer:
"""Thread-safe bounded buffer using semaphores."""
def __init__(self, capacity: int):
self.capacity = capacity
self.buffer = deque()
self.mutex = threading.Semaphore(1) # Binary semaphore for mutual exclusion
self.empty = threading.Semaphore(capacity) # Counts empty slots
self.full = threading.Semaphore(0) # Counts filled slots
def produce(self, item, producer_id: int):
"""Producer adds an item to the buffer."""
self.empty.acquire() # Wait for empty slot
self.mutex.acquire() # Enter critical section
self.buffer.append(item)
print(f"[Producer {producer_id}] Produced: {item} | Buffer: {list(self.buffer)}")
self.mutex.release() # Exit critical section
self.full.release() # Signal that buffer has item
def consume(self, consumer_id: int):
"""Consumer removes an item from the buffer."""
self.full.acquire() # Wait for available item
self.mutex.acquire() # Enter critical section
item = self.buffer.popleft()
print(f"[Consumer {consumer_id}] Consumed: {item} | Buffer: {list(self.buffer)}")
self.mutex.release() # Exit critical section
self.empty.release() # Signal that buffer has space
return item
def producer(buffer, producer_id, num_items):
for i in range(num_items):
item = f"P{producer_id}-Item{i}"
buffer.produce(item, producer_id)
time.sleep(random.uniform(0.1, 0.5))
def consumer(buffer, consumer_id, num_items):
for _ in range(num_items):
item = buffer.consume(consumer_id)
time.sleep(random.uniform(0.2, 0.6))
# Run simulation
buffer = BoundedBuffer(capacity=5)
threads = [
threading.Thread(target=producer, args=(buffer, 1, 5)),
threading.Thread(target=producer, args=(buffer, 2, 5)),
threading.Thread(target=consumer, args=(buffer, 1, 5)),
threading.Thread(target=consumer, args=(buffer, 2, 5)),
]
for t in threads:
t.start()
for t in threads:
t.join()
print("Simulation complete!")
5. File System Operations
import os
import struct
from datetime import datetime
from pathlib import Path
class SimpleFileSystem:
"""Simplified file system simulation."""
def __init__(self, disk_size=1024*1024): # 1 MB
self.disk_size = disk_size
self.block_size = 512
self.total_blocks = disk_size // self.block_size
self.fat = [-1] * self.total_blocks # File Allocation Table
self.root = {
'type': 'directory',
'children': {},
'created': datetime.now(),
}
def allocate_blocks(self, num_blocks: int) -> list:
"""Allocate contiguous blocks."""
allocated = []
for i in range(self.total_blocks):
if self.fat[i] == -1: # Free block
allocated.append(i)
if len(allocated) == num_blocks:
break
if len(allocated) < num_blocks:
raise OSError("Disk full")
# Mark blocks as allocated
for i in range(len(allocated)):
if i < len(allocated) - 1:
self.fat[allocated[i]] = allocated[i + 1] # Chain
else:
self.fat[allocated[i]] = -2 # End of file marker
return allocated
def create_file(self, path: str, size: int):
"""Create a file and allocate blocks."""
num_blocks = (size + self.block_size - 1) // self.block_size
blocks = self.allocate_blocks(num_blocks)
parts = path.strip('/').split('/')
current = self.root
# Navigate to parent directory
for part in parts[:-1]:
if part not in current['children']:
current['children'][part] = {
'type': 'directory',
'children': {},
'created': datetime.now(),
}
current = current['children'][part]
# Create file
current['children'][parts[-1]] = {
'type': 'file',
'size': size,
'blocks': blocks,
'created': datetime.now(),
'modified': datetime.now(),
}
print(f"Created {path} ({size} bytes, {num_blocks} blocks)")
return blocks
def delete_file(self, path: str):
"""Delete a file and free blocks."""
parts = path.strip('/').split('/')
current = self.root
for part in parts[:-1]:
if part not in current['children']:
raise FileNotFoundError(f"Path not found: {path}")
current = current['children'][part]
if parts[-1] not in current['children']:
raise FileNotFoundError(f"File not found: {path}")
file_info = current['children'][parts[-1]]
# Free blocks
if file_info['type'] == 'file':
for block in file_info['blocks']:
self.fat[block] = -1
del current['children'][parts[-1]]
print(f"Deleted {path}")
def get_disk_usage(self):
"""Calculate disk usage statistics."""
used_blocks = sum(1 for b in self.fat if b != -1)
return {
"total_size": self.disk_size,
"used": used_blocks * self.block_size,
"free": (self.total_blocks - used_blocks) * self.block_size,
"usage_percent": used_blocks / self.total_blocks * 100,
}
def list_directory(self, path='/'):
"""List contents of a directory."""
current = self.root
if path != '/':
for part in path.strip('/').split('/'):
current = current['children'][part]
for name, info in current['children'].items():
type_char = 'd' if info['type'] == 'directory' else 'f'
print(f"{type_char} {name:30} {info.get('size', 'N/A')}")
# Usage
fs = SimpleFileSystem(disk_size=10240) # 10 KB
fs.create_file('/documents/report.txt', 1024)
fs.create_file('/documents/budget.xlsx', 2048)
fs.create_file('/photos/vacation.jpg', 3072)
fs.list_directory('/documents')
print(fs.get_disk_usage())
fs.delete_file('/documents/budget.xlsx')
6. Shell Implementation (C)
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/wait.h>
#include <fcntl.h>
#define MAX_ARGS 64
#define MAX_CMD 1024
void parse_command(char *cmd, char **args) {
int i = 0;
char *token = strtok(cmd, " \t\n");
while (token != NULL && i < MAX_ARGS - 1) {
args[i++] = token;
token = strtok(NULL, " \t\n");
}
args[i] = NULL;
}
int execute_command(char **args) {
pid_t pid = fork();
if (pid < 0) {
perror("fork failed");
return -1;
} else if (pid == 0) {
// Child process
execvp(args[0], args);
perror("exec failed"); // Only reached if exec fails
exit(1);
} else {
// Parent process
int status;
waitpid(pid, &status, 0);
return WEXITSTATUS(status);
}
}
int execute_pipeline(char *cmd1[], char *cmd2[]) {
int pipefd[2];
pipe(pipefd);
pid_t pid1 = fork();
if (pid1 == 0) {
// First command: write to pipe
dup2(pipefd[1], STDOUT_FILENO);
close(pipefd[0]);
close(pipefd[1]);
execvp(cmd1[0], cmd1);
exit(1);
}
pid_t pid2 = fork();
if (pid2 == 0) {
// Second command: read from pipe
dup2(pipefd[0], STDIN_FILENO);
close(pipefd[0]);
close(pipefd[1]);
execvp(cmd2[0], cmd2);
exit(1);
}
close(pipefd[0]);
close(pipefd[1]);
waitpid(pid1, NULL, 0);
waitpid(pid2, NULL, 0);
return 0;
}
int main() {
char cmd[MAX_CMD];
char *args[MAX_ARGS];
printf("mini-shell> ");
while (fgets(cmd, MAX_CMD, stdin)) {
parse_command(cmd, args);
if (args[0] == NULL) continue;
if (strcmp(args[0], "exit") == 0) break;
if (strcmp(args[0], "cd") == 0) {
chdir(args[1]);
} else if (strcmp(args[0], "pwd") == 0) {
char cwd[1024];
getcwd(cwd, sizeof(cwd));
printf("%s\n", cwd);
} else {
execute_command(args);
}
printf("mini-shell> ");
}
return 0;
}
7. Thread Pool Implementation
import threading
import queue
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable, Any
class ThreadPool:
"""Custom thread pool implementation."""
def __init__(self, num_threads: int = 4):
self.tasks = queue.Queue()
self.results = {}
self.lock = threading.Lock()
self.task_id = 0
self.workers = []
self._shutdown = False
# Create worker threads
for i in range(num_threads):
worker = threading.Thread(target=self._worker_loop,
args=(i,), daemon=True)
worker.start()
self.workers.append(worker)
def _worker_loop(self, worker_id: int):
"""Main loop for worker threads."""
while not self._shutdown:
try:
task = self.tasks.get(timeout=1)
if task is None: # Shutdown signal
break
task_id, func, args, kwargs = task
try:
result = func(*args, **kwargs)
with self.lock:
self.results[task_id] = ('success', result)
except Exception as e:
with self.lock:
self.results[task_id] = ('error', str(e))
finally:
self.tasks.task_done()
except queue.Empty:
continue
def submit(self, func: Callable, *args, **kwargs) -> int:
"""Submit a task to the thread pool."""
with self.lock:
current_id = self.task_id
self.task_id += 1
self.tasks.put((current_id, func, args, kwargs))
return current_id
def get_result(self, task_id: int, timeout: float = None) -> Any:
"""Get result of a task (blocking)."""
start = time.time()
while timeout is None or time.time() - start < timeout:
with self.lock:
if task_id in self.results:
status, result = self.results.pop(task_id)
if status == 'success':
return result
else:
raise RuntimeError(result)
time.sleep(0.01)
raise TimeoutError("Task did not complete in time")
def shutdown(self):
"""Gracefully shutdown the thread pool."""
self._shutdown = True
for _ in self.workers:
self.tasks.put(None) # Signal workers to exit
for worker in self.workers:
worker.join()
# Using concurrent.futures (Python built-in)
def parallel_processing_example():
"""Using ThreadPoolExecutor for parallel tasks."""
def process_image(image_id: int) -> dict:
time.sleep(0.1) # Simulate processing
return {"image_id": image_id, "status": "processed"}
with ThreadPoolExecutor(max_workers=4) as executor:
# Submit tasks
futures = {executor.submit(process_image, i): i
for i in range(20)}
# Collect results as they complete
for future in as_completed(futures):
image_id = futures[future]
try:
result = future.result()
print(f"Image {image_id}: {result['status']}")
except Exception as e:
print(f"Image {image_id} failed: {e}")
# Usage
pool = ThreadPool(num_threads=4)
# Submit tasks
task_ids = [pool.submit(pow, base, exp)
for base, exp in [(2, 10), (3, 5), (5, 3), (7, 2)]]
# Get results
for tid in task_ids:
print(f"Result: {pool.get_result(tid)}")
pool.shutdown()
8. Disk Scheduling Algorithm (SCAN / Elevator)
class DiskScheduler:
"""Simulates disk head movement using SCAN algorithm."""
def __init__(self, disk_size=200, initial_head=50):
self.disk_size = disk_size
self.head = initial_head
self.direction = 1 # 1 = toward higher cylinders, -1 = lower
self.total_movement = 0
self.seek_sequence = [initial_head]
def scan(self, requests: list) -> dict:
"""SCAN (Elevator) algorithm."""
requests = sorted(set(requests))
left = [r for r in requests if r < self.head]
right = [r for r in requests if r >= self.head]
if self.direction == 1:
# Go right first, then left
sequence = right + left[::-1]
else:
# Go left first, then right
sequence = left[::-1] + right
for cylinder in sequence:
self.total_movement += abs(cylinder - self.head)
self.head = cylinder
self.seek_sequence.append(cylinder)
return {
"algorithm": "SCAN",
"sequence": self.seek_sequence,
"total_movement": self.total_movement,
"avg_seek_length": self.total_movement / len(requests)
}
def sstf(self, requests: list) -> dict:
"""Shortest Seek Time First algorithm."""
remaining = list(requests)
while remaining:
# Find closest request
closest = min(remaining, key=lambda r: abs(r - self.head))
self.total_movement += abs(closest - self.head)
self.head = closest
self.seek_sequence.append(self.head)
remaining.remove(closest)
return {
"algorithm": "SSTF",
"sequence": self.seek_sequence,
"total_movement": self.total_movement,
"avg_seek_length": self.total_movement / len(requests)
}
# Compare algorithms
requests = [98, 183, 37, 122, 14, 124, 65, 67]
print("Requests:", requests)
print()
scan = DiskScheduler(disk_size=200, initial_head=50)
scan_result = scan.scan(requests)
print(f"SCAN: Total movement = {scan_result['total_movement']}, "
f"Seq: {scan_result['sequence']}")
sstf = DiskScheduler(disk_size=200, initial_head=50)
sstf_result = sstf.sstf(requests)
print(f"SSTF: Total movement = {sstf_result['total_movement']}, "
f"Seq: {sstf_result['sequence']}")
# Example output:
# SCAN: Total movement = 236, Seq: [50, 65, 67, 98, 122, 124, 183, 37, 14]
# SSTF: Total movement = 236, Seq: [50, 65, 67, 37, 14, 98, 122, 124, 183]
9. Virtual Memory & TLB Simulation
class TLB:
"""Translation Lookaside Buffer simulation."""
def __init__(self, size=16):
self.size = size
self.entries = {} # page_num -> frame_num
self.access_order = [] # LRU order
self.hits = 0
self.misses = 0
def lookup(self, page_num: int) -> int:
"""Look up page in TLB. Returns frame number or -1 on miss."""
if page_num in self.entries:
self.hits += 1
self.access_order.remove(page_num)
self.access_order.append(page_num)
return self.entries[page_num]
else:
self.misses += 1
return -1
def insert(self, page_num: int, frame_num: int):
"""Insert a page-table entry into TLB."""
if len(self.entries) >= self.size:
# Evict LRU
evicted = self.access_order.pop(0)
del self.entries[evicted]
self.entries[page_num] = frame_num
self.access_order.append(page_num)
@property
def hit_rate(self):
total = self.hits + self.misses
return self.hits / total if total > 0 else 0
class VirtualMemoryManager:
"""Simulates virtual memory with paging and TLB."""
def __init__(self, physical_frames=4, tlb_size=16):
self.physical_frames = physical_frames
self.tlb = TLB(tlb_size)
self.page_table = {} # page_num -> frame_num
self.frame_table = [None] * physical_frames # frame -> page
self.free_frames = list(range(physical_frames))
self.page_faults = 0
def translate_address(self, virtual_address: int, page_size=4096) -> int:
"""Translate virtual address to physical address."""
page_num = virtual_address // page_size
offset = virtual_address % page_size
# Check TLB first
frame_num = self.tlb.lookup(page_num)
if frame_num == -1:
# TLB miss — check page table
if page_num in self.page_table:
frame_num = self.page_table[page_num]
self.tlb.insert(page_num, frame_num)
else:
# Page fault!
self.page_faults += 1
frame_num = self._handle_page_fault(page_num)
self.tlb.insert(page_num, frame_num)
physical_address = frame_num * page_size + offset
return physical_address
def _handle_page_fault(self, page_num: int) -> int:
"""Handle page fault by allocating a frame."""
if self.free_frames:
frame = self.free_frames.pop(0)
else:
# No free frames — use FIFO replacement
frame = self._fifo_replace()
self.page_table[page_num] = frame
self.frame_table[frame] = page_num
return frame
def _fifo_replace(self) -> int:
"""FIFO page replacement (simplified)."""
# In real OS: track load time, evict oldest
victim_frame = 0 # Simplified
victim_page = self.frame_table[victim_frame]
del self.page_table[victim_page]
return victim_frame
# Usage
vmm = VirtualMemoryManager(physical_frames=4, tlb_size=16)
# Simulate address accesses
import random
random.seed(42)
addresses = [random.randint(0, 65536) for _ in range(1000)]
for addr in addresses:
vmm.translate_address(addr)
print(f"Page Faults: {vmm.page_faults}/1000 ({vmm.page_faults/10:.1f}%)")
print(f"TLB Hit Rate: {vmm.tlb.hit_rate*100:.1f}%")
10. System Calls Wrapper (Python)
import os
import signal
import resource
import mmap
from contextlib import contextmanager
class SystemCalls:
"""Python wrappers for common POSIX system calls."""
# ===== Process Management =====
@staticmethod
def fork_example():
"""Demonstrate fork() system call."""
pid = os.fork()
if pid == 0:
# Child process
print(f"Child PID: {os.getpid()}, Parent PID: {os.getppid()}")
os._exit(0)
else:
# Parent process
_, status = os.waitpid(pid, 0)
print(f"Parent: child {pid} exited with status {status}")
@staticmethod
def exec_example():
"""Demonstrate exec() replacing process image."""
pid = os.fork()
if pid == 0:
# Child: replace with /bin/echo
os.execvp('echo', ['echo', 'Hello from exec!'])
else:
os.waitpid(pid, 0)
# ===== File I/O =====
@staticmethod
def low_level_io():
"""Demonstrate open/read/write/close system calls."""
# Open file
fd = os.open('/tmp/test.txt', os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
try:
# Write
os.write(fd, b'Hello, System Calls!\n')
finally:
os.close(fd)
# Read
fd = os.open('/tmp/test.txt', os.O_RDONLY)
try:
data = b''
while True:
chunk = os.read(fd, 1024)
if not chunk:
break
data += chunk
print(f"Read: {data.decode()}")
finally:
os.close(fd)
@staticmethod
def mmap_example():
"""Memory-mapped file I/O."""
# Create a file
with open('/tmp/mmap_test.dat', 'wb') as f:
f.write(b'\x00' * 4096)
# Memory-map the file
with open('/tmp/mmap_test.dat', 'r+b') as f:
mm = mmap.mmap(f.fileno(), 4096)
# Write through memory
mm[0:5] = b'Hello'
mm[100:105] = b'World'
# Read back
print(f"Data: {mm[0:5]}, {mm[100:105]}")
mm.close()
# ===== Process Synchronization =====
@staticmethod
def pipe_example():
"""Inter-process communication using pipes."""
read_fd, write_fd = os.pipe()
pid = os.fork()
if pid == 0:
# Child: write to pipe
os.close(read_fd)
os.write(write_fd, b'From child process')
os.close(write_fd)
os._exit(0)
else:
# Parent: read from pipe
os.close(write_fd)
data = os.read(read_fd, 1024)
print(f"Received: {data.decode()}")
os.close(read_fd)
os.waitpid(pid, 0)
# ===== Signal Handling =====
@staticmethod
def signal_example():
"""Handle OS signals gracefully."""
def handler(signum, frame):
print(f"Received signal {signum} ({signal.Signals(signum).name})")
signal.signal(signal.SIGUSR1, handler)
signal.signal(signal.SIGINT, handler)
print(f"PID: {os.getpid()} — send SIGUSR1 to test")
# kill -USR1 from another terminal
signal.pause() # Wait for signal
# ===== Resource Limits =====
@staticmethod
def set_resource_limits():
"""Set process resource limits (ulimit)."""
# Limit CPU time to 1 second
resource.setrlimit(resource.RLIMIT_CPU, (1, 1))
# Limit memory to 100 MB
resource.setrlimit(resource.RLIMIT_AS, (100*1024*1024, 100*1024*1024))
# Limit open files to 10
resource.setrlimit(resource.RLIMIT_NOFILE, (10, 10))
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
print(f"Open file limit: soft={soft}, hard={hard}")
# SystemCalls.low_level_io()
# SystemCalls.mmap_example()
# SystemCalls.pipe_example()
Core Concepts
Process vs Thread
Process: Independent execution unit with its own address space, file descriptors, and resources. Heavyweight to create/switch.
Thread: Lightweight unit of execution within a process. Shares address space and resources with other threads in the same process.
- Process Creation: fork() creates copy; exec() replaces image
- Thread Creation: pthread_create() — much faster than fork
- IPC: Pipes, shared memory, message queues, sockets
- Synchronization: Mutex, semaphore, condition variable
CPU Scheduling Algorithms
- FCFS (First-Come First-Served): Simple queue, non-preemptive, convoy effect
- SJF (Shortest Job First): Optimal average wait time, preemptive version is SRTF
- Round Robin: Time quantum, fair, good for time-sharing
- Priority Scheduling: Higher priority first, risk of starvation
- Multilevel Queue: Multiple queues with different priorities
- Multilevel Feedback Queue: Processes move between queues based on behavior
Memory Management
Paging: Fixed-size blocks (pages) mapped to physical frames. Eliminates external fragmentation.
Segmentation: Variable-sized blocks based on logical divisions (code, data, stack).
Virtual Memory: Uses disk as extension of RAM; demand paging loads pages only when needed.
Page Replacement: FIFO, LRU, Optimal (MIN), Clock algorithm.
TLB: Hardware cache for page table entries — speeds up address translation.
Deadlock
Four Necessary Conditions:
- Mutual Exclusion: Resource can only be held by one process
- Hold and Wait: Process holds resource while waiting for another
- No Preemption: Resource cannot be forcibly taken away
- Circular Wait: Circular chain of processes waiting
Handling Strategies: Prevention (eliminate one condition), Avoidance (Banker's Algorithm), Detection & Recovery, Ignorance (Ostrich algorithm)
File Systems
- FAT32: Simple, portable, 4GB file limit
- NTFS: Windows default, journaling, encryption
- ext4: Linux default, journaling, 16TB max file
- APFS: Apple, copy-on-write, snapshots
- ZFS: Enterprise, data integrity, snapshots
Key Concepts: Inodes (metadata), directories, journaling, mounting, permissions
Process Synchronization
Critical Section: Code accessing shared resources — must be mutually exclusive.
Solutions:
- Mutex: Binary lock — only one thread enters critical section
- Semaphore: Counter-based — allows N threads
- Monitor: High-level abstraction with condition variables
- Spinlock: Busy-wait — efficient for short waits
I/O Management
- Programmed I/O: CPU polls device status — wasteful
- Interrupt-driven I/O: Device interrupts CPU when ready
- DMA (Direct Memory Access): Device transfers data directly to memory without CPU
- I/O Scheduling: SCAN, C-SCAN, LOOK for disk head optimization
Protection & Security
User vs Kernel Mode: Ring 0 (kernel) has full access; Ring 3 (user) restricted.
Access Control: ACLs (who can do what), Capabilities (tokens representing rights).
Authentication: Passwords, biometrics, tokens, certificates.
Worked Examples
Example 1: Round Robin Scheduling Calculation
Problem: Calculate completion, turnaround, and waiting times for processes using Round Robin (time quantum = 4).
Processes: P0(Burst=8, Arrival=0), P1(Burst=4, Arrival=1), P2(Burst=9, Arrival=2), P3(Burst=5, Arrival=3)
Solution:
- Execution order: P0(4)→P1(4)→P2(4)→P3(4)→P0(4)→P2(4)→P3(1)→P2(4)→P2(1)
- Gantt: 0-4:P0, 4-8:P1, 8-12:P2, 12-16:P3, 16-20:P0, 20-24:P2, 24-25:P3, 25-29:P2, 29-30:P2
- P0: Completion=20, Turnaround=20-0=20, Waiting=20-8=12
- P1: Completion=8, Turnaround=8-1=7, Waiting=7-4=3
- P2: Completion=30, Turnaround=30-2=28, Waiting=28-9=19
- P3: Completion=25, Turnaround=25-3=22, Waiting=22-5=17
- Avg Waiting Time: (12+3+19+17)/4 = 12.75
Example 2: LRU Page Replacement Walkthrough
Problem: Given reference string [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2, 1, 2, 0, 1, 7, 0, 1] with 3 frames, simulate LRU.
Solution:
- Access 7: FAULT → [7] (Faults: 1)
- Access 0: FAULT → [7, 0] (Faults: 2)
- Access 1: FAULT → [7, 0, 1] (Faults: 3)
- Access 2: FAULT → evict LRU(7) → [0, 1, 2] (Faults: 4)
- Access 0: HIT → [1, 2, 0]
- Access 3: FAULT → evict LRU(1) → [2, 0, 3] (Faults: 5)
- Access 0: HIT → [2, 3, 0]
- Access 4: FAULT → evict LRU(2) → [3, 0, 4] (Faults: 6)
- Access 2: FAULT → evict LRU(3) → [0, 4, 2] (Faults: 7)
- Access 3: FAULT → evict LRU(0) → [4, 2, 3] (Faults: 8)
- Access 0: FAULT → evict LRU(4) → [2, 3, 0] (Faults: 9)
- Access 3: HIT, Access 2: HIT (no change)
- Access 1: FAULT → evict LRU(3) → [2, 0, 1] (Faults: 10)
- Access 2: HIT, Access 0: HIT, Access 1: HIT
- Access 7: FAULT → evict LRU(2) → [0, 1, 7] (Faults: 11)
- Access 0: HIT, Access 1: HIT
- Total: 12 page faults, 8 hits. Hit rate: 40%
Example 3: Banker's Algorithm Safety Check
Problem: 5 processes, 3 resources (A=10, B=5, C=7). Determine if system is in safe state.
Solution:
- Available: [3, 3, 2]
- Max: P0=[7,5,3], P1=[3,2,2], P2=[9,0,2], P3=[2,2,2], P4=[4,3,3]
- Allocation: P0=[0,1,0], P1=[2,0,0], P2=[3,0,2], P3=[2,1,1], P4=[0,0,2]
- Need: P0=[7,4,3], P1=[1,2,2], P2=[6,0,0], P3=[0,1,1], P4=[4,3,1]
- Safety Algorithm:
- Work=[3,3,2]. P1 need[1,2,2] ≤ Work? Yes → Work=[5,3,2], Finish[P1]=T
- P3 need[0,1,1] ≤ [5,3,2]? Yes → Work=[7,4,3], Finish[P3]=T
- P4 need[4,3,1] ≤ [7,4,3]? Yes → Work=[7,4,5], Finish[P4]=T
- P0 need[7,4,3] ≤ [7,4,5]? Yes → Work=[7,5,5], Finish[P0]=T
- P2 need[6,0,0] ≤ [7,5,5]? Yes → Work=[10,5,7], Finish[P2]=T
- Result: Safe! Sequence: P1 → P3 → P4 → P0 → P2
Example 4: Disk Scheduling Comparison
Problem: Head starts at cylinder 50, requests: [98, 183, 37, 122, 14, 124, 65, 67]. Compare FCFS, SSTF, and SCAN.
Solution:
- FCFS: 50→98→183→37→122→14→124→65→67
- Movement: 48+85+146+85+108+110+59+2 = 643 cylinders
- SSTF: 50→65→67→37→14→98→122→124→183
- Movement: 15+2+30+23+84+24+2+59 = 239 cylinders
- SCAN: 50→65→67→98→122→124→183→37→14
- Movement: 15+2+31+24+2+59+146+23 = 302 cylinders
- Conclusion: SSTF best (239), but may starve. SCAN is fair.
Example 5: Address Translation with Paging
Problem: Page size = 4 KB. Page table maps: Page 0→Frame 3, Page 1→Frame 7, Page 2→Frame 5. Translate virtual address 12345.
Solution:
- Page size = 4096 bytes
- Virtual address 12345
- Page number = 12345 / 4096 = 3 (integer division)
- Offset = 12345 % 4096 = 57
- Look up page 3 in page table → NOT PRESENT (only pages 0-2 mapped)
- Result: Page Fault! OS must load page 3 from disk.
Second example: Virtual address 8200
- Page number = 8200 / 4096 = 2
- Offset = 8200 % 4096 = 8
- Page 2 → Frame 5
- Physical address = 5 × 4096 + 8 = 20488
Practice Problems
Problem 1: Implement Priority Scheduling
Write a Priority Scheduling algorithm (lower number = higher priority). Handle tie-breaking by arrival time.
from dataclasses import dataclass
import heapq
@dataclass
class Process:
pid: int
arrival_time: int
burst_time: int
priority: int
def priority_schedule(processes):
processes.sort(key=lambda p: (p.arrival_time, p.priority))
time = 0
completed = []
remaining = list(processes)
while remaining:
# Find highest priority arrived process
available = [p for p in remaining if p.arrival_time <= time]
if not available:
time = remaining[0].arrival_time
continue
current = min(available, key=lambda p: p.priority)
time += current.burst_time
remaining.remove(current)
current.completion_time = time
current.turnaround_time = time - current.arrival_time
current.waiting_time = current.turnaround_time - current.burst_time
completed.append(current)
return completed
# Test
procs = [
Process(1, 0, 6, 2),
Process(2, 1, 3, 1),
Process(3, 2, 1, 3),
Process(4, 3, 2, 4),
]
for p in priority_schedule(procs):
print(f"P{p.pid}: CT={p.completion_time}, TAT={p.turnaround_time}, WT={p.waiting_time}")
Problem 2: Implement Readers-Writers Problem
Create a solution where multiple readers can read simultaneously, but writers need exclusive access.
import threading
import time
import random
class ReadWriteLock:
def __init__(self):
self.readers = 0
self.writers_waiting = 0
self.lock = threading.Condition()
self.write_locked = False
def acquire_read(self):
with self.lock:
while self.write_locked or self.writers_waiting > 0:
self.lock.wait() # Readers wait if writer is active or waiting
self.readers += 1
def release_read(self):
with self.lock:
self.readers -= 1
if self.readers == 0:
self.lock.notify_all()
def acquire_write(self):
with self.lock:
self.writers_waiting += 1
while self.readers > 0 or self.write_locked:
self.lock.wait()
self.writers_waiting -= 1
self.write_locked = True
def release_write(self):
with self.lock:
self.write_locked = False
self.lock.notify_all()
# Usage
rw_lock = ReadWriteLock()
data = {"value": 0}
def reader(reader_id):
rw_lock.acquire_read()
print(f"Reader {reader_id} reading: {data['value']}")
time.sleep(random.uniform(0.1, 0.5))
rw_lock.release_read()
def writer(writer_id, new_value):
rw_lock.acquire_write()
print(f"Writer {writer_id} writing: {new_value}")
time.sleep(random.uniform(0.1, 0.3))
data["value"] = new_value
rw_lock.release_write()
# Run threads
threads = []
for i in range(5):
threads.append(threading.Thread(target=reader, args=(i,)))
for i in range(2):
threads.append(threading.Thread(target=writer, args=(i, i * 100)))
for t in threads:
t.start()
for t in threads:
t.join()
Problem 3: Calculate Page Fault Rate
Given reference string [1, 2, 3, 4, 1, 2, 5, 1, 2, 3, 4, 5] with 3 frames, compute FIFO and LRU page faults.
Answer:
- FIFO: 9 page faults (frames: [1,2,3]→evict 1→[4,2,3]→evict 2→[4,1,3]→...)
- LRU: 10 page faults (more because of the cyclic pattern)
- Optimal (MIN): 7 page faults (theoretical best)
Problem 4: Implement a Simple Shell
Write a C program that implements a basic shell with fork, exec, and wait. Support pipes (|).
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/wait.h>
#define MAX_CMD 1024
#define MAX_ARGS 64
void parse(char *input, char **args) {
char *token = strtok(input, " \t\n");
int i = 0;
while (token) { args[i++] = token; token = strtok(NULL, " \t\n"); }
args[i] = NULL;
}
int main() {
char input[MAX_CMD];
char *args[MAX_ARGS];
while (1) {
printf("shell> ");
if (!fgets(input, MAX_CMD, stdin)) break;
parse(input, args);
if (!args[0]) continue;
if (strcmp(args[0], "exit") == 0) break;
pid_t pid = fork();
if (pid == 0) {
execvp(args[0], args);
perror("exec failed");
exit(1);
}
wait(NULL);
}
return 0;
}
Problem 5: Memory Allocation (Best Fit)
Simulate dynamic memory allocation using Best Fit strategy.
class MemoryBlock:
def __init__(self, start, size, is_free=True):
self.start = start
self.size = size
self.is_free = is_free
class BestFitAllocator:
def __init__(self, total_size):
self.blocks = [MemoryBlock(0, total_size)]
def allocate(self, size):
best = None
best_idx = -1
for i, block in enumerate(self.blocks):
if block.is_free and block.size >= size:
if best is None or block.size < best.size:
best = block
best_idx = i
if best is None:
return None # No suitable block found
if best.size == size:
best.is_free = False
else:
new_block = MemoryBlock(best.start, size, is_free=False)
best.start += size
best.size -= size
self.blocks.insert(best_idx, new_block)
return best.start if best.size == size else self.blocks[best_idx].start
def deallocate(self, start, size):
for i, block in enumerate(self.blocks):
if block.start == start and not block.is_free:
block.is_free = True
# Coalesce with adjacent free blocks
self._coalesce()
return True
return False
def _coalesce(self):
i = 0
while i < len(self.blocks) - 1:
if self.blocks[i].is_free and self.blocks[i+1].is_free:
self.blocks[i].size += self.blocks[i+1].size
self.blocks.pop(i+1)
else:
i += 1
Problem 6: Detect Deadlock
Write an algorithm to detect deadlocks using resource allocation graph.
def detect_deadlock(processes, resources, allocation, request):
"""Detect deadlock using cycle detection in wait-for graph."""
# Build wait-for graph
wait_for = {p: set() for p in processes}
for p in processes:
for r in resources:
if request[p][r] > 0: # Process p wants resource r
# p waits for processes holding r
for q in processes:
if allocation[q][r] > 0:
wait_for[p].add(q)
# Detect cycle using DFS
visited = set()
rec_stack = set()
def has_cycle(node):
visited.add(node)
rec_stack.add(node)
for neighbor in wait_for[node]:
if neighbor not in visited:
if has_cycle(neighbor):
return True
elif neighbor in rec_stack:
return True
rec_stack.remove(node)
return False
for p in processes:
if p not in visited:
if has_cycle(p):
return True # Deadlock detected
return False
Problem 7: Implement POSIX Threads
Write a C program using pthreads to compute the sum of an array in parallel.
#include <pthread.h>
#include <stdio.h>
#define NUM_THREADS 4
#define ARRAY_SIZE 1000
int array[ARRAY_SIZE];
long partial_sums[NUM_THREADS];
pthread_mutex_t sum_mutex = PTHREAD_MUTEX_INITIALIZER;
typedef struct {
int start;
int end;
int thread_id;
} ThreadArgs;
void *partial_sum(void *arg) {
ThreadArgs *args = (ThreadArgs *)arg;
long sum = 0;
for (int i = args->start; i < args->end; i++) {
sum += array[i];
}
partial_sums[args->thread_id] = sum;
return NULL;
}
int main() {
// Initialize array
for (int i = 0; i < ARRAY_SIZE; i++) array[i] = i + 1;
pthread_t threads[NUM_THREADS];
ThreadArgs args[NUM_THREADS];
int chunk = ARRAY_SIZE / NUM_THREADS;
for (int i = 0; i < NUM_THREADS; i++) {
args[i] = (ThreadArgs){i * chunk, (i+1) * chunk, i};
pthread_create(&threads[i], NULL, partial_sum, &args[i]);
}
long total = 0;
for (int i = 0; i < NUM_THREADS; i++) {
pthread_join(threads[i], NULL);
total += partial_sums[i];
}
printf("Sum: %ld (expected: %d)\n", total, ARRAY_SIZE * (ARRAY_SIZE + 1) / 2);
return 0;
}
Problem 8: Segmentation with Paging
Explain how combined segmentation and paging works in Intel x86 architecture.
Answer:
- Logical Address: (Segment Selector, Offset)
- Segment Translation: Segment selector → GDT/LDT → Linear Address
- Paging: Linear Address → Page Directory → Page Table → Physical Frame
- Two-level lookup: Segment base + offset = linear, then page translation
- TLB caches recent translations for speed
Problem 9: File Allocation Methods
Compare contiguous, linked, and indexed allocation for a 10-block file.
Answer:
- Contiguous: File stored in blocks 5-14. Fast sequential access, external fragmentation.
- Linked: Each block points to next (5→8→3→12→...). No fragmentation, slow random access.
- Indexed: Index block stores pointers to all file blocks (index→[5,8,3,12,...]). Fast random access, extra block for index.
Problem 10: Implement Spinlock
Write a spinlock implementation using atomic compare-and-swap.
import threading
import time
class SpinLock:
def __init__(self):
self._locked = False
self._owner = None
self._lock = threading.Lock()
def acquire(self, timeout=None):
start = time.time()
while True:
with self._lock:
if not self._locked:
self._locked = True
self._owner = threading.current_thread().name
return True
if timeout and time.time() - start > timeout:
return False
# Busy wait — yield to avoid starving other threads
time.sleep(0.0001)
def release(self):
with self._lock:
self._locked = False
self._owner = None
def __enter__(self):
self.acquire()
return self
def __exit__(self, *args):
self.release()
# Usage
lock = SpinLock()
counter = 0
def increment():
global counter
for _ in range(10000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Counter: {counter}") # 40000