无锁多读单写答案

【问题标题】：Lock free multiple readers single writer无锁多读单写
【发布时间】：2009-05-26 07:48:56
【问题描述】：

我有一个内存数据结构，它可以被多个线程读取，并且只能由一个线程写入。目前我正在使用一个关键部分来使这个访问线程安全。不幸的是，即使只有另一个读者正在访问它，这也会阻止读者。

有两种解决方法：

使用 TMultiReadExclusiveWriteSynchronizer
使用无锁方法消除任何阻塞

对于 2. 到目前为止，我得到了以下内容（任何无关紧要的代码都被省略了）：

type
  TDataManager = class
  private
    FAccessCount: integer;
    FData: TDataClass;
  public
    procedure Read(out _Some: integer; out _Data: double);
    procedure Write(_Some: integer; _Data: double);
  end;

procedure TDataManager.Read(out _Some: integer; out _Data: double);
var
  Data: TDAtaClass;
begin
  InterlockedIncrement(FAccessCount);
  try
    // make sure we get both values from the same TDataClass instance
    Data := FData;
    // read the actual data
    _Some := Data.Some;
    _Data := Data.Data;
  finally
    InterlockedDecrement(FAccessCount);
  end;
end;

procedure TDataManager.Write(_Some: integer; _Data: double);
var
  NewData: TDataClass;
  OldData: TDataClass;
  ReaderCount: integer;
begin
  NewData := TDataClass.Create(_Some, _Data);
  InterlockedIncrement(FAccessCount);
  OldData := TDataClass(InterlockedExchange(integer(FData), integer(NewData));
  // now FData points to the new instance but there might still be
  // readers that got the old one before we exchanged it.
  ReaderCount := InterlockedDecrement(FAccessCount);
  if ReaderCount = 0 then
    // no active readers, so we can safely free the old instance
    FreeAndNil(OldData)
  else begin
    /// here is the problem
  end;
end;

不幸的是，OldData 实例被替换后会出现一个小问题。如果 Read 方法中当前没有其他线程（ReaderCount=0），则可以安全地处理它，仅此而已。但如果不是这样，我该怎么办？我可以将它存储到下一次调用并在那里处理它，但是 Windows 调度理论上可以让读取器线程在 Read 方法中休眠并且仍然有对 OldData 的引用。

如果您发现上述代码有任何其他问题，请告诉我。这是要在多核的计算机上运行，并且上面的方法会被非常频繁地调用。

如果这很重要：我正在使用带有内置内存管理器的 Delphi 2007。我知道内存管理器在创建新类时可能会强制执行一些锁定，但我想暂时忽略它。

编辑：从上面可能不清楚：对于 TDataManager 对象的整个生命周期，只有一个线程写入数据，而不是几个可能竞争写访问的线程。所以这是 MREW 的一个特例。

【问题讨论】：

我对自己编写的无锁代码持谨慎态度，几乎不可能做到正确。至于 TMREWS：没有办法在典型机器上为您的用例计时，因为有不同的方法来实现它们，而 VCL 只为您提供一种。有关比较不同实现（包括时间）的文章，请参阅codeproject.com/KB/threads/testing_rwlocks.aspx

标签： multithreading delphi lock-free

【解决方案1】：

我不知道任何可以在 Intel86 代码上实现的无锁（或上面示例中的微锁）MREW 方法。

对于小型（快过期）锁，OmniThreadLibrary 的旋转方法可以正常工作：

type
TOmniMREW = record
strict private
  omrewReference: integer;      //Reference.Bit0 is 'writing in progress' flag
public
  procedure EnterReadLock; inline;
  procedure EnterWriteLock; inline;
  procedure ExitReadLock; inline;
  procedure ExitWriteLock; inline;
end; { TOmniMREW }

procedure TOmniMREW.EnterReadLock;
var
  currentReference: integer;
begin
  //Wait on writer to reset write flag so Reference.Bit0 must be 0 than increase Reference
  repeat
    currentReference := omrewReference AND NOT 1;
  until currentReference = InterlockedCompareExchange(omrewReference, currentReference + 2, currentReference);
end; { TOmniMREW.EnterReadLock }

procedure TOmniMREW.EnterWriteLock;
var
  currentReference: integer;
begin
  //Wait on writer to reset write flag so omrewReference.Bit0 must be 0 then set omrewReference.Bit0
  repeat
    currentReference := omrewReference AND NOT 1;
  until currentReference = InterlockedCompareExchange(omrewReference, currentReference + 1, currentReference);
  //Now wait on all readers
  repeat
  until omrewReference = 1;
end; { TOmniMREW.EnterWriteLock }

procedure TOmniMREW.ExitReadLock;
begin
  //Decrease omrewReference
  InterlockedExchangeAdd(omrewReference, -2);
end; { TOmniMREW.ExitReadLock }

procedure TOmniMREW.ExitWriteLock;
begin
  omrewReference := 0;
end; { TOmniMREW.ExitWriteLock }

我刚刚注意到这里可能存在对齐问题 - 代码应检查 omrewReference 是否为 4 对齐。会通知作者。

【讨论】：

如果我没记错的话，你会通知自己的 ;-) 顺便说一句，这是一个不错的图书馆。
@gabr：对于多核系统来说，这是一个非常好的工具箱，+1。不过，它确实与 Vista 引入的超薄 R/W 锁有一个共同点：访问不能从读取升级到写入。如果我正确阅读了这段代码，这样做会导致无限循环。也许值得为此添加注释。
@Davy：不，我不是作者，GJ 是 - 还写了无锁（或者更确切地说，微锁）堆栈和队列的人。
@DavyLandman，当我读到 gabr 说的时候，我也有同样的问题 :)
@mghie 你是对的，这把锁会永远旋转。对应的 Windows slim R/W 锁只会旋转 1024 次。之后它将回退到一个快速用户空间互斥体；无证的Keyed Event.

【解决方案2】：

只是一个补充 - 您在此处查看的内容通常称为 Hazard Pointers。我不知道你是否可以在 Delphi 中做类似的事情。

【讨论】：

【解决方案3】：

自从我接触 Delphi 以来已经有一段时间了，所以在使用之前验证这一点，但是......从内存中，如果你使用接口和使用 TInterfacedObject 的实现，你可以获得引用计数行为。

type
    IDataClass = interface
        function GetSome: integer;
        function GetData: double;

        property Some: integer read GetSome;
        property Data: double read GetData;
    end;

    TDataClass = class(TInterfacedObject, IDataClass)
    private
        FSome: integer;
        FData: double;
    protected
        function GetSome: integer;
        function GetData: double;
    public
        constructor Create(ASome: integer; AData: double);
    end;

然后您将所有变量都设为 ISomeData 类型（混合 ISomeData 和 TSomeData 是一个非常糟糕的主意...您很容易遇到引用计数问题）。

基本上，这会导致引用计数在您的阅读器代码中自动增加，它会在其中加载对数据的本地引用，并在变量离开范围时减少，此时它将在那里取消分配。

我知道在接口和类实现中复制数据类的 API 有点乏味，但这是获得所需行为的最简单方法。

【讨论】：

不幸的是，接口的引用计数不是线程安全的。
引用计数是线程安全的。在多个线程之间共享单个接口变量不是线程安全的。
这确实使事情本身有点复杂。我显然需要深入研究 Delphi 以验证在线程代码中使用 TInterfacedObject 时什么是安全的。

【解决方案4】：

我有一个潜在的解决方案给你；它让新读者可以随时开始，直到作者想写为止。然后作者等待读者完成并执行其写入。写完后读者可以再看一遍。

此外，此解决方案不需要锁或互斥锁，但它确实需要原子测试和设置操作。我不了解 Delphi，我用 Lisp 编写了我的解决方案，所以我将尝试用伪代码来描述它。

（CAPS 是函数名，所有这些函数都接受和返回不带参数）

integer access-mode = 1; // start in reader mode. 

WRITE  loop with current = accessmode, 
            with new = (current & 0xFFFFFFFe) 
            until test-and-set(access-mode, current to new)
       loop until access-mode = 0; 

ENDWRITE assert( access-mode = 0)
         set access-mode to 1

READ loop with current = ( accessmode | 1 ),
          with new = (current + 2),
          until test-and-set(access-mode, current to new)
ENDREAD loop with current = accessmode
             with new = (current - 2),
             until test-and-set(access-mode, current to new)

要使用，阅读器在阅读前调用 READ，完成后调用 ENDREAD。单独的写入者在写入之前调用 WRITE，完成后调用 ENDWRITE。

这个想法是一个称为访问模式的整数，它在最低位保存一个布尔值，并在较高的位。 WRITE 将该位设置为 0，然后旋转直到足够的 ENDREAD 将访问模式倒计时为零。 Endwrite 将访问模式设置回 1。将当前访问模式与 1 进行 READ OR 运算，因此只有在低位开始时它们的测试和设置才会通过。我加减 2 以保留低位。

要计算读者数量，只需将访问模式右移一位即可。

【讨论】：