Менеджер потоков

Класс менеджера потоков:

TGsvThreadManager = classprivate FLatch: TRTLCriticalSection; FHashTable: array of TGsvThread; FCurHandle: Integer; FCount: Integer; FOnEmpty: TNotifyEvent; public constructor Create(aCapacity: Integer = 64); destructor Destroy; override; function Add(aThread: TGsvThread; aStart: Boolean = True): Integer; procedure Release(aHandle: Integer); function Lock(aHandle: Integer): TGsvThread; function TerminateAll: Integer; function ActiveThreadList: IGsvThreadList; property OnEmpty: TNotifyEvent read FOnEmpty write FOnEmpty;end;

Критическая секция FLatch используется для предоставления исключительного доступа к внутренним данным менеджера для того, чтобы избежать ситуацию наложения конкурирующих потоков, которые одновременно могут вызывать методы менеджера. Хэш-таблица организует быстрое преобразование дескриптора потока в его адрес, по сравнению с другими способами поиска хеш-таблица обладает прозводительностью O(1), то есть, время поиска не зависит от числа элементов. В менеджере реализована не совсем обычная хеш-таблица - вместо хеш-функции используется само значение дескриптора потока. Размер таблицы задается в конструкторе и обычно выбирается в полтора-два раза большим, чем предполагаемое число потоков. Дескриптор потока назначается с помощью целочисленной переменной, которая каждый раз увеличивается на 1. Это дает очень хорошую расстановку хеш-ключей. При многократном создании-уничтожении потоков может произойти ситуация, при которой два и более различных дескриптора будут иметь один и тот же хеш-индекс (который вычисляется как остаток от деления дескриптора на размер хеш-таблицы). В этом случае для хеш-индекса создается цепочка коллизий, которая выглядит как односвязный список.

Метод Add связывает объект потока с дескриптором, устанавливает его счетчик ссылок в 1 и добавляет объект потока в хеш-таблицу:

function TGsvThreadManager.Add(aThread: TGsvThread; aStart: Boolean): Integer;var hash: Integer; // хеш-код дескриптора: индекс в хеш-таблицеbegin // делаем все операции внутри критической секции чтобы исключить // наложение параллельных потоков EnterCriticalSection(FLatch); try Inc(FCurHandle); // создаем следующий дескриптор hash:= FCurHandle mod Length(FHashTable); aThread.FManager:= Self; aThread.FGsvHandle:= FCurHandle; aThread.FRefCount:= 1; // включаем объект в начало цепочки коллизий (если она есть) и // вносим объект в хеш-таблицу aThread.FCollision:= FHashTable[hash]; FHashTable[hash]:= aThread; Inc(FCount); Result:= FCurHandle; finally LeaveCriticalSection(FLatch); end; // активизируем параллельное выполнение потока if aStart then aThread.Resume;end;

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

Метод Release находит объект потока по его дескриптору, уменьшает счетчик ссылок и уничтожает объект, если он больше никому не нужен:

procedure TGsvThreadManager.Release(aHandle: Integer);var hash: Integer; th: TGsvThread; // поток, связанный с дескриптором prev: TGsvThread; // предыдущий поток в цепочке коллизийbegin EnterCriticalSection(FLatch); try // поиск объекта, связанного с дескриптором и предыдущего объекта в // цепочке коллизий. Предыдущий объект нужен из-за того, что // цепочка коллизий представляет собой односвязный список, а нам может // потребоваться удаление объекта из середины или конца списка hash:= aHandle mod Length(FHashTable); prev:= nil; th:= FHashTable[hash]; while Assigned(th) do begin if th.FGsvHandle = aHandle then Break; // проходим по цепочке коллизий prev:= th; th:= th.FCollision; end; if Assigned(th) then begin // объект потока еще существует, уменьшаем его счетчик ссылок Dec(th.FRefCount); // используем сравнение (<= 0) для защиты от ошибки повторного уничтожения if th.FRefCount <= 0 then begin // объект потока больше никому не нужен if not th.FFinished then begin // объект потока еще не завершен. Завершаем его th.Terminate; // после своего завершения поток самостоятельно вызовет Release, // поэтому временно увеличиваем его счетчик ссылок без уничтожения // объекта (не допускаем отрицательного значения счетчика ссылок) Inc(th.FRefCount); end else begin // объект потока завершен, // удаляем объект из хеш-таблицы и из цепочки коллизий if Assigned(prev) then prev.FCollision:= th.FCollision // объект в середине или в конце else FHashTable[hash]:= th.FCollision; // объект в начале цепочки Dec(FCount); // уничтожаем объект потока th.Free; end; end; end; // else - объекта с таким дескриптором не существует, ничего не делаем // Если список потоков пуст, то вызываем событие OnEmpty if (FCount = 0) and Assigned(FOnEmpty) then FOnEmpty(Self); finally LeaveCriticalSection(FLatch); end;end;

Особо отметим два момента. Если поток еще не завершен, но его счетчик ссылок уже равен 0, то вместо грубого уничтожения потока мы делаем мягкое действие - предлагаем потоку завершиться самостоятельно, вызывая его метод Terminate. Это достаточно важный фактор, так как поток при своем завершении может требовать некоторых действий, которые обязательно нужно сделать для корректного завершения работы потока. Второй момент - когда программа закрывается, то мы должны сначала инициировать завершение работы всех параллельных потоков и дождаться их действительного завершения. Обычно эта работа делается в методе формы CloseQuery, но для завершения параллельных потоков эта задача несколько усложняется, так как мы не можем перевести основной поток в состояние ожидания - при своем завершении параллельный поток может требовать взаимодействия с основным потоком и перевод основного потока в состояние ожидания приведет к дедлоку и мертвому зависанию программы. Для решения этой задачи и предназначено событие OnEmpty. Приведем пример кода, который решает задачу ожидания завершения параллельных потоков без риска дедлока:

procedure TForm1.FormCloseQuery(Sender: TObject; var CanClose: Boolean);begin FThreadManager.OnEmpty:= OnEmpty; CanClose:= FThreadManager.TerminateAll = 0;end; procedure TForm1.OnEmpty(Sender: TObject);begin PostMessage(Handle, WM_CLOSE_APP, 0, 0);end; procedure TForm1.OnCloseApp(var aMessage: TMessage);begin Close;end;

При завершении программы мы устанавливаем обработчик OnEmpty, который будет вызван менеджером потоков при завершении всех потоков, а затем вызываем метод TerminateAll, который инициирует завершение всех имеющихся параллельных потоков и возвращает количество еще не завершенных потоков. Если незавершенные потоки есть, то закрытие формы запрещается. Когда менеджер потоков обнаружит завершение последнего потока, он вызовет обработчик события OnEmpty. Мы не можем сразу в этом обработчике закрыть форму, так как завершение последнего потока выполняется еще в контексте его параллельного потока, а не основного VCL-потока. Поэтому посылаем форме асинхронное сообщение (PostMessage) - оно будет обрабатываться в основном потоке.

function TGsvThreadManager.TerminateAll: Integer;var hash: Integer; th: TGsvThread;begin Result:= 0; EnterCriticalSection(FLatch); try // обходим всю хеш-таблицу for hash:= Low(FHashTable) to High(FHashTable) do begin th:= FHashTable[hash]; while Assigned(th) do begin // завершаем поток. Если он уже завершен, то Terminate не вызывает // побочных действий th.Terminate; if not th.FFinished then Inc(Result); // проходим по цепочке коллизий th:= th.FCollision; end; end; finally LeaveCriticalSection(FLatch); end;end;

Обсудив создание и завершение потоков, перейдем к их жизни и более подробно рассмотрим метод Lock:

function TGsvThreadManager.Lock(aHandle: Integer): TGsvThread;var hash: Integer;begin EnterCriticalSection(FLatch); try hash:= aHandle mod Length(FHashTable); // поиск объекта потока по его дескриптору Result:= FHashTable[hash]; while Assigned(Result) do begin if Result.FGsvHandle = aHandle then Break; Result:= Result.FCollision; end; // объект существует, увеличиваем его счетчик ссылок, так как у объекта // появился еще один "пользователь" if Assigned(Result) then Inc(Result.FRefCount); finally LeaveCriticalSection(FLatch); end;end;

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

IGsvThreadList = interface['{2B09399A-07E9-47F5-9CB7-3E34230D37D1}'] function Count: Integer; function GetItem(aIndex: Integer): TGsvThread; property Items[aIndex: Integer]: TGsvThread read GetItem; default;end;

Все потоки, которые внесены в список, получают приращение счетчика ссылок, поэтому даже если сразу после создания списка они будут завершены, их объекты останутся действительными вплоть до уничтожения списка, а при уничтожении списка для каждого из потоков будет вызван метод Release. Может вызвать некоторое удивление, что список доступен через интерфейс, причем реализация списка скрыта в секции implementation. Этому есть очень простое объяснение. Использование интерфейса позволяет автоматически управлять временем жизни списка, так как компилятор Delphi автоматически уничтожит объект списка, когда ссылка на его интерфейс выйдет из области видимости. Приведем фрагмент кода:

var list: IGsvThreadList; i: Integer;begin list:= FThreadManager.ActiveThreadList; for i:= 0 to Pred(list.Count) do begin.... // некоторое действие с list[i] end;

Нам не требуется явно уничтожать список. Независимо от исключительных ситуаций, которые могут возникнуть внутри цикла for, объект списка будет уничтожен автоматически с помощью механизма подсчета ссылок, который компилятор Delphi применяет к интерфейсам. Использование интерфейса упрощает код и делает его менее подверженным ошибкам.

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

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