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

Код на:JavaC#PythonJavaScript

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

Проблем

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

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

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

lostSales();
System.out.println("Продадени (очаквано 4000):");
System.out.println(sold);

Изход

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

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

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

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

Идеята

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

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

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

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

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

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

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

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            waitAt(start);
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

// Чака на бариерата, докато дойдат всички (най-много 1 s).
static void waitAt(CyclicBarrier start) {
    try {
        start.await(1, TimeUnit.SECONDS);
    } catch (Exception e) {
        throw new IllegalStateException(e);
    }
}

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

Изход

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

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

CyclicBarrier(4): await чака, докато го извикат 4 нишки, после пуска всички. waitAt чака най-много 1 секунда: тест не трябва да виси, ако някоя нишка не дойде.

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

static int sold = 0;
static final Object lock = new Object();

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            waitAt(start);
            for (int k = 0; k < 1_000; k++) {
                synchronized (lock) {
                    sold++;
                }
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

Изход

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

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

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

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

Пробвай

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

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

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        t.join();
    }
    return sold != 4_000;
}

Изход

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

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

1static int sold = 0;23static boolean lostSales() throws InterruptedException {4 sold = 0;5 for (int i = 0; i < 4; i++) {6 Thread t = new Thread(() -> {7 for (int k = 0; k < 1_000; k++) {8 sold++;9 }10 });11 t.start();12 t.join();13 }14 return sold != 4_000;15}

Провери

Пробвай

Попълни

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

static int sold = 0;

static boolean lostSales() throws InterruptedException {
    sold = 0;
    CyclicBarrier start = new CyclicBarrier(4);
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < 4; i++) {
        Thread t = new Thread(() -> {
            
            for (int k = 0; k < 1_000; k++) {
                sold++;
            }
        });
        t.start();
        threads.add(t);
    }
    for (Thread t : threads) {
        t.join();
    }
    return sold != 4_000;
}

Провери

Идеята

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

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

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

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

class Stock {
    int count;
    Runnable pause = () -> {};

    Stock(int count) {
        this.count = count;
    }

    void buy() {
        int left = count;
        pause.run();   // шев за теста: тук тестът може да спре нишката
        count = left - 1;
    }
}

static String forcedTest(Stock stock) throws InterruptedException {
    CountDownLatch read = new CountDownLatch(1);
    CountDownLatch go = new CountDownLatch(1);
    stock.pause = () -> {
        if (Thread.currentThread().getName().equals("first")) {
            read.countDown();
            waitFor(go);
        }
    };
    Thread first = new Thread(stock::buy, "first");
    Thread second = new Thread(stock::buy, "second");
    first.start();
    waitFor(read);      // first прочете 10 и чака
    second.start();
    second.join(200);   // second купува докрай или чака ключалката
    go.countDown();     // first записва
    first.join(1000);
    second.join(1000);
    String verdict = stock.count == 8 ? "МИНАВА" : "ПАДА";
    return "Наличност: " + stock.count + " (очаквано 8): " + verdict;
}

static void waitFor(CountDownLatch latch) {
    try {
        latch.await(1, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

Изход

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

Шевът е полето pause. Тестът слага в него код, който спира само нишката с име first: отваря резето read и чака резето go. CountDownLatch(1) е резе: countDown() го отваря, await чака да се отвори. waitFor чака най-много 1 секунда.

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

class Stock {
    int count;
    Runnable pause = () -> {};

    Stock(int count) {
        this.count = count;
    }

    synchronized void buy() {
        int left = count;
        pause.run();   // шев за теста: тук тестът може да спре нишката
        count = left - 1;
    }
}

Изход

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

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

Пробвай

Трима на шева

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

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

Stock stock = new Stock(10);
CountDownLatch allRead = new CountDownLatch(3);
stock.pause = () -> {
    allRead.countDown();
    waitFor(allRead);
};
List<Thread> buyers = new ArrayList<>();
for (int i = 0; i < 3; i++) {
    Thread t = new Thread(stock::buy);
    t.start();
    buyers.add(t);
}
for (Thread t : buyers) {
    t.join();
}
System.out.println("Наличност: " + stock.count);

Отговор

Наличност: 7

Наличност: 8

Наличност: 9

Провери

Идеята

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

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

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

Shop shop = new Shop(100);
int[] bought = new int[8];
CyclicBarrier start = new CyclicBarrier(8);
List<Thread> buyers = new ArrayList<>();
for (int i = 0; i < 8; i++) {
    int me = i;
    Thread t = new Thread(() -> {
        waitAt(start);
        for (int k = 0; k < 50; k++) {
            if (shop.tryBuy()) {
                bought[me]++;
            }
        }
    });
    t.start();
    buyers.add(t);
}
for (Thread t : buyers) {
    t.join();
}
int total = Arrays.stream(bought).sum();
System.out.println("Наличност: " + shop.stock() + ", продадени: " + shop.sold());
System.out.println(check("наличност + продадени = 100", shop.stock() + shop.sold() == 100));
System.out.println(check("наличност >= 0", shop.stock() >= 0));
System.out.println(check("купените = продадените", total == shop.sold()));

Изход

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

Магазинът:

class Shop {
    private int stock;
    private int sold = 0;

    Shop(int stock) {
        this.stock = stock;
    }

    synchronized boolean tryBuy() {
        if (stock == 0) {
            return false;
        }
        stock--;
        sold++;
        return true;
    }

    synchronized int stock() {
        return stock;
    }

    synchronized int sold() {
        return sold;
    }
}

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

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

Идеята

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

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

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

static final Object sofia = new Object();
static final Object plovdiv = new Object();

static void transfer(Object from, Object to) {
    synchronized (from) {
        sleep(50);
        synchronized (to) {
            // премести стоката
        }
    }
}

static void sleep(int ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

static boolean finishesWithin(long ms, Runnable... parts) throws InterruptedException {
    List<Thread> threads = new ArrayList<>();
    for (Runnable part : parts) {
        Thread t = new Thread(part);
        t.setDaemon(true);
        t.start();
        threads.add(t);
    }
    long end = System.currentTimeMillis() + ms;
    for (Thread t : threads) {
        t.join(Math.max(1, end - System.currentTimeMillis()));
    }
    return threads.stream().noneMatch(Thread::isAlive);
}

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

Изход

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

join(ms) чака най-много ms. После isAlive() казва дали нишката още върви. setDaemon(true) позволява на JVM да спре, макар двете нишки да са заключени завинаги.

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

В рамките за тестове таймаутът е готов: @Timeout(1) или assertTimeoutPreemptively в JUnit 5. Как се пише тест в езика, виж в урока за тестовете.

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

• 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