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 должен уничтожить все эти объекты и освободить всю выделенную память.
Полный исходный код реализации класса TtdProduceManyConsumeSync можно найти на Web-сайте издательства, в разделе материалов. После выгрузки материалов отыщите среди них файл TDPCSync.pas.
В качестве примера программы мы рассмотрим подпрограмму многопоточного копирования, выполняющую копирование потока в три других потока. Как и в случае примера, приведенного в листинге 12.14, производитель будет считывать исходный поток в буфера, количество которых может доходить до 20. Потребители, количество которых теперь равняется трем, будут считывать буфера и выполнять запись в собственные потоки.
Класс TQueuedBuffers (листинг 12.19) должен быть несколько изменен, поскольку ему необходимо хранить указатель начала очереди для нескольких потребителей и, следовательно, он должен содержать массив таких указателей.
Листинг 12.19. Класс TQueuedBuffers для модели с несколькими потребителями type
PBuffer = ^TBuffer;
TBuffer = packed record
bCount : longint;
bBlock : array [0..pred(BufferSize) ] of byte;
end;
PBufferArray = ^TBufferArray;
TBufferArray = array [0..pred(MaxBuffers) ] of PBuffer;
TQueuedBuffers = class private
FBufCount : integer;
FBuffers : PBufferArray;
FConsumerCount : integer;
FHead : array [0..pred(MaxConsumers)] of integer;
FTail : integer;
protected
function qbGetHead(aInx : integer): PBuffer;
function qbGetTail : PBuffer;
public
constructor Create(aBufferCount : integer;
aConsumerCount : integer);
destructor Destroy; override;
procedureAdvanceHead(aConsumerId : integer);
procedure AdvanceTail;
property Head [aInx : integer] : PBuffer read qbGetHead;
property Tail : PBuffer read qbGetTail;
property ConsumerCount : integer read FConsumerCount;
end;
constructor TQueuedBuffers.Create(aBufferCount : integer;
aConsumerCount : integer);
var
i : integer;
begin
inherited Create;
{распределить буферы}
FBuffers := AllocMem(aBufferCount * sizeof(pointer));
for i := 0 to pred(aBufferCount) do
GetMem(FBuffers^[i], sizeof(TBuffer));
FBufCount := aBufferCount;
FConsumerCount := aConsumerCount;
end;
destructor TQueuedBuffers.Destroy;
var
i : integer;
begin
{освободить буферы}
if (FBuffers <> nil) then begin
for i := 0 to pred(FBufCount) do
if (FBuffers^[i] <> nil) then
FreeMem(FBuffers^[i], sizeof(TBuffer));
FreeMem(FBuffers, FBufCount * sizeof(pointer));
end;
inherited Destroy;
end;
procedure TQueuedBuffers.AdvanceHead(aConsumerId : integer);
begin
inc(FHead[aConsumerId]);
if (FHead[aConsumerId] = FBufCount) then
FHead[aConsumerId] := 0;
end;
procedure TQueuedBuffers.AdvanceTail;
begin
inc(FTail);
if (FTail = FBufCount) then
FTail := 0;
end;
function TQueuedBuffers.qbGetHead(aInx : integer): PBuffer;
begin
Result := FBuffers^[FHead[aInx]];
end;
function TQueuedBuffers.qbGetTail : PBuffer;
begin
Result := FBuffers^ [FTail];
end;
Следующей мы рассмотрим реализацию классов производителя и потребителя (листинг 12.20). Класс производителя претерпел не слишком много изменений по сравнению с предыдущей реализацией, в то время как класс потребителя теперь содержит идентификационный номер, посредством которого он обращается к объекту буферов для получения нужного указателя начала очереди.
Листинг 12.20. Классы производителя и потребителя
type
TProducer * class(TThread) private
FBuffers : TQueuedBuffers;
FStream : TStream;
FSyncObj : TtdProduceManyConsumeSync;
protected
procedure Execute; override;
public
constructor Create(aStream : TStream;
aSyncObj : TtdProduceManyConsumeSync;
aBuffers : TQueuedBuffers);
end;
constructor TProducer.Create(aStream : TStream;
aSyncObj : TtdProduceManyConsumeSync;
aBuffers : TQueuedBuffers);
begin
inherited Create (true);
FStream := aStream;
FSyncObj := aSyncObj;
FBuffers := aBuffers;
end;
procedure TProducer.Execute;
var
Tail : PBuffer;
begin
{выполнять до тех nop, пока поток не будет исчерпан...}
repeat
{передать сигнал о готовности к началу генерации данных}
FSyncObj.StartProducing;
{выполнить считывание блока из потока в конечный буфер очереди}
Tail := FBuffers.Tail;
Tail74.bCount := FStream.Read (Tail^.ЬВ1оск, 1024);
{переместить указатель конца очереди}
FBuffers.AdvanceTail;
{передать сигнал о прекращении генерации данных}
FSyncObj.StopProducing;
until (Tail^.bCount = 0);
end;
type
TConsumer = class (TThread) private
FBuffers : TQueuedBuffers;
FID : integer;
FStream : TStream;
FSyncObj : TtdProduceManyConsumeSync;
protected
procedure Execute; override;
public
constructor Create(aStream : TStream;
aSyncObj : TtdProduceManyConsumeSync;
aBuffers : TQueuedBuffers;
alD : integer);
end;
constructor TConsumer.Create(aStream : TStream;
aSyncObj : TtdProduceManyConsumeSync;
aBuffers : TQueuedBuffers;
alD : integer);
begin
inherited Create (true);
FStream := aStream;
FSyncObj := aSyncObj;
FBuffers := aBuffers;
FID := alD;
end;
procedure TConsumer.Execute;
var
Head : PBuffer;
begin
{передать сигнал о готовности к началу потребления данных}
