Джулиан Бакнелл - Фундаментальные алгоритмы и структуры данных в Delphi
TtdProduceConsumeSync = class private
FHasData : THandle;
{семафор}
FNeedsData : THandle;
{семафор}
protected
public
constructor Create(aBufferCount : integer);
destructor Destroy; override;
procedure StartConsuming;
procedure StartProducing;
procedure StopConsuming;
procedure StopProducing;
end;
Первым делом, мы рассмотрим метод StartProducing (см. листинг 12.8), вызываемый производителем для запуска генерирования данных. Метод будет вызывать блокировку, если потребитель не успел использовать достаточно данных, чтобы производитель мог заменить их новыми. Метод достаточно прост: он просто ожидает передачи семафора "требуются данные". Как мы увидим, этот семафор будет передаваться потребителем.
Листинг 12.8. Метод StartProducing
procedure TtdProduceConsumeSync.StartProducing;
begin
{чтобы генерирование было начато, должен быть передан семафор "требуются данные"}
WaitForSingleObject(FNeedsData, INFINITE);
end;
Производитель будет вызывать второй метод, StopProducing (см. листинг 12.9), сообщающий потребителю о том, что он сгенерировал определенные (возможно все) данные, и что, следовательно, существуют данные, которые нужно использовать. Его реализация также проста: код просто передает семафор "имеются данные", ожидаемый потребителем.
Листинг 12.9. Метод StopProducing
procedure TtdProduceConsumeSync.StopProducing;
begin
{при генерировании каких-либо дополнительных данных потребителю нужно сообщить о необходимости их использования}
ReleaseSemaphore(FHasData, 1, nil);
end;
Третий метод, StartConsuming (листинг 12.10), вызывается потребителем перед тем, как он приступит к потреблению сгенерированных производителем данных. Метод будет вызывать блокировку на время ожидания семафора "имеются данные", который будет передаваться немедленно, если производитель уже сгенерировал какие-либо данные.
Листинг 12.10. Метод StartConcuming
procedure TtdProduceConsumeSync.StartConsuming;
begin
{чтобы можно было начать потребление данных, должен быть передан семафор "имеются данные"}
WaitForSingleObject(FHasData, INFINITE);
end;
Последний метод, StopConcuming (листинг 12.11), вызывается потребителем при считывании им достаточного объема (или всех) данных, чтобы производитель мог сгенерировать дополнительные данные. Очевидно, что этот метод всего лишь передает семафор "требуются данные", который будет предоставлять свободу действий производителю, если тот находится в состоянии ожидания.
Листинг 12.11. Метод StopConcuming
procedure TtdProduceConsumeSync.StopConsuming;
begin
{если какие-либо данные были использованы, нужно сигнализировать производителю о необходимости генерации дополнительных данных}
ReleaseSemaphore(FNeedsData, 1, nil);
end;
Полный исходный код класса TtdProduceConsumeSync можно найти на Web-сайте издательства, в разделе материалов. После выгрузки материалов отыщите среди них файл TDPCSync.pas.
Обратите внимание, что при использовании объекта семафора Windows неявно предполагается, что данные могут храниться только в 127 или меньшем количестве буферов, поскольку каждый раз, когда производитель сообщает, что потребитель может использовать какие-либо дополнительные данные, значение семафора "имеются данные" увеличивается на единицу (а его максимальное значение ограничено величиной, равной 127). Аналогичные соображения справедливы и по отношению к семафору "требуются данные". Однако в целом, это не столь уж большое ограничение. Во множестве сценариев с применением производителя-потребителя для передачи данных используется всего один буфер, а подпрограмма копирования потока, которую мы будем рассматривать, использует очередь буферов, содержащую 20 элементов.
Очередь буферов, используемая в рассматриваемом примере копирования потока, реализована в виде циклической очереди. Очередь создается с заранее выделенными всеми ее буферами. Код реализации этого класса приведен в листинге 12.12.
Обратите внимание, что мы не будем использовать диспетчер кучи во время процесса копирования потока, поскольку критический раздел защищает диспетчер кучи в многопоточной подпрограмме. Если начать вызывать подпрограммы распределения и освобождения памяти из потоков, они слишком легко смогут блокировать одна другую и, возможно, препятствовать достижению основной цели применения класса синхронизации производителя-потребителя.
Производитель будет заполнять буфер в начале очереди, а затем перемещать указатель начала очереди. С другой стороны, потребитель будет считывать, данные из буфера в конце очереди, а затем перемещать конец очереди. Процессы заполнения и считывания могут происходить одновременно, поскольку они используют различные буферы.
Листинг 12.12. Класс TQueuedBuffers, предназначенный для выполнения копирования потока
type
PBuffer= ^TBuffer;
TBuffer = packed record
bCount : longint;
bBlock : array [0..pred(BufferSize)] of byte;
end;
PBufferArray = ^TBufferArray;
TBufferArray = array [0..1023] of PBuffer;
type
TQueuedBuffers = class private
FBufCount : integer;
FBuffers : PBufferArray;
FHead : integer;
FTail : integer;
protected
function qbGetHead : PBuffer;
function qbGetTail : PBuffer;
public
constructor Create(aBufferCount : integer);
destructor Destroy; override;
procedure AdvanceHead;
procedure AdvanceTail;
property Head : PBuffer read qbGetHead;
property Tail : PBuffer read qbGetTail;
end;
constructor TQueuedBuffer s.Create(aBufferCount : 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;
end;
destructor TQueuedBuffers.Destroy;
var
i : integer;
begin
{освободить буферы}
if (FBuffers <> nil) then begin
for i := 0 to pred( FBuf Count) do
if (FBuffers^[i] <> nil) then
FreeMem(FBuffers^[i], sizeof(TBuffer));
FreeMem(FBuffers, FBufCount * sizeof(pointer));
end;
inherited Destroy;
end;
procedure TQueuedBuffers.AdvanceHead;
begin
inc(FHead);
if (FHead = FBufCount) then
FHead := 0;
end;
procedure TQueuedBuffers.AdvanceTail;
begin
inc(FTail);
if (FTail = FBuf Count) then
FTail := 0;
end;
function TQueuedBuffers.qbGetHead : PBuffer;
begin
Result := FBuffers^[FHead];
end;
function TQueuedBuffers.qbGetTail : PBuffer;
begin
Result := FBuffers^[FTail];
end;
Менее очевидно то, что указатели начала и конца очереди не должны быть защищены от изменений критическими разделами или какими-то аналогичными элементами. На первый взгляд это кажется противоречащим здравому смыслу и всем правилам совместного использования данных в различных потоках. Однако поток потребителя никогда не будет обращаться к указателю конца очереди. О наличии данных, которые нужно считать из указателя начала очереди, ему будет сообщать поток производителя (в этот момент времени указатели начала и конца очереди будут различными). Аналогично, поток производителя никогда не будет обращаться к указателю начала очереди, поскольку о наличии места для добавления данных в конце очереди ему будет сообщать поток потребителя.
Коды реализации классов производителя и потребителя приведены в листинге 12.13. Эти классы являются производными от класса TThread. Код реализации каждого из перекрытых методов Execute не отличается от ранее описанного. Поток производителя входит в цикл. На каждом шаге цикла он вызывает метод StartProducer объекта синхронизации, а затем считывает блок данных из исходного потока в буфер в конце очереди. После этого он смещает указатель конца очереди. И, в заключение, он вызывает метод StopProducing и повторяет цикл с начала. Выполнение цикла прекращается, как только поток производителя устанавливает буфер в состояние, соответствующее отсутствию в нем каких-либо данных (потребитель воспринимает это состояние в качестве признака "конец потока").
В свою очередь, цикл потока потребителя выполняется следующим образом. Вначале поток вызывает метод StartConsuming объекта синхронизации. Возврат из этого метода свидетельствует об отсутствии данных для считывания в объекте поставленных в очередь буферов. Поток считывает данные из буфера, определяемого указателем начала очереди, и записывает их в поток назначения. Затем он смещает указатель начала очереди. Сразу после считывания всех данных из заполненного буфера он вызывает метод StopConsuming объекта синхронизации и повторяет цикл сначала. Работа потребителя останавливается при получении им пустого буфера.
Листинг 12.13. Классы производителя и потребителя
type
TProducer = class (TThread) private
FBuffers : TQueuedBuffers;
FStream : TStream;
FSyncObj : TtdProduceConsumeSync;
protected
procedure Execute; override;
public
constructor Create(aStream : TStream;
aSyncObj : TtdProduceConsumeSync;
aBuffers : TQueuedBuffers);
end;
constructor TProducer.Create(aStream : TStream;
aSyncObj : TtdProduceConsumeSync;
aBuffers : TQueuedBuffers);
begin
inherited Create (true);
FStream := aStream;
FSyncObj :=,aSyncObj;
FBuffers aBuffers;
end;
procedure TProducer.Execute;
var
Tail : PBuffer;
begin
{выполнять до момента опустошения потока...}
repeat
{сигнализировать о готовности к началу генерирования данных}
FSyncObj.StartProducing;
{считать блок из потока в конечный буфер}
Tail FBuffers.Tail;
Tail^.bCount := FStream.Read(Tail^.bBlock, BufferSize);
{переместить указатель конца очереди}
FBuffers.AdvanceTail;
{поскольку выполняется запись нового буфера, необходимо сигнализировать о созданных данных}
