Earliest Deadline First (EDF): Real-Time Scheduling Algorithm Explained

import heapq

class Task:
    def __init__(self, name, period, execution_time):
        self.name = name
        self.period = period
        self.execution_time = execution_time
        self.deadline = period  # implicit deadline
        self.remaining_time = execution_time
        self.next_release = 0

    def __lt__(self, other):
        return self.deadline < other.deadline

def edf_schedule(tasks, simulation_time):
    ready_queue = []
    time = 0
    while time < simulation_time:
        # Release tasks whose release time has arrived
        for task in tasks:
            if task.next_release <= time:
                task.deadline = time + task.period
                task.remaining_time = task.execution_time
                heapq.heappush(ready_queue, task)
                task.next_release += task.period
        if not ready_queue:
            time += 1
            continue
        current = heapq.heappop(ready_queue)
        # Execute for 1 time unit
        current.remaining_time -= 1
        print(f"Time {time}: {current.name} (deadline {current.deadline})")
        if current.remaining_time > 0:
            heapq.heappush(ready_queue, current)
        else:
            if time + 1 > current.deadline:
                print(f"Deadline miss for {current.name} at time {time+1}")
        time += 1

if __name__ == "__main__":
    tasks = [Task("A", 5, 2), Task("B", 8, 3)]
    edf_schedule(tasks, 20)

def edf_schedulable(tasks):
    # tasks: list of (C, T, D) with D <= T
    # Returns True if schedulable under EDF
    total_util = sum(C / T for C, T, D in tasks)
    if total_util > 1:
        return False
    # For constrained deadlines, use demand bound function
    # Check up to hyperperiod (LCM of periods) or a limit
    from math import gcd
    from functools import reduce
    def lcm(a, b):
        return a * b // gcd(a, b)
    hyperperiod = reduce(lcm, [T for _, T, _ in tasks])
    for t in range(1, hyperperiod + 1):
        demand = 0
        for C, T, D in tasks:
            if t >= D:
                demand += ((t - D) // T + 1) * C
        if demand > t:
            return False
    return True

# Example
if __name__ == "__main__":
    tasks = [(2, 5, 5), (3, 8, 8)]  # C, T, D
    print(edf_schedulable(tasks))  # True

# Compare RMS and EDF for a task set
# Task set: A (C=2, T=5), B (C=3, T=8)
# RMS: A higher priority (shorter period)
# EDF: dynamic
# Simulate both and count deadline misses

def simulate_rms(tasks, sim_time):
    # tasks: list of (name, C, T)
    # Fixed priority: shorter T higher
    tasks_sorted = sorted(tasks, key=lambda x: x[2])  # by period
    time = 0
    releases = {t[0]: 0 for t in tasks_sorted}
    remaining = {t[0]: 0 for t in tasks_sorted}
    deadlines = {t[0]: 0 for t in tasks_sorted}
    misses = 0
    while time < sim_time:
        # Release tasks
        for name, C, T in tasks_sorted:
            if releases[name] <= time:
                remaining[name] = C
                deadlines[name] = time + T
                releases[name] += T
        # Find highest priority task with remaining time
        selected = None
        for name, C, T in tasks_sorted:
            if remaining[name] > 0:
                selected = name
                break
        if selected:
            remaining[selected] -= 1
            if remaining[selected] == 0 and time + 1 > deadlines[selected]:
                misses += 1
        time += 1
    return misses

# Similarly for EDF (omitted for brevity)
print("RMS misses:", simulate_rms([("A",2,5),("B",3,8)], 100))

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sched.h>
#include <linux/sched/types.h>

int main() {
    struct sched_attr attr;
    attr.size = sizeof(attr);
    attr.sched_policy = SCHED_DEADLINE;
    attr.sched_runtime = 2 * 1000 * 1000; // 2ms in ns
    attr.sched_period = 5 * 1000 * 1000;  // 5ms
    attr.sched_deadline = 5 * 1000 * 1000; // implicit deadline
    if (sched_setattr(0, &attr, 0) != 0) {
        perror("sched_setattr");
        return 1;
    }
    // Do real-time work
    while (1) {
        // Busy work for 2ms
        volatile int i;
        for (i = 0; i < 1000000; i++);
        // Yield to avoid overrun
        sched_yield();
    }
    return 0;
}

class CBS:
    def __init__(self, budget, period):
        self.budget = budget
        self.period = period
        self.remaining_budget = budget
        self.deadline = period
        self.active = False

    def replenish(self, time):
        if time >= self.deadline:
            self.remaining_budget = self.budget
            self.deadline += self.period
            self.active = True

    def consume(self, amount):
        if self.active and self.remaining_budget >= amount:
            self.remaining_budget -= amount
            return True
        return False

# Usage in scheduler
cbs = CBS(2, 5)
time = 0
while time < 20:
    cbs.replenish(time)
    if cbs.consume(1):
        print(f"Time {time}: task runs")
    else:
        print(f"Time {time}: idle")
    time += 1

# Simulate tracing scheduling events
import time

def trace_schedule(tasks, sim_time):
    # Simplified trace
    events = []
    time = 0
    while time < sim_time:
        # ... scheduling logic ...
        events.append((time, "context_switch", "A"))
        time += 1
    return events

events = trace_schedule([], 10)
for e in events:
    print(e)