Асинхронное взаимодействие

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

Важная проблема эффективности многопоточной обработки возникает в том случае, если поток должен обрабатывать несколько очередей входящих сообщений или несколько событий. Приведем цитату из [1]: "пассажиру, ожидающему автобуса номер 127 в общем случае придется ждать дольше, чем тому, кто может воспользоваться как 19-м, так и 127-м маршрутом автобуса, в зависимости от того, который из них раньше придет к остановке. В предположении что автобусы приходят в случайном порядке, у пассажира, имеющего выбор, время ожидания оказывается вдвое короче - парадоксально, но получается, что он ждет вдвое "быстрее"! Единственный способ достигнуть этого - ожидать именно первого из многих возможных событий; приобретение же более быстрого компьютера здесь не поможет".

Учитывая важность выбора первого из многих возможных событий, введем понятие альтернативного ожидания и дадим его практическое воплощение.

TGsvSelectMethod = procedure of object; TGsvSelect = classprivate FEvents: array[0..MAXIMUM_WAIT_OBJECTS - 1] of THandle; FMethods: array[0..MAXIMUM_WAIT_OBJECTS - 1] of TGsvSelectMethod; FCount: Cardinal; public constructor Create(aThread: TGsvThread); procedure Init; procedure Add(aEvent: THandle; aMethod: TGsvSelectMethod); function Wait(aTimeout: Cardinal = INFINITE): Boolean;end;

Альтернатива могут быть добавлена к списку возможных альтернатив с помощью метода Add, который принимает 2 аргумента - дескриптор события и метод-обработчик. Альтернативы могут добавляться статически, на этапе создания объекта, а могут динамически, причем список альтернатив может быть различным в зависимости от различных условий. В языке Ada это делает оператор отбора. В Delphi мы можем воспользоваться оператором if или case. Список альтернатив инициализируется вызовом Init, а альтернативное ожидание - вызовом Wait. Если в процессе ожидания возникло событие, то будет вызван метод, связанный с данным событием и Wait вернет True, в противном случае Wait вернет False. Обработчик события - это обычный метод класса параллельного потока. Функция Wait принимает в качестве аргумента значение таймаута (в миллисекундах), если за время таймаута никаких событий не произошло, то Wait вернет False. Отметим важную деталь - в список ожидаемых событий неявно вносится событие FTerminationEvent, которое делает возможным форсированное завершение ожидания при завершении потока - в этом случае Wait также вернет False. Приведем пример кода, в котором параллельный поток создает набор альтернатив и выполняет ожидание. Предполагается, что в конструкторе объекта потока создан объект FSelect класса TGsvSelect:

FSelect.Init;if условие1 then FSelect.Add(событие1, метод1); if условие2 then FSelect.Add(событие2, метод2); FSelect.Add(безусловное_событие3, метод3);if not FSelect.Wait(100) then begin if Terminated then // обработка завершения потока else // обработка таймаутаend;

Теперь рассмотрим более подробно реализацию класса TGsvSelect:

constructor TGsvSelect.Create(aThread: TGsvThread);begin inherited Create; FEvents[0]:= aThread.FTerminationEvent; FMethods[0]:= nil; FCount:= 1;end; procedure TGsvSelect.Init;begin FCount:= 1;end; procedure TGsvSelect.Add(aEvent: THandle; aMethod: TGsvSelectMethod);begin Assert(FCount <= High(FEvents)); FEvents[FCount]:= aEvent; FMethods[FCount]:= aMethod; Inc(FCount);end; function TGsvSelect.Wait(aTimeout: Cardinal): Boolean;var res, i: Cardinal;begin Result:= False; res:= WaitForMultipleObjects(FCount, @FEvents[0], False, aTimeout); if res < (WAIT_OBJECT_0 + FCount) then begin Result:= res > WAIT_OBJECT_0; if Result then begin i:= res - WAIT_OBJECT_0; if Assigned(FMethods[i]) then FMethods[i](); end; end;end;

При создании объекта первым событием в списке становится завершающее событие. Процедура Add регистрирует событие - добавляет его в список (в массив) событий и добавляет в список методов связанный с этим событием метод. Функция Wait выполняет альтернативное ожидание по всем зарегистрированным событиям, включая завершающее событие. Следует отметить, что функция Wait может вернуть False не только при таймауте или при завершающем событии. Это может случиться также в том случае, когда в качестве события используется мьютекс, от которого отказался создавший его процесс. Но поскольку мы ограничились случаем взаимодействия потоков в рамках одного приложения (одного процесса операционной системы), то это событие у нас не возникнет никогда.

Теперь пора уточнить термин "событие", который мы использовали до этого момента без особого разъяснения. Под событием мы будем понимать изменение состояния объекта ядра операционной системы. В качестве событий можно использовать:

1. Event - объект ядра, который так и называется - событие. Ввиду его важности мы подробно рассмотрим этот вид события немного ниже.
2. Mutex (мьютекс) - напоминает критическую секцию, то есть, только один из конкурирующий потоков может захвать мьютекс, а остальные потоки будут приостановлены. Мьютекс имеет перед критической секцией ряд преимуществ, в частности, время ожидания мьютекса можно ограничивать таймаутом, мьютекс можно использовать для синхронизации параллельных потоков из разных приложений. См. в MSDN функции CreateMutex, OpenMutex, ReleaseMutex.
3. Semaphore (семафор) - напоминает мьютекс, но разрешает захват ресурса не одним, а несколькими потоками, причем число потоков ограничено доступным количеством ресурса. См. функции CreateSemaphore, OpenSemaphore, ReleaseSemaphore.
4. Change notification (уведомление файловой системы) - событие возникает при различных изменениях в файловой системе. См. FindFirstChangeNotification.
5. Process - событие возбуждается при завершении процесса. См. CreateProcess.
6. Thread - событие возбуждается при завершении потока. Ожидание этого события позволяет родительскому потоку дождаться завершения дочерних потоков. В классе TGsvThread дескриптор этого события называется SysHandle.

Кроме перечисленных событий операционная система Windows предоставляет еще несколько событий, но они доступны только в Windows NT. Особо можно еще отметить сокетные события, которые создаются функцией WSACreateEvent - эти события также можно использовать при альтернативных ожиданиях на сокетах.

Рассмотрим более подробно событие "Event":

TGsvEvent = classprivate FHandle: THandle; procedure SetState(aState: Boolean); public constructor Create; destructor Destroy; override; function Wait(aThread: TGsvThread; aTimeout: Cardinal = INFINITE): Boolean; property Handle: THandle read FHandle; property State: Boolean write SetState;end;

Событие может находиться в двух состояниях - активном (сигнализирущее состояние) и сброшенном (несигнализирующее состояние). В активное состояние событие переводится установкой свойства State в True, а сбрасывается событие установкой State в False. Эти редкий тип свойства, которое доступно только для установки, но не доступно для получения. Событие автоматически сбрасывается при каком-либо успешном ожидании, после чего его нужно явно перевести в активное состояние. Событие может быть внутренним объектом потока, а может не принадлежать никакому потоку конкретно. Поток может ожидать только данного события с помощью метода Wait, либо использовать его дескриптор Handle для альтернативного ожидания. Приведем реализацию метода Wait:

function TGsvEvent.Wait(aThread: TGsvThread; aTimeout: Cardinal): Boolean;var objs: array[0..1] of THandle; cnt: Integer;begin objs[0]:= FHandle; cnt:= 1; if Assigned(aThread) then begin objs[1]:= aThread.FTerminationEvent; cnt:= 2; end; Result:= WaitForMultipleObjects(cnt, @objs[0], False, aTimeout) = WAIT_OBJECT_0;end;

Если событие ожидает поток aThread, то кроме собственно события также будет ожидаться завершающее событие потока, которое необходимо для форсированного прекращения ожидания. Если поток не задан (aThread = nil), то метод Wait будет ожидать только своего события - это может пригодиться основному VCL-потоку.

Чтобы суммировать материал этого раздела, приведем реализацию очереди ограниченной длины, которую можно использовать для взаимодействия обменом сообщениями. Будем предполагать, что все сообщения являются объектами. Сообщения добавляются в конец очереди, а извлекаются из начала очереди. Исключение из правила - высокоприоритетное сообщение, которое помещается в начало очереди без учета ограничения на длину очереди.

Класс очереди:

TGsvQueue = classprivate FGetEvent: TGsvEvent; FPutEvent: TGsvEvent; FLatch: TGsvLatch; FList: TList; FMaxCount: Integer; function GetCount: Integer; procedure SetEvents; public constructor Create(aMaxCount: Integer); destructor Destroy; override; function Get(aThread: TGsvThread; aTimeout: Cardinal = INFINITE): TObject; function Put(aThread: TGsvThread; aMessage: TObject; aTimeout: Cardinal = INFINITE): Boolean; procedure PutOutOfTurn(aMessage: TObject); property GetEvent: TGsvEvent read FGetEvent; property PutEvent: TGsvEvent read FPutEvent; property Count: Integer read GetCount; property MaxCount: Integer read FMaxCount;end;

Рассмотрим подробнее метод Get:

function TGsvQueue.Get(aThread: TGsvThread; aTimeout: Cardinal): TObject;begin Result:= nil; if not FGetEvent.Wait(aThread, aTimeout) then Exit; FLatch.Lock; try if FList.Count <> 0 then begin Result:= TObject(FList.Items[0]); FList.Delete(0); SetEvents; end; finally FLatch.Unlock; end;end;

Поток, вызвавший метод Get, сразу переходит в состояние ожидания непустой очереди. Если очередь содержит хоть одно сообщение, то поток продолжит свое выполнение без ожидания. Первое сообщение извлекается из списка внутри критической секции, а затем производится переустановка событий:

procedure TGsvQueue.SetEvents;begin FGetEvent.State:= FList.Count <> 0; FPutEvent.State:= FList.Count < FMaxCount;end;

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

Дата добавления: 2015-07-20