Реализовать рассмотренное приложение, в котором используется модель "производитель-потребитель", было достаточно просто. Теперь рассмотрим модель с одним производителем и несколькими потребителями. В этом случае имеется поток, который создает данные. Предположим, что существует несколько потоков, которым требуется считывать созданные данные. В упомянутом ранее примере использовались два потребителя, которые сжимали данные с применением разных алгоритмов. Еще одним примером мог бы служить браузер. Будем считать, что производитель выгружает \¥еЬ-страницу из удаленного сайта, а один потребитель считывает НТМЬ-код, чтобы выполнить его сохранение на диске, второй считывает код для его отображения на экране, а третий - с целью отображения индикатора выполнения. Создание этих процессов как отдельных потребителей упрощает написание кода, поскольку каждый процесс должен выполнять только одну задачу.

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

Так как же быть с производителем? Каким образом он узнает, что можно снова заполнять буфер данных? Понятно, что он может это делать только после того, как последний (предположительно самый медленный) потребитель прочитал достаточный объем данных, чтобы появилось место для его заполнения новыми данными (иначе говоря, как только буфер снова освободится). Это, в свою очередь, предполагает, что должен существовать счетчик потребителей для каждого буфера данных. Каждый раз, когда потребитель считывает данные из буфера, он уменьшает значение этого счетчика (число потребителей, которым еще только предстоит выполнить считывание данных из этого буфера). Таким образом, когда последний потребитель приступает к использованию каких-либо данных, известно, что он является последним, поскольку после уменьшения значение счетчика должно быть равно нулю. Обратите внимание, что потребители являются потоками и, следовательно, уменьшение значения счетчика следует выполнять безопасным для потоков образом.

Код этого расширенного класса TtdProduceManyConsumeSync, который позволяет нескольким потребителям потреблять данные, сгенерированные единственным производителем, приведен в листинге 12.15. Предполагается, что каждый поток потребителя имеет уникальный, начинающийся с нуля, идентификатор (на практике этого легко добиться, но при необходимости класс можно было бы расширить, чтобы потребители могли регистрироваться и отменять свою регистрацию, и чтобы идентификаторы присваивались им "на лету"). Затем потребитель использует этот идентификатор (числовое значение) при обращении к методам StartConsumer и StopConsumer.

Листинг 12.15. Класс синхронизации одного производителя и нескольких потребителей

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

Метод StartProducing, показанный в листинге 12.16, работает во многом аналогично описанному в предыдущем случае: он просто дожидается передачи ему семафора "требуются данные". (Этот семафор содержит значение, равное количеству буферов, что позволяет производителю заполнить все буфера.) type TtdProduceManyConsumeSync = class private FBufferCount : integer; {счетчик буферов данных) FBufferlnfo : TList; {циклическая очередь информации о буферах]

FBufferTail : integer; {конец циклической очереди буферов) FConsumerCount : integer; {счетчик потребителей) FConsumerlnfo : TList; {информация для каждого потребителя) FNeedsData : THandle; {семафор) protected public

constructor Create (aBufferCount : integer;
aConsumerCount : integer);
destructor Destroy;
override; procedure StartConsuming(aid : integer);
procedure StartProducing;
procedure StopConsuming(aid : integer);
procedure StopProducing;
end;

Метод StopProducing, также показанный в листинге 12.16, на этот раз должен выполнить несколько больший объем работы. Во-первых, счетчик использования потребителей только что заполненного им буфера должен быть установлен равным количеству потребителей. Обратите внимание, что поток производителя должен передавать все семафоры "имеются данные" (по одному для каждого потребителя), тем самым сообщая о наличии еще одного буфера, готового к использованию.

Листинг 12.16. Методы StartProducing и StopProducing

type PBuf fer Info = ATBuf ferInfo; TBuf f erlnf о « packed record

biToUseCount : integer; {счетчик потребителей, которым еще предстоит использовать буфер)
end;
type PConsumerlnfo = ATConsumerInfo;
TConsumerlnf о «
packed record ciHasData : THandle; {семафор)
ciHead : integer; {указатель на начало очереди) ends-procedure TtdProduceManyConsumeSync.StartProducing;
begin
{чтобы можно было начать генерацию данных, необходимо передать семафор "требуются данные")
WaitForSingleObject(FNeedsData, INFINITE);
ends-procedure TtdProduceManyConsumeSync.StopProducing;
var
i : integer;
Buflnfo : PBufferInfo;
Consumerlnfo : PConsumerlnfo;
begin
{в случае генерации каких-либо дополнительных данных необходимо установить счетчик потребителей буфера в конце очереди, чтобы тем самым обеспечить правильную обработку всех буферов)
Buflnfo := PBufferlnfо(FBufferlnfо[FBufferTail]);
BufInfoA.biToUseCount := FConsumerCount;
inc(FBufferTail);
if (FBufferTail >= FBufferCount) then FBufferTail := 0;
{теперь всем потребителям необходимо сообщить о наличии дополнительных данных)
for i := 0 to pred(FConsumerCount) do begin Consumerlnfo := PConsumerlnfo(FConsumerInfо[i]);
ReleaseSemaphore(ConsumerInfoA.ciHasData/ 1, nil);
ends-end;

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

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

procedure TtdProduceManyConsumeSync. StartConsuming (aid : integer); var
ConsumerInfо : PConsumerlnfo;
begin
{чтобы можно было начать потребление данных, потребителю с данным конкретным идентификатором должен быть передан семафор "имеются данные") Consumerlnfo := PConsumerlnfo(FConsumerlnfо[aid]); WaitForSingleObject(ConsumerInfоA.ciHasData, INFINITE);
end;
procedure TtdProduceManyConsumeSync. StopConsuming (aid : integer); var
Buflnfo : PBufferlnfo;
Consumerlnfo : PConsumerlnfo;
NumToRead : integer;
begin
{мы выполнили считывание данных в буфере, на который указывает указатель начала очереди) Consumerlnfo := PConsumerlnfo(FConsumerlnfо[aid]);
Buflnfo := PBufferInfo(FBufferInfo[ConsumerInfo/4.ciHead]);
NumToRead := InterlockedDecrement(BufInfoA.biToUseCount); {переместить указатель начала очереди) inc(ConsumerInfoA.ciHead);
if (Consumer Inf о A. ciHead >= FBuf ferCount) then
ConsumerlnfoA.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;
Buflnfo : PBufferlnfo;
ConsumerInfo : PConsumerlnfo;
begin
inherited Create;
{создать семафор "требуются данные"} GetRandomObjName (NameZ, 1 tdPMC. Needs Data1) ;
FNeedsData := CreateSemaphore(nil, aBufferCount, aBufferCount, NameZ);
if (FNeedsData = INVALID_HANDLE_VALUE) then
RaiseLastWin32Error; {создать циклическую очередь буферов и заполнить ее} FBufferCount := aBufferCount;
FBufferlnfo := TList.Create;
FBufferlnfо.Count : = aBufferCount;
for i := 0 to pred(aBuf ferCount) do begin
New(Buflnfo);
BufInfoA.biToUseCount :=0; FBufferlnfo[i] := Buflnfo;
end;
{создать информационный список потребителей и заполнить его} FConsumerCount := aConsumerCount;
FConsumerlnfo := TList.Create;
FConsumerInfо.Count : = aConsumerCount;
for i : = 0 to pred (aConsumerCount) do begin
New(ConsumerInfo);
FConsumerlnfo[i] := ConsumerInfo;
GetRandomObjName(NameZ, 1tdPMC.HasData4);
ConsumerlnfoA.ciHasData :=
CreateSemaphore(nil, 0, aBufferCount, NameZ);
if (Consumer Inf oA. ciHasData = INVALID__HANDLE__VALUE) then
RaiseLastWin32Error;
ConsumerlnfoA.ciHead := 0; ends-end;
destructor TtdProduceManyConsumeSync. Destroy;
var
i : integer;
Buflnfo : PBufferlnfo;
ConsumerInfo : PConsumerlnfo;
begin
{уничтожить семафор "требуются данные")
if (FNeedsData о INVALID_HANDLE_VALUE) then
CloseHandle(FNeedsData); {уничтожить информационный список потребителей) if (FConsumerlnfo <>
nil) then begin for i : = 0 to pred(FConsumerCount) do begin ConsumerInfo := PConsumerlnfo(FConsumerlnfo[i]);
if (Consumerlnfo <>
nil) then begin if (ConsumerInfоA.ciHasData о INVALID__HANDLE__VALUE) then
CloseHandle(ConsumerlnfoA.ciHasData);
Dispose(Consumerlnfo);
end;
end;
FConsumerlnfo.Free;
end;
{уничтожить информационный список буферов) if (FBufferlnfo <>
nil) then begin for i : = 0 to pred(FBufferCount) do begin Buflnfo := PBufferlnfo(FBufferlnfo[i]);
if (Buflnfo <>
nil) then Dispose(Buflnfo);
end;
FBufferlnfo.Free;
end;
inherited Destroy;
end;

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

Полный исходный код реализации класса TtdProduceManyConsumeSync можно найти на Web-сайте издательства, в разделе материалов. После выгрузки материалов отыщите среди них файл TDPCSync.pas.

В качестве примера программы мы рассмотрим подпрограмму многопоточного копирования, выполняющую копирование потока в три других потока. Как и в случае примера, приведенного в листинге 12.14, производитель будет считывать исходный поток в буфера, количество которых может доходить до 20. Потребители, количество которых теперь равняется трем, будут считывать буфера и выполнять запись в собственные потоки.

Класс TQueuedBuffers (листинг 12.19) должен быть несколько изменен, поскольку ему необходимо хранить указатель начала очереди для нескольких потребителей и, следовательно, он должен содержать массив таких указателей.

Листинг 12.19. Класс TQueuedBuffers для модели с несколькими потребителями type PBuf fer = ATBuf fer; TBuf fer = packed record

bCount : longint;
bBlock : array [0. .pred(BufferSize) ] of byte;
end;
PBuf ferArray = ATBuf ferArray;
TBufferArray = array [0. .pred(MaxBuffers) ] of PBuffer;
TQueuedBuffers = class private FBufCount : integer;
FBuffers : PBufferArray;
FConsumerCount : integer;
FHead : array [0..pred(MaxConsumers)] of integer;
FTail : integers-protected
function qbGetHead(alnx : integer) : PBuffer;
function qbGetTail : PBuffer;
public
constructor Create (aBufferCount : integer ;
aConsumerCount : integer);
destructor Destroy;
override; procedureAdvanceHead(aConsumerId : integer);
procedure AdvanceTail;
property Head [alnx : integer] : PBuf fer read qbGetHead;
property Tail : PBuffer read qbGetTail;
property ConsumerCount : integer read FConsumerCount;
ends-constructor TQueuedBuffers. Create (aBufferCount : integer;
aConsumerCount : integer);
var
i : integers-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 TQueuedBuf fers. Destroy;
var
i : integer;
begin
{освободить буферы] if (FBuffers onil) then begin for i := 0 to pred(FBufCount) do if (FBuffersA[i] onil) then FreeMem(FBuffersA[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 TQueuedBuf fers.AdvanceTail;
begin
inc(FTail);
if (FTail = FBufCount) then FTail := 0;
end;
function TQueuedBuffers.qbGetHead(aInx : integer) : PBuffer;
begin
Result := FBuffersA[FHead[aInx]];
end;
function TQueuedBuffers.qbGetTail : PBuffer;
begin
Result := FBuffersA [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);
ends-constructor TProducer. Create (aStream : TStream;
aSyncObj : TtdProduceManyConsumeSync;
aBuffers : TQueuedBuffers);
begin
inherited Create (true) ;
FStream := aStream;
FSyncObj : = aSyncObj;

FBuffers := aBuffers; ends-procedure TProducer.Execute; var Tail : PBuffer; begin {выполнять до тех nop, пока поток не будет исчерпан. . .} repeat

{передать сигнал о готовности к началу генерации данных) FSyncObj.StartProducing;
{выполнить считывание блока из потока в конечный буфер очереди) Tail := FBuffers.Tails-Tail74.bCount :== FStream. Read (TailA .ЬВ1оск, 1024); {переместить указатель конца очереди) FBuffers.AdvanceTail;
{передать сигнал о прекращении генерации данных)
FSyncObj.StopProducing;
until (TailA.bCount = 0);
ends-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
{передать сигнал о готовности к началу потребления данных)
FSyncObj.StartConsuming(FID);
{выполнить считывание начального буфера очереди)
Head := FBuffers.Head[FID];
{до тех пор, пока начальный буфер не пуст...)
while (HeadA.bCount <>0) do begin
{выполнить запись блока из начального буфера очереди в поток)
FStream.Write(HeadA.bBlock, HeadA.bCount);
{переместить указатель начала очереди)
FBuffers.AdvanceHead(FID);
{обработка этого буфера завершена)
FSyncObj.StopConsuming(FID);
{передать сигнал о повторной готовности к началу потребления данных) FSyncObj.StartConsuming(FID); {выполнить считывание начального буфера очереди) Head := FBuffers.Head[FID];
end;
{обработка последнего буфера завершена) FSyncObj.StopConsuming(FID);
end;

И, наконец, рассмотрим подпрограмму копирования потоков, код которой показан в листинге 12.21.

Листинг 12.21. Копирование потоков с применением модели "производитель-потребитель"

procedure ThreadedMultiCopyStream(aSrcStream : TStream;
aDestCount : integer;
aDestStreams : PStreamArray);
var
i : integer;
SyncObj : TtdProduceManyConsumeSync;
Buffers : TQueuedBuffers;
Producer : TProducer;
Consumer : array [0. .pred(MaxConsumers) ] of TConsumer;
WaitArray : array [0. .MaxConsumers] of THandle;
begin
SyncObj nils-Buffers nils-Producer :=nil;
for i := 0 to pred(MaxConsumers) do
Consumer[i] := nil;
for i : = 0 to MaxConsumers do
WaitArray[i] := 0; try
{создать объект синхронизации)
SyncObj : * TtdProduceManyConsumeSync.Create(20, aDestCount);
{создать объект буфера с очередью)
Buffers := TQueuedBuffers.Create(20, aDestCount);
{создать поток производителя и сохранить его дескриптор)
Producer : = TProducer.Create(aSrcStream, SyncObj, Buffers);
WaitArray[0] : = Producer.Handle;

{создать потоки потребителей и сохранить их дескрипторы) for i := 0 to pred(aDestCount) do begin Consumer [ i ] : = TConsumer. Create (

aDestStreamsA[i], SyncObj, Buffers, i);
WaitArray[i+1] : - Consumer[i].Handle;
end;
{запустить потоки)
for i := 0 to pred(aDestCount) do
Consumer[i].Resume;
Producer.Resume; {ожидать завершения потоков)
WaitForMultipleObjects(l+aDestCount, @WaitArray, true, INFINITE);
finally Producer.Free;
for i :« 0 to pred(aDestCount) do
Consumer[i].Free;
Buffers.Free;
SyncObj.Free;
ends-end;

Большая часть кода предназначена для выполнения тех же рутинных задач, что и в модели с одним потребителем, представленной в листинге 12.14, за исключением того, что на этот раз необходимо заботиться о нескольких потребителях. Полный код подпрограммы находится в файлах TstNCpy.dpr и TstNCpyu.pas на Web-сайте издательства, в разделе материалов.

Модель с одним производителем и одним потребителем || Оглавление || Поиск различий между двумя файлами


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



Новости за месяц

  • Июнь
    2019
  • Пн
  • Вт
  • Ср
  • Чт
  • Пт
  • Сб
  • Вс