Синхронизация пула потоков

Я уже говорил раньше, что блокирующие операции губительны для параллельной обработки. Однако зачастую приходится ждать, когда станет доступным некий ресурс, или, возможно, вы реализуете протокол, в котором предусматривается определенный период ожидания до повторной отправки сетевого пакета. Что же тогда делать? Вы могли бы задействовать критическую секцию (critical section), вызывать функции вроде Sleep и WaitForSingleObject и т. д. Конечно, это означает, что ваши потоки снова будут простаивать на блокировке. Здесь нужен какой-то способ переложить ожидание на пул потоков, не затрагивая его лимиты параллельных потоков, о чем шла речь в моей статье за сентябрь (msdn.microsoft.com/magazine/hh394144). Тогда пул потоков может поставить в очередь обратный вызов, как только ресурс станет доступным или истечет время ожидания.

Сегодня я покажу, как реализовать именно такой вариант. Наряду с объектами работы, о которых я рассказывал в своей статье за август (msdn.microsoft.com/magazine/hh335066), API пула потоков предоставляет ряд других объектов, генерирующих обратные вызовы, и мы рассмотрим, как использовать объекты ожидания.

Объекты ожидания

Объект ожидания (wait object) пула потоков используется для синхронизации. Вместо блокировки на критической секции вы можете ждать, когда синхронизирующий объект ядра (обычно событие или семафор) перейдет в свободное состояние (become signaled).

Хотя вы можете использовать WaitForSingleObject и ее свиту, объект ожидания отлично интегрируется с остальной частью API пула потоков. Это делается весьма эффективно, группируя вместе любые переданные вами объекты ожидания, что уменьшает количество необходимых потоков и объем кода, который вам придется писать и отлаживать. Благодаря этому вы можете использовать среду пула потоков и группы очистки и избавиться от нужды выделять один или более потоков под ожидание перехода объектов в свободное состояние. В связи с рядом усовершенствований в той части пула потоков, которая относится к ядру, в некоторых случаях этого можно добиваться даже без использования потоков.

Объект ожидания создается функцией CreateThreadpoolWait. Если функция завершается успешно, она возвращает непрозрачный указатель, представляющий объект ожидания. А если ее выполнение заканчивается неудачей, функция возвращает нулевое значение указателя и предоставляет дополнительную информацию через функцию GetLastError. Функция CloseThreadpoolWait сообщает пулу потоков, что объект ожидания может быть освобожден. Эта функция не возвращает значение и для большей эффективности предполагает, что объект ожидания действителен.

Все эти детали берет на себя шаблон класса unique_handle, который я представил в своей статье за июль (msdn.microsoft.com/magazine/hh288076).

Вот класс traits, который можно использовать совместно с unique_handle, а также как typedef для удобства:

struct wait_traits
{
  static PTP_WAIT invalid() throw()
  {
    return nullptr;
  }

  static void close(PTP_WAIT value) throw()
  {
    CloseThreadpoolWait(value);
  }
};

typedef unique_handle<PTP_WAIT, wait_traits> wait;

Теперь я могу использовать typedef и создать объект wait следующим образом:

void * context = ...
wait w(CreateThreadpoolWait(callback, context, nullptr));
check_bool(w);

Как обычно, последний (необязательный) параметр принимает указатель на среду, чтобы вы могли сопоставить объект wait со средой (environment), как я описывал в статье за сентябрь. Первый параметр принимает функцию обратного вызова, которая будет поставлена в очередь пула потоков, как только ожидание завершится. Она объявляется так:

void CALLBACK callback(PTP_CALLBACK_INSTANCE, void * context,
  PTP_WAIT, TP_WAIT_RESULT);

Аргумент TP_WAIT_RESULT обратного вызова — целое значение без знака, сообщающее о причине, по которой завершилось ожидание. Значение WAIT_OBJECT_0 указывает, что ожидание завершилось успешно, так как синхронизирующий объект перешел в свободное состояние. Значение WAIT_TIMEOUT говорит о том, что период ожидания истек до того, как синхронизирующий объект перешел в свободное состояние. Как можно задать период ожидания и конкретный синхронизирующий объект, на котором вы будете ждать? Это работа на удивление сложной функции SetThreadpoolWait. Точнее, она достаточно проста, пока вы не пытаетесь указать период ожидания. Рассмотрим пример:

handle e(CreateEvent( ... ));
check_bool(e);

SetThreadpoolWait(w.get(), e.get(), nullptr);

Сначала я создаю объект события, используя unique_handle typedef из своей статьи за июль. Функция SetThreadpoolWait задает синхронизирующий объект, на котором должен ждать объект wait. Последний (необязательный) параметр определяет период ожидания, но в этом примере я передаю нулевое значение указателя (nullptr), указывая, что пул потоков должен ждать неопределенно долго.

Структура FILETIME

Как же указать конкретный период ожидания? В этом-то и заключается вся сложность. Такие функции, как WaitForSingleObject, позволяют задавать период ожидания в миллисекундах как целое значение без знака. Однако функция SetThreadpoolWait ожидает указатель на структуру FILETIME, которая ставит перед разработчиком несколько трудных задач. Структура FILETIME является 64-битным значением, представляющим абсолютные дату и время от начала 1601 года в виде 100-наносекундных интервалов (на основе Coordinated Universal Time).

Для принятия относительных временных интервалов SetThreadpoolWait интерпретирует структуру FILETIME как знаковое 64-битное значение. Если вы передаете отрицательное значение, она воспринимает беззнаковую часть как период, относительный текущему времени, — опять же в 100-наносекундных интервалах. Стоит отметить, что относительный таймер (relative timer) прекращает отсчет при переходе компьютера в ждущий или спящий режим. Абсолютные значения периода ожидания, очевидно, не затрагиваются такими событиями. В любом случае использование FILETIME неудобно ни с абсолютными, ни с относительными значениями интервалов ожидания.

По-видимому, самый простой способ выражения интервалов ожидания в абсолютных значениях — заполнение структуры SYSTEMTIME и последующая подготовка структуры FILETIME без вашего участия — вызовом функции SystemTimeToFileTime:

SYSTEMTIME st = {};
st.wYear = ...
st.wMonth = ...
st.wDay = ...
st.wHour = ...
// и т. д.

FILETIME ft;
check_bool(SystemTimeToFileTime(&st, &ft));

SetThreadpoolWait(w.get(), e.get(), &ft);

В случае относительных значений тайм-аута придется попотеть немного больше. Сначала вы должны преобразовать относительное время в 100-наносекундные интервалы, а затем преобразовать их отрицательное 64-битное значение. Последнее сложнее, чем кажется. Вспомните, что компьютеры представляют знаковые целые, используя систему поразрядного дополнения до двух, а это означает, что у отрицательного числа должен быть установлен самый старший бит. Теперь добавьте к этому тот факт, что FILETIME на самом деле состоит из двух 32-битных значений. То есть вам придется позаботиться о корректном выравнивании при их обработке как 64-битного значения, а иначе произойдет сбой из-за ошибки выравнивания. Кроме того, нельзя просто использовать младшие 32 бита для хранения значения, так как самый старший бит находится в старших 32 битах.

Преобразование относительных значений периода ожидания

Обычно относительные периоды ожидания выражают в миллисекундах, поэтому позвольте мне продемонстрировать это преобразование здесь. Вспомните, что миллисекунда — это одна тысячная секунды, а наносекунда — одна миллиардная часть секунды. Можно взглянуть на это и по-другому: миллисекунда — это 1000 микросекунд, а микросекунда — 1000 наносекунд. В таком случае миллисекунда равна 10 000 100-наносекундных интервалов (такую единицу измерения ожидает функция SetThreadpoolWait). Выразить это можно разными способами, но вот один из подходов, который хорошо работает:

DWORD milliseconds = ...
auto ft64 = -static_cast<INT64>(milliseconds) * 10000;

FILETIME ft;
memcpy(&ft, &ft64, sizeof(INT64));

SetThreadpoolWait(w.get(), e.get(), &ft);

Заметьте, что я для надежности преобразую DWORD до умножения, чтобы избежать переполнения целого значения. Я также использую memcpy, поскольку reinterpret_cast потребовал бы выравнивания FILETIME по границе, кратной восьми байтам. Конечно, вы могли бы прибегнуть к этому варианту, но мой вариант немного изящнее. Еще более простой подход базируется на том факте, что компилятор Visual C++ выравнивает объединение (union) по самым большим требованиям к выравниванию любого из членов объединения. По сути, если вы корректно упорядочиваете члены объединения, то можете сделать это всего одной строкой:

union FILETIME64
{
  INT64 quad;
  FILETIME ft;
};

FILETIME64 ft = { -static_cast<INT64>(milliseconds) * 10000 };

SetThreadpoolWait(w.get(), e.get(), &ft.ft);

Но хватит о трюках с компилятором. Вернемся к пулу потоков. У вас может возникнуть соблазн попытаться использовать нулевой период ожидания. Обычно это делается в случае с WaitForSingleObject, чтобы определить, перешел ли синхронизирующий объект в свободное состояние, без реальной блокировки и ожидания. Однако эта процедура не поддерживается пулом потоков, так что лучше выбросьте это из головы.

Если вы хотите, чтобы конкретный объект работы прекратил ожидание своего синхронизирующего объекта, просто вызовите SetThreadpoolWait с нулевым значением указателя в ее втором параметре. Остерегайтесь создания явных условий для конкуренции потоков.

Последняя функция, относящаяся к объектам ожидания, — WaitForThreadpoolWaitCallbacks. Поначалу она может показаться похожей на функцию WaitForThreadpoolWorkCallbacks, используемую с объектами работы (я рассказывал о ней в статье за август). Но не дайте себя обмануть. Функция WaitForThreadpoolWaitCallbacks делает буквально то, что предполагает ее имя. Она ожидает любые обратные вызовы от конкретного объекта wait.

Подвох в том, что объект wait поставит обратный вызов в очередь, только когда соответствующий синхронизирующий объект перейдет в свободное состояние или истечет период ожидания. А до тех пор никакие обратные вызовы в очередь не ставятся, и функции Wait нечего ждать. Решение заключается в том, чтобы сначала вызывать SetThreadpoolWait с нулевыми значениями указателей, сообщая объекту wait о прекращении ожидания, а затем вызывать WaitForThreadpoolWaitCallbacks, чтобы избежать появления конкуренции между потоками:

SetThreadpoolWait(w.get(), nullptr, nullptr);
WaitForThreadpoolWaitCallbacks(w.get(), TRUE);

Как и следовало ожидать, второй параметр определяет, будут ли отменены любые необработанные обратные вызовы, которые могли поступить в очередь, но их выполнение пока не начато. Естественно, объекты ожидания нормально работают с группами очистки (cleanup groups). О том, как использовать группы очистки, см. в моей статье за октябрь (msdn.microsoft.com/magazine/hh456398). В более крупных проектах они реально помогают упростить множество трудных операций отмены и очистки.