哲学家进餐问题实现异常：锁机制失效导致无法正常进食

我正在实现哲学家进餐问题的解决方案，现在遇到的问题是哲学家大部分时间都在思考，几乎不进食，不确定当前实现是否正确。我的需求是先锁定可用的左侧叉子，再锁定可用的右侧叉子后开始进食，预期输出类似：

thinking
thinking
thinking
thinking
thinking
eating number 2
thinking
eating number 1
thinking
eating number 5

但实际运行后始终只显示思考状态，该如何排查？附上我的实现代码：

import threading
import random
import time

class Fork:
    def __init__(self, index: int):
        self.index: int = index
        self.lock: threading.Lock = threading.Lock()
        self.picked_up: bool = False
        self.owner: int = -1

    def __enter__(self):
        return self

    def __call__(self, owner: int):
        if self.lock.acquire():
            self.owner = owner
            self.picked_up = True
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.lock.release()
        self.picked_up = False
        self.owner = -1

    def __str__(self):
        return f"F{self.index:2d} ({self.owner:2d})"


class Philosopher(threading.Thread):
    def __init__(self, index: int, left_fork: Fork, right_fork: Fork,eat_max:int):
        super().__init__()
        self.index: int = index
        self.left_fork: Fork = left_fork
        self.right_fork: Fork = right_fork
        self.spaghetti: int = 0
        self.eat_max:int= eat_max 

    def run(self):
        while self.spaghetti < self.eat_max:
            self.think()
            self.eat()

    def think(self):
        time.sleep(random.uniform(1, 2))
        print(f"philosopher {self.index} thinking")
        

    def eat(self):
        if not self.left_fork.picked_up:
            with self.left_fork(self.index):
                time.sleep(random.uniform(1, 2))
                if not self.right_fork.picked_up:
                    with self.right_fork(self.index):
                        self.spaghetti += 1
                        time.sleep(random.uniform(2, 2.5))
                        print(f"philosopher {self.index} eating")
                        


def main() -> None:
    n: int = 5
    m: int = 2
    forks: list[Fork] = [Fork(i) for i in range(n)]
    philosophers: list[Philosopher] = [
        Philosopher(i, forks[i], forks[(i + 1) % n], m) for i in range(n)
    ]
    for philosopher in philosophers:
        philosopher.start()
    
    for philosopher in philosophers:
        philosopher.join()

    for philosopher in philosophers:
        print(
            f"philosopher {philosopher.index} has eaten {philosopher.spaghetti} pieces"
        )

if __name__ == "__main__":
    main()

问题排查与修复 #

核心问题1：竞态条件导致锁获取逻辑失效 #

你在eat()方法中提前判断left_fork.picked_up和right_fork.picked_up，但这个判断和后续的锁获取不是原子操作——在判断完成到调用锁的间隙，其他线程可能已经拿走叉子，导致你拿着左叉子却永远拿不到右叉子，最终无法进入进食逻辑，只能释放左叉子回到思考循环。

核心问题2：Fork类的锁操作逻辑有漏洞 #

Fork.__call__方法中，如果lock.acquire()失败（锁被持有），仍然会返回自身，但此时picked_up和owner并未设置，导致状态混乱。

核心问题3：未处理死锁场景 #

所有哲学家都先拿左叉子的话，会形成循环等待，直接导致死锁，没人能拿到两个叉子进食。

修复后的代码 #

import threading
import random
import time

class Fork:
    def __init__(self, index: int):
        self.index: int = index
        self.lock: threading.Lock = threading.Lock()
        self.owner: int = -1

    def __enter__(self):
        self.lock.acquire()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.owner = -1
        self.lock.release()

    def set_owner(self, owner: int):
        self.owner = owner

class Philosopher(threading.Thread):
    def __init__(self, index: int, left_fork: Fork, right_fork: Fork, eat_max: int, is_last: bool = False):
        super().__init__()
        self.index: int = index
        # 最后一个哲学家先拿右叉子，打破循环等待避免死锁
        self.first_fork = right_fork if is_last else left_fork
        self.second_fork = left_fork if is_last else right_fork
        self.spaghetti: int = 0
        self.eat_max: int = eat_max

    def run(self):
        while self.spaghetti < self.eat_max:
            self.think()
            self.eat()

    def think(self):
        time.sleep(random.uniform(0.1, 0.5))  # 缩短思考时间，便于观察进食输出
        print(f"philosopher {self.index} thinking")

    def eat(self):
        # 直接尝试获取第一个叉子，无需提前判断状态
        with self.first_fork:
            self.first_fork.set_owner(self.index)
            # 获取第二个叉子
            with self.second_fork:
                self.second_fork.set_owner(self.index)
                self.spaghetti += 1
                time.sleep(random.uniform(0.2, 0.5))
                print(f"philosopher {self.index} eating")

def main() -> None:
    n: int = 5
    m: int = 2
    forks: list[Fork] = [Fork(i) for i in range(n)]
    philosophers: list[Philosopher] = []
    for i in range(n):
        # 标记最后一个哲学家，调整叉子获取顺序
        is_last = (i == n - 1)
        philosophers.append(Philosopher(i, forks[i], forks[(i + 1) % n], m, is_last))
    
    for philosopher in philosophers:
        philosopher.start()
    
    for philosopher in philosophers:
        philosopher.join()

    for philosopher in philosophers:
        print(f"philosopher {philosopher.index} has eaten {philosopher.spaghetti} pieces")

if __name__ == "__main__":
    main()

关键修复说明 #

1. 移除了eat()方法中多余的picked_up判断，直接通过with语句原子性获取锁，避免竞态条件。
2. 调整最后一个哲学家的叉子获取顺序，打破循环等待，解决死锁问题。
3. 简化Fork类逻辑，__enter__直接获取锁，__exit__释放锁并重置状态，避免状态不一致。
4. 缩短了思考和进食的睡眠时间，更容易观察到进食输出（可根据需求改回原时长）。

内容的提问来源于stack exchange，提问作者01d