11. Implement FCFS, SJF, and Round Robin CPU schedulers in Python.
Implement First-Come, First-Served, Shortest Job First, and Round Robin scheduling. Define the process inputs and output metrics, handle arrival and burst times, and apply the Round Robin quantum correctly.
I would use one validated process model and return the same metrics from all three schedulers. FCFS runs jobs by arrival order. Nonpreemptive SJF chooses the shortest burst among jobs that have arrived. Round Robin uses a deque and runs each ready job for at most one positive quantum. The key detail is to move the clock across idle periods and add arrivals from a completed slice before requeuing unfinished work.
See the Code while reading this explanation.
Use one process model with a unique id, an arrival time, and a positive burst time. Return start, completion, turnaround, waiting, and response time for every process. FCFS sorts by arrival and uses original input order when arrivals are equal. It runs each process to completion. SJF here is nonpreemptive. At each decision point, it selects the arrived process with the smallest burst. If no process is ready, the clock jumps to the next arrival. Round Robin keeps ready processes in collections.deque. It runs the front process for the smaller of its remaining burst and the positive quantum. Arrivals that occur during that slice enter the queue before the unfinished process is added back. Turnaround is completion minus arrival. Waiting is turnaround minus burst. Response is first start minus arrival. FCFS is simple but short work may wait behind long work. SJF can reduce average waiting when burst estimates are useful, but long work may wait too long. Round Robin improves fairness and response, but a small quantum causes more queue turns and real context switches. This code is a deterministic simulator. A production operating system also handles priorities, interrupts, context switch cost, and changing workloads.
- Should I focus on Python language behavior, or also explain the runtime and standard library?
- Which Python version and execution environment should I assume?
- Would you like a small code example together with production tradeoffs and edge cases?
The code validates every process and returns results in original input order. FCFS uses one stable sort and takes O(n log n) time. The shown SJF scans pending work for every selection and takes O(n squared) time. Round Robin sorts arrivals once and performs one constant time queue turn per slice. Its time is O(n log n plus k), where k is the total number of slices, equal to the sum of ceiling of burst divided by quantum for all processes. Every scheduler uses O(n) extra memory. For P1 with arrival 0 and burst 5, P2 with arrival 1 and burst 3, and P3 with arrival 2 and burst 1, FCFS gives completion times 5, 8, and 9. Nonpreemptive SJF gives 5, 9, and 6. Round Robin with quantum 2 gives 9, 8, and 5. All three use the same turnaround, waiting, and response formulas.
from collections import deque
from dataclasses import dataclass
from typing import Callable
@dataclass(frozen=True)
class Process:
pid: str
arrival: int
burst: int
MetricRow = dict[str, int | str]
def validate(processes: list[Process]) -> None:
# Every process needs a unique id, a nonnegative arrival, and positive work.
if len({process.pid for process in processes}) != len(processes):
raise ValueError("Process ids must be unique")
for process in processes:
if process.arrival < 0:
raise ValueError("Arrival time cannot be negative")
if process.burst <= 0:
raise ValueError("Burst time must be positive")
def make_result(
process: Process,
start: int,
completion: int,
) -> MetricRow:
# Turnaround is the total time from arrival through completion.
turnaround = completion - process.arrival
# Waiting excludes the time spent using the CPU.
waiting = turnaround - process.burst
# Response is the delay before the first CPU service.
response = start - process.arrival
return {
"pid": process.pid,
"arrival": process.arrival,
"burst": process.burst,
"start": start,
"completion": completion,
"turnaround": turnaround,
"waiting": waiting,
"response": response,
}
def fcfs(processes: list[Process]) -> list[MetricRow]:
validate(processes)
# Python sorting is stable. The index makes the equal arrival rule explicit.
ordered = sorted(
enumerate(processes),
key=lambda item: (item[1].arrival, item[0]),
)
time = 0
by_index: dict[int, MetricRow] = {}
for index, process in ordered:
# Jump over an idle period when the next process has not arrived.
start = max(time, process.arrival)
completion = start + process.burst
by_index[index] = make_result(process, start, completion)
time = completion
# Return every scheduler result in original input order.
return [by_index[index] for index in range(len(processes))]
def sjf(processes: list[Process]) -> list[MetricRow]:
validate(processes)
# This is nonpreemptive SJF. A started process runs to completion.
pending = list(enumerate(processes))
time = 0
by_index: dict[int, MetricRow] = {}
while pending:
ready = [item for item in pending if item[1].arrival <= time]
if not ready:
# No process is ready, so move directly to the next arrival.
time = min(process.arrival for _, process in pending)
ready = [item for item in pending if item[1].arrival <= time]
index, process = min(
ready,
key=lambda item: (
item[1].burst,
item[1].arrival,
item[0],
),
)
start = time
completion = start + process.burst
by_index[index] = make_result(process, start, completion)
time = completion
pending.remove((index, process))
return [by_index[index] for index in range(len(processes))]
def round_robin(
processes: list[Process],
quantum: int,
) -> list[MetricRow]:
validate(processes)
if quantum <= 0:
raise ValueError("Quantum must be positive")
ordered = sorted(
enumerate(processes),
key=lambda item: (item[1].arrival, item[0]),
)
remaining = {index: process.burst for index, process in ordered}
first_start: dict[int, int] = {}
completion: dict[int, int] = {}
ready: deque[int] = deque()
time = 0
next_arrival = 0
while next_arrival < len(ordered) or ready:
if not ready:
# Jump over an idle period.
time = max(time, ordered[next_arrival][1].arrival)
# Add every process that is available at the new time.
while next_arrival < len(ordered) and ordered[next_arrival][1].arrival <= time:
ready.append(next_arrival)
next_arrival += 1
# The queue stores positions inside the arrival ordered list.
position = ready.popleft()
original_index, process = ordered[position]
first_start.setdefault(original_index, time)
run_time = min(quantum, remaining[original_index])
time += run_time
remaining[original_index] -= run_time
# Add arrivals from this slice before requeuing unfinished work.
while next_arrival < len(ordered) and ordered[next_arrival][1].arrival <= time:
ready.append(next_arrival)
next_arrival += 1
if remaining[original_index] > 0:
ready.append(position)
else:
completion[original_index] = time
return [
make_result(
process,
first_start[index],
completion[index],
)
for index, process in enumerate(processes)
]
def print_schedule(
name: str,
scheduler: Callable[[], list[MetricRow]],
) -> None:
print(f"\n{name}")
for row in scheduler():
print(row)
if __name__ == "__main__":
sample = [
Process("P1", arrival=0, burst=5),
Process("P2", arrival=1, burst=3),
Process("P3", arrival=2, burst=1),
]
print_schedule("FCFS", lambda: fcfs(sample))
print_schedule("SJF", lambda: sjf(sample))
print_schedule(
"Round Robin with quantum 2",
lambda: round_robin(sample, quantum=2),
)These schedulers are useful in operating system teaching tools, workload simulators, interview exercises, and tests for queue based dispatch logic. Similar ideas appear in worker pools, job runners, request queues, and time sharing services. Real production schedulers usually add priorities, cancellation, resource limits, context switch cost, and dynamic workload information.
Interviewers ask this question to test whether a candidate can turn scheduling rules into correct Python state changes. It checks stable sorting, queue operations with collections.deque, clock movement, validation, tie handling, and metric calculation. It also tests whether the candidate understands arrivals, remaining work, fairness, idle periods, and the effect of the Round Robin quantum.
Common mistakes include choosing an SJF process before checking its arrival, treating SJF as preemptive without saying so, ignoring idle CPU periods, allowing a zero or negative quantum, and resetting burst time instead of tracking remaining work. Other errors include calculating response from the last start, requeuing the current Round Robin process before adding arrivals from its completed slice, using completion minus burst as waiting when arrival is not zero, and leaving equal arrival ties undefined.
State the assumptions first. Say that SJF is nonpreemptive, equal ties use original input order, and the quantum must be positive. Then explain the ready set, clock movement, and the metric formulas. Walk through one Round Robin slice and state exactly when new arrivals enter the deque.
