操作系统核心算法实战：5大调度与3种死锁处理代码实现

操作系统核心算法实战：5大调度与3种死锁处理代码实现

在计算机科学领域，操作系统作为连接硬件与软件的桥梁，其核心算法的理解与实现是每位开发者必须掌握的技能。本文将聚焦处理机调度与死锁处理两大核心主题，通过Python代码实现五种经典调度算法和三种死锁处理策略，帮助读者从理论走向实践。

1. 处理机调度算法原理与实现

处理机调度算法决定了系统如何分配CPU时间给各个进程，直接影响系统的吞吐量、响应时间和公平性。我们将实现五种经典算法，并分析其适用场景。

1.1 先来先服务(FCFS)调度

FCFS是最简单的调度算法，按照进程到达的顺序进行服务。其特点是实现简单但可能导致"护航效应"。

def fcfs_scheduling(processes): current_time = 0 waiting_time = 0 processes.sort(key=lambda x: x['arrival_time']) for process in processes: if current_time < process['arrival_time']: current_time = process['arrival_time'] process['start_time'] = current_time process['finish_time'] = current_time + process['burst_time'] process['waiting_time'] = current_time - process['arrival_time'] waiting_time += process['waiting_time'] current_time = process['finish_time'] avg_waiting = waiting_time / len(processes) return processes, avg_waiting

提示：FCFS在负载较轻且进程执行时间相近时表现良好，但在长短进程混合的场景下性能较差。

1.2 短作业优先(SJF)调度

SJF选择预计执行时间最短的进程优先执行，可以最小化平均等待时间。

def sjf_scheduling(processes): current_time = 0 waiting_time = 0 completed = 0 n = len(processes) remaining_time = [p['burst_time'] for p in processes] while completed != n: ready_processes = [i for i in range(n) if processes[i]['arrival_time'] <= current_time and remaining_time[i] > 0] if not ready_processes: current_time += 1 continue shortest = min(ready_processes, key=lambda x: remaining_time[x]) processes[shortest]['start_time'] = current_time current_time += remaining_time[shortest] processes[shortest]['finish_time'] = current_time processes[shortest]['waiting_time'] = ( processes[shortest]['start_time'] - processes[shortest]['arrival_time']) waiting_time += processes[shortest]['waiting_time'] remaining_time[shortest] = 0 completed += 1 avg_waiting = waiting_time / n return processes, avg_waiting

1.3 优先级调度(Priority Scheduling)

每个进程被赋予优先级，系统总是选择优先级最高的进程执行。

def priority_scheduling(processes): current_time = 0 waiting_time = 0 processes.sort(key=lambda x: (x['arrival_time'], x['priority'])) ready_queue = [] completed = 0 n = len(processes) while completed < n: # 添加到达的进程到就绪队列 while processes and processes[0]['arrival_time'] <= current_time: ready_queue.append(processes.pop(0)) if not ready_queue: current_time += 1 continue # 选择优先级最高的进程(数字越小优先级越高) ready_queue.sort(key=lambda x: x['priority']) current_process = ready_queue.pop(0) current_process['start_time'] = current_time current_process['finish_time'] = current_time + current_process['burst_time'] current_process['waiting_time'] = current_time - current_process['arrival_time'] waiting_time += current_process['waiting_time'] current_time = current_process['finish_time'] completed += 1 avg_waiting = waiting_time / n return processes, avg_waiting

1.4 轮转调度(Round Robin)

RR算法为每个进程分配一个时间片，当时间片用完时，将CPU分配给下一个进程。

def round_robin(processes, time_quantum): current_time = 0 waiting_time = 0 remaining_time = [p['burst_time'] for p in processes] arrival_time = [p['arrival_time'] for p in processes] n = len(processes) ready_queue = [] completed = 0 while completed < n: # 添加到达的进程 for i in range(n): if (arrival_time[i] <= current_time and remaining_time[i] > 0 and i not in ready_queue): ready_queue.append(i) if not ready_queue: current_time += 1 continue current_process = ready_queue.pop(0) execution_time = min(time_quantum, remaining_time[current_process]) if remaining_time[current_process] == processes[current_process]['burst_time']: processes[current_process]['start_time'] = current_time remaining_time[current_process] -= execution_time current_time += execution_time if remaining_time[current_process] == 0: processes[current_process]['finish_time'] = current_time processes[current_process]['waiting_time'] = ( processes[current_process]['finish_time'] - processes[current_process]['arrival_time'] - processes[current_process]['burst_time']) waiting_time += processes[current_process]['waiting_time'] completed += 1 else: ready_queue.append(current_process) avg_waiting = waiting_time / n return processes, avg_waiting

1.5 多级反馈队列(MLFQ)调度

MLFQ结合了多种调度算法的优点，通过多级队列和动态优先级调整来平衡响应时间和吞吐量。

def mlfq_scheduling(processes, queues=3, base_quantum=4): current_time = 0 waiting_time = 0 n = len(processes) remaining_time = [p['burst_time'] for p in processes] arrival_time = [p['arrival_time'] for p in processes] queue_levels = [[] for _ in range(queues)] completed = 0 # 初始化进程状态 for i in range(n): if arrival_time[i] <= current_time: queue_levels[0].append(i) while completed < n: for level in range(queues): quantum = base_quantum * (2 ** level) if queue_levels[level]: current_process = queue_levels[level].pop(0) exec_time = min(quantum, remaining_time[current_process]) if remaining_time[current_process] == processes[current_process]['burst_time']: processes[current_process]['start_time'] = current_time remaining_time[current_process] -= exec_time current_time += exec_time # 检查是否有新进程到达 for i in range(n): if (arrival_time[i] <= current_time and remaining_time[i] == processes[i]['burst_time'] and i != current_process): queue_levels[0].append(i) if remaining_time[current_process] == 0: processes[current_process]['finish_time'] = current_time processes[current_process]['waiting_time'] = ( processes[current_process]['finish_time'] - processes[current_process]['arrival_time'] - processes[current_process]['burst_time']) waiting_time += processes[current_process]['waiting_time'] completed += 1 else: if level < queues - 1: queue_levels[level + 1].append(current_process) else: queue_levels[level].append(current_process) break else: current_time += 1 avg_waiting = waiting_time / n return processes, avg_waiting

2. 死锁处理策略与实现

死锁是指多个进程因竞争资源而造成的一种互相等待的现象，若无外力作用，这些进程都将无法向前推进。我们将实现三种主要的死锁处理策略。

2.1 死锁预防

死锁预防通过破坏四个必要条件中的一个或多个来避免死锁发生。

class DeadlockPrevention: def __init__(self, processes, resources): self.processes = processes # 进程列表 self.resources = resources # 资源总量 self.allocated = {r: 0 for r in resources} # 已分配资源 self.max_claim = {p: {r: 0 for r in resources} for p in processes} # 最大需求 self.need = {p: {r: 0 for r in resources} for p in processes} # 还需要 def set_max_claim(self, process, claims): """设置进程的最大资源需求""" for resource, amount in claims.items(): self.max_claim[process][resource] = amount self.need[process][resource] = amount def request_resources(self, process, requests): """进程请求资源""" # 检查请求是否超过声明 for resource, amount in requests.items(): if amount > self.need[process][resource]: raise ValueError("请求超过最大声明") if amount > (self.resources[resource] - self.allocated[resource]): # 采用资源预分配策略，拒绝请求 return False # 尝试分配 temp_allocated = self.allocated.copy() temp_need = {p: n.copy() for p, n in self.need.items()} for resource, amount in requests.items(): temp_allocated[resource] += amount temp_need[process][resource] -= amount # 检查系统是否处于安全状态 if self.is_safe(temp_allocated, temp_need): # 实际分配资源 for resource, amount in requests.items(): self.allocated[resource] += amount self.need[process][resource] -= amount return True else: # 拒绝分配，预防死锁 return False def is_safe(self, allocated, need): """检查系统是否处于安全状态""" work = {r: self.resources[r] - allocated[r] for r in self.resources} finish = {p: False for p in self.processes} while True: found = False for p in self.processes: if not finish[p] and all(need[p][r] <= work[r] for r in self.resources): # 可以执行完成并释放资源 for r in self.resources: work[r] += allocated[r] if p in allocated else 0 finish[p] = True found = True break if not found: break return all(finish.values())

2.2 死锁避免（银行家算法）

银行家算法通过动态检查资源分配状态来避免系统进入不安全状态。

class BankersAlgorithm: def __init__(self, processes, resources): self.processes = processes self.resources = resources self.max = {p: {r: 0 for r in resources} for p in processes} self.allocation = {p: {r: 0 for r in resources} for p in processes} self.need = {p: {r: 0 for r in resources} for p in processes} self.available = resources.copy() def set_max(self, process, max_claims): """设置进程的最大资源需求""" for resource, amount in max_claims.items(): self.max[process][resource] = amount self.need[process][resource] = amount def request_resources(self, process, request): """处理资源请求""" # 步骤1: 检查请求是否超过声明 for resource in self.resources: if request[resource] > self.need[process][resource]: raise ValueError(f"进程{process}请求超过其最大声明") # 步骤2: 检查是否有足够可用资源 for resource in self.resources: if request[resource] > self.available[resource]: return False # 必须等待 # 步骤3: 尝试分配 old_available = self.available.copy() old_allocation = {p: a.copy() for p, a in self.allocation.items()} old_need = {p: n.copy() for p, n in self.need.items()} for resource in self.resources: self.available[resource] -= request[resource] self.allocation[process][resource] += request[resource] self.need[process][resource] -= request[resource] # 步骤4: 检查安全性 if not self.is_safe(): # 回滚 self.available = old_available self.allocation = old_allocation self.need = old_need return False return True def is_safe(self): """检查当前状态是否安全""" work = self.available.copy() finish = {p: False for p in self.processes} safe_sequence = [] while True: found = False for p in self.processes: if not finish[p] and all(self.need[p][r] <= work[r] for r in self.resources): # 可以执行完成并释放资源 for r in self.resources: work[r] += self.allocation[p][r] finish[p] = True safe_sequence.append(p) found = True break if not found: break if all(finish.values()): print(f"安全序列: {safe_sequence}") return True else: return False

2.3 死锁检测与恢复

当系统允许死锁发生时，需要通过检测算法识别死锁并采取恢复措施。

class DeadlockDetection: def __init__(self, processes, resources): self.processes = processes self.resources = resources self.allocation = {p: {r: 0 for r in resources} for p in processes} self.request = {p: {r: 0 for r in resources} for p in processes} self.available = resources.copy() def set_allocation(self, process, allocation): """设置已分配资源""" for resource, amount in allocation.items(): self.allocation[process][resource] = amount self.available[resource] -= amount def set_request(self, process, request): """设置进程的资源请求""" for resource, amount in request.items(): self.request[process][resource] = amount def detect(self): """检测死锁""" work = self.available.copy() finish = {p: False for p in self.processes} deadlocked = [] # 初始化finish: 没有已分配资源的进程视为已完成 for p in self.processes: if all(self.allocation[p][r] == 0 for r in self.resources): finish[p] = True while True: found = False for p in self.processes: if not finish[p] and all(self.request[p][r] <= work[r] for r in self.resources): # 可以执行完成并释放资源 for r in self.resources: work[r] += self.allocation[p][r] finish[p] = True found = True break if not found: break # 未完成的进程处于死锁状态 deadlocked = [p for p in self.processes if not finish[p]] return deadlocked def recovery(self, deadlocked): """死锁恢复策略""" if not deadlocked: return "无死锁" print(f"检测到死锁进程: {deadlocked}") # 策略1: 终止所有死锁进程 # for p in deadlocked: # self.terminate(p) # return "已终止所有死锁进程" # 策略2: 逐个终止进程直到死锁解除 terminated = [] while deadlocked: # 选择牺牲者(这里简单选择第一个) victim = deadlocked[0] self.terminate(victim) terminated.append(victim) # 重新检测 deadlocked = self.detect() if not deadlocked: return f"已终止进程{terminated}解除死锁" return "恢复失败" def terminate(self, process): """终止进程并释放资源""" for r in self.resources: self.available[r] += self.allocation[process][r] self.allocation[process][r] = 0 self.request[process][r] = 0

3. 性能对比与场景分析

了解不同算法的特性后，我们需要在实际场景中选择合适的策略。本节将通过实验对比各算法的性能指标。

3.1 调度算法性能指标

我们定义以下关键指标来评估调度算法：

3.2 实验设计与结果

我们设计了三组不同特性的进程集合进行测试：

测试集1: CPU密集型进程

processes_cpu = [ {'pid': 1, 'arrival_time': 0, 'burst_time': 8, 'priority': 3}, {'pid': 2, 'arrival_time': 1, 'burst_time': 6, 'priority': 1}, {'pid': 3, 'arrival_time': 2, 'burst_time': 4, 'priority': 2}, {'pid': 4, 'arrival_time': 3, 'burst_time': 2, 'priority': 4} ]

测试集2: I/O密集型进程

processes_io = [ {'pid': 1, 'arrival_time': 0, 'burst_time': 4, 'priority': 2}, {'pid': 2, 'arrival_time': 0, 'burst_time': 2, 'priority': 1}, {'pid': 3, 'arrival_time': 2, 'burst_time': 6, 'priority': 3}, {'pid': 4, 'arrival_time': 4, 'burst_time': 8, 'priority': 4}, {'pid': 5, 'arrival_time': 6, 'burst_time': 2, 'priority': 2} ]

测试集3: 混合型进程

processes_mixed = [ {'pid': 1, 'arrival_time': 0, 'burst_time': 10, 'priority': 3}, {'pid': 2, 'arrival_time': 1, 'burst_time': 2, 'priority': 1}, {'pid': 3, 'arrival_time': 2, 'burst_time': 5, 'priority': 2}, {'pid': 4, 'arrival_time': 3, 'burst_time': 1, 'priority': 4}, {'pid': 5, 'arrival_time': 4, 'burst_time': 8, 'priority': 2}, {'pid': 6, 'arrival_time': 5, 'burst_time': 3, 'priority': 3} ]

实验结果对比如下：

3.3 死锁处理策略对比

同样，我们对三种死锁处理策略进行比较：

4. 现代操作系统中的调度优化

现代操作系统如Linux、Windows等采用了更为复杂的调度策略，本节分析这些系统中的优化点。

4.1 Linux CFS调度器

完全公平调度器(CFS)通过虚拟运行时间(vruntime)来实现公平性：

// 简化的CFS核心逻辑 struct sched_entity { u64 vruntime; // 虚拟运行时间 u64 exec_start; // 本次开始执行时间 // 其他字段... }; void update_curr(struct cfs_rq *cfs_rq) { struct sched_entity *curr = cfs_rq->curr; u64 now = rq_clock_task(rq_of(cfs_rq)); u64 delta_exec; delta_exec = now - curr->exec_start; curr->exec_start = now; curr->vruntime += calc_delta_fair(delta_exec, curr); update_min_vruntime(cfs_rq); } struct sched_entity *pick_next_entity(struct cfs_rq *cfs_rq) { // 选择vruntime最小的进程 struct rb_node *left = rb_first_cached(&cfs_rq->tasks_timeline); return rb_entry(left, struct sched_entity, run_node); }

4.2 Windows优先级调度

Windows采用基于优先级的抢占式调度，包含32个优先级级别：

Windows优先级层次： ------------------------------- 31 | 实时级(Real-time) 30 | ...| 16 | 高(High) 15 | ...| 1 | 低(Low) 0 | 零页线程(Zero Page Thread)

4.3 多核处理器调度挑战

多核环境下面临的新问题及解决方案：

1. 缓存亲和性：尽量使进程在同一个CPU核上运行
2. 负载均衡：避免某些核过载而其他核空闲
3. NUMA架构：考虑内存访问延迟差异

# 简化的多核负载均衡逻辑 def load_balance(rq): busiest = find_busiest_queue() if not busiest: return imbalance = calculate_imbalance(rq, busiest) if imbalance < THRESHOLD: return tasks = get_tasks_to_move(busiest, imbalance) for task in tasks: migrate_task(task, busiest, rq)