如何修复TSparseArray <T>？

Question

由于未修复的错误System.Generics.Collections.TArray.Copy<T>(取决于已报告的错误System.CopyArray),有时会使用线程库引发异常.

方法中引发异常System.Threading.TSparseArray<T>.Add:

function TSparseArray<T>.Add(const Item: T): Integer;
var
  I: Integer;
  LArray, NewArray: TArray<T>;
begin
  ...
          TArray.Copy<T>(LArray, NewArray, I + 1); // <- Exception here
  ...
end;

好吧,这是预计的错误System.CopyArray.因此,在尝试修复此问题时,我的第一个想法是简单地复制数组:

// TArray.Copy<T>(LArray, NewArray, I + 1); // <- Exception here
for LIdx := Low( LArray ) to High( LArray ) do
  NewArray[LIdx] := LArray[LIdx];

奇迹般有效.但在那之后我想知道为什么需要数组副本:

LArray := FArray; // copy array reference from field
...
SetLength(NewArray, Length(LArray) * 2);
TArray.Copy<T>(LArray, NewArray, I + 1);
NewArray[I + 1] := Item;
Exit(I + 1);

元素被复制到NewArray(局部变量),就是这样.没有任何作业FArray,所以对我来说,NewArray超出范围时最终确定.

现在我有三个错误修正选择:

1. 只需更换 TArray.Copy

   SetLength(NewArray, Length(LArray) * 2);
   // TArray.Copy<T>(LArray, NewArray, I + 1); // <- Exception here
   for LIdx := Low( LArray ) to High( LArray ) do
     NewArray[LIdx] := LArray[LIdx];
   NewArray[I + 1] := Item;
   Exit(I + 1);
   

2. 替换TArray.Copy并保存NewArray

   SetLength(NewArray, Length(LArray) * 2);
   // TArray.Copy<T>(LArray, NewArray, I + 1); // <- Exception here
   for LIdx := Low( LArray ) to High( LArray ) do
     NewArray[LIdx] := LArray[LIdx];
   NewArray[I + 1] := Item;
   FArray := NewArray;
   Exit(I + 1);
   

3. 注释掉所有不必要的代码部分(因为它们只是在浪费时间)

   // SetLength(NewArray, Length(LArray) * 2);
   // TArray.Copy<T>(LArray, NewArray, I + 1); // <- Exception here
   // NewArray[I + 1] := Item;
   Exit(I + 1);
   

我检查了所有三个修复程序,其中包含一堆任务,查找未使用的工作线程或未执行的任务.但我没有找到任何一个.该库按预期工作(现在没有任何例外).

你能指点我在这里失踪的东西吗？

---

为了达到这个例外,你已经运行了一堆任务,让TTaskPool创建越来越多TWorkerQueueThreads.通过TaskManager检查线程数,并TArray.Copy在TSparseArray<T>.Add方法中使用断点.这里,当应用程序的线程数超过25个线程时,我得到此异常.

// Hit the button very fast until the debugger stops 
// at TSparseArray<T>.Add method to copy the array
procedure TForm1.Button1Click( Sender : TObject );
var
  LIdx : Integer;
begin
  for LIdx := 1 to 20 do
    TTask.Run( 
      procedure 
      begin
        Sleep( 50 );
      end );
end;

Answer 1

项目是否写入并不重要TSparseArray<T>，因为只有当一个工作线程完成了委托给他的所有任务并且另一个工作线程尚未完成时才需要它。此时，空闲线程正在查看池内其他线程的队列并尝试窃取一些工作。

如果任何队列未进入该数组，则空闲线程不可见，因此无法共享工作负载。

为了解决这个问题，我选择选项 2

function TSparseArray<T>.Add(const Item: T): Integer;
...
SetLength(NewArray, Length(LArray) * 2);
TArray.Copy<T>(LArray, NewArray, I + 1); // <- No Exception here with XE7U1
NewArray[I + 1] := Item;
{$IFDEF USE_BUGFIX}
FArray := NewArray;
{$ENDIF}
Exit(I + 1);

但在没有任何锁定的情况下，窃取部分是有风险的

procedure TThreadPool.TQueueWorkerThread.Execute;

...

if Signaled then
begin
  I := 0;
  while I < Length(ThreadPool.FQueues.Current) do
  begin
    if (ThreadPool.FQueues.Current[I] <> nil) 
      and (ThreadPool.FQueues.Current[I] <> WorkQueue)
      and ThreadPool.FQueues.Current[I].TrySteal(Item) 
    then
      Break;
    Inc(I);
  end;
  if I <> Length(ThreadPool.FQueues.Current) then
    Break;
  LookedForSteals := True;
end

数组长度只会增长

while I < Length(ThreadPool.FQueues.Current) do

和

if I <> Length(ThreadPool.FQueues.Current) then

应该足够安全了。

if Signaled then
begin
  I := 0;
  while I < Length(ThreadPool.FQueues.Current) do
  begin
    {$IFDEF USE_BUGFIX}
    TMonitor.Enter(ThreadPool.FQueues);
    try
    {$ENDIF}
      if (ThreadPool.FQueues.Current[I] <> nil) and (ThreadPool.FQueues.Current[I] <> WorkQueue) and ThreadPool.FQueues.Current[I].TrySteal(Item) then
        Break;
    {$IFDEF USE_BUGFIX}
    finally
      TMonitor.Exit(ThreadPool.FQueues);
    end;
    {$ENDIF}
    Inc(I);
  end;
  if I <> Length(ThreadPool.FQueues.Current) then
    Break;
  LookedForSteals := True;
end

现在我们需要一个测试环境来观察窃取行为：

program WatchStealingTasks;

{$APPTYPE CONSOLE}
{$R *.res}

uses
  Winapi.Windows,
  System.SysUtils,
  System.Threading,
  System.Classes,
  System.Math;

procedure OutputDebugStr( const AStr: string ); overload;
begin
  OutputDebugString( PChar( AStr ) );
end;

procedure OutputDebugStr( const AFormat: string; const AParams: array of const ); overload;
begin
  OutputDebugStr( Format( AFormat, AParams ) );
end;

function CreateInnerTask( AThreadId: Cardinal; AValue: Integer; APool: TThreadPool ): ITask;
begin
  Result := TTask.Run(
      procedure
    begin
      Sleep( AValue );
      if AThreadId <> TThread.CurrentThread.ThreadID
      then
        OutputDebugStr( '[%d] executed stolen task from [%d]', [TThread.CurrentThread.ThreadID, AThreadId] )
      else
        OutputDebugStr( '[%d] executed task', [TThread.CurrentThread.ThreadID] );
    end, APool );
end;

function CreateTask( AValue: Integer; APool: TThreadPool ): ITask;
begin
  Result := TTask.Run(
    procedure
    var
      LIdx: Integer;
      LTasks: TArray<ITask>;
    begin
      // Create three inner tasks per task
      SetLength( LTasks, 3 );
      for LIdx := Low( LTasks ) to High( LTasks ) do
        begin
          LTasks[LIdx] := CreateInnerTask( TThread.CurrentThread.ThreadID, AValue, APool );
        end;
      OutputDebugStr( '[%d] waiting for tasks completion', [TThread.CurrentThread.ThreadID] );
      TTask.WaitForAll( LTasks );
      OutputDebugStr( '[%d] task finished', [TThread.CurrentThread.ThreadID] );
    end, APool );
end;

procedure Test;
var
  LPool: TThreadPool;
  LIdx: Integer;
  LTasks: TArray<ITask>;
begin
  OutputDebugStr( 'Test started' );
  try
    LPool := TThreadPool.Create;
    try
      // Create three tasks
      SetLength( LTasks, 3 );
      for LIdx := Low( LTasks ) to High( LTasks ) do
        begin
          // Let's put some heavy work (200ms) on the first tasks shoulder
          // and the other tasks just some light work (20ms) to do
          LTasks[LIdx] := CreateTask( IfThen( LIdx = 0, 200, 20 ), LPool );
        end;
      TTask.WaitForAll( LTasks );
    finally
      LPool.Free;
    end;
  finally
    OutputDebugStr( 'Test completed' );
  end;
end;

begin
  try
    Test;
  except
    on E: Exception do
      Writeln( E.ClassName, ': ', E.Message );
  end;
  ReadLn;

end.

调试日志是

Debug-Ausgabe：测试已启动 Prozess WatchStealingTasks.exe (4532)
线程启动：线程 ID：2104。Prozess WatchStealingTasks.exe (4532)
线程启动：线程 ID：2188。Prozess WatchStealingTasks.exe (4532)
线程启动：线程 ID：4948。Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2188] 等待任务完成 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2104] 等待任务完成 Prozess WatchStealingTasks.exe (4532)
线程启动：线程 ID：2212。Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[4948] 等待任务完成 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2188] 执行任务 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[4948] 执行任务 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2188] 执行任务 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[4948] 执行任务 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2188] 执行任务 Prozess WatchStealingTasks.exe (4532)
调试-Ausgabe：[2188] 任务已完成 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[4948] 执行任务 Prozess WatchStealingTasks.exe (4532)
调试-Ausgabe：[4948] 任务已完成 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2104] 执行任务 Prozess WatchStealingTasks.exe (4532)
Debug-Ausgabe：[2188] 执行了从 [2104] Prozess WatchStealingTasks.exe 窃取的任务 (4532)
Debug-Ausgabe：[4948] 执行了从 [2104] Prozess WatchStealingTasks.exe 窃取的任务 (4532)
调试-Ausgabe：[2104] 任务已完成 Prozess WatchStealingTasks.exe (4532)
调试-Ausgabe：线程退出：2188 Prozess WatchStealingTasks.exe (4532)
调试-Ausgabe：线程退出：4948 Prozess WatchStealingTasks.exe (4532)
线程结束：线程 ID：4948。Prozess WatchStealingTasks.exe (4532)
线程结束：线程 ID：2188。Prozess WatchStealingTasks.exe (4532)
线程结束：线程 ID：2212。Prozess WatchStealingTasks.exe (4532)

好的，窃取现在应该可以在任意数量的工作线程上进行，所以一切都好吗？

不

这个小测试应用程序不会结束，因为它现在冻结在线程池的析构函数中。最后一个工作线程不会终止，原因是

procedure TThreadPool.TQueueWorkerThread.Execute;

...

if ThreadPool.FWorkerThreadCount = 1 then
begin
  // it is the last thread after all tasks executed, but
  // FQueuedRequestCount is still on 7 - WTF
  if ThreadPool.FQueuedRequestCount = 0 then
  begin

这里还有一个需要修复的错误...因为当等待任务时，Task.WaitForAll您现在正在等待的所有任务都会在内部执行，但不会减少FQueuedRequestCount.

解决这个问题

function TThreadPool.TryRemoveWorkItem(const WorkerData: IThreadPoolWorkItem): Boolean;
begin
  Result := (QueueThread <> nil) and (QueueThread.WorkQueue <> nil);
  if Result then
    Result := QueueThread.WorkQueue.LocalFindAndRemove(WorkerData);
  {$IFDEF USE_BUGFIX}
  if Result then
    DecWorkRequestCount;
  {$ENDIF}
end;

现在它就像应该立即完成一样运行。

更新

作为 Uwe 的评论，我们还需要修复固定的System.Generics.Collections.TArray.Copy<T>

class procedure TArray.Copy<T>(const Source, Destination: array of T; SourceIndex, DestIndex, Count: NativeInt);
{$IFDEF USE_BUGFIX}
begin
  CheckArrays(Pointer(@Source[0]), Pointer(@Destination[0]), SourceIndex, Length(Source), DestIndex, Length(Destination), Count);
  if IsManagedType(T) then
    System.CopyArray(Pointer(@Destination[DestIndex]), Pointer(@Source[SourceIndex]), TypeInfo(T), Count)
  else
    System.Move(Pointer(@Source[SourceIndex])^,Pointer(@Destination[DestIndex])^, Count * SizeOf(T) );
end;
{$ELSE}
begin
  CheckArrays(Pointer(@Source[0]), Pointer(@Destination[0]), SourceIndex, Length(Source), DestIndex, Length(Destination), Count);
  if IsManagedType(T) then
    System.CopyArray(Pointer(@Destination[SourceIndex]), Pointer(@Source[SourceIndex]), TypeInfo(T), Count)
  else
    System.Move(Pointer(@Destination[SourceIndex])^, Pointer(@Source[SourceIndex])^, Count * SizeOf(T));
end;
{$ENDIF}

一个简单的检查来测试：

procedure TestArrayCopy;
var
  LArr1, LArr2: TArray<Integer>;
begin
  LArr1 := TArray<Integer>.Create( 10, 11, 12, 13 );
  LArr2 := TArray<Integer>.Create( 20, 21 );
  // copy the last 2 elements from LArr1 to LArr2
  TArray.Copy<Integer>( LArr1, LArr2, 2, 0, 2 );
end;

• 使用 XE7 你会得到一个例外
• 使用 XE7 Update1 你会得到

  LArr1 = ( 10, 11, 0, 0 )
  LArr2 = ( 20, 21 )
  

• 通过上面的修复将得到

  LArr1 = ( 10, 11, 12, 13 )
  LArr2 = ( 12, 13 )