Джулиан Бакнелл - Фундаментальные алгоритмы и структуры данных в Delphi

WaitForSingleObject(FNeedsData, INFINITE);

end;

procedure TtdProduceManyConsumeSync.StopProducing;

var

i : integer;

BufInfo : PBufferInfo;

ConsumerInfo : PConsumerInfo;

begin

{в случае генерации каких-либо дополнительных данных необходимо установить счетчик потребителей буфера в конце очереди, чтобы тем самым обеспечить правильную обработку всех буферов}

BufInfo := PBufferInfo(FBufferInfo[FBufferTail]);

BufInfo^.biToUseCount := FConsumerCount;

inc(FBufferTail);

if (FBufferTail >= FBufferCount) then

FBufferTail := 0;

{теперь всем потребителям необходимо сообщить о наличии дополнительных данных}

for i := 0 to pred(FConsumerCount) do

begin

ConsumerInfo := PConsumerInfo(FConsumerInfo[i]);

ReleaseSemaphore(ConsumerInfo^.ciHasData/ 1, nil);

end;

end;

Чтобы разобраться с работой алгоритма с точки зрения потребителя, взгляните на листинг 12.17. Метод StartConsuming должен дождаться передачи семафора "имеются данные", предназначенного для соответствующего потока потребителя (каждому потоку присвоен идентификатор потребителя). Метод StopConsuming -наиболее сложный во всем классе синхронизации. Вначале он извлекает информационную запись о буфере, соответствующую его собственному указателю на начало очереди. Затем он уменьшает значение счетчика потребителей, которым еще предстоит выполнить считывание (потребить) данный буфер. (подпрограмма InterlockedDecrement - это составная часть интерфейса WIN32 API. Она уменьшает значение своего параметра безопасным для потоков образом и возвращает новое значение параметра.) Затем метод увеличивает указатель на начало очереди для данного потока потребителя и, если теперь число потребителей, которым еще предстоит выполнить считывание этого буфера, равно нулю, передает производителю семафор "требуются данные", чтобы побудить его сгенерировать новые данные.

Листинг 12.17. Методы StartConsuming и StopConsuming

procedure TtdProduceManyConsumeSync.StartConsuming(aId : integer);

var

ConsumerInfo : PConsumerInfo;

begin

{чтобы можно было начать потребление данных, потребителю с данным конкретным идентификатором должен быть передан семафор "имеются данные"}

ConsumerInfo := PConsumerInfo(FConsumerInfo[aId]);

WaitForSingleObject(ConsumerInfo^.ciHasData, INFINITE);

end;

procedure TtdProduceManyConsumeSync.StopConsuming(aId : integer);

var

BufInfo : PBufferInfo;

ConsumerInfo : PConsumerInfo;

NumToRead : integer;

begin

{мы выполнили считывание данных в буфере, на который указывает указатель начала очереди}

ConsumerInfo := PConsumerInfo(FConsumerInfo[aId]);

BufInfo := PBufferInfo(FBufferInfo[ConsumerInfo^.ciHead]);

NumToRead := InterLockedDecrement(BufInfo^.biToUseCount);

{переместить указатель начала очереди}

inc(ConsumerInfo^.ciHead);

if (ConsumerInfo^.ciHead >= FBufferCount) then

ConsumerInfo^.ciHead := 0;

{если данный поток был последним, который должен был использовать этот буфер, производителю нужно сигнализировать о необходимости генерирования новых данных}

if (NumToRead = 0) then

ReleaseSemaphore(FNeedsData, 1, nil);

end;

Конструктор и деструктор этого класса должны создавать и уничтожать большое количество объектов синхронизации, а также всю информацию о буфере и потребителе.

Листинг 12.18. Создание и уничтожение объекта синхронизации

constructor TtdProduceManyConsumeSync.Create(aBufferCount : integer;

aConsumerCount : integer);

var

NameZ : array [0..MAX_PATH] of AnsiChar;

i : integer;

BufInfo : PBufferInfo;

ConsumerInfo : PConsumerInfo;

begin

inherited Create;

{создать семафор "требуются данные"}

GetRandomObjName(NameZ, 'tdPMC.Needs Data');

FNeedsData := CreateSemaphore(nil, aBufferCount, aBufferCount, NameZ);

if (FNeedsData = INVALID_HANDLE_VALUE) then

RaiseLastWin32Error;

{создать циклическую очередь буферов и заполнить ее}

FBufferCount := aBufferCount;

FBufferInfo := TList.Create;

FBufferInfo.Count := aBufferCount;

for i := 0 to pred(aBufferCount) do

begin

New(BufInfo);

BufInfo^.biToUseCount :=0;

FBufferInfo[i] := BufInfo;

end;

{создать информационный список потребителей и заполнить его}

FConsumerCount := aConsumerCount;

FConsumerInfo := TList.Create;

FConsumerInfo.Count := aConsumerCount;

for i := 0 to pred(aConsumerCount) do

begin

New(ConsumerInfo);

FConsumerInfo[i] := ConsumerInfo;

GetRandomObjName(NameZ, 'tdPMC.HasData');

ConsumerInfo^.ciHasData :=

CreateSemaphore(nil, 0, aBufferCount, NameZ);

if (Consumer Info^.ciHasData = INVALID__HANDLE__VALUE) then

RaiseLastWin32Error;

ConsumerInfo^.ciHead := 0;

end;

end;

destructor TtdProduceManyConsumeSync.Destroy;

var

i : integer;

BufInfo : PBufferInfo;

ConsumerInfo : PConsumerInfo;

begin

{уничтожить семафор "требуются данные"}

if (FNeedsData <> INVALID_HANDLE_VALUE) then

CloseHandle(FNeedsData);

{уничтожить информационный список потребителей}

if (FConsumerInfo <> nil) then begin

for i := 0 to pred(FConsumerCount) do

begin

ConsumerInfo := PConsumerInfo(FConsumerInfo[i]);

if (ConsumerInfo <> nil) then begin

if (ConsumerInfo^.ciHasData <> INVALID__HANDLE__VALUE) then

CloseHandle(ConsumerInfo^.ciHasData);

Dispose(ConsumerInfo);

end;

end;

FConsumerInfo.Free;

end;

{уничтожить информационный список буферов}

if (FBufferInfo <> nil) then begin

for i := 0 to pred(FBufferCount) do

begin

BufInfo := PBufferInfo(FBufferInfo[i]);

if (BufInfo <> nil) then

Dispose(BufInfo);

end;

FBufferInfo.Free;

end;

inherited Destroy;

end;

Хотя, на первый взгляд, кажется, что в программе листинга 12.18 выполняется множество действий, в действительности все достаточно просто. Конструктор Create должен создать список буферов и заполнить его требуемым числом записей о буферах. Он должен также создать список потребителей и заполнить его соответствующим количеством записей о потребителях. Для каждой записи потребителя должен быть создан отдельный семафор. Деструктор Destroy должен уничтожить все эти объекты и освободить всю выделенную память.