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

Incrmnt

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

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

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

競爭條件

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

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

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

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

# 線程1和線程2在不同的進(jìn)程中同時(shí)執(zhí)行
# 為了展示的目的，在這里并排放置
# 在垂直方向分開它們，以說明在每個(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)榫€程安全的，但是在你那么做之前，你還想寫一個(gè)測試證明競爭的存在。

重現(xiàn)競爭

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

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

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

簡單起見，我們試著重現(xiàn)這個(gè)嵌套的情形。這里整 個(gè)Thread 2 在 Thread 1 的首個(gè) get_count 調(diào)用之后執(zhí)行：

# Thread 1             # Thread 2
def 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è)庫，它提供了幫助重現(xiàn)這種情形的工具。它可以在一個(gè)函數(shù)之前或之后插入任意代碼。

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

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

我們的測試代碼如下：
 

# test_incrmnt.py
 
import unittest
 
import before_after
 
import db
import 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 來插入另外一個(gè) increment 的調(diào)用。

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

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

防止競爭

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

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

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

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

現(xiàn)在我們增加這樣一個(gè)測試：
 

# 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ù)了。

減緩競爭

我們這里還有一個(gè)問題，通過上邊這種方式，如果兩個(gè)請求沖突，一個(gè)不會被登記。為了緩解這個(gè)問題，我們可以讓 increment 重新鏈接服務(wù)器（有一個(gè)簡潔的方式，就是用類似 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ù)庫中，讓其在遠(yuǎn)離我們的應(yīng)用程序的地方承擔(dān)責(zé)任。

總結(jié)

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

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

最新評論