Python 模擬死鎖的常見範例詳解

前言

常見的例子是在銀行賬戶上：假如要在兩個銀行賬戶之間執行交易，你必須確保兩個賬戶都被鎖定，不受其他交易的影響，以達到正確的資金轉移量。在這裡，這個類比並不完全成立--哲學家對應的是鎖定賬戶的交易（分叉）--但同樣的技術困難也會出現。

其他的例子包括電商秒殺系統，多個使用者搶一個商品，不允許一個資料庫被多個客戶同時修改。

死鎖也是由一個並行程式需要同時具備的條件來定義的，這樣才會發生死鎖。這些條件是由電腦科學家Edward G. Coffman, Jr .首先提出的，因此被稱為 Coffman 條件。這些條件如下：

• 至少有一個資源必須處於不可共用的狀態。這意味著該資源被一個單獨的程序（或執行緒）持有，不能被其他人存取； 在任何時間內，該資源只能被單個的程序（或執行緒）存取和持有。這個條件也被稱為相互排斥。
• 有一個程序（或執行緒）同時存取一個資源並等待其他程序（或執行緒）持有的另一個資源。換句話說，這個程序（或執行緒）需要存取兩個資源來執行其指令，其中一個它已經持有，另一個它正在等待其他程序（或執行緒）。這種情況被稱為保持和等待。
• 只有在有特定指令讓程序（或執行緒）釋放資源的情況下，才能由持有這些資源的程序（或執行緒）來釋放。這就是說，除非程序（或執行緒）自願主動地釋放資源，否則該資源仍處於不可共用的狀態。這就是無搶佔條件。
• 最後一個條件叫做迴圈等待。顧名思義，這個條件規定了一組程序（或執行緒）的存在，因此這組程序中的第一個程序（或執行緒）正在等待第二個程序（或執行緒）釋放資源，而第二個程序（或執行緒）又需要等待第三個程序（或執行緒）；最後，這組程序中的最後一個程序（或執行緒）正在等待第一個程序。

造成執行緒死鎖的常見例子包括：

• 一個在自己身上等待的執行緒（例如，試圖兩次獲得同一個互斥鎖）
• 互相等待的執行緒（例如，A 等待 B，B 等待 A）
• 未能釋放資源的執行緒（例如，互斥鎖、號誌、屏障、條件、事件等）
• 執行緒以不同的順序獲取互斥鎖（例如，未能執行鎖排序）

模擬死鎖1：執行緒等待本身

導致死鎖的一個常見原因是執行緒在自己身上等待。

我們並不打算讓這種死鎖發生，例如，我們不會故意寫程式碼，導致執行緒自己等待。相反，由於一系列的函數呼叫和變數的傳遞，這種情況會意外地發生。

一個執行緒可能會因為很多原因而在自己身上等待，比如：

• 等待獲得它已經獲得的互斥鎖
• 等待自己被通知一個條件
• 等待一個事件被自己設定
• 等待一個訊號被自己釋放

開發一個 task() 函數，直接嘗試兩次獲取同一個 mutex 鎖。也就是說，該任務將獲取鎖，然後再次嘗試獲取鎖。

# task to be executed in a new thread
def task(lock):
    print('Thread acquiring lock...')
    with lock:
        print('Thread acquiring lock again...')
        with lock:
            # will never get here
            pass

這將導致死鎖，因為執行緒已經持有該鎖，並將永遠等待自己釋放該鎖，以便它能再次獲得該鎖, task() 試圖兩次獲取同一個鎖並觸發死鎖。

在主執行緒中，可以建立鎖:

# create the mutex lock
lock = Lock()

然後我們將建立並設定一個新的執行緒，在一個新的執行緒中執行我們的 task() 函數，然後啟動這個執行緒並等待它終止，而它永遠不會終止。

# create and configure the new thread
thread = Thread(target=task, args=(lock,))
# start the new thread
thread.start()
# wait for threads to exit...
thread.join()

完整程式碼如下：

from threading import Thread
from threading import Lock
# task to be executed in a new thread
def task(lock):
    print('Thread acquiring lock...')
    with lock:
        print('Thread acquiring lock again...')
        with lock:
            # will never get here
            pass
# create the mutex lock
lock = Lock()
# create and configure the new thread
thread = Thread(target=task, args=(lock,))
# start the new thread
thread.start()
# wait for threads to exit...
thread.join()

執行結果如下：

首先建立鎖，然後新的執行緒被混淆並啟動，主執行緒阻塞，直到新執行緒終止，但它從未這樣做。

新執行緒執行並首先獲得了鎖。然後它試圖再次獲得相同的互斥鎖並阻塞。

它將永遠阻塞，等待鎖被釋放。該鎖不能被釋放，因為該執行緒已經持有該鎖。因此，該執行緒已經陷入死鎖。

該程式必須被強制終止，例如，通過 Control-C 殺死終端。

模擬死鎖2：執行緒互相等待

一個常見的例子就是兩個或多個執行緒互相等待。例如：執行緒 A 等待執行緒 B，執行緒 B 等待執行緒 A。

如果有三個執行緒，可能會出現執行緒迴圈等待，例如：

• 執行緒 A：等待執行緒 B
• 執行緒 B：等待執行緒 C
• 執行緒 C：等待執行緒 A

如果你設定了執行緒來等待其他執行緒的結果，這種死鎖是很常見的，比如在一個流水線或工作流中，子任務的一些依賴關係是不符合順序的。

from threading import current_thread
from threading import Thread
# task to be executed in a new thread
def task(other):
    # message
    print(f'[{current_thread().name}] waiting on [{other.name}]...n')
    other.join()
# get the current thread
main_thread = current_thread()
# create the second thread
new_thread = Thread(target=task, args=(main_thread,))
# start the new thread
new_thread.start()
# run the first thread
task(new_thread)

首先得到主執行緒的範例 main_thread，然後建立一個新的執行緒 new_thread，並呼叫傳遞給主執行緒的 task() 函數。新執行緒返回一條資訊並等待主執行緒停止，主執行緒用新執行緒的範例呼叫 task()函數，並等待新執行緒的終止。每個執行緒都在等待另一個執行緒終止，然後自己才能終止，這導致了一個死鎖。

執行結果：

[Thread-1] waiting on [MainThread]...
[MainThread] waiting on [Thread-1]...

模擬死鎖3：以錯誤的順序獲取鎖

導致死鎖的一個常見原因是，兩個執行緒同時以不同的順序獲得鎖。例如，我們可能有一個受鎖保護的關鍵部分，在這個關鍵部分中，我們可能有程式碼或函數呼叫受第二個鎖保護。

可能會遇到這樣的情況：一個執行緒獲得了鎖 1 ，然後試圖獲得鎖 2，然後有第二個執行緒呼叫獲得鎖 2 的功能，然後試圖獲得鎖 1。如果這種情況同時發生，執行緒 1 持有鎖 1，執行緒 2 持有鎖 2，那麼就會有一個死鎖。

• 執行緒1: 持有鎖 1, 等待鎖 2
• 執行緒2 : 持有鎖 2, 等待鎖 1

from time import sleep
from threading import Thread
from threading import Lock
# task to be executed in a new thread
def task(number, lock1, lock2):
    # acquire the first lock
    print(f'Thread {number} acquiring lock 1...')
    with lock1:
        # wait a moment
        sleep(1)
        # acquire the next lock
        print(f'Thread {number} acquiring lock 2...')
        with lock2:
            # never gets here..
            pass
# create the mutex locks
lock1 = Lock()
lock2 = Lock()
# create and configure the new threads
thread1 = Thread(target=task, args=(1, lock1, lock2))
thread2 = Thread(target=task, args=(2, lock2, lock1))
# start the new threads
thread1.start()
thread2.start()
# wait for threads to exit...
thread1.join()
thread2.join()

執行這個例子首先建立了兩個鎖。然後兩個執行緒都被建立，主執行緒等待執行緒的終止。

第一個執行緒接收 lock1 和 lock2 作為引數。它獲得了鎖 1 並 sleep。

第二個執行緒接收 lock2 和 lock1 作為引數。它獲得了鎖 2 並 sleep。

第一個執行緒醒來並試圖獲取鎖 2，但它必須等待，因為它已經被第二個執行緒獲取。第二個執行緒醒來並試圖獲取鎖 1，但它必須等待，因為它已經被第一個執行緒獲取。

結果是一個死鎖：

Thread 1 acquiring lock 1...
Thread 2 acquiring lock 1...
Thread 1 acquiring lock 2...
Thread 2 acquiring lock 2...

解決辦法是確保鎖在整個程式中總是以相同的順序獲得。這就是所謂的鎖排序。

模擬死鎖4：鎖未釋放

導致死鎖的另一個常見原因是執行緒未能釋放一個資源。這通常是由執行緒在關鍵部分引發錯誤或異常造成的，這種方式會阻止執行緒釋放資源，包括：

• 未能釋放一個鎖
• 未能釋放一個訊號器
• 未能到達一個 barrier
• 未能在一個條件上通知執行緒
• 未能設定一個事件

# example of a deadlock caused by a thread failing to release a lock
from time import sleep
from threading import Thread
from threading import Lock
# task to be executed in a new thread
def task(lock):
    # acquire the lock
    print('Thread acquiring lock...')
    lock.acquire()
    # fail
    raise Exception('Something bad happened')
    # release the lock (never gets here)
    print('Thread releasing lock...')
    lock.release()
# create the mutex lock
lock = Lock()
# create and configure the new thread
thread = Thread(target=task, args=(lock,))
# start the new thread
thread.start()
# wait a while
sleep(1)
# acquire the lock
print('Main acquiring lock...')
lock.acquire()
# do something...
# release lock (never gets here)
lock.release()

執行該例子時，首先建立鎖，然後建立並啟動新的執行緒。然後主執行緒阻塞。新執行緒執行。它首先獲得了鎖，然後引發了一個異常而失敗。該執行緒解開了鎖，但卻沒有解開鎖的程式碼。新的執行緒終止了。最後，主執行緒被喚醒，然後試圖獲取鎖。由於鎖沒有被釋放，主執行緒永遠阻塞，導致了死鎖。

Thread acquiring lock...
Exception in thread Thread-1:
Traceback (most recent call last):
  ...
Exception: Something bad happened
Main acquiring lock...

總結

本文首先通過實際案例中可能出現死鎖的情況，介紹了死鎖的概念及條件，並通過 Python 程式碼模擬死鎖的四種情況，更多關於Python 模擬死鎖的資料請關注it145.com其它相關文章！