Синхронизация процессов при работе с Windows. Waitable timer (таймер ожидания)

Таймер ожидания переходит в сигнальное состояние по завершении заданного интервала времени.
Для его создания используется функция createwaitabletimer:

function createwaitabletimer(
lptimerattributes: psecurityattributes; // Адрес структуры
// tsecurityattributes
bmanualreset: bool; // Задает, будет ли таймер переходить в
// сигнальное состояние по завершении функции
// ожидания
lptimername: pchar // Имя объекта
): thandle; stdcall;
Когда параметр bmanualreset равен true, то таймер после срабатывания функции ожидания остается в сигнальном состоянии до явного вызова setwaitabletimer, если false-таймер автоматически переходит в несигнальное состояние.

Если lptimername совпадает с именем уже существующего в системе таймера, то функция возвращает его идентификатор, позволяя использовать объект для синхронизации между процессами. Имя таймера не должно совпадать с именем уже существующих объектов типов event, semaphore, mutex, job или file-mapping.

Идентификатор уже существующего таймера можно получить функцией:

function openwaitabletimer(
dwdesiredaccess: dword; // Задает права доступа к объекту
binherithandle: bool; // Задает, может ли объект наследоваться
// дочерними процессами
lptimername: pchar // Имя объекта
): thandle; stdcall;
Параметр dwdesiredaccess может принимать следующие значения:

timer_all_access
Разрешает полный доступ к объекту

timer_modify_state
Разрешает изменять состояние таймера функциями setwaitabletimer и cancelwaitabletimer

synchronize
Только для windows nt — разрешает использовать таймер в функциях ожидания
 

После получения идентификатора таймера поток может задать время его срабатывания функцией setwaitabletimer:

function setwaitabletimer(
htimer: thandle; // Идентификатор таймера
const lpduetime: tlargeinteger; // Время срабатывания
lperiod: longint; // Период повторения срабатывания
pfncompletionroutine: tfntimerapcroutine; // Процедура-обработчик
lpargtocompletionroutine: pointer;// Параметр процедуры-обработчика
fresume: bool // Задает, будет ли операционная
// система «пробуждаться»
): bool; stdcall;

Рассмотрим параметры подробнее.

lpduetime
Задает время срабатывания таймера. Время задается в формате tfiletime и базируется на coordinated universal time (utc), то есть должно указываться по Гринвичу. Для преобразования системного времени в tfiletime используется функция systemtimetofiletime. Если время имеет положительный знак, оно трактуется как абсолютное, если отрицательный — как относительное от момента запуска таймера.

lperiod
Задает срок между повторными срабатываниями таймера. Если lperiod равен 0, то таймер сработает один раз.

pfncompletionroutine
Адрес функции, объявленной как:

procedure timerapcproc(
lpargtocompletionroutine: pointer; // данные
dwtimerlowvalue: dword; // младшие 32 разряда значения таймера
dwtimerhighvalue: dword; // старшие 32 разряда значения таймера
); stdcall;
Эта функция вызывается, когда срабатывает таймер, если поток, ожидающий его срабатывания, использует функцию ожидания, поддерживающую асинхронный вызов процедур. В функцию передаются три параметра:

lpargtocompletionroutine — значение, переданное в качестве одноименного параметра в функцию setwaitabletimer. Приложение может использовать его для передачи в процедуру обработки адреса блока данных, необходимых для ее работы
dwtimerlowvalue и dwtimerhighvalue – соответственно члены dwlowdatetime и dwhighdatetime структуры tfiletime. Они описывают время срабатывания таймера. Время задается в utc-формате (по Гринвичу).
Если дополнительная функция обработки не нужна, в качестве этого параметра можно передать nil.

lpargtocompletionroutine
Это значение передается в функцию pfncompletionroutine при ее вызове.

fresume
Определяет необходимость «пробуждения» системы, если на момент срабатывания таймера она находится в режиме экономии электроэнергии (suspended). Если операционная система не поддерживает пробуждение и fresume равно true, то функция setwaitabletimer выполнится успешно, однако последующий вызов getlasterror вернет результат error_not_supported.

Если необходимо перевести таймер в неактивное состояние, это можно сделать функцией:

function cancelwaitabletimer(htimer: thandle): bool; stdcall;
Эта функция не изменяет состояния таймера и не приводит к срабатыванию функций ожидания и вызову процедур-обработчиков.

По завершении работы объект должен быть уничтожен функцией closehandle.

Создадим класс, который ожидает в отдельном потоке наступления заданного времени, а затем вызывает процедуру главного потока приложения. Такой класс может использоваться, например, в планировщике заданий (поскольку таймер ожидания позволяет задавать время срабатывания в абсолютных величинах, отпадает необходимость постоянно анализировать текущее время, используя обычный таймер windows):

unit waitthread;

interface

uses classes, windows;

type
twaitthread = class(tthread)
waituntil: tdatetime;
procedure execute; override;
end;

implementation

uses sysutils;

procedure twaitthread.execute;
var
timer: thandle;
systemtime: tsystemtime;
filetime, localfiletime: tfiletime;
begin
timer := createwaitabletimer(nil, false, nil);
try
datetimetosystemtime(waituntil, systemtime);
systemtimetofiletime(systemtime, localfiletime);
localfiletimetofiletime(localfiletime, filetime);
setwaitabletimer(timer, tlargeinteger(filetime), 0,
nil, nil, false);
waitforsingleobject(timer, infinite);
finally
closehandle(timer);
end;
end;

end.

Использовать этот класс можно, например, следующим образом:

type
tform1 = class(tform)
button1: tbutton;
procedure button1click(sender: tobject);
private
procedure timerfired(sender: tobject);
end;

...

implementation

uses waitthread;

procedure tform1.button1click(sender: tobject);
var
t: tdatetime;
begin
with twaitthread.create(true) do
begin
onterminate := timerfired;
freeonterminate := true;
// Срок ожидания закончится через 5 секунд
waituntil := now + 1 / 24 / 60 / 60 * 5;
resume;
end;
end;

procedure tform1.timerfired(sender: tobject);
begin
showmessage('timer fired !');
end;
 

Дополнительные объекты синхронизации
Некоторые объекты win32 api не предназначены исключительно для целей синхронизации, однако могут использоваться с функциями синхронизации.
Такими объектами являются:

Сообщение об изменении папки (change notification)
windows позволяет организовать слежение за изменениями объектов файловой системы. Для этого служит функция findfirstchangenotification:

function findfirstchangenotification(
lppathname: pchar; // Путь к папке, изменения в которой нас
// интересуют
bwatchsubtree: bool; // Задает необходимость слежения за
// изменениями во вложенных папках
dwnotifyfilter: dword // Фильтр событий
): thandle; stdcall;
Параметр dwnotifyfilter — это битовая маска из одного или нескольких следующих значений:

file_notify_change_file_name
Слежение ведется за любым изменением имени файла, в том числе за созданием и удалением файлов
file_notify_change_dir_name
Слежение ведется за любым изменением имени папки, в том числе за созданием и удалением папок
file_notify_change_attributes
Слежение ведется за любым изменением атрибутов
file_notify_change_size
Слежение ведется за изменением размера файлов. Изменение размера происходит при записи в файл. Функция ожидания срабатывает только после успешного сброса дискового кэша
file_notify_change_last_write
Слежение ведется за изменением времени последней записи в файл, то есть фактически за любой записью в файл. Функция ожидания срабатывает только после успешного сброса дискового кэша
file_notify_change_security
Слежение ведется за любыми изменениями дескрипторов защиты

Идентификатор, возвращенный этой функцией, может использоваться в любой функции ожидания.
Он переходит в сигнальное состояние, когда в папке происходят изменения, запрошенные для слежения.
Можно продолжить слежение, используя функцию findnextchangenotification:

function findnextchangenotification(
hchangehandle: thandle
): bool; stdcall;
По завершении работы идентификатор должен быть закрыт при помощи функции findclosechangenotification:

function findclosechangenotification(
hchangehandle: thandle
): bool; stdcall;
Чтобы не блокировать исполнение основного потока программы функцией ожидания, удобно реализовать ожидание изменений в отдельном потоке. Реализуем поток на базе класса tthread. Для того чтобы можно было прервать исполнение потока методом terminate, необходимо, чтобы функция ожидания, реализованная в методе execute, также прерывалась при вызове terminate. Для этого будем использовать вместо waitforsingleobject функцию waitformultipleobjects и прерывать ожидание по событию (event), устанавливаемому в terminate:

type
tcheckfolder = class(tthread)
private
fonchange: tnotifyevent;
handles: array[0..1] of thandle; // Идентификаторы объектов
// синхронизации
procedure doonchange;
protected
procedure execute; override;
public
constructor create(createsuspended: boolean;
pathtomonitor: string; waitsubtree: boolean;
onchange: tnotifyevent; notifyfilter: dword);
destructor destroy; override;
procedure terminate;
end;

procedure tcheckfolder.doonchange;
// Эта процедура вызывается в контексте главного потока приложения
// В ней можно использовать вызовы vcl, изменять состояние формы,
// например перечитать содержимое tlistbox, отображающего файлы
begin
if assigned(fonchange) then
fonchange(self);
end;

procedure tcheckfolder.terminate;
begin
inherited; // Вызываем tthread.terminate, устанавливаем
// terminated = true
setevent(handles[1]); // Сигнализируем о необходимости
// прервать ожидание
end;

constructor tcheckfolder.create(createsuspended: boolean;
pathtomonitor: string; waitsubtree: boolean;
onchange: tnotifyevent; notifyfilter: dword);
var
boolforwin95: integer;
begin
// Создаем поток остановленным
inherited create(true);
// windows 95 содержит не очень корректную реализацию функции
// findfirstchangenotification. Для корректной работы необходимо,
// чтобы:
// - lppathname - не содержал завершающего слэша "" для
// некорневого каталога
// - bwatchsubtree - true должен передаваться как bool(1)
if waitsubtree then
boolforwin95 := 1
else
boolforwin95 := 0;
if (length(pathtomonitor) > 1) and
(pathtomonitor[length(pathtomonitor)] = ‘’) and
(pathtomonitor[length(pathtomonitor)-1] <> ‘:’) then
delete(pathtomonitor, length(pathtomonitor), 1);
handles[0] := findfirstchangenotification(
pchar(pathtomonitor), bool(boolforwin95), notifyfilter);
handles[1] := createevent(nil, true, false, nil);
fonchange := onchange;
// И, при необходимости, запускаем
if not createsuspended then
resume;
end;

destructor tcheckfolder.destroy;
begin
findclosechangenotification(handles[0]);
closehandle(handles[1]);
inherited;
end;

procedure tcheckfolder.execute;
var
reason: integer;
dummy: integer;
begin
repeat
// Ожидаем изменения в папке либо сигнала о завершении
// потока
reason := waitformultipleobjects(2, @handles, false, infinite);
if reason = wait_object_0 then begin
// Изменилась папка, вызываем обработчик в контексте
// главного потока приложения
synchronize(doonchange);
// И продолжаем поиск
findnextchangenotification(handles[0]);
end;
until terminated;
end;

Поскольку метод tthread.terminate не виртуальный, этот класс нельзя использовать с переменной типа tthread, так как в этом случае будет вызываться метод terminate класса tthread, который не может прервать ожидания, и поток будет выполняться до изменения в папке, за которой ведется слежение.

Устройство стандартного ввода с консоли (console input)
Идентификатор стандартного устройства ввода с консоли, полученный при помощи вызова функции getstdhandle(std_input_handle), можно использовать в функциях ожидания. Он находится в сигнальном состоянии, если очередь ввода консоли не пустая, и в несигнальном — если пустая. Это позволяет организовать ожидание ввода символов или при помощи функции waitformultipleobjects совместить его с ожиданием каких-либо других событий.

Задание (job)
job — это новый механизм windows 2000, позволяющий объединить группу процессов в одно задание и манипулировать ими одновременно. Идентификатор задания находится в сигнальном состоянии, если все процессы, ассоциированные с ним, завершились по причине истечения лимита времени на выполнение задания.

Процесс (process)
Идентификатор процесса, полученный при помощи функции createprocess, переходит в сигнальное состояние по завершении процесса, что позволяет организовать ожидание завершения процесса, например, при запуске из приложения внешней программы:

var
pi: tprocessinformation;
si: tstartupinfo;
...
fillchar(si, sizeof(si), 0);
si.cb := sizeof(si);
win32check(createprocess(nil, 'command.com', nil,
nil, false, 0, nil, nil, si, pi));
// Задерживаем исполнение программы до завершения процесса
waitforsingleobject(pi.hprocess, infinite);
closehandle(pi.hprocess);
closehandle(pi.hthread);
Следует понимать, что в этом случае вызывающий процесс будет заморожен полностью и не сможет обрабатывать сообщения. Поэтому, если дочерний процесс может выполняться в течение длительного времени, лучше использовать более корректный вариант ожидания, описанный в разделе, посвященном функции msgwaitformultipleobjects.

Поток (thread)
Идентификатор потока находится в несигнальном состоянии до тех пор, пока поток выполняется. По его завершении идентификатор переходит в сигнальное состояние. Это позволяет легко узнать, завершился ли поток, либо при помощи функции, ожидающей несколько объектов, организовать ожидание завершения одного или всех интересующих потоков.

Дополнительные механизмы синхронизации
Критические секции
Критические секции — это механизм, предназначенный для синхронизации потоков внутри одного процесса. Как и мьютекс, критическая секция может в один момент времени принадлежать только одному потоку, однако она предоставляет более быстрый и эффективный механизм, чем мьютексы. Перед использованием критической секции необходимо инициализировать ее функцией:

procedure initializecriticalsection(
var lpcriticalsection: trtlcriticalsection
); stdcall;
После создания объекта поток, перед доступом к защищаемому ресурсу, должен вызвать функцию:

procedure entercriticalsection(
var lpcriticalsection: trtlcriticalsection
); stdcall;
Если в этот момент ни один из потоков в процессе не владеет объектом, то поток становится владельцем критической секции и продолжает выполнение. Если секция уже захвачена другим потоком, то выполнение потока, вызвавшего функцию, приостанавливается до ее освобождения.

Поток, владеющий критической секцией, может повторно вызывать функцию entercriticalsection без блокирования своего исполнения. По завершении работы с защищаемым ресурсом поток должен вызвать функцию leavecriticalsection:

procedure leavecriticalsection(
var lpcriticalsection: trtlcriticalsection
); stdcall;
Эта функция освобождает объект независимо от количества предыдущих вызовов потоком функции entercriticalsection. Если имеются другие потоки, ожидающие освобождения секции, один из них становится ее владельцем и продолжает исполнение. Если поток завершился, не освободив критическую секцию, то ее состояние становится неопределенным, что может вызвать блокировку работы программы.

Имеется возможность попытаться захватить объект без замораживания потока. Для этого служит функция tryentercriticalsection:

function tryentercriticalsection(
var lpcriticalsection: trtlcriticalsection
): bool; stdcall;
Она проверяет, захвачена ли секция в момент ее вызова. Если да — функция возвращает false, в противном случае — захватывает секцию и возвращает true.

По завершении работы с критической секцией она должна быть уничтожена вызовом функции deletecriticalsection:

procedure deletecriticalsection(
var lpcriticalsection: trtlcriticalsection
); stdcall;
Рассмотрим пример приложения, осуществляющего в нескольких потоках загрузку данных по сети. Глобальные переменные bytessummary и timesummary хранят общее количество загруженных байтов и время загрузки. Эти переменные каждый поток обновляет по мере считывания данных; для предотвращения конфликтов приложение должно защитить общий ресурс при помощи критической секции:

var
// Глобальные переменные
criticalsection: trtlcriticalsection;
bytessummary: cardinal;
timesummary: tdatetime;
averagespeed: float;
...
// При инициализации приложения
initializecriticalsection(criticalsection);
bytessummary := 0;
timesummary := 0;
averagespeed := 0;

//В методе execute потока, загружающего данные.
repeat
bytesread := readdatablockfromnetwork;
entercriticalsection(criticalsection);
try
bytessummary := bytessummary + bytesread;
timesummary := timesummary + (now - threadstarttime);
if timesummary > 0 then
averagespeed := bytessummary / (timesummary/24/60/60);
finally
leavecriticalsection(criticalsection)
end;
until loadcomplete;

// При завершении приложения
deletecriticalsection(criticalsection);
delphi предоставляет класс, инкапсулирующий функциональность критической секции. Класс объявлен в модуле syncobjs.pas:

type
tcriticalsection = class(tsynchroobject)
public
constructor create;
destructor destroy; override;
procedure acquire; override;
procedure release; override;
procedure enter;
procedure leave;
end;
Методы enter и leave являются синонимами методов acquire и release соответственно и добавлены для лучшей читаемости исходного кода:

procedure tcriticalsection.enter;
begin
acquire;
end;

procedure tcriticalsection.leave;
begin
release;
end;
 

Защищенный доступ к переменным (interlocked variable access)
Часто возникает необходимость в совершении операций над разделяемыми между потоками 32-разрядными переменными. В целях упрощения решения этой задачи windows api предоставляет функции для защищенного доступа к ним, не требующие использования дополнительных (и более сложных) механизмов синхронизации. Переменные, используемые в этих функциях, должны быть выровнены на границу 32-разрядного слова. Применительно к delphi это означает, что если переменная объявлена внутри записи (record), то эта запись не должна быть упакованной (packed) и при ее объявлении должна быть активна директива компилятора {$a+}. Несоблюдение данного требования может привести к возникновению ошибок на многопроцессорных конфигурациях.

type
tpackedrecord = packed record
a: byte;
b: integer;
end;
// tpackedrecord.b нельзя использовать в функциях interlockedxxx

tnotpackedrecord = record
a: byte;
b: integer;
end;

{$a-}
var
a1: tnotpackedrecord;
// a1.b нельзя использовать в функциях interlockedxxx
i: integer
// i можно использовать в функциях interlockedxxx, так как переменные в
// delphi всегда выравниваются на границу слова безотносительно
// к состоянию директивы компилятора $a

{$a+}
var
a2: tnotpackedrecord;
// a2.b можно использовать в функциях interlockedxxx

function interlockedincrement(
var addend: integer
): integer; stdcall;
Функция увеличивает переменную addend на 1. Возвращаемое значение зависит от операционной системы:

windows 98, windows nt 4.0 и старше — возвращается новое значение переменной addend;
windows 95, windows nt 3.51:

если после изменения addend < 0, то возвращается отрицательное число, не обязательно равное addend;
если addend = 0, то возвращается 0;
если после изменения addend > 0, то возвращается положительное число, не обязательно равное addend.
 

function interlockeddecrement(
var addend: integer
): integer; stdcall;
Функция уменьшает переменную addend на 1. Возвращаемое значение аналогично функции interlockedincrement.

function interlockedexchange(
var target: integer;
value: integer
): integer; stdcall;
Функция записывает в переменную target значение value и возвращает предыдущее значение target.

Следующие функции для выполнения требуют windows 98 или windows nt 4.0 и старше.

function interlockedcompareexchange(
var destination: pointer;
exchange: pointer;
comperand: pointer
): pointer; stdcall;
Функция сравнивает значения destination и comperand. Если они совпадают, значение exchange записывается в destination. Функция возвращает начальное значение destination.

function interlockedexchangeadd(
addend: plongint;
value: longint
): longint; stdcall;
Функция добавляет к переменной, на которую указывает addend, значение value и возвращает начальное значение addend.

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

Если приложения или потоки одного процесса изменяют общий ресурс — защищайте доступ к нему при помощи критических секций или мьютексов.
Если доступ осуществляется только на чтение — защищать ресурс не обязательно
Критические секции более эффективны, но применимы только внутри одного процесса; мьютексы могут использоваться для синхронизации между процессами.
Используйте семафоры для ограничения количества обращений к одному ресурсу.
Используйте события (event) для информирования потока о наступлении какого-либо события.
Если разделяемый ресурс — 32-битная переменная, то для синхронизации доступа к нему можно использовать функции, обеспечивающие разделяемый доступ к переменным.
Многие объекты win32 позволяют организовать эффективное слежение за своим состоянием при помощи функций ожидания. Это наиболее эффективный с точки зрения расхода системных ресурсов метод.
Если ваш поток создает (даже неявно, при помощи coinitialize или функций dde) окна, то он должен обрабатывать сообщения. Не используйте в таком потоке функции, не позволяющие прервать ожидание по приходу сообщения с большим или неограниченным периодом ожидания. Используйте функции msgwaitforxxx.

Анатолий Тенцер
КомпьютерПресс http://www.compress.ru/