Delphi TMonitor.Wait 多线程问题

Question

我们在后端服务中遇到了这个多线程问题：

在具有 30 多个线程的多线程 Windows 服务应用程序中，SysUtils.EventCache 出现问题。问题是 NewWaitObj 函数有时会返回 NIL 而不是 Event 对象。此功能用于 TMonitor 同步方法 Wait。 TMonitor.Wait 在事件对象获得 NIL 时停止工作。这会影响许多 VCL 和 RTL 线程同步源代码，并在多线程应用程序中导致不同的侧面问题，例如 TThreadedQueue.PopItem 不等待新项目到达队列并立即返回超时结果。

NewWaitObj函数出现问题：

function NewWaitObj: Pointer;
var
  EventItem: PEventItemHolder;
begin
  EventItem := Pop(EventCache);
  if EventItem <> nil then
  begin
    Result := EventItem.Event;
    EventItem.Event := nil;
    Push(EventItemHolders, EventItem);
  end else
    Result := NewSyncWaitObj;
  ResetSyncWaitObj(Result);
end;

看起来 Pop 函数在繁重的多线程应用程序中没有得到很好的保护，并且在一定数量的并发线程中它开始将一个和相同的 EventItem 实例返回到两个（或更多）线程。然后在 NewWaitObj 中发生竞争条件：

1. 一个线程获取 EventItem.Event 并将其作为 Result 返回并用 NIL 将其归零，竞赛并行线程得到相同的 EventItem.Event 但它已经被第一个线程清除了。
2. 这会导致其中一个竞赛线程返回有效的事件句柄，而其他竞赛线程返回 NIL。
3. TMonitor.Wait 函数不起作用，因为它将 NIL 作为事件句柄。
4. TThreadedQueue.PopItem 不等待，其他同步方法也无法正常工作。

由于某些原因，当应用程序有许多并发线程时，Pop 方法中的线程同步不起作用：

function Pop(var Stack: PEventItemHolder): PEventItemHolder;
begin
  repeat
    Result := Stack;
    if Result = nil then
      Exit;
  until AtomicCmpExchange(Pointer(Stack), Result.Next, Result) = Result;
end;

在 60 个测试线程的测试应用程序中，问题在大约 10-20 秒内出现，30 个线程更难发生，通常需要 5-10 分钟。一旦出现问题 - 它永远不会停止，直到重新启动应用程序。在线程同步被破坏后的测试应用程序中 - EventCache 的每 5 个操作中大约有一个返回 NIL。 AtomicCmpExchange 中似乎有问题，我检查了生成的代码 - 它只是一个 CMPXCHG 指令，还有几个用于设置寄存器的指令。我不太确定导致问题的原因 - 例如，一个线程可以在设置寄存器以调用 CMPXCHG 时或在调用之后处理结果时获得其他线程的干预吗？

试图了解导致问题的原因，以便找到最佳解决方法。现在我打算用我自己的替换原来的 NewWaitObj，它只会调用原来的版本，直到它返回有效的对象。这个问题在我们的开发、测试和生产环境中经常出现，对于生产服务器上的真正中间件服务，需要几个小时（有时几天）才能出现问题，之后只需重新启动即可解决问题。 测试应用可以从 Embarcadero JIRA 的 issue 中下载：https://quality.embarcadero.com/browse/RSP-31154

编辑：TestApp：https://quality.embarcadero.com/secure/attachment/31605/EventCacheBug.zip

Delphi 源代码示例：

unit FormMainEventCacheBugU;

interface

uses
  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes, Vcl.Graphics,
  Vcl.Controls, Vcl.Forms, Vcl.Dialogs, Math, Vcl.StdCtrls;

const
   MaxProducers = 60;

type
  TFormEventCacheBug = class(TForm)
    BtnMaxProducers: TButton;
    BtnRemoveProducer: TButton;
    BtnAddProducer: TButton;
    procedure BtnMaxProducersClick(Sender: TObject);
    procedure BtnRemoveProducerClick(Sender: TObject);
    procedure BtnAddProducerClick(Sender: TObject);
  private
    { Private declarations }
  public
    { Public declarations }
  end;

   TEventEater = class(TThread)
   private
      SleepTime: Integer;
      SMsg, EMsg, NMsg: PChar;
      procedure EatEvent;
   protected
      procedure Execute; override;
   public
      constructor Create;
   end;

var
  FormEventCacheBug: TFormEventCacheBug;
  Producers: array[1..MaxProducers] of TThread;
  ProdCount: Integer;

implementation

{$R *.dfm}

procedure AddProducer;
begin
   if ProdCount < MaxProducers then
   begin
      Inc(ProdCount);
      Producers[ProdCount] := TEventEater.Create;
      Producers[ProdCount].FreeOnTerminate := True;
   end;
end;

procedure RemoveProducer;
begin
   if ProdCount > 0 then
   begin
      Producers[ProdCount].Terminate;
      Dec(ProdCount);
   end;
end;

{ TEventEater }

constructor TEventEater.Create;
begin
   inherited Create(False);
   SleepTime := RandomRange(1, 3);
end;

procedure TEventEater.EatEvent;
var
   EventHandle: Pointer;
begin
   //OutputDebugString(SMsg);
   EventHandle := System.MonitorSupport.NewWaitObject;
   try
      if EventHandle = nil then
         OutputDebugString('NIL');
      Sleep(SleepTime);
   finally
      if EventHandle <> nil then
         System.MonitorSupport.FreeWaitObject(EventHandle);
//      OutputDebugString(EMsg);
   end;
end;

procedure TEventEater.Execute;
begin
   SMsg := PChar('S:' + GetCurrentThreadId.ToString);
   EMsg := PChar('E:' + GetCurrentThreadId.ToString);
   NMsg := PChar('NIL:' + GetCurrentThreadId.ToString);
   while not Terminated do
   begin
      EatEvent;
      Sleep(SleepTime);
   end;
end;

procedure TFormEventCacheBug.BtnAddProducerClick(Sender: TObject);
begin
   AddProducer;
end;

procedure TFormEventCacheBug.BtnRemoveProducerClick(Sender: TObject);
begin
   RemoveProducer;
end;

procedure TFormEventCacheBug.BtnMaxProducersClick(Sender: TObject);
var
   i: Integer;
begin
   for i := ProdCount + 1 to MaxProducers do
      AddProducer;
end;

end.

感谢您的任何想法，

Answer 1

@MiroslavPenchev，谢谢你的帖子！ 在 XE2 中工作并有类似的问题。 Delphi 10.4.1 使用带有计数器和 128 位比较交换的链表头解决了 TMonitor ABA 问题。 不幸的是，这对 XE2 来说不是一个简单的选择。

再次感谢您建议重写一些调用原始方法的 MonitorSupport 方法。

以下是我正在使用的解决方案。它不是 100% 完美的，因为涉及到锁定，但是对于较少并发的环境，它至少可以使系统稳定并且没有 100% 的 CPU 问题。

var
  MonitorSupportFix: TMonitorSupport;
  OldMonitorSupport: PMonitorSupport;
  NewWaitObjCS: TCriticalSection;

function NewWaitObjFix: Pointer;
begin
  if Assigned(NewWaitObjCS) then
    NewWaitObjCS.Enter;
  try
    Result := OldMonitorSupport.NewWaitObject;
  finally
    if Assigned(NewWaitObjCS) then
      NewWaitObjCS.Leave;
  end;
end;

procedure FreeWaitObjFix(WaitObject: Pointer);
begin
  if Assigned(NewWaitObjCS) then
    NewWaitObjCS.Enter;
  try
    OldMonitorSupport.FreeWaitObject(WaitObject);
  finally
    if Assigned(NewWaitObjCS) then
      NewWaitObjCS.Leave;
  end;
end;

procedure InitMonitorSupportFix;
begin
  OldMonitorSupport := System.MonitorSupport;
  MonitorSupportFix := OldMonitorSupport^;
  MonitorSupportFix.NewWaitObject := NewWaitObjFix;
  MonitorSupportFix.FreeWaitObject := FreeWaitObjFix;

  System.MonitorSupport := @MonitorSupportFix;
end;

initialization
  NewWaitObjCS := TCriticalSection.Create;
  InitMonitorSupportFix;
finalization
  FreeAndNil(NewWaitObjCS);
end.