6727220087

Committed 02 Nov 2023 01:39AM UTC coverage: 98.56% (+0.02%) from 98.539%

Build # 6727220087

Build Type

push

github

Committed by

web-flow

Commit Message

enable code analysis (#446)

Run Details

863 of 892 branches covered (0.0%)

Branch coverage included in aggregate %.

8 of 10 new or added lines in 5 files covered. (80.0%)

5 existing lines in 2 files now uncovered.

3722 of 3760 relevant lines covered (98.99%)

21194350.5 hits per line

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

94.39

/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 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>
        /// <remarks>
        /// Not thread safe.
        /// </remarks>
        public void Clear()
        {
            slots = new Slot[slots.Length];
            headAndTail = new PaddedHeadAndTail();
        }

        [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 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)	450✔
25	{	450✔
26	if (boundedLength < 0)	450✔
27	Throw.ArgOutOfRange(nameof(boundedLength));	3✔
28
29	// must be power of 2 to use & slotsMask instead of %
30	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	447✔
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];	447✔
35	slotsMask = boundedLength - 1;	447✔
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++)	87,774✔
52	{	43,440✔
53	slots[i].SequenceNumber = i;	43,440✔
54	}	43,440✔
55	}	447✔
56
57	/// <summary>
58	/// Gets the number of items contained in the buffer.
59	/// </summary>
60	public int Count
61	{
62	get
63	{	137✔
64	var spinner = new SpinWait();	137✔
65	while (true)	137✔
66	{	137✔
67	var headNow = Volatile.Read(ref headAndTail.Head);	137✔
68	var tailNow = Volatile.Read(ref headAndTail.Tail);	137✔
69
70	if (headNow == Volatile.Read(ref headAndTail.Head) &&	137!
71	tailNow == Volatile.Read(ref headAndTail.Tail))	137✔
72	{	137✔
73	return GetCount(headNow, tailNow);	137✔
74	}
75
UNCOV 76	spinner.SpinOnce();	×
UNCOV 77	}	×
78	}	137✔
79	}
80
81	private int GetCount(int head, int tail)
82	{	137✔
83	if (head != tail)	137✔
84	{	131✔
85	head &= slotsMask;	131✔
86	tail &= slotsMask;	131✔
87
88	return head < tail ? tail - head : slots.Length - head + tail;	131✔
89	}
90	return 0;	6✔
91	}	137✔
92
93	/// <summary>
94	/// The bounded capacity.
95	/// </summary>
96	public int Capacity => slots.Length;	3✔
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	{	84,106✔
105	// Get the head at which to try to dequeue.
106	var currentHead = Volatile.Read(ref headAndTail.Head);	84,106✔
107	var slotsIndex = currentHead & slotsMask;	84,106✔
108
109	// Read the sequence number for the head position.
110	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	84,106✔
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);	84,106✔
115	if (diff == 0)	84,106✔
116	{	84,075✔
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)	84,075✔
126	{	83,950✔
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;	83,950✔
130
131	slots[slotsIndex].Item = default;	83,950✔
132	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentHead + slots.Length);	83,950✔
133
134	return BufferStatus.Success;	83,950✔
135	}
136	}	125✔
137	else if (diff < 0)	31✔
138	{	30✔
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);	30✔
148	if (currentTail - currentHead <= 0)	30!
149	{	30✔
150	item = default;	30✔
151	return BufferStatus.Empty;	30✔
152	}
153	}	×
154
155	item = default;	126✔
156	return BufferStatus.Contended;	126✔
157	}	84,106✔
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	{	4,865,128✔
166	// Get the tail at which to try to return.
167	var currentTail = Volatile.Read(ref headAndTail.Tail);	4,865,128✔
168	var slotsIndex = currentTail & slotsMask;	4,865,128✔
169
170	// Read the sequence number for the tail position.
171	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	4,865,128✔
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;	4,865,128✔
176	if (diff == 0)	4,865,128✔
177	{	138,914✔
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)	138,914✔
187	{	121,118✔
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;	121,118✔
191	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentTail + 1);	121,118✔
192	return BufferStatus.Success;	121,118✔
193	}
194	}	17,796✔
195	else if (diff < 0)	4,726,214✔
196	{	4,726,166✔
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;	4,726,166✔
203	}
204
205	return BufferStatus.Contended;	17,844✔
206	}	4,865,128✔
207
208	/// <summary>
209	/// Removes all values from the buffer.
210	/// </summary>
211	/// <remarks>
212	/// Not thread safe.
213	/// </remarks>
214	public void Clear()
215	{	3✔
216	slots = new Slot[slots.Length];	3✔
217	headAndTail = new PaddedHeadAndTail();	3✔
218	}	3✔
219
220	[StructLayout(LayoutKind.Auto)]
221	[DebuggerDisplay("Item = {Item}, SequenceNumber = {SequenceNumber}")]
222	internal struct Slot
223	{
224	/// <summary>The item.</summary>
225	public T Item;
226	/// <summary>The sequence number for this slot, used to synchronize between enqueuers and dequeuers.</summary>
227	public int SequenceNumber;
228	}
229	}
230	}

bitfaster / BitFaster.Caching / 6727220087

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