當(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上找到）。
|
1
2
3
4
5
6
7
8
9
10
|
# 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
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
# 線程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í)行：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
# 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 依賴于 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è)試代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
# 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ù)：
|
1
2
3
4
5
|
# 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è)試：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
# 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 的東西）：
|
1
2
3
4
5
6
7
8
|
# 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)條件。
