multithreading - Delphi中两个线程相互同步的最佳方法

Question

我目前正在尝试找到最好的(*)方法来让两个线程交替运行并让它们互相等待。

(*) 速度快且 CPU 成本低的最佳组合

到目前为止，我发现了三种方法，我将它们放在一些演示应用程序中来展示我发现的问题。

由于所有的锁定，使用遵循经典等待/脉冲模式的 TMonitor 的性能不是很好(根据 SamplingProfiler 在这些函数中大部分时间都是烧毁的)。我使用 Windows 事件 (SyncObjs.TEvent) 尝试了相同的操作，但它的表现类似(即不好)。

使用调用 TThread.Yield 的等待循环性能最佳，但显然会疯狂地消耗 CPU 周期。如果切换发生得非常快，这并不重要，但当线程实际上正在等待时，这会很麻烦(您可以在演示中看到这一点)。

使用 TSpinWait 效果很好(如果不是这三个中最好的)，但前提是切换发生得非常快。由于 TSpinWait 的工作原理，切换时间越长，性能就越差。

由于多线程不是我的优势之一，我想知道是否有这些方法的某种组合或一些完全不同的方法来在两种情况下(快速和慢速切换)实现良好的性能。

program PingPongThreads;

{$APPTYPE CONSOLE}

{$R *.res}

uses
  Classes,
  Diagnostics,
  SyncObjs,
  SysUtils;

type
  TPingPongThread = class(TThread)
  private
    fCount: Integer;
  protected
    procedure Execute; override;
    procedure Pong; virtual;
  public
    procedure Ping; virtual;
    property Count: Integer read fCount;
  end;

  TPingPongThreadClass = class of TPingPongThread;

  TMonitorThread = class(TPingPongThread)
  protected
    procedure Pong; override;
    procedure TerminatedSet; override;
  public
    procedure Ping; override;
  end;

  TYieldThread = class(TPingPongThread)
  private
    fState: Integer;
  protected
    procedure Pong; override;
  public
    procedure Ping; override;
  end;

  TSpinWaitThread = class(TPingPongThread)
  private
    fState: Integer;
  protected
    procedure Pong; override;
  public
    procedure Ping; override;
  end;

{ TPingPongThread }

procedure TPingPongThread.Execute;
begin
  while not Terminated do
    Pong;
end;

procedure TPingPongThread.Ping;
begin
  TInterlocked.Increment(fCount);
end;

procedure TPingPongThread.Pong;
begin
  TInterlocked.Increment(fCount);
end;

{ TMonitorThread }

procedure TMonitorThread.Ping;
begin
  inherited;
  TMonitor.Enter(Self);
  try
    if Suspended then
      Start
    else
      TMonitor.Pulse(Self);
    TMonitor.Wait(Self, INFINITE);
  finally
    TMonitor.Exit(Self);
  end;
end;

procedure TMonitorThread.Pong;
begin
  inherited;
  TMonitor.Enter(Self);
  try
    TMonitor.Pulse(Self);
    if not Terminated then
      TMonitor.Wait(Self, INFINITE);
  finally
    TMonitor.Exit(Self);
  end;
end;

procedure TMonitorThread.TerminatedSet;
begin
  TMonitor.Enter(Self);
  try
    TMonitor.Pulse(Self);
  finally
    TMonitor.Exit(Self);
  end;
end;

{ TYieldThread }

procedure TYieldThread.Ping;
begin
  inherited;
  if Suspended then
    Start
  else
    fState := 3;
  while TInterlocked.CompareExchange(fState, 2, 1) <> 1 do
    TThread.Yield;
end;

procedure TYieldThread.Pong;
begin
  inherited;
  fState := 1;
  while TInterlocked.CompareExchange(fState, 0, 3) <> 3 do
    if Terminated then
      Abort
    else
      TThread.Yield;
end;

{ TSpinWaitThread }

procedure TSpinWaitThread.Ping;
var
  w: TSpinWait;
begin
  inherited;
  if Suspended then
    Start
  else
    fState := 3;
  w.Reset;
  while TInterlocked.CompareExchange(fState, 2, 1) <> 1 do
    w.SpinCycle;
end;

procedure TSpinWaitThread.Pong;
var
  w: TSpinWait;
begin
  inherited;
  fState := 1;
  w.Reset;
  while TInterlocked.CompareExchange(fState, 0, 3) <> 3 do
    if Terminated then
      Abort
    else
      w.SpinCycle;
end;

procedure TestPingPongThread(threadClass: TPingPongThreadClass; quickSwitch: Boolean);
const
  MAXCOUNT = 10000;
var
  t: TPingPongThread;
  i: Integer;
  sw: TStopwatch;
  w: TSpinWait;
begin
  t := threadClass.Create(True);
  try
    for i := 1 to MAXCOUNT do
    begin
      t.Ping;

      if not quickSwitch then
      begin
        // simulate some work
        w.Reset;
        while w.Count < 20 do
          w.SpinCycle;
      end;

      if i = 1 then
      begin
        if not quickSwitch then
        begin
          Writeln('Check CPU usage. Press <Enter> to continue');
          Readln;
        end;
        sw := TStopwatch.StartNew;
      end;
    end;
    Writeln(threadClass.ClassName, ' quick switches: ', quickSwitch);
    Writeln('Duration: ', sw.ElapsedMilliseconds, ' ms');
    Writeln('Call count: ', t.Count);
    Writeln;
  finally
    t.Free;
  end;
end;

procedure Main;
begin
  TestPingPongThread(TMonitorThread, False);
  TestPingPongThread(TYieldThread, False);
  TestPingPongThread(TSpinWaitThread, False);

  TestPingPongThread(TMonitorThread, True);
  TestPingPongThread(TYieldThread, True);
  TestPingPongThread(TSpinWaitThread, True);
end;

begin
  try
    Main;
  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
  Writeln('Press <Enter> to exit');
  Readln;
end.

更新:

我想出了一个事件和 spinwait 的组合:

constructor TSpinEvent.Create;
begin
  inherited Create(nil, False, False, '');
end;

procedure TSpinEvent.SetEvent;
begin
  fState := 1;
  inherited;
end;

procedure TSpinEvent.WaitFor;
var
  startCount: Cardinal;
begin
  startCount := TThread.GetTickCount;
  while TInterlocked.CompareExchange(fState, 0, 1) <> 1 do
  begin
    if (TThread.GetTickCount - startCount) >= YieldTimeout then // YieldTimeout = 10
      inherited WaitFor(INFINITE)
    else
      TThread.Yield;
  end;
end;

在进行快速切换时，其执行速度仅比基于光纤的实现慢大约 5 到 6 倍，而在 Ping 调用之间添加一些工作时，速度慢不到 1%。当使用光纤时，它当然运行在 2 个核心上，而不是只有一个。

Answer 1

当我发现自己处于这种情况时，我喜欢使用 Windows 事件。 它们在 Delphi 中使用 TEvent 类公开，您可以使用 WaitForSingleObject。

因此，您可以使用两个事件:Thread1NotActive 和 Thread2NotActive。 一旦 Thread1 完成，它就会设置 Thread1NotActive 标志，该标志由 Thread2 等待。 相反，如果 Thread2 停止处理，则会设置 Thread2NotActive，该状态由 Thread1 监视。

这应该可以让您避免竞争条件(这就是为什么我建议使用两个事件而不是 1 个)，并且应该让您在此过程中保持理智，同时不会消耗过多的 CPU 时间。

如果您需要更完整的示例，请明天等待:)