當(dāng)你有多個(gè)進(jìn)程或線程訪問(wèn)相同的數(shù)據(jù)時(shí)，競(jìng)爭(zhēng)條件是一個(gè)威脅。本文探討了在發(fā)現(xiàn)競(jìng)爭(zhēng)條件后如何測(cè)試它們。

Incrmnt

你在一個(gè)名為“Incrmnt”的火熱新創(chuàng)公司工作，該公司只做一件事情，并且做得比較好。

你展示一個(gè)全局計(jì)數(shù)器和一個(gè)加號(hào)，用戶可以點(diǎn)擊加號(hào)，此時(shí)計(jì)數(shù)器加一。這太簡(jiǎn)單了，而且容易使人上癮。毫無(wú)疑問(wèn)這就是接下來(lái)的大事情。

投資者們爭(zhēng)先恐后的進(jìn)入了董事會(huì)，但你有一個(gè)大問(wèn)題。

競(jìng)爭(zhēng)條件

在你的內(nèi)測(cè)中，Abraham和Belinda是如此的興奮，以至于每個(gè)人都點(diǎn)了100次加號(hào)按鈕。你的服務(wù)器日志顯示了200次請(qǐng)求，但計(jì)數(shù)器卻顯示為173。很明顯，有一些請(qǐng)求沒(méi)有被加上。

先將“Incrmnt變成了一坨屎”的新聞拋到腦后，你檢查下代碼（本文用到的所有代碼都能在Github上找到）。
 

# incrmnt.pyimport db def increment():  count = db.get_count()   new_count = count + 1  db.set_count(new_count)   return new_count

你的Web服務(wù)器使用多進(jìn)程處理流量請(qǐng)求，所以這個(gè)函數(shù)能在不同的線程中同時(shí)執(zhí)行。如果你沒(méi)掌握好時(shí)機(jī)，將會(huì)發(fā)生：
 

# 線程1和線程2在不同的進(jìn)程中同時(shí)執(zhí)行# 為了展示的目的，在這里并排放置# 在垂直方向分開(kāi)它們，以說(shuō)明在每個(gè)時(shí)間點(diǎn)上執(zhí)行什么代碼# Thread 1（線程1）         # Thread 2（線程2）def increment():                  def increment():  # get_count returns 0  count = db.get_count()                    # get_count returns 0 again                    count = db.get_count()  new_count = count + 1  # set_count called with 1  db.set_count(new_count)                    new_count = count + 1                    # set_count called with 1 again                    db.set_count(new_count)

所以盡管增加了兩次計(jì)數(shù)，但最終只增加了1。

你知道你可以修改這個(gè)代碼，變?yōu)榫€程安全的，但是在你那么做之前，你還想寫(xiě)一個(gè)測(cè)試證明競(jìng)爭(zhēng)的存在。

重現(xiàn)競(jìng)爭(zhēng)

在理想情況下，測(cè)試應(yīng)該盡可能的重現(xiàn)上面的場(chǎng)景。競(jìng)爭(zhēng)的關(guān)鍵因素是：

?兩個(gè) get_count 調(diào)用必須在兩個(gè) set_count 調(diào)用之前執(zhí)行，從而使得兩個(gè)線程中的計(jì)數(shù)具有相同的值。

set_count 調(diào)用，什么時(shí)候執(zhí)行都沒(méi)關(guān)系，只要它們都在 get_count 調(diào)用之后即可。

簡(jiǎn)單起見(jiàn)，我們?cè)囍噩F(xiàn)這個(gè)嵌套的情形。這里整 個(gè)Thread 2 在 Thread 1 的首個(gè) get_count 調(diào)用之后執(zhí)行：

# Thread 1             # Thread 2def increment():  # get_count returns 0  count = db.get_count()                  def increment():                    # get_count returns 0 again                    count = db.get_count()                     # set_count called with 1                    new_count = count + 1                    db.set_count(new_count)  # set_count called with 1 again  new_count = count + 1  db.set_count(new_count)

before_after 是一個(gè)庫(kù)，它提供了幫助重現(xiàn)這種情形的工具。它可以在一個(gè)函數(shù)之前或之后插入任意代碼。

before_after 依賴(lài)于 mock 庫(kù)，它用來(lái)補(bǔ)充一些功能。如果你不熟悉 mock，我建議閱讀一些優(yōu)秀的文檔。文檔中特別重要的部分是 Where To Patch。

我們希望，Thread 1 調(diào)用 get_count 后，執(zhí)行全部的 Thread 2 ，之后恢復(fù)執(zhí)行 Thread 1。

我們的測(cè)試代碼如下：
 

# test_incrmnt.py import unittest import before_after import dbimport incrmnt class TestIncrmnt(unittest.TestCase):  def setUp(self):    db.reset_db()   def test_increment_race(self):    # after a call to get_count, call increment    with before_after.after('incrmnt.db.get_count', incrmnt.increment):      # start off the race with a call to increment      incrmnt.increment()     count = db.get_count()    self.assertEqual(count, 2)

在首次 get_count 調(diào)用之后，我們使用 before_after 的上下文管理器 after 來(lái)插入另外一個(gè) increment 的調(diào)用。

在默認(rèn)情況下，before_after只調(diào)用一次 after 函數(shù)。在這個(gè)特殊的情況下這是很有用的，因?yàn)榉駝t的話堆棧會(huì)溢出（increment調(diào)用get_count，get_coun t也調(diào)用 increment，increment 又調(diào)用get_count…）。

這個(gè)測(cè)試失敗了，因?yàn)橛?jì)數(shù)等于1，而不是2。現(xiàn)在我們有一個(gè)重現(xiàn)了競(jìng)爭(zhēng)條件的失敗測(cè)試，一起來(lái)修復(fù)。

防止競(jìng)爭(zhēng)

我們將要使用一個(gè)簡(jiǎn)單的鎖機(jī)制來(lái)減緩競(jìng)爭(zhēng)。這顯然不是理想的解決方案，更好的解決方法是使用原子更新進(jìn)行數(shù)據(jù)存儲(chǔ)――但這種方法能更好地示范 before_after 在測(cè)試多線程應(yīng)用程序上的作用。

在 incrmnt.py 中添加一個(gè)新函數(shù)：

# incrmnt.py def locking_increment():  with db.get_lock():    return increment()

它保證在同一時(shí)間只有一個(gè)線程對(duì)計(jì)數(shù)進(jìn)行讀寫(xiě)操作。如果一個(gè)線程試圖獲取鎖，而鎖被另外一個(gè)線程保持，將會(huì)引發(fā) CouldNotLock 異常。

現(xiàn)在我們?cè)黾舆@樣一個(gè)測(cè)試：
 

# test_incrmnt.py def test_locking_increment_race(self):  def erroring_locking_increment():    # Trying to get a lock when the other thread has it will cause a    # CouldNotLock exception - catch it here or the test will fail    with self.assertRaises(db.CouldNotLock):      incrmnt.locking_increment()   with before_after.after('incrmnt.db.get_count', erroring_locking_increment):    incrmnt.locking_increment()   count = db.get_count()  self.assertEqual(count, 1)

現(xiàn)在在同一時(shí)間，就只有一個(gè)線程能夠增加計(jì)數(shù)了。

減緩競(jìng)爭(zhēng)

我們這里還有一個(gè)問(wèn)題，通過(guò)上邊這種方式，如果兩個(gè)請(qǐng)求沖突，一個(gè)不會(huì)被登記。為了緩解這個(gè)問(wèn)題，我們可以讓 increment 重新鏈接服務(wù)器（有一個(gè)簡(jiǎn)潔的方式，就是用類(lèi)似 funcy retry 的東西）：
 

# incrmnt.py def retrying_locking_increment():  @retry(tries=5, errors=db.CouldNotLock)  def _increment():    return locking_increment()   return _increment()

當(dāng)我們需要比這種方法提供的更大規(guī)模的操作時(shí)，可以將 increment 作為一個(gè)原子更新或事務(wù)轉(zhuǎn)移到我們的數(shù)據(jù)庫(kù)中，讓其在遠(yuǎn)離我們的應(yīng)用程序的地方承擔(dān)責(zé)任。

總結(jié)

Incrmnt 現(xiàn)在不存在競(jìng)爭(zhēng)了，人們可以愉快地點(diǎn)擊一整天，而不用擔(dān)心自己不被計(jì)算在內(nèi)。

這是一個(gè)簡(jiǎn)單的例子，但是 before_after 可以用于更復(fù)雜的競(jìng)爭(zhēng)條件，以確保你的函數(shù)能正確地處理所有情形。能夠在單線程環(huán)境中測(cè)試和重現(xiàn)競(jìng)爭(zhēng)條件是一個(gè)關(guān)鍵，它能讓你更確定你正在正確地處理競(jìng)爭(zhēng)條件。