TThreadedQueue (Generics.Collections) を単一のプロデューサーの複数のコンシューマー スキームで使用しようとしています。(Delphi-XE)。アイデアは、オブジェクトをキューにプッシュし、いくつかのワーカー スレッドがキューを排出できるようにすることです。
ただし、期待どおりには機能しません。2 つ以上のワーカー スレッドが 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.
更新: TThreadedQueue のクラッシュの原因となった TMonitor のエラーは、Delphi XE2 で修正されました。
更新 2 : 上記のテストでは、空の状態でキューにストレスがかかりました。Darian Miller さんは、満杯の状態でキューに負荷をかけても、XE2 でエラーを再現できることを発見しました。エラーは TMonitor に再び表示されます。詳細については、以下の彼の回答を参照してください。また、QC101114 へのリンクもあります。
アップデート 3 : Delphi-XE2 アップデート 4TMonitorでは、 の問題を解決する修正プログラムが発表されましたTThreadedQueue。これまでの私のテストでは、エラーを再現できなくなりましたTThreadedQueue。キューが空でいっぱいのときに、単一のプロデューサー/複数のコンシューマー スレッドをテストしました。複数のプロデューサー/複数のコンシューマーもテストしました。リーダー スレッドとライター スレッドを 1 から 100 まで変化させましたが、問題はありませんでした。しかし、歴史を知っているので、あえて他の人を壊しTMonitorます。