死锁的常见例子及 Python 模拟

1. 死锁常见例子

死锁（Deadlock）通常发生在多任务系统中，指由于系统中多个进程（或线程）相互请求对方占有的资源而造成的一种僵局，处于该状态的进程导致系统不能继续运行。下面介绍几个常见的死锁例子：

1.1 多线程死锁

在多线程编程中，死锁是很容易出现的问题。一个简单的例子就是两个线程同时尝试获得对方持有的锁，导致彼此阻塞，在没有外部干预的情况下永远无法继续执行：

import threading
def func1():
    lock1.acquire()
    print("func1 acquire lock1")
    lock2.acquire()
    print("func1 acquire lock2")
    lock2.release()
    lock1.release()
def func2():
    lock2.acquire()
    print("func2 acquire lock2")
    lock1.acquire()
    print("func2 acquire lock1")
    lock1.release()
    lock2.release()
lock1 = threading.Lock()
lock2 = threading.Lock()
t1 = threading.Thread(target=func1)
t2 = threading.Thread(target=func2)
t1.start()
t2.start()
t1.join()
t2.join()

上述代码会产生死锁，因为 func1() 和 func2() 分别锁定了 lock1 和 lock2，并在尝试获取另一个锁时阻塞，使整个程序陷入僵局。

1.2 进程间通信死锁

进程间通信（IPC）也是死锁经常发生的地方。例如，两个进程需要互相传递数据，使用 pipe 可以实现。但如果其中一个进程写入数据而另一个进程还未读取它，就会导致写进程挂起，从而产生死锁：

import os
def func1(fd):
    os.write(fd, bytes("Hello from func1\n", 'utf-8'))
    print("func1 write data")
def func2(fd):
    while True:
        data = os.read(fd, 1024)
        if not data:
            break
        print("func2 receive data: ", str(data, 'utf-8'))
pipe_in, pipe_out = os.pipe()
pid1 = os.fork()
if pid1 == 0:
    # child process 1
    os.close(pipe_out)
    func1(pipe_in)
    os._exit(0)
pid2 = os.fork()
if pid2 == 0:
    # child process 2
    os.close(pipe_in)
    func2(pipe_out)
    os._exit(0)
# parent process
os.close(pipe_in)
os.close(pipe_out)
_, status1 = os.waitpid(pid1, 0)
_, status2 = os.waitpid(pid2, 0)
print("func1 exit with status: ", status1)
print("func2 exit with status: ", status2)

上述代码中，func1 向 pipe 写入数据，func2 从 pipe 读取数据。但如果 func1 在 func2 开始读取之前就写入了数据，那么 func1 会在 os.write() 处永远阻塞，进而导致整个程序死锁。

1.3 资源不足死锁

资源不足也是产生死锁的常见原因。在一个系统中，如果某个进程占有了部分资源而又需要另外一些资源，而这些资源已被其他进程占有，那么该进程就会被挂起等待，导致死锁。

假设有两个进程 A 和 B，两个进程都需要并发地使用打印机、扫描仪、复印机等三个资源。如果 A 占有了打印机，但需要扫描仪，而此时扫描仪已被进程 B 占有，那么 A 就会被挂起等待进程 B 释放扫描仪。而进程 B 又需要打印机，但打印机已被 A 占有……如此循环下去，产生死锁。

2. Python 模拟死锁

为了更好地理解死锁的原理以及如何避免死锁，我们可以手动编写一些 Python 代码来模拟死锁的过程。

2.1 锁和条件变量

在 Python 中，可以使用 threading 模块来实现多线程编程。在多线程编程中，为了避免线程之间的竞争和冲突，可以使用锁和条件变量。

锁是多线程编程中最基本的同步机制。它能够确保同一时刻只有一个线程在执行关键代码段。在 Python 中，可以使用 threading.Lock 类来创建锁：

import threading
lock = threading.Lock()

当一个线程调用 lock.acquire() 时，如果当前锁还未被其他线程占用，那么该线程会立即占用这个锁，并继续执行。如果当前锁已被其他线程占用，那么此时该线程会阻塞，一直等到这个锁被释放为止。锁的释放可以通过调用 lock.release() 实现。

条件变量则用于在多个线程之间传递信号。它通常被用于生产者-消费者模型中，用于控制消费者仅在有数据时才执行操作。在 Python 中，可以使用 threading.Condition 类实现条件变量：

import threading
cond = threading.Condition()

条件变量的主要方法包括 wait()、notify() 以及 notify_all()。wait() 会使当前线程阻塞，直到其他线程调用 notify() 或 notify_all() 为止。notify() 则会随机唤醒一个等待线程，而 notify_all() 会唤醒所有等待线程。

2.2 模拟多线程死锁

以下代码模拟了两个线程之间的死锁：

import threading
import time
def func1():
    lock1.acquire()
    print("func1 acquire lock1")
    time.sleep(1)
    lock2.acquire()
    print("func1 acquire lock2")
    lock2.release()
    lock1.release()
def func2():
    lock2.acquire()
    print("func2 acquire lock2")
    time.sleep(1)
    lock1.acquire()
    print("func2 acquire lock1")
    lock1.release()
    lock2.release()
lock1 = threading.Lock()
lock2 = threading.Lock()
t1 = threading.Thread(target=func1)
t2 = threading.Thread(target=func2)
t1.start()
t2.start()
t1.join()
t2.join()

该代码中，func1 和 func2 分别尝试获得 lock1 和 lock2。如果 t1 先获得了 lock1，但在尝试获取 lock2 时被 t2 占用，而 t2 又在等待 lock1 的释放，双方都无法继续执行，产生死锁。

2.3 模拟进程间通信死锁

以下代码模拟了两个进程之间通过 pipe 通信的死锁：

import os
import time
def func1(fd):
    os.write(fd, bytes("Hello from func1\n", 'utf-8'))
    print("func1 write data")
    time.sleep(1)
    os.read(fd, 1024)
def func2(fd):
    while True:
        data = os.read(fd, 1024)
        if not data:
            break
        print("func2 receive data: ", str(data, 'utf-8'))
    os.write(fd, bytes("Hello from func2\n", 'utf-8'))
pipe_in, pipe_out = os.pipe()
pid1 = os.fork()
if pid1 == 0:
    # child process 1
    os.close(pipe_out)
    func1(pipe_in)
    os._exit(0)
pid2 = os.fork()
if pid2 == 0:
    # child process 2
    os.close(pipe_in)
    func2(pipe_out)
    os._exit(0)
# parent process
os.close(pipe_in)
os.close(pipe_out)
_, status1 = os.waitpid(pid1, 0)
_, status2 = os.waitpid(pid2, 0)
print("func1 exit with status: ", status1)
print("func2 exit with status: ", status2)

该代码中，func1 向 pipe 写入数据并等待 1 秒，然后再尝试从 pipe 读取数据。func2 则不断从 pipe 读取数据，直到读到空数据为止，然后向 pipe 写入数据。

当 func1 向 pipe 写入数据后，它会立即挂起等待从 pipe 中读取数据，但此时 func2 尚未开始读取数据，因此 func1 会一直阻塞在 os.read() 处，直到 func2 开始读取为止。而 func2 则会在读到 func1 发送的数据后尝试向 pipe 写入数据，但此时 pipe 已被 func1 占用，而 func1 又在等待从 pipe 中读取数据，双方都无法继续执行，产生死锁。

2.4 避免死锁的方法

为避免死锁，应该尽可能遵循以下规则：

尽可能减少持锁时间。

避免多个锁的情况。

确保加锁顺序的一致性。

使用超时机制防止死锁。

使用死锁检测和恢复算法。

例如，在上文中的多线程死锁例子中，可以调整加锁顺序，例如使用以下代码：

def func1():
    lock1.acquire()
    print("func1 acquire lock1")
    time.sleep(1)
    lock2.acquire()
    print("func1 acquire lock2")
    lock2.release()
    lock1.release()
def func2():
    lock1.acquire()
    print("func2 acquire lock1")
    time.sleep(1)
    lock2.acquire()
    print("func2 acquire lock2")
    lock2.release()
    lock1.release()

在该代码中，func1 和 func2 都按照相同的顺序占用了锁，因此不会产生死锁。

而在进程间通信的死锁例子中，则可以使用 select 等机制检查 pipe 是否可写，避免写进程被阻塞，例如以下代码：

import select
def func1(fd):
    while True:
        ret = select.select([], [fd], [], 1)
        if fd in ret[1]:
            os.write(fd, bytes("Hello from func1\n", 'utf-8'))
            print("func1 write data")
            break
    time.sleep(1)
    while True:
        ret = select.select([fd], [], [], 10)
        if fd in ret[0]:
            data = os.read(fd, 1024)
            print("func1 receive data: ", str(data, 'utf-8'))
            break
def func2(fd):
    while True:
        ret = select.select([fd], [], [], 1)
        if fd in ret[0]:
            data = os.read(fd, 1024)
            print("func2 receive data: ", str(data, 'utf-8'))
            os.write(fd, bytes("Hello from func2\n", 'utf-8'))
            break
pipe_in, pipe_out = os.pipe()
pid1 = os.fork()
if pid1 == 0:
    # child process 1
    os.close(pipe_out)
    func1(pipe_in)
    os._exit(0)
pid2 = os.fork()
if pid2 == 0:
    # child process 2
    os.close(pipe_in)
    func2(pipe_out)
    os._exit(0)
# parent process
os.close(pipe_in)
os.close(pipe_out)
_, status1 = os.waitpid(pid1, 0)
_, status2 = os.waitpid(pid2, 0)
print("func1 exit with status: ", status1)
print("func2 exit with status: ", status2)

通过使用 select 等机制，可以在写进程阻塞之前检查 pipe 是否可写，避免了写进程被阻塞的情况。

总结

死锁是多任务系统中一个常见且难以解决的问题。为了避免死锁，应该遵循尽可能减少持锁时间、避免多个锁的情况、确保加锁顺序的一致性、使用超时机制防止死锁以及使用死锁检测和恢复算法等原则。在编写多线程和进程间通信代码时，还可以使用锁和条件变量来避免竞争和冲突。需要注意的是，死锁的产生通常需要复杂的条件和操作，因此在编写代码时要仔细考虑各种情况，以避免死锁的发生。