Към съдържанието
wannacode

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

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

Проблем

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

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

if (isMainThread) {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), { workerData: { sold } });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  console.log("Продадени (очаквано 4000):");
  console.log(sold[0]);
} else {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}
Изход
Продадени (очаквано 4000):
4000

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

Worker-ите тръгват бавно (всеки е нов V8), затова първият често е свършил своите 1000 продажби, преди следващият да е тръгнал. На машината с 4 ядра и натоварване от разпродажбата е друго.

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

Идеята

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

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

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

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}


function worker() {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}
Стъпка по стъпка, с думи
  1. Тестът пуска 4 Worker-а, всяки продава 1000 пъти без ключалка. Без бариера всеки Worker тръгва, щом е създаден.
  2. 20 пускания: в този пример 6 от 20 червени. Worker-ът често свършва своите 1000 продажби, преди следващият да е тръгнал.
  3. Бариера: всеки Worker чака на старта, докато дойдат и четирите. После тръгват заедно.
  4. 20 пускания с бариера: 20 от 20 червени. Същият счупен брояч, но сега тестът го хваща всеки път.

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

// Бариера за Worker-и: всеки чака, докато стигнат всички (най-много 1 s).
function arrive(gate, parties) {
  Atomics.add(gate, 0, 1);
  Atomics.notify(gate, 0);
  const end = Date.now() + 1000;
  let n;
  while ((n = Atomics.load(gate, 0)) < parties && Date.now() < end) {
    Atomics.wait(gate, 0, n, 50);
  }
}


async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}


function worker() {
  const { sold, gate } = workerData;
  arrive(gate, 4);
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}


if (isMainThread) {
  let failed = 0;
  for (let run = 0; run < 20; run++) {
    if (await lostSales()) {
      failed++;
    }
  }
  console.log("Пускания с изгубени продажби (от 20):");
  console.log(failed);
} else {
  worker();
}
Изход
Пускания с изгубени продажби (от 20):
20

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

Бариерата е брояч в SharedArrayBuffer: всеки Worker го увеличава с Atomics.add, събужда другите с Atomics.notify и чака с Atomics.wait, докато броячът стане 4. Atomics.wait е позволено в Worker; в главната нишка на браузъра не е. Чакането е най-много 1 секунда.

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

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}


function worker() {
  const { sold, gate } = workerData;
  arrive(gate, 4);
  for (let k = 0; k < 1_000; k++) {
    Atomics.add(sold, 0, 1);
  }
}
Изход
Пускания с изгубени продажби: 0 от 20
По-точно, за любопитните

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

Как е в другите езици: Java има готова бариера (CyclicBarrier), C# също (Barrier), Python също (threading.Barrier). В Node я пишем сами, защото Atomics е най-ниското ниво.

Пробвай

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

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

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold },
    });
    await once(w, "exit");
  }
  return sold[0] !== 4_000;
}


function worker() {
  const { sold } = workerData;
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}
Изход
Пускания с изгубени продажби: 0 от 20

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

Пробвай

Попълни

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

async function lostSales() {
  const sold = new Int32Array(new SharedArrayBuffer(4));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  const done = [];
  for (let i = 0; i < 4; i++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { sold, gate },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  return sold[0] !== 4_000;
}

function worker() {
  const { sold, gate } = workerData;
  
  for (let k = 0; k < 1_000; k++) {
    sold[0]++;
  }
}

Идеята

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

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

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

  1. Първият купувач прочита 10 и спира на шева.
  2. Вторият купува докрай: прочита 10, записва 9.
  3. Първият продължава и записва своето 10 − 1 = 9.
class Stock {
  constructor(count) {
    this.count = count;
    this.pause = async () => {};
  }


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


async function forcedTest(stock) {
  const read = latch();
  const go = latch();
  stock.pause = async (who) => {
    if (who === "first") {
      read.open();
      await go.wait(1000);
    }
  };
  const first = stock.buy("first");
  await read.wait(1000);   // first прочете 10 и чака
  const second = stock.buy("second");
  await Promise.race([second, sleep(200)]);   // second купува докрай или чака ключалката
  go.open();   // first записва
  await Promise.all([first, second]);
  const verdict = stock.count === 8 ? "МИНАВА" : "ПАДА";
  return `Наличност: ${stock.count} (очаквано 8): ${verdict}`;
}
Изход
Наличност: 9 (очаквано 8): ПАДА
В Node общото състояние в обикновен код се дели между await-ове (урокът async), затова и шевът е await this.pause(who). Резето е обещание: open() го изпълнява, а wait(ms) чака него или таймера, което дойде първо.

Наличността е 9 вместо 8, при всяко пускане. Сега поправката: четенето и записът без await между тях:

class Stock {
  constructor(count) {
    this.count = count;
    this.pause = async () => {};
  }


  async buy(who) {
    this.count--;   // без await между четенето и записа
    await this.pause(who);   // шев за теста: тук тестът може да спре нишката
  }
}
Изход
Наличност: 8 (очаквано 8): МИНАВА

Тестът е същият. Първият купувач вече е намалил наличността, преди да спре на шева, затова вторият вижда 9. Тест, който падаше всеки път преди поправката и минава всеки път след нея, е точно това, което искаш в CI.

Пробвай

Трима на шева

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

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

const stock = new Stock(10);
const allRead = countdown(3);
stock.pause = async () => {
  allRead.arrive();
  await allRead.done;
};
await Promise.all([stock.buy(), stock.buy(), stock.buy()]);
console.log(`Наличност: ${stock.count}`);
Отговор

Идеята

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

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

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

if (isMainThread) {
  const shop = new Int32Array(new SharedArrayBuffer(8));
  const bought = new Int32Array(new SharedArrayBuffer(4 * 8));
  const gate = new Int32Array(new SharedArrayBuffer(4));
  shop[0] = 100;
  const done = [];
  for (let me = 0; me < 8; me++) {
    const w = new Worker(new URL(import.meta.url), {
      workerData: { shop, bought, gate, me },
    });
    done.push(once(w, "exit"));
  }
  await Promise.all(done);
  const total = bought.reduce((a, b) => a + b, 0);
  console.log(`Наличност: ${shop[0]}, продадени: ${shop[1]}`);
  console.log(check("наличност + продадени = 100", shop[0] + shop[1] === 100));
  console.log(check("наличност >= 0", shop[0] >= 0));
  console.log(check("купените = продадените", total === shop[1]));
} else {
  const { shop, bought, gate, me } = workerData;
  arrive(gate, 8);
  for (let k = 0; k < 50; k++) {
    if (tryBuy(shop)) {
      bought[me]++;
    }
  }
}
Изход
Наличност: 0, продадени: 100
наличност + продадени = 100: МИНАВА
наличност >= 0: МИНАВА
купените = продадените: МИНАВА

Магазинът:

// shop[0] е наличността, shop[1] са продадените.
function tryBuy(shop) {
  while (true) {
    const left = Atomics.load(shop, 0);
    if (left === 0) return false;
    if (Atomics.compareExchange(shop, 0, left, left - 1) === left) {
      Atomics.add(shop, 1, 1);
      return true;
    }
  }
}
Тук tryBuy е без ключалка: Atomics.compareExchange намалява наличността само ако тя още е тази, която сме прочели (урокът за атомарните операции). Всеки Worker пише в свое място в bought, затова там не трябва нищо.

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

Идеята

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

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

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

// Ключалка в SharedArrayBuffer: 0 е свободна, 1 е заета.
function lock(locks, i) {
  while (Atomics.compareExchange(locks, i, 0, 1) !== 0) {
    Atomics.wait(locks, i, 1, 50);
  }
}


function transfer({ locks, from, to }) {
  lock(locks, from);
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50);   // пауза 50 ms
  lock(locks, to);
  // премести стоката
}


async function finishesWithin(ms, jobs) {
  const workers = jobs.map((job) => new Worker(new URL(import.meta.url), { workerData: job }));
  const done = Promise.all(workers.map((w) => once(w, "exit"))).then(() => true);
  const ok = await Promise.race([done, sleep(ms).then(() => false)]);
  await Promise.all(workers.map((w) => w.terminate()));
  return ok;
}


if (isMainThread) {
  const locks = new Int32Array(new SharedArrayBuffer(8));
  const ok = await finishesWithin(1000, [
    { locks, from: 0, to: 1 },
    { locks, from: 1, to: 0 },
  ]);
  console.log("Тест „прехвърляне между складовете“: " + (ok ? "МИНАВА" : "не свърши за 1 s: ПАДА"));
} else {
  transfer(workerData);
}
Изход
Тест „прехвърляне между складовете“: не свърши за 1 s: ПАДА
Ключалките са в SharedArrayBuffer: Atomics.compareExchange взима ключалката, ако е свободна, а Atomics.wait чака. Promise.race чака или двата Worker-а, или таймера. После terminate() спира и двата, иначе програмата никога не би свършила.
По-точно, за любопитните

В рамките за тестове таймаутът е готов: { timeout: 1000 } в node:test. Как се пише тест в езика, виж в урока за тестовете.

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

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

Упражнение

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

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

Касата на разпродажбата има Atomics.add(register, 0, euros). Напиши стрес тест:

  • 20 пускания. Във всяко 4 Worker-а с бариера на старта; всеки продава 500 пъти по 2 €. Очакваният сбор е 4000 €.
  • Пускане, в което Worker-ите не свършат за 1 секунда, се брои за таймаут, а тестът продължава.
  • Накрая отпечатай колко пускания са с грешен сбор и колко с таймаут, и дали касата минава.
Изход
Пускания: 20
С грешен сбор: 0
С таймаут: 0
Касата: МИНАВА

Още задачи

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

Тест

Провери се

1. Тестът на брояча без ключалка минава. Какво значи това?
2. За какво е бариерата на старта на стрес теста?
3. Стрес тестът минава 20 от 20 пъти. Безопасен ли е кодът?
4. Как тестът да пада всеки път на една и съща race condition?
5. Защо тест с нишки трябва да има таймаут?

Създаден от Иво Гергов като част от кампанията на Evo Tech за достъпно обучение по програмиране ·