7076079219

Committed 03 Dec 2023 09:51AM UTC coverage: 99.1% (-0.1%) from 99.236%

Build # 7076079219

Build Type

push

github

Committed by

web-flow

Commit Message

Make buffer Clear() thread safe (#523)

Run Details

945 of 967 branches covered (0.0%)

Branch coverage included in aggregate %.

8 of 8 new or added lines in 3 files covered. (100.0%)

5 existing lines in 2 files now uncovered.

4118 of 4142 relevant lines covered (99.42%)

51179989.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.55

/BitFaster.Caching/Buffers/MpmcBoundedBuffer.cs

﻿using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;

namespace BitFaster.Caching.Buffers
{
    /// <summary>
    /// Provides a multi-producer, multi-consumer thread-safe ring buffer. When the buffer is full,
    /// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
    /// </summary>
    /// Based on the Segment internal class from the .NET ConcurrentQueue class.
    public sealed class MpmcBoundedBuffer<T>
    {
        private readonly Slot[] slots;
        private readonly int slotsMask;
        private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly

        /// <summary>
        /// Initializes a new instance of the MpmcBoundedBuffer class with the specified bounded capacity.
        /// </summary>
        /// <param name="boundedLength">The bounded length.</param>
        /// <exception cref="ArgumentOutOfRangeException"></exception>
        public MpmcBoundedBuffer(int boundedLength)
        {
            if (boundedLength < 0)
                Throw.ArgOutOfRange(nameof(boundedLength));

            // must be power of 2 to use & slotsMask instead of %
            boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);

            // Initialize the slots and the mask.  The mask is used as a way of quickly doing "% _slots.Length",
            // instead letting us do "& _slotsMask".
            slots = new Slot[boundedLength];
            slotsMask = boundedLength - 1;

            // Initialize the sequence number for each slot.  The sequence number provides a ticket that
            // allows dequeuers to know whether they can dequeue and enqueuers to know whether they can
            // enqueue.  An enqueuer at position N can enqueue when the sequence number is N, and a dequeuer
            // for position N can dequeue when the sequence number is N + 1.  When an enqueuer is done writing
            // at position N, it sets the sequence number to N + 1 so that a dequeuer will be able to dequeue,
            // and when a dequeuer is done dequeueing at position N, it sets the sequence number to N + _slots.Length,
            // so that when an enqueuer loops around the slots, it'll find that the sequence number at
            // position N is N.  This also means that when an enqueuer finds that at position N the sequence
            // number is < N, there is still a value in that slot, i.e. the segment is full, and when a
            // dequeuer finds that the value in a slot is < N + 1, there is nothing currently available to
            // dequeue. (It is possible for multiple enqueuers to enqueue concurrently, writing into
            // subsequent slots, and to have the first enqueuer take longer, so that the slots for 1, 2, 3, etc.
            // may have values, but the 0th slot may still be being filled... in that case, TryDequeue will
            // return false.)
            for (var i = 0; i < slots.Length; i++)
            {
                slots[i].SequenceNumber = i;
            }
        }

        /// <summary>
        /// Gets the number of items contained in the buffer.
        /// </summary>
        public int Count
        {
            get
            {
                var spinner = new SpinWait();
                while (true)
                {
                    var headNow = Volatile.Read(ref headAndTail.Head);
                    var tailNow = Volatile.Read(ref headAndTail.Tail);

                    if (headNow == Volatile.Read(ref headAndTail.Head) &&
                        tailNow == Volatile.Read(ref headAndTail.Tail))
                    {
                        return GetCount(headNow, tailNow);
                    }

                    spinner.SpinOnce();
                }
            }
        }

        private int GetCount(int head, int tail)
        {
            if (head != tail)
            {
                head &= slotsMask;
                tail &= slotsMask;

                return head < tail ? tail - head : slots.Length - head + tail;
            }
            return 0;
        }

        /// <summary>
        /// The bounded capacity.
        /// </summary>
        public int Capacity => slots.Length;

        /// <summary>
        /// Tries to remove an item.
        /// </summary>
        /// <param name="item">The item to be removed.</param>
        /// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
        public BufferStatus TryTake(out T item)
        {
            // Get the head at which to try to dequeue.
            var currentHead = Volatile.Read(ref headAndTail.Head);
            var slotsIndex = currentHead & slotsMask;

            // Read the sequence number for the head position.
            var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);

            // We can dequeue from this slot if it's been filled by an enqueuer, which
            // would have left the sequence number at pos+1.
            var diff = sequenceNumber - (currentHead + 1);
            if (diff == 0)
            {
                // We may be racing with other dequeuers.  Try to reserve the slot by incrementing
                // the head.  Once we've done that, no one else will be able to read from this slot,
                // and no enqueuer will be able to read from this slot until we've written the new
                // sequence number. WARNING: The next few lines are not reliable on a runtime that
                // supports thread aborts. If a thread abort were to sneak in after the CompareExchange
                // but before the Volatile.Write, enqueuers trying to enqueue into this slot would
                // spin indefinitely.  If this implementation is ever used on such a platform, this
                // if block should be wrapped in a finally / prepared region.
                if (Interlocked.CompareExchange(ref headAndTail.Head, currentHead + 1, currentHead) == currentHead)
                {
                    // Successfully reserved the slot.  Note that after the above CompareExchange, other threads
                    // trying to dequeue from this slot will end up spinning until we do the subsequent Write.
                    item = slots[slotsIndex].Item;

                    slots[slotsIndex].Item = default;
                    Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentHead + slots.Length);

                    return BufferStatus.Success;
                }
            }
            else if (diff < 0)
            {
                // The sequence number was less than what we needed, which means this slot doesn't
                // yet contain a value we can dequeue, i.e. the segment is empty.  Technically it's
                // possible that multiple enqueuers could have written concurrently, with those
                // getting later slots actually finishing first, so there could be elements after
                // this one that are available, but we need to dequeue in order.  So before declaring
                // failure and that the segment is empty, we check the tail to see if we're actually
                // empty or if we're just waiting for items in flight or after this one to become available.
                //bool frozen = _frozenForEnqueues;
                var currentTail = Volatile.Read(ref headAndTail.Tail);
                if (currentTail - currentHead <= 0)
                {
                    item = default;
                    return BufferStatus.Empty;
                }
            }

            item = default;
            return BufferStatus.Contended;
        }

        /// <summary>
        /// Tries to add the specified item.
        /// </summary>
        /// <param name="item">The item to be added.</param>
        /// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
        public BufferStatus TryAdd(T item)
        {
            // Get the tail at which to try to return.
            var currentTail = Volatile.Read(ref headAndTail.Tail);
            var slotsIndex = currentTail & slotsMask;

            // Read the sequence number for the tail position.
            var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);

            // The slot is empty and ready for us to enqueue into it if its sequence
            // number matches the slot.
            var diff = sequenceNumber - currentTail;
            if (diff == 0)
            {
                // We may be racing with other enqueuers.  Try to reserve the slot by incrementing
                // the tail.  Once we've done that, no one else will be able to write to this slot,
                // and no dequeuer will be able to read from this slot until we've written the new
                // sequence number. WARNING: The next few lines are not reliable on a runtime that
                // supports thread aborts. If a thread abort were to sneak in after the CompareExchange
                // but before the Volatile.Write, other threads will spin trying to access this slot.
                // If this implementation is ever used on such a platform, this if block should be
                // wrapped in a finally / prepared region.
                if (Interlocked.CompareExchange(ref headAndTail.Tail, currentTail + 1, currentTail) == currentTail)
                {
                    // Successfully reserved the slot.  Note that after the above CompareExchange, other threads
                    // trying to return will end up spinning until we do the subsequent Write.
                    slots[slotsIndex].Item = item;
                    Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentTail + 1);
                    return BufferStatus.Success;
                }
            }
            else if (diff < 0)
            {
                // The sequence number was less than what we needed, which means this slot still
                // contains a value, i.e. the segment is full.  Technically it's possible that multiple
                // dequeuers could have read concurrently, with those getting later slots actually
                // finishing first, so there could be spaces after this one that are available, but
                // we need to enqueue in order.
                return BufferStatus.Full;
            }

            return BufferStatus.Contended;
        }

        /// <summary>
        /// Removes all values from the buffer.
        /// </summary>
        public void Clear()
        {
            while (TryTake(out _) != BufferStatus.Empty)
            { 
            }
        }

        [StructLayout(LayoutKind.Auto)]
        [DebuggerDisplay("Item = {Item}, SequenceNumber = {SequenceNumber}")]
        internal struct Slot
        {
            /// <summary>The item.</summary>
            public T Item;
            /// <summary>The sequence number for this slot, used to synchronize between enqueuers and dequeuers.</summary>
            public int SequenceNumber;
        }
    }
}

1	using System;
2	using System.Diagnostics;
3	using System.Runtime.InteropServices;
4	using System.Threading;
5
6	namespace BitFaster.Caching.Buffers
7	{
8	/// <summary>
9	/// Provides a multi-producer, multi-consumer thread-safe ring buffer. When the buffer is full,
10	/// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
11	/// </summary>
12	/// Based on the Segment internal class from the .NET ConcurrentQueue class.
13	public sealed class MpmcBoundedBuffer<T>
14	{
15	private readonly Slot[] slots;
16	private readonly int slotsMask;
17	private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly
18
19	/// <summary>
20	/// Initializes a new instance of the MpmcBoundedBuffer class with the specified bounded capacity.
21	/// </summary>
22	/// <param name="boundedLength">The bounded length.</param>
23	/// <exception cref="ArgumentOutOfRangeException"></exception>
24	public MpmcBoundedBuffer(int boundedLength)	596✔
25	{	596✔
26	if (boundedLength < 0)	596✔
27	Throw.ArgOutOfRange(nameof(boundedLength));	4✔
28
29	// must be power of 2 to use & slotsMask instead of %
30	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	592✔
31
32	// Initialize the slots and the mask. The mask is used as a way of quickly doing "% _slots.Length",
33	// instead letting us do "& _slotsMask".
34	slots = new Slot[boundedLength];	592✔
35	slotsMask = boundedLength - 1;	592✔
36
37	// Initialize the sequence number for each slot. The sequence number provides a ticket that
38	// allows dequeuers to know whether they can dequeue and enqueuers to know whether they can
39	// enqueue. An enqueuer at position N can enqueue when the sequence number is N, and a dequeuer
40	// for position N can dequeue when the sequence number is N + 1. When an enqueuer is done writing
41	// at position N, it sets the sequence number to N + 1 so that a dequeuer will be able to dequeue,
42	// and when a dequeuer is done dequeueing at position N, it sets the sequence number to N + _slots.Length,
43	// so that when an enqueuer loops around the slots, it'll find that the sequence number at
44	// position N is N. This also means that when an enqueuer finds that at position N the sequence
45	// number is < N, there is still a value in that slot, i.e. the segment is full, and when a
46	// dequeuer finds that the value in a slot is < N + 1, there is nothing currently available to
47	// dequeue. (It is possible for multiple enqueuers to enqueue concurrently, writing into
48	// subsequent slots, and to have the first enqueuer take longer, so that the slots for 1, 2, 3, etc.
49	// may have values, but the 0th slot may still be being filled... in that case, TryDequeue will
50	// return false.)
51	for (var i = 0; i < slots.Length; i++)	116,896✔
52	{	57,856✔
53	slots[i].SequenceNumber = i;	57,856✔
54	}	57,856✔
55	}	592✔
56
57	/// <summary>
58	/// Gets the number of items contained in the buffer.
59	/// </summary>
60	public int Count
61	{
62	get
63	{	188✔
64	var spinner = new SpinWait();	188✔
65	while (true)	188✔
66	{	188✔
67	var headNow = Volatile.Read(ref headAndTail.Head);	188✔
68	var tailNow = Volatile.Read(ref headAndTail.Tail);	188✔
69
70	if (headNow == Volatile.Read(ref headAndTail.Head) &&	188!
71	tailNow == Volatile.Read(ref headAndTail.Tail))	188✔
72	{	188✔
73	return GetCount(headNow, tailNow);	188✔
74	}
75
UNCOV 76	spinner.SpinOnce();	×
UNCOV 77	}	×
78	}	188✔
79	}
80
81	private int GetCount(int head, int tail)
82	{	188✔
83	if (head != tail)	188✔
84	{	180✔
85	head &= slotsMask;	180✔
86	tail &= slotsMask;	180✔
87
88	return head < tail ? tail - head : slots.Length - head + tail;	180✔
89	}
90	return 0;	8✔
91	}	188✔
92
93	/// <summary>
94	/// The bounded capacity.
95	/// </summary>
96	public int Capacity => slots.Length;	4✔
97
98	/// <summary>
99	/// Tries to remove an item.
100	/// </summary>
101	/// <param name="item">The item to be removed.</param>
102	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
103	public BufferStatus TryTake(out T item)
104	{	198,272✔
105	// Get the head at which to try to dequeue.
106	var currentHead = Volatile.Read(ref headAndTail.Head);	198,272✔
107	var slotsIndex = currentHead & slotsMask;	198,272✔
108
109	// Read the sequence number for the head position.
110	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	198,272✔
111
112	// We can dequeue from this slot if it's been filled by an enqueuer, which
113	// would have left the sequence number at pos+1.
114	var diff = sequenceNumber - (currentHead + 1);	198,272✔
115	if (diff == 0)	198,272✔
116	{	198,259✔
117	// We may be racing with other dequeuers. Try to reserve the slot by incrementing
118	// the head. Once we've done that, no one else will be able to read from this slot,
119	// and no enqueuer will be able to read from this slot until we've written the new
120	// sequence number. WARNING: The next few lines are not reliable on a runtime that
121	// supports thread aborts. If a thread abort were to sneak in after the CompareExchange
122	// but before the Volatile.Write, enqueuers trying to enqueue into this slot would
123	// spin indefinitely. If this implementation is ever used on such a platform, this
124	// if block should be wrapped in a finally / prepared region.
125	if (Interlocked.CompareExchange(ref headAndTail.Head, currentHead + 1, currentHead) == currentHead)	198,259✔
126	{	198,058✔
127	// Successfully reserved the slot. Note that after the above CompareExchange, other threads
128	// trying to dequeue from this slot will end up spinning until we do the subsequent Write.
129	item = slots[slotsIndex].Item;	198,058✔
130
131	slots[slotsIndex].Item = default;	198,058✔
132	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentHead + slots.Length);	198,058✔
133
134	return BufferStatus.Success;	198,058✔
135	}
136	}	201✔
137	else if (diff < 0)	13✔
138	{	12✔
139	// The sequence number was less than what we needed, which means this slot doesn't
140	// yet contain a value we can dequeue, i.e. the segment is empty. Technically it's
141	// possible that multiple enqueuers could have written concurrently, with those
142	// getting later slots actually finishing first, so there could be elements after
143	// this one that are available, but we need to dequeue in order. So before declaring
144	// failure and that the segment is empty, we check the tail to see if we're actually
145	// empty or if we're just waiting for items in flight or after this one to become available.
146	//bool frozen = _frozenForEnqueues;
147	var currentTail = Volatile.Read(ref headAndTail.Tail);	12✔
148	if (currentTail - currentHead <= 0)	12!
149	{	12✔
150	item = default;	12✔
151	return BufferStatus.Empty;	12✔
152	}
153	}	×
154
155	item = default;	202✔
156	return BufferStatus.Contended;	202✔
157	}	198,272✔
158
159	/// <summary>
160	/// Tries to add the specified item.
161	/// </summary>
162	/// <param name="item">The item to be added.</param>
163	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
164	public BufferStatus TryAdd(T item)
165	{	6,056,346✔
166	// Get the tail at which to try to return.
167	var currentTail = Volatile.Read(ref headAndTail.Tail);	6,056,346✔
168	var slotsIndex = currentTail & slotsMask;	6,056,346✔
169
170	// Read the sequence number for the tail position.
171	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	6,056,346✔
172
173	// The slot is empty and ready for us to enqueue into it if its sequence
174	// number matches the slot.
175	var diff = sequenceNumber - currentTail;	6,056,346✔
176	if (diff == 0)	6,056,346✔
177	{	301,462✔
178	// We may be racing with other enqueuers. Try to reserve the slot by incrementing
179	// the tail. Once we've done that, no one else will be able to write to this slot,
180	// and no dequeuer will be able to read from this slot until we've written the new
181	// sequence number. WARNING: The next few lines are not reliable on a runtime that
182	// supports thread aborts. If a thread abort were to sneak in after the CompareExchange
183	// but before the Volatile.Write, other threads will spin trying to access this slot.
184	// If this implementation is ever used on such a platform, this if block should be
185	// wrapped in a finally / prepared region.
186	if (Interlocked.CompareExchange(ref headAndTail.Tail, currentTail + 1, currentTail) == currentTail)	301,462✔
187	{	247,629✔
188	// Successfully reserved the slot. Note that after the above CompareExchange, other threads
189	// trying to return will end up spinning until we do the subsequent Write.
190	slots[slotsIndex].Item = item;	247,629✔
191	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentTail + 1);	247,629✔
192	return BufferStatus.Success;	247,629✔
193	}
194	}	53,833✔
195	else if (diff < 0)	5,754,884✔
196	{	5,753,775✔
197	// The sequence number was less than what we needed, which means this slot still
198	// contains a value, i.e. the segment is full. Technically it's possible that multiple
199	// dequeuers could have read concurrently, with those getting later slots actually
200	// finishing first, so there could be spaces after this one that are available, but
201	// we need to enqueue in order.
202	return BufferStatus.Full;	5,753,775✔
203	}
204
205	return BufferStatus.Contended;	54,942✔
206	}	6,056,346✔
207
208	/// <summary>
209	/// Removes all values from the buffer.
210	/// </summary>
211	public void Clear()
212	{	4✔
213	while (TryTake(out _) != BufferStatus.Empty)	12✔
214	{	8✔
215	}	8✔
216	}	4✔
217
218	[StructLayout(LayoutKind.Auto)]
219	[DebuggerDisplay("Item = {Item}, SequenceNumber = {SequenceNumber}")]
220	internal struct Slot
221	{
222	/// <summary>The item.</summary>
223	public T Item;
224	/// <summary>The sequence number for this slot, used to synchronize between enqueuers and dequeuers.</summary>
225	public int SequenceNumber;
226	}
227	}
228	}

bitfaster / BitFaster.Caching / 7076079219

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous