1:
2:
3:
4:
5:
6:
7:
8:
9:
10:
11:
12:
13:
14:
15:
16:
17:
18:
19:
20:
21:
22:
23:
24:
25:
26:
27:
28:
29:
30:
31:
32:
33:
34:
35:
36:
37:
38:
39:
40:
41:
42:
43:
44:
45:
46:
47:
48:
49:
50:
51:
52:
53:
54:
55:
56:
57:
58:
59:
60:
61:
62:
63:
64:
65:
66:
67:
68:
69:
70:
71:
72:
73:
74:
75:
76:
77:
78:
79:
80:
81:
82:
83:
84:
85:
86:
87:
88:
89:
90:
91:
92:
93:
94:
95:
96:
97:
98:
99:
100:
101:
102:
103:
104:
105:
106:
107:
108:
109:
110:
111:
112:
113:
114:
115:
116:
117:
118:
119:
120:
121:
122:
123:
124:
125:
126:
127:
128:
129:
130:
131:
132:
133:
134:
135:
136:
137:
138:
139:
140:
141:
142:
143:
144:
145:
146:
147:
148:
149:
150:
151:
152:
153:
154:
155:
156:
157:
158:
159:
160:
161:
162:
163:
164:
165:
166:
167:
168:
169:
170:
171:
172:
173:
174:
175:
176:
177:

unit u_dzMultiWriteSingleReadLockFreeQueue;

interface

uses
  Windows,
  SysUtils,
  SyncObjs;

type
  TIntObj = class(TObject)
  public
    Value: Integer;
    constructor Create(_Value: integer);
  end;

type
  _ItemType_ = TIntObj;

type
  ///<summary> Do not use this class directly, use
  ///          TMultiWriteSingleReadLockFreeQueue instead </summary>
  TInternalQueue = class
  private
    FReadIdx: integer;
    FWriteIdx: integer;
    FItems: array of _ItemType_;
  public
    constructor Create(_Length: integer);
    function Enqueue(_Item: _ItemType_): boolean;
    function Dequeue(out _Item: _ItemType_): boolean;
  end;

type
  TMultiWriteSingleReadLockFreeQueue = class
  private
    FWriteQueue: TInternalQueue;
    FReadQueue: TInternalQueue;
    FActiveWriters: integer;
    FActiveWritersZeroEvent: TEvent;
    FLength: integer;
  public
    constructor Create(_Length: integer);
    destructor Destroy; override;
    function Enqueue(_Item: _ItemType_): boolean;
    function Dequeue(out _Item: _ItemType_): boolean;
  end;

implementation

{ TIntObj }

constructor TIntObj.Create(_Value: integer);
begin
  inherited Create;
  Value := _Value;
end;

{ TInternalQueue }

constructor TInternalQueue.Create(_Length: integer);
begin
  inherited Create;
  SetLength(FItems, _Length);
  FWriteIdx := -1;
end;

function TInternalQueue.Enqueue(_Item: _ItemType_): boolean;
var
  Idx: integer;
begin
  // InterlockedIncrement increments FWriteIdx and returns the
  // new value. Only one thread at a time can execute this function
  // so we can be sure that the returned Idx is unique within all
  // threads that call Enqeue.
  Idx := InterlockedIncrement(FWriteIdx);
  // Here is the catch: We cannot increase the length of the array
  // in a thread safe manner without locking it, so it is possible
  // that Enqueue fails.
  Result := Idx < Length(FItems);
  if Result then
    FItems[Idx] := _Item;
end;

function TInternalQueue.Dequeue(out _Item: _ItemType_): boolean;
begin
  // We assume that we are the owner of this instance,
  // all write operations have finished and no more write
  // operations are possible, so we just read until we reach
  // FWriteIdx. No synchronization is necessary.
  Result := FReadIdx <= FWriteIdx;
  if Result then begin
    _Item := FItems[FReadIdx];
    Inc(FReadIdx);
  end;
end;

{ TMultiWriteSingleReadLockFreeQueue }

constructor TMultiWriteSingleReadLockFreeQueue.Create(_Length: integer);
begin
  inherited Create;
  FActiveWritersZeroEvent := TEvent.Create;
  FLength := _Length;
  FWriteQueue := TInternalQueue.Create(_Length);
  FReadQueue := nil;
end;

destructor TMultiWriteSingleReadLockFreeQueue.Destroy;
var
  Item: _ItemType_;
begin
  while Dequeue(Item) do
    Item.Free;
  FActiveWritersZeroEvent.Free;
  inherited;
end;

function TMultiWriteSingleReadLockFreeQueue.Enqueue(_Item: _ItemType_): boolean;
var
  Queue: TInternalQueue;
begin
  // this is the simple case: The WriteQueue can be written to
  // by any number of threads in parallel, no locking is required
  // In addition we count the number of threads that are executing
  // this method and therefore hold a reference to the WriteQueue.
  // This is so the reader thread knows when it can safely
  // use its copy.
  InterlockedIncrement(FActiveWriters);
  try
    // First, we get our local reference of the WriteQueue,
    // this is probably not necessary but just to be sure.
    Queue := FWriteQueue;
    // Then we enqueue our item to it
    Result := Queue.Enqueue(_Item);
  finally
    // decrement the ActiveWriters, if we reach zero,
    // notify the reading thread that it now owns he
    // reading queue, in case it is waiting for it.
    if InterlockedDecrement(FActiveWriters) = 0 then
      FActiveWritersZeroEvent.SetEvent;
  end;
end;

function TMultiWriteSingleReadLockFreeQueue.Dequeue(out _Item: _ItemType_): boolean;
var
  NewQueue: TInternalQueue;
begin
  if not Assigned(FReadQueue) then begin
    // we create a new queue for all the writer threads to use
    // and assign it as the new WriteQueue, the old one is assigned
    // to ReadQueue. InterlockedExchange assures that nobody
    // can access the WriteQueue while we are changing it.
    NewQueue := TInternalQueue.Create(FLength);
    FReadQueue := TInternalQueue(InterlockedExchange(Integer(FWriteQueue), Integer(NewQueue)));

    // Unfortunately it is possible that other threads still hold a
    // reference to the old queue.
    // To be 100% sure we need to wait until the ActiveWriters count
    // drops to 0 (once is enough, because all new calls to Enqueue
    // will use the new WriteQueue)
    // We reset the event, just in case we need it
    FActiveWritersZeroEvent.ResetEvent;
    // If there currently are writers, we wait for the event
    // which will be set by the first writer that decrements
    // ActiveWriters to 0
    if FActiveWriters <> 0 then
      FActiveWritersZeroEvent.WaitFor(INFINITE);
    // OK, now we own the read queue, just Dequeue from it
  end;
  Result := FReadQueue.Dequeue(_Item);
  // If the ReadQueue is empty, free it
  if not Result then
    FreeAndNil(FReadQueue);
end;

end.