import threading
import time

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

def thread_1():
    print("線(xiàn)程1: 嘗試獲取 lock_a...")
    with lock_a:
        print("線(xiàn)程1: 已持有 lock_a，等待 lock_b...")
        time.sleep(0.1)  # 模擬處理時(shí)間，給線(xiàn)程2機(jī)會(huì)持有 lock_b
        with lock_b:     # ← 此處永遠(yuǎn)等待，lock_b 被線(xiàn)程2持有
            print("線(xiàn)程1: 同時(shí)持有 lock_a 和 lock_b")

def thread_2():
    print("線(xiàn)程2: 嘗試獲取 lock_b...")
    with lock_b:
        print("線(xiàn)程2: 已持有 lock_b，等待 lock_a...")
        time.sleep(0.1)
        with lock_a:     # ← 此處永遠(yuǎn)等待，lock_a 被線(xiàn)程1持有
            print("線(xiàn)程2: 同時(shí)持有 lock_a 和 lock_b")

t1 = threading.Thread(target=thread_1, name="Thread-1")
t2 = threading.Thread(target=thread_2, name="Thread-2")
t1.start()
t2.start()
t1.join()  # 主線(xiàn)程永遠(yuǎn)阻塞在這里
t2.join()
print("這行代碼永遠(yuǎn)不會(huì)被執(zhí)行")  # ← 程序已死鎖

from collections import defaultdict
from typing import Optional


class DeadlockDetector:
    """
    基于資源分配圖的死鎖檢測(cè)器。
    維護(hù)兩種關(guān)系：
    - allocation: 線(xiàn)程當(dāng)前持有的鎖（已分配）
    - waiting: 線(xiàn)程正在等待的鎖
    """

    def __init__(self):
        # {thread_id: lock_id}  線(xiàn)程持有的鎖
        self.allocation: dict[str, str] = {}
        # {thread_id: lock_id}  線(xiàn)程等待的鎖
        self.waiting: dict[str, str] = {}

    def thread_acquired(self, thread_id: str, lock_id: str):
        """記錄線(xiàn)程成功獲取鎖"""
        self.allocation[thread_id] = lock_id
        # 從等待列表移除
        self.waiting.pop(thread_id, None)

    def thread_waiting(self, thread_id: str, lock_id: str):
        """記錄線(xiàn)程正在等待鎖"""
        self.waiting[thread_id] = lock_id

    def thread_released(self, thread_id: str):
        """記錄線(xiàn)程釋放鎖"""
        self.allocation.pop(thread_id, None)

    def _find_lock_holder(self, lock_id: str) -> Optional[str]:
        """找到持有某個(gè)鎖的線(xiàn)程"""
        for tid, lid in self.allocation.items():
            if lid == lock_id:
                return tid
        return None

    def detect_deadlock(self) -> Optional[list[str]]:
        """
        檢測(cè)是否存在死鎖，返回死鎖環(huán)中的線(xiàn)程列表。
        算法：對(duì)每個(gè)等待中的線(xiàn)程，沿等待鏈追蹤，
        如果回到起點(diǎn)則發(fā)現(xiàn)死鎖。
        """
        for start_thread in self.waiting:
            visited = []
            current = start_thread

            while current is not None:
                if current in visited:
                    # 找到環(huán)！從環(huán)的起點(diǎn)截取
                    cycle_start = visited.index(current)
                    return visited[cycle_start:] + [current]
                visited.append(current)

                # 當(dāng)前線(xiàn)程在等待哪個(gè)鎖？
                waiting_for_lock = self.waiting.get(current)
                if not waiting_for_lock:
                    break  # 不再等待，沒(méi)有死鎖（沿此路徑）

                # 誰(shuí)持有那個(gè)鎖？
                current = self._find_lock_holder(waiting_for_lock)

        return None  # 未檢測(cè)到死鎖

    def status_report(self) -> str:
        lines = ["=== 資源分配狀態(tài) ==="]
        lines.append("持有情況:")
        for tid, lid in self.allocation.items():
            waiting = self.waiting.get(tid, "無(wú)")
            lines.append(f"  {tid} 持有 {lid}，等待 {waiting}")
        deadlock = self.detect_deadlock()
        if deadlock:
            lines.append(f"\n??  檢測(cè)到死鎖！環(huán)路: {' → '.join(deadlock)}")
        else:
            lines.append("\n? 未檢測(cè)到死鎖")
        return "\n".join(lines)


# 演示死鎖檢測(cè)
detector = DeadlockDetector()

# 模擬死鎖場(chǎng)景
detector.thread_acquired("Thread-1", "Lock-A")
detector.thread_acquired("Thread-2", "Lock-B")
detector.thread_waiting("Thread-1", "Lock-B")  # T1 等待被 T2 持有的 Lock-B
detector.thread_waiting("Thread-2", "Lock-A")  # T2 等待被 T1 持有的 Lock-A

print(detector.status_report())
# 輸出：
# === 資源分配狀態(tài) ===
# 持有情況:
#   Thread-1 持有 Lock-A，等待 Lock-B
#   Thread-2 持有 Lock-B，等待 Lock-A
# ??  檢測(cè)到死鎖！環(huán)路: Thread-1 → Thread-2 → Thread-1

import threading
from contextlib import contextmanager

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

# 給每個(gè)鎖分配唯一 ID，用于排序
LOCK_ORDER = {id(lock_a): 0, id(lock_b): 1}

@contextmanager
def acquire_locks_in_order(*locks):
    """
    按固定順序獲取多個(gè)鎖，徹底消除循環(huán)等待。
    無(wú)論調(diào)用者以什么順序傳入鎖，內(nèi)部都會(huì)按 ID 排序。
    """
    sorted_locks = sorted(locks, key=lambda l: id(l))
    acquired = []
    try:
        for lock in sorted_locks:
            lock.acquire()
            acquired.append(lock)
        yield
    finally:
        # 逆序釋放
        for lock in reversed(acquired):
            lock.release()

def safe_thread_1():
    with acquire_locks_in_order(lock_a, lock_b):  # 傳入順序無(wú)關(guān)緊要
        print("線(xiàn)程1: 安全地同時(shí)持有兩個(gè)鎖")

def safe_thread_2():
    with acquire_locks_in_order(lock_b, lock_a):  # 傳入順序不同，但內(nèi)部會(huì)排序
        print("線(xiàn)程2: 安全地同時(shí)持有兩個(gè)鎖")

t1 = threading.Thread(target=safe_thread_1)
t2 = threading.Thread(target=safe_thread_2)
t1.start(); t2.start()
t1.join(); t2.join()
print("程序正常結(jié)束，沒(méi)有死鎖！")

import threading
import time
import random
import logging

logging.basicConfig(level=logging.INFO, format="%(threadName)s: %(message)s")

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

def thread_with_timeout(first_lock, second_lock, name):
    """帶超時(shí)的鎖獲取，失敗后退避重試"""
    max_retries = 5
    for attempt in range(max_retries):
        acquired_first = first_lock.acquire(timeout=0.5)
        if not acquired_first:
            logging.info(f"獲取第一個(gè)鎖超時(shí)，第 {attempt+1} 次重試")
            time.sleep(random.uniform(0.01, 0.1))  # 隨機(jī)退避，避免活鎖
            continue

        try:
            acquired_second = second_lock.acquire(timeout=0.5)
            if not acquired_second:
                logging.info(f"獲取第二個(gè)鎖超時(shí)，釋放第一個(gè)鎖，第 {attempt+1} 次重試")
                first_lock.release()
                time.sleep(random.uniform(0.01, 0.1))
                continue

            try:
                # 成功同時(shí)持有兩個(gè)鎖
                logging.info(f"? 成功獲取兩個(gè)鎖，執(zhí)行臨界區(qū)操作")
                time.sleep(0.2)  # 模擬操作
                return True
            finally:
                second_lock.release()
        finally:
            if first_lock.locked():
                try:
                    first_lock.release()
                except RuntimeError:
                    pass

    logging.warning(f"??  達(dá)到最大重試次數(shù)，放棄執(zhí)行")
    return False

t1 = threading.Thread(target=thread_with_timeout, args=(lock_a, lock_b, "T1"), name="Thread-1")
t2 = threading.Thread(target=thread_with_timeout, args=(lock_b, lock_a, "T2"), name="Thread-2")
t1.start(); t2.start()
t1.join(); t2.join()

from dataclasses import dataclass
from typing import Optional
import copy


@dataclass
class SystemState:
    """系統(tǒng)資源狀態(tài)快照"""
    n_threads: int          # 線(xiàn)程數(shù)量
    n_resources: int        # 資源類(lèi)型數(shù)量
    available: list[int]    # 當(dāng)前可用資源向量
    allocation: list[list[int]]  # allocation[i][j]：線(xiàn)程i持有資源j的數(shù)量
    max_need: list[list[int]]    # max_need[i][j]：線(xiàn)程i最多需要資源j的數(shù)量

    @property
    def need(self) -> list[list[int]]:
        """計(jì)算每個(gè)線(xiàn)程還需要的資源量 = max_need - allocation"""
        return [
            [self.max_need[i][j] - self.allocation[i][j]
             for j in range(self.n_resources)]
            for i in range(self.n_threads)
        ]


class BankersAlgorithm:
    """
    銀行家算法實(shí)現(xiàn)。
    在分配資源前，模擬分配后的狀態(tài)，
    若系統(tǒng)仍處于安全狀態(tài)則批準(zhǔn)，否則拒絕。
    """

    def __init__(self, state: SystemState):
        self.state = state
        self.thread_names = [f"Thread-{i}" for i in range(state.n_threads)]

    def find_safe_sequence(self, state: SystemState) -> Optional[list[int]]:
        """
        安全性算法：尋找安全執(zhí)行序列。
        使用貪心策略：找到一個(gè)當(dāng)前需求能被滿(mǎn)足的線(xiàn)程，
        模擬其執(zhí)行完畢并釋放資源，重復(fù)直到所有線(xiàn)程完成。
        """
        available = state.available.copy()
        need = state.need
        finished = [False] * state.n_threads
        safe_sequence = []

        for _ in range(state.n_threads):
            # 在未完成的線(xiàn)程中，找到需求能被當(dāng)前可用資源滿(mǎn)足的
            found = False
            for i in range(state.n_threads):
                if finished[i]:
                    continue
                # 檢查線(xiàn)程 i 的所有資源需求是否都能被滿(mǎn)足
                if all(need[i][j] <= available[j] for j in range(state.n_resources)):
                    # 可以滿(mǎn)足：模擬線(xiàn)程 i 執(zhí)行完畢，釋放其持有的資源
                    for j in range(state.n_resources):
                        available[j] += state.allocation[i][j]
                    finished[i] = True
                    safe_sequence.append(i)
                    found = True
                    break

            if not found:
                # 這輪沒(méi)有任何線(xiàn)程能推進(jìn)，系統(tǒng)處于不安全狀態(tài)
                return None

        return safe_sequence  # 返回安全序列

    def request_resources(self, thread_id: int, request: list[int]) -> tuple[bool, str]:
        """
        資源請(qǐng)求算法：線(xiàn)程申請(qǐng)資源。
        返回 (是否批準(zhǔn), 原因說(shuō)明)
        """
        need = self.state.need
        thread_name = self.thread_names[thread_id]

        # 步驟1：檢查請(qǐng)求是否超過(guò)聲明的最大需求
        if any(request[j] > need[thread_id][j] for j in range(self.state.n_resources)):
            return False, f"? 拒絕：{thread_name} 的請(qǐng)求超過(guò)其聲明的最大需求"

        # 步驟2：檢查請(qǐng)求是否超過(guò)當(dāng)前可用資源
        if any(request[j] > self.state.available[j] for j in range(self.state.n_resources)):
            return False, f"? 等待：{thread_name} 請(qǐng)求的資源暫不足，需要等待"

        # 步驟3：試探性分配，檢查分配后是否仍處于安全狀態(tài)
        # 創(chuàng)建試探狀態(tài)（深拷貝）
        trial_state = copy.deepcopy(self.state)
        for j in range(self.state.n_resources):
            trial_state.available[j] -= request[j]
            trial_state.allocation[thread_id][j] += request[j]
            trial_state.max_need[thread_id][j] -= request[j]  # 更新剩余需求

        safe_sequence = self.find_safe_sequence(trial_state)

        if safe_sequence is None:
            return False, (
                f"?? 拒絕：分配給 {thread_name} 后系統(tǒng)將進(jìn)入不安全狀態(tài)，"
                f"存在死鎖風(fēng)險(xiǎn)"
            )

        # 分配安全，提交試探狀態(tài)
        self.state = trial_state
        seq_names = [self.thread_names[i] for i in safe_sequence]
        return True, (
            f"? 批準(zhǔn)：分配后系統(tǒng)仍安全，"
            f"安全執(zhí)行序列: {' → '.join(seq_names)}"
        )

    def status_report(self):
        """打印系統(tǒng)當(dāng)前狀態(tài)"""
        need = self.state.need
        print("\n" + "="*60)
        print("系統(tǒng)資源狀態(tài)報(bào)告")
        print("="*60)
        print(f"可用資源:  {self.state.available}")
        print(f"\n{'線(xiàn)程':<12} {'已分配':<20} {'最大需求':<20} {'還需要':<20}")
        print("-"*60)
        for i in range(self.state.n_threads):
            print(
                f"{self.thread_names[i]:<12} "
                f"{str(self.state.allocation[i]):<20} "
                f"{str(self.state.max_need[i]):<20} "
                f"{str(need[i]):<20}"
            )
        safe_seq = self.find_safe_sequence(self.state)
        if safe_seq:
            seq_names = [self.thread_names[i] for i in safe_seq]
            print(f"\n安全狀態(tài): ?  安全序列: {' → '.join(seq_names)}")
        else:
            print("\n安全狀態(tài): ??  不安全！存在死鎖風(fēng)險(xiǎn)")
        print("="*60 + "\n")

# 構(gòu)建初始系統(tǒng)狀態(tài)
# 3 個(gè)線(xiàn)程，4 種資源（A, B, C, D）
initial_state = SystemState(
    n_threads=3,
    n_resources=4,
    available=[3, 2, 2, 1],       # 當(dāng)前可用資源
    allocation=[                    # 當(dāng)前已分配
        [0, 1, 0, 0],              # Thread-0 已持有
        [2, 0, 1, 1],              # Thread-1 已持有
        [0, 0, 1, 0],              # Thread-2 已持有
    ],
    max_need=[                      # 最大需求聲明
        [3, 2, 1, 1],              # Thread-0 最多需要
        [4, 1, 2, 2],              # Thread-1 最多需要
        [1, 1, 2, 0],              # Thread-2 最多需要
    ],
)

banker = BankersAlgorithm(initial_state)
banker.status_report()

# 場(chǎng)景1：Thread-0 請(qǐng)求 [1, 0, 1, 0]
print("Thread-0 請(qǐng)求資源 [1, 0, 1, 0]:")
approved, reason = banker.request_resources(0, [1, 0, 1, 0])
print(reason)

# 場(chǎng)景2：Thread-1 請(qǐng)求超出需求的資源
print("\nThread-1 請(qǐng)求資源 [3, 0, 0, 0]（超出聲明上限）:")
approved, reason = banker.request_resources(1, [3, 0, 0, 0])
print(reason)

# 場(chǎng)景3：Thread-2 請(qǐng)求可能導(dǎo)致不安全狀態(tài)的資源
print("\nThread-2 請(qǐng)求資源 [0, 1, 1, 0]:")
approved, reason = banker.request_resources(2, [0, 1, 1, 0])
print(reason)

banker.status_report()

import threading

# 普通 Lock：同一線(xiàn)程兩次獲取會(huì)死鎖
# lock = threading.Lock()

# RLock（可重入鎖）：同一線(xiàn)程可以多次獲取，不會(huì)死鎖
rlock = threading.RLock()

def recursive_function(n):
    with rlock:  # 同一線(xiàn)程第二次獲取 RLock，安全！
        if n > 0:
            recursive_function(n - 1)
        print(f"層級(jí) {n} 執(zhí)行完畢")

recursive_function(3)  # 正常執(zhí)行，不會(huì)死鎖

import threading
import time

# 信號(hào)量：限制同時(shí)訪(fǎng)問(wèn)某資源的線(xiàn)程數(shù)量
db_semaphore = threading.Semaphore(3)  # 最多3個(gè)線(xiàn)程同時(shí)訪(fǎng)問(wèn)數(shù)據(jù)庫(kù)

def query_database(thread_id):
    with db_semaphore:
        print(f"Thread-{thread_id}: 獲得數(shù)據(jù)庫(kù)連接")
        time.sleep(1)  # 模擬查詢(xún)
        print(f"Thread-{thread_id}: 釋放數(shù)據(jù)庫(kù)連接")

threads = [threading.Thread(target=query_database, args=(i,)) for i in range(8)]
for t in threads: t.start()
for t in threads: t.join()