更快的TMultiReadExclusiveWriteSynchronizer？

Question

有没有更快的TMultiReadExclusiveWriteSynchronizer类型呢？比如FastCode？

从Windows Vista开始，微软添加了Slim Reader/Writer lock。它比Delphi的TMultiReadExclusiveWriteSynchronizer表现更好。不幸的是，它只存在于Windows Vista及以后版本中，而这些版本的客户并不多。

可以假设在Slim Reader/Writer lock内使用的概念可以在本地Delphi代码中重新实现，但是否有人已经这样做了呢？

我遇到了这样一种情况：获取和释放TMultiReadExclusiveWriteSynchronizer上的锁（即使没有争用-单个线程），会导致100%的开销（操作时间加倍）。我可以不锁定运行，但这时我的类就不再是线程安全的。

我需要翻译的内容是：“是否有更快的TMultiReadExclusiveWriteSynchronizer？” 注意：如果我使用TCriticalSection，只会导致2%的性能损失（尽管在获取成功时，即单线程且没有争用时，临界区被认为是快速的。）。 CS的缺点是我失去了“多个读取器”的功能。

测量结果

使用TMultiReadExclusiveWriteSynchronizer，在BeginRead和EndRead内花费了相当多的时间：

我接着将代码移植到使用Windows自己的SlimReaderWriter Lock(需要进行一些代码重写，因为它不支持递归锁定)，并对结果进行了分析：

• TMultiReadExclusiveWriteSynchronizer: 每次迭代耗时10,698纳秒
  迭代1,000,000次总计耗时10,697,772,613纳秒

• SRWLock: 每次迭代耗时8,802纳秒
  迭代1,000,000次总计耗时8,801,678,339纳秒

• Omni Reader-Writer lock: 每次迭代耗时8,941纳秒
  迭代1,000,000次总计耗时8,940,552,487纳秒

使用SRWLocks（也称为Omni's spinning lock）后性能提高了17%。

现在，我无法永久切换代码以使用Windows Vista SRWLocks，因为还有一些企业客户仍在使用Windows XP。

这个“Slim locks”只是小心谨慎地使用了InterlockedCompareExchange函数; 但比我能够成功使用的更加小心谨慎。 我距离窃取所涉及的 140 条机器指令并使其完成还有很远的路要走。

额外阅读

• 临界区始终更快吗？
• 如何编写快速的多线程Delphi应用程序

- Ian Boyd

---

为什么你会感到惊讶，因为很少有商业客户升级到Windows 7？我有一些客户希望他们能尽快将SQL Server数据库从Windows 2000 Server迁移。 - Ian Boyd

我通常通过设计来解决这个问题，以避免出现需要锁定的情况。我知道这听起来不真诚，但我发现有两种主要情况需要使用多线程，我将其称为“响应式UI”和“网络服务”。响应式UI通常使用消息队列处理，而经过适当设计的服务会话不需要锁定，因为给定会话实际上是单线程的。我发现唯一真正需要锁定的情况是在维护网络服务中活动会话的集合时。 - Peter Wone

如果您频繁进行锁定和释放，以至于锁定导致明显的性能问题，那么您为并发性牺牲了太多单线程性能。有很多方法可以将权衡推向另一个方向。例如，不要在每个操作上获取/释放锁定，而是跨多个操作保持它。 - David Schwartz

由于您没有经历争用延迟，所有锁定和解锁（为避免争用而进行）都是不必要的。只需保持锁定状态即可。您会遇到更多的争用（这很好），并且可以避免所有无谓的锁定/解锁，从而提高单线程性能。（正如我们在业界所说，您需要增加锁的范围。） - David Schwartz

不知道你需要这些锁的原因，很难推荐特定的技术。你的使用场景是什么？ - Ross Judson

Answer 1

TOmniMREW来自OmniThreadLibrary，声称更快、更轻量级：

顺便说一下，OTL是一个很棒的线程库。

示例代码

TOmniReaderWriterLock = class(TInterfacedObject, IReaderWriterLock)
private
   omrewReference: Integer;
public
   { IReaderWriterLock }
   procedure BeginRead;
   procedure EndRead;
   procedure BeginWrite;
   procedure EndWrite;
end;

{ TOmniReaderWriterLock }

procedure TOmniReaderWriterLock.BeginRead;
var
  currentReference: Integer;
begin
    //Wait on writer to reset write flag so Reference.Bit0 must be 0 than increase Reference
    repeat
        currentReference := Integer(omrewReference) and not 1;
    until currentReference = Integer(InterlockedCompareExchange(Pointer(omrewReference), Pointer(Integer(currentReference) + 2), Pointer(currentReference)));
end;

procedure TOmniReaderWriterLock.EndRead;
begin
    //Decrease omrewReference
    InterlockedExchangeAdd(@omrewReference, -2);
end;

procedure TOmniReaderWriterLock.BeginWrite;
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 = Integer(InterlockedCompareExchange(Pointer(omrewReference), Pointer(currentReference+1), Pointer(currentReference)));

    //Now wait on all readers
    repeat
    until omrewReference = 1;
end;

procedure TOmniReaderWriterLock.EndWrite;
begin
    omrewReference := 0;
end;

- Edwin Yip

Answer 2

最终，我采用了一种折中的解决方案。 Omni 读写锁使用了“轻量级”原则（自旋位操作）。与Windows自身的锁不同的是，它不支持锁升级。我已经测试过了，并且它似乎没有出现崩溃或死锁的情况。

最终，我采用了备选方案。最通用的通用接口来支持“读写”概念：

IReaderWriterLock = interface
   ['{6C4150D0-7B13-446D-9D8E-866B66723320}']
   procedure BeginRead;
   procedure EndRead;
   procedure BeginWrite;
   procedure EndWrite;
end;

然后我们在运行时决定使用哪个实现。如果我们在Windows Vista或更高版本上运行，则使用Windows自己的SlimReaderWriter，否则回退到Omni版本：

TReaderWriterLock = class(TObject)
public
   class function Create: IReaderWriterLock;
end;

class function TReaderWriterLock.Create: IReaderWriterLock;
begin
   if Win32MajorVersion >= 6 then //SRWLocks were introduced with Vista/Server 2008 (Windows version 6.0)
   begin
      //Use the Windows built-in Slim ReaderWriter lock
      Result := TSlimReaderWriterLock.Create;
   end
   else
   begin
      //XP and earlier fallback to Omni equivalent
      Result := TOmniReaderWriterLock.Create;
   end;
end;

注意：任何代码均已发布至公共领域。不需要署名。

- Ian Boyd

Answer 3

Delphi中的TMultiReadExclusiveWriteSynchronizer非常复杂 - 它可以被递归获取，您可以从Read更新到Write。

这是有代价的，也就是说需要为每个线程管理一个共享状态桶。由于Windows线程本地机制（通过threadvar访问）对于此太过简单（无法处理多个MREWS实例），因此以一种相当低效的方式进行处理 - 可以查看RTL或JCL源代码 - 实现非常相似，存在性能不佳和更新死锁风险。

首先确保您真正需要MREWS功能 - 根据锁定开销与工作负载的比例，我假设您使用TCriticalSection会更好。

如果您确实需要它，请使用Delphi实现，并注意可能在BeginWrite中隐藏的解锁 - 请查看其文档和返回值含义。

可以使用Interlocked函数或内联汇编语言实现类似于Vista的SRW，但在大多数情况下不值得付出努力。

- Viktor Svub

---

TMREWSync的读状态真的可以更新为写状态吗？ - Alex Egorov

Answer 4

JCL有一个MREWS，这是一种不同的实现，可能适合您。不确定它需要哪个版本的Windows。

http://wiki.delphi-jedi.org/wiki/JCL_Help:TJclMultiReadExclusiveWrite

http://wiki.delphi-jedi.org/index.php?title=JEDI_Code_Library

- IvoTops

Answer 5

试试这个？它可以像普通变量一样使用：

type myclass=class
              Lock:TOBRWLock;
              function ReadIt:Integer;
              procedure WriteIt(A:Integer);
             end;  
function ReadIt:Integer;
begin;
 Lock.LockRead;
 Result:=GetVal;
 Lock.UnLockRead;
end;

还有很大的改进空间，您可以从这里构建各种偏向于读取而非写入或根据需要以不同方式运行的变体。

const ldFree    = 0;
      ldReading = 1;
      ldWriting = 2;
type TOBRWLock          = record
                 [Volatile]WritersWaiting,
                           ReadersWaiting,
                           ReadersReading,
                           Disposition    : Integer;
                           procedure LockRead;
                           procedure LockWrite;
                           procedure UnlockRead;
                           procedure UnlockWrite;
                           procedure UnReadWrite;
                           procedure UnWriteRead;
                          end;

procedure TOBRWLock.LockRead;
var SpinCnt : NativeUInt;
    I       : Integer;
begin
 SpinCnt:=0;
 TInterlocked.Increment(ReadersWaiting);
 repeat
  if (Disposition=ldReading)
     then begin
           I:=TInterlocked.Increment(ReadersReading);
           if (Disposition<>ldReading) or (I=1)(*Only 1 read reference or Disposition changed, suspicious, rather retry*)
              then begin
                    TInterlocked.Decrement(ReadersReading);
                    continue;
                   end
              else begin(*Success*)
                    TInterlocked.Decrement(ReadersWaiting);
                    break;
                   end;
          end;
  if (WritersWaiting<>0)or(Disposition<>ldFree)
     then begin
           SpinBackoff(SpinCnt);
           continue;
          end;
  if TInterlocked.CompareExchange(Disposition,ldReading,ldFree)=ldFree
     then begin
           TInterlocked.Increment(ReadersReading);
           TInterlocked.Decrement(ReadersWaiting);
           break;
          end;
  SpinBackoff(SpinCnt);
 until False;
end;

procedure TOBRWLock.LockWrite;
var SpinCnt : NativeUInt;
begin
 SpinCnt:=0;
 TInterlocked.Increment(WritersWaiting);
 repeat
  if (Disposition<>ldFree)
     then begin
           SpinBackoff(SpinCnt);
           continue;
          end;
  if TInterlocked.CompareExchange(Disposition,ldWriting,ldFree)=ldFree
     then begin
           TInterlocked.Decrement(WritersWaiting);
           break;
          end
     else SpinBackoff(SpinCnt);
 until False;
end;

procedure TOBRWLock.UnlockRead;
begin
 {$IFDEF DEBUG}
 if Disposition<>ldReading
    then raise Exception.Create('UnlockRead a lock that is not Reading');
 {$ENDIF}
 TInterlocked.Decrement(ReadersReading);
 if ReadersReading=0
    then begin;
          if TInterlocked.CompareExchange(Disposition,ldFree,ldReading)<>ldReading
             then raise Exception.Create('Impossible 310');
         end;
end;

procedure TOBRWLock.UnlockWrite;
begin
 {$IFDEF DEBUG}
 if Disposition<>ldWriting
    then raise Exception.Create('UnlockWrite a lock that is not Writing');
 {$ENDIF}
 if TInterlocked.CompareExchange(Disposition,ldFree,ldWriting)<>ldWriting
    then raise Exception.Create('Impossible 321');
end;

procedure TOBRWLock.UnReadWrite;
var SpinCnt : NativeUInt;
begin
 {$IFDEF DEBUG}
 if Disposition<>ldReading
    then raise Exception.Create('UnReadWrite a lock that is not Reading');
 {$ENDIF}
 TInterlocked.Increment(WritersWaiting);
 SpinCnt:=0;
 repeat
  if ReadersReading=1(*Only me reading*)
     then begin;
           if TInterlocked.CompareExchange(Disposition,ldWriting,ldReading)<>ldReading(*Must always succeed*)
              then raise Exception.Create('Impossible 337');
           TInterlocked.Decrement(ReadersReading);
           TInterlocked.Decrement(WritersWaiting);
           break;
          end;
  SpinBackoff(SpinCnt);
 until False;
end;

- AntonE