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

Код на:JavaC#PythonJavaScript

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

Проблем

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

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

int sold = 0;

bool LostSales()
{
    sold = 0;
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

LostSales();
Console.WriteLine("Продадени (очаквано 4000):");
Console.WriteLine(sold);

Изход

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

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

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

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

Идеята

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

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

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

int sold = 0;

bool LostSales()
{
    sold = 0;
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

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

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

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

int sold = 0;

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            start.SignalAndWait(TimeSpan.FromSeconds(1));
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

int failed = 0;
for (int run = 0; run < 20; run++)
{
    if (LostSales())
    {
        failed++;
    }
}
Console.WriteLine("Пускания с изгубени продажби (от 20):");
Console.WriteLine(failed);

Изход

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

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

Barrier(4): SignalAndWait чака, докато го извикат 4 нишки, после пуска всички. Таймаутът от 1 секунда пази теста да не виси, ако някоя нишка не дойде.

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

int sold = 0;
object sync = new object();

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            start.SignalAndWait(TimeSpan.FromSeconds(1));
            for (int k = 0; k < 1_000; k++)
            {
                lock (sync)
                {
                    sold++;
                }
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

Изход

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

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

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

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

Пробвай

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

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

int sold = 0;

bool LostSales()
{
    sold = 0;
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        t.Join();
    }
    return sold != 4_000;
}

Изход

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

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

1int sold = 0;23bool LostSales()4{5 sold = 0;6 for (int i = 0; i < 4; i++)7 {8 var t = new Thread(() =>9 {10 for (int k = 0; k < 1_000; k++)11 {12 sold++;13 }14 });15 t.Start();16 t.Join();17 }18 return sold != 4_000;19}

Провери

Пробвай

Попълни

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

int sold = 0;

bool LostSales()
{
    sold = 0;
    var start = new Barrier(4);
    var threads = new List<Thread>();
    for (int i = 0; i < 4; i++)
    {
        var t = new Thread(() =>
        {
            
            for (int k = 0; k < 1_000; k++)
            {
                sold++;
            }
        });
        t.Start();
        threads.Add(t);
    }
    foreach (var t in threads)
    {
        t.Join();
    }
    return sold != 4_000;
}

Провери

Идеята

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

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

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

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

string ForcedTest(Stock stock)
{
    var read = new ManualResetEventSlim();
    var go = new ManualResetEventSlim();
    stock.Pause = () =>
    {
        if (Thread.CurrentThread.Name == "first")
        {
            read.Set();
            go.Wait(TimeSpan.FromSeconds(1));
        }
    };
    var first = new Thread(stock.Buy) { Name = "first" };
    var second = new Thread(stock.Buy) { Name = "second" };
    first.Start();
    read.Wait(TimeSpan.FromSeconds(1));   // first прочете 10 и чака
    second.Start();
    second.Join(200);   // second купува докрай или чака ключалката
    go.Set();           // first записва
    first.Join(1000);
    second.Join(1000);
    string verdict = stock.Count == 8 ? "МИНАВА" : "ПАДА";
    return $"Наличност: {stock.Count} (очаквано 8): {verdict}";
}

class Stock
{
    public int Count;
    public Action Pause = () => { };

    public Stock(int count) => Count = count;

    public void Buy()
    {
        int left = Count;
        Pause();   // шев за теста: тук тестът може да спре нишката
        Count = left - 1;
    }
}

Изход

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

Шевът е полето Pause. Тестът слага в него код, който спира само нишката с име first: отваря read и чака go. ManualResetEventSlim е резе: Set() го отваря, Wait чака да се отвори, тук най-много 1 секунда.

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

class Stock
{
    public int Count;
    public Action Pause = () => { };
    private readonly object sync = new object();

    public Stock(int count) => Count = count;

    public void Buy()
    {
        lock (sync)
        {
            int left = Count;
            Pause();   // шев за теста: тук тестът може да спре нишката
            Count = left - 1;
        }
    }
}

Изход

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

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

Пробвай

Трима на шева

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

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

var stock = new Stock(10);
var allRead = new CountdownEvent(3);
stock.Pause = () =>
{
    allRead.Signal();
    allRead.Wait(TimeSpan.FromSeconds(1));
};
var buyers = new List<Thread>();
for (int i = 0; i < 3; i++)
{
    var t = new Thread(stock.Buy);
    t.Start();
    buyers.Add(t);
}
foreach (var t in buyers)
{
    t.Join();
}
Console.WriteLine($"Наличност: {stock.Count}");

Отговор

Наличност: 9

Наличност: 8

Наличност: 7

Провери

Идеята

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

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

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

var shop = new Shop(100);
int[] bought = new int[8];
var start = new Barrier(8);
var buyers = new List<Thread>();
for (int i = 0; i < 8; i++)
{
    int me = i;
    var t = new Thread(() =>
    {
        start.SignalAndWait(TimeSpan.FromSeconds(1));
        for (int k = 0; k < 50; k++)
        {
            if (shop.TryBuy())
            {
                bought[me]++;
            }
        }
    });
    t.Start();
    buyers.Add(t);
}
foreach (var t in buyers)
{
    t.Join();
}
Console.WriteLine($"Наличност: {shop.Stock}, продадени: {shop.Sold}");
Console.WriteLine(Check("наличност + продадени = 100", shop.Stock + shop.Sold == 100));
Console.WriteLine(Check("наличност >= 0", shop.Stock >= 0));
Console.WriteLine(Check("купените = продадените", bought.Sum() == shop.Sold));

Изход

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

Магазинът:

class Shop
{
    private int stock;
    private int sold = 0;
    private readonly object sync = new object();

    public Shop(int stock) => this.stock = stock;

    public bool TryBuy()
    {
        lock (sync)
        {
            if (stock == 0) return false;
            stock--;
            sold++;
            return true;
        }
    }

    public int Stock { get { lock (sync) return stock; } }
    public int Sold { get { lock (sync) return sold; } }
}

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

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

Идеята

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

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

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

object sofia = new object();
object plovdiv = new object();

void Transfer(object from, object to)
{
    lock (from)
    {
        Thread.Sleep(50);
        lock (to)
        {
            // премести стоката
        }
    }
}

bool FinishesWithin(int ms, params ThreadStart[] parts)
{
    var threads = parts.Select(part => new Thread(part) { IsBackground = true }).ToList();
    threads.ForEach(t => t.Start());
    var end = DateTime.Now.AddMilliseconds(ms);
    foreach (var t in threads)
    {
        var left = end - DateTime.Now;
        t.Join(left > TimeSpan.Zero ? left : TimeSpan.FromMilliseconds(1));
    }
    return threads.All(t => !t.IsAlive);
}

bool ok = FinishesWithin(1000,
    () => Transfer(sofia, plovdiv),
    () => Transfer(plovdiv, sofia));
Console.WriteLine("Тест „прехвърляне между складовете“: " + (ok ? "МИНАВА" : "не свърши за 1 s: ПАДА"));

Изход

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

Join(TimeSpan) чака най-много толкова и връща дали нишката е свършила. IsBackground = true позволява на програмата да свърши, макар двете нишки да са заключени завинаги.

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

В рамките за тестове таймаутът е готов: [Fact(Timeout = 1000)] в xUnit (за async тестове) или [Timeout] в NUnit. Как се пише тест в езика, виж в урока за тестовете.

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

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

Упражнение

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

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

Касата на разпродажбата има Sell(int 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