使用 WaitForMultipleObjects 等待多个线程答案

【问题标题】：Waiting for multiples threads using WaitForMultipleObjects使用 WaitForMultipleObjects 等待多个线程
【发布时间】：2011-10-15 13:50:43
【问题描述】：

我正在使用WaitForMultipleObjects 函数等待几个线程的完成，但我做错了，因为结果不是预期的

查看此示例代码

type
  TForm1 = class(TForm)
    Memo1: TMemo;
    Button1: TButton;
    procedure Button1Click(Sender: TObject);
  private
  end;

  TFoo = class(TThread)
  private
    Factor: Double;
    procedure ShowData;
  protected
    procedure Execute; override;
    constructor Create(AFactor : Double);
  end;


var
  Form1: TForm1;

implementation

Uses
 Math;

{$R *.dfm}

{ TFoo }

constructor TFoo.Create(AFactor: Double);
begin
  inherited Create(False);
  Factor := AFactor;
  FreeOnTerminate := True;

end;

procedure TFoo.Execute;
const
  Max=100000000;
var
  i : Integer;
begin
  inherited;
  for i:=1 to Max do
    Factor:=Sqrt(Factor);

  Synchronize(ShowData);
end;

procedure TFoo.ShowData;
begin
  Form1.Memo1.Lines.Add(FloatToStr(Factor));
end;

procedure TForm1.Button1Click(Sender: TObject);
const
 nThreads=5;
Var
 tArr  : Array[1..nThreads]  of TFoo;
 hArr  : Array[1..nThreads]  of THandle;
 i     : Integer;
 rWait : Cardinal;
begin
  for i:=1  to nThreads do
   begin
     tArr[i]:=TFoo.Create(Pi*i);
     hArr[i]:=tArr[i].Handle;
   end;

  repeat
    rWait:= WaitForMultipleObjects(nThreads, @hArr, True, 100);
    Application.ProcessMessages;
  until rWait<>WAIT_TIMEOUT;
  //here I want to show this message when all the threads are terminated    
  Memo1.Lines.Add('Wait done');
end;

end.

这是演示应用的当前输出

1
Wait done
1
1
1
1

但我想要这样的东西

1
1
1
1
1
Wait done

我必须如何使用WaitForMultipleObjects 函数等待所有线程终止？

【问题讨论】：

标签： multithreading delphi delphi-xe

【解决方案1】：

不要像最后一个参数那样传递这么短的超时时间。

根据MSDN

dw毫秒 [in] 超时间隔，以毫秒为单位。如果间隔已过，即使 bWaitAll 参数指定的条件不满足，该函数也会返回。如果 dwMilliseconds 为零，则该函数测试指定对象的状态并立即返回。如果 dwMilliseconds 为 INFINITE，则函数的超时间隔永远不会过去。

特别注意第二句。您告诉它等待所有句柄，但在 100 毫秒后超时。所以将 INFINITE 作为最后一个参数传递，并使用WAIT_OBJECT_0 而不是WAIT_TIMEOUT 作为退出测试。

【讨论】：

我认为他想继续定期调用 Application.processmessages，因此超时时间很短。
@Warren：调用ProcessMessages 破坏了使用线程的全部目的，并且使用它应该成为终止的理由。 :) 线程本身可以发布消息以更新其状态，并且他们已经在调用Synchronize。
@Ken：但是如果主线程在WaitForMultipleObjects(INFINITE) 调用上被阻塞，它永远不会将堆栈展开回 TApplication 中的消息泵，以便从线程中检索这些消息，并且应用程序将死锁。唯一可行的解决方案是将整个事情移出主线程。
我会将整个循环移动到单独的线程（并删除我个人的噩梦Application.ProcessMessages），并在所有子线程完成后通知主线程（例如再次通过同步）。在单独的线程中，您可以使用WaitForMultipleObjects(INFINITE) 平静地等待
或者只是避免阻塞轮询，并在最后一个线程终止时向主窗体发送一条消息。

【解决方案2】：

修复：删除 FreeOnTerminate。

当您仍然需要句柄时，您的代码会导致线程被释放。这是一个大错误，您可能会在代码中的其他地方遇到访问冲突，或者从您的 WaitFormMultipleObjects 返回错误返回代码。

当 TThread 被释放时，TThread.handle 变得无效，这会提前终止你的等待循环，因为句柄不再有效。如果您在后台释放 TThread 后尝试访问它，您也可能会遇到访问访问冲突，因此我认为最好在已知时间有意释放它们。

将线程句柄用作事件句柄可以正常工作，但您不应在线程终止时使用 FreeOnTerminate 来释放线程，因为这会过早破坏句柄。

我也同意那些说使用 Application.Processmessages 执行忙等待循环的人的观点非常丑陋。还有其他方法可以做到这一点。

unit threadUnit2;

interface

uses Classes, SyncObjs,Windows, SysUtils;

type
  TFoo = class(TThread)
  private
    FFactor: Double;
    procedure ShowData;
  protected
    procedure Execute; override;
    constructor Create(AFactor : Double);
    destructor Destroy; override;
  end;

  procedure WaitForThreads;


implementation

Uses
 Forms,
 Math;

procedure Trace(msg:String);
begin
  if Assigned(Form1) then
    Form1.Memo1.Lines.Add(msg);
end;



{ TFoo }

constructor TFoo.Create(AFactor: Double);
begin
  inherited Create(False);
  FFactor := AFactor;
//  FreeOnTerminate := True;

end;

destructor TFoo.Destroy;
begin
  inherited;
end;

procedure TFoo.Execute;
const
  Max=100000000;
var
  i : Integer;
begin
  inherited;
  for i:=1 to Max do
    FFactor:=Sqrt(FFactor);


  Synchronize(ShowData);
end;


procedure TFoo.ShowData;
begin

  Trace(FloatToStr(FFactor));
end;

procedure WaitForThreads;
const
 nThreads=5;
Var
 tArr  : Array[1..nThreads]  of TFoo;
 hArr  : Array[1..nThreads]  of THandle;
 i     : Integer;
 rWait : Cardinal;
begin
  for i:=1  to nThreads do
   begin
     tArr[i]:=TFoo.Create(Pi*i);
     hArr[i]:=tArr[i].handle; // Event.Handle;
   end;

  repeat
    rWait:= WaitForMultipleObjects(nThreads, @hArr[1],{waitAll} True, 150);
    Application.ProcessMessages;
  until rWait<>WAIT_TIMEOUT;
  Sleep(0);
  //here I want to show this message when all the threads are terminated
  Trace('Wait done');

  for i:=1  to nThreads do
   begin
     tArr[i].Free;
   end;

end;

end.

【讨论】：

1st，WaitForMultipleObjecs 中没有访问冲突之类的东西，它要么失败并返回 WAIT_FAILED，要么成功并返回其他值之一。在任何情况下，它都不会引发 AV。第二，错误不是在需要句柄时释放线程。错误在于使用无效句柄。 OP 可以选择保留不再需要的线程，或者避免将无效句柄传递给 WaitForMultipleObjects。
谁说句柄因为其他原因而无效？谁说释放线程后可以保持句柄有效？我要说的是，与其试图保持句柄有效，一种选择是，不要将无效的线程句柄传递给 WaitForMultipleObjects。无论如何，-1 表示没有纠正这种“WaitForMultipleObjects 中的 AV”谬误。
我不想拖太多，但我脑海中的谬论仍然被答案第一段中提到的神秘“WaitForMultipleObjects中的AVs”所困扰.即使 OP 没有缓存线程句柄，而是访问已释放线程上的句柄属性，一旦线程实例指针指向有效的内存位置（堆栈/堆上的变量），他将不会获得访问句柄的 AV，因为句柄属性只是读取一个静态字段。而且 API 本身也不会引发 AV。
好吧，好吧，我把那个复杂的句子弄得更复杂了。你还快乐吗？

【解决方案3】：

在您忽略的重复循环中有一个条件满足您的“直到”条件，WAIT_FAILED：

until rWait<>WAIT_TIMEOUT; 
Memo1.Lines.Add('Wait done');

由于您的超时时间有点紧，一个（或多个）线程完成并释放自身，呈现一个（或多个）句柄对于下一个 WaitForMultipleObjects 无效，这导致它返回“WAIT_FAILED”，从而导致“等待”显示“完成”消息。

对于重复循环中的每次迭代，您应该从hArr 中删除已完成线程的句柄。然后在任何情况下都不要忘记测试“WAIT_FAILED”。

编辑：
下面是一些示例代码，展示了如何做到这一点。这种方法而不是保持线程活动的不同之处在于，它不会留下未使用的内核和 RTL 对象。这对于手头的样本来说无关紧要，但对于许多从事冗长业务的线程来说，它可能是首选。

在代码中，WaitForMultipleObjects 被调用并为 'bWaitAll' 参数传递 'false' 以便能够删除线程句柄，而无需使用额外的 API 调用来确定它是否无效。但它允许其他情况，因为代码还必须能够处理在等待调用之外完成的线程。

procedure TForm1.Button1Click(Sender: TObject);

const
  nThreads=5;

Var
  tArr  : Array[1..nThreads]  of TFoo;
  hArr  : Array[1..nThreads]  of THandle;
  i     : Integer;
  rWait : Cardinal;

  hCount: Integer;  // total number of supposedly running threads
  Flags: DWORD;     // dummy variable used in a call to find out if a thread handle is valid

  procedure RemoveHandle(Index: Integer); // Decrement valid handle count and leave invalid handle out of range
  begin
    if Index <> hCount then
      hArr[Index] := hArr[hCount];
    Dec(hCount);
  end;

begin
  Memo1.Clear;

  for i:=1  to nThreads do
   begin
     tArr[i]:=TFoo.Create(Pi*i);
     hArr[i]:=tArr[i].Handle;
   end;
   hCount := nThreads;

  repeat
    rWait:= WaitForMultipleObjects(hCount, @hArr, False, 100);

    case rWait of

      // one of the threads satisfied the wait, remove its handle
      WAIT_OBJECT_0..WAIT_OBJECT_0 + nThreads - 1: RemoveHandle(rWait + 1);

      // at least one handle has become invalid outside the wait call, 
      // or more than one thread finished during the previous wait,
      // find and remove them
      WAIT_FAILED:
        begin
          if GetLastError = ERROR_INVALID_HANDLE then
          begin
            for i := hCount downto 1 do 
              if not GetHandleInformation(hArr[i], Flags) then // is handle valid?
                RemoveHandle(i);
          end
          else
            // the wait failed because of something other than an invalid handle
            RaiseLastOSError;
        end;

      // all remaining threads continue running, process messages and loop.
      // don't process messages if the wait returned WAIT_FAILED since we didn't wait at all
      // likewise WAIT_OBJECT_... may return soon
      WAIT_TIMEOUT: Application.ProcessMessages; 
    end;

  until hCount = 0;  // no more valid thread handles, we're done

  Memo1.Lines.Add('Wait done');
end;

请注意，这是为了回答被问到的问题。我宁愿使用 TThreads 的OnTerminate 事件来减少计数器并在达到“0”时输出“等待完成”消息。这或者正如其他人所建议的那样，将等待转移到它自己的线程中，会更容易并且可能更干净，并且可以避免Application.ProcessMessages 的需要。

【讨论】：

【解决方案4】：

我在例程的末尾添加了以下几行：

memo1.Lines.add(intToHex(rWait, 2));
if rWait = $FFFFFFFF then
  RaiseLastOSError;

原来WaitForMultipleObjects 因访问被拒绝错误而失败，这很可能是因为一些但不是所有线程正在完成并在迭代之间自行清理。

这里有一个棘手的问题。您需要保持消息泵运行，否则 Synchronize 调用将不起作用，因此您无法像 Ken 建议的那样通过 INFINITE。但是如果你做你现在正在做的事情，你就会遇到这个问题。

解决方案是将WaitForMultipleObjects 调用及其周围的代码也移动到它自己的线程中。它应该等待 INFINITE，然后当它完成时，它应该以某种方式向 UI 线程发出信号，让它知道它已经完成。（例如，当您单击按钮时，禁用该按钮，然后当监控线程完成时，它会再次启用该按钮。）

【讨论】：

【解决方案5】：

当你等待并且涉及到消息时，你必须使用 MsgWait... 并指定处理预期消息的掩码

repeat
    rWait:= MsgWaitForMultipleObjects(nThreads, @hArr[1], True, INFINITE, QS_ALLEVENTS);
    Application.ProcessMessages;
 until (rWait<>WAIT_TIMEOUT) and (rWait <> (WAIT_OBJECT_0 + nThreads));

n个线程

【讨论】：

我坚决支持MsgWaitForMultipleObjects，但是有两个问题。首先是语法，您需要将hArr[1] 而不是@hArr[1] 传递给Delphi API 翻译。其次，使用 Synchronize 结合MsgWaitForMultipleObjects 和INFINITE 超时死锁。我自己，我只是 Queue() ShowData 事件。

【解决方案6】：

如果您真的想了解多线程的工作原理，那么您就走在了正确的道路上 - 通过代码学习并提出问题，就像您在此处所做的那样。但是，如果您只想在应用程序中使用多线程，则可以使用OmniThreadLibrary 以更简单的方式完成，前提是您至少使用 Delphi 2009。

uses
  Math,
  OtlTask,
  OtlParallel;

function Calculate(factor: real): real;
const
  Max = 100000000;
var
  i: integer;
begin
  Result := factor;
  for i := 1 to Max do
    Result := Sqrt(Result);
end;

procedure TForm35.btnClick(Sender: TObject);
const
  nThreads = 5;
begin
  Parallel.ForEach(1, nThreads).Execute(
    procedure (const task: IOmniTask; const value: integer)
    var
      res: real;
    begin
      res := Calculate(Pi*value);
      task.Invoke(
        procedure begin
          Form35.Memo1.Lines.Add(FloatToStr(res));
        end
      );
    end
  );
  Memo1.Lines.Add('All done');
end;

【讨论】：

一个很好的观点。重新发明轮子是一种很好的学习方式，但是在生产代码中，您应该使用经过时间测试和单元测试的轮子，而 OmniThreadLibrary 是我对线程库的想法，它提供了足够的设施来完成常见的工作（比如制作工作线程) 需要比 Delphi 程序员（由于 VCL 导致精神变软和松弛）需要编写更多的代码，才能安全地使用线程。因此，Omni 线程库是一件很棒的事情。

【解决方案7】：

这就是正在发生的事情。

您的代码正在从 WaitForMultipleObjects 返回 WAIT_FAILED。
调用GetLastError 导致错误代码6，句柄无效。
您传递给WaitForMultipleObjects 的唯一句柄是线程句柄，因此其中一个线程句柄无效。
线程句柄之一变为无效的唯一方法是它已被关闭。
正如其他人所指出的，您正在通过设置 FreeOnTerminate 来关闭句柄。

故事的寓意是从所有函数中正确检查您的返回值，并让GetLastError 引导您找到问题的根本原因。

【讨论】：

【解决方案8】：

我不能利用这个机会来创建一个启动几个线程并使用消息传递将结果报告回 GUI 的工作示例。

将要启动的线程声明为：

type
  TWorker = class(TThread)
  private
    FFactor: Double;
    FResult: Double;
    FReportTo: THandle;
  protected
    procedure Execute; override;
  public
    constructor Create(const aFactor: Double; const aReportTo: THandle);

    property Factor: Double read FFactor;
    property Result: Double read FResult;
  end;

构造函数只是设置私有成员并将 FreeOnTerminate 设置为 False。这是必不可少的，因为它将允许主线程查询实例以获取结果。 execute 方法进行计算，然后发布一条消息到它在其构造函数中接收到的句柄，说它已完成。

procedure TWorker.Execute;
const
  Max = 100000000;
var
  i : Integer;
begin
  inherited;

  FResult := FFactor;
  for i := 1 to Max do
    FResult := Sqrt(FResult);

  PostMessage(FReportTo, UM_WORKERDONE, Self.Handle, 0);
end;

自定义 UM_WORKERDONE 消息的声明声明为：

const
  UM_WORKERDONE = WM_USER + 1;

type
  TUMWorkerDone = packed record
    Msg: Cardinal;
    ThreadHandle: Integer;
    unused: Integer;
    Result: LRESULT;
  end;

启动线程的表单在其声明中添加了以下内容：

  private
    FRunning: Boolean;
    FThreads: array of record
      Instance: TThread;
      Handle: THandle;
    end;
    procedure StartThreads(const aNumber: Integer);
    procedure HandleThreadResult(var Message: TUMWorkerDone); message UM_WORKERDONE;

FRunning 用于防止在工作进行时单击按钮。 FThreads 用于保存实例指针和创建线程的句柄。

启动线程的过程有一个非常简单的实现：

procedure TForm1.StartThreads(const aNumber: Integer);
var
  i: Integer;
begin
  Memo1.Lines.Add(Format('Starting %d worker threads', [aNumber]));
  SetLength(FThreads, aNumber);
  for i := 0 to aNumber - 1 do
  begin
    FThreads[i].Instance := TWorker.Create(pi * (i+1), Self.Handle);
    FThreads[i].Handle := FThreads[i].Instance.Handle;
  end;
end;

乐趣在于 HandleThreadResult 的实现：

procedure TForm1.HandleThreadResult(var Message: TUMWorkerDone);
var
  i: Integer;
  ThreadIdx: Integer;
  Thread: TWorker;
  Done: Boolean;
begin
  // Find thread in array
  ThreadIdx := -1;
  for i := Low(FThreads) to High(FThreads) do
    if FThreads[i].Handle = Cardinal(Message.ThreadHandle) then
    begin
      ThreadIdx := i;
      Break;
    end;

  // Report results and free the thread, nilling its pointer so we can detect
  // when all threads are done.
  if ThreadIdx > -1 then
  begin
    Thread := TWorker(FThreads[i].Instance);
    Memo1.Lines.Add(Format('Thread %d returned %f', [ThreadIdx, Thread.Result]));
    FreeAndNil(FThreads[i].Instance);
  end;

  // See whether all threads have finished.
  Done := True;
  for i := Low(FThreads) to High(FThreads) do
    if Assigned(FThreads[i].Instance) then
    begin
      Done := False;
      Break;
    end;
  if Done then
    Memo1.Lines.Add('Work done');
end;

享受...

【讨论】：

这并不相关，因为您没有使用等待。
+1 表示无忙等待或死锁，或 application.processmessage 调用。但是，您仍然在循环遍历所有对象并检查是否分配了对象；你真的确定没有比赛条件吗？我认为应该可以使用更简单的方法，使用 OmniThreadLibrary，所以我不建议任何人在生产代码中使用这种方法。
@Warren：是的，我正在遍历数组，但唯一修改它的代码是 StartThreads，它在每个指针/句柄启动之前分配指针/句柄，而 HandleThreadResults 则释放并清除它们。两者都在主线程中运行......所以我看不出任何竞争条件是如何出现的。
@David：你是对的，它没有回答如何使用 WaitForMultipleObjects。但是，我从更广泛的意义上理解了 OP 的问题：如何等待所有线程完成并只有在所有线程实际完成后才“等待完成”。

【解决方案9】：

您可以重构代码以只等待一个对象而不是多个对象。

我想向您介绍一个小帮手，它通常会在这种情况下帮助我。这次他的名字是IFooMonitor：

IFooMonitor = interface
  function WaitForAll(ATimeOut: Cardinal): Boolean;
  procedure ImDone;
end;

TFoo 和 IFooMonitor 将成为朋友：

TFoo = class(TThread)
strict private
  FFactor: Double;
  FMonitor: IFooMonitor;
  procedure ShowData;
protected
  procedure Execute; override;
public
  constructor Create(const AMonitor: IFooMonitor; AFactor: Double);
end;

constructor TFoo.Create(const ACountDown: ICountDown; AFactor: Double);
begin
  FCountDown := ACountDown;
  FFactor := AFactor;
  FreeOnTerminate := True;
  inherited Create(False);// <- call inherited constructor at the end!
end;

当 TFoo 完成他的工作时，它会告诉他的新朋友：

procedure TFoo.Execute;
const
  Max = 100000000;
var
  i: Integer;
begin
  for i := 1 to Max do
    FFactor := Sqrt(FFactor);

  Synchronize(ShowData);

  FMonitor.ImDone(); 
end;

现在我们可以将事件处理程序重构为如下所示：

procedure TForm1.Button1Click(Sender: TObject);
const
  nThreads = 5;
var
 i: Integer;
 monitor: IFooMonitor;
begin
  monitor := TFooMonitor.Create(nThreads); // see below for the implementation.

  for i := 1 to nThreads do
    TFoo.Create(monitor, Pi*i);

  while not monitor.WaitForAll(100) do
    Application.ProcessMessages;

  Memo1.Lines.Add('Wait done');
end;

这就是我们可以实现 IFooMonitor 的方式：

uses
  SyncObjs;

TFooMonitor = class(TInterfacedObject, IFooMonitor)
strict private
  FCounter: Integer;
  FEvent: TEvent;
  FLock: TCriticalSection;
private
  { IFooMonitor }
  function WaitForAll(ATimeOut: Cardinal): Boolean;
  procedure ImDone;
public
  constructor Create(ACount: Integer);
  destructor Destroy; override;   
end;

constructor TFooMonitor.Create(ACount: Integer);
begin
  inherited Create;
  FCounter := ACount;
  FEvent := TEvent.Create(nil, False, False, '');
  FLock := TCriticalSection.Create;
end;

procedure TFooMonitor.ImDone;
begin
  FLock.Enter;
  try
    Assert(FCounter > 0);
    Dec(FCounter);
    if FCounter = 0 then
      FEvent.SetEvent;
  finally
    FLock.Leave
  end;
end;

destructor TFooMonitor.Destroy;
begin
  FLock.Free;
  FEvent.Free;
  inherited;
end;

function TFooMonitor.WaitForAll(ATimeOut: Cardinal): Boolean;
begin
  Result := FEvent.WaitFor(ATimeOut) = wrSignaled 
end;

【讨论】：