Delphi TMonitor。等待多线程问题

Question

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

在具有30个线程的多线程Windows服务应用程序中，SysUtils出现问题。事件缓存出现。问题是NewWaitObj函数有时返回NIL而不是Event对象。此函数用于TMonitor sync methods Wait。t监视器。Wait在事件对象为零时停止工作。这会影响许多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。事件并将其作为结果返回，然后将其归零为零，赛车并行线程将获得相同的EventItem。事件，但它已被第一个线程清除
2. 这会导致一个竞速线程返回有效的事件句柄，而另一个竞速线程返回NIL
3. t监视器。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替换原来的NewWaitObj，它只调用原来的版本，直到它返回有效的对象为止。这个问题在我们的开发、测试和生产环境中经常发生，对于生产服务器上的真实中间件服务，需要几个小时（有时几天）才能出现问题，之后只需重新启动即可修复问题。测试应用程序可从Embarcadero JIRA的发行版下载：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方法。

下面是我正在使用的解决方案。它并不是百分之百完美的，因为它涉及到锁定，但对于并发性较差的环境，它至少使系统稳定，并且没有百分之百的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.