Попытка использовать TThreadedQueue (Generics.Collections) в единой схеме множественного потребления. (Delphi XE). Идея состоит в том, чтобы вставлять объекты в очередь и пропускать несколько рабочих потоков в очередь.
Однако он работает не так, как ожидалось. Когда два или более рабочих потока вызывают PopItem, нарушения доступа выводятся из TThreadedQueue.
Если вызов PopItem сериализуется с критическим разделом, все в порядке.
Конечно, TThreadedQueue должен иметь возможность обрабатывать несколько пользователей, так что я что-то упускаю или это чистая ошибка в TThreadedQueue?
Вот простой пример для получения ошибки.
program TestThreadedQueue;
{$APPTYPE CONSOLE}
uses
// FastMM4 in '..\..\..\FastMM4\FastMM4.pas',
Windows,
Messages,
Classes,
SysUtils,
SyncObjs,
Generics.Collections;
type TThreadTaskMsg =
class(TObject)
private
threadID : integer;
threadMsg : string;
public
Constructor Create( ID : integer; const msg : string);
end;
type TThreadReader =
class(TThread)
private
fPopQueue : TThreadedQueue<TObject>;
fSync : TCriticalSection;
fMsg : TThreadTaskMsg;
fException : Exception;
procedure DoSync;
procedure DoHandleException;
public
Constructor Create( popQueue : TThreadedQueue<TObject>;
sync : TCriticalSection);
procedure Execute; override;
end;
Constructor TThreadReader.Create( popQueue : TThreadedQueue<TObject>;
sync : TCriticalSection);
begin
fPopQueue:= popQueue;
fMsg:= nil;
fSync:= sync;
Self.FreeOnTerminate:= FALSE;
fException:= nil;
Inherited Create( FALSE);
end;
procedure TThreadReader.DoSync ;
begin
WriteLn(fMsg.threadMsg + ' ' + IntToStr(fMsg.threadId));
end;
procedure TThreadReader.DoHandleException;
begin
WriteLn('Exception ->' + fException.Message);
end;
procedure TThreadReader.Execute;
var signal : TWaitResult;
begin
NameThreadForDebugging('QueuePop worker');
while not Terminated do
begin
try
{- Calling PopItem can return empty without waittime !? Let other threads in by sleeping. }
Sleep(20);
{- Serializing calls to PopItem works }
if Assigned(fSync) then fSync.Enter;
try
signal:= fPopQueue.PopItem( TObject(fMsg));
finally
if Assigned(fSync) then fSync.Release;
end;
if (signal = wrSignaled) then
begin
try
if Assigned(fMsg) then
begin
fMsg.threadMsg:= '<Thread id :' +IntToStr( Self.threadId) + '>';
fMsg.Free; // We are just dumping the message in this test
//Synchronize( Self.DoSync);
//PostMessage( fParentForm.Handle,WM_TestQueue_Message,Cardinal(fMsg),0);
end;
except
on E:Exception do begin
end;
end;
end;
except
FException:= Exception(ExceptObject);
try
if not (FException is EAbort) then
begin
{Synchronize(} DoHandleException; //);
end;
finally
FException:= nil;
end;
end;
end;
end;
Constructor TThreadTaskMsg.Create( ID : Integer; Const msg : string);
begin
Inherited Create;
threadID:= ID;
threadMsg:= msg;
end;
var
fSync : TCriticalSection;
fThreadQueue : TThreadedQueue<TObject>;
fReaderArr : array[1..4] of TThreadReader;
i : integer;
begin
try
IsMultiThread:= TRUE;
fSync:= TCriticalSection.Create;
fThreadQueue:= TThreadedQueue<TObject>.Create(1024,1,100);
try
{- Calling without fSync throws exceptions when two or more threads calls PopItem
at the same time }
WriteLn('Creating worker threads ...');
for i:= 1 to 4 do fReaderArr[i]:= TThreadReader.Create( fThreadQueue,Nil);
{- Calling with fSync works ! }
//for i:= 1 to 4 do fReaderArr[i]:= TThreadReader.Create( fThreadQueue,fSync);
WriteLn('Init done. Pushing items ...');
for i:= 1 to 100 do fThreadQueue.PushItem( TThreadTaskMsg.Create( i,''));
ReadLn;
finally
for i:= 1 to 4 do fReaderArr[i].Free;
fThreadQueue.Free;
fSync.Free;
end;
except
on E: Exception do
begin
Writeln(E.ClassName, ': ', E.Message);
ReadLn;
end;
end;
end.
Обновить. Ошибка в TMonitor, которая вызвала сбой TThreadedQueue, исправлена в Delphi XE2.
Обновление 2. Вышеупомянутый тест подчеркнул очередь в пустом состоянии. Дарьян Миллер обнаружил, что подчеркивание очереди в полном состоянии, все же может воспроизвести ошибку в XE2. Ошибка снова находится в TMonitor. См. Его ответ ниже для получения дополнительной информации. А также ссылку на QC101114.
Обновление 3:
С обновлением Delphi-XE2 4 было объявлено исправление для TMonitor
, которое устранило бы проблемы в TThreadedQueue
. Мои тесты до сих пор больше не могут воспроизвести какие-либо ошибки в TThreadedQueue
.
Протестированные потоки одного производителя/нескольких потребителей, когда очередь пуста и заполнена.
Также проверено несколько производителей/несколько потребителей. Я менял нити считывателя и писательские потоки от 1 до 100 без каких-либо сбоев. Но, зная историю, я осмеливаюсь, чтобы другие нарушили TMonitor
.