Contribute to IDE0059 (unnecessary assignment)
[mono-project.git] / netcore / System.Private.CoreLib / shared / System / Collections / Concurrent / ConcurrentQueue.cs
blob39e77904b3b368ee6b9a70a35243c9da711c69c8
1 // Licensed to the .NET Foundation under one or more agreements.
2 // The .NET Foundation licenses this file to you under the MIT license.
3 // See the LICENSE file in the project root for more information.
5 using System.Collections.Generic;
6 using System.Diagnostics;
7 using System.Diagnostics.CodeAnalysis;
8 using System.Threading;
10 namespace System.Collections.Concurrent
12 /// <summary>
13 /// Represents a thread-safe first-in, first-out collection of objects.
14 /// </summary>
15 /// <typeparam name="T">Specifies the type of elements in the queue.</typeparam>
16 /// <remarks>
17 /// All public and protected members of <see cref="ConcurrentQueue{T}"/> are thread-safe and may be used
18 /// concurrently from multiple threads.
19 /// </remarks>
20 [DebuggerDisplay("Count = {Count}")]
21 [DebuggerTypeProxy(typeof(IProducerConsumerCollectionDebugView<>))]
22 public class ConcurrentQueue<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>
24 // This implementation provides an unbounded, multi-producer multi-consumer queue
25 // that supports the standard Enqueue/TryDequeue operations, as well as support for
26 // snapshot enumeration (GetEnumerator, ToArray, CopyTo), peeking, and Count/IsEmpty.
27 // It is composed of a linked list of bounded ring buffers, each of which has a head
28 // and a tail index, isolated from each other to minimize false sharing. As long as
29 // the number of elements in the queue remains less than the size of the current
30 // buffer (Segment), no additional allocations are required for enqueued items. When
31 // the number of items exceeds the size of the current segment, the current segment is
32 // "frozen" to prevent further enqueues, and a new segment is linked from it and set
33 // as the new tail segment for subsequent enqueues. As old segments are consumed by
34 // dequeues, the head reference is updated to point to the segment that dequeuers should
35 // try next. To support snapshot enumeration, segments also support the notion of
36 // preserving for observation, whereby they avoid overwriting state as part of dequeues.
37 // Any operation that requires a snapshot results in all current segments being
38 // both frozen for enqueues and preserved for observation: any new enqueues will go
39 // to new segments, and dequeuers will consume from the existing segments but without
40 // overwriting the existing data.
42 /// <summary>Initial length of the segments used in the queue.</summary>
43 private const int InitialSegmentLength = 32;
44 /// <summary>
45 /// Maximum length of the segments used in the queue. This is a somewhat arbitrary limit:
46 /// larger means that as long as we don't exceed the size, we avoid allocating more segments,
47 /// but if we do exceed it, then the segment becomes garbage.
48 /// </summary>
49 private const int MaxSegmentLength = 1024 * 1024;
51 /// <summary>
52 /// Lock used to protect cross-segment operations, including any updates to <see cref="_tail"/> or <see cref="_head"/>
53 /// and any operations that need to get a consistent view of them.
54 /// </summary>
55 private readonly object _crossSegmentLock;
56 /// <summary>The current tail segment.</summary>
57 private volatile ConcurrentQueueSegment<T> _tail;
58 /// <summary>The current head segment.</summary>
59 private volatile ConcurrentQueueSegment<T> _head; // SOS's ThreadPool command depends on this name
61 /// <summary>
62 /// Initializes a new instance of the <see cref="ConcurrentQueue{T}"/> class.
63 /// </summary>
64 public ConcurrentQueue()
66 _crossSegmentLock = new object();
67 _tail = _head = new ConcurrentQueueSegment<T>(InitialSegmentLength);
70 /// <summary>
71 /// Initializes a new instance of the <see cref="ConcurrentQueue{T}"/> class that contains elements copied
72 /// from the specified collection.
73 /// </summary>
74 /// <param name="collection">
75 /// The collection whose elements are copied to the new <see cref="ConcurrentQueue{T}"/>.
76 /// </param>
77 /// <exception cref="System.ArgumentNullException">The <paramref name="collection"/> argument is null.</exception>
78 public ConcurrentQueue(IEnumerable<T> collection)
80 if (collection == null)
82 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.collection);
85 _crossSegmentLock = new object();
87 // Determine the initial segment size. We'll use the default,
88 // unless the collection is known to be larger than that, in which
89 // case we round its length up to a power of 2, as all segments must
90 // be a power of 2 in length.
91 int length = InitialSegmentLength;
92 if (collection is ICollection<T> c)
94 int count = c.Count;
95 if (count > length)
97 length = Math.Min(ConcurrentQueueSegment<T>.RoundUpToPowerOf2(count), MaxSegmentLength);
101 // Initialize the segment and add all of the data to it.
102 _tail = _head = new ConcurrentQueueSegment<T>(length);
103 foreach (T item in collection)
105 Enqueue(item);
109 /// <summary>
110 /// Copies the elements of the <see cref="ICollection"/> to an <see
111 /// cref="Array"/>, starting at a particular <see cref="Array"/> index.
112 /// </summary>
113 /// <param name="array">
114 /// The one-dimensional <see cref="Array">Array</see> that is the destination of the
115 /// elements copied from the <see cref="ConcurrentQueue{T}"/>. <paramref name="array"/> must have
116 /// zero-based indexing.
117 /// </param>
118 /// <param name="index">The zero-based index in <paramref name="array"/> at which copying begins.</param>
119 /// <exception cref="ArgumentNullException"><paramref name="array"/> is a null reference (Nothing in
120 /// Visual Basic).</exception>
121 /// <exception cref="ArgumentOutOfRangeException"><paramref name="index"/> is less than
122 /// zero.</exception>
123 /// <exception cref="ArgumentException">
124 /// <paramref name="array"/> is multidimensional. -or-
125 /// <paramref name="array"/> does not have zero-based indexing. -or-
126 /// <paramref name="index"/> is equal to or greater than the length of the <paramref name="array"/>
127 /// -or- The number of elements in the source <see cref="ICollection"/> is
128 /// greater than the available space from <paramref name="index"/> to the end of the destination
129 /// <paramref name="array"/>. -or- The type of the source <see
130 /// cref="ICollection"/> cannot be cast automatically to the type of the
131 /// destination <paramref name="array"/>.
132 /// </exception>
133 void ICollection.CopyTo(Array array, int index)
135 // Special-case when the Array is actually a T[], taking a faster path
136 if (array is T[] szArray)
138 CopyTo(szArray, index);
139 return;
142 // Validate arguments.
143 if (array == null)
145 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.array);
148 // Otherwise, fall back to the slower path that first copies the contents
149 // to an array, and then uses that array's non-generic CopyTo to do the copy.
150 ToArray().CopyTo(array, index);
153 /// <summary>
154 /// Gets a value indicating whether access to the <see cref="ICollection"/> is
155 /// synchronized with the SyncRoot.
156 /// </summary>
157 /// <value>true if access to the <see cref="ICollection"/> is synchronized
158 /// with the SyncRoot; otherwise, false. For <see cref="ConcurrentQueue{T}"/>, this property always
159 /// returns false.</value>
160 bool ICollection.IsSynchronized => false; // always false, as true implies synchronization via SyncRoot
162 /// <summary>
163 /// Gets an object that can be used to synchronize access to the <see
164 /// cref="ICollection"/>. This property is not supported.
165 /// </summary>
166 /// <exception cref="NotSupportedException">The SyncRoot property is not supported.</exception>
167 object ICollection.SyncRoot { get { ThrowHelper.ThrowNotSupportedException(ExceptionResource.ConcurrentCollection_SyncRoot_NotSupported); return default; } }
169 /// <summary>Returns an enumerator that iterates through a collection.</summary>
170 /// <returns>An <see cref="IEnumerator"/> that can be used to iterate through the collection.</returns>
171 IEnumerator IEnumerable.GetEnumerator() => ((IEnumerable<T>)this).GetEnumerator();
173 /// <summary>
174 /// Attempts to add an object to the <see cref="Concurrent.IProducerConsumerCollection{T}"/>.
175 /// </summary>
176 /// <param name="item">The object to add to the <see
177 /// cref="Concurrent.IProducerConsumerCollection{T}"/>. The value can be a null
178 /// reference (Nothing in Visual Basic) for reference types.
179 /// </param>
180 /// <returns>true if the object was added successfully; otherwise, false.</returns>
181 /// <remarks>For <see cref="ConcurrentQueue{T}"/>, this operation will always add the object to the
182 /// end of the <see cref="ConcurrentQueue{T}"/>
183 /// and return true.</remarks>
184 bool IProducerConsumerCollection<T>.TryAdd(T item)
186 Enqueue(item);
187 return true;
190 /// <summary>
191 /// Attempts to remove and return an object from the <see cref="Concurrent.IProducerConsumerCollection{T}"/>.
192 /// </summary>
193 /// <param name="item">
194 /// When this method returns, if the operation was successful, <paramref name="item"/> contains the
195 /// object removed. If no object was available to be removed, the value is unspecified.
196 /// </param>
197 /// <returns>true if an element was removed and returned successfully; otherwise, false.</returns>
198 /// <remarks>For <see cref="ConcurrentQueue{T}"/>, this operation will attempt to remove the object
199 /// from the beginning of the <see cref="ConcurrentQueue{T}"/>.
200 /// </remarks>
201 bool IProducerConsumerCollection<T>.TryTake(out T item) => TryDequeue(out item);
203 /// <summary>
204 /// Gets a value that indicates whether the <see cref="ConcurrentQueue{T}"/> is empty.
205 /// </summary>
206 /// <value>true if the <see cref="ConcurrentQueue{T}"/> is empty; otherwise, false.</value>
207 /// <remarks>
208 /// For determining whether the collection contains any items, use of this property is recommended
209 /// rather than retrieving the number of items from the <see cref="Count"/> property and comparing it
210 /// to 0. However, as this collection is intended to be accessed concurrently, it may be the case
211 /// that another thread will modify the collection after <see cref="IsEmpty"/> returns, thus invalidating
212 /// the result.
213 /// </remarks>
214 public bool IsEmpty
218 // IsEmpty == !TryPeek. We use a "resultUsed:false" peek in order to avoid marking
219 // segments as preserved for observation, making IsEmpty a cheaper way than either
220 // TryPeek(out T) or Count == 0 to check whether any elements are in the queue.
221 return !TryPeek(out _, resultUsed: false);
225 /// <summary>Copies the elements stored in the <see cref="ConcurrentQueue{T}"/> to a new array.</summary>
226 /// <returns>A new array containing a snapshot of elements copied from the <see cref="ConcurrentQueue{T}"/>.</returns>
227 public T[] ToArray()
229 // Snap the current contents for enumeration.
230 ConcurrentQueueSegment<T> head, tail;
231 int headHead, tailTail;
232 SnapForObservation(out head, out headHead, out tail, out tailTail);
234 // Count the number of items in that snapped set, and use it to allocate an
235 // array of the right size.
236 long count = GetCount(head, headHead, tail, tailTail);
237 T[] arr = new T[count];
239 // Now enumerate the contents, copying each element into the array.
240 using (IEnumerator<T> e = Enumerate(head, headHead, tail, tailTail))
242 int i = 0;
243 while (e.MoveNext())
245 arr[i++] = e.Current;
247 Debug.Assert(count == i);
250 // And return it.
251 return arr;
254 /// <summary>
255 /// Gets the number of elements contained in the <see cref="ConcurrentQueue{T}"/>.
256 /// </summary>
257 /// <value>The number of elements contained in the <see cref="ConcurrentQueue{T}"/>.</value>
258 /// <remarks>
259 /// For determining whether the collection contains any items, use of the <see cref="IsEmpty"/>
260 /// property is recommended rather than retrieving the number of items from the <see cref="Count"/>
261 /// property and comparing it to 0.
262 /// </remarks>
263 public int Count
267 var spinner = new SpinWait();
268 while (true)
270 // Capture the head and tail, as well as the head's head and tail.
271 ConcurrentQueueSegment<T> head = _head;
272 ConcurrentQueueSegment<T> tail = _tail;
273 int headHead = Volatile.Read(ref head._headAndTail.Head);
274 int headTail = Volatile.Read(ref head._headAndTail.Tail);
276 if (head == tail)
278 // There was a single segment in the queue. If the captured segments still
279 // match, then we can trust the values to compute the segment's count. (It's
280 // theoretically possible the values could have looped around and still exactly match,
281 // but that would required at least ~4 billion elements to have been enqueued and
282 // dequeued between the reads.)
283 if (head == _head &&
284 tail == _tail &&
285 headHead == Volatile.Read(ref head._headAndTail.Head) &&
286 headTail == Volatile.Read(ref head._headAndTail.Tail))
288 return GetCount(head, headHead, headTail);
291 else if (head._nextSegment == tail)
293 // There were two segments in the queue. Get the positions from the tail, and as above,
294 // if the captured values match the previous reads, return the sum of the counts from both segments.
295 int tailHead = Volatile.Read(ref tail._headAndTail.Head);
296 int tailTail = Volatile.Read(ref tail._headAndTail.Tail);
297 if (head == _head &&
298 tail == _tail &&
299 headHead == Volatile.Read(ref head._headAndTail.Head) &&
300 headTail == Volatile.Read(ref head._headAndTail.Tail) &&
301 tailHead == Volatile.Read(ref tail._headAndTail.Head) &&
302 tailTail == Volatile.Read(ref tail._headAndTail.Tail))
304 return GetCount(head, headHead, headTail) + GetCount(tail, tailHead, tailTail);
307 else
309 // There were more than two segments in the queue. Fall back to taking the cross-segment lock,
310 // which will ensure that the head and tail segments we read are stable (since the lock is needed to change them);
311 // for the two-segment case above, we can simply rely on subsequent comparisons, but for the two+ case, we need
312 // to be able to trust the internal segments between the head and tail.
313 lock (_crossSegmentLock)
315 // Now that we hold the lock, re-read the previously captured head and tail segments and head positions.
316 // If either has changed, start over.
317 if (head == _head && tail == _tail)
319 // Get the positions from the tail, and as above, if the captured values match the previous reads,
320 // we can use the values to compute the count of the head and tail segments.
321 int tailHead = Volatile.Read(ref tail._headAndTail.Head);
322 int tailTail = Volatile.Read(ref tail._headAndTail.Tail);
323 if (headHead == Volatile.Read(ref head._headAndTail.Head) &&
324 headTail == Volatile.Read(ref head._headAndTail.Tail) &&
325 tailHead == Volatile.Read(ref tail._headAndTail.Head) &&
326 tailTail == Volatile.Read(ref tail._headAndTail.Tail))
328 // We got stable values for the head and tail segments, so we can just compute the sizes
329 // based on those and add them. Note that this and the below additions to count may overflow: previous
330 // implementations allowed that, so we don't check, either, and it is theoretically possible for the
331 // queue to store more than int.MaxValue items.
332 int count = GetCount(head, headHead, headTail) + GetCount(tail, tailHead, tailTail);
334 // Now add the counts for each internal segment. Since there were segments before these,
335 // for counting purposes we consider them to start at the 0th element, and since there is at
336 // least one segment after each, each was frozen, so we can count until each's frozen tail.
337 // With the cross-segment lock held, we're guaranteed that all of these internal segments are
338 // consistent, as the head and tail segment can't be changed while we're holding the lock, and
339 // dequeueing and enqueueing can only be done from the head and tail segments, which these aren't.
340 for (ConcurrentQueueSegment<T> s = head._nextSegment!; s != tail; s = s._nextSegment!)
342 Debug.Assert(s._frozenForEnqueues, "Internal segment must be frozen as there's a following segment.");
343 count += s._headAndTail.Tail - s.FreezeOffset;
346 return count;
352 // We raced with enqueues/dequeues and captured an inconsistent picture of the queue.
353 // Spin and try again.
354 spinner.SpinOnce();
359 /// <summary>Computes the number of items in a segment based on a fixed head and tail in that segment.</summary>
360 private static int GetCount(ConcurrentQueueSegment<T> s, int head, int tail)
362 if (head != tail && head != tail - s.FreezeOffset)
364 head &= s._slotsMask;
365 tail &= s._slotsMask;
366 return head < tail ? tail - head : s._slots.Length - head + tail;
368 return 0;
371 /// <summary>Gets the number of items in snapped region.</summary>
372 private static long GetCount(ConcurrentQueueSegment<T> head, int headHead, ConcurrentQueueSegment<T> tail, int tailTail)
374 // All of the segments should have been both frozen for enqueues and preserved for observation.
375 // Validate that here for head and tail; we'll validate it for intermediate segments later.
376 Debug.Assert(head._preservedForObservation);
377 Debug.Assert(head._frozenForEnqueues);
378 Debug.Assert(tail._preservedForObservation);
379 Debug.Assert(tail._frozenForEnqueues);
381 long count = 0;
383 // Head segment. We've already marked it as frozen for enqueues, so its tail position is fixed,
384 // and we've already marked it as preserved for observation (before we grabbed the head), so we
385 // can safely enumerate from its head to its tail and access its elements.
386 int headTail = (head == tail ? tailTail : Volatile.Read(ref head._headAndTail.Tail)) - head.FreezeOffset;
387 if (headHead < headTail)
389 // Mask the head and tail for the head segment
390 headHead &= head._slotsMask;
391 headTail &= head._slotsMask;
393 // Increase the count by either the one or two regions, based on whether tail
394 // has wrapped to be less than head.
395 count += headHead < headTail ?
396 headTail - headHead :
397 head._slots.Length - headHead + headTail;
400 // We've enumerated the head. If the tail is different from the head, we need to
401 // enumerate the remaining segments.
402 if (head != tail)
404 // Count the contents of each segment between head and tail, not including head and tail.
405 // Since there were segments before these, for our purposes we consider them to start at
406 // the 0th element, and since there is at least one segment after each, each was frozen
407 // by the time we snapped it, so we can iterate until each's frozen tail.
408 for (ConcurrentQueueSegment<T> s = head._nextSegment!; s != tail; s = s._nextSegment!)
410 Debug.Assert(s._preservedForObservation);
411 Debug.Assert(s._frozenForEnqueues);
412 count += s._headAndTail.Tail - s.FreezeOffset;
415 // Finally, enumerate the tail. As with the intermediate segments, there were segments
416 // before this in the snapped region, so we can start counting from the beginning. Unlike
417 // the intermediate segments, we can't just go until the Tail, as that could still be changing;
418 // instead we need to go until the tail we snapped for observation.
419 count += tailTail - tail.FreezeOffset;
422 // Return the computed count.
423 return count;
426 /// <summary>
427 /// Copies the <see cref="ConcurrentQueue{T}"/> elements to an existing one-dimensional <see
428 /// cref="Array">Array</see>, starting at the specified array index.
429 /// </summary>
430 /// <param name="array">The one-dimensional <see cref="Array">Array</see> that is the
431 /// destination of the elements copied from the
432 /// <see cref="ConcurrentQueue{T}"/>. The <see cref="Array">Array</see> must have zero-based
433 /// indexing.</param>
434 /// <param name="index">The zero-based index in <paramref name="array"/> at which copying
435 /// begins.</param>
436 /// <exception cref="ArgumentNullException"><paramref name="array"/> is a null reference (Nothing in
437 /// Visual Basic).</exception>
438 /// <exception cref="ArgumentOutOfRangeException"><paramref name="index"/> is less than
439 /// zero.</exception>
440 /// <exception cref="ArgumentException"><paramref name="index"/> is equal to or greater than the
441 /// length of the <paramref name="array"/>
442 /// -or- The number of elements in the source <see cref="ConcurrentQueue{T}"/> is greater than the
443 /// available space from <paramref name="index"/> to the end of the destination <paramref
444 /// name="array"/>.
445 /// </exception>
446 public void CopyTo(T[] array, int index)
448 if (array == null)
450 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.array);
452 if (index < 0)
454 ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.index);
457 // Snap for enumeration
458 ConcurrentQueueSegment<T> head, tail;
459 int headHead, tailTail;
460 SnapForObservation(out head, out headHead, out tail, out tailTail);
462 // Get the number of items to be enumerated
463 long count = GetCount(head, headHead, tail, tailTail);
464 if (index > array.Length - count)
466 ThrowHelper.ThrowArgumentException(ExceptionResource.Arg_ArrayPlusOffTooSmall);
469 // Copy the items to the target array
470 int i = index;
471 using (IEnumerator<T> e = Enumerate(head, headHead, tail, tailTail))
473 while (e.MoveNext())
475 array[i++] = e.Current;
478 Debug.Assert(count == i - index);
481 /// <summary>Returns an enumerator that iterates through the <see cref="ConcurrentQueue{T}"/>.</summary>
482 /// <returns>An enumerator for the contents of the <see
483 /// cref="ConcurrentQueue{T}"/>.</returns>
484 /// <remarks>
485 /// The enumeration represents a moment-in-time snapshot of the contents
486 /// of the queue. It does not reflect any updates to the collection after
487 /// <see cref="GetEnumerator"/> was called. The enumerator is safe to use
488 /// concurrently with reads from and writes to the queue.
489 /// </remarks>
490 public IEnumerator<T> GetEnumerator()
492 ConcurrentQueueSegment<T> head, tail;
493 int headHead, tailTail;
494 SnapForObservation(out head, out headHead, out tail, out tailTail);
495 return Enumerate(head, headHead, tail, tailTail);
498 /// <summary>
499 /// Gets the head and tail information of the current contents of the queue.
500 /// After this call returns, the specified region can be enumerated any number
501 /// of times and will not change.
502 /// </summary>
503 private void SnapForObservation(out ConcurrentQueueSegment<T> head, out int headHead, out ConcurrentQueueSegment<T> tail, out int tailTail)
505 lock (_crossSegmentLock) // _head and _tail may only change while the lock is held.
507 // Snap the head and tail
508 head = _head;
509 tail = _tail;
510 Debug.Assert(head != null);
511 Debug.Assert(tail != null);
512 Debug.Assert(tail._nextSegment == null);
514 // Mark them and all segments in between as preserving, and ensure no additional items
515 // can be added to the tail.
516 for (ConcurrentQueueSegment<T> s = head; ; s = s._nextSegment!)
518 s._preservedForObservation = true;
519 if (s == tail) break;
520 Debug.Assert(s._frozenForEnqueues); // any non-tail should already be marked
522 tail.EnsureFrozenForEnqueues(); // we want to prevent the tailTail from moving
524 // At this point, any dequeues from any segment won't overwrite the value, and
525 // none of the existing segments can have new items enqueued.
527 headHead = Volatile.Read(ref head._headAndTail.Head);
528 tailTail = Volatile.Read(ref tail._headAndTail.Tail);
532 /// <summary>Gets the item stored in the <paramref name="i"/>th entry in <paramref name="segment"/>.</summary>
533 private T GetItemWhenAvailable(ConcurrentQueueSegment<T> segment, int i)
535 Debug.Assert(segment._preservedForObservation);
537 // Get the expected value for the sequence number
538 int expectedSequenceNumberAndMask = (i + 1) & segment._slotsMask;
540 // If the expected sequence number is not yet written, we're still waiting for
541 // an enqueuer to finish storing it. Spin until it's there.
542 if ((segment._slots[i].SequenceNumber & segment._slotsMask) != expectedSequenceNumberAndMask)
544 var spinner = new SpinWait();
545 while ((Volatile.Read(ref segment._slots[i].SequenceNumber) & segment._slotsMask) != expectedSequenceNumberAndMask)
547 spinner.SpinOnce();
551 // Return the value from the slot.
552 return segment._slots[i].Item;
555 private IEnumerator<T> Enumerate(ConcurrentQueueSegment<T> head, int headHead, ConcurrentQueueSegment<T> tail, int tailTail)
557 Debug.Assert(head._preservedForObservation);
558 Debug.Assert(head._frozenForEnqueues);
559 Debug.Assert(tail._preservedForObservation);
560 Debug.Assert(tail._frozenForEnqueues);
562 // Head segment. We've already marked it as not accepting any more enqueues,
563 // so its tail position is fixed, and we've already marked it as preserved for
564 // enumeration (before we grabbed its head), so we can safely enumerate from
565 // its head to its tail.
566 int headTail = (head == tail ? tailTail : Volatile.Read(ref head._headAndTail.Tail)) - head.FreezeOffset;
567 if (headHead < headTail)
569 headHead &= head._slotsMask;
570 headTail &= head._slotsMask;
572 if (headHead < headTail)
574 for (int i = headHead; i < headTail; i++) yield return GetItemWhenAvailable(head, i);
576 else
578 for (int i = headHead; i < head._slots.Length; i++) yield return GetItemWhenAvailable(head, i);
579 for (int i = 0; i < headTail; i++) yield return GetItemWhenAvailable(head, i);
583 // We've enumerated the head. If the tail is the same, we're done.
584 if (head != tail)
586 // Each segment between head and tail, not including head and tail. Since there were
587 // segments before these, for our purposes we consider it to start at the 0th element.
588 for (ConcurrentQueueSegment<T> s = head._nextSegment!; s != tail; s = s._nextSegment!)
590 Debug.Assert(s._preservedForObservation, "Would have had to been preserved as a segment part of enumeration");
591 Debug.Assert(s._frozenForEnqueues, "Would have had to be frozen for enqueues as it's intermediate");
593 int sTail = s._headAndTail.Tail - s.FreezeOffset;
594 for (int i = 0; i < sTail; i++)
596 yield return GetItemWhenAvailable(s, i);
600 // Enumerate the tail. Since there were segments before this, we can just start at
601 // its beginning, and iterate until the tail we already grabbed.
602 tailTail -= tail.FreezeOffset;
603 for (int i = 0; i < tailTail; i++)
605 yield return GetItemWhenAvailable(tail, i);
610 /// <summary>Adds an object to the end of the <see cref="ConcurrentQueue{T}"/>.</summary>
611 /// <param name="item">
612 /// The object to add to the end of the <see cref="ConcurrentQueue{T}"/>.
613 /// The value can be a null reference (Nothing in Visual Basic) for reference types.
614 /// </param>
615 public void Enqueue(T item)
617 // Try to enqueue to the current tail.
618 if (!_tail.TryEnqueue(item))
620 // If we're unable to, we need to take a slow path that will
621 // try to add a new tail segment.
622 EnqueueSlow(item);
626 /// <summary>Adds to the end of the queue, adding a new segment if necessary.</summary>
627 private void EnqueueSlow(T item)
629 while (true)
631 ConcurrentQueueSegment<T> tail = _tail;
633 // Try to append to the existing tail.
634 if (tail.TryEnqueue(item))
636 return;
639 // If we were unsuccessful, take the lock so that we can compare and manipulate
640 // the tail. Assuming another enqueuer hasn't already added a new segment,
641 // do so, then loop around to try enqueueing again.
642 lock (_crossSegmentLock)
644 if (tail == _tail)
646 // Make sure no one else can enqueue to this segment.
647 tail.EnsureFrozenForEnqueues();
649 // We determine the new segment's length based on the old length.
650 // In general, we double the size of the segment, to make it less likely
651 // that we'll need to grow again. However, if the tail segment is marked
652 // as preserved for observation, something caused us to avoid reusing this
653 // segment, and if that happens a lot and we grow, we'll end up allocating
654 // lots of wasted space. As such, in such situations we reset back to the
655 // initial segment length; if these observations are happening frequently,
656 // this will help to avoid wasted memory, and if they're not, we'll
657 // relatively quickly grow again to a larger size.
658 int nextSize = tail._preservedForObservation ? InitialSegmentLength : Math.Min(tail.Capacity * 2, MaxSegmentLength);
659 var newTail = new ConcurrentQueueSegment<T>(nextSize);
661 // Hook up the new tail.
662 tail._nextSegment = newTail;
663 _tail = newTail;
669 /// <summary>
670 /// Attempts to remove and return the object at the beginning of the <see
671 /// cref="ConcurrentQueue{T}"/>.
672 /// </summary>
673 /// <param name="result">
674 /// When this method returns, if the operation was successful, <paramref name="result"/> contains the
675 /// object removed. If no object was available to be removed, the value is unspecified.
676 /// </param>
677 /// <returns>
678 /// true if an element was removed and returned from the beginning of the
679 /// <see cref="ConcurrentQueue{T}"/> successfully; otherwise, false.
680 /// </returns>
681 public bool TryDequeue([MaybeNullWhen(false)] out T result) =>
682 _head.TryDequeue(out result) || // fast-path that operates just on the head segment
683 TryDequeueSlow(out result); // slow path that needs to fix up segments
685 /// <summary>Tries to dequeue an item, removing empty segments as needed.</summary>
686 private bool TryDequeueSlow([MaybeNullWhen(false)] out T item)
688 while (true)
690 // Get the current head
691 ConcurrentQueueSegment<T> head = _head;
693 // Try to take. If we're successful, we're done.
694 if (head.TryDequeue(out item))
696 return true;
699 // Check to see whether this segment is the last. If it is, we can consider
700 // this to be a moment-in-time empty condition (even though between the TryDequeue
701 // check and this check, another item could have arrived).
702 if (head._nextSegment == null)
704 item = default!;
705 return false;
708 // At this point we know that head.Next != null, which means
709 // this segment has been frozen for additional enqueues. But between
710 // the time that we ran TryDequeue and checked for a next segment,
711 // another item could have been added. Try to dequeue one more time
712 // to confirm that the segment is indeed empty.
713 Debug.Assert(head._frozenForEnqueues);
714 if (head.TryDequeue(out item))
716 return true;
719 // This segment is frozen (nothing more can be added) and empty (nothing is in it).
720 // Update head to point to the next segment in the list, assuming no one's beat us to it.
721 lock (_crossSegmentLock)
723 if (head == _head)
725 _head = head._nextSegment;
731 /// <summary>
732 /// Attempts to return an object from the beginning of the <see cref="ConcurrentQueue{T}"/>
733 /// without removing it.
734 /// </summary>
735 /// <param name="result">
736 /// When this method returns, <paramref name="result"/> contains an object from
737 /// the beginning of the <see cref="Concurrent.ConcurrentQueue{T}"/> or default(T)
738 /// if the operation failed.
739 /// </param>
740 /// <returns>true if and object was returned successfully; otherwise, false.</returns>
741 /// <remarks>
742 /// For determining whether the collection contains any items, use of the <see cref="IsEmpty"/>
743 /// property is recommended rather than peeking.
744 /// </remarks>
745 public bool TryPeek([MaybeNullWhen(false)] out T result) => TryPeek(out result, resultUsed: true);
747 /// <summary>Attempts to retrieve the value for the first element in the queue.</summary>
748 /// <param name="result">The value of the first element, if found.</param>
749 /// <param name="resultUsed">true if the result is needed; otherwise false if only the true/false outcome is needed.</param>
750 /// <returns>true if an element was found; otherwise, false.</returns>
751 private bool TryPeek([MaybeNullWhen(false)] out T result, bool resultUsed)
753 // Starting with the head segment, look through all of the segments
754 // for the first one we can find that's not empty.
755 ConcurrentQueueSegment<T> s = _head;
756 while (true)
758 // Grab the next segment from this one, before we peek.
759 // This is to be able to see whether the value has changed
760 // during the peek operation.
761 ConcurrentQueueSegment<T>? next = Volatile.Read(ref s._nextSegment);
763 // Peek at the segment. If we find an element, we're done.
764 if (s.TryPeek(out result, resultUsed))
766 return true;
769 // The current segment was empty at the moment we checked.
771 if (next != null)
773 // If prior to the peek there was already a next segment, then
774 // during the peek no additional items could have been enqueued
775 // to it and we can just move on to check the next segment.
776 Debug.Assert(next == s._nextSegment);
777 s = next;
779 else if (Volatile.Read(ref s._nextSegment) == null)
781 // The next segment is null. Nothing more to peek at.
782 break;
785 // The next segment was null before we peeked but non-null after.
786 // That means either when we peeked the first segment had
787 // already been frozen but the new segment not yet added,
788 // or that the first segment was empty and between the time
789 // that we peeked and then checked _nextSegment, so many items
790 // were enqueued that we filled the first segment and went
791 // into the next. Since we need to peek in order, we simply
792 // loop around again to peek on the same segment. The next
793 // time around on this segment we'll then either successfully
794 // peek or we'll find that next was non-null before peeking,
795 // and we'll traverse to that segment.
798 result = default!;
799 return false;
802 /// <summary>
803 /// Removes all objects from the <see cref="ConcurrentQueue{T}"/>.
804 /// </summary>
805 public void Clear()
807 lock (_crossSegmentLock)
809 // Simply substitute a new segment for the existing head/tail,
810 // as is done in the constructor. Operations currently in flight
811 // may still read from or write to an existing segment that's
812 // getting dropped, meaning that in flight operations may not be
813 // linear with regards to this clear operation. To help mitigate
814 // in-flight operations enqueuing onto the tail that's about to
815 // be dropped, we first freeze it; that'll force enqueuers to take
816 // this lock to synchronize and see the new tail.
817 _tail.EnsureFrozenForEnqueues();
818 _tail = _head = new ConcurrentQueueSegment<T>(InitialSegmentLength);