Простейшие Lock-Free объекты для двух потоков

program LockFreeTest;

{$APPTYPE CONSOLE}

{$R *.res}

uses
  SysUtils, Classes, Windows, Messages;

// Lock-free work thread state ////////////////////////////////////////////////
type
  TLockFreeWorkState = class
  protected
    FNewest: Boolean;
  public
    class function Read(var CurrentItem, ReadItem): Boolean;
    class procedure Write(var CurrentItem, WriteItem);
    property Newest: Boolean read FNewest write FNewest;
  end;

class function TLockFreeWorkState.Read(var CurrentItem, ReadItem): Boolean;
begin
  if TLockFreeWorkState(CurrentItem).Newest then begin
    pointer(ReadItem) := InterlockedExchangePointer(pointer(CurrentItem), pointer(ReadItem));
    TLockFreeWorkState(ReadItem).Newest := False;
    Result := True;
  end else
    Result := False;
end;

class procedure TLockFreeWorkState.Write(var CurrentItem, WriteItem);
begin
  TLockFreeWorkState(WriteItem).Newest := True;
  pointer(WriteItem) := InterlockedExchangePointer(pointer(CurrentItem), pointer(WriteItem));
end;

// Lock-free work thread queue ////////////////////////////////////////////////
type
  TLockFreeWorkQueue = class
  public const
    QueueCapacity = 4; // Small value for test purposes
  public type
    TLockFreeWorkQueueItems = array[0..QueueCapacity - 1] of TObject;
  public
    Head: Integer; // Access from main thread only
    Tail: Integer; // Access from work thread only
    NextQueue: TLockFreeWorkQueue;
    Items: TLockFreeWorkQueueItems;
  public
    destructor Destroy; override;
    class procedure Add(var WriteQueue: TLockFreeWorkQueue; Item: TObject); static;
    class function Extract(var ReadQueue: TLockFreeWorkQueue): TObject; static;
  end;

destructor TLockFreeWorkQueue.Destroy;
var
  i: Integer;
begin
  // Free non-extracted items
  for i := 0 to QueueCapacity - 1 do
    Items[i].Free;
  // Free NextQueue if exists
  NextQueue.Free;
  inherited;
end;

class procedure TLockFreeWorkQueue.Add(var WriteQueue: TLockFreeWorkQueue; Item: TObject);
var
  NewWriteQueue: TLockFreeWorkQueue;
begin
  // Check item assigned (can't add empty items)
  if not Assigned(Item) or not Assigned(WriteQueue) then
    Exit;
  if InterlockedCompareExchangePointer(pointer(WriteQueue.Items[WriteQueue.Tail]), pointer(Item), Nil) = Nil then begin
    // Added successfully
    Inc(WriteQueue.Tail);
    if WriteQueue.Tail = QueueCapacity then
      WriteQueue.Tail := 0;
  end else begin
    // WriteQueue full. Create new chained queue.
    NewWriteQueue := TLockFreeWorkQueue.Create;
    NewWriteQueue.Items[0] := Item;
    Inc(NewWriteQueue.Tail);
    if NewWriteQueue.Tail = QueueCapacity then // Check single-item queue
      NewWriteQueue.Tail := 0;
    InterlockedExchangePointer(pointer(WriteQueue.NextQueue), NewWriteQueue);
    WriteQueue := NewWriteQueue;
  end;
end;

class function TLockFreeWorkQueue.Extract(var ReadQueue: TLockFreeWorkQueue): TObject;
var
  NewReadQueue: TLockFreeWorkQueue;
begin
  Result := Nil;
  if not Assigned(ReadQueue) then
    Exit;
  repeat
    Result := InterlockedExchangePointer(pointer(ReadQueue.Items[ReadQueue.Head]), Nil);
    if Result = Nil then begin
      // No new items in this queue. Check next queue is available
      NewReadQueue := InterlockedExchangePointer(pointer(ReadQueue.NextQueue), Nil);
      if Assigned(NewReadQueue) then begin
        ReadQueue.Free;
        ReadQueue := NewReadQueue;
      end else
        // No new item in queue
        Exit;
    end;
  until Result <> Nil;
  // Item extracted successfully
  Inc(ReadQueue.Head);
  if ReadQueue.Head = QueueCapacity then
    ReadQueue.Head := 0;
end;

// Test work thread ///////////////////////////////////////////////////////////
const
  WM_MAINNOTIFY = WM_USER + 1;

type
  TWorkThreadState = class(TLockFreeWorkState)
  public
    Progress: Integer;
  end;

  TWorkThreadQueueItem = class
  public
    ItemData: Integer;
  end;

  TWorkThread = class(TThread)
  protected
    FMainHandle: THandle;
    FMainNotified: Integer;
    // State fields
    FStateRead: TWorkThreadState;
    FStateCurrent: TWorkThreadState;
    FStateWrite: TWorkThreadState;
    // Queue fields
    FQueueRead: TLockFreeWorkQueue;
    FQueueWrite: TLockFreeWorkQueue;
    // Debug (test) fiels
    FDebugReadQueue: Boolean;
    procedure Execute; override;
    procedure SetState;
    procedure AddQueueItem(Item: TWorkThreadQueueItem);
    procedure NotifyMain;
  public
    constructor Create(CreateSuspended: Boolean);
    destructor Destroy; override;
    function GetState: TWorkThreadState;
    function ExtractQueueItem: TWorkThreadQueueItem;
    procedure NotificationProcessed;
    property MainHandle: THandle read FMainHandle;
  end;

constructor TWorkThread.Create(CreateSuspended: Boolean);
begin
  inherited Create(CreateSuspended);
  // State objects
  FStateRead := TWorkThreadState.Create;
  FStateCurrent := TWorkThreadState.Create;
  FStateWrite := TWorkThreadState.Create;
  // Queue objects
  FQueueRead := TLockFreeWorkQueue.Create;
  FQueueWrite := FQueueRead;
end;

destructor TWorkThread.Destroy;
begin
  inherited;
  FStateRead.Free;
  FStateCurrent.Free;
  FStateWrite.Free;
  // Always destroy read queue only: only read queue may have NextQueue reference
  FQueueRead.Free;
end;

procedure TWorkThread.NotifyMain;
begin
  if InterlockedExchange(FMainNotified, 1) = 0 then
    PostMessage(FMainHandle, WM_MAINNOTIFY, 0, 0);
end;

procedure TWorkThread.NotificationProcessed;
begin
  InterlockedExchange(FMainNotified, 0);
end;

function TWorkThread.GetState: TWorkThreadState;
begin
  TLockFreeWorkState.Read(FStateCurrent, FStateRead);
  Result := FStateRead;
end;

procedure TWorkThread.SetState;
begin
  TLockFreeWorkState.Write(FStateCurrent, FStateWrite);
end;

procedure TWorkThread.AddQueueItem(Item: TWorkThreadQueueItem);
begin
  TLockFreeWorkQueue.Add(FQueueWrite, Item);
end;

function TWorkThread.ExtractQueueItem: TWorkThreadQueueItem;
begin
  Result := TWorkThreadQueueItem(TLockFreeWorkQueue.Extract(FQueueRead));
end;

procedure TWorkThread.Execute;
const
  TestQueueCountToFlush = 10;
var
  ProgressIndex: Integer;
  TestQueueCount: Integer;
  Item: TWorkThreadQueueItem;
begin
  TestQueueCount := 0;
  ProgressIndex := 0;
  while not Terminated do begin
    // Send current progress
    if FStateWrite.Progress <> ProgressIndex then begin
      // All state object fields initialization required
      FStateWrite.Progress := ProgressIndex;
      SetState;
      NotifyMain;
    end;
    // Emulate calculation
    Sleep(500);
    Inc(ProgressIndex);
    // Put intermediate result in queue
    Item := TWorkThreadQueueItem.Create;
    Item.ItemData := ProgressIndex;
    AddQueueItem(Item);
    Inc(TestQueueCount);
    if TestQueueCount = TestQueueCountToFlush then begin
      TestQueueCount := 0;
      // Allow queue reading from main thread
      FDebugReadQueue := True;
      NotifyMain;
    end;
  end;
end;

// Test application ///////////////////////////////////////////////////////////
type
  TMain = class
  protected
    FHandle: THandle;
    FThread: TWorkThread;
    procedure WndProc(var Message: TMessage);
  public
    constructor Create;
    destructor Destroy; override;
    function Run: Boolean;
    property Handle: THandle read FHandle;
  end;

var
  Main: TMain;

constructor TMain.Create;
begin
  FHandle := AllocateHWnd(WndProc);
  FThread := TWorkThread.Create(True);
  FThread.FMainHandle := Handle;
  FThread.Start;
  writeln('Work thread started');
end;

destructor TMain.Destroy;
begin
  writeln('Stopping work thread...');
  FThread.Free;
  writeln('Work thread stopped');
  DeallocateHWnd(FHandle);
  inherited;
end;

procedure TMain.WndProc(var Message: TMessage);
var
  State: TWorkThreadState;
  Item: TWorkThreadQueueItem;
begin
  if Message.Msg = WM_MAINNOTIFY then begin
    FThread.NotificationProcessed;
    State := FThread.GetState;
    // Show current progress
    writeln('Work progress ', State.Progress);
    // Check queue reading allowed
    if FThread.FDebugReadQueue then begin
      writeln('Read queue...');
      repeat
        Item := FThread.ExtractQueueItem;
        try
          if Assigned(Item) then
            writeln('Queue item: ', Item.ItemData);
        finally
          Item.Free;
        end;
      until not Assigned(Item);
      FThread.FDebugReadQueue := False;
    end;
  end else
    Message.Result := DefWindowProc(Handle, Message.Msg, Message.wParam, Message.lParam);
end;

function TMain.Run: Boolean;
var
  Msg: TMsg;
begin
  writeln('Start message loop (Ctrl+C to break)');
  Result := True;
  while Result do
    case Integer(GetMessage(Msg, Handle, 0, 0)) of
      0:
        Break;
      -1:
        Result := False;
      else
        begin
          TranslateMessage(Msg);
          DispatchMessage(Msg);
        end;
    end;
end;

// Console event handler //////////////////////////////////////////////////////

function ConsoleEventProc(CtrlType: DWORD): BOOL; stdcall;
begin
  Result := False;
  case CtrlType of
    CTRL_CLOSE_EVENT,
    CTRL_C_EVENT,
    CTRL_BREAK_EVENT:
      if Assigned(Main) then begin
        PostMessage(Main.Handle, WM_QUIT, 0, 0);
        Result := True;
      end;
  end;
end;

// Main procedure /////////////////////////////////////////////////////////////

begin
  {$IFDEF DEBUG}
  ReportMemoryLeaksOnShutdown := True;
  {$ENDIF}
  try
    SetConsoleCtrlHandler(@ConsoleEventProc, True);
    Main := TMain.Create;
    try
      Main.Run;
    finally
      FreeAndNil(Main);
    end;
  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
end.