10.2.5. Структурирование многопоточного тестового кода

В разделе 10.2.2 я говорил о том, что нужно придумать, как обеспечить надлежащий порядок планирования для циклов «while» в тестах. Сейчас самое время поговорить о возникающих здесь вопросах.

Основная проблема — организовать набор потоков таким образом, чтобы каждый исполнял выбранный фрагмент кода в указанный вами момент времени. В простейшем случае потоков всего два, но решение легко обобщается и на большее число. На первом этапе нужно определиться, как устроен каждый тест:

• код общей настройки, исполняемый в самом начале;

• потоковый код настройки, исполняемый в каждом потоке;

• содержательный код, исполняемый в параллельно работающих потоках;

• код, исполняемый по завершении параллельного исполнения; может включать утверждения о состоянии программы.

Для определённости рассмотрим пример из списка в разделе 10.2.2: один поток вызывает push() для пустой очереди, а второй в это время вызывает pop().

Код общей настройки очевиден: надо создать очередь. В потоке, исполняющем pop(), нет потокового кода настройки. Потоковый код настройки для потока, исполняющего push(), зависит от интерфейса очереди и типа сохраняемого в ней объекта. Если конструировать сохраняемый объект дорого или память для него должна выделяться из кучи, то лучше сделать это в потоковом коде настройки, чтобы не оказывать влияния на сам тест. С другой стороны, если в очереди хранятся всего лишь значения типа int, то мы ничего не выиграем от их конструирования в коде настройки. Собственно тестируемый код тоже прост — вызвать push() в одном потоке и pop() в другом. А вот как быть с кодом, «исполняемым по завершении»?

В данном случае всё зависит от того, что должна делать функция pop(). Если предполагается, что она блокирует поток до появления данных в очереди, то, очевидно, мы ожидаем, что будут возвращены данные, переданные функции push(), и что очередь в итоге окажется пустой. Если же pop() не блокирует поток и может вернуть управление, даже когда очередь пуста, то требуется проверить два возможных исхода: либо pop() вернула данные, переданные push(), и очередь пуста, либо pop() известила об отсутствии данных и в очереди есть один элемент. Истинно должно быть ровно одно утверждение; чего мы точно не хотим, так это ситуации, когда pop() говорит «нет данных», но очередь пуста, или когда pop() вернула значение, а очередь все равно не пуста. Для упрощения теста предположим, что функция pop() блокирующая. Тогда в завершающем коде должно быть утверждение вида «извлеченное значение совпадает с помещённым и очередь пуста».

Определившись со структурой кода, мы должны постараться, чтобы все работало в соответствии с планом. Один из путей - воспользоваться набором объектов std::promise, обозначающих, что все готово. Каждый поток устанавливает обещание, сообщая, что он готов, а затем ждет (копии) будущего результата std::shared_future, полученного из третьего объекта std::promise; главный поток ждет обещаний от всех потоков, а затем запускает потоки, устанавливая go. Тем самым гарантируется, что каждый поток запущен и находится в точке, непосредственно предшествующей коду, который должен выполняться параллельно; весь потоковый код настройки должен завершиться до установки обещания go. Наконец, главный поток ждет завершения других потоков и проверяет получившееся состояние. Мы также должны принять во внимание исключения и гарантировать, что ни один поток не будет ждать сигнала go, который никогда не поступит. В листинге ниже приведён один из возможных способов структурирования этого теста.

Листинг 10.1. Пример теста, проверяющего параллельное выполнение функций очереди push() и pop()

void test_concurrent_push_and_pop_on_empty_queue() {

 threadsafe_queue<int> q; ← (1)

 std::promise<void> go, push_ready, pop_ready;← (2)

 std::shared_future<void>

  ready(go.get_future()); ← (3)

 std: :future<void> push_done; ← (4)

 std::future<int> pop_done;

 try {

  push_done = std::async(std::launch::async, ← (5)

   [&q, ready, &push_ready]() {

    push_ready.set_value();

    ready.wait();

    q.push(42);

   }

  );

  pop_done = std::async(std::launch::async, ← (6)

   [&q, ready, &pop_ready]() {

    pop_ready.set_value();

    ready.wait();

    return q.pop(); ←(7)

   }

  );

  push_ready.get_future().wait(); ← (8)

  pop_ready.get_future().wait();

  go.set_value(); ← (9)

  push_done.get(); ← (10)

  assert(pop_done.get() == 42); ← (11)

  assert(q.empty());

 } catch (...) {

  go.set_value(); ← (12)

  throw;

 }

}

Структура кода в точности соответствует описанной выше. Сначала, в коде общей настройки, мы создаем пустую очередь (1). Затем создаем все объекты-обещания для сигналов ready (готово) (2) и получаем std::shared_future для сигнала go (3). После этого создаются будущие результаты, означающие, что потоки завершили исполнение (4). Они должны быть созданы вне блока try, чтобы сигнал go можно было установить в случае исключения, не ожидая завершения потоков (что привело бы к взаимоблокировке — вещь, абсолютно недопустимая в тесте).

Внутри блока try мы затем можем создать потоки (5), (6) — использование std::launch::async гарантирует, что каждая задача работает в отдельном потоке. Отметим, что благодаря использованию std::async обеспечить безопасность относительно исключений проще, чем в случае простого std::thread, потому что деструктор будущего результата присоединит поток. В переменных, захваченных лямбда-функцией, хранится ссылка на очередь, соответствующее обещание для подачи сигнала о готовности, а также копия будущего результата ready, полученного из обещания go.

Как было описано выше, каждая задача устанавливает свой сигнал ready, а затем ждет общего сигнала ready, прежде чем начать исполнение тестируемого кода. Главный поток делает всё наоборот — ждет сигналов от обоих потоков (8), а затем сигнализирует им о том, что можно переходить к исполнению тестируемого кода (9).

Напоследок главный поток вызывает функцию get() обоих будущих результатов, возвращенных асинхронными вызовами, чтобы дождаться завершения задач (10), (11) и проверить получившееся состояние. Отметим, что задача pop возвращает извлеченное из очереди значение в будущем результате (7), чтобы мы могли проверить его в утверждении (11).

В случае исключения мы устанавливаем сигнал go, чтобы не оказалось висячего потока, и возбуждаем исключение повторно (12). Будущие результаты, соответствующие обеим задачам (4), были объявлены последними, поэтому уничтожаются первыми, и их деструкторы ждут завершения задач, если они еще не завершились.

Хотя служебного кода многовато для тестирования двух простых вызовов, что-то в этом роде все равно необходимо, чтобы проверить именно то, что мы хотим проверить. Например, запуск потока занимает довольно много времени, поэтому если бы мы не заставили потоки ждать сигнала go, то поток, помещающий данные, вполне мог бы завершиться еще до запуска потока, извлекающего данные, а это шло бы вразрез с целью данного теста. Благодаря использованию будущих результатов мы можем быть уверены, что оба потока запущены и блокированы в ожидании одного и того же будущего. Как только это будущее наступит, оба потока начнут работать. Привыкнув к этой структуре, вы без труда напишете и другие тесты. Продемонстрированный принцип без труда обобщается на случай, когда в каком-то тесте требуется более двух потоков.

До сих пор мы говорили о корректности многопоточного кода. Это, конечно, самая важная, но не единственная цель тестирования. Существенна также его производительность, и далее мы займемся этим вопросом.