Delphi高效读写锁

原作者: [db:作者] 来自: [db:来源] 收藏邀请

https://github.com/search?q=GetQueuedCompletionStatus+WaitforMultipleObjects&type=Code

本人设计了一个高效读写锁，可实现多个线程读一个线程写的锁，应该比Delphi自带的读写锁高效，本人没有做对比测试。

本文的锁不可以在一个线程里重入，否则会锁死，另外读写锁最多支持65535个线程同时读。

// HeZiHang@cnblogs
// 跨平台简易高效锁
 
unit utLocker;
 
interface
 
type
  // 多读单写锁
  // 1.写的时候阻塞其他所有写和读
  // 2.读的时候不阻塞其他读，但阻塞所有写，当阻塞了一个或以上的写后，将阻塞所有后来新的读
  TMultiReadSingleWriteLocker = class
  protected
    [Volatile]
    FLocker: Integer;
  public
    procedure LockRead;
    procedure UnLockRead; inline;
    procedure LockWrite;
    procedure UnLockWrite; inline;
    function TryLockRead: Boolean; inline;
    function TryLockWrite: Boolean; inline;
    constructor Create;
  end;
 
  TSimpleLocker = class
  protected
    [Volatile]
    FLocker: Integer;
  public
    procedure Lock;
    procedure UnLock; inline;
    function TryLock: Boolean; inline;
  end;
 
implementation
 
uses System.SyncObjs, System.SysUtils, System.Classes;
 
type
  TSpinWait = record
  private const
    YieldThreshold = 10;
    Sleep1Threshold = 20;
    Sleep0Threshold = 5;
  private
    FCount: Integer;
    function GetNextSpinCycleWillYield: Boolean; inline;
  public
    procedure Reset;inline;
    procedure SpinCycle;inline;
 
    property Count: Integer read FCount;
    property NextSpinCycleWillYield: Boolean read GetNextSpinCycleWillYield;
  end;
 
  { TSpinWait }
 
function TSpinWait.GetNextSpinCycleWillYield: Boolean;
begin
  Result := (FCount > YieldThreshold) or (CPUCount = 1);
end;
 
procedure TSpinWait.Reset;
begin
  FCount := 0;
end;
 
procedure TSpinWait.SpinCycle;
var
  SpinCount: Integer;
begin
  if NextSpinCycleWillYield then
  begin
    if FCount >= YieldThreshold then
      SpinCount := FCount - YieldThreshold
    else
      SpinCount := FCount;
    if SpinCount mod Sleep1Threshold = Sleep1Threshold - 1 then
      TThread.Sleep(1)
    else if SpinCount mod Sleep0Threshold = Sleep0Threshold - 1 then
      TThread.Sleep(0)
    else
      TThread.Yield;
  end
  else
    TThread.SpinWait(4 shl FCount);
  Inc(FCount);
  if FCount < 0 then
    FCount := YieldThreshold + 1;
end;
 
{ TMultiReadSingleWriteLocker }
 
procedure TMultiReadSingleWriteLocker.LockRead;
var
  CurLock: Integer;
  Wait: TSpinWait;
begin
  Wait.Reset;
  while True do
  begin
    CurLock := FLocker;
    if CurLock <= $FFFF then
    begin
      if TInterlocked.CompareExchange(FLocker, CurLock + 1, CurLock) = CurLock
      then
        Exit;
    end;
    Wait.SpinCycle;
  end;
end;
 
procedure TMultiReadSingleWriteLocker.LockWrite;
var
  CurLock: Integer;
  Wait: TSpinWait;
begin
  Wait.Reset;
  while True do
  begin
    CurLock := FLocker;
    if CurLock <= $FFFF then
    begin
      if TInterlocked.CompareExchange(FLocker, CurLock + $10000, CurLock) = CurLock
      then
        Exit;
    end;
    Wait.SpinCycle;
  end;
end;
 
function TMultiReadSingleWriteLocker.TryLockRead: Boolean;
var
  CurLock: Integer;
begin
  CurLock := FLocker;
  if CurLock <= $FFFF then
    Result := TInterlocked.CompareExchange(FLocker, CurLock + 1, CurLock)
      = CurLock
  else
    Result := False;
end;
 
function TMultiReadSingleWriteLocker.TryLockWrite: Boolean;
var
  CurLock: Integer;
begin
  CurLock := FLocker;
  if CurLock <= $FFFF then
    Result := TInterlocked.CompareExchange(FLocker, CurLock + $10000, CurLock)
      = CurLock
  else
    Result := False;
end;
 
procedure TMultiReadSingleWriteLocker.UnLockWrite;
begin
  if FLocker < $10000 then
    raise Exception.Create('TMultiReadSingleWriteLocker Error');
 
  TInterlocked.Add(FLocker, -$10000);
end;
 
procedure TMultiReadSingleWriteLocker.UnLockRead;
begin
  TInterlocked.Decrement(FLocker);
end;
 
constructor TMultiReadSingleWriteLocker.Create;
begin
  FLocker := 0;
end;
 
{ TSimpleLocker }
 
procedure TSimpleLocker.Lock;
var
  Wait: TSpinWait;
begin
  Wait.Reset;
  while True do
  begin
    if FLocker = 0 then
    begin
      if TInterlocked.CompareExchange(FLocker, 1, 0) = 0 then
        Exit;
    end;
    Wait.SpinCycle;
  end;
end;
 
function TSimpleLocker.TryLock: Boolean;
begin
  if FLocker = 0 then
  begin
    Result := TInterlocked.CompareExchange(FLocker, 1, 0) = 0;
  end
  else
    Result := False;
end;
 
procedure TSimpleLocker.UnLock;
begin
  if TInterlocked.CompareExchange(FLocker, 0, 1) <> 1 then
    raise Exception.Create('TSimpleLocker Error');
end;
 
end.

一个简易无锁池

1.所有读写无等待，不需要判断条件直接读写(除自动扩充容量时），效率是一般带锁或带条件判断池的两倍以上。

2.预先开辟2的幂大小容量，可自增，每次翻倍

3.仅提供思路，工程应用可靠性还不确定。

// 无锁池
// hezihang @cnblogs.com

// 20160228 增加代引用计数器内存块的池，增加编译指令POOLGROW功能，可打开关闭池的自动翻倍增长功能
// 20160225 修正Grow中FWritePtr没有增长Bug
// 20140609 增加Grow临界区，减少等待时间
// 20140608 修正可能存在同时Grow的Bug

unit Iocp.AtomPool;

interface

{ .$DEFINE POOLGROW }

Uses
  System.SysUtils,
  System.SyncObjs;

Type
  Int32 = Integer;
  UInt32 = Cardinal;

  TAtomPoolAbstract = class
  private
    FWritePtr: Int32;
    FReadPtr: Int32;
    FHighBound: UInt32;
    FData: array of Pointer;
{$IFDEF POOLGROW}
    FCs: TCriticalSection;
    FLock: Int32;
    procedure CheckGrow; inline;
    procedure Grow; inline;
{$ENDIF}
  Protected
    function AllocItemResource: Pointer; virtual; abstract;
    procedure FreeItemResource(Item: Pointer); virtual; abstract;
    function GetCapacity: UInt32;
    procedure FreeResources;
  Public
    procedure AllocResources;
    function Get: Pointer;
    procedure Put(Item: Pointer);
    Constructor Create(Capacity: UInt32); Virtual;
    Destructor Destroy; Override;
    property Capacity: UInt32 read GetCapacity;
  End;

  TAtomPoolMem4K = class(TAtomPoolAbstract)
    function AllocItemResource: Pointer; override;
    procedure FreeItemResource(Item: Pointer); override;
  end;

  // 内存块带引用计数器的池,池容量恒定不能增长
  TAtomMemoryPoolRef = class
  private
    FMemory: PByteArray;
    FWritePtr: Int32;
    FReadPtr: Int32;
    FHighBound: UInt32;
    FMemSize: UInt32;
    FData: array of Pointer;
    FDataRef: array of Int32;
  Protected
    function GetCapacity: UInt32;
    procedure AllocResources;
    procedure FreeResources;
  Public
    function Get: Pointer;
    procedure Put(Item: Pointer);
    function IncRef(Item: Pointer): Int32;
    function DecRef(var Item: Pointer): Int32;
    Constructor Create(Capacity: UInt32; MemSize: UInt32);
    Destructor Destroy; Override;
    property Capacity: UInt32 read GetCapacity;
    property MemSize:UInt32 read FMemSize;
  End;

Implementation

const
  MAXTHREADCOUNT = 1000; // 从池中申请资源最大线程数
  // 创建池，大小必须是2的幂,并且必须大于MAXTHREADCOUNT

Constructor TAtomPoolAbstract.Create(Capacity: UInt32);
var
  OK: Boolean;
Begin
  Inherited Create;
  OK := (Capacity and (Capacity - 1) = 0);
  OK := OK and (Capacity > MAXTHREADCOUNT);

  if not OK then
    raise Exception.Create(Format('池长度必须大于%d并为2的幂', [MAXTHREADCOUNT]));
{$IFDEF POOLGROW}
  FCs := TCriticalSection.Create;
{$ENDIF}
  FHighBound := Capacity - 1;
  FReadPtr := 0;
End;

Destructor TAtomPoolAbstract.Destroy;
Begin
  FreeResources;
  SetLength(FData, 0);
{$IFDEF POOLGROW}
  FCs.Free;
{$ENDIF}
  Inherited;
End;

procedure TAtomPoolAbstract.AllocResources;
var
  i: UInt32;
begin
  try
    SetLength(FData, Capacity);
    for i := 0 to FHighBound do
      FData[i] := AllocItemResource;
  except
    Raise Exception.Create('池申请内存失败');
  end;
end;

procedure TAtomPoolAbstract.FreeResources;
var
  i: UInt32;
begin
  for i := FHighBound downto 0 do
    Self.FreeItemResource(FData[i]);
end;

procedure TAtomPoolAbstract.Put(Item: Pointer);
var
  N: UInt32;
begin
{$IFDEF POOLGROW}
  CheckGrow;
{$ENDIF}
  N := TInterlocked.Increment(FWritePtr);
  FData[N and FHighBound] := Item;
end;

Function TAtomPoolAbstract.Get: Pointer;
var
{$IFDEF POOLGROW}
  N, M, K: UInt32;
{$ELSE}
  N: UInt32;
{$ENDIF}
begin
{$IFDEF POOLGROW}
  N := FWritePtr and FHighBound;
  M := FReadPtr and FHighBound;
  K := (M + MAXTHREADCOUNT) and FHighBound;
  if (N > M) and (N < K) then
  // if ((N > M) and (N < K)) or ((N < M) and (N > K)) then
  begin
    Grow
  end;
{$ENDIF}
  N := TInterlocked.Increment(FReadPtr);
  Result := FData[N and FHighBound];
end;

function TAtomPoolAbstract.GetCapacity: UInt32;
begin
  Result := FHighBound + 1;
end;

{$IFDEF POOLGROW}

procedure TAtomPoolAbstract.CheckGrow;
begin
  if TInterlocked.Add(FLock, 0) > 0 then
  begin
    while FLock = 1 do
      Sleep(0);
    FCs.Enter;
    FCs.Leave;
  end;
end;

procedure TAtomPoolAbstract.Grow;
var
  i, N: Integer;
begin
  if TInterlocked.CompareExchange(FLock, 1, 0) = 0 then // 加锁
  begin
    FCs.Enter;
    TInterlocked.Increment(FLock);
    N := Length(FData);
    SetLength(FData, N + N);
    for i := N to High(FData) do
      FData[i] := AllocItemResource;
    TInterlocked.Increment(FLock);
    FHighBound := High(FData);
    FWritePtr := FHighBound;
    FCs.Leave;
    TInterlocked.Exchange(FLock, 0);
  end
  else
    CheckGrow;
end;
{$ENDIF}
{ TAtomPoolMem4K }

function TAtomPoolMem4K.AllocItemResource: Pointer;
begin
  GetMem(Result, 4096);
end;

procedure TAtomPoolMem4K.FreeItemResource(Item: Pointer);
begin
  FreeMem(Item, 4096);
end;

Constructor TAtomMemoryPoolRef.Create(Capacity: UInt32; MemSize: UInt32);
var
  OK: Boolean;
Begin
  Inherited Create;
  OK := (Capacity and (Capacity - 1) = 0);
  OK := OK and (Capacity > MAXTHREADCOUNT);

  if not OK then
    raise Exception.Create(Format('池长度必须大于%d并为2的幂', [MAXTHREADCOUNT]));
  if FMemSize and $10 <> 0 then
    raise Exception.Create('内存块大小必须是16的倍数');

  FMemSize := MemSize;
  try
    AllocResources;
    FHighBound := Capacity - 1;
    FWritePtr := FHighBound;
    FReadPtr := 0;
  except
    Raise Exception.Create('池申请内存失败');
  end;
End;

function TAtomMemoryPoolRef.DecRef(var Item: Pointer): Int32;
var
  N: Integer;
begin
  N := (NativeUInt(Item) - NativeUInt(FMemory)) div FMemSize;
  if (N>=0) and (N<=FHighBound) then
  begin
    Result := TInterlocked.Decrement(FDataRef[N]);
    if Result = 0 then
    begin
      Put(Item);
      Item := nil;
    end;
  end
  else Result:=-1;
end;

Destructor TAtomMemoryPoolRef.Destroy;
Begin
  FreeResources;
  Inherited;
End;

procedure TAtomMemoryPoolRef.AllocResources;
var
  i: UInt32;
  P: PByteArray;
begin
  SetLength(FData, Capacity);
  SetLength(FDataRef, Capacity);
  FillChar(FDataRef[0], Capacity * Sizeof(FDataRef[0]), 0);
  GetMem(FMemory, Length(FData) * FMemSize); // 一次申请所有内存
  P := FMemory;
  for i := 0 to FHighBound do
  begin
    FData[i] := P;
    Inc(P, FMemSize);
  end;
end;

procedure TAtomMemoryPoolRef.FreeResources;
begin
  FreeMem(FMemory, Length(FData) * FMemSize);
  SetLength(FData, 0);
  SetLength(FDataRef, 0);
end;

procedure TAtomMemoryPoolRef.Put(Item: Pointer);
var
  N: UInt32;
begin
  N := TInterlocked.Increment(FWritePtr);
  FData[N and FHighBound] := Item;
end;

Function TAtomMemoryPoolRef.Get: Pointer;
var
  N: UInt32;
begin
  N := TInterlocked.Increment(FReadPtr);
  Result := FData[N and FHighBound];
end;

function TAtomMemoryPoolRef.GetCapacity: UInt32;
begin
  Result := FHighBound + 1;
end;

function TAtomMemoryPoolRef.IncRef(Item: Pointer): Int32;
var
  N: Integer;
begin
  N := (NativeInt(Item) - NativeInt(FMemory)) div FMemSize;
  if (N>=0) and (N<=FHighBound) then
    Result := TInterlocked.Increment(FDataRef[N])
  else
    Result:=-1;
end;

End.

简易高效的Delphi原子队列

本文提供Delphi一个基于原子操作的无锁队列，简易高效。适用于多线程大吞吐量操作的队列。

可用于Android系统和32，64位Windows系统。

感谢歼10和qsl提供了修改建议！

有如下问题：

1.必须事先足够大开辟内存，大到不会出现队列溢出了。

2.队列大小必须是2的幂

3.不能压入空指针

4.本程序还未经过工程应用考验

unit Iocp.AtomQueue;

interface

Uses
  SysUtils,
  SyncObjs;

Type
  TAtomFIFO = Class
  Protected
    FWritePtr: Integer;
    FReadPtr: Integer;
    FCount:Integer;
    FHighBound:Integer;
    FisEmpty:Integer;
    FData: array of Pointer;
    function GetSize:Integer;
  Public
    procedure Push(Item: Pointer);
    function Pop: Pointer;
    Constructor Create(Size: Integer); Virtual;
    Destructor Destroy; Override;
    Procedure Empty;
    property Size: Integer read GetSize;
    property UsedCount:Integer read FCount;
  End;

Implementation

//创建队列，大小必须是2的幂，需要开辟足够大的队列，防止队列溢出

Constructor TAtomFIFO.Create(Size: Integer);
var
  i:NativeInt;
  OK:Boolean;
Begin
  Inherited Create;
  OK:=(Size and (Size-1)=0);

  if not OK then raise Exception.Create('FIFO长度必须大于等于256并为2的幂');

  try
    SetLength(FData, Size);
    FHighBound:=Size-1;
  except
    Raise Exception.Create('FIFO申请内存失败');
  end;
End;

Destructor TAtomFIFO.Destroy;
Begin
  SetLength(FData, 0);
  Inherited;
End;

procedure TAtomFIFO.Empty;
begin
  while (TInterlocked.Exchange(FReadPtr, 0)<>0) and
  (TInterlocked.Exchange(FWritePtr, 0)<>0) and
  (TInterlocked.Exchange(FCount, 0)<>0) do;
end;

function TAtomFIFO.GetSize: Integer;
begin
  Result:=FHighBound+1;
end;

procedure TAtomFIFO.Push(Item:Pointer);
var
  N:Integer;
begin
  if Item=nil then Exit;

  N:=TInterlocked.Increment(FWritePtr) and FHighBound;
  FData[N]:=Item;
  TInterlocked.Increment(FCount);
end;

Function TAtomFIFO.Pop:Pointer;
var
  N:Integer;
begin
  if TInterlocked.Decrement(FCount)<0 then
  begin
    TInterlocked.Increment(FCount);
    Result:=nil;
  end
  else
  begin
    N:=TInterlocked.Increment(FReadPtr) and FHighBound;
    //假设线程A调用了Push,并且正好是第1个push，
    //执行了N:=TInterlocked.Increment(FWritePtr) and FHighBound,
    //还没执行FData[N]:=Item, 被切换到其他线程
    //此时假设线程B调用了Push，并且正好是第2个push,并且执行完毕，这样出现FCount=1,第2个Item不为空，而第一个Item还是nil（线程A还没执行赋值）
    //假设线程C执行Pop，由于Count>0（线程B的作用）所以可以执行到这里，但此时FData[N]=nil（线程A还没执行赋值），
    //因此线程C要等待线程A完成FData[N]:=Item后，才能取走FData[N]
    //出现这种情况的概率应该比较小，基本上不会浪费太多CPU
    while FData[N]=nil do Sleep(1);
    Result:=FData[N];

    FData[N]:=nil;
  end;
end;

End.

性能测试：

采用天地弦提供的评估程序，进行了一些修改，分别对使用不同的临界区的队列进行对比结果如下：

其中Swith是因队列读空，进行线程上下文切换的次数

Delphi的FIFO实现

FIFO主要用于多个不同线程或进程之间数据交换时做缓冲区用，尤其适合实时数据通讯应用中的数据缓冲，接收线程（进程）将数据写入FIFO，处理线程（进程）从FIFO取出数据

本单元中：

TMemoryFIFO类适用于单进程内不同线程之间交换数据

TMapFileFIFO类适用于不同进程之间交换数据

Unit UtFIFO;
 
Interface
 
Uses
  Windows,
  SysUtils,
  SyncObjs;
 
Type
  PFIFOStruct= ^TFIFOStruct;
 
  TFIFOStruct= Record
    FSize: Integer;
    FWritePtr: Integer;
    FReadPtr: Integer;
    FBuffer: TByteArray;
  End;
 
  TFIFOReadFunc= Function(Buf: Pointer; Count: Integer): Integer;
  TFIFOReadFuncOfObject= Function(const Buf;  Count: Integer): Integer Of Object;
 
  TAbstractFIFO= Class
  Protected
    FSelfAccess: Boolean;
    FDataStruct: PFIFOStruct; // 数据区指针
    Procedure AllocateResource(Size: Integer); Virtual; Abstract;
    Procedure FreeResources; Virtual; Abstract;
    Procedure Lock; Virtual; Abstract;
    Procedure UnLock; Virtual; Abstract;
  Public
    Function FIFOFreeSpace: Integer;
    Function FIFOUsedSpace: Integer;
    Function CheckFIFOFull: Boolean;
    Function CheckFIFOEmpty: Boolean;
    Function WriteData(const Buf: Pointer; Count: Integer): Integer; Virtual;
    Function ReadData(Buf: Pointer; Count: Integer): Integer; Virtual;
    Function ReadDataByFunc(Func: TFIFOReadFuncOfObject;
      Count: Integer): Integer; Virtual;
    Constructor Create(Size: Integer); Virtual;
    Destructor Destroy; Override;
    Procedure Empty;
    Function Size: Integer;
  End;
 
  TMemoryFIFO= Class(TAbstractFIFO)
  Protected
    FLocker: TCriticalSection;
    Procedure AllocateResource(Size: Integer); Override;
    Procedure FreeResources; Override;
    Procedure Lock; Override;
    Procedure UnLock; Override;
  Public
    Constructor Create(Size: Integer); Override;
    Destructor Destroy; Override;
  End;
 
  TFileMapFIFO= Class(TAbstractFIFO)
  Private
    FMaster:Boolean;
    FMapHandle: THandle; // 内存映射文件句柄
    FMutexHandle: THandle; // 互斥句柄
    FMapName: String; // 内存映射对象
    FPVHandl