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

Полный исходный код реализации класса 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

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

FSyncObj.StartConsuming(FID);

{выполнить считывание начального буфера очереди}

Head := FBuffers.Head[FID];

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

while (Head^.bCount <> 0) do

begin

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

FStream.Write(Head^.bBlock, Head^.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 nil;

Buffers nil;

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(

aDestStreams^[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;

end;

end;

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

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

Существует множество программ, выполняющих подобные функции. В их числе и программа diff, которую можно считать прародительницей всех программ сравнения файлов. Пакет Microsoft Windows SDK содержит программу, названную WinDiff. Программа Visual SourceSafe, поставляемая компанией Microsoft, также предоставляет функцию, которая позволяет выбрать две версии файла, хранящиеся в базе данных, и просмотреть различия между ними.