06 - 死锁¶

import numpy as np

class BankersAlgorithm:
    """银行家算法完整实现"""

    def __init__(self, available, max_matrix, allocation_matrix):
        self.available = np.array(available)
        self.max_matrix = np.array(max_matrix)
        self.allocation = np.array(allocation_matrix)
        self.need = self.max_matrix - self.allocation
        self.n_processes = len(max_matrix)
        self.n_resources = len(available)

    def is_safe(self):
        """安全性检查算法，返回 (是否安全, 安全序列)"""
        work = self.available.copy()
        finish = [False] * self.n_processes
        safe_sequence = []

        while True:
            found = False
            for i in range(self.n_processes):
                if not finish[i] and all(self.need[i] <= work):
                    # 进程 i 可以完成
                    work += self.allocation[i]
                    finish[i] = True
                    safe_sequence.append(f"P{i}")
                    found = True
                    break  # 从头重新扫描

            if not found:
                break

        is_safe = all(finish)
        return is_safe, safe_sequence

    def request_resources(self, process_id, request):
        """进程 process_id 请求资源"""
        request = np.array(request)
        pid = process_id

        # 步骤1：检查请求是否超过需求
        if any(request > self.need[pid]):
            return False, "错误：请求超过最大需求"

        # 步骤2：检查资源是否足够
        if any(request > self.available):
            return False, "资源不足，进程需等待"

        # 步骤3：试探性分配
        self.available -= request
        self.allocation[pid] += request
        self.need[pid] -= request

        # 步骤4：安全性检查
        safe, sequence = self.is_safe()

        if safe:
            return True, f"分配成功！安全序列: {' → '.join(sequence)}"
        else:
            # 回滚
            self.available += request
            self.allocation[pid] -= request
            self.need[pid] += request
            return False, "分配会导致不安全状态，已回滚"

    def display_state(self):
        """显示当前系统状态"""
        print("=" * 60)
        print(f"Available (可用资源): {self.available}")
        print()
        headers = [f"R{j}" for j in range(self.n_resources)]
        print(f"{'进程':<6} {'Allocation':<20} {'Max':<20} {'Need':<20}")
        print("-" * 60)
        for i in range(self.n_processes):
            alloc_str = str(self.allocation[i])
            max_str = str(self.max_matrix[i])
            need_str = str(self.need[i])
            print(f"P{i:<5} {alloc_str:<20} {max_str:<20} {need_str:<20}")
        print("=" * 60)

# ==================== 测试示例 ====================
if __name__ == "__main__":
    # 系统有 3 种资源 (A, B, C)
    available = [3, 3, 2]

    # 5 个进程的最大需求
    max_matrix = [
        [7, 5, 3],  # P0
        [3, 2, 2],  # P1
        [9, 0, 2],  # P2
        [2, 2, 2],  # P3
        [4, 3, 3],  # P4
    ]

    # 当前已分配资源
    allocation_matrix = [
        [0, 1, 0],  # P0
        [2, 0, 0],  # P1
        [3, 0, 2],  # P2
        [2, 1, 1],  # P3
        [0, 0, 2],  # P4
    ]

    banker = BankersAlgorithm(available, max_matrix, allocation_matrix)

    print("【初始状态】")
    banker.display_state()

    # 安全性检查
    safe, sequence = banker.is_safe()
    print(f"\n系统是否安全: {safe}")
    print(f"安全序列: {' → '.join(sequence)}")

    # P1 请求 (1, 0, 2)
    print("\n【P1 请求资源 (1, 0, 2)】")
    success, msg = banker.request_resources(1, [1, 0, 2])
    print(msg)

    if success:
        banker.display_state()

    # P4 请求 (3, 3, 0)
    print("\n【P4 请求资源 (3, 3, 0)】")
    success, msg = banker.request_resources(4, [3, 3, 0])
    print(msg)

def detect_deadlock(available, allocation, request):
    """
    死锁检测算法
    available: 当前可用资源向量
    allocation: 当前分配矩阵
    request: 当前请求矩阵（各进程等待的资源）
    返回: (是否有死锁, 死锁进程列表)
    """
    n = len(allocation)
    work = available.copy()
    finish = [False] * n

    # 初始：Allocation[i] 全为 0 的进程标记完成（不占资源）
    for i in range(n):
        if all(a == 0 for a in allocation[i]):
            finish[i] = True

    changed = True
    while changed:
        changed = False
        for i in range(n):
            if not finish[i] and all(request[i][j] <= work[j]
                                     for j in range(len(work))):
                # 进程 i 的请求可被满足 → 假设其完成并释放资源
                work = [work[j] + allocation[i][j] for j in range(len(work))]
                finish[i] = True
                changed = True

    deadlocked = [f"P{i}" for i in range(n) if not finish[i]]
    return len(deadlocked) > 0, deadlocked

# 测试
available = [0, 0, 0]
allocation = [
    [0, 1, 0],
    [2, 0, 0],
    [3, 0, 3],
    [2, 1, 1],
    [0, 0, 2],
]
request = [
    [0, 0, 0],
    [2, 0, 2],
    [0, 0, 0],
    [1, 0, 0],
    [0, 0, 2],
]

has_deadlock, processes = detect_deadlock(available, allocation, request)
print(f"存在死锁: {has_deadlock}")
print(f"死锁进程: {processes}")

import threading
import time

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread1_func():
    lock_a.acquire()
    print("Thread1: 持有 lock_a")
    time.sleep(0.1)

    print("Thread1: 等待 lock_b...")
    lock_b.acquire()  # 阻塞！Thread2 持有 lock_b
    print("Thread1: 获得 lock_b")

    lock_b.release()
    lock_a.release()

def thread2_func():
    lock_b.acquire()
    print("Thread2: 持有 lock_b")
    time.sleep(0.1)

    print("Thread2: 等待 lock_a...")
    lock_a.acquire()  # 阻塞！Thread1 持有 lock_a
    print("Thread2: 获得 lock_a")

    lock_a.release()
    lock_b.release()

t1 = threading.Thread(target=thread1_func)
t2 = threading.Thread(target=thread2_func)
t1.start()
t2.start()
t1.join(timeout=5)
t2.join(timeout=5)

if t1.is_alive() or t2.is_alive():
    print("检测到死锁！程序无法正常结束。")

# 使用 timeout 避免永久等待
if not lock_b.acquire(timeout=2):
    print("获取 lock_b 超时，释放已持有的锁")
    lock_a.release()

import threading

class DiningPhilosophers:
    """哲学家就餐问题 — 奇偶编号法"""

    def __init__(self, n=5):
        self.n = n
        self.chopsticks = [threading.Lock() for _ in range(n)]

    def philosopher(self, pid):
        left = pid
        right = (pid + 1) % self.n

        # 奇数编号的哲学家先拿右后拿左（破坏循环等待）
        if pid % 2 == 0:
            first, second = left, right
        else:
            first, second = right, left

        for _ in range(3):  # 吃3次
            # 思考
            print(f"哲学家{pid} 正在思考")

            # 拿筷子
            self.chopsticks[first].acquire()
            self.chopsticks[second].acquire()

            print(f"哲学家{pid} 正在吃饭 🍜")

            # 放下筷子
            self.chopsticks[second].release()
            self.chopsticks[first].release()

        print(f"哲学家{pid} 吃饱了")

if __name__ == "__main__":
    dp = DiningPhilosophers(5)
    threads = []
    for i in range(5):
        t = threading.Thread(target=dp.philosopher, args=(i,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("所有哲学家都吃饱了，没有发生死锁！")

import threading

chopsticks = [threading.Lock() for _ in range(5)]
semaphore = threading.Semaphore(4)  # 最多4人同时尝试拿筷子

def philosopher(pid):
    left = pid
    right = (pid + 1) % 5

    for _ in range(3):
        semaphore.acquire()  # 进入餐厅（最多4人）
        chopsticks[left].acquire()
        chopsticks[right].acquire()

        print(f"哲学家{pid} 正在吃饭")

        chopsticks[right].release()
        chopsticks[left].release()
        semaphore.release()  # 离开餐厅

threads = [threading.Thread(target=philosopher, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("无死锁完成！")

# ✗ 错误：两个线程以不同顺序获取锁
def transfer_bad(from_acc, to_acc, amount):
    from_acc.lock.acquire()
    to_acc.lock.acquire()
    # 如果另一个线程同时执行 transfer(to_acc, from_acc, ...) → 死锁！

# ✓ 正确：始终按 id 从小到大获取锁
def transfer_good(from_acc, to_acc, amount):
    first = min(from_acc, to_acc, key=lambda x: x.id)  # lambda匿名函数：简洁的单行函数
    second = max(from_acc, to_acc, key=lambda x: x.id)
    first.lock.acquire()
    second.lock.acquire()
    # 执行转账...
    second.lock.release()
    first.lock.release()

# Python: 使用 timeout
if lock.acquire(timeout=5):
    try:  # try/except捕获异常
        # 临界区操作
        pass
    finally:
        lock.release()
else:
    print("获取锁超时，进行重试或其他处理")

import threading  # 线程池/多线程：并发执行任务

lock1 = threading.Lock()
lock2 = threading.Lock()
lock3 = threading.Lock()

def worker_a():
    lock1.acquire()
    lock2.acquire()
    lock3.acquire()
    print("Worker A done")
    lock3.release()
    lock2.release()
    lock1.release()

def worker_b():
    lock2.acquire()
    lock3.acquire()
    lock1.acquire()
    print("Worker B done")
    lock1.release()
    lock3.release()
    lock2.release()

def worker_c():
    lock3.acquire()
    lock1.acquire()
    lock2.acquire()
    print("Worker C done")
    lock2.release()
    lock1.release()
    lock3.release()