2つのスレッド用のシンプルなロックフリーオブジェクト

 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.