Детальное рассмотрение Блокировочно-Асинхронного Буфера (баб)
Сделав всю подготовительную работу, теперь можно рассказать о БАБ более подробно. Он содержит двунаправленный буфер и два потока, один чтения и один записи. Эти потоки выполняют действия записи и чтения с ограниченным буфером от имени потока VCL. Выполнение всех потоков можно представить графически с некоторыми отклонениями от существующих соглашений:
Эта
диаграмма выглядит довольно устрашающе,
так что, видимо, легче рассмотреть
рабочий пример. Давайте возьмем тот
случай, когда рабочий поток выполняет
блокирующую запись в БАБ.
Рабочий поток выполняет блокирующую запись.
Поток чтения БАБ в данный момент блокирован при попытке прочесть из двунаправленного буфера. В результате записи он разблокируется и успешно проводит чтение.
Он копирует прочитанные данные во вспомогательный буфер, локальный для класса потока, и генерирует событие прохождения данных, обрабатываемое БАБ.
Код обработки события прохождения данных БАБ, исполняемый в контексте потока чтения, посылает сообщение дескриптору своего окна, свидетельствуя, что данные были успешно прочитаны потоком чтения.
Поток чтения теперь ждет семафора, который сигнализирует о том, что данные прочитаны главным потоком VCL.
Несколько позже главный поток VCL обрабатывает сообщения из очереди, предназначенные компоненту, точно таким же образом, как и для всех компонентов, имеющих дескриптор окна.
Среди сообщений для компонента есть сообщение уведомления, посланное потоком чтения. Оно обрабатывается и генерирует событие OnRead для компонента.
Событие OnRead обрабатывается логикой приложения (вероятно, главной формой), и в результате, очевидно, поток VCL пытается прочитать данные.
Поток VCL вызывает метод БАБ AsyncRead.
AsyncRead копирует данные из вспомогательного буфера в главный поток VCL. Затем он освобождает семафор, которым блокирован поток чтения, разрешая ему попытаться осуществить новую операцию чтения из двунаправленного буфера.
Точно так же БАБ работает и при записи. Она осуществляется асинхронно потоком VCL, внутреннй поток записи БАБ пробуждается и проводит блокирующую запись в двунаправленный буфер, и по окончании записи поток VCL уведомляется через событие, что можно пытаться делать новые операции записи.
В сущности интерфейс между блокировкой и асинхронной операцией через отправления сообщения идентичен тому, который был неформально введен в более ранних примерах. Отличие для этого компонента в том, что детали спрятаны от конечного пользователя, и проблема решена более формальным, хорошо определенным способом.
Вот код этого компонента
unit BlockToAsyncBuf;
{ Martin Harvey 10/5/2000 }
Interface
uses Classes, Forms, Messages, Windows, BiDirBuf;
const
InternalBufferSize = 4;
WM_BLOCK_ASYNC = WM_USER + 2876;
Type
{ With this component, as with previous buffering schemes, one cannot read
or write nil pointers. }
TThreadNotify = (tnReaderDataFlow, tnWriterDataFlow);
TBlockAsyncThread = class(TThread)
Private
FDataSection: TRTLCriticalSection;
FIdleSemaphore: THandle;
FInterimBuf: Pointer;
FOnDataFlow: TNotifyEvent;
FBuffer: TBiDirBuf;
Protected
procedure DataFlow; virtual;
function GetItemsInTransit: integer;
public
constructor Create(CreateSuspended: boolean);
destructor Destroy; override;
published
property OnDataFlow: TNotifyEvent read FOnDataFlow write FOnDataFlow;
property Buffer: TBiDirBuf write FBuffer;
property ItemsInTransit: integer read GetItemsInTransit;
end;
TBAWriterThread = class(TBlockAsyncThread)
Protected
procedure Execute; override;
public
function WriteItem(Item: Pointer): boolean;
end;
TBAReaderThread = class(TBlockAsyncThread)
Private
Protected
procedure Execute; override;
public
function ReadItem: pointer;
end;
TBlockToAsyncBuf = class(TComponent)
Private
FHWND: THandle;
FBuffer: TBiDirBuf;
FReaderThread: TBAReaderThread;
FWriterThread: TBAWriterThread;
FOnRead, FOnWrite: TNotifyEvent;
Protected
procedure MessageHandler(var Msg: TMessage);
procedure ReaderDataFlow(Sender: TObject);
procedure WriterDataFlow(Sender: TObject);
procedure Read; virtual;
procedure Write; virtual;
function GetItemsInTransit: integer;
public
constructor Create(AOwner: TComponent); override;
destructor Destroy; override;
function BlockingRead: pointer;
function BlockingWrite(Item: pointer): boolean;
function AsyncRead: pointer;
function AsyncWrite(Item: pointer): boolean;
procedure ResetState;
published
property OnRead: TNotifyEvent read FOnRead write FOnRead;
property OnWrite: TNotifyEvent read FOnWrite write FOnWrite;
property ItemsInTransit: integer read GetItemsInTransit;
end;
implementation
procedure TBlockAsyncThread.DataFlow;
begin
if Assigned(FOnDataFlow) then FOnDataFlow(Self);
end;
constructor TBlockAsyncThread.Create(CreateSuspended: boolean);
begin
inherited Create(CreateSuspended);
InitializeCriticalSection(FDataSection);
FIdleSemaphore := CreateSemaphore(nil, 0, High(Integer), nil);
end;
destructor TBlockAsyncThread.Destroy;
begin
ReleaseSemaphore(FIdleSemaphore, 1, nil);
WaitFor;
DeleteCriticalSection(FDataSection);
CloseHandle(FIdleSemaphore);
end;
function TBlockAsyncThread.GetItemsInTransit: integer;
begin
EnterCriticalSection(FDataSection);
if Assigned(FInterimBuf) then
result := 1
else
result := 0;
LeaveCriticalSection(FDataSection);
end;
{ Buffer error handling needs to be discussed }
procedure TBAWriterThread.Execute;
var
Temp: Pointer;
Begin
while not Terminated do
begin
DataFlow;
WaitForSingleObject(FIdleSemaphore, INFINITE);
EnterCriticalSection(FDataSection);
Temp := FInterimBuf;
FInterimBuf := nil;
LeaveCriticalSection(FDataSection);
if not FBuffer.PutItem(bsSideA, Temp) then Terminate;
end;
end;
function TBAWriterThread.WriteItem(Item: Pointer): boolean;
begin
result := false;
EnterCriticalSection(FDataSection);
if not Assigned(FInterimBuf) then
begin
FInterimBuf := Item;
result := true;
end;
LeaveCriticalSection(FDataSection);
if Result then ReleaseSemaphore(FIdleSemaphore, 1, nil);
end;
procedure TBAReaderThread.Execute;
var
Temp: Pointer;
Begin
while not Terminated do
begin
Temp := FBuffer.GetItem(bsSideA);
if Assigned(Temp) then
begin
EnterCriticalSection(FDataSection);
FInterimBuf := Temp;
LeaveCriticalSection(FDataSection);
DataFlow;
WaitForSingleObject(FIdleSemaphore, INFINITE);
End
else Terminate;
end;
end;
function TBAReaderThread.ReadItem: pointer;
begin
EnterCriticalSection(FDataSection);
result := FInterimBuf;
LeaveCriticalSection(FDataSection);
if Assigned(Result) then ReleaseSemaphore(FIdleSemaphore, 1, nil);
end;
procedure TBlockToAsyncBuf.MessageHandler(var Msg: TMessage);
begin
if (Msg.Msg = WM_BLOCK_ASYNC) then
begin
case TThreadNotify(Msg.LParam) of
tnReaderDataflow: Read;
tnWriterDataflow: Write;
else
Assert(false);
end;
end;
end;
procedure TBlockToAsyncBuf.ReaderDataFlow(Sender: TObject);
begin
PostMessage(FHWND, WM_BLOCK_ASYNC, 0, Integer(tnReaderDataflow));
end;
procedure TBlockToAsyncBuf.WriterDataFlow(Sender: TObject);
begin
PostMessage(FHWND, WM_BLOCK_ASYNC, 0, Integer(tnWriterDataflow));
end;
procedure TBlockToAsyncBuf.Read;
begin
if Assigned(FOnRead) then FOnRead(Self);
end;
procedure TBlockToAsyncBuf.Write;
begin
if Assigned(FOnWrite) then FOnWrite(Self);
end;
constructor TBlockToAsyncBuf.Create(AOwner: TComponent);
begin
inherited Create(AOwner);
FHWND := AllocateHWnd(MessageHandler);
FBuffer := TBiDirBuf.Create;
FBuffer.Size := InternalBufferSize;
FReaderThread := TBAReaderThread.Create(true);
FReaderThread.Buffer := Self.FBuffer;
FReaderThread.OnDataFlow := ReaderDataFlow;
FWriterThread := TBAWriterThread.Create(true);
FWriterThread.Buffer := Self.FBuffer;
FWriterThread.OnDataFlow := WriterDataFlow;
FReaderThread.Resume;
FWriterThread.Resume;
end;
procedure TBlockToAsyncBuf.ResetState;
begin
if Assigned(FReaderThread) then FReaderThread.Terminate;
if Assigned(FWriterThread) then FWriterThread.Terminate;
FBuffer.ResetState;
FReaderThread.Free;
FWriterThread.Free;
FReaderThread := nil;
FWriterThread := nil;
end;
destructor TBlockToAsyncBuf.Destroy;
begin
{ A few destruction subtleties here }
ResetState;
FBuffer.Free;
DeallocateHWnd(FHWND);
inherited Destroy;
end;
function TBlockToAsyncBuf.BlockingRead: pointer;
begin
result := FBuffer.GetItem(bsSideB);
end;
function TBlockToAsyncBuf.BlockingWrite(Item: pointer): boolean;
begin
result := FBuffer.PutItem(bsSideB, Item);
end;
function TBlockToAsyncBuf.AsyncRead: pointer;
begin
result := FReaderThread.ReadItem;
end;
function TBlockToAsyncBuf.AsyncWrite(Item: pointer): boolean;
begin
result := FWriterThread.WriteItem(Item);
end;
function TBlockToAsyncBuf.GetItemsInTransit: integer;
var
Entries: integer;
Begin
result := FReaderThread.ItemsInTransit + FWriterThread.ItemsInTransit;
if FBuffer.GetEntriesUsed(bsSideA, Entries) then
Inc(result, Entries);
if FBuffer.GetEntriesUsed(bsSideB, Entries) then
Inc(result, Entries);
end;
end.
Некоторые моменты стоит отметить особо. Обычно потомки TThread нечасто используют наследование, однако в данном случае потоки чтения и записи имеют много общего в своей функциональности, которая и реализована в базовом классе TBlockAsyncThread. Он содержит:
Вспомогательный буфер, содержащий единственный указатель.
Критическую секцию для обеспечения атомарного доступа ко вспомогательному буферу.
Указатель на двунаправленный буфер для использования операций блокировки. Он устанавливается в компоненте, указывая на двунаправленный буфер, используемый внутри БАБ.
Событие "OnDataFlow", которое обрабатывается компонентом БАБ.
Семафор ожидания. Он используется для реализации операций "Ждать записи VCL" и "Ждать чтения VCL" общепринятым способом.
Базовый класс потока также обеспечивает необходимый минимум общей функциональности: создание потока,его уничтожение, и переключатель события для OnDataFlow. У базового класса есть два потомка: TBAWriterThread и TBAReaderThread. Они реализуют конкретные методы выполнения потока, а также предоставляют методы чтения и записи, которые косвенно будут исполняться потоком VCL. Сам компонент БАБ хранит двунаправленный буфер и два потока. Кроме того, в нем также хранится дескриптор окна FHWND, который используется для специализированной обработки сообщений.