Как се тества конкурентен код

Код на:JavaC#PythonJavaScript

Накратко: Конкурентният код се тества със стрес тест с много пускания, бариера на старта, принудено преплитане и проверка на инварианта, защото едно зелено пускане не доказва нищо.

Проблем

Минава на моята машина

Броячът на продажбите от урока за race condition е без ключалка. Колега пише тест: четири нишки продават по 1000 пъти, а накрая продадените трябва да са 4000:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        self.sold += 1


def sell_once():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = [threading.Thread(target=buyer) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return counter.sold


print("Продадени (очаквано 4000):")
print(sell_once())

Изход

Продадени (очаквано 4000):
4000

При него тестът минава. Минава и в CI. Кодът отива в разпродажбата, и в първия час магазинът е продал повече бройки, отколкото броячът показва.

В Python това е особено лесно: заради GIL нишките се сменят на около 5 ms, а self.sold += 1 е няколко инструкции. Шансът смяната да падне точно между четенето и записа е малък. Малък, но не нула: на сървър с много заявки за час той се случва.

Един зелен тест на конкурентен код не доказва почти нищо: при това пускане нишките просто не са се срещнали точно между четенето и записа. Как се пише тест, който хваща такава грешка?

Идеята

Стрес тест: много пускания и бариера

Три неща правят грешката видима:

• Много пускания. Тестът не е едно пускане, а 20 (или 1000 в CI през нощта). Брои се колко са червени.
• Много нишки наведнъж. Повече нишки от ядрата, за да се сменят по средата на работата.
• Бариера на старта. Всяка нишка чака, докато дойдат всички, и тръгват заедно. Така се застъпват през цялото време, а не само в края.

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # тук друга нишка влиза
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Стъпка по стъпка, с думи

1. Тестът пуска 4 нишки, всяка продава 1000 пъти без ключалка. Без бариера всяка нишка тръгва, щом е създадена.
2. 20 пускания: 20 от 20 червени. Тук time.sleep(0) отваря широка пролука и пада и без бариера. В истински код пролуката е тясна.
3. Бариера: всяка нишка чака на старта, докато дойдат и четирите. После тръгват заедно.
4. С бариера: 20 от 20 червени, колкото и без нея: sleep(0) държи пролуката широка. При тясна пролука бариерата решава.

Тестът с бариера:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # тук друга нишка влиза
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000


failed = 0
for run in range(20):
    if lost_sales():
        failed += 1
print("Пускания с изгубени продажби (от 20):")
print(failed)

Изход

Пускания с изгубени продажби (от 20):
20

Изходът е „най-много 20“: колко пускания стават червени зависи от машината. На моята машина излизат 20 от 20 и с бариера, и без нея.

threading.Barrier(4, timeout=1): wait() чака, докато го извикат 4 нишки, после пуска всички. Ако някоя не дойде за 1 секунда, хвърля BrokenBarrierError, вместо да виси.

Тук sell пуска GIL с time.sleep(0) между четенето и записа (както в урока за race condition). Пролуката е толкова широка, че тестът пада и без бариера. В истински код пролуката е няколко инструкции, и там бариерата и многото пускания решават дали изобщо ще я видиш.

Същият тест срещу брояча с ключалка:

class Counter:
    def __init__(self):
        self.sold = 0
        self.lock = threading.Lock()

    def sell(self):
        with self.lock:
            left = self.sold
            time.sleep(0)  # тук друга нишка влиза
            self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Изход

Пускания с изгубени продажби: 0 от 20

По-точно, за любопитните

Стрес тестът намира грешки, но не доказва, че ги няма. „0 от 20“ значи „този път не я видяхме“. Затова в CI такива тестове вървят хиляди пъти, на машини с много ядра, а прегледът на кода пита отделно: кое е общо и коя ключалка го пази.

Как е в другите езици: в Java бариерата е CyclicBarrier, в C# Barrier, в Node се пише с Atomics.wait и Atomics.notify. Вместо бариера често се ползва и threading.Event, което главната нишка отваря.

Пробвай

Намери грешката

Друг колега е написал стрес тест без бариера. Той минава 20 от 20 пъти, при всяко пускане, на всяка машина, макар че броячът няма ключалка:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # тук друга нишка влиза
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        t.join()
    return counter.sold != 4000

Изход

Пускания с изгубени продажби: 0 от 20

Избери реда с грешката.

1class Counter:2 def __init__(self):3 self.sold = 045 def sell(self):6 left = self.sold7 time.sleep(0) # тук друга нишка влиза8 self.sold = left + 191011def lost_sales():12 counter = Counter()1314 def buyer():15 for _ in range(1000):16 counter.sell()1718 for _ in range(4):19 t = threading.Thread(target=buyer)20 t.start()21 t.join()22 return counter.sold != 4000

Провери

Пробвай

Попълни

Кой ред кара нишките да тръгнат заедно?

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # тук друга нишка влиза
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Провери

Идеята

Принуди преплитането

Стрес тестът намира грешката понякога. За тест, който пада всеки път, тестът сам избира реда на нишките. За това кодът има шев (seam): място, където тестът може да спре нишката. В истинската програма шевът не прави нищо.

Сценарият е грешката от урока за race condition: остават 10 бройки, двама купуват по една.

1. Първият купувач прочита 10 и спира на шева.
2. Вторият купува докрай: прочита 10, записва 9.
3. Първият продължава и записва своето 10 − 1 = 9.

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None

    def buy(self):
        left = self.count
        self.pause()  # шев за теста: тук тестът може да спре нишката
        self.count = left - 1


def forced_test(stock):
    read = threading.Event()
    go = threading.Event()

    def pause():
        if threading.current_thread().name == "first":
            read.set()
            go.wait(timeout=1)

    stock.pause = pause
    first = threading.Thread(target=stock.buy, name="first")
    second = threading.Thread(target=stock.buy, name="second")
    first.start()
    read.wait(timeout=1)  # first прочете 10 и чака
    second.start()
    second.join(timeout=0.2)  # second купува докрай или чака ключалката
    go.set()  # first записва
    first.join(timeout=1)
    second.join(timeout=1)
    verdict = "МИНАВА" if stock.count == 8 else "ПАДА"
    return f"Наличност: {stock.count} (очаквано 8): {verdict}"

Изход

Наличност: 9 (очаквано 8): ПАДА

Шевът е полето pause. Тестът слага в него функция, която спира само нишката с име first: отваря read и чака go. threading.Event е резе: set() го отваря, wait(timeout=1) чака да се отвори, най-много 1 секунда.

Наличността е 9 вместо 8, при всяко пускане. Сега поправката: with self.lock: около четенето и записа:

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None
        self.lock = threading.Lock()

    def buy(self):
        with self.lock:
            left = self.count
            self.pause()  # шев за теста: тук тестът може да спре нишката
            self.count = left - 1

Изход

Наличност: 8 (очаквано 8): МИНАВА

Тестът е същият. Първият купувач държи ключалката, докато е спрял на шева. Вторият чака ключалката, затова second.join(timeout=0.2) изтича, без той да е купил. Чак когато първият запише 9 и пусне ключалката, вторият прочита 9 и записва 8. Тест, който падаше всеки път преди поправката и минава всеки път след нея, е точно това, което искаш в CI.

Пробвай

Трима на шева

Същият Stock без ключалка. Сега шевът чака, докато и тримата купувачи прочетат наличността:

Какво ще отпечата програмата?

stock = Stock(10)
all_read = threading.Barrier(3, timeout=1)
stock.pause = all_read.wait
buyers = [threading.Thread(target=stock.buy) for _ in range(3)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Наличност: {stock.count}")

Отговор

Наличност: 9

Наличност: 8

Наличност: 7

Провери

Идеята

Проверявай инварианта, не реда

При 8 купувача кой какво е купил зависи от пускането. Тест, който очаква „първият купувач купи 13“, ще пада и при верен код. Затова тестът проверява инвариант: нещо, което трябва да е вярно при всеки ред.

За магазина това е: наличността плюс продадените е колкото е имало в началото, наличността не е под нула, а покупките на купувачите събрани са колкото продадените. 8 купувача правят по 50 опита за 100 бройки:

shop = Shop(100)
bought = [0] * 8
start = threading.Barrier(8, timeout=1)


def buyer(me):
    start.wait()
    for _ in range(50):
        if shop.try_buy():
            bought[me] += 1


buyers = [threading.Thread(target=buyer, args=(i,)) for i in range(8)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Наличност: {shop.stock}, продадени: {shop.sold}")
print(check("наличност + продадени = 100", shop.stock + shop.sold == 100))
print(check("наличност >= 0", shop.stock >= 0))
print(check("купените = продадените", sum(bought) == shop.sold))

Изход

Наличност: 0, продадени: 100
наличност + продадени = 100: МИНАВА
наличност >= 0: МИНАВА
купените = продадените: МИНАВА

Магазинът:

class Shop:
    def __init__(self, stock):
        self.stock = stock
        self.sold = 0
        self.lock = threading.Lock()

    def try_buy(self):
        with self.lock:
            if self.stock == 0:
                return False
            self.stock -= 1
            self.sold += 1
            return True

Всяка нишка пише в свое място bought[me], затова списъкът не иска ключалка. Сборът е верен при всеки ред; кой колко е купил не е.

400 опита за 100 бройки значи, че накрая наличността е 0 при всеки ред. Това също е инвариант, затова изходът е един и същ при всяко пускане.

Идеята

Таймаут: deadlock-ът проваля теста

Два склада прехвърлят стока един на друг. Всяко прехвърляне заключва склада, от който взима, после този, в който слага, в обратен ред. Това е deadlock: всяка нишка държи едната ключалка и чака другата вечно.

Тест без таймаут не пада: той виси. CI чака час, докато го убие, и никой не знае кой тест е бил. Затова тестът с нишки винаги чака с таймаут:

sofia = threading.Lock()
plovdiv = threading.Lock()


def transfer(source, target):
    with source:
        time.sleep(0.05)
        with target:
            pass  # премести стоката


def finishes_within(seconds, *parts):
    threads = [threading.Thread(target=part, daemon=True) for part in parts]
    for t in threads:
        t.start()
    end = time.monotonic() + seconds
    for t in threads:
        t.join(timeout=max(0, end - time.monotonic()))
    return not any(t.is_alive() for t in threads)


ok = finishes_within(1, lambda: transfer(sofia, plovdiv), lambda: transfer(plovdiv, sofia))
print("Тест „прехвърляне между складовете“: " + ("МИНАВА" if ok else "не свърши за 1 s: ПАДА"))

Изход

Тест „прехвърляне между складовете“: не свърши за 1 s: ПАДА

join(timeout=...) чака най-много толкова. После is_alive() казва дали нишката още върви. daemon=True позволява на Python да спре, макар двете нишки да са заключени завинаги.

По-точно, за любопитните

В рамките за тестове таймаутът е готов: pytest-timeout (@pytest.mark.timeout(1)) или faulthandler.dump_traceback_later, който отпечатва стека на всички нишки, когато тестът е заседнал. Как се пише тест в езика, виж в урока за тестовете.

Има и инструменти, които търсят грешките вместо теб. Тук ги назоваваме само, без да ги ползваме:

• jcstress (Java): пуска малки парчета код милиони пъти във всички редове, които JVM-ът позволява, и брои резултатите.
• ThreadSanitizer (C, C++, Go, Rust) и race detector-ът на Go (go test -race): следят всяко четене и запис по време на теста и казват „тези две нишки пипат едно и също без ключалка“, дори ако този път резултатът е бил верен.
• Microsoft.Coyote (.NET): пуска async и многонишков код под свой планировчик, опитва много редове и може да повтори реда, който е паднал.
• faulthandler (Python): при заседнал тест отпечатва къде е всяка нишка.

Упражнение

Стрес тест на касата

Реши задачата в редактора си.

Касата на разпродажбата има sell(euros). Напиши стрес тест:

• 20 пускания. Във всяко 4 нишки с бариера на старта; всяка продава 500 пъти по 2 €. Очакваният сбор е 4000 €.
• Пускане, в което нишките не свършат за 1 секунда, се брои за таймаут, а тестът продължава.
• Накрая отпечатай колко пускания са с грешен сбор и колко с таймаут, и дали касата минава.

Изход

Пускания: 20
С грешен сбор: 0
С таймаут: 0
Касата: МИНАВА

Още задачи

• Счупи касата. Махни ключалката и виж колко пускания стават червени. Пусни го няколко пъти: числото се мени.
• Принудено преплитане. Добави шев в sell и напиши тест, който пада всеки път без ключалката и минава с нея.
• Deadlock в касата. Добави второ заключване (касата и складът) в обратен ред и провери, че тестът ти отчита таймаут, а не виси.

Тест

Провери се

1. Тестът на брояча без ключалка минава. Какво значи това?

Почти нищо: при това пускане нишките не са се застъпилиБроячът е безопасен на тази машинаБроячът е безопасен

2. За какво е бариерата на старта на стрес теста?

Тестът да е по-бързВсички нишки да тръгнат заедно и да се застъпятНишките да вървят една след друга

3. Стрес тестът минава 20 от 20 пъти. Безопасен ли е кодът?

Да, 20 пускания стигатНе е доказано: тестът показва грешки, но не доказва, че ги нямаДа, ако бариерата е била включена

4. Как тестът да пада всеки път на една и съща race condition?

Thread.sleep с различни паузиРезета в шев на кода: първата нишка спира между четенето и записа, докато втората минеПовече пускания

5. Защо тест с нишки трябва да има таймаут?

Защото нишките не свършват самиПри deadlock тестът иначе виси вечно и спира целия CIЗа да е по-бърз

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: Атомарни операции и неизменими обекти · Deadlock · Пул от нишки и задачи · async не е нишка
• Задачи по темата: Нишки и race conditions