ffsshhttiikk What I do
- Design hard/soft real-time systems
- Implement scheduling algorithms
- Ensure timing constraints
- Handle interrupts safely
When to use me
When building systems with strict timing requirements.
Real-Time Concepts
Task Scheduling
from enum import Enum
from typing import List
import heapq
class TaskPriority(Enum):
CRITICAL = 0 # Hard real-time
HIGH = 1
NORMAL = 2
LOW = 3
class RealTimeTask:
"""Real-time task representation"""
def __init__(self, name: str, period: float,
execution_time: float, deadline: float,
priority: TaskPriority = TaskPriority.NORMAL):
self.name = name
self.period = period
self.execution_time = execution_time
self.deadline = deadline
self.priority = priority
self.release_time = 0.0
self.completion_time = None
def utilization(self) -> float:
"""CPU utilization of task"""
return self.execution_time / self.period
class RateMonotonicScheduler:
"""Rate Monotonic Scheduling (RMS)"""
def __init__(self):
self.tasks: List[RealTimeTask] = []
def add_task(self, task: RealTimeTask):
self.tasks.append(task)
def is_schedulable(self) -> bool:
"""Check schedulability using RMS"""
n = len(self.tasks)
# Sort by period (shorter = higher priority)
sorted_tasks = sorted(self.tasks,
key=lambda t: t.period)
# Utilization bound test
total_util = sum(t.utilization() for t in sorted_tasks)
bound = n * (2 ** (1/n) - 1)
if total_util <= bound:
return True
# Response time analysis
return all(self._response_time(t, sorted_tasks)
<= t.deadline for t in sorted_tasks)
def _response_time(self, task: RealTimeTask,
tasks: List[RealTimeTask]) -> float:
"""Calculate worst-case response time"""
# Response time = execution + interference
response = task.execution_time
for higher in tasks:
if higher.period < task.period:
response += higher.execution_time
return response
class EarliestDeadlineFirst:
"""EDF Scheduling"""
def __init__(self):
self.ready_queue = []
self.current_time = 0.0
def schedule(self, tasks: List[RealTimeTask]) -> List[tuple]:
"""EDF schedule"""
schedule = []
for task in tasks:
heapq.heappush(
self.ready_queue,
(task.deadline, task)
)
while self.ready_queue:
deadline, task = heapq.heappop(self.ready_queue)
schedule.append((self.current_time, task.name))
self.current_time += task.execution_time
return schedule
Priority Inheritance
class PriorityInheritance:
"""Priority inheritance protocol"""
def __init__(self):
self.locks = {}
def acquire(self, task: str, lock_id: str,
task_priority: int):
"""Acquire lock with priority inheritance"""
if lock_id in self.locks:
# Lock held by another task
holder = self.locks[lock_id]
# Boost holder's priority
holder.inherited_priority = max(
holder.priority, task_priority
)
self.locks[lock_id] = TaskContext(task, task_priority)
def release(self, task: str, lock_id: str):
"""Release lock and reset priority"""
if lock_id in self.locks:
ctx = self.locks[lock_id]
ctx.priority = ctx.original_priority
del self.locks[lock_id]
class TaskContext:
def __init__(self, task: str, priority: int):
self.task = task
self.original_priority = priority
self.priority = priority
self.inherited_priority = priority
Real-Time Mutex
import threading
import ctypes
class RealTimeMutex:
"""Priority ceiling protocol mutex"""
def __init__(self, ceiling_priority: int):
self.ceiling_priority = ceiling_priority
self.lock = threading.Lock()
self.owner = None
self.original_priority = None
def acquire(self, task_priority: int):
"""Acquire with priority ceiling"""
# Boost priority to ceiling
if self.owner is None:
self.owner = threading.current_thread()
self.lock.acquire()
def release(self):
"""Release mutex"""
self.lock.release()
Interrupt Handling
class InterruptHandler:
"""Real-time interrupt handling"""
def __init__(self):
self.handlers = {}
self.interrupt_level = 0
def register_handler(self, interrupt_number: int,
handler: Callable, priority: int):
"""Register interrupt handler"""
self.handlers[interrupt_number] = {
"handler": handler,
"priority": priority,
"enabled": True
}
def handle_interrupt(self, interrupt_number: int):
"""Handle interrupt"""
if interrupt_number not in self.handlers:
return
info = self.handlers[interrupt_number]
if not info["enabled"]:
return
# Disable interrupts at same or lower priority
self._disable_interrupts(info["priority"])
try:
info["handler"]()
finally:
self._enable_interrupts(info["priority"])
def _disable_interrupts(self, priority: int):
"""Disable interrupts up to priority level"""
self.interrupt_level = priority
def _enable_interrupts(self, priority: int):
"""Re-enable interrupts"""
self.interrupt_level = 0
Timing Analysis
class TimingAnalysis:
"""Real-time timing analysis"""
@staticmethod
def wcet_analysis(code: List[str]) -> float:
"""Worst-Case Execution Time analysis"""
# Static analysis of code
wcet = 0.0
for instruction in code:
if "loop" in instruction:
wcet += 10.0 # Conservative estimate
elif "branch" in instruction:
wcet += 2.0
else:
wcet += 1.0
return wcet
@staticmethod
def schedulability_test(tasks: List[RealTimeTask]) -> Dict:
"""Test if task set is schedulable"""
rms = RateMonotonicScheduler()
for task in tasks:
rms.add_task(task)
return {
"schedulable": rms.is_schedulable(),
"total_utilization": sum(t.utilization() for t in tasks),
"theoretical_bound": len(tasks) * (2**(1/len(tasks)) - 1)
}